AI 电子伴侣
创建时间: 2026-04-03最后更新: 2026-08-04

1. 图一旦真的跑业务,失败就是日常

前面这一章一路写下来,图已经能分支、暂停、恢复、拆子图,也能组织多 Agent 协作。

但只要开始接真实业务,另一个问题就会立刻冒出来:

图不会一直顺着你的理想路径往下跑。

模型可能超时,工具可能报错,外部接口可能返回空值,人工审批可能迟迟不回来,某个并行节点也可能只成功了一半。开发阶段最容易忽略这一点,因为示例里的流程通常都很干净。可一旦放进真实系统,失败不是例外,而是常态。

所以这一篇要处理的,不是怎么把图写出来,而是图出问题以后,怎么别一下子整条断掉。

2. 先把“错误”分成几类看

这件事最好别一上来就想成一个大词。图里的失败来源,其实经常是不同层的。

有些失败来自节点内部。比如工具调用报错、模型返回异常、数据解析失败。

有些失败来自图的流程设计。比如本来应该等两个并行节点都回来,结果汇总节点过早开始执行;或者某个字段本来应该追加,结果写成了覆盖。

还有些失败来自外部系统。比如恢复时 thread 配错了、审批结果迟迟不回来、长期记忆没读到。

把这些都叫“错误”当然没问题,但处理方式不一样。
有的要在节点里兜住,有的要在状态里留痕,有的要给图留一条备用路径。

正在加载图示...

所以这一篇不会只讲 try/catch,而是把几种最常见的容错位置都走一遍。

3. 第一层兜底,通常就在节点里

最直接的失败,通常就发生在节点内部。比如某个节点里调了外部工具,结果工具超时了。

这时候最稳的第一步,往往不是急着改整张图,而是先让节点自己别炸掉。

safe-node.ts
01
import type { GraphNode } from '@langchain/langgraph'
02
03
const callSearchApi: GraphNode<typeof State> = async (state) => {
04
try {
05
// 这里模拟一个可能失败的外部调用
06
const result = await fetchSearchResult(state.query)
07
08
return {
09
searchResult: result,
10
status: 'search-success',
11
lastError: '',
12
}
13
} catch (error) {
14
// 节点先把失败写回状态,而不是直接把整条图打崩
15
return {
16
searchResult: '',
17
status: 'search-failed',
18
lastError: error instanceof Error ? error.message : '搜索接口调用失败',
19
}
20
}
21
}

这一层的关键,不是把错误吞掉,而是把错误留在状态里。

只要状态里还有 statuslastError 这些字段,后面的节点、条件边或者汇总节点就还有机会决定:是重试、降级,还是结束。

4. 只会 try/catch 还不够,后面还得知道怎么走

如果节点里只是 catch 住错误,但后面的图根本不知道失败了,那这层兜底就只做了一半。

更完整的写法,通常是:

先让节点把成功或失败写进状态,再让下一步根据状态决定往哪走。

error-routing.ts
01
import {
02
StateGraph,
03
StateSchema,
04
START,
05
END,
06
} from '@langchain/langgraph'
07
import type { GraphNode, ConditionalEdgeRouter } from '@langchain/langgraph'
08
import { z } from 'zod'
09
10
const State = new StateSchema({
11
query: z.string().default(''),
12
searchResult: z.string().default(''),
13
status: z.string().default('idle'),
14
lastError: z.string().default(''),
15
})
16
17
const callSearchApi: GraphNode<typeof State> = async (state) => {
18
try {
19
const result = await fetchSearchResult(state.query)
20
21
return {
22
searchResult: result,
23
status: 'search-success',
24
lastError: '',
25
}
26
} catch (error) {
27
return {
28
searchResult: '',
29
status: 'search-failed',
30
lastError: error instanceof Error ? error.message : '搜索接口调用失败',
31
}
32
}
33
}
34
35
const fallbackAnswer: GraphNode<typeof State> = (state) => {
36
return {
37
searchResult: `降级回答:当前外部搜索暂时不可用,错误信息是 ${state.lastError}`,
38
}
39
}
40
41
const shouldContinue: ConditionalEdgeRouter<typeof State, 'fallbackAnswer'> = (state) => {
42
if (state.status === 'search-failed') return 'fallbackAnswer'
43
return END
44
}
45
46
const graph = new StateGraph(State)
47
.addNode('callSearchApi', callSearchApi)
48
.addNode('fallbackAnswer', fallbackAnswer)
49
.addEdge(START, 'callSearchApi')
50
.addConditionalEdges('callSearchApi', shouldContinue, ['fallbackAnswer'])
51
.addEdge('fallbackAnswer', END)
52
.compile()

这里的变化很重要。错误不再只是节点内部的一次异常,而是变成了图可以继续读取的一种状态。
一旦图能读到它,就可以把失败也纳入流程控制里。

5. 如果可以重试,就把重试写进图里

有些失败不是立刻就该降级,而是值得再试一次。

比如模型超时、搜索接口临时抖一下、外部服务偶发返回空。碰到这种情况,最常见的做法就是在状态里加一个计数器,再留一条回边。

