Multi-Agent 多智能体

Multi-Agent（多智能体）系统通过多个 Agent 的协作来解决复杂问题。LangGraph 的图结构天然支持多智能体编排。

多智能体的模式

1. Supervisor（监督者）模式

一个"主管"Agent 负责协调多个"子"Agent：

               → Agent A（搜索）
用户 → Supervisor  → Agent B（计算）→ 汇总 → 输出
               → Agent C（翻译）

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Literal, Annotated, List
import operator

class SupervisedState(TypedDict):
    messages: Annotated[List[dict], operator.add]
    next_agent: str
    task: str

# 监督者节点
def supervisor_node(state: SupervisedState):
    """决定哪个 Agent 执行下一步"""
    prompt = f"""
    当前任务: {state['task']}
    消息历史: {state['messages']}

    选择下一个执行的 agent: search, calculate, translate, 或 finish
    """
    decision = llm.invoke(prompt)

    return {"next_agent": decision.content.strip()}

# 子 Agent 节点
def search_agent(state: SupervisedState):
    result = search_tool.invoke(state["task"])
    return {"messages": [{"role": "assistant", "content": f"搜索结果: {result}"}]}

def calculate_agent(state: SupervisedState):
    result = calculate_tool.invoke(state["task"])
    return {"messages": [{"role": "assistant", "content": f"计算结果: {result}"}]}

def translate_agent(state: SupervisedState):
    result = translate_tool.invoke(state["task"])
    return {"messages": [{"role": "assistant", "content": f"翻译结果: {result}"}]}

def is_finish(state: SupervisedState):
    if state["next_agent"] == "finish":
        return END
    return state["next_agent"]

