Appearance
多智能体系统
单个智能体再强也有上限:工具一多就乱、上下文一长就慢、任务一复杂就顾此失彼。解决办法是 多个智能体协作——每个 agent 专精一件事,组合起来完成大任务。本章讲多智能体系统的常见模式与实现。
一、为什么需要多智能体
| 单 Agent 痛点 | 多 Agent 解法 |
|---|---|
| 工具多,模型选错 | 每个 agent 只装 2-3 个相关工具 |
| 上下文长,token 贵 | 各 agent 各自维护短上下文 |
| 任务多步骤,难调试 | 职责分离,单点失败容易定位 |
| 想扩展新能力 | 加一个 agent 节点即可,不动旧的 |
典型场景:写作助手(研究员查资料、写手写正文、编辑校对)、客服系统(路由 agent + 退换货 agent + 售后 agent)、研发团队(前端 + 后端 + 测试)。
二、四种协作模式总览
1. 网络型(Network)
所有 agent 互相可调用,去中心化。灵活但容易乱,不推荐新手用。
2. Supervisor 主管型
一个主管 agent 决定下一步交给哪个 worker,worker 干完回到主管。最常用、最稳定,下一章Supervisor 模式专门讲。
3. 层级型(Hierarchical)
主管上面还有主管,形成树状结构。适合大型任务,见层级智能体。
4. 手拉手(Sequential)
agent A → agent B → agent C,固定流水线。最简单,适合流程明确的任务。
mermaid
flowchart LR
subgraph 网络型
A1((A)) <--> A2((B))
A2 <--> A3((C))
A1 <--> A3
end
subgraph 主管型
S1[Supervisor] --> W1[Worker1]
S1 --> W2[Worker2]
W1 --> S1
W2 --> S1
end
subgraph 手拉手
Q1[A] --> Q2[B] --> Q3[C]
end三、状态在 agent 间共享
多 agent 协作的核心是共享状态。我们自定义一个状态,包含 messages 和 next(下一步路由):
python
from typing import TypedDict, Annotated, Literal
from langgraph.graph import MessagesState
class AgentState(TypedDict):
messages: Annotated[list, "add"] # 消息累加
next: Literal["researcher", "writer", "FINISH"] # 下一步去哪
Annotated[list, "add"]表示这个字段用"追加"而非"覆盖"策略更新,和MessagesState内置行为一致。
四、手拉手示例:研究员 + 写手
我们做一个最简单的两 agent 流水线:研究员查资料→写手根据资料写文章。
4.1 定义两个 agent
每个 agent 都是一个 create_react_agent,装备不同工具、不同人设:
python
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
# 研究员的"搜索"工具(模拟)
@tool
def search(topic: str) -> str:
"""搜索某个主题的资料,返回一段文字。"""
return f"关于【{topic}】的资料:它是一种2023年兴起的技术,特点是A、B、C。"
researcher = create_react_agent(
model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
tools=[search],
prompt="你是研究员。用 search 工具查资料,把要点列出来,不要自己写文章。",
)
# 写手不配工具,纯靠 LLM
writer = create_react_agent(
model=ChatOpenAI(model="gpt-4o-mini", temperature=0.7),
tools=[],
prompt="你是写手。根据研究员提供的资料,写一篇 200 字左右的科普短文。",
)4.2 用子图组装
把两个 agent 作为节点放进一个外层 StateGraph:
python
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage, SystemMessage
def researcher_node(state: AgentState):
# 把用户问题交给研究员
result = researcher.invoke({"messages": state["messages"]})
# 研究员的最终回复作为新消息
last = result["messages"][-1]
return {"messages": [last], "next": "writer"}
def writer_node(state: AgentState):
# 给写手一个明确指令:基于上面资料写文章
msgs = state["messages"] + [
SystemMessage(content="请根据以上资料,写一篇 200 字科普短文。")
]
result = writer.invoke({"messages": msgs})
last = result["messages"][-1]
return {"messages": [last], "next": "FINISH"}
builder = StateGraph(AgentState)
builder.add_node("researcher", researcher_node)
builder.add_node("writer", writer_node)
builder.add_edge(START, "researcher")
builder.add_edge("researcher", "writer")
builder.add_edge("writer", END)
team = builder.compile()4.3 调用
python
result = team.invoke({
"messages": [HumanMessage(content="请介绍一下 LangGraph。")],
"next": "researcher",
})
print(result["messages"][-1].content)流程:用户问题 → 研究员调 search 拿资料 → 写手基于资料写文章。整个协作图:
mermaid
flowchart LR
U([用户问题]) --> R[研究员<br/>调 search 工具]
R --> W[写手<br/>写科普短文]
W --> O([最终文章])五、动态路由:让 LLM 决定下一个 agent
手拉手是写死的。更灵活的做法是加一个 router 节点,让 LLM 根据当前状态决定下一步去哪。这就是 Supervisor 的雏形(详见下一章):
python
from typing import Literal
def router(state: AgentState) -> Literal["researcher", "writer", "END"]:
last_msg = state["messages"][-1].content.lower()
if "查" in last_msg or "搜索" in last_msg or "资料" in last_msg:
return "researcher"
if "写" in last_msg or "文章" in last_msg:
return "writer"
return "END"
builder.add_conditional_edges("router", router)六、协作图(完整结构)
mermaid
flowchart TB
S([START]) --> R[router 路由节点]
R -->|查资料| RE[researcher]
R -->|写文章| WR[writer]
RE --> R
WR --> R
R -->|完成| E([END])七、常见踩坑
1. 死循环
agent A 把任务丢给 B,B 又丢回 A,无限循环。表现是 recursion_limit 报错。
防范:
- 给 router 加"完成"出口,明确结束条件。
- 在状态里加
iteration计数器,超过 N 次强制结束。 - 用
Literal类型约束next字段的可选值,编译期就能发现笔误。
2. 职责重叠
两个 agent 都会写代码,结果互相覆盖、互相拆台。原则:一个能力只交给一个 agent,人设在 prompt 里写死边界。
3. 状态字段丢失
子 agent 内部状态(如它的中间思考)默认不传到外层。如果外层需要,要在 node 函数里显式提取并放进返回值。例如:
python
def researcher_node(state):
result = researcher.invoke({"messages": state["messages"]})
# 只取最后一条,中间的 tool_calls 不会被外层看到
return {"messages": [result["messages"][-1]]}4. 子 agent 也消耗 recursion_limit
外层图的递归上限是整条调用链共享的。子 agent 内部 ReAct 循环也算步数。复杂系统要把 recursion_limit 调到 100+。
八、小结
- 多智能体解决"单 agent 能力有限"问题,核心是职责分离 + 共享状态。
- 四种模式:网络、Supervisor、层级、手拉手。新手从手拉手和 Supervisor 开始。
- 用
StateGraph把多个create_react_agent当节点组装即可。 - 踩坑:死循环、职责重叠、状态丢失、递归上限。
下一章深入讲最常用的 Supervisor 模式。