Agent 之多智能体：从一人包办到流水线协作

上篇你给 Agent 装上了 Plan-and-Execute——它能先规划再执行，复杂任务有了章法。但当任务本身需要「研究 + 写作 + 审稿」多种能力时，让同一个 Agent 一人包办，质量往往上不去。这篇讲多智能体——把工作拆给专职角色，用流水线协作出稿。

小明想写一篇科普文

小明用上篇的 /plan 模式，给 Agent 下了个任务：

你：/plan 帮我写一篇关于人工智能的科普文章

计划出来了，看起来很靠谱：

[计划]
1. 搜集 AI 核心概念与常见误区
2. 按大纲写出完整初稿
3. 检查逻辑与可读性，输出终稿

执行完，文章有了——但小明越看越别扭：开头像研究报告，中间像教科书，结尾又突然变得很「鸡汤」。同一段里既想讲清原理，又想煽情，语气来回跳。

他拿着初稿去找老张：”明明计划分了三步，为什么写出来还是像一个人硬扛所有角色？”

老张笑了：”因为你只有一个 Agent。Planner 把步骤拆开了，但执行时还是同一个脑子在干三件事——研究、写作、编辑。计划拆的是步骤，不是角色。”

“那怎么办？”

“需要多智能体：让不同 Agent 各干一件事，中间用协议传结果。”

什么是多智能体

老张在白板上画了一条流水线：

用户输入
   ↓
[Orchestrator]  调度员：只拆解、传递、汇总，不写正文
   ↓
[Researcher] → research_output   （研究）
   ↓
[Writer]   → draft       （写作）
   ↓
[Editor]   → 终稿        （审稿）

“你可以把多智能体想成一家小报社：

• 调度员（Orchestrator）——接稿、派活、盯进度，自己不动笔
• 研究员（Researcher）——只负责把主题拆成要点和写作建议
• 写手（Writer）——只根据研究报告写初稿
• 编辑（Editor）——只盯住最关键的一条问题，改出终稿

Plan-and-Execute 是一个人对自己列清单再干活；多智能体是几个人按工种协作。前者解决’步骤乱’，后者解决’角色糊’。”

小明点点头：”所以多智能体的关键不是’多几个模型’，而是职责边界 + 交接协议。”

“对。代码里这两件事分别落在 agents/ 和 protocol.py。”

代码实现：先定协议，再组流水线

老张打开写作助手writing_assistant项目目录：

writing_assistant/
├── main.py
├── protocol.py      ← Agent 间通信协议
├── llm.py
└── agents/
  ├── orchestrator.py  ← 调度员
  ├── researcher.py
  ├── writer.py
  └── editor.py

1. Task / TaskResult：Agent 之间只认这两张单

@dataclass
class Task:
  """Orchestrator 下发给 Worker Agent 的任务单。"""
  task_id: str
  agent_name: str
  instruction: str
  context: dict[str, Any] = field(default_factory=dict)
  created_at: float = field(default_factory=time.time)

  @classmethod
  def new(cls, agent_name: str, instruction: str, **context: Any) -> "Task":
    return cls(
      task_id=str(uuid.uuid4())[:8],
      agent_name=agent_name,
      instruction=instruction,
      context=dict(context),
    )


@dataclass
class TaskResult:
  """Worker 完成后回传的结果单。"""
  task_id: str
  agent_name: str
  status: TaskStatus
  output: str
  duration_ms: int = 0
  metadata: dict[str, Any] = field(default_factory=dict)

“看清楚：Worker 不互相调用。研究员不认识写手，写手也不认识编辑。大家只认 Orchestrator 发来的 Task，办完交回 TaskResult。

context 才是上下文传递口——研究报告塞进 research_output，初稿塞进 draft。增删一个 Worker，只要 Orchestrator 改派活顺序，其他模块不用改。这叫松耦合。”

小明：”那 Task.new(..., research_output=...) 里的关键字参数呢？”

