# 使用LangChain+LangGraph开发AI应用（10） - 多智能体编排


![](https://static.xiongneng.me/langchain-langgraph-cover-20261017193538.png)

## 学习进度

单个 agent 解决一个完整任务，任务复杂到需要查资料、写稿、审稿三种角色各干一摊时，把它们排进同一张图，就是多智能体编排。这一篇讲的就是怎么排。

底座是三样现成的东西。每个 agent 就是一个节点，或者一个 compile 过的子图；节点之间的控制流转靠 Command(goto=)；共享数据靠状态和 Reducer。没有新引擎，只有几套把节点组织起来的成熟阵型。示例在 `source/hello_multi.py`，八个 demo，全部纯函数。完整案例是一个编辑部流水线 `source/newsroom.py`，主管派活，写手写稿，审稿带禁词检查，被打回就改稿重审，四个角色一张图跑完。

## 前言用途

什么时候需要多个 agent。单 agent 加一筐工具能干很多事，但工具超过二十个，模型选错的概率肉眼可见地涨；系统提示词塞进角色设定、业务规则、输出规范，越长越顾此失彼；同一个节点的提示词里既要它会查证又要它会修辞，两头都做不精。

拆成多个 agent，每个 agent 只背自己的工具子集和角色提示词，选工具的准确率和回答质量都会回来。代价是编排复杂度，谁先谁后、谁把结果交给谁、出了分歧听谁的，这些问题从模型脑子里搬到了图结构里。图结构是确定的、可测试的，这反而比把一切都押在单条系统提示词上更稳。

常见阵型就三种。监督者模式，一个主管节点看状态决定派谁，干完的把控制权交回来，适合流程有明确阶段的场景。群集模式，没有主管，agent 之间多对多互相移交，谁干完谁决定下一步，适合角色平权的协作。子图嵌套，把一张完整图当另一个图的节点，适合团队分层，主编只关心撰稿子图的产出，不管它内部几道工序。三种阵型底层的路由件是同一个，Command。

## 原理概念

多智能体在 LangGraph 里没有专用 API。一张图里放几个节点，每个节点背后是一个完整的 agent（模型加工具循环，或者干脆一个子图），节点间怎么连就是编排模式本身。条件边和 Command(goto)，在这里凑成主力。

子图是第一个关键词。compile 过的图能直接 add_node 进另一张图，前提是两边共享状态 schema，字段名对得上，数据就自动流动。想隐藏内部细节，子图可以配 input_schema 和 output_schema，对外只露两个窄接口，外层图看不到子图的全量状态。这一招在团队分层时特别有用，写手 agent 内部可能有查资料、列提纲、成稿三道工序，主编只需要看到最终稿。

控制权移交是第二个关键词。节点返回 `Command(goto="另一个节点", update={...})`，状态更新和控制权一起交出去，这就是 agent 之间的「干完了，该你了」。默认的 goto 只在本图内生效。子图里的节点想直接指挥父图的节点，得在 Command 里加 `graph=Command.PARENT`，更新随控制权一起上浮。

![](https://static.xiongneng.me/multi-agent-arch-20261017193600.png)

有两处行为和直觉对不上。一是注解，写 `Command[Literal["x"]]` 能让静态检查器校验 goto 目标，但校验范围是本图节点，跨图目标写了注解，编译都过不去。二是更新合并，PARENT 上浮的更新要并进父图状态，多个并行子图同时写一个共享 key 时必须有 Reducer 兜着，否则撞车报错；串行场景下单个子图上浮一个无 Reducer 的普通字段，实测是可以的。写并行分支时，共享字段一律配 Reducer，这条别赌。

## 核心用法

八个用法，示例在 `source/hello_multi.py`，全部纯函数不依赖模型。

### 1. 子图当节点

compile 过的图直接当节点，这是最朴素的组合方式。

```python
from langgraph.graph import END, START, StateGraph

sub = StateGraph(Base)
sub.add_node("upper", lambda s: {"text": s["text"].upper()})
sub.add_edge(START, "upper")
sub.add_edge("upper", END)
sub_app = sub.compile()

outer = StateGraph(Base)
outer.add_node("sub", sub_app)      # 子图直接进图
outer.add_node("wrap", lambda s: {"text": f"[{s['text']}]"})
```

```text
== 子图当节点 ==
  [HELLO]
```

文本从外层进子图变大写，出来再被 wrap 加上括号。两边共用同一个 Base schema，字段名 text 全程对得上，数据自动流。子图内部可以有自己的节点、边、条件路由，对外只是图里的一个点。

### 2. 输入输出收窄

子图配 input_schema 与 output_schema，对外只露两个窄接口。

```python
class In(TypedDict):
    raw: str

class Out(TypedDict):
    clean: str

class Overall(TypedDict):
    raw: str
    clean: str

sub = StateGraph(Overall, input_schema=In, output_schema=Out)
sub.add_node("clean", lambda s: {"clean": s["raw"].strip()})
```

```text
== 输入输出收窄 ==
  子图吃 raw 得 clean: 'padded'
```

子图内部看得到 Overall 全部字段，外人只能送进 In、拿回 Out。invoke 完返回值里只剩 clean 一个键，内部字段 raw 不会漏出去。给 agent 子图套上这一层，上层团队图的提示词里就只会出现该看的字段。

### 3. 同图移交

节点返回 Command，把控制权连同更新一起交出去。

```python
from langgraph.types import Command

def agent_a(state: Team) -> Command[Literal["b"]]:
    return Command(goto="b", update={"log": ["a 干完移交 b"]})

def agent_b(state: Team):
    return {"log": ["b 接手完成"], "current": "b"}
```

```text
== 同图移交 ==
  ['a 干完移交 b', 'b 接手完成']
```

a 返回的 Command 里 goto 指向 b，log 的更新随行。类型注解 `Command[Literal["b"]]` 声明本节点可能的去向，图编译时会校验 b 确实是本图节点，写错名字编译就报错。这个注解是可选的，但多智能体图里我建议都写，移交关系即文档。

### 4. 跨图导航

子图节点想直接指挥父图的节点，Command 里加 graph=Command.PARENT。

```python
def inner_cmd(state: Shared):
    # 没有 Command[Literal] 注解，目标在父图，注解过不了编译
    return Command(goto="outer_end", update={"items": ["child"]},
                   graph=Command.PARENT)

sub.add_node("cmd", inner_cmd)
outer.add_node("sub", sub.compile())
outer.add_node("outer_end", lambda s: {"items": ["outer"]})
```

```text
== 跨图导航 ==
  子图与父图的更新都算数: ['init', 'child', 'outer']
```

子图 cmd 的更新 child 上浮进父图，父图接着跑 outer_end 再补一笔 outer，加上初始的 init，三笔更新按顺序合并。注意这个节点函数没写 Command[Literal[...]] 注解，编译期校验的范围是本图节点表，父图里的目标过不了编译。跨图导航是双刃剑，它让子图能穿透封装直接插手外层流程，用多了子图就失去了独立可测试性，我的原则是子图只上浮数据，控制权尽量留在子图内部收尾。

### 5. PARENT 误用

graph=Command.PARENT 写在顶层图节点里，直接炸。

```python
def bad(state: Team):
    return Command(goto="b", graph=Command.PARENT)

g = StateGraph(Team)    # g 自己就是顶层，没有父图
g.add_node("a", bad)
```

```text
== PARENT 误用坑 ==
  顶层图抛 ParentCommand，没有父图接得住
  graph=Command.PARENT 只能写在子图节点里
```

LangGraph 抛的是专门的 ParentCommand 异常，语义很清楚，你要求上浮，可你在顶层。demo 里把异常接住打了一行说明，实际代码里这个错发生在运行期，图跑起来才炸。规避靠纪律，Command.PARENT 只出现在明确知道自己是子图的节点函数里。

### 6. Supervisor

主管节点决定下一个干活的，干完的把控制权交回主管。

```python
decisions = ["research", "write", "__end__"]

def supervisor(state: Flow):
    return Command(goto=decisions.pop(0))   # demo 用固定序列占位

def research(state: Flow):
    return {"steps": ["查资料"], "answer": "资料显示"}

def write(state: Flow):
    return {"steps": ["写稿"], "answer": state["answer"] + "，成稿"}

g.add_edge(START, "supervisor")
g.add_edge("supervisor", END)
for agent in ("research", "write"):
    g.add_edge(agent, "supervisor")         # 干完交回主管
```

```text
== Supervisor ==
  调度顺序 ['查资料', '写稿']，产出「资料显示，成稿」
```

结构上有三个特征。主管有到每个 agent 的出边（demo 里走 Command），每个 agent 有一条回主管的边，主管到 END 的边让流程有出口。demo 里主管用固定序列占位，真实实现换成模型的结构化输出，输入是当前状态摘要，输出是下一个 agent 的名字，路由机制一行不用改。主管决策读状态，agent 产出写状态，状态是唯一的沟通媒介，这是监督者模式最值得维持的纪律。

### 7. Swarm 网络

没有主管，agent 之间多对多，谁干完谁决定交给谁。

```python
def agent_z(state: Net) -> Command[Literal["y"]]:
    # z 判断自己不擅长写稿，移交给 y
    return Command(goto="y", update={"log": ["z 改不了稿，移交 y"]})

g.add_edge(START, "x")
g.add_edge("x", "z")
g.add_edge("z", "y")
g.add_edge("y", END)
```

```text
== Swarm 网络 ==
  · x 查完事实
  · z 改不了稿，移交 y
  · y 写完初稿
```

Swarm 的移交决策写在各 agent 自己的函数里，每个 agent 知道自己擅长什么、干完后该轮到谁。demo 里移交关系恰好能画成一条直线，真实场景里可以是网状的，x 干完可能去 y 也可能去 z。移交函数多了以后，公共的移交逻辑（比如把对话历史带上）可以抽成工厂函数，官方文档里叫 handoff tools 的做法就是把移交包装成工具给模型选。

### 8. 可视化

compile 过的图自带一张结构图，`get_graph().draw_mermaid()` 生成文本图。

```python
mermaid = g.compile().get_graph().draw_mermaid()
print(mermaid)
```

```text
graph TD;
	__start__ --> supervisor;
	research --> supervisor;
	supervisor -.-> research;
	supervisor -.-> writer;
```

静态边画实线，Command 这类动态路由画虚线，一眼看出控制权怎么流。零依赖零联网，把输出粘进任何支持 mermaid 的渲染器（GitHub 的 markdown 原生支持）就能看图。`draw_png` 要装 pygraphviz，`draw_mermaid_png` 要联网调用渲染服务，本地开发我固定用文本版。多智能体图的移交关系比单图复杂得多，写完先出图看一眼，能省掉大半接线错误。

## 完整案例

八个用法都过了一遍。收尾搭一个真的能协作的编辑部，`newsroom.py`，四个角色一张图。researcher 查资料，writer 写稿，reviewer 审稿带禁词检查，supervisor 看状态派活。

```bash
uv run python newsroom.py "LangGraph 是什么" --trace
```

```text
协作轨迹:
  [supervisor]
  [research] notes=['关于「LangGraph 是什么」的查证要点 3 条']
      · researcher: 收集资料
  [supervisor]
  [writer] draft=初稿：LangGraph 是什么 值得了解，理由来自资料。
      · writer: 出初稿
  [supervisor]
  [reviewer] final=定稿：初稿：LangGraph 是什么 值得了解，理由来自资料。
      · reviewer: 审校通过
  [supervisor]

协作步骤: researcher: 收集资料 -> writer: 出初稿 -> reviewer: 审校通过
成稿: 定稿：初稿：LangGraph 是什么 值得了解，理由来自资料。
```

supervisor 是纯状态驱动的决策器，代码短得像伪码。

```python
def supervisor(state: NewsroomState) -> Command[Literal[
        "research", "writer", "reviewer", "__end__"]]:
    if not state["notes"]:
        return Command(goto="research")
    if not state["draft"]:
        return Command(goto="writer")
    if state["review_note"]:
        return Command(goto="writer")       # 有批注说明被打回过，先改稿
    if not state["final"]:
        return Command(goto="reviewer")
    return Command(goto="__end__")
```

没有队列没有计数，每个判断对应一个阶段，notes 空就查资料，draft 空就写稿，有批注先改稿，没定稿先送审，全齐了收工。reviewer 是质检关口，草稿里出现禁词就地打回。

```python
def reviewer(state: NewsroomState) -> Command[Literal["supervisor"]]:
    draft = state["draft"]
    hit = next((w for w in BANNED if w in draft), None)
    if hit:
        return Command(update={
            "review_note": f"禁词「{hit}」请替换",
            "revisions": state["revisions"] + 1,
        }, goto="supervisor")
    return Command(update={"final": f"定稿：{draft}"}, goto="supervisor")
```

打回不直接跳 writer，控制权先回 supervisor，由它决定下一步。这是我在多 agent 图里守的一条规矩，agent 之间不做横向移交，全部经主管中转，移交关系只存在于 supervisor 一个函数里，改流程只改一处。试试喂一个故意带禁词的选题。

```bash
uv run python newsroom.py "写一篇绝对要用'赋能'这个词的稿子"
```

```text
协作步骤: researcher: 收集资料 -> writer: 出初稿 -> reviewer: 打回（发现禁词 赋能） -> writer: 按批注出第 2 版 -> reviewer: 审校通过
成稿: 定稿：初稿：写一篇绝对要用助力这个词的稿子 值得了解，理由来自资料。
```

reviewer 抓到禁词打回，writer 真的把那个词换成了助力，二审通过。批注、改稿、复审，整条返工链在 trace 里看得清清楚楚。writer 的改稿必须是真改（把禁词从稿子里换掉），如果只是把批注文字拼在稿子后面，禁词还在，reviewer 每轮都会打回，图就死循环了，我在开发这个案例时真踩过。 supervisor 里还有一个兜底，revisions 达到上限就 goto 结束并写转人工，多智能体系统里每个循环都必须有硬上限，这是从那次死循环里换来的纪律。

结构图用 demo 8 的方法打出来看一眼。

```text
graph TD;
	__start__ --> supervisor;
	supervisor -.-> research;
	supervisor -.-> writer;
	supervisor -.-> reviewer;
	supervisor -.-> __end__;
	research --> supervisor;
	writer --> supervisor;
	reviewer -.-> supervisor;
```

星型拓扑，四个 agent 围着 supervisor，虚线全是动态路由。测试在 `tests/test_hello_multi.py` 七条加 `tests/test_newsroom.py` 八条，子图合并、收窄接口、移交路由、ParentCommand、注解编译校验、打回改稿再审、返工上限都钉住了。加上前九篇的，全仓一百六十七条，全绿。

## 源码

[https://github.com/yidao620c/xneng-langchain-langgraph](https://github.com/yidao620c/xneng-langchain-langgraph)

## 参考

- [LangGraph 官方文档 - Multi-agent](https://docs.langchain.com/oss/python/langgraph/multi-agent)
- [LangGraph 官方文档 - Subgraphs](https://docs.langchain.com/oss/python/langgraph/subgraphs)
- [LangGraph 官方文档 - Command](https://docs.langchain.com/oss/python/langgraph/command)
- [LangGraph 官方文档 - How to visualize](https://docs.langchain.com/oss/python/langgraph/visualization)


---

> 作者: XiongNeng  
> URL: https://xiongneng.me/posts/ai/langchain/langchain-langgraph-10-multi-agent/