AI 电子伴侣

创建时间: 2026-04-02最后更新: 2026-04-23作者: yangbo

1. 条件边有一个不方便的地方

前面几篇用条件边完成了大量路由逻辑。

ReAct Agent 的工具循环、质量门控的重试循环，核心都是 addConditionalEdges。

条件边能力上没问题，但用多了会发现一个不太方便的地方：路由逻辑和节点逻辑是分开的。

比如前面的重试循环：

separate-logic.ts

01import type { GraphNode, ConditionalEdgeRouter } from '@langchain/langgraph'
02
03// 验证节点：检查草稿质量，把结果写入状态
04const validate: GraphNode<typeof State> = (state) => {
05  if (state.draft.length >= 30) {
06    return { isValid: true, feedback: '' }
07  }
08  return { isValid: false, feedback: '草稿太短' }
09}
10
11// 路由函数：读取验证结果，决定下一步
12const checkResult: ConditionalEdgeRouter<typeof State, 'generate'> = (state) => {
13  if (state.isValid) return END
14  if (state.retryCount >= 3) return END
15  return 'generate'
16}

validate 节点已经知道草稿合不合格了，但它没法直接说「下一步去哪」。它只能先把 isValid 写入状态，然后等条件边的路由函数再去读这个字段。

做同一个决策，代码分散在两个地方。

当图变复杂以后，这种分离会越来越明显：

决策的上下文在节点里
但路由的执行在条件边里
两边要通过状态字段做间接通信

Command 就是用来解决这件事的：让节点直接说「我要更新这些状态，并且下一步走到那个节点」。

2. Command：在节点里直接决定下一步去哪

Command 是一个类。节点不再返回普通的状态对象，而是返回一个 Command 实例。

command-shape.ts

1import { Command } from '@langchain/langgraph'
2
3return new Command({
4  update: { isValid: true },  // 状态更新，和节点正常返回的对象一样
5  goto: 'output',             // 下一步走到哪个节点
6})

两个字段：

update：状态更新，和节点正常返回的状态对象一样
goto：下一步要走到的节点名

先看一个最简单的完整例子：

command-basic.ts

01import {
02  StateGraph,
03  StateSchema,
04  START,
05  END,
06  Command,
07} from '@langchain/langgraph'
08import type { GraphNode } from '@langchain/langgraph'
09import { z } from 'zod'
10
11const State = new StateSchema({
12  value: z.string().default(''),
13  route: z.string().default(''),
14})
15
16// nodeA 根据输入决定下一步去哪
17const nodeA: GraphNode<typeof State> = (state) => {
18  if (state.value === 'urgent') {
19    return new Command({
20      update: { route: 'fast' },
21      goto: 'fastTrack',
22    })
23  }
24  return new Command({
25    update: { route: 'normal' },
26    goto: 'normalTrack',
27  })
28}
29
30const fastTrack: GraphNode<typeof State> = (state) => {
31  return { value: `${state.value} → 快速通道处理完成` }
32}
33
34const normalTrack: GraphNode<typeof State> = (state) => {
35  return { value: `${state.value} → 普通通道处理完成` }
36}
37
38const graph = new StateGraph(State)
39  // ends 声明 nodeA 可能会跳到哪些节点
40  .addNode('nodeA', nodeA, { ends: ['fastTrack', 'normalTrack'] })
41  .addNode('fastTrack', fastTrack)
42  .addNode('normalTrack', normalTrack)
43  .addEdge(START, 'nodeA')
44  .addEdge('fastTrack', END)
45  .addEdge('normalTrack', END)
46  .compile()
47
48const result = await graph.invoke({ value: 'urgent' })
49console.log(result.value)
50// → urgent → 快速通道处理完成
51console.log(result.route)
52// → fast

这里最关键的一行：

addNode-ends.ts

1.addNode('nodeA', nodeA, { ends: ['fastTrack', 'normalTrack'] })

当节点用 Command 做路由时，这篇里的写法都会通过 ends 告诉 LangGraph 这个节点可能会跳到哪些目标。因为 goto 是运行时才确定的，编译阶段先把候选目标声明出来，图才能顺利通过结构检查。

和条件边对比

同样的逻辑，用条件边需要三样东西：

节点函数——把决策结果写入状态
路由函数——读状态字段，返回节点名
addConditionalEdges——把路由函数挂到边上

用 Command 只需要两样：

节点函数——在内部直接决定状态更新和下一步目标
addNode 的 ends 选项——声明可能的目标

逻辑更集中，代码也更短。

用 Command 重写重试循环

把第四篇的重试循环用 Command 改写。对比一下哪些代码可以省掉。

command-retry.ts

