多 Agent 系统
问题
什么是多 Agent 系统?多个 Agent 如何协作?
答案
多 Agent 系统是由多个专业化 Agent 协作完成复杂任务的架构。每个 Agent 有特定的角色、工具和职责。
一、为什么需要多 Agent
| 问题 | 单 Agent 局限 | 多 Agent 优势 |
|---|---|---|
| 任务复杂 | 一个 Prompt 难以覆盖所有领域 | 专业分工 |
| 工具太多 | >20 个工具选择准确率下降 | 每个 Agent 只需少量工具 |
| 质量保障 | 自评可能盲区 | Agent 间交叉检查 |
| 流程明确 | 黑盒决策难调试 | 角色清晰、流程可追踪 |
二、协作模式
1. 主管模式(Supervisor)
- 一个中心 Agent 协调分配任务
- 适合有明确管理层级的场景
2. 对等协作模式(Peer)
- Agent 之间平等通信
- 适合需要多角度讨论/辩论的场景
3. 流水线模式(Pipeline)
- 固定流程,前一阶段输出是后一阶段输入
- 适合 DevOps、内容创作等流水线场景
三、LangGraph 多 Agent 实现
from langgraph.graph import StateGraph, MessagesState, START, END
# 定义各 Agent 节点
def researcher(state: MessagesState):
"""研究 Agent:搜索收集信息"""
response = research_llm.invoke(state["messages"])
return {"messages": [response]}
def writer(state: MessagesState):
"""写作 Agent:基于研究信息撰写"""
response = writer_llm.invoke(state["messages"])
return {"messages": [response]}
def reviewer(state: MessagesState):
"""评审 Agent:检查质量"""
response = reviewer_llm.invoke(state["messages"])
return {"messages": [response]}
# 路由函数:根据评审结果决定下一步
def route_after_review(state: MessagesState):
last_message = state["messages"][-1].content
if "APPROVED" in last_message:
return END
return "writer" # 返回修改
# 构建图
graph = StateGraph(MessagesState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")
graph.add_conditional_edges("reviewer", route_after_review)
app = graph.compile()
四、通信机制
| 方式 | 说明 | 优点 | 缺点 |
|---|---|---|---|
| 共享状态 | 所有 Agent 读写同一个 State | 简单直接 | 可能冲突 |
| 消息传递 | Agent 间发送消息 | 解耦清晰 | 消息格式要统一 |
| 黑板模式 | 共享黑板,Agent 各取所需 | 灵活 | 需要协调 |
常见面试问题
Q1: 多 Agent 的主要挑战是什么?
答案:
- 通信开销:Agent 间消息传递增加 Token 成本和延迟
- 协调复杂:死循环、互相等待、任务分配不均
- 错误传播:一个 Agent 的错误可能影响全局
- 调试困难:多个 Agent 的决策链路难以追踪
Q2: 什么时候用多 Agent 而不是单 Agent?
答案:
- 用单 Agent:任务相对简单、工具
<10个、不需要多角色 - 用多 Agent:需要专业分工、工具
>20个、需要交叉验证、流程固定(如 DevOps)