LangGraph 18：多智能体协作

发表于 更新于 分类于
用 Supervisor 编排与 Handoff 移交两种模式组织多个 Agent，并讲解共享 checkpointer 下的多智能体会话。

前言

一张图里只放一个 ReAct 循环，任务一大就变成「一个模型干所有事」。
把不同职责拆成多个 Agent（研究、写码、客服……），再用编排层决定谁上场，是 LangGraph 多智能体的核心。
官方文档里多智能体主要有两种写法：Supervisor 编排（中心仲裁）与 Handoff 移交（Swarm 式互相转手）。
本文两种都写可跑骨架，最后用共享 checkpointer 串成可续话的会话。
单个 Agent 的构造与工具挂载见 《LangGraph 06：工具调用》；跨会话记忆见 《LangGraph 08：Checkpoint 持久化》。
示例统一对接 火山方舟 Coding Plan，模型用 ark-code-latest。
下文需要 Python 3.12+，依赖用 uv 管理。

概要

先用 create_react_agent 造两个专职 Agent：研究员（查天气/时间）与编码员（查代码）。
再分别演示两种多智能体图：

  1. Supervisor：一个「主管」节点用 bind_tools 选人，Command(goto=...) 把人叫上场，最后自己收尾。
  2. Handoff：每个 Agent 手里都有一把 create_handoff_tool，任务不对就转给下家，像接线台一样互相移交。

最后给图挂 InMemorySaver，同一条 thread_id 能连续聊多轮。

依赖

建议使用 Python 3.12 及以上。

1
2
3
4
uv init langgraph-multi-agent
cd langgraph-multi-agent
uv venv --python 3.12
uv add "langgraph>=1.0,<2.0" "langchain>=1.0,<2.0" langchain-openai python-dotenv rich

在项目根目录创建 .env ，写入 Coding Plan 的 Key 与专用 Base URL。

1
2
OPENAI_API_KEY=你的火山方舟 API Key
OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

请勿把 Base URL 写成普通方舟 .../api/v3 ，以免无法抵扣 Coding Plan 额度。

专职 Agent

先把两个会互相转手的角色各做成一棵独立的 create_react_agent。
create_react_agent 返回的是编译好的图，内部就是「模型 ↔ 工具」循环，可当节点直接挂进大图。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.prebuilt import create_react_agent

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """根据城市查询当日天气。

    Args:
        city: 城市名称
    """
    return f"{city} 今天天气不错"


@tool(parse_docstring=True)
def get_time(city: str) -> str:
    """根据城市查询当地时间。

    Args:
        city: 城市名称
    """
    return f"{city} 现在是下午 3 点"


@tool(parse_docstring=True)
def check_code(code: str) -> str:
    """对一段 Python 代码做静态检查。

    Args:
        code: 待检查的 Python 代码
    """
    if "def " in code:
        return "检查通过：函数定义完整。"
    return "提示：缺少函数定义。"

研究者只挂天气/时间工具，编码员只挂代码检查工具。
prompt 决定角色分工，模型只在自己的工具集里干活。

1
2
3
4
5
6
7
8
9
10
11
researcher = create_react_agent(
    model,
    tools=[get_weather, get_time],
    prompt="你是研究员，负责查询天气与时间，用中文简洁作答。",
)

coder = create_react_agent(
    model,
    tools=[check_code],
    prompt="你是编码员，负责检查 Python 代码，用中文给出结论。",
)

把上面保存为 src/agents.py ，两种图都从这里 import。
model 是共享的，两个 Agent 各挂各的工具，互不干扰。

Supervisor 编排

主管不直接干活，而是拿「转交给谁」当工具。
transfer_to_* 每个工具对应一个 Agent 节点，主管模型调用哪个工具，Command(goto=...) 就进哪个节点。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from langchain_core.tools import tool
from langgraph.types import Command


@tool(parse_docstring=True)
def transfer_to_researcher(task: str) -> str:
    """把任务交给研究员处理。

    Args:
        task: 交给研究员的问题
    """
    return f"已转交研究员：{task}"


@tool(parse_docstring=True)
def transfer_to_coder(task: str) -> str:
    """把任务交给编码员处理。

    Args:
        task: 交给编码员的问题
    """
    return f"已转交编码员：{task}"


supervisor_tools = [transfer_to_researcher, transfer_to_coder]
supervisor_model = model.bind_tools(supervisor_tools)

主管节点根据模型挑出的工具名决定去哪个节点；
没有工具调用就 Command(goto=END) ，整张图结束。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from typing import Literal

from langchain.messages import HumanMessage
from langgraph.graph import END, MessagesState, START, StateGraph

NodeName = Literal["researcher", "coder", "__end__"]


