Appearance
第17章 多 Agent 模式实战
17.1 引言
单个 Agent 能力有限——它只有一组工具、一种提示、一个执行循环。当任务复杂度上升时(如"帮我分析这家公司的财报,然后写一份投资建议书"),单个 Agent 要么因为工具太多导致 LLM 选择困难,要么因为系统提示过长影响推理质量。多 Agent 系统通过分工协作解决这个问题:不同的 Agent 负责不同的能力域,通过明确的通信机制协同完成复杂任务。
LangGraph 的底层基础设施——StateGraph、Send、Command、子图——天然支持多 Agent 模式的构建。图的节点可以是子图(即嵌套的 Agent),条件边可以实现动态路由,Send 支持一对多的任务分发,Command 支持跨图的状态更新和导航。
本章将系统介绍基于 LangGraph 实现的四种典型多 Agent 架构:Supervisor(监督者)、Swarm(蜂群)、分层 Agent 和协作 Agent。每种模式都有其适用场景和权衡,我们会结合源码和实战代码深入分析它们的设计原理。
本章要点
- Supervisor 模式——中央调度者分配任务给专业 Agent
- Swarm 模式——Agent 之间点对点的控制权交接
- 分层 Agent——多层级的 Supervisor 树形结构
- 协作 Agent——共享状态的对等协作
- 基于 LangGraph 原语实现各种模式的核心技术
17.2 多 Agent 的核心抽象
17.2.1 Agent 作为子图
在 LangGraph 中,一个 Agent 本质上就是一个 CompiledStateGraph。多 Agent 系统就是一个图中嵌套了多个子图:
python
# 每个 Agent 是一个独立的图
research_agent = create_react_agent(model, research_tools, name="researcher")
writer_agent = create_react_agent(model, writing_tools, name="writer")
# 多 Agent 系统是一个包含子图的父图
builder = StateGraph(TeamState)
builder.add_node("researcher", research_agent)
builder.add_node("writer", writer_agent)17.2.2 通信机制的三种模式
17.3 Supervisor 模式
17.3.1 架构概述
Supervisor 模式是最直观的多 Agent 架构:一个"主管"Agent 接收用户请求,分析任务需求,然后将子任务分配给专业 Agent。每个专业 Agent 完成后将结果返回给 Supervisor,Supervisor 决定是否需要更多工作。
17.3.2 实现方式
python
from typing import Annotated, Literal, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import create_react_agent
from langgraph.types import Command
class SupervisorState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
next_agent: str
# 创建专业 Agent
research_agent = create_react_agent(model, research_tools, name="researcher")
writer_agent = create_react_agent(model, writing_tools, name="writer")
def supervisor_node(state: SupervisorState) -> Command:
"""Supervisor 决定下一步交给哪个 Agent"""
response = supervisor_llm.invoke([
SystemMessage(content="""你是一个任务分配主管。
分析用户请求,决定交给哪个 Agent:
- researcher: 负责信息检索和数据分析
- writer: 负责内容创作和文档编写
- FINISH: 任务完成,返回最终结果"""),
*state["messages"]
])
# 解析 LLM 的决策
next_agent = parse_decision(response.content)
if next_agent == "FINISH":
return Command(goto=END, update={"messages": [response]})
else:
return Command(
goto=next_agent,
update={"messages": [response]}
)
# 构建 Supervisor 图
builder = StateGraph(SupervisorState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", research_agent)
builder.add_node("writer", writer_agent)
builder.add_edge(START, "supervisor")
builder.add_edge("researcher", "supervisor") # 完成后返回 Supervisor
builder.add_edge("writer", "supervisor")
graph = builder.compile()