祥哥的技术博客

主题

📅 2026年8月19日 🏷️ AI 应用 ⏱️ 12 分钟阅读 Multi-AgentLangGraphCrewAIA2AAgent周报

Multi-Agent 协作实战踩坑：从"一个 Agent 干所有"到"Agent 团队" ​

一个 8 年 Java 后端，用 LangGraph 把单个 ReAct Agent 拆成 Planner + Searcher + Writer + Critic 四个 Agent，踩了死循环、State 膨胀、角色越界三个坑。

核心结论：Multi-Agent 的难点不在"怎么跑起来"，在"怎么让 Agent 之间不互相捣乱"。

为什么要从 Single Agent 拆成 Multi-Agent ​

Week 1-5 的 Agent 都是单 Agent：一个 ReAct 循环里，LLM 又要理解问题、又要搜索资料、又要写答案、又要检查质量。这在简单任务上没问题，但任务一复杂，三个问题同时出现：

  • 上下文爆炸：搜索结果、草稿、修改意见全塞在一个 messages 列表里，token 数很快突破模型窗口
  • 角色混淆：LLM 在"搜索"和"写作"之间切换，prompt 里写"你是搜索专家"又写"你要写报告"，模型在两个角色之间摇摆
  • 工具过多：搜索工具 + 格式化工具 + 质量检查工具全挂在同一个 Agent 上，LLM 经常选错工具

Multi-Agent 的思路很直接：把一个什么都干的 Agent 拆成几个各司其职的 Agent，每个只管自己那一段。

一、四种主流架构模式 ​

1. Supervisor / Worker ​

用户 -> Supervisor -> Worker A
                  -> Worker B
                  -> Worker C
         Supervisor 汇总结果 -> 用户

Supervisor 是调度中心，不干活，只决定"把任务分给谁"。Worker 各自有工具和 prompt，干完活把结果交回 Supervisor。

适用场景：任务可以明确拆分成独立子任务（如"分析这三篇论文各自的优缺点"）。

坑：Supervisor 如果也用 LLM，它分配任务本身就有不确定性。有时候 Worker A 明明更合适，它偏分给 Worker B。

2. Router ​

用户 -> Router -> 判断类型 -> 专家 Agent A
                          -> 专家 Agent B
                          -> 专家 Agent C

Router 只做一件事：判断该走哪条路。然后对应的专家 Agent 全程处理。

适用场景：客服系统（技术问题走技术 Agent，退费问题走售后 Agent）。

坑：Router 判断错了就全错了。需要在 Router 后面加一个 fallback Agent 兜底。

3. Hierarchical ​

Manager -> Team Lead A -> Worker A1, A2
       -> Team Lead B -> Worker B1, B2

多层分发，像公司组织架构。

适用场景：大型复杂任务（如"写一份竞品分析报告"-> 拆成市场分析、技术分析、商业模式分析三个子团队）。

坑：层数一多，信息层层传递失真严重。Manager 不知道 Worker 的实际执行细节，Worker 不知道 Manager 的全局意图。

4. Network / Debate ​

Agent A <-> Agent B <-> Agent C

Agent 之间自由通信，互相审查、辩论。

适用场景：需要多角度验证的任务（如代码 Review：安全 Agent 和性能 Agent 各自审查，分歧时辩论）。

坑：最容易死循环。Agent A 说"这样写不安全"，Agent B 说"但这样写性能好"，来回辩论停不下来。

二、Planner -> Searcher -> Writer -> Critic 实现 ​

我选了 Supervisor/Worker 的变体：流水线模式，用 LangGraph 实现。四个 Agent 串行协作：

用户问题 -> Planner -> [子任务1, 子任务2, ...]
                      |
                      v
          Searcher -> [搜索结果1, 搜索结果2, ...]
                      |
                      v
            Writer  -> 草稿报告
                      |
                      v
            Critic   -> 通过？--Yes--> 输出
                      |
                      No --> 回到 Writer（带修改意见）

State 设计 ​

python
from typing import TypedDict, List, Optional

class ResearchState(TypedDict):
    query: str                    # 用户原始问题
    subtasks: List[str]           # Planner 拆解的子任务
    search_results: List[str]     # Searcher 的搜索结果
    draft: str                    # Writer 的草稿
    review_feedback: Optional[str] # Critic 的反馈
    iteration: int                # 当前修订轮次
    max_iterations: int           # 最大修订次数

关键决策：全量 State 传递。 每个 Agent 都能看到完整的 State。好处是 Writer 能看到原始 query 和搜索结果，Critic 能看到草稿和搜索结果。坏处是 State 会越来越大，需要定期清理（比如搜索结果太长就做摘要后再存）。

Node 定义 ​

python
def planner_node(state: ResearchState) -> ResearchState:
    """拆解用户问题为 3-5 个子任务"""
    prompt = f"""你是研究规划师。把以下问题拆成 3-5 个子任务。
    每个子任务应该是可以独立搜索的。
    问题：{state['query']}
    输出格式：JSON 数组"""
    subtasks = call_llm(prompt)
    return {**state, "subtasks": subtasks}