# 构建
builder = StateGraph(SupervisedState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("search", search_agent)
builder.add_node("calculate", calculate_agent)
builder.add_node("translate", translate_agent)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges(
    "supervisor",
    is_finish,
    ["search", "calculate", "translate", END]
)
builder.add_edge("search", "supervisor")   # 回到主管
builder.add_edge("calculate", "supervisor")
builder.add_edge("translate", "supervisor")

graph = builder.compile()

2. 网络模式

Agent 之间自由通信，没有中心协调者：

Agent A ←→ Agent B
   ↕        ↕
Agent C ←→ Agent D

3. 流水线模式

每个 Agent 处理一个阶段，输出传递给下一个：

Agent A（分析）→ Agent B（搜索）→ Agent C（生成）→ Agent D（质检）

class PipelineState(TypedDict):
    query: str
    analysis: str
    search_results: str
    draft: str
    final_output: str

def analyst(state: PipelineState):
    analysis = analyze_tool.invoke(state["query"])
    return {"analysis": analysis}

def searcher(state: PipelineState):
    results = search_tool.invoke(state["analysis"])
    return {"search_results": results}

def writer(state: PipelineState):
    draft = llm.invoke(f"基于分析: {state['analysis']} 和搜索结果: {state['search_results']} 写一份报告")
    return {"draft": draft}

def reviewer(state: PipelineState):
    feedback = llm.invoke(f"审核以下报告: {state['draft']}")
    return {"final_output": f"{state['draft']}\n\n审核反馈: {feedback}"}

builder = PipelineGraph(PipelineState)
builder.add_node("analyst", analyst)
builder.add_node("searcher", searcher)
builder.add_node("writer", writer)
builder.add_node("reviewer", reviewer)

builder.add_edge(START, "analyst")
builder.add_edge("analyst", "searcher")
builder.add_edge("searcher", "writer")
builder.add_edge("writer", "reviewer")
builder.add_edge("reviewer", END)

4. 竞争模式（投票）

多个 Agent 独立处理同一任务，然后通过投票选择最佳结果：

class VotingState(TypedDict):
    task: str
    responses: Annotated[List[str], operator.add]
    final_answer: str

def agent_a(state: VotingState):
    return {"responses": [agent_a_solve(state["task"])]}

def agent_b(state: VotingState):
    return {"responses": [agent_b_solve(state["task"])]}

def agent_c(state: VotingState):
    return {"responses": [agent_c_solve(state["task"])]}

def voter(state: VotingState):
    best = llm.invoke(f"从以下回答中选择最好的: {state['responses']}")
    return {"final_answer": best}

builder = StateGraph(VotingState)
builder.add_node("agent_a", agent_a)
builder.add_node("agent_b", agent_b)
builder.add_node("agent_c", agent_c)
builder.add_node("voter", voter)

# 并行执行三个 Agent
builder.add_edge(START, "agent_a")
builder.add_edge(START, "agent_b")
builder.add_edge(START, "agent_c")

# 全部完成后投票
builder.add_edge("agent_a", "voter")
builder.add_edge("agent_b", "voter")
builder.add_edge("agent_c", "voter")
builder.add_edge("voter", END)

多 Agent 的通信方式

通过 State 传递消息

最简单的方式，每个 Agent 读取和写入共享 State：

class SharedState(TypedDict):
    messages: Annotated[list, operator.add]
    tasks: dict
    results: Annotated[list, operator.add]

通过消息通道

高级方式，使用专门的通道进行通信：

from langgraph.graph import StateGraph, MessageGraph

通过子图隔离

每个 Agent 有自己的子图，父图协调通信：

class TeamState(TypedDict):
    messages: Annotated[list, operator.add]
    current_agent: str

# Agent A 子图
agent_a_graph = create_agent_a_subgraph()
# Agent B 子图
agent_b_graph = create_agent_b_subgraph()

# 父图
builder = StateGraph(TeamState)
builder.add_node("agent_a", agent_a_graph)
builder.add_node("agent_b", agent_b_graph)

实际案例：多 Agent 研究助手

class ResearchState(TypedDict):
    topic: str
    plan: str
    search_results: Annotated[List[str], operator.add]
    draft: str
    reviewed: bool
    final: str

def planner(state: ResearchState):
    """制定研究计划"""
    plan = llm.invoke(f"制定研究计划: {state['topic']}")
    return {"plan": plan}

def researcher_web(state: ResearchState):
    results = web_search_tool.invoke(state["plan"])
    return {"search_results": [f"网页: {results}"]}

def researcher_docs(state: ResearchState):
    results = doc_search_tool.invoke(state["plan"])
    return {"search_results": [f"文档: {results}"]}

def writer(state: ResearchState):
    draft = llm.invoke(f"""
    研究计划: {state['plan']}
    搜索结果: {state['search_results']}
    撰写研究报告
    """)
    return {"draft": draft}

def reviewer(state: ResearchState):
    feedback = llm.invoke(f"审核报告: {state['draft']}")
    if "需要修改" in feedback:
        return {"reviewed": False}
    return {"reviewed": True, "final": state["draft"]}

builder = StateGraph(ResearchState)
builder.add_node("planner", planner)
builder.add_node("web_researcher", researcher_web)
builder.add_node("doc_researcher", researcher_docs)
builder.add_node("writer", writer)
builder.add_node("reviewer", reviewer)

builder.add_edge(START, "planner")
builder.add_edge("planner", "web_researcher")
builder.add_edge("planner", "doc_researcher")
builder.add_edge("web_researcher", "writer")
builder.add_edge("doc_researcher", "writer")
builder.add_edge("writer", "reviewer")
builder.add_conditional_edges(
    "reviewer",
    lambda s: "writer" if not s["reviewed"] else END,
    ["writer", END]
)

最佳实践

1. 从 Supervisor 模式开始，最简单可控
2. 使用子图隔离 Agent，每个 Agent 有自己的 context
3. 定义清晰的通信协议，Agent 之间通过 State 传递结构化数据
4. 限制 Agent 权限，不要给 Agent 不需要的工具
5. 监控和日志：多 Agent 系统调试难度高，务必开启 tracing

参考

• 官方文档：Multi-agent systems
• 官方文档：LangGraph Overview