大模型实战指南（13）——多 Agent 协作实战：从单兵到军团

大模型实战指南（13）——多 Agent 协作实战：从单兵到军团

上一篇你已经有了一个能离线调用工具的本地 AI 助手——它能查文件、读文件、写文件。但你有没有发现一个问题：它一次只能干一件事。你让它“收集本周行业新闻→写 500 字摘要→翻译成英文→排版成 Markdown 文件”，它得一步步来，每一步都要你手动喂上一步的结果，中途出错整个重来。

这就像你雇了一个助手，什么都会干，但只能串行干活，不能并行，也不会自己分配任务。一个人的效率天花板就在那了。

2026 年 8 月，AI 行业正式进入“企业智能体规模化落地元年”——这是麦肯锡在 7 月发布的全球 AI 敏捷度调查中的判断。但数据同时揭示了一个残酷现实：Gartner 预测到 2027 年底，超过 40% 的 AI Agent 项目将被取消；德勤调查显示 38% 企业试点了 Agent，但只有 11% 真正投产。卡点不在模型能力，在于系统集成、数据治理和协作架构——“能不能稳定干活”才是真正的战场。

这一篇，我们从单 Agent 走向多 Agent，做四件事：

不扯虚的，跟着做，做完你的电脑上就有一个多 Agent 协作系统。

一、为什么单 Agent 不够用

1.1 一个真实场景

假设你是技术内容创作者，每周要出一篇行业分析报告。工作流程是这样的：

用第 12 篇的单 Agent 助手，你得这样干：先让它搜索，手动把搜索结果喂给它让它总结，再把总结喂给它让它写摘要，再把摘要喂给它让它翻译……每一步都是你手动驱动。

问题在哪？单 Agent 没有分工意识，没有流水线能力，不能自己管理多步任务的状态。它就像一个全能但单线程的员工——什么都会，但一次只能做一件事，不会自己规划执行顺序。

1.2 多 Agent 怎么解决

多 Agent 系统的核心思想是分工。把上面 5 个步骤拆给 5 个专门的 Agent：

• 规划官：接收主题，制定报告大纲
• 研究员：根据大纲搜索和整理素材
• 写手：基于素材写摘要
• 翻译官：把摘要翻译成英文
• 质检员：检查质量，不合格就退回重写

每个 Agent 只做自己擅长的事，做完把结果传给下一个。质检员不合格就退回写手——这就是“流水线+反馈循环”。

1.3 数据说话：什么时候该用多 Agent

先别急着上多 Agent——它复杂度高、成本高、调试难。先做个自查：

4 条自查清单：

勾中 3 条以上才需要多 Agent。 如果你的任务只是“帮我总结这段话”，单 Agent 就够了。

几个关键数据帮你建立预期（均为 2026 年公开报告）：

最后一条特别值得注意——95% 的试点没有带来可衡量回报。这意味着大多数 AI Agent 项目停在了 Demo 阶段，没有真正进入生产。本篇的目标就是帮你跨过这道鸿沟。

二、多 Agent 协作的三种模式

进入实操之前，先搞懂概念。多 Agent 的协作模式有三种主流范式，理解了这三种，后面所有框架你都能快速上手。

2.1 模式一：图/状态机编排（LangGraph）

把 Agent 工作流建模成一张有状态的有向图。每个 Agent 是图中的一个节点，Agent 之间的协作关系是图中的边。

核心概念就三个：

• State（状态）：整个图运行过程中的共享数据快照，所有节点都能读写
• Node（节点）：执行具体逻辑的地方——一个 Agent 就是一个节点
• Edge（边）：决定节点之间的执行顺序，支持条件分支（质检不通过就回到写手节点）

这就像一条流水线——零件从起点出发，经过每个工位（Agent）加工，质检不合格就退回上游工位，合格就流向下游。你提前把整个流程画成图，运行时按图执行。

LangGraph 是这种模式的代表框架，核心 API 就四个：StateGraph（创建图）、add_node（添加节点）、add_edge（添加边）、add_conditional_edges（添加条件边）。

2.2 模式二：消息传递（AutoGen）

Agent 之间通过对话消息协作。Agent A 说“这是搜索结果，请总结”，Agent B 收到后回复“这是总结，请审阅”，Agent C 收到后说“审阅通过”。

这更像一个群聊——每个 Agent 是群里的一个人，通过发消息来协作。谁先说话、谁后说话、说什么内容，由对话流程控制。

AutoGen 是这种模式的代表。它的特点是灵活但不可预测——对话可能走向你意想不到的方向。好处是探索性强，坏处是你很难保证每次执行路径都一样。

2.3 模式三：任务委派（CrewAI）

你定义一个团队——每个 Agent 有角色（Role）、目标（Goal）、背景故事（Backstory）。然后定义任务（Task），指定由哪个 Agent 负责。框架自动按顺序执行任务，一个 Agent 做完把结果交给下一个。

这更像一个项目组——你是项目经理，给每个人分配角色和任务，每个人做完自己的部分交接给下一个人。

CrewAI 是这种模式的代表，对小白最友好。你只需要声明“这个人是研究员”“那个任务是写摘要”，框架自动组装流程。

2.4 三种模式怎么选

本篇实操选用 LangGraph——原因有三个：① 它的 StateGraph 模式天然适合“流水线+反馈循环”结构；② 条件边原生支持“质检不通过退回重写”这种逻辑；③ 从第 12 篇的纯 Python 代码过渡到 LangGraph 是最自然的——你还是在写 Python 函数，只是用 LangGraph 把它们串成图。

三、环境准备：你需要装什么

3.1 基础环境

延续第 12 篇的环境——你需要：

• Python 3.10+
• Ollama 已安装并能跑 qwen3:8b（建议 8B 以上，多 Agent 场景对模型理解力要求更高）
• pip 安装 LangGraph 和 LangChain 的 Ollama 集成：

pip install langgraph langchain-ollama

为什么建议 8B 而不是 4B？ 多 Agent 场景中，每个 Agent 都需要理解自己的角色、任务和上下文。4B 模型在角色扮演场景下容易“串角色”——研究员突然开始写摘要，质检员忘了自己在质检。8B 以上的模型在角色一致性上明显更可靠。

3.2 验证安装

装完后跑一下这个最小示例，确认 LangGraph 能正常工作：

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class State(TypedDict):
counter: int

def increment(state):
return {"counter": state["counter"] + 1}

graph = StateGraph(State)
graph.add_node("inc", increment)
graph.add_edge(START, "inc")
graph.add_edge("inc", END)
app = graph.compile()

result = app.invoke({"counter": 0})
print(result)