def searcher_node(state: ResearchState) -> ResearchState:
    """对每个子任务执行搜索"""
    results = []
    for task in state["subtasks"]:
        result = search_tool(task)
        results.append(result)
    return {**state, "search_results": results}

def writer_node(state: ResearchState) -> ResearchState:
    """基于搜索结果写草稿"""
    prompt = f"""你是技术写作专家。基于以下搜索结果写一份报告。
    要求：结构清晰、有数据支撑、不超过 800 字。
    
    原始问题：{state['query']}
    搜索结果：{state['search_results']}
    
    {'上一版草稿的修改意见：' + state['review_feedback'] if state['review_feedback'] else ''}
    """
    draft = call_llm(prompt)
    return {**state, "draft": draft, "iteration": state["iteration"] + 1}

def critic_node(state: ResearchState) -> ResearchState:
    """审查草稿质量"""
    prompt = f"""你是质量审查员。审查以下报告。
    检查：1.是否回答了原始问题 2.是否有事实错误 3.结构是否清晰
    如果合格，输出 PASS。如果不合格，输出具体修改意见。
    
    原始问题：{state['query']}
    报告：{state['draft']}
    """
    review = call_llm(prompt)
    return {**state, "review_feedback": review}

Edge / 路由 ​

python
def should_revise(state: ResearchState) -> str:
    """Critic 后的路由：通过则结束，不通过则回到 Writer"""
    if "PASS" in state["review_feedback"]:
        return "end"
    if state["iteration"] >= state["max_iterations"]:
        return "end"  # 超过最大迭代次数，强制结束
    return "revise"

# 构建图
graph = StateGraph(ResearchState)
graph.add_node("planner", planner_node)
graph.add_node("searcher", searcher_node)
graph.add_node("writer", writer_node)
graph.add_node("critic", critic_node)

graph.add_edge("planner", "searcher")
graph.add_edge("searcher", "writer")
graph.add_edge("writer", "critic")
graph.add_conditional_edges("critic", should_revise, {
    "end": END,
    "revise": "writer"
})

app = graph.compile()

三、踩坑 Top 3 ​

坑 1：Critic 死循环 ​

现象：Critic 永远不满意，Writer 改了 5 轮还在"还可以更好"。

根因：Critic 的 prompt 写的是"审查质量"，没有定义什么叫"合格"。LLM 天然倾向于找出可以改进的地方，所以永远能挑出毛病。

解法：

  • Critic prompt 里明确"只有出现事实错误或严重遗漏才不通过，措辞和风格不算"
  • 硬性限制 max_iterations=3，超过强制输出
  • 质量阈值：要求 Critic 打分（1-10），>=7 分就算通过
python
def should_revise(state):
    if state["iteration"] >= state["max_iterations"]:
        return "end"
    score = extract_score(state["review_feedback"])
    if score >= 7:
        return "end"
    return "revise"

坑 2：Planner 过度拆解 ​

现象：用户问"Spring AI 和 LangGraph4j 怎么选"，Planner 拆出 12 个子任务，Searcher 调了 12 次 API，结果一半是重复的。

根因：Planner 的 prompt 写的是"拆成子任务"，没限制数量。LLM 觉得"拆得越多越全面"。

解法：

  • prompt 里硬性限制"3-5 个子任务，不要重复"
  • Planner 输出前加一步去重：如果两个子任务的语义相似度 > 0.8，合并
  • 子任务数量和搜索成本直接挂钩，在 prompt 里告诉 LLM "每个子任务会触发一次搜索，请控制成本"

坑 3：Writer 忽略搜索结果 ​

现象：Writer 写的草稿里出现了搜索结果里没有的信息--LLM 在用训练数据"补脑"。

根因：Writer 的 prompt 虽然写了"基于搜索结果写报告"，但没有强制约束"不得使用搜索结果外的信息"。LLM 觉得搜索结果不够全面，就自己补了。

解法：

  • prompt 改成"只能使用以下搜索结果中的信息，不得添加任何外部知识。如果搜索结果不足以回答某部分，标注'资料不足'"
  • 这和 Week 5 RAG 的拒答约束是同一个思路：宁可说不知道，也不能编

四、Agent 间通信：三种方式对比 ​

方式实现优点缺点适用场景
共享 StateLangGraph 的 TypedDict简单直观，全局可见State 膨胀，耦合度高流水线式协作
消息传递Agent 间发 JSON 消息松耦合，可扩展需要消息路由，复杂度高分布式 Agent
黑板模式共享一个 key-value 存储灵活，Agent 可随时加入并发控制难，需锁并行探索

我的选择：共享 State。原因：LangGraph 原生支持，四个 Agent 在同一个进程里跑，不需要分布式通信。如果以后要跨进程或跨机器，再切消息传递。

五、A2A 协议：Agent 间通信的标准化尝试 ​

A2A（Agent-to-Agent）是 Google 在 2025 年提出的协议，解决的是"不同框架的 Agent 怎么互相通信"的问题。

