S21-X14 · 多 Agent 协作：角色分工与消息传递

把任务拆给四个 agent，运行三轮后开始互相追问"你上一步的输出是什么"，然后一个人写、三个人评论，最后谁也没交付。多 agent 的失败很少是模型不够强，而是契约没写、消息没归属、终止条件没定义。这篇只讲这三件事怎么落地。

1. 什么情况下才该拆

拆分的正当理由只有三条，其余都是"看起来更智能"。

理由	判断方式	反例
上下文隔离	每个角色只需读一小部分材料，混在一起会互相干扰	材料本来就短，拆了反而看不全
专业分工	不同角色需要不同的工具集/提示词/权限	只是换了个角色名字
并行探索	多条独立路径可同时验证，再择优	有先后依赖却硬并行

反过来的信号：如果子任务之间有强顺序依赖，或者一个 agent 已经能覆盖 80% 的输入，拆开只会增加通信开销与失败点。

一条实用判据：画出信息流。 如果每条边上的消息都可以只用一句话描述清楚（"订单号与状态"、"候选方案列表"），拆分是健康的；如果某条边上要传"全部上下文"，说明拆错了。

2. 三种拓扑

拓扑	结构	通信开销	适合
主管-工人（supervisor-worker）	一个调度者分派、汇总	中（每轮过一次主管）	任务可枚举、需要统一裁决
流水线（pipeline）	A→B→C 顺序交接	低（只传接口数据）	阶段清晰、契约明确
对等讨论（peer）	多角色互评直到收敛	高（O(n²) 消息）	需要多视角质疑、且能判定收敛

主管模式是最稳的默认值：唯一入口、唯一年出口，出错时容易定位是哪个工人还是调度判错。

# LangGraph 里的主管拓扑骨架
g.add_node("supervisor", route_next)
for w in ("researcher", "coder", "reviewer"):
    g.add_node(w, make_worker(w))
    g.add_edge(w, "supervisor")          # 工人干完必回主管
g.add_conditional_edges("supervisor", pick, {
    "researcher": "researcher", "coder": "coder",
    "reviewer": "reviewer", "done": END,
})

对等讨论只在有明确裁决机制时才用，否则就是无限互评。

3. 角色契约：输入输出必须结构化

角色描述写成"你是一个资深工程师"，等于没写。契约要能回答：给我什么、还你什么、什么算完成、失败怎么报。

契约要素	反例	合格写法
输入	"根据上下文处理"	`{"order_id": str, "policy_text": str}`
输出	"给出分析"	JSON：`{"verdict": "pass 或 fail", "reasons": [str]}`
完成判据	"分析清楚即可"	`verdict` 非空且 `reasons` 至少一条
失败上报	直接抛异常	`{"error": "...", "needs": "缺少 policy_text"}`
边界	未声明	只判断合规性，不改写原文

from pydantic import BaseModel, Field

class ReviewIn(BaseModel):
    order_id: str
    policy_text: str = Field(description="规范原文，不得为空")

class ReviewOut(BaseModel):
    verdict: str = Field(pattern="^(pass|fail)$")
    reasons: list[str] = Field(min_length=1)
    needs: str | None = None      # 缺信息时填这里，而不是反问你

needs 字段是关键设计：它把"互相追问"变成"结构化缺件"，主管一次补齐，不用来回聊天。

4. 消息传递的三种机制

机制	形态	一致性	风险
共享黑板	所有 agent 读写同一份状态	需要显式写锁/单写者	信息污染、互相覆盖
点对点	只按契约传字段	高（结构固定）	拓扑变复杂后边数爆炸
事件总线	发布/订阅，异步	最终一致	难以复现、顺序不定

# 黑板 + 单写者：只有主管能提交，工人只返回建议
class Board:
    def __init__(self):
        self.state = {"facts": {}, "decisions": [], "open_questions": []}

    def submit(self, patch: dict, author: str):
        if author != "supervisor":
            raise PermissionError("只有主管能写入黑板")
        self.state["decisions"].append({"by": author, **patch})

    def snapshot(self, keys: list[str]) -> dict:
        return {k: self.state["facts"].get(k) for k in keys}   # 按需切片下发

"单写者 + 切片下发"能同时解决两个问题：写冲突没了，工人也只看到自己该看的那部分。

5. 上下文隔离：价值与代价

好处	代价
每个 agent 的提示词更短，干扰更少	信息不共享，重复检索
权限可按角色裁剪	跨角色的事实可能不一致
单角色可独立回归测试	需要额外的"事实汇总"节点

隔离不是免费的。典型代价是同一份材料被检索两三次：检索成本上升，但换来的是判断质量稳定。

折中做法：设一个"事实层"（facts），由检索 agent 写入、其他 agent 只读。事实只写一次，观点各自生成——避免两个人对同一个数字给出不同版本。