如果输出 {'counter': 1}，说明 LangGraph 装好了。这段代码虽然简单，但包含了 LangGraph 的全部核心概念：定义状态 → 写节点函数 → 画图 → 编译 → 运行。

3.3 连接 Ollama

接下来确认 LangChain 能连上你本地的 Ollama：

from langchain_ollama import ChatOllama

llm = ChatOllama(model="qwen3:8b", base_url="http://localhost:11434")
response = llm.invoke("你好，用一句话介绍你自己")
print(response.content)

如果模型回了一句话，环境就准备好了。接下来进入正式开发。

四、核心实操：用 LangGraph 写“内容生产工厂”

这是本篇的重头戏。我们要从零搭建一个多 Agent 系统——“内容生产工厂”。给它一个主题，4 个 Agent 自动接力完成：规划大纲 → 研究素材 → 撰写初稿 → 质检审阅。质检不通过自动退回重写，最多重试 3 次。

4.1 系统设计：4 个 Agent 的分工

先想清楚架构，再写代码。我们的“内容生产工厂”有 4 个工位：

执行流程是一条流水线，但质检员可以通过条件边把稿件退回给写手：

START → 规划官 → 研究员 → 写手 → 质检员 →(通过)→ END
↑ ↓(不通过)
└────┘ (退回重写，最多3次)

4.2 定义共享状态：Agent 之间怎么传数据

多 Agent 系统的第一个关键问题：Agent 之间怎么共享数据？LangGraph 的答案是用一个全局的 State 对象——每个 Agent 函数接收当前 State，返回需要更新的字段，LangGraph 自动合并。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langchain_ollama import ChatOllama
import operator

class FactoryState(TypedDict):
# 输入
topic: str
# 各 Agent 的产出
outline: str
research: str
draft: str
review_comment: str
# 质检控制
review_passed: bool
revision_count: int
# 审计日志（用 operator.add 实现追加）
log: Annotated[list[str], operator.add]

这里有个关键设计：log 字段用 Annotated[list[str], operator.add] 标注。意思是——每次 Agent 返回新的 log 列表时，LangGraph 不会覆盖旧值，而是用 operator.add（也就是列表拼接）把新日志追加到旧日志后面。这就是消息累加模式，在多 Agent 系统中非常常用。

其他字段（如 outline、draft）没有特殊标注，每次返回就直接覆盖——因为这些字段只需要最新值，不需要历史记录。

4.3 实现 4 个 Agent 节点

现在写 4 个 Agent 的核心逻辑。每个 Agent 就是一个 Python 函数：接收 State，调用 LLM，返回更新后的字段。

先初始化模型：

llm = ChatOllama(
model="qwen3:8b",
base_url="http://localhost:11434",
temperature=0.7,
)

temperature=0.7 是个折中——规划官和研究员可以用 0.7 保持一定创造性，质检员应该用更低的 temperature 保证判断一致性。后面我们会在优化章节中做差异化配置。

Agent 1：规划官——把主题拆成大纲：

def planner(state: FactoryState) –> dict:
topic = state["topic"]
prompt = f"""你是一位技术内容规划官。给定主题，制定一份报告大纲。

主题：{topic}

要求：
1. 拆出 3-5 个核心子方向
2. 每个子方向用一句话说明要调研什么
3. 输出格式为编号列表

直接输出大纲，不要多余解释。"""

response = llm.invoke(prompt)
outline = response.content

return {
"outline": outline,
"log": [f"[规划官] 已生成大纲，共 {outline.count(chr(10)) + 1} 个子方向"],
}

注意返回值——我们只返回需要更新的字段，不需要把整个 State 都返回。LangGraph 会自动把 outline 和 log 合并到 State 中。

Agent 2：研究员——根据大纲整理素材：

def researcher(state: FactoryState) –> dict:
outline = state["outline"]
prompt = f"""你是一位技术研究员。根据以下大纲，为每个子方向整理关键信息。

大纲：
{outline}

要求：
1. 针对每个子方向，列出 3-5 条核心要点
2. 标注信息的来源类型（官方文档/行业报告/社区讨论）
3. 如果某个子方向信息不足，标注"需补充"

直接输出研究笔记。"""

response = llm.invoke(prompt)
research = response.content

return {
"research": research,
"log": [f"[研究员] 已完成素材整理"],
}

这里我们模拟研究员“从已有知识中整理素材”。在实际项目中，你可以给研究员装上搜索工具——让它真正去网上搜索。我们会在 4.7 节展示怎么给 Agent 加工具。

Agent 3：写手——把研究笔记写成报告：

def writer(state: FactoryState) –> dict:
topic = state["topic"]
outline = state["outline"]
research = state["research"]
revision_count = state.get("revision_count", 0)

if revision_count > 0:
review_comment = state.get("review_comment", "")
prompt = f"""你是一位技术写手。你的初稿未通过质检，请根据反馈修改。

主题：{topic}
大纲：{outline}
研究笔记：{research}
质检反馈：{review_comment}

这是第 {revision_count + 1} 版。请根据反馈修改初稿。"""
else:
prompt = f"""你是一位技术写手。根据以下信息撰写一篇技术报告。

主题：{topic}
大纲：{outline}
研究笔记：{research}

要求：
1. 800-1200 字
2. 结构清晰，分节标题
3. 语言准确，避免空话

直接输出报告正文。"""

response = llm.invoke(prompt)
draft = response.content

return {
"draft": draft,
"log": [f"[写手] 已完成第 {revision_count + 1} 版初稿"],
}

注意写手函数里的分支逻辑：如果是第一次写（revision_count == 0），正常生成初稿；如果是退回重写（revision_count > 0），会把质检员的反馈也喂进去，让模型针对性修改。这就是反馈循环的核心——让 Agent 知道上次哪里没做好。

Agent 4：质检员——审核初稿，决定通过或退回：

def reviewer(state: FactoryState) –> dict:
draft = state["draft"]
outline = state["outline"]
revision_count = state.get("revision_count", 0)

prompt = f"""你是一位严格的质检编辑。审核以下报告初稿。

报告大纲：
{outline}

报告初稿：
{draft}

审核标准：
1. 完整性：是否覆盖了大纲中的所有子方向？
2. 准确性：有无明显的事实错误？
3. 可读性：结构是否清晰，语言是否流畅？

先给出审核意见，然后在最后一行输出判定：
PASS 或 FAIL"""

response = llm.invoke(prompt)
result = response.content

passed = "PASS" in result.split("\\n")[–1].upper()

return {
"review_comment": result,
"review_passed": passed,
"revision_count": revision_count + (0 if passed else 1),
"log": [f"[质检员] 第 {revision_count + 1} 版审核结果：{'通过' if passed else '未通过'}"],
}