retry-loop.ts
01
import {
02
StateGraph,
03
StateSchema,
04
START,
05
END,
06
Command,
07
} from '@langchain/langgraph'
08
import type { GraphNode } from '@langchain/langgraph'
09
import { z } from 'zod'
10
11
const State = new StateSchema({
12
query: z.string().default(''),
13
result: z.string().default(''),
14
retryCount: z.number().default(0),
15
status: z.string().default('idle'),
16
lastError: z.string().default(''),
17
})
18
19
const callSearchApi: GraphNode<typeof State> = async (state) => {
20
try {
21
const result = await fetchSearchResult(state.query)
22
23
return {
24
result,
25
status: 'success',
26
lastError: '',
27
}
28
} catch (error) {
29
// 到这里说明这一次调用失败了,
30
// 下面要决定的是继续重试,还是直接切到降级节点
31
if (state.retryCount >= 2) {
32
return new Command({
33
update: {
34
status: 'failed',
35
lastError: error instanceof Error ? error.message : '搜索接口调用失败',
36
},
37
goto: 'fallbackAnswer',
38
})
39
}
40
41
return new Command({
42
update: {
43
retryCount: state.retryCount + 1,
44
status: 'retrying',
45
lastError: error instanceof Error ? error.message : '搜索接口调用失败',
46
},
47
goto: 'callSearchApi',
48
})
49
}
50
}
51
52
const fallbackAnswer: GraphNode<typeof State> = (state) => {
53
return {
54
result: `最终降级:重试 ${state.retryCount} 次后仍然失败。`,
55
}
56
}
57
58
const graph = new StateGraph(State)
59
.addNode('callSearchApi', callSearchApi, { ends: ['callSearchApi', 'fallbackAnswer'] })
60
.addNode('fallbackAnswer', fallbackAnswer)
61
.addEdge(START, 'callSearchApi')
62
.addEdge('fallbackAnswer', END)
63
.compile()

这一段和前面的差别在于,失败以后图没有立刻结束,而是明确地沿着另一条路径再试一次。

只要重试条件和退出条件写清楚,图里的失败处理就会比散落在各处的 if/else 更稳。

这里也能顺手把边界看清楚:

  • 值得重试的问题,通常是超时、抖动、偶发空结果
  • 不值得重试的问题,通常是参数错误、权限错误、输入本身就不成立

6. 有些错误不该重试,而该直接降级

不是所有失败都值得重试。

如果是权限错误、参数不合法、用户输入本身就不完整,那你再试三次通常也不会变好。这个时候更合适的办法往往不是回边,而是直接切到降级节点。

比如:

工具不可用时,先给一个保守回答;外部搜索失败时,退回本地知识库;审批节点拿不到结果时,先把状态标记成待人工处理。

这类节点和前面的 fallbackAnswer 一样,真正的价值不是「回答得多聪明」,而是图没有因此直接死掉。

7. 并行场景里,最怕的是一边成功一边失败

前面讲层级团队和并行协作时,图里最容易被忽略的一个现实问题是:

两个并行节点,不一定会一起成功。

比如研究员拿回来了竞品信息,但风险分析接口超时了。这个时候,汇总节点就会面临一个选择:

是拿到一半结果就继续收口,还是先标记失败,还是再补一次重试。

所以并行图里最好别只存「结果」,还要存每个分支自己的状态。

parallel-status.ts
1
const State = new StateSchema({
2
researchResult: z.string().default(''),
3
riskResult: z.string().default(''),
4
researchStatus: z.string().default('idle'),
5
riskStatus: z.string().default('idle'),
6
finalAnswer: z.string().default(''),
7
})

这样到了汇总节点,你才能看清楚到底是哪一边出了问题,而不是只看到「最终答案怎么少了一块」。

8. 人工介入,其实也是一种容错方式

前面那篇审批流已经讲过,interrupt() 能让图暂停,等外部决定以后再继续。

把它放到这一篇里看,会发现它其实也是一种容错。

有些问题,系统不应该自己硬判断。比如:

模型打分处在灰区,检索结果彼此冲突,或者工具执行结果看起来可疑。

这时候最稳的做法往往不是再让模型自己猜,而是停下来,把当前状态交给人工确认。
所以图里的容错不一定都是自动恢复,有时候恰恰是「别自动往下跑」。

9. 写这类图时最容易出的问题

最常见的问题,是只在日志里记错误,不把错误写进状态。这样节点内部看起来已经兜住了,图本身却根本不知道发生了什么,后面也就没法继续做判断。

第二个问题,是重试没有出口。只要失败就往回跳,却没有最大次数、没有降级路径,最后图会一直在原地打转。

还有一种情况,是把所有失败都归成一种。模型超时、参数错误、人工审批未完成、外部工具 500,全被塞进同一个 error 字段里。这样后面虽然看起来「有错误状态」,但实际上你还是不知道该怎么处理。

10. 总结

走到这里,这一章已经不只是讲怎么把图写出来了,也开始讲怎么让图在真实环境里活下来。

前面的状态、路由、暂停、子图、多 Agent,解决的是图怎么组织;这一篇的重点是图出问题以后怎么继续工作。节点内兜底、状态留痕、重试、降级、人工介入,这几层合在一起,图才会真正有韧性。

下一篇就进入最后一篇实战:用 LangGraph 重构 AI 伴侣管线。到那时,前面这些状态、控制流、长期记忆、多 Agent 和容错能力,都会落到同一条完整系统里。