01import {
02  StateGraph,
03  StateSchema,
04  START,
05  END,
06  Command,
07} from '@langchain/langgraph'
08import type { GraphNode } from '@langchain/langgraph'
09import { z } from 'zod'
10
11const State = new StateSchema({
12  draft: z.string().default(''),
13  retryCount: z.number().default(0),
14  feedback: z.string().default(''),
15})
16
17// generate 不需要改，它只负责生成草稿
18const generate: GraphNode<typeof State> = (state) => {
19  if (state.retryCount === 0) {
20    return { draft: 'LangGraph 是一个框架。', retryCount: 1 }
21  }
22  return {
23    draft: `LangGraph 是一个基于图的 AI 工作流编排框架，支持状态管理、条件路由和持久化。（第 ${state.retryCount + 1} 版）`,
24    retryCount: state.retryCount + 1,
25  }
26}
27
28// validate 现在直接决定下一步去哪，不需要单独的路由函数了
29const validate: GraphNode<typeof State> = (state) => {
30  if (state.draft.length >= 30) {
31    return new Command({
32      update: { feedback: '' },
33      goto: 'output',
34    })
35  }
36
37  if (state.retryCount >= 3) {
38    return new Command({
39      update: { feedback: '已达最大重试次数' },
40      goto: 'output',
41    })
42  }
43
44  return new Command({
45    update: { feedback: `草稿太短（${state.draft.length} 字），至少需要 30 字` },
46    goto: 'generate',
47  })
48}
49
50const output: GraphNode<typeof State> = (state) => {
51  return { draft: `最终结果：${state.draft}` }
52}
53
54const graph = new StateGraph(State)
55  .addNode('generate', generate)
56  // ends 声明 validate 可能跳到的所有目标
57  .addNode('validate', validate, { ends: ['generate', 'output'] })
58  .addNode('output', output)
59  .addEdge(START, 'generate')
60  .addEdge('generate', 'validate')
61  .addEdge('output', END)
62  .compile()
63
64const result = await graph.invoke({})
65console.log(result.draft)
66// → 最终结果：LangGraph 是一个基于图的 AI 工作流编排框架...

和第四篇对比，省掉了：

isValid 状态字段——不需要了，validate 内部直接判断完就跳转
checkResult 路由函数——不需要了，路由逻辑在 validate 节点里
addConditionalEdges——不需要了，换成 addNode 的 ends

注意 generate → validate 之间还是用的普通边。因为 generate 执行完一定走 validate，没有条件分支。只有 validate 的下一步不确定，所以只有它用了 Command。

什么时候用 Command，什么时候用条件边

不需要一刀切，按这个思路选：

路由只是简单地读一个布尔值或枚举字段 → 条件边就够了
路由逻辑和节点逻辑紧密绑定，「算完就知道该走哪条路」 → Command 更自然
需要在同一步里既更新状态又决定路由 → 用 Command 最顺手

两种方式可以在同一张图里混用。

3. Send：让同一个节点并行跑多份

Command 解决的是路由问题。Send 解决的是另一个问题：一个节点需要并行跑多份，每份拿到不同的输入。

最典型的场景是 map-reduce：

先把一个任务拆成 N 份（map）
每份并行处理
最后把结果汇总（reduce）

比如用户给了三个主题，你想同时为三个主题生成摘要，而不是排队生成。

Send 的构造函数只有两个参数：

send-shape.ts

1import { Send } from '@langchain/langgraph'
2
3new Send('targetNode', { topics: ['React'] })
4//        ↑ 目标节点名    ↑ 传给这个节点的状态

每个 Send 实例代表「调用一次目标节点，传入这份特定的状态」。返回多个 Send，就是并行调用多次。

在条件边里使用 Send

Send 最基础的用法是在条件边的路由函数里返回一个 Send 数组：

send-basic.ts

01import {
02  StateGraph,
03  StateSchema,
04  ReducedValue,
05  Send,
06  START,
07  END,
08} from '@langchain/langgraph'
09import type { GraphNode } from '@langchain/langgraph'
10import { z } from 'zod'
11
12const appendStrings = (current: string[], update: string[]) => {
13  return [...current, ...update]
14}
15
16const State = new StateSchema({
17  topics: z.array(z.string()).default([]),
18  // 多个并行节点都会写 summaries，所以必须用 reducer 追加
19  summaries: new ReducedValue(
20    z.array(z.string()).default([]),
21    { reducer: appendStrings },
22  ),
23})
24
25// 这个节点会被并行调用多次，每次拿到一个不同的 topic
26const summarize: GraphNode<typeof State> = (state) => {
27  const topic = state.topics[0]
28  return {
29    summaries: [`「${topic}」的摘要：这是一段关于 ${topic} 的分析...`],
30  }
31}
32
33// 路由函数返回 Send 数组，触发并行扇出（fan-out）
34// 这里的“扇出”是分布式系统里的常用说法，意思是把一份输入拆成多份子任务，同时分发给多个并行执行的分支
35const fanOut = (state: typeof State.type) => {
36  return state.topics.map(
37    (topic) => new Send('summarize', { topics: [topic] })
38  )
39}
40
41const graph = new StateGraph(State)
42  .addNode('summarize', summarize)
43  .addConditionalEdges(START, fanOut)
44  .addEdge('summarize', END)
45  .compile()
46
47const result = await graph.invoke({
48  topics: ['React', 'LangGraph', 'Zustand'],
49})
50
51console.log(result.summaries)
52// [
53//   '「React」的摘要：这是一段关于 React 的分析...',
54//   '「LangGraph」的摘要：这是一段关于 LangGraph 的分析...',
55//   '「Zustand」的摘要：这是一段关于 Zustand 的分析...',
56// ]