和 MCP 的关系 ​

┌─────────────────────────────────┐
│  A2A 协议（Agent <-> Agent）      │  上层：Agent 间通信
├─────────────────────────────────┤
│  Agent 框架（LangGraph/CrewAI）   │  中层：编排
├─────────────────────────────────┤
│  MCP 协议（Agent <-> Tool）       │  下层：工具调用
└─────────────────────────────────┘
  • MCP 管 Agent 怎么调工具（Agent -> Database、Agent -> Search API）
  • A2A 管 Agent 怎么调另一个 Agent（Planner Agent -> Searcher Agent）
  • 两者正交，不冲突

Agent Card 结构 ​

json
{
  "name": "Searcher Agent",
  "description": "搜索网络信息并返回结构化结果",
  "capabilities": ["web_search", "content_extraction"],
  "endpoint": "https://agent.example.com/a2a/searcher",
  "authentication": "bearer_token"
}

类似微服务里的服务注册。每个 Agent 发布自己的 Agent Card，其他 Agent 通过 Card 找到它并调用。

当前生态（2026-08） ​

A2A 还在早期阶段。规范在演进，主流框架还没原生支持。但方向是对的：Agent 要走向生产，必须有标准化的通信协议，就像微服务时代的 HTTP + JSON。

六、CrewAI vs LangGraph 多 Agent ​

维度LangGraphCrewAI
编排方式图（State + Node + Edge）角色扮演（Agent + Task + Crew）
灵活性极高，可以画任意拓扑中等，预定义了几种模式
学习曲线陡（要理解 State/Edge/Checkpoint）平缓（声明式配置）
适用场景复杂流程、需要精确控制快速验证、角色明确的场景
Java 生态LangGraph4j（社区版）无（Python only）
HITL 支持原生 interrupt需要自己实现
可观测性LangSmith / LangfuseCrewAI Dashboard（基础）

我的选择：LangGraph。原因：祥哥走 Java 路线，LangGraph4j 虽然是社区版但能用；Week 3 已经有 LangGraph 基础；需要 Checkpoint 和 HITL 这些高级特性。

七、生产级 Multi-Agent 的三个考量 ​

1. 容错 ​

一个 Agent 挂了怎么办？

  • 超时：每个 Agent Node 设超时，超时返回 fallback 结果
  • 重试：搜索工具调失败，重试 2 次后降级到本地知识库
  • 熔断：如果某个 Agent 连续失败 N 次，跳过它走兜底逻辑
python
from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def searcher_node(state):
    try:
        results = [search_tool(t) for t in state["subtasks"]]
    except Exception:
        results = ["搜索失败，使用本地知识库"]  # fallback
    return {**state, "search_results": results}

2. 可观测性 ​

多 Agent 系统是黑盒中的黑盒。出问题时，必须能看到每个 Agent 的输入输出。

  • 每个 Node 打日志：Agent 名、输入 State 摘要、输出 State 摘要、耗时、token 数
  • 链路追踪：给每次执行分配 trace_id，所有 Agent 的日志关联到同一个 trace
  • State 快照：每次 Node 执行完保存 State 快照，支持回放

LangGraph 的 Checkpoint 机制天然支持 State 快照，配合 LangSmith/Langfuse 可以做到全链路追踪。

3. 成本控制 ​

4 个 Agent = 4 倍 LLM 调用。如果 Critic 修订 3 轮，就是 4 + 3 = 7 次 LLM 调用。

  • Planner 用小模型：任务拆解不需要最强模型，用便宜的
  • Critic 用大模型：质量审查需要判断力，用最强的
  • Searcher 不用 LLM：搜索工具直接调 API，不需要 LLM 参与
  • 设 token 预算：整个流程超过 N token 就强制结束

总结 ​

Multi-Agent 不是银弹。它解决的是 Single Agent 在复杂任务上的上下文爆炸和角色混淆，但代价是更复杂的编排、更高的成本、更难的可观测性。

三个最值钱的结论：

  • Critic 必须有明确的终止条件：要么质量阈值 + 打分，要么硬性迭代上限，两个都有最好。否则死循环
  • Agent 之间通过共享 State 通信最简单：但要注意 State 膨胀，长搜索结果要摘要后再存
  • A2A 是方向，MCP 是当下：A2A 还在早期，MCP 已经是生产标配。先做好 MCP 工具层，A2A 关注规范演进

如果面试官问"你怎么设计 Multi-Agent 系统"，我的回答是：先用 Supervisor/Worker 模式跑通流水线，共享 State 传递信息，Critic 用打分 + 最大迭代双保险防死循环，每个 Node 打日志做全链路追踪。不追求花哨的拓扑，能跑通、能调试、能停得住就是好架构。

参考：LangGraph 文档 / A2A 协议规范 / Week 3 LangGraph 基础笔记 / Week 5 RAG 评估笔记

手撕 → 框架 → 生产化，一步步把 AI Agent 工程化搞透。
粤ICP备2026104257号-1

Copyright © 2026 祥哥