“工厂方法把 **context 打进 context 字典。调度时写得短，Worker 取时用 task.context.get("research_output")。”

2. 三个 Worker：同一套接口，三套人设

研究员温度偏低，偏稳：

class ResearcherAgent:
  def run(self, task: Task) -> TaskResult:
    topic = task.instruction
    prompt = f"写作主题：{topic}\n\n请帮我分析这个主题，整理关键要点和写作建议。"
    messages = [system(_SYSTEM_PROMPT), user(prompt)]
    try:
      output = chat(messages, temperature=0.3)
      return TaskResult.success(task, output=output)
    except Exception as e:
      return TaskResult.failure(task, reason=str(e))

写手温度偏高，敢发挥；并且强制吃研究报告：

class WriterAgent:
  def run(self, task: Task) -> TaskResult:
    topic = task.instruction
    research = task.context.get("research_output", "（无研究资料）")
    prompt = f"""\
写作主题：{topic}

以下是研究员整理的参考资料，请基于这些内容写出完整文章：

{research}

直接输出文章正文。
"""
    messages = [system(_SYSTEM_PROMPT), user(prompt)]
    try:
      draft = chat(messages, temperature=0.8)
      return TaskResult.success(task, output=draft)
    except Exception as e:
      return TaskResult.failure(task, reason=str(e))

编辑只改最关键的一条，再吐终稿——避免改成「十条意见、一篇没改」：

_SYSTEM_PROMPT = """\
你是一位严格但高效的文章编辑。

你的工作分两步：
1. 给出【一条】最重要的改进建议（一句话，直接说问题在哪、怎么改）
2. 紧接着输出改进后的完整文章全文
...
"""

def _extract_final(editor_response: str) -> str:
  marker = "【终稿】"
  if marker in editor_response:
    return editor_response[editor_response.index(marker) + len(marker):].strip()
  return editor_response.strip()

“三个 Worker 的 run(task) -> TaskResult 签名完全一样。差别只在 System Prompt、温度，以及从 context 里取什么。接口统一，才能被 Orchestrator 当成可替换零件。”

3. Orchestrator：不写字，只调度 + 重试

class OrchestratorAgent:
  MAX_RETRIES = 2

  def run(self, user_request: str) -> str:
    print(f"\n[Orchestrator] 收到任务：{user_request}")

    # Step 1: 研究
    research_result = self._run_with_retry(
      agent=self._researcher,
      agent_name="researcher",
      instruction=user_request,
    )
    if research_result.status == TaskStatus.FAILED:
      return f"研究阶段失败：{research_result.output}"

    # Step 2: 写作（携带研究报告）
    write_result = self._run_with_retry(
      agent=self._writer,
      agent_name="writer",
      instruction=user_request,
      research_output=research_result.output,
    )
    if write_result.status == TaskStatus.FAILED:
      return f"写作阶段失败：{write_result.output}"

    # Step 3: 编辑（携带初稿）
    edit_result = self._run_with_retry(
      agent=self._editor,
      agent_name="editor",
      instruction=user_request,
      draft=write_result.output,
    )
    if edit_result.status == TaskStatus.FAILED:
      print(f"[Orchestrator]   Editor 失败，保留初稿。")
      return write_result.output

    return edit_result.output

“三处设计值得记：

1. 上游输出变下游输入——research_output、draft 显式传递，不靠’大家共享一份巨大 messages’
2. 编辑可降级——审稿挂了，至少交初稿，不把整单打成零
3. 每步可重试——_run_with_retry 失败睡 1 秒再试，最多额外 2 次”

def _run_with_retry(self, agent, agent_name, instruction, **context_kwargs) -> TaskResult:
  last_result: TaskResult | None = None
  for attempt in range(1, self.MAX_RETRIES + 2):  # 1 次正常 + 2 次重试
    task = Task.new(agent_name=agent_name, instruction=instruction, **context_kwargs)
    result = agent.run(task)
    if result.status == TaskStatus.DONE:
      return result
    last_result = result
    if attempt <= self.MAX_RETRIES:
      print(f"  [{agent_name}] 第{attempt}次失败：{result.output}，1s 后重试...")
      time.sleep(1)
    else:
      print(f"  [{agent_name}] 已达最大重试次数，放弃。")
  return last_result