def supervisor_node(state: MessagesState) -> Command[NodeName]:
    response = supervisor_model.invoke(state["messages"])
    calls = response.tool_calls
    if not calls:
        return Command(goto=END, update={"messages": [response]})
    name = calls[0]["name"]
    goto = {"transfer_to_researcher": "researcher", "transfer_to_coder": "coder"}[name]
    return Command(goto=goto, update={"messages": [response]})

把主管、两个 Agent 都挂进同一张 MessagesState 图。
Agent 节点之间不连边，谁上场完全由主管的 Command(goto=...) 决定。

1
2
3
4
5
6
7
builder = StateGraph(state_schema=MessagesState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_edge(START, "supervisor")

supervisor_graph = builder.compile()

保存为 src/supervisor.py 。
跑一个混合问题，让主管先派研究员、再派编码员：

1
2
3
4
5
6
7
8
9
10
from rich import print as rprint

from src.supervisor import supervisor_graph

result = supervisor_graph.invoke(
    {"messages": [HumanMessage(content="北京天气怎么样？顺便帮我检查这段代码：def hello(): pass")]},
    config={"recursion_limit": 20},
)
for msg in result["messages"]:
    msg.pretty_print()

recursion_limit 给足，多轮转手别把递归打爆。
想让人在转手前点头，参照 《LangGraph 11：HITL 人机协同》 在主管节点里加 interrupt。

Handoff 移交

Swarm 式写法不给主管，而是给每个 Agent 一把转手工具。
create_handoff_tool 造的工具执行后返回 Command(goto=agent_name) ，控制权直接转给目标 Agent。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from langgraph.prebuilt import create_handoff_tool


def make_agent(name: str, system_prompt: str, tools: list) -> object:
    handoff_tool = create_handoff_tool(
        agent_name=name,
        description=f"把问题移交给 {name}，由它直接回答或继续转交。",
    )
    return create_react_agent(
        model,
        tools=list(tools) + [handoff_tool],
        prompt=system_prompt,
    )


triage = make_agent(
    "triage",
    "你是接线台。天气/时间问题转给研究员，代码问题转给编码员。",
    [],
)
researcher = make_agent(
    "researcher",
    "你是研究员，查天气和时间。不适合你的问题转给编码员。",
    [get_weather, get_time],
)
coder = make_agent(
    "coder",
    "你是编码员，检查 Python 代码。不适合你的问题转给研究员。",
    [check_code],
)

Handoff 图把三个角色连成环：边只表示「允许转去的方向」，实际去哪由工具返回的 Command(goto=...) 决定。

1
2
3
4
5
6
7
8
9
builder = StateGraph(state_schema=MessagesState)
builder.add_node("triage", triage)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_edge(START, "triage")
builder.add_edge("triage", "researcher")
builder.add_edge("triage", "coder")
builder.add_edge("researcher", "coder")
builder.add_edge("coder", "researcher")

保存为 src/swarm.py ，连同共享 checkpointer 一起编译。

共享记忆

多智能体也可以有会话：给编译的图挂 checkpointer，用 thread_id 串多轮对话。
Supervisor 与 Swarm 两种图都可以这么挂。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import MessagesState, START, StateGraph

from src.swarm import triage, researcher, coder

builder = StateGraph(state_schema=MessagesState)
builder.add_node("triage", triage)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_edge(START, "triage")
builder.add_edge("triage", "researcher")
builder.add_edge("triage", "coder")
builder.add_edge("researcher", "coder")
builder.add_edge("coder", "researcher")

swarm_graph = builder.compile(checkpointer=InMemorySaver())

同一 thread_id 下先问天气再问代码，Agent 们记得上一轮上下文。

1
2
3
4
5
6
7
8
from rich import print as rprint

from src.swarm import swarm_graph

config = {"configurable": {"thread_id": "s-1"}, "recursion_limit": 20}

rprint(swarm_graph.invoke({"messages": [HumanMessage(content="北京现在几点？")]}, config)["messages"][-1])
rprint(swarm_graph.invoke({"messages": [HumanMessage(content="那这段代码呢：def hello(): pass")]}, config)["messages"][-1])

内存版只适合演示。
生产环境换成 《LangGraph 08：Checkpoint 持久化》 里的 PostgresSaver，多智能体会话才能跨进程续跑。

总结

  1. 多智能体两种主写法：Supervisor 中心编排，Handoff 互相转手。
  2. Supervisor：bind_tools 的转交工具 + Command(goto=...)；无工具调用即 goto END。
  3. Handoff：create_handoff_tool 给每个 Agent 一把转手工具，环状边只约束可转方向。
  4. 专职 Agent 用 create_react_agent 构造，prompt 分工，可当节点直接进大图。
  5. 共享 checkpointer + thread_id 让多智能体会话可续聊；生产用 Postgres。