facts = {"order_status": "PAID", "amount": 199.00}
ctx_for_coder = {"facts": facts, "task": "生成退款单"}
ctx_for_reviewer = {"facts": facts, "candidate": candidate, "policy_text": policy}

6. 终止条件：三种判据，至少选两种

判据	实现	单独使用的问题
主管判定	主管输出 `done`	主管自己也可能反复
投票	多角色多数同意	平票无解、容易串通
轮次上限	`max_rounds` 硬截断	到上限仍没结论，需降级策略

def should_stop(state) -> str:
    if state["rounds"] >= MAX_ROUNDS:          # 硬上限永远要有
        return "degrade"
    if state.get("verdict") in ("pass", "fail") and state.get("reasons"):
        return "done"
    if state["open_questions"] == []:
        return "done"
    return "continue"

硬上限必须配降级动作：输出当前最佳结论 + 未决问题清单，而不是抛错。用户宁可拿到"部分结论 + 待确认项"，也不要一个 500。

7. 冲突：两个 agent 结论相反时谁裁决

冲突类型	裁决方	手段
事实冲突（数字/状态不一致）	事实层	回到数据源重查，只有一份权威
判断冲突（通过 vs 不通过）	主管或规则	优先级规则、按条款号取严
方案冲突（A 方案 vs B 方案）	人	交给人工选择，别让模型投票

原则：事实交给数据，判断交给规则，取舍交给人。 让两个模型互相说服，是最贵且最不稳定的方案。

def arbitrate(a: dict, b: dict) -> dict:
    if a["facts_hash"] != b["facts_hash"]:
        raise DataInconsistent("事实不一致，回到 facts 层重查")
    if a["verdict"] != b["verdict"]:
        return {"verdict": "escalate_to_human", "candidates": [a, b]}
    return a

8. 成本结构：为什么会翻倍

调用次数不是"agent 数"，而是三者相乘的结果：

总调用 ≈ Σ(每轮 agent 数 × 轮次) + Σ(工具调用次数)

结构	调用量级	说明
单 agent	1 × 轮次	基线
主管-3 工人	约 4 × 轮次	主管每轮都要跑一次
对等 4 角色 × 3 轮	约 12 + 仲裁	消息数按 O(n²) 增长
加一个"评审"环节	再乘一次	评审本身也要看全文

要压成本，优先级是：减少轮次 > 减少角色 > 缩短每个角色的上下文 > 换小模型。把评审角色换成规则校验，往往比换模型更有效。

9. 可观测：每条轨迹要能对上因果

要记录	为什么
`trace_id` + `span_id`	跨 agent 串成一条链
`agent_id` / `role`	定位是谁做的决定
输入摘要 + 输出全文	复盘时不用猜
工具名 + 入参 + 结果	判断"该调没调"
轮次序号	看是否在打转

with tracer.start_as_current_span(f"agent.{role}") as span:
    span.set_attribute("trace.round", state["rounds"])
    span.set_attribute("agent.role", role)
    span.set_attribute("io.input_digest", digest(payload))
    span.set_attribute("io.output", out_json[:2000])

跨 agent 的因果链比单 agent 轨迹更重要：没有 trace_id 串起来，你只能看到四次独立失败，看不到"主管把任务分错了"。

10. 排错：四个高频坑

现象	根因	处理
两个 agent 无限互相询问	契约没定义"缺件"表达方式	加 `needs` 字段，由主管一次补齐
讨论收敛不了	终止条件只有"投票"	加轮次硬上限 + 降级输出
后一个 agent 被前一个的错结论带跑	共享上下文，错误被当事实继承	事实层与观点层分离，事实单写者
并行后整体更慢	并行段太短，调度与合并开销占主导	只并行真正独立的长任务；短的串行做

前两条本质是同一件事：你没有定义一个"可以结束的状态"。多 agent 系统的设计重心不在角色，而在状态的收敛性。

11. 边界：这些情况本就不该拆

场景	更合适	理由
单步抽取/改写	单 agent 单次调用	拆分只增加失败点
严格顺序的业务流程	工作流代码 + 一次模型调用	不需要协商
需要强一致的事务	后端服务内完成	agent 之间没有事务
只是想让结果"更可信"	加校验步骤或规则	多角色不等于更可信
团队还没有 trace	先做单 agent 可观测	多 agent 会让黑盒更难查

12. 自测

你的拆分理由是上下文隔离、专业分工、并行探索中的哪一条？能举出反例吗？
角色契约里缺了"缺件"字段会发生什么？你怎么补？
共享黑板为什么要求单写者？多写者会带来哪两类问题？
终止条件只用投票为什么不够？硬上限触发后该输出什么？
两个 agent 对同一笔金额给出不同数字，该由谁裁决？
多 agent 的成本里，哪一项最容易失控，先砍哪个？