质检员是整个系统中最关键的 Agent——它决定了稿件是流向终点还是退回重写。review_passed 布尔值会被条件边读取，决定下一步走哪个节点。

4.4 组装图：把 4 个 Agent 串成流水线

4 个 Agent 函数写好了，现在用 LangGraph 的图 API 把它们串起来：

def should_revise(state: FactoryState) –> str:
"""条件边路由函数：决定质检后下一步去哪"""
if state["review_passed"]:
return "end"
if state["revision_count"] >= 3:
return "end"
return "revise"

graph = StateGraph(FactoryState)

# 添加 4 个节点
graph.add_node("planner", planner)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)

# 添加固定边：串行流水线
graph.add_edge(START, "planner")
graph.add_edge("planner", "researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")

# 添加条件边：质检后的分支
graph.add_conditional_edges(
"reviewer",
should_revise,
{
"end": END,
"revise": "writer",
},
)

# 编译
app = graph.compile()

这段代码的精髓在最后两部分：

should_revise 函数是路由逻辑的核心：

• 质检通过 → "end"，流程结束
• 质检不通过但已重试 3 次 → "end"，放弃重试，输出当前版本
• 质检不通过且重试次数不足 → "revise"，回到写手重写

这个“最多重试 3 次”的设计非常重要——没有它，如果质检员永远不满意，系统就会陷入死循环（写手和质检员互相踢皮球到天荒地老）。这是后面踩坑章节要讲的第一个坑。

4.5 运行：一键自动化生产报告

一切就绪，跑一下：

import pprint

initial_state = {
"topic": "2026 年端侧 AI 大模型的落地挑战与机遇",
"revision_count": 0,
"log": [],
}

result = app.invoke(initial_state, {"recursion_limit": 20})

print("=" * 60)
print("最终报告")
print("=" * 60)
print(result["draft"])
print("\\n" + "=" * 60)
print("审计日志")
print("=" * 60)
for entry in result["log"]:
print(entry)
print(f"\\n总修订次数：{result['revision_count']}")
print(f"质检结果：{'通过' if result['review_passed'] else '未通过（已达上限）'}")

如果一切正常，你会看到类似这样的输出：

============================================================
最终报告
============================================================
# 2026 年端侧 AI 大模型的落地挑战与机遇

## 一、端侧算力瓶颈：NPU 能力参差不齐
…

============================================================
审计日志
============================================================
[规划官] 已生成大纲，共 4 个子方向
[研究员] 已完成素材整理
[写手] 已完成第 1 版初稿
[质检员] 第 1 版审核结果：通过

总修订次数：0
质检结果：通过

如果质检没通过，日志会多出几行修订记录：

[规划官] 已生成大纲，共 4 个子方向
[研究员] 已完成素材整理
[写手] 已完成第 1 版初稿
[质检员] 第 1 版审核结果：未通过
[写手] 已完成第 2 版初稿
[质检员] 第 2 版审核结果：通过

总修订次数：1
质检结果：通过

看到没有？写手和质检员的交互完全是自动的——你只给了一个主题，系统自己规划、研究、写稿、质检、退回重写、再质检，全程不需要你介入。这就是多 Agent 协作的力量。

4.6 给研究员装上搜索工具

上面的研究员只是从模型内置知识中整理素材。在实际项目中，你需要让它真正去搜索。这里展示如何给 Agent 节点加工具调用能力——延续第 12 篇的 Function Calling 思路。

import json

def search_web(query: str) –> str:
"""模拟搜索工具（实际项目中替换为真实搜索 API）"""
return f"搜索结果：关于「{query}」的最新信息——这是一个模拟结果，实际使用时替换为搜索 API。"

def researcher_with_tools(state: FactoryState) –> dict:
outline = state["outline"]

# 第一步：让 LLM 决定要搜索什么
search_prompt = f"""你是一位技术研究员。根据以下大纲，列出需要搜索的关键词。

大纲：
{outline}

只输出搜索关键词列表，每行一个。"""

search_response = llm.invoke(search_prompt)
search_queries = [q.strip() for q in search_response.content.strip().split("\\n") if q.strip()]

# 第二步：执行搜索
search_results = []
for query in search_queries[:5]:
result = search_web(query)
search_results.append(f"【{query}】\\n{result}")

# 第三步：让 LLM 基于搜索结果整理研究笔记
research_prompt = f"""你是一位技术研究员。基于以下搜索结果整理研究笔记。

大纲：
{outline}

搜索结果：
{chr(10).join(search_results)}

针对每个子方向整理 3-5 条核心要点。"""

research_response = llm.invoke(research_prompt)

return {
"research": research_response.content,
"log": [f"[研究员] 完成搜索，共查询 {len(search_queries[:5])} 个关键词"],
}

这个实现用的是 “Plan → Search → Synthesize”三步模式——先让 LLM 规划搜索什么，再执行搜索，最后让 LLM 综合搜索结果整理笔记。这比直接把搜索工具丢给 LLM 让它自己决定怎么用要可控得多。

如果要替换成真实搜索，只需要修改 search_web 函数——用 Tavily API、SerpAPI 或者本地 SearXNG 都行。Agent 的逻辑完全不用改。

4.7 流式输出：实时看到每个 Agent 在干什么

生产环境中你不能干等 2 分钟然后突然蹦出一个结果——你需要实时看到进度。LangGraph 支持流式执行，可以逐节点查看状态变化：

def run_with_streaming(topic: str):
initial_state = {
"topic": topic,
"revision_count": 0,
"log": [],
}

for output in app.stream(initial_state, {"recursion_limit": 20}):
for node_name, node_output in output.items():
log_entry = node_output.get("log", [""])
print(f"[{node_name}] {log_entry[–1] if log_entry else '完成'}")

print("\\n" + "=" * 60)
print("最终报告：")
print("=" * 60)
print(result["draft"])

run_with_streaming("LangGraph 多 Agent 系统的设计模式")

app.stream() 会在每个节点执行完毕后立即 yield 该节点的输出，你可以在回调中实时打印日志。终端输出效果：

[planner] [规划官] 已生成大纲，共 4 个子方向
[researcher] [研究员] 已完成素材整理
[writer] [写手] 已完成第 1 版初稿
[reviewer] [质检员] 第 1 版审核结果：通过

============================================================
最终报告：
============================================================
# LangGraph 多 Agent 系统的设计模式
…

如果你想进一步看到每个 Agent 内部的逐 token 输出（而不仅仅是节点完成后的日志），可以把 llm.invoke() 换成 llm.stream()，然后在每个 Agent 函数内部遍历 token 块实时打印。不过这会增加代码复杂度，建议先跑通基础版再考虑。

4.8 可视化你的 Agent 图

LangGraph 支持把你的图导出为 Mermaid 格式，方便文档化和团队沟通：