这里有三个要点。

第一，fanOut 不是返回一个节点名，而是返回一个 Send 数组。每个 Send 都指向同一个 summarize 节点，但传入的状态不同。LangGraph 会把它们并行执行。

第二，summaries 字段必须声明 reducer。因为三个并行节点都在往这个字段里写，如果没有 reducer，结果会互相覆盖，只留下最后一个。

第三，Send 的第二个参数是传给目标节点的状态。这里传的是 { topics: [topic] }，只包含一个 topic。目标节点 summarize 拿到的 state.topics 就只有一个元素。

在 Command 的 goto 里使用 Send

Send 不只能在条件边里用。它也可以放在 Command 的 goto 字段里，这样就能一步完成状态更新 + 并行扇出（fan-out，也就是把一份任务拆成多份后同时派发出去）：

command-send.ts

01import {
02  StateGraph,
03  StateSchema,
04  ReducedValue,
05  Command,
06  Send,
07  START,
08  END,
09} from '@langchain/langgraph'
10import type { GraphNode } from '@langchain/langgraph'
11import { z } from 'zod'
12
13const appendStrings = (current: string[], update: string[]) => {
14  return [...current, ...update]
15}
16
17const State = new StateSchema({
18  topics: z.array(z.string()).default([]),
19  summaries: new ReducedValue(
20    z.array(z.string()).default([]),
21    { reducer: appendStrings },
22  ),
23  status: z.string().default(''),
24})
25
26// dispatcher 既更新状态，又触发并行扇出
27const dispatcher: GraphNode<typeof State> = (state) => {
28  return new Command({
29    update: { status: `正在处理 ${state.topics.length} 个主题` },
30    goto: state.topics.map(
31      (topic) => new Send('summarize', { topics: [topic] })
32    ),
33  })
34}
35
36const summarize: GraphNode<typeof State> = (state) => {
37  const topic = state.topics[0]
38  return {
39    summaries: [`${topic} 分析完成`],
40  }
41}
42
43const graph = new StateGraph(State)
44  .addNode('dispatcher', dispatcher, { ends: ['summarize'] })
45  .addNode('summarize', summarize)
46  .addEdge(START, 'dispatcher')
47  .addEdge('summarize', END)
48  .compile()
49
50const result = await graph.invoke({
51  topics: ['性能优化', '错误处理', '状态管理'],
52})
53
54console.log(result.status)
55// → 正在处理 3 个主题
56console.log(result.summaries)
57// → ['性能优化 分析完成', '错误处理 分析完成', '状态管理 分析完成']

这就是 Command + Send 的组合：一步完成状态更新 + 并行扇出。

4. 实战：多城市天气并行查询

把前面的能力综合起来，做一个更接近实际的例子。

场景：用户问多个城市的天气 → 解析出城市列表 → 并行查询每个城市 → 汇总回复。

parallel-weather.ts

