Skip to content

多智能体系统

单个智能体再强也有上限:工具一多就乱、上下文一长就慢、任务一复杂就顾此失彼。解决办法是 多个智能体协作——每个 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 协作的核心是共享状态。我们自定义一个状态,包含 messagesnext(下一步路由):

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 模式