print(app.get_graph().draw_mermaid())

输出类似：

#mermaid-svg-j0Du5lT3kdCEJh8D{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-j0Du5lT3kdCEJh8D .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-j0Du5lT3kdCEJh8D .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-j0Du5lT3kdCEJh8D .error-icon{fill:#552222;}#mermaid-svg-j0Du5lT3kdCEJh8D .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-j0Du5lT3kdCEJh8D .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-j0Du5lT3kdCEJh8D .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-j0Du5lT3kdCEJh8D .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-j0Du5lT3kdCEJh8D .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-j0Du5lT3kdCEJh8D .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-j0Du5lT3kdCEJh8D .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-j0Du5lT3kdCEJh8D .marker{fill:#333333;stroke:#333333;}#mermaid-svg-j0Du5lT3kdCEJh8D .marker.cross{stroke:#333333;}#mermaid-svg-j0Du5lT3kdCEJh8D svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-j0Du5lT3kdCEJh8D p{margin:0;}#mermaid-svg-j0Du5lT3kdCEJh8D .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-j0Du5lT3kdCEJh8D .cluster-label text{fill:#333;}#mermaid-svg-j0Du5lT3kdCEJh8D .cluster-label span{color:#333;}#mermaid-svg-j0Du5lT3kdCEJh8D .cluster-label span p{background-color:transparent;}#mermaid-svg-j0Du5lT3kdCEJh8D .label text,#mermaid-svg-j0Du5lT3kdCEJh8D span{fill:#333;color:#333;}#mermaid-svg-j0Du5lT3kdCEJh8D .node rect,#mermaid-svg-j0Du5lT3kdCEJh8D .node circle,#mermaid-svg-j0Du5lT3kdCEJh8D .node ellipse,#mermaid-svg-j0Du5lT3kdCEJh8D .node polygon,#mermaid-svg-j0Du5lT3kdCEJh8D .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-j0Du5lT3kdCEJh8D .rough-node .label text,#mermaid-svg-j0Du5lT3kdCEJh8D .node .label text,#mermaid-svg-j0Du5lT3kdCEJh8D .image-shape .label,#mermaid-svg-j0Du5lT3kdCEJh8D .icon-shape .label{text-anchor:middle;}#mermaid-svg-j0Du5lT3kdCEJh8D .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-j0Du5lT3kdCEJh8D .rough-node .label,#mermaid-svg-j0Du5lT3kdCEJh8D .node .label,#mermaid-svg-j0Du5lT3kdCEJh8D .image-shape .label,#mermaid-svg-j0Du5lT3kdCEJh8D .icon-shape .label{text-align:center;}#mermaid-svg-j0Du5lT3kdCEJh8D .node.clickable{cursor:pointer;}#mermaid-svg-j0Du5lT3kdCEJh8D .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-j0Du5lT3kdCEJh8D .arrowheadPath{fill:#333333;}#mermaid-svg-j0Du5lT3kdCEJh8D .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-j0Du5lT3kdCEJh8D .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-j0Du5lT3kdCEJh8D .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-j0Du5lT3kdCEJh8D .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-j0Du5lT3kdCEJh8D .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-j0Du5lT3kdCEJh8D .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-j0Du5lT3kdCEJh8D .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-j0Du5lT3kdCEJh8D .cluster text{fill:#333;}#mermaid-svg-j0Du5lT3kdCEJh8D .cluster span{color:#333;}#mermaid-svg-j0Du5lT3kdCEJh8D div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-j0Du5lT3kdCEJh8D .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-j0Du5lT3kdCEJh8D rect.text{fill:none;stroke-width:0;}#mermaid-svg-j0Du5lT3kdCEJh8D .icon-shape,#mermaid-svg-j0Du5lT3kdCEJh8D .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-j0Du5lT3kdCEJh8D .icon-shape p,#mermaid-svg-j0Du5lT3kdCEJh8D .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-j0Du5lT3kdCEJh8D .icon-shape .label rect,#mermaid-svg-j0Du5lT3kdCEJh8D .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-j0Du5lT3kdCEJh8D .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-j0Du5lT3kdCEJh8D .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-j0Du5lT3kdCEJh8D :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

end

revise

__start__

planner

researcher

writer

reviewer

__end__

你可以把这段 Mermaid 贴到任何支持 Mermaid 的编辑器（VS Code、GitHub、Notion）中渲染成流程图。在项目初期设计阶段，先用 Mermaid 画好流程图，团队确认后再写代码——这是多 Agent 项目的最佳实践。

为什么不用 LangGraph Studio？ LangGraph Studio 是官方的可视化调试工具，能交互式地查看图的执行过程、状态变化和节点输出。但它需要 LangSmith 账号，且主要面向云端部署场景。本篇聚焦本地化方案，所以用 Mermaid + 终端日志的方式做可视化，对大多数人够用了。

五、2026 年 8 月的 Agent 协议前沿：MCP、A2A、Agent Plugins

你已经会写多 Agent 系统了。但 Agent 之间怎么通信、Agent 怎么调用外部工具、Agent 怎么被打包分发——这些问题在 2026 年有了新的标准答案。这一章带你走进 Agent 协议的前沿战场。

5.1 MCP（Model Context Protocol）：Agent 和工具的“USB 接口”

第 12 篇你已经接触过 Function Calling——Agent 通过调用函数来使用工具。但 Function Calling 有个问题：每个应用都要自己定义工具接口，换个应用就得重写一遍。

MCP 解决的就是这个问题。它是一个标准化的工具协议，让任何 Agent 都能通过统一接口连接任何工具——就像 USB 接口一样，你不用关心鼠标是什么牌子，插上就能用。

MCP 的核心架构：

Agent（MCP Client） ←→ MCP Protocol ←→ MCP Server（封装工具）

MCP Server 封装了具体工具的能力（搜索、数据库查询、文件操作等），通过 MCP 协议对外暴露。任何支持 MCP 的 Agent 都可以直接连这个 Server，不需要关心底层工具怎么实现。

2026 年的关键更新：无状态化

2026 年 MCP 候选版本引入了一个重要改进——无状态化请求。之前 MCP 会为每个客户端会话保持状态，这在云端部署时带来大麻烦：负载均衡器把请求分到不同服务器，状态不一致就出错。无状态化后，每个请求自包含所有信息，随便哪台服务器都能处理，负载均衡、无服务器部署、故障切换都变得简单了。

实战中怎么用 MCP

在实际项目中，你可以用 MCP 替代手写的工具函数。比如你的研究员 Agent 需要搜索能力，不用自己写 search_web 函数，而是连接一个搜索类 MCP Server：

from langchain_mcp_adapters.client import MultiServerMCPClient