01import {
02  StateGraph,
03  StateSchema,
04  MessagesValue,
05  ReducedValue,
06  Command,
07  Send,
08  START,
09  END,
10} from '@langchain/langgraph'
11import type { GraphNode } from '@langchain/langgraph'
12import { z } from 'zod'
13
14const appendStrings = (current: string[], update: string[]) => {
15  return [...current, ...update]
16}
17
18const State = new StateSchema({
19  messages: MessagesValue,
20  cities: z.array(z.string()).default([]),
21  weatherResults: new ReducedValue(
22    z.array(z.string()).default([]),
23    { reducer: appendStrings },
24  ),
25})
26
27// 1. 解析节点：从用户消息里提取城市列表，然后扇出到查询节点
28const parseCities: GraphNode<typeof State> = (state) => {
29  const userMsg = state.messages.at(-1)?.content?.toString() ?? ''
30
31  // 实际项目里这里会用 LLM 做实体提取，这里简化处理
32  const knownCities = ['北京', '上海', '深圳', '广州', '杭州']
33  const found = knownCities.filter((c) => userMsg.includes(c))
34
35  if (found.length === 0) {
36    // 没找到城市，直接结束
37    return new Command({
38      update: {
39        messages: [{ role: 'assistant', content: '没有识别到城市名称。' }],
40      },
41      goto: END,
42    })
43  }
44
45  // 找到了城市，更新城市列表 + 并行扇出查询
46  return new Command({
47    update: { cities: found },
48    goto: found.map((city) => new Send('queryWeather', { cities: [city] })),
49  })
50}
51
52// 2. 查询节点：查单个城市的天气（会被并行调用多次）
53const queryWeather: GraphNode<typeof State> = (state) => {
54  const city = state.cities[0]
55  const data: Record<string, string> = {
56    北京: '晴，12-25°C',
57    上海: '小雨，17-22°C',
58    深圳: '多云，26-30°C',
59    广州: '阵雨，24-29°C',
60    杭州: '晴转多云，15-27°C',
61  }
62  return {
63    weatherResults: [`${city}：${data[city] ?? '暂无数据'}`],
64  }
65}
66
67// 3. 汇总节点：把所有天气结果组织成最终回复
68const formatReport: GraphNode<typeof State> = (state) => {
69  const report = state.weatherResults.join('\n')
70  return {
71    messages: [{
72      role: 'assistant',
73      content: `查询到 ${state.cities.length} 个城市的天气：\n\n${report}`,
74    }],
75  }
76}
77
78const graph = new StateGraph(State)
79  .addNode('parseCities', parseCities, { ends: ['queryWeather'] })
80  .addNode('queryWeather', queryWeather)
81  .addNode('formatReport', formatReport)
82  .addEdge(START, 'parseCities')
83  .addEdge('queryWeather', 'formatReport')
84  .addEdge('formatReport', END)
85  .compile()
86
87const result = await graph.invoke({
88  messages: [{ role: 'user', content: '北京、上海和深圳今天天气怎么样？' }],
89})
90
91console.log(result.messages.at(-1)?.content)
92// → 查询到 3 个城市的天气：
93// →
94// → 北京：晴，12-25°C
95// → 上海：小雨，17-22°C
96// → 深圳：多云，26-30°C

这张图的执行流程：

parseCities 从用户消息里提取出「北京、上海、深圳」三个城市
用 Command 同时完成两件事：把城市列表写入状态 + 用 Send 扇出三个并行查询
三个 queryWeather 并行执行，各自查一个城市，结果通过 reducer 汇总到 weatherResults
formatReport 拿到全部结果，拼成最终回复

如果没有 Send，你只能顺序查三次；如果没有 Command，你需要额外的条件边和状态字段来完成路由。两者组合在一起，代码更紧凑，语义也更清晰。

5. 使用注意

Command 的 goto 会追加一条动态出边

如果你同时给一个节点加了普通边和 Command 路由：

gotcha-edge.ts

1graph
2  .addEdge('nodeA', 'nodeB')                          // 普通边
3  .addNode('nodeA', nodeA, { ends: ['nodeC'] })       // nodeA 返回 Command({ goto: 'nodeC' })

那 nodeB 和 nodeC 都会执行。Command 的 goto 不会取消已有的普通边，它只是额外追加了一条动态边。所以，如果你打算把某个节点的出路完全交给 Command，就不要再给它保留普通出边。

ends 要一起写上

用 Command 做路由时，addNode 的 ends 最好和节点一起声明。这样一来，图编译时能先知道这个节点可能跳到哪些目标；二来读代码时也能一眼看清这个节点的出口范围。

ends 的作用是告诉 LangGraph：这个节点可能会动态跳到哪些目标。图编译时需要这个信息来验证结构完整性和生成可视化。

Send 的并行结果需要 reducer

多个 Send 并行写同一个状态字段时，必须给这个字段声明 reducer（第三篇讲过的 ReducedValue）。没有 reducer 的话，多个并行写入会互相覆盖，最后只留下其中一个的值。

Command 和条件边可以混用

一张图里可以同时有用 Command 路由的节点和用条件边路由的节点。选哪种取决于那个节点的具体场景，不需要全图统一。

6. 总结

这一篇往前多走了一步：前面的条件边，解决的是“根据状态决定下一步去哪”；这里的 Command，解决的是“节点自己在更新状态的同时，直接决定下一步去哪”；Send 则是把“一份任务拆成多份并行去跑”这件事补上。

所以它们的关系不是替代，而是分工更细：

简单条件分叉，条件边就够了
节点里已经算完了，也知道下一步去哪，用 Command
同一个节点要并行跑很多份，用 Send
既要改状态又要并行扇出，就把 Command 和 Send 组合起来

下一篇讲 interrupt：暂停与恢复机制。到那时，图不只是“继续往下走”，还会在运行中停下来，等外部输入后再继续。