第 28 章 案例四 多智能体协作系统

本章要解决的问题

行业分析报告需要研究员、数据员、写手协作完成——多智能体系统的拓扑、通信、可观测性实战。

章节大纲

• 28.1 多 Agent 拓扑设计（主从 + 流水线）
• 28.2 通信协议与任务编排
• 28.3 可观测性与调试
• 28.4 上线与迭代
• 🛠 解决方案：多智能体上下文隔离方案 + 协作效率度量

28.1 场景与拓扑设计

28.1.1 场景：行业分析报告生成

业务：投研团队要生成行业分析报告，人工流程：

研究员查资料 → 数据分析师算指标 → 写手撰写 → 分析师复核

为什么适合多智能体（第 16 章三个信号全中）：

• 专业隔离：研究员/分析师/写手的角色与工具完全不同。
• 上下文隔离：各角色的工作上下文都很大，混在一起会互相污染。
• 并行收益：部分子任务可并行。

28.1.2 拓扑选择：主从 + 流水线混合

                    ┌──► <span>[研究员 Agent]</span> 查资料 → 事实摘要
                    │        │
<span>[主管 Agent]</span> 接任务 ─┼──► <span>[数据分析师]</span> 算指标 → 数据结论
    │ 拆解/派活/汇总  │        │
    │               └──► <span>[写手 Agent]</span> 撰写报告
    │                      │
    └──► <span>[复核 Agent]</span> 质检 → 通过 → 交付 / 不通过 → 打回

图 1：主从+流水线混合拓扑

设计要点（第 16 章原则）：

• 主管-工人：主管负责拆解、派活、汇总——职责清晰。
• 流水线尾部：研究员/分析师 → 写手 → 复核，顺序推进。
• 复核即质检：复核 Agent 是对整个系统的"最后防线"（呼应第 17 章自检）。

28.1.3 角色定义（第 16 章角色四原则）

示例代码：以下代码演示核心结构，省略了异常处理、日志和完整 import。

ROLES = {
    <span>"supervisor"</span>: {<span>"role"</span>: <span>"项目主管：拆解任务、派发给 worker、汇总质检"</span>},
    <span>"researcher"</span>: {<span>"role"</span>: <span>"行业研究员：检索资料，输出事实与数据摘要"</span>,
                   <span>"tools"</span>: [<span>"search"</span>, <span>"web_fetch"</span>]},
    <span>"analyst"</span>:   {<span>"role"</span>: <span>"数据分析师：计算行业指标，输出趋势判断"</span>,
                  <span>"tools"</span>: [<span>"calculator"</span>, <span>"data_query"</span>]},
    <span>"writer"</span>:    {<span>"role"</span>: <span>"报告写手：基于研究+分析结果撰写结构化报告"</span>,
                  <span>"tools"</span>: []},
    <span>"reviewer"</span>:  {<span>"role"</span>: <span>"报告复核员：检查报告准确性、完整性、逻辑性"</span>,
                  <span>"tools"</span>: []},
}

角色单一职责：写手没有检索工具（最小权限），复核员不写内容（避免"既当运动员又当裁判"）。

28.2 通信协议与任务编排

28.2.1 通信模式：结构化传递（第 16 章）

不用自由对话（群聊会失控），用结构化传递——每个 worker 只输出 JSON 结果：

<span># 研究员输出（数据契约）</span>
researcher_output = {
    <span>"findings"</span>: [{<span>"topic"</span>: <span>"市场规模"</span>, <span>"fact"</span>: <span>"..."</span>, <span>"source"</span>: <span>"..."</span>}],
    <span>"summary"</span>: <span>"行业规模约 X 亿元，增速 Y%"</span>,
}

<span># 分析师输出</span>
analyst_output = {
    <span>"metrics"</span>: {<span>"market_size"</span>: <span>1200</span>, <span>"growth"</span>: <span>0.15</span>, <span>"concentration"</span>: <span>"高"</span>},
    <span>"trend"</span>: <span>"头部集中度提升"</span>,
}

<span># 写手输入：只接收上面两个结构化结果（不接收原始聊天）</span>
writer_input = {<span>"research"</span>: researcher_output, <span>"analysis"</span>: analyst_output}

28.2.2 主管编排：任务图（第 15/16 章）

TASK_GRAPH = {
    <span>"researcher"</span>: {<span>"depends_on"</span>: [], <span>"parallel_group"</span>: <span>1</span>},
    <span>"analyst"</span>:    {<span>"depends_on"</span>: [], <span>"parallel_group"</span>: <span>1</span>},   <span># 与研究员并行</span>
    <span>"writer"</span>:     {<span>"depends_on"</span>: [<span>"researcher"</span>, <span>"analyst"</span>]},
    <span>"reviewer"</span>:   {<span>"depends_on"</span>: [<span>"writer"</span>]},
}

<span>def</span> <span>orchestrate</span>(<span>goal, budget</span>):
    <span># 伪代码：Agent、ROLES、run_worker 为编排框架的封装，此处仅示意调度结构</span>
    supervisor = Agent(role=ROLES[<span>"supervisor"</span>])
    results = {}
    <span># 第一波：并行跑研究员 + 分析师（第12章）</span>
    group1 = [run_worker(<span>"researcher"</span>, goal), run_worker(<span>"analyst"</span>, goal)]
    results[<span>"researcher"</span>], results[<span>"analyst"</span>] = asyncio.gather(*group1)
    <span># 第二波：写手（依赖前两者）</span>
    results[<span>"writer"</span>] = run_worker(<span>"writer"</span>, results)
    <span># 第三波：复核（质检，第17章）</span>
    review = run_worker(<span>"reviewer"</span>, results)
    <span>if</span> <span>not</span> review[<span>"pass"</span>]:
        results[<span>"writer"</span>] = revise_writer(results, review[<span>"issues"</span>])
        review = run_worker(<span>"reviewer"</span>, results)   <span># 复核重跑</span>
    <span>return</span> synthesize(results, review)