# 连接 MCP Server（本地启动或远程均可）
client = MultiServerMCPClient({
"search": {
"url": "http://localhost:8000/sse",
"transport": "sse",
}
})

# 获取工具列表
tools = await client.get_tools()

# 把 MCP 工具绑定到 LLM
llm_with_tools = llm.bind_tools(tools)

# Agent 就能自动使用 MCP Server 提供的搜索能力了
response = llm_with_tools.invoke("搜索 LangGraph 最新版本")

你甚至可以让多个 Agent 共享同一个 MCP Server——研究员用搜索 Server 查资料，写手用翻译 Server 翻译，质检员用查重 Server 检查原创性。每个 Agent 各取所需，工具零重复开发。

5.2 A2A（Agent-to-Agent）协议：Agent 之间的“电话协议”

MCP 解决的是“Agent → 工具”的通信。但 Agent 之间怎么通信？在第 12 篇的单 Agent 系统中这不是问题，但在多 Agent 系统中——尤其当 Agent 运行在不同机器上、不同框架中时——通信就成了大问题。

A2A（Agent-to-Agent）协议就是为此而生。它定义了 Agent 之间互相发现、互相对话、互相委派任务的标准协议。

A2A 要解决的核心问题：

• 互操作性：LangGraph 写的 Agent 能不能和 CrewAI 写的 Agent 协作？A2A 说能——只要双方都实现 A2A 协议
• 远程协作：Agent A 在你的笔记本上跑，Agent B 在公司服务器上跑，它们怎么通信？A2A 定义了基于 HTTP 的远程通信标准
• 任务委派：Agent A 能不能把子任务交给 Agent B？A2A 定义了任务委派和结果回收的协议

一个具体场景：假设你的规划官跑在本地笔记本（LangGraph），研究员跑在公司的 GPU 服务器（CrewAI），写手用的云端 API（OpenAI Agents SDK）。没有 A2A，这三个 Agent 根本没法协作。有了 A2A，它们可以通过标准协议互相发现和对话。

A2A 的核心概念：

2026 年 A2A 由 Google 和 Anthropic 联合推动标准化，目前已有 50+ 企业参与。虽然协议规范还在快速迭代中，但核心设计思路已经稳定——把 Agent 当作网络服务，用标准协议通信。

5.3 Agent Plugins 1.0：Agent 的“App Store”

2026 年 8 月，Agent Plugins 1.0 规范正式落地。你可以把它理解为 Agent 的 App Store——开发者把一个“技能”打包成插件，任何兼容的 Agent 平台都能安装使用。

一个 Agent Plugin 封装什么：

Agent Plugin = 技能定义 + MCP Server + 权限声明

• 技能定义：描述这个插件能让 Agent 做什么（“搜索学术论文”“生成 SQL 查询”“分析代码质量”）
• MCP Server：插件内置一个 MCP Server，提供具体工具能力
• 权限声明：声明这个插件需要什么权限（网络访问、文件读写、数据库连接等）

为什么这很重要？之前给 Agent 加能力，你得自己写工具函数、定义接口、处理异常。有了 Agent Plugins，你只需要“安装”一个插件——就像手机装 App 一样。企业 IT 部门可以维护一个允许列表，只批准经过安全审计的插件——统一治理、统一管控。

已有的落地场景（2026 年 8 月）：

5.4 三种协议的关系：一张图看懂

把三个协议放一起看，关系就清楚了：

• MCP：Agent 和工具之间的通信协议——“我怎么用你的工具？”
• A2A：Agent 和 Agent 之间的通信协议——“我跟你怎么协作？”
• Agent Plugins：技能的打包和分发标准——“我的能力怎么给别人用？”

三者不互斥，而是互补关系。一个成熟的多 Agent 系统会同时用到三种协议：Agent 之间用 A2A 通信，每个 Agent 通过 MCP 调用工具，开发者把通用能力打包成 Agent Plugins 分发。

5.5 这跟你的本地多 Agent 系统有什么关系

你可能会想：这些协议看着很高大上，但我只是在本地跑一个多 Agent 系统，用得上吗？

答案是：现阶段你不需要实现这些协议，但你的架构设计要为它们留好扩展空间。 具体来说：

这些设计原则在第 12 篇就埋了种子（工具函数与执行逻辑分离），本篇把它们扩展到了多 Agent 层面。

六、把第 12 篇的本地助手升级成多 Agent 团队

第 12 篇你做了一个能查文件、读文件、写文件的单 Agent 助手。现在我们把它升级——不只是一个助手干活，而是一整支团队。

6.1 升级思路：从“全能选手”到“专业团队”

第 12 篇的助手长这样：

用户输入 → [单一 Agent：理解意图 + 调用工具 + 生成回复] → 输出

升级后的多 Agent 团队长这样：

用户输入 → [调度官：理解意图，决定分配给谁]
↓
┌───────┼───────┐
↓ ↓ ↓
[文件管家] [搜索员] [写作员]
↓ ↓ ↓
└───────┼───────┘
↓
[质检员：整合结果，检查质量]
↓
输出

调度官负责理解你的意图，决定把任务分配给哪个 Agent；各专业 Agent 各司其职；质检员收集结果做最终检查。

6.2 升级代码：在 LangGraph 中实现动态路由

核心改动是引入一个调度节点，用条件边动态分配任务：

from enum import Enum

class AgentType(Enum):
FILE_MANAGER = "file_manager"
SEARCH_AGENT = "search_agent"
WRITER_AGENT = "writer_agent"

class AssistantState(TypedDict):
user_input: str
agent_type: str
file_result: str
search_result: str
write_result: str
final_output: str
log: Annotated[list[str], operator.add]

def dispatcher(state: AssistantState) –> dict:
"""调度官：分析用户意图，决定路由到哪个 Agent"""
user_input = state["user_input"]

prompt = f"""分析用户的请求，判断应该交给哪个 Agent 处理。

用户请求：{user_input}

可选 Agent：
– FILE_MANAGER：文件操作（查文件、读文件、写文件、列目录）
– SEARCH_AGENT：信息搜索（搜索资料、查找信息）
– WRITER_AGENT：内容写作（写摘要、写报告、翻译）

只输出 Agent 名称，不要解释。"""

response = llm.invoke(prompt)
agent_name = response.content.strip().upper()

return {
"agent_type": agent_name,
"log": [f"[调度官] 路由到 {agent_name}"],
}

def route_to_agent(state: AssistantState) –> str:
"""条件边路由函数"""
agent_type = state["agent_type"]
if "FILE" in agent_type:
return "file_manager"
elif "SEARCH" in agent_type:
return "search_agent"
elif "WRITER" in agent_type:
return "writer_agent"
return "file_manager"