入口很薄——用户一句话，全交给调度员：

orchestrator = OrchestratorAgent()
article = orchestrator.run(request)
print(article)

完整运行示例

“看一次真实流水线：”

python main.py "帮我写一篇关于人工智能的科普文章"

[Orchestrator] 收到任务：帮我写一篇关于人工智能的科普文章
[Orchestrator] → 交给 Researcher...
[Orchestrator]   Researcher 完成（18240ms）
[Orchestrator] → 交给 Writer...
[Orchestrator]   Writer 完成（25610ms）
[Orchestrator] → 交给 Editor...
[Orchestrator]   Editor 完成（19880ms）
[Orchestrator] 全部流程完成。

============================================================
**剥开AI的魔法外衣：普通人如何驾驭这场技术狂飙？**

想象这样一个场景：你坐在电脑前，面对一份毫无头绪的季度汇报...
（终稿：开头抓场景 → 讲清概率本质 → 划清能力边界 → 给普通人行动建议）
============================================================

老张说：”注意日志里的箭头——你看到的是调度轨迹，不是某个万能 Agent 的内心独白。研究员先把角度和要点钉死，写手再发挥，编辑最后只拧最关键的那一颗螺丝。所以终稿不会一会儿论文腔、一会儿口号腔。”

小明：”如果 Editor 挂了呢？”

“代码会打印 Editor 失败，保留初稿，然后把 Writer 的稿子直接返回。生产里这叫有损降级：宁可少一道工序，也不整单空白。”

单 Agent vs 多智能体：别混成一件事

常见坑

坑 1：Worker 互相直接调用

症状：Writer 里 import ResearcherAgent，自己去跑研究。

原因：耦合回单 Agent，协议形同虚设，重试和降级都没法统一做。

修法：Worker 只实现 run(task)；谁先谁后只写在 Orchestrator。

坑 2：上下文靠「口头提醒」而不是 context

症状：只把主题传给 Writer，研究报告丢在日志里。

原因：模型看不到上游产物，等于重新开写。

修法：像课件一样显式传 research_output=research_result.output。

坑 3：编辑提十条建议却不改稿

症状：输出全是意见列表，没有可用正文。

原因：编辑 Prompt 太「顾问」、太发散。

修法：强制「一条建议 +【终稿】全文」，再用 _extract_final 抠正文。

总结

老张说：”多智能体不是把模型多开几个窗口，而是把组织方式写进代码。”

“三个核心理解：

1. Orchestrator + Worker —— 调度员不写内容，Worker 不互相认识；边界清楚，才好扩展
2. Task / TaskResult 协议 —— 任务单带 instruction + context，结果单带 status + output；流水线靠协议传话，不靠共享大脑
3. 重试与降级 —— 研究/写作失败可重试；编辑失败可回退初稿，保证链路韧性”

系列走到这里：

Function Calling → 会用工具
    ↓
工具注册表     → 方便用很多工具
    ↓
MCP        → 用全世界的工具
    ↓
记忆       → 记住对话历史
    ↓
ReAct 循环     → 自己决定用几次、用什么
    ↓
Plan-and-Execute → 先规划后执行
    ↓
多智能体     → 多角色流水线协作
    ↓
完整 Agent     → 能力合一 + 可观测上线

小明说：”我明白了。Plan 解决’先想清楚再做’；多智能体解决’别让一个人同时当记者、作家和主编’。”

“对。下一篇可以把记忆、ReAct、Plan、多 Agent 收进一个完整项目——再补上评估与日志，才算往生产靠近。”

Agent 的进化路线：会调用工具 → 会自主决策 → 会规划执行 → 会分工协作。从「能干的个体」走到「可编排的团队」。