图 2：任务编排图

编排要点：并行组用 asyncio.gather；依赖关系用任务图（depends_on）；复核不通过打回写手（闭环）。

28.2.3 上下文隔离：按需投喂（第 16 章核心）

<span>def</span> <span>run_worker</span>(<span>name, context</span>):
    <span># 只给该 worker 需要的部分（不广播全量）</span>
    <span>if</span> name == <span>"researcher"</span>:
        feed = {<span>"goal"</span>: context[<span>"goal"</span>]}                    <span># 只要目标</span>
    <span>elif</span> name == <span>"analyst"</span>:
        feed = {<span>"goal"</span>: context[<span>"goal"</span>], <span>"data"</span>: context[<span>"raw_data"</span>]}
    <span>elif</span> name == <span>"writer"</span>:
        feed = {<span>"research"</span>: context[<span>"researcher"</span>],          <span># 只吃结构化结果</span>
                <span>"analysis"</span>: context[<span>"analyst"</span>]}
    <span>return</span> worker_call(name, feed)

隔离三手段（第 16 章）：输入隔离（按需投喂）+ 输出隔离（只交换 JSON）+ 状态隔离（各 worker 独立）。

28.3 可观测性与调试

28.3.1 多智能体的可观测性：比单 Agent 难 10 倍

多智能体出问题，可能是：主管拆解错、某个 worker 算错、上下文串扰、通信协议断……没有全链路 trace，根本无从查起。

28.3.2 分层 trace（第 21 章扩展）

TRACE = {
    <span>"trace_id"</span>: <span>"t-multi-001"</span>,
    <span>"task"</span>: <span>"生成半导体行业分析报告"</span>,
    <span>"spans"</span>: [
        {<span>"span_id"</span>: <span>"supervisor-plan"</span>, <span>"type"</span>: <span>"llm"</span>,
         <span>"worker"</span>: <span>"supervisor"</span>, <span>"output"</span>: <span>"计划: 研究员+分析师并行→写手→复核"</span>},
        {<span>"span_id"</span>: <span>"researcher-1"</span>, <span>"type"</span>: <span>"llm"</span>, <span>"worker"</span>: <span>"researcher"</span>,
         <span>"input_feed"</span>: <span>"目标摘要"</span>,          <span># 记录投喂内容（可查串扰）</span>
         <span>"output"</span>: <span>"发现3条事实"</span>},
        {<span>"span_id"</span>: <span>"analyst-1"</span>, <span>"type"</span>: <span>"llm"</span>, <span>"worker"</span>: <span>"analyst"</span>,
         <span>"input_feed"</span>: <span>"原始指标数据"</span>, <span>"output"</span>: <span>"结论: 增长 15%"</span>},
        {<span>"span_id"</span>: <span>"writer-1"</span>, <span>"type"</span>: <span>"llm"</span>, <span>"worker"</span>: <span>"writer"</span>,
         <span>"input_feed"</span>: <span>"研究+分析结果"</span>, <span>"output"</span>: <span>"报告草稿"</span>},
        {<span>"span_id"</span>: <span>"reviewer-1"</span>, <span>"type"</span>: <span>"llm"</span>, <span>"worker"</span>: <span>"reviewer"</span>,
         <span>"output"</span>: <span>"不通过: 市场规模数据无出处"</span>},
    ],
}

图 3：多智能体分层 trace

关键记录点：每个 worker span 要记录投喂了什么（input_feed）——串扰排查时，看"这个 worker 到底看到了什么"。

28.3.3 常见多智能体故障定位（第 21 章分类）

28.4 上线与迭代

28.4.1 上线检查（第 22 章清单应用）

28.4.2 协作效率度量（第 20 章四维指标扩展）

多智能体要多看两个指标：

图 4：协作效率指标

"多智能体不是越多越好"：如果协作效率低（大量空转调用），说明任务不需要这么多 Agent——回退到单 Agent 或提示链（第 16 章立场：先证明单 Agent 不行再上）。

28.4.3 迭代节奏

上线 → 每周抽 10 份报告人工评质量（第20章抽评）
  → 收集失败案例 → 更新角色提示词/任务图
  → 评测回归 → 灰度 → 上线

🛠 解决方案：多智能体上下文隔离方案 + 协作效率度量

常见问题

1. "worker 互相污染上下文"：广播投喂。对策：按需投喂 + 结构化传递（28.2.3）。
2. "复核反复打回，任务卡死"：无打回上限。对策：打回上限 2 次 + 超限转人工（28.3.3）。
3. "写手瞎编数据"：数据没传给写手或传了原文。对策：写手只吃结构化结果 + 复核查出处（28.2.2）。
4. "某 worker 越权调工具"：权限没按角色配。对策：RBAC + 工具白名单（第 22 章）。
5. "多智能体比单 Agent 还慢还贵"：任务不需要多 Agent。对策：评估协作效率，回退到单 Agent/提示链（28.4.2）。

解决方案速查表

实战提示

1. 默认主从 + 流水线混合：主管拆解、worker 干活、复核兜底——最稳形态（第 16 章）。
2. 上下文隔离是第一原则：input_feed 记录投喂内容，串扰一眼可见。
3. 复核要"独立裁判"：复核员不写内容、不给写手工具，保持中立。
4. 打回要有上限：2 次打回后转人工，别让系统自己绕圈。
5. 用协作效率说话：多智能体跑得值不值，看指标，不看"看起来高级"。