三个专业 Agent 的实现：

def file_manager(state: AssistantState) –> dict:
"""文件管家：处理文件操作"""
user_input = state["user_input"]
prompt = f"""你是文件管家。用户请求：{user_input}

你可以使用以下工具：
– list_files(path): 列出目录下的文件
– read_file(path): 读取文件内容
– write_file(path, content): 写入文件

根据用户请求，说明你要执行什么操作。"""

response = llm.invoke(prompt)
return {
"file_result": response.content,
"final_output": response.content,
"log": [f"[文件管家] 已处理文件操作请求"],
}

def search_agent(state: AssistantState) –> dict:
"""搜索员：处理信息搜索"""
user_input = state["user_input"]
prompt = f"""你是搜索研究员。用户需要搜索：{user_input}

整理相关信息并给出结果。"""

response = llm.invoke(prompt)
return {
"search_result": response.content,
"final_output": response.content,
"log": [f"[搜索员] 已完成搜索"],
}

def writer_agent(state: AssistantState) –> dict:
"""写作员：处理内容创作"""
user_input = state["user_input"]
prompt = f"""你是技术写手。用户需要：{user_input}

请根据要求创作内容。"""

response = llm.invoke(prompt)
return {
"write_result": response.content,
"final_output": response.content,
"log": [f"[写作员] 已完成写作"],
}

组装图的代码：

assistant_graph = StateGraph(AssistantState)

assistant_graph.add_node("dispatcher", dispatcher)
assistant_graph.add_node("file_manager", file_manager)
assistant_graph.add_node("search_agent", search_agent)
assistant_graph.add_node("writer_agent", writer_agent)

assistant_graph.add_edge(START, "dispatcher")

assistant_graph.add_conditional_edges(
"dispatcher",
route_to_agent,
{
"file_manager": "file_manager",
"search_agent": "search_agent",
"writer_agent": "writer_agent",
},
)

assistant_graph.add_edge("file_manager", END)
assistant_graph.add_edge("search_agent", END)
assistant_graph.add_edge("writer_agent", END)

assistant_app = assistant_graph.compile()

运行：

result = assistant_app.invoke({
"user_input": "帮我写一段关于 LangGraph 的简介",
"log": [],
})
print(result["final_output"])

调度官会识别出这是写作任务，路由到写作员，写作员生成内容后直接输出。试试换成“帮我搜索一下 Python 异步编程的资料”，调度官会自动路由到搜索员。

6.3 进阶：让 Agent 之间能互相委派

上面的设计是“调度官 → 专业 Agent → 输出”的扁平结构。但现实中的需求往往更复杂——搜索员找到资料后可能需要写手帮忙整理成文档，写手写完可能需要文件管家帮忙保存。这就需要 Agent 之间能互相委派任务。

在 LangGraph 中实现这个，关键是让每个 Agent 都能通过条件边跳转到其他 Agent：

def route_after_agent(state: AssistantState) –> str:
"""每个专业 Agent 执行完后的路由"""
user_input = state["user_input"]
# 让 LLM 判断是否需要后续处理
prompt = f"""用户原始请求：{user_input}
当前 Agent 已完成。是否需要其他 Agent 继续处理？

可选：done（完成）、search（需要搜索）、write（需要写作）、file（需要文件操作）
只输出一个选项。"""

response = llm.invoke(prompt)
choice = response.content.strip().lower()

if "search" in choice:
return "search_agent"
elif "write" in choice:
return "writer_agent"
elif "file" in choice:
return "file_manager"
return "done"

然后把专业 Agent 的出口边也改成条件边：

assistant_graph.add_conditional_edges(
"search_agent",
route_after_agent,
{"done": END, "search_agent": "search_agent", "writer_agent": "writer_agent", "file_manager": "file_manager"},
)
assistant_graph.add_conditional_edges(
"writer_agent",
route_after_agent,
{"done": END, "search_agent": "search_agent", "writer_agent": "writer_agent", "file_manager": "file_manager"},
)
assistant_graph.add_conditional_edges(
"file_manager",
route_after_agent,
{"done": END, "search_agent": "search_agent", "writer_agent": "writer_agent", "file_manager": "file_manager"},
)

这样 Agent 之间就能互相委派任务了——搜索员找到资料后可以交给写手整理，写手写完可以交给文件管家保存。系统从“调度官单向分发”进化为“Agent 自主协作网络”。

注意：互委派模式虽然灵活，但调试难度陡增。建议先用 6.2 节的扁平模式跑通，再逐步加入互委派能力。另外一定要设 recursion_limit 防止 Agent 来回踢皮球陷入死循环。

七、三个真坑，每个都付过费

多 Agent 系统的坑比单 Agent 多一个数量级——因为 bug 会发生在 Agent 之间的交互上，而不是单个 Agent 内部。下面三个坑是我踩过最疼的。

7.1 坑一：Token 黑洞——成本失控

场景：你的内容生产工厂跑得好好的，突然一天的 Token 消耗翻了 5 倍。打开日志一看，质检员连续退回 3 次，写手每次都把完整大纲 + 完整研究笔记 + 完整初稿 + 完整质检反馈一起发给 LLM——每一轮修订，输入 Token 都在膨胀。

根因：每个 Agent 在调用 LLM 时，都会把前面的上下文塞进 prompt。第一轮：大纲 + 研究笔记（约 2000 token）。第二轮：大纲 + 研究笔记 + 初稿 + 质检反馈（约 4000 token）。第三轮：大纲 + 研究笔记 + 初稿 V1 + 质检反馈 V1 + 初稿 V2 + 质检反馈 V2（约 6000 token）。Token 消耗是平方级增长，不是线性增长。

解决方案：

方案一：限制上下文窗口。不要把所有历史都喂给 LLM，只传必要的最新信息：

def writer_slim(state: FactoryState) –> dict:
# 只传必要的最新信息，不传完整历史
topic = state["topic"]
outline = state["outline"]
research = state["research"]
# 只传最新一版的质检反馈，不传历史反馈
review_comment = state.get("review_comment", "无")
revision_count = state.get("revision_count", 0)

prompt = f"""主题：{topic}
大纲：{outline}
研究笔记：{research}

{"这是第" + str(revision_count + 1) + "版，质检反馈：" + review_comment if revision_count > 0 else "请撰写初稿。"}

直接输出报告。"""

response = llm.invoke(prompt)
return {"draft": response.content}

方案二：设置 Token 预算。在每个 Agent 函数中统计 Token 用量，超过预算就熔断：

total_tokens = {"count": 0}
TOKEN_BUDGET = 50000

def check_budget():
if total_tokens["count"] > TOKEN_BUDGET:
raise RuntimeError(f"Token 预算耗尽：已用 {total_tokens['count']}，上限 {TOKEN_BUDGET}")

def llm_invoke_with_budget(prompt: str) –> str:
check_budget()
response = llm.invoke(prompt)
# ChatOllama 的 usage 信息
usage = response.response_metadata.get("usage", {})
total_tokens["count"] += usage.get("total_tokens", len(prompt) // 4 + 100)
return response.content

方案三：差量传递。质检员不把完整初稿退回给写手，只告诉写手“第二段缺少数据支撑”——写手自己从 State 中读取初稿修改，不需要在 prompt 里重复完整初稿。

7.2 坑二：上下文膨胀——State 越来越大

场景：你的多 Agent 系统跑了 20 轮，State 里的 log 列表已经有几百条，research 字段存了所有历史版本的研究笔记——每个节点都要序列化整个 State 传给 LLM，最终导致单次请求 prompt 超过模型上下文窗口，直接报错。

根因：Annotated[list, operator.add] 是个双刃剑——它让日志累加很方便，但如果 State 中有多个累加字段，每个都在膨胀，整个 State 就像一个不断长大的气球。

解决方案：

方案一：定期清理 State。在设计图中加入一个“清理节点”，每隔 N 轮压缩历史数据：

def state_cleaner(state: FactoryState) –> dict:
"""每隔几轮执行一次，压缩历史"""
log = state.get("log", [])
if len(log) > 20:
# 只保留最近 10 条日志 + 前面 2 条（起始记录）
log = log[:2] + ["…(已压缩历史日志)…"] + log[–10:]

return {
"log": log,
"research": state["research"], # 保留最新版即可
}

方案二：State 分层。把 State 分为“热数据”（当前轮需要的）和“冷数据”（历史归档）。热数据放在 LangGraph State 中，冷数据写到本地文件或数据库：

import json
from pathlib import Path

def archive_state(state: FactoryState, round_num: int):
"""把历史状态归档到本地文件"""
archive_dir = Path(".temp/archive")
archive_dir.mkdir(parents=True, exist_ok=True)
archive_file = archive_dir / f"round_{round_num}.json"
archive_file.write_text(json.dumps({
"draft": state.get("draft", ""),
"review_comment": state.get("review_comment", ""),
"log": state.get("log", []),
}, ensure_ascii=False, indent=2), encoding="utf-8")

方案三：摘要替代全文。每次 Agent 执行后，把上一轮的完整输出做一个摘要存入 State，不存全文：

def summarize_for_state(full_text: str) –> str:
"""把长文本压缩成摘要存入 State"""
prompt = f"用 100 字以内总结以下内容：\\n{full_text}"
return llm.invoke(prompt).content

7.3 坑三：死循环——Agent 互相踢皮球

场景：质检员说“不够详细”，写手加了内容。质检员又说“太啰嗦了”，写手删了内容。质检员说“不够详细”……写手和质检员陷入无限循环，系统跑了一小时还没出结果。

根因：两个 Agent 的评判标准互相矛盾——质检员同时要求“详细”和“简洁”，这本身就不可能同时满足。没有循环上限的话，Agent 会一直跑下去。

解决方案：

方案一：硬性循环上限（基础防御）。我们 4.4 节已经做了——revision_count >= 3 就强制结束。这是每个多 Agent 系统都必须有的底线防御：

def should_revise(state: FactoryState) –> str:
if state["review_passed"]:
return "end"
if state["revision_count"] >= 3: # 硬上限
return "end"
return "revise"

方案二：质检标准结构化。把质检员的评判标准从自由文本改成结构化检查项，避免主观矛盾：

def reviewer_structured(state: FactoryState) –> dict:
draft = state["draft"]
outline = state["outline"]

prompt = f"""你是一位质检编辑。逐项检查以下报告，每项给出 PASS 或 FAIL。

检查项（每项独立判断，不能互相矛盾）：
1. 完整性：是否覆盖大纲所有子方向？PASS/FAIL
2. 字数：是否在 800-1200 字之间？PASS/FAIL
3. 无明显事实错误：PASS/FAIL

报告初稿：
{draft}

大纲：
{outline}

输出格式：
完整性：PASS/FAIL
字数：PASS/FAIL
事实：PASS/FAIL

所有检查项都 PASS 才算通过。"""

response = llm.invoke(prompt)
result = response.content
passed = result.count("PASS") >= 3 and "FAIL" not in result.split("完整性：")[–1].split("\\n")[0]

return {
"review_comment": result,
"review_passed": passed,
"revision_count": state.get("revision_count", 0) + (0 if passed else 1),
}

方案三：退回时携带具体修改指令。质检员不只能说“不通过”，必须给出具体可执行的修改指令——写手拿到指令后就知道改什么，而不是猜：

# 在质检员的 prompt 中加入：
"""如果不通过，必须给出具体修改指令，格式：
修改指令：
1. 在第二段补充 XX 的数据
2. 删除第四段多余的比喻
3. 把开头改成 XX 风格"""

方案四：Escalation 机制。如果重试 2 次还不通过，不退回写手，而是升级给“高级编辑”——用一个更强模型或更详细的 prompt 做最终修改。这在企业场景中很常见——不是所有问题都能在同级 Agent 之间解决。

def should_revise_with_escalation(state: FactoryState) –> str:
if state["review_passed"]:
return "end"
if state["revision_count"] >= 2:
return "escalate" # 升级给高级编辑
return "revise"

# 在图的定义中加入 escalation 路径
graph.add_node("senior_editor", senior_editor_node)
graph.add_conditional_edges(
"reviewer",
should_revise_with_escalation,
{"end": END, "revise": "writer", "escalate": "senior_editor"},
)
graph.add_edge("senior_editor", END)

八、性能优化与调试技巧

8.1 差异化 Temperature 策略

不同 Agent 的任务性质不同，应该用不同的 temperature：

实现方式——为每个 Agent 初始化独立的 LLM 实例：

llm_creative = ChatOllama(model="qwen3:8b", base_url="http://localhost:11434", temperature=0.7)
llm_precise = ChatOllama(model="qwen3:8b", base_url="http://localhost:11434", temperature=0.1)

def planner(state: FactoryState) –> dict:
# 用创造性模型
response = llm_creative.invoke(prompt)
...

def reviewer(state: FactoryState) –> dict:
# 用精确模型
response = llm_precise.invoke(prompt)
...

8.2 并行执行：让不冲突的 Agent 同时跑

默认情况下 LangGraph 按边顺序串行执行。但有些 Agent 之间没有依赖关系，可以并行。比如研究员和翻译官可以同时工作——研究员整理中文素材，翻译官提前准备术语表。

from langgraph.graph import StateGraph, START, END

# 用 fan-out 模式实现并行
graph.add_edge(START, "planner")
# planner 之后同时启动 researcher 和 translator
graph.add_edge("planner", "researcher")
graph.add_edge("planner", "translator")
# 两者都完成后汇聚到 writer
graph.add_edge("researcher", "writer")
graph.add_edge("translator", "writer")
graph.add_edge("writer", "reviewer")

LangGraph 编译器会自动识别这种 fan-out/fan-in 模式，并行执行 researcher 和 translator，等两者都完成后再执行 writer。这对于减少总执行时间很有效——尤其是当各 Agent 的 LLM 调用时间都较长时。

8.3 调试技巧：打断点看 State

多 Agent 系统最难调试的是“不知道哪个 Agent 出了问题”。LangGraph 的 stream 模式是你最好的调试工具——它在每个节点执行后输出当前 State 快照，你可以看到数据是怎么一步步流转的：

# 调试模式：逐步查看 State 变化
for event in app.stream(initial_state, {"recursion_limit": 20}, stream_mode="updates"):
for node_name, node_state in event.items():
print(f"\\n{'='*40}")
print(f"节点：{node_name}")
print(f"State 更新：")
for key, value in node_state.items():
if key == "log":
print(f" {key}: [{len(value)} 条日志]")
else:
preview = str(value)[:200] if value else "空"
print(f" {key}: {preview}{'…' if len(str(value)) > 200 else ''}")

这段代码会在每个节点执行后打印 State 的变化情况——哪些字段被更新了、值的前 200 字预览。你一眼就能看出“规划官的大纲正常吗”“研究员的笔记有内容吗”“写手的初稿是不是空的”——快速定位问题节点。

8.4 性能数据参考

以 qwen3:8b 在联想昭阳 X5（16GB 内存，Intel Core Ultra 5）上的实测数据为参考：

一次完整执行（无修订）：约 60-125 秒，总 Token 消耗约 12,000。如果质检退回重写一次，时间翻倍，Token 增加 50%（因为写手和质检员都要重跑）。

对比云端 API：同样 4 个 Agent 调用 GPT-4o 级别的 API，首 Token 延迟更低（0.5-1 秒），但单次完整执行成本约 0.05-0.10 美元。本地 Ollama 的优势是免费 + 数据不出门——在需要保密或高频调用的场景中优势明显。

九、总结与经验清单

9.1 一张图回顾全文

从第 12 篇的单 Agent 助手，到本篇的多 Agent 内容生产工厂——你经历了这几个关键跃迁：

9.2 五条经验清单

把全文浓缩成 5 条可以带走的经验：

1. 先画图再写代码

多 Agent 系统的复杂度不在单个 Agent 的代码，而在 Agent 之间的连接关系。动手写代码之前，先用 Mermaid 或纸笔画出流程图——有哪些节点、哪些边、哪些条件分支、哪些循环。团队确认流程图后再写代码，能省掉 80% 的返工。

2. 循环上限是保命符

任何有反馈循环的多 Agent 系统都必须设循环上限——revision_count >= N 就强制结束。没有这个上限的系统迟早会死循环。设 3 次是合理的起点——大部分质量问题在前两次修订中就能解决，第三次还不过说明质检标准和写手能力不匹配，继续循环没意义。

3. Token 预算比 Token 优化更重要

先加 Token 预算熔断（比如 50,000 token 上限），再做 Token 优化（差量传递、摘要替代）。预算是兜底保障——确保即使出了 bug，成本也不会失控。优化是锦上添花——在预算内进一步提效。

4. 质检标准要结构化

质检员用自由文本做判断是最常见的翻车原因——“不够详细”和“太啰嗦”可以无限矛盾。把质检标准拆成独立的结构化检查项（完整性 PASS/FAIL、字数 PASS/FAIL、事实 PASS/FAIL），每项独立判断不互相矛盾。退回时必须给具体修改指令，不能只说“不行重写”。

5. 本地优先，协议预留

在本地用 Ollama + LangGraph 跑通多 Agent 系统，这是性价比最高的方案。但架构设计要为未来留好空间——工具调用走标准接口（为 MCP 预留）、Agent 间通信走 State 而非直接调用（为 A2A 预留）、Agent 能力模块化（为 Agent Plugins 预留）。今天多花 10% 的时间做解耦，明天省 90% 的时间做升级。

9.3 下篇预告

下一篇我们换个方向——从 Agent 协作转向 Agent 记忆。你可能已经发现，本篇的多 Agent 系统每次运行都是“从零开始”——规划官不记得上次规划过什么，质检员不记得上次审过什么。下一篇，我们给 Agent 加上长期记忆：让它们记住过往经验、避免重复犯错、在迭代中持续进化。

下一篇：大模型实战指南（14）——Agent 记忆系统：让 AI 拥有长期记忆

互动问题：你的多 Agent 系统遇到过什么奇葩 bug？欢迎在评论区分享，下一篇我会选最典型的案例做拆解。

9.4 数据来源声明

本文所有事实性数据均来自公开渠道，截至 2026 年 8 月 28 日核实：

• LangGraph：LangChain 官方文档及 GitHub 仓库（github.com/langchain-ai/langgraph），版本 0.3+
• CrewAI：PyPI 页面（pypi.org/project/crewai），版本 1.15.12（2026-08-06），GitHub Star 约 45k
• AutoGen / MAF：Microsoft Reactor 官方公告，AutoGen v0.4 已整合进 Microsoft Agent Framework (MAF) 1.0（2026-04 GA），GitHub Star 约 40k
• OpenAI Agents SDK：OpenAI 官方文档，2026 年 3 月发布 Python 版及独立 JS/TS 版
• MCP 无状态化：Model Context Protocol 官方规范候选版
• A2A 协议：Google + Anthropic 联合推动，A2A 协议规范文档
• Agent Plugins 1.0：2026 年 8 月在 VS Code、Copilot CLI、Copilot SDK 落地
• Gartner 预测：到 2027 年底 40%+ AI Agent 项目将被取消
• 德勤调查：38% 企业试点 Agent，仅 11% 投产
• 麦肯锡调查：62% 企业试验 Agent，低于 10% 规模化部署；70% 员工已使用 AI 工具，仅 27% 管理层认为组织完成面向 Agent 的结构性调整
• AWS/Netflix 联合报告：《分布式 Agent 系统韧性报告》，10+ Agent 系统中单一故障 30 秒内触发级联崩塌概率 42%，从局部异常到全系统不可用中位时间 4.7 秒
• MIT 研究：95% 生成式 AI 试点未带来可衡量回报
• 性能数据：基于联想昭阳 X5-14 IRL（Intel Core Ultra 5 / 16GB）+ Ollama qwen3:8b 本地实测