做一个多智能体客服系统最怕的不是模型答不好而是整个流程跑到一半直接崩了。我最近用 LangGraph 搭了一套具备自愈能力的客服架构这里把完整思路、核心代码和踩坑记录都放出来希望能给正在做 Agent 编排或者客服系统的朋友一些参考。所谓“可自愈”不是加个 try-except 那么简单。它要涵盖网络抖动时自动重试、第三方接口超时后优雅降级、某个智能体崩溃时不影响整条链路、崩溃后能从最近的检查点恢复现场以及会话快照能支撑人工无缝接管。这套东西用传统的状态机或者裸的 LLM 循环来写会非常痛苦而 LangGraph 恰好把状态、节点、边和持久化都抽象好了拿来改造一轮效果相当明显。1. 为什么选 LangGraph 做客服系统从编排痛点说起1.1 传统客服系统的死板与自愈的缺失我最早做客服机器人用的是“意图识别- 对话管理 - 回答生成”这种线性流水线。线上跑起来后问题很快就暴露了。接口偶尔超时整个流程就得用户重新发起用户中途换了个说法对话状态就乱了更头疼的是当我要接入两个辅助模型一个做大模型推理、一个做敏感词审核两者必须串行只要其中一个挂了整个客服就“呆滞”了。这种架构天然缺乏自愈能力因为它没有一个统一的调度中心去感知每个环节的健康状态。就算你写一堆监控埋点那也只是事后报警做不到“自动恢复正常”。客服场景最不能接受的就是用户说到一半机器人突然不理人。1.2 LangGraph 的图编排能力如何解决这些痛点LangGraph 的核心思想是把智能体流程抽象成一张图节点是执行单元边是流转条件而整个流程的状态统一放在一个 State 对象里。这跟我以前写的“类状态机”完全不是一个量级。它有几个非常契合自愈设计的特性一个流程实例可以被打断也可以从断点继续LangGraph 的持久化 checkpoint 天然支持“会话现场恢复”。节点之间通过条件边流转你可以把重试、降级、熔断当作图上的路径来设计而不是散落在代码里的分支。多智能体之间也是通过图来共享状态协调逻辑可视化程度高出了问题直接把图打印出来就能定位不用瞎猜。我选择 LangGraph 还有一个现实理由它不需要自己维护一个复杂的状态机轮子。队列、重试、超时这些中间件能力我可以嵌入到图节点或边里但图的整体执行顺序和分支策略完全交给 LangGraph 处理这本身就减少了一堆自我实现的 bug。注意我这里说的“可自愈”并不是玄学。LangGraph 提供了持久化与断点能力而超时重试、熔断降级这类策略需要在节点的内部代码中配合实现。两者结合才是完整的自愈闭环。2. 整体架构设计可自愈的多智能体拓扑2.1 智能体划分意图识别、业务处理、服务保障、监控仲裁在设计之初我参考了团队里一个稳定的客服小组的分工把整个在线客服拆成四个智能体每个智能体只做一类事。第一个是意图识别智能体负责判断用户到底想干什么。它的输入是当前消息和整个会话历史输出是用户意图、实体槽位和优先级。这个智能体我用了独立的轻量模型因为它的实时性要求很高差不多一秒内要出结果否则后面所有环节都得等着。第二个是业务处理智能体这是核心执行者。它根据意图识别结果决定调用哪几个业务工具。比如查订单、改地址、算退款金额都由它负责。它需要做得非常细因为它面对的是直接业务逻辑任何一步错了用户都会感知到。第三个是服务保障智能体专门负责兜底。当业务处理智能体抛异常、工具调用失败、或者多次重试仍然拿不到有效结果时服务保障智能体启动。它做的事包括换一种表述重新问大模型、降级到静态 FAQ 问答、生成“稍后人工介入”的工单。它的存在让整个流程形成“自愈闭环”正常情况下它不执行异常时它自动被激活。第四个是监控仲裁智能体它其实不直接对用户说话它只做两件事评估上一个回复是否靠谱以及检查整个链路是否还有必要继续执行。如果同一个分支重试超过三次它会强制切换到降级路径如果识别到用户情绪非常激动它会直接拉起人工客服并且把当前对话的摘要和状态包好交给人工。这个智能体的核心价值就是决策权防止系统陷入无意义的内部循环。四个智能体之间的协作我用了“路由器-执行器-兜底”的模式。意图识别先走业务处理接着走只要出错服务保障智能体就介入监控仲裁智能体全程看着。这套拓扑最大的优点是故障被限制在局部某一个智能体挂了边上会有另一条路绕过去。2.2 自愈机制的分层设计消息重试、节点熔断、记忆快照、会话迁移我把自愈分成四层每一层解决的问题都不一样。第一层是消息重试解决的是临时性故障。比如某个 HTTP 工具调用超时网络抖动了一下这种故障重试一两次就能过去。LangGraph 的节点内我封装了一个带指数退避的重试包装器每当我们调用外部接口时它会根据失败次数动态调整等待时间。重试次数上限一般是 3 次超过上限就认为这个节点的“当前执行”失败。第二层是节点熔断解决的是持续性故障。如果某个智能体连续失败超过预设阈值我设的是 5 次监控仲裁智能体会把该节点熔断直接走另一个路线。比如业务处理智能体依赖的订单系统宕机了此时熔断后系统进入“降级响应模式”不再反复请求订单系统而是直接告诉用户“系统繁忙请稍后”同时生成工单。第三层是记忆快照解决的是状态丢失问题。LangGraph 的持久化检查点在每个节点执行前后都会保存当前状态包括消息历史、槽位、中间结果。一旦实例崩溃重启后可以从最近的检查点恢复。这一步很关键因为我们客服系统通常是无状态微服务部署真正有状态的是对话流程本身。第四层是会话迁移解决的是流程中断后的接管问题。当一个流程实例始终无法恢复或者用户要求转人工时监控仲裁智能体把当前对话的快照、用户槽位、尝试过的操作清单全部打包交给人工客服平台。人工客服直接看到完整的上下文不需要用户重新解释一遍。2.3 关键数据结构状态、检查点、事件流LangGraph 里所有智能体共享一个 State 对象这个对象的结构直接决定了自愈能否顺利实现。我定义的状态大致包含这样几部分messages一个消息列表存储用户和 AI 的所有对话记录这是一个累积字段不会被覆盖。slots一个字典存储用户实体的临时槽位比如用户 ID、订单号、退货原因。current_intent当前意图由意图识别智能体写入。history_steps一个列表记录每一个已经执行过的节点名称和结果状态用于追踪流程路径。fault_info一个字典记录最近发生的异常类型、重试次数、熔断标记这是自愈逻辑的核心输入。checkpoint_ids用于关联不同的检查点根据每个节点执行完后的快照 ID 进行回放。这里有个很关键的设计细节检查点不是整个状态的拷贝而是按变更记录存储的。LangGraph 的状态更新是叠加式的你能看到一个节点执行后哪些字段变化了。这对自愈非常重要因为当流程恢复到某个检查点时你不需要重新构造所有数据只需要回放变更。另外它还带图形记录功能定位问题时能看到还原后的执行路径。事件流方面我用 LangGraph 的 stream 模式处理实时输出。每个节点执行时都向外发出“节点开始、节点结束、节点失败”这样的事件这些事件是自愈监控仲裁智能体的数据来源。通过订阅这些事件能在节点发生异常后的第一时间触发重试逻辑。关于状态合并有一点特别容易踩坑。LangGraph 的 state update 默认是“覆盖整个字段”。如果你在一个节点里想给messages追加数据必须明确使用 Annotated 类型并定义 reducer 函数否则你辛辛苦苦维护的消息历史会被下一次节点更新覆盖掉。这个问题在并行执行分支时尤其致命我后面会展开说。3. 核心细节解析LangGraph 中的自愈实现3.1 图状态设计与条件边的容错语义状态是图流转的基础。我先定义了一个 dataclass 类型的客服状态然后通过 LangGraph 的StateGraph把四个智能体挂上去。先看状态定义from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class KumatoAgentState(TypedDict): messages: Annotated[list, add_messages] current_intent: str slots: dict fault_info: dict history_steps: list final_answer: str handover_required: bool这里有个细节messages字段用add_messages作为 reducer它的作用是自动把新消息追加到历史列表里而不是覆盖。fault_info字段则直接覆盖因为每次故障都是一个新状态。接下去定义节点间的条件边。条件边是自愈分支的灵魂它决定下一步走哪条路。LangGraph 天然支持这种“异常分支”def route_by_outcome(state: KumatoAgentState) - str: fault state[fault_info] retries fault.get(retries, 0) is_failed fault.get(is_failed, False) if is_failed and retries 3: return retry elif is_failed and retries 3: return fallback elif state.get(handover_required): return handover else: return continue这个自定义函数就是条件边的核心LangGraph 会根据它的返回值把执行流送到不同的节点。我把重试上限设为 3超过 3 次就进入降级节点而不是顺着主流程继续跑。这就是“容错误义”落到代码上的体现它不是简单的 if-else而是图级别的路径重规划。3.2 节点装饰器与重试策略代码与实现逻辑我给每个节点都包了一个“自愈装饰器”这样每个节点的内部不用写一堆 try-except只管实现业务逻辑。装饰器会统一处理异常捕获、重试等待、故障写入和熔断标记。import time import functools def self_healing_node(step_name, max_retries3): def decorator(func): functools.wraps(func) def wrapper(state: KumatoAgentState) - KumatoAgentState: fault_info state.get(fault_info, {}) retries fault_info.get(retries, 0) if retries max_retries: return {fault_info: {**fault_info, is_failed: True, fallback: True}} try: result func(state) return {**result, fault_info: {**result.get(fault_info, {}), is_failed: False}} except Exception as e: retries 1 wait_time 2 ** retries # 指数退避 time.sleep(wait_time) state[history_steps].append(f{step_name}_retry_{retries}) return {fault_info: {**fault_info, retries: retries, last_error: str(e), is_failed: True}} return wrapper return decorator这里有一个细节要注意state是 TypedDict它是不可变对象还是可变对象取决于你对节点的定义。如果你直接修改state[history_steps]那这个修改必须通过返回值传出去否则 LangGraph 不会把它写入新的检查点。我的实际项目里统一在装饰器内更新state并把新状态返回保证下一次执行能看到最新的重试次数。装饰器里的等待时间使用指数退避第一次失败等 2 秒第二次等 4 秒第三次等 8 秒。如果连续重试三次仍然失败就不再浪费时间直接设置fallback标记让路由把执行引导到降级节点。这个装饰器还有设计上的妙处它可以把每个节点的错误都集中上报到监控仲裁智能体。fault_info虽然是覆盖式更新但每次更新时保留了 last_error 和 retries这些信息可以为后面的根因分析提供依据。3.3 检查点与断点恢复从内存存储到持久化LangGraph 默认的执行模式是跑完整个图就结束如果你想实现“自愈恢复”必须开启持久化。我用的是SQLiteSaver把每个检查点写到本地磁盘文件这样即使整个服务重启也能恢复到之前的现场。from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(checkpoints/chatbot.sqlite) as saver: graph builder.compile(checkpointersaver)我在跑起来后发现一个误解不是每个节点自动保存检查点而是你需要在编译时指定interrupt_before或interrupt_after来告诉图在哪里保存检查点。我把主要的业务处理节点后面都加了interrupt_after这样每一次业务操作完成后都有一个快照可以回退graph builder.compile( checkpointersaver, interrupt_after[intent_agent, business_agent, fault_triage_node] )这样的好处是如果用户在业务处理节点完成后突然中断网络下一次对话可以通过thread_id从该检查点继续而不是从头开始跑两个智能体。自愈能力里最核心的“恢复现场”靠的就是这个机制。恢复现场有个技巧。LangGraph 的update_state可以修改某个检查点之后的状态但尽量别在恢复时直接改历史消息否则容易造成上下文不一致。我在做会话迁移时采用的是“快照加摘要”的方式把状态里的完整消息和槽位交给人工客服而不是让 AI 重新生成。3.4 超时和熔断如何接到图上客服系统最怕外部接口慢。LangGraph 本身不内置超时所以在节点装饰器内部我给所有外部调用统一套了超时控制。比如调用订单系统我用的是带超时的 HTTP 客户端from httpx import Timeout, AsyncClient async def call_order_api(order_id): timeout Timeout(5.0) async with AsyncClient(timeouttimeout) as client: resp await client.post(https://order-api.example.com/query, json{id: order_id}) resp.raise_for_status() return resp.json()如果在 5 秒内没有响应就会抛出超时异常这个异常会被自愈装饰器捕获然后触发重试。在重试几次仍然超时后fault_info里的fallback会被置为 True路由条件边就会把请求引到服务保障智能体。熔断逻辑放在监控仲裁智能体里。它不是靠业务节点触发而是靠监控事件驱动。节点每执行一次就会把结果发送到监控通道监控仲裁智能体统计最近 5 次事件。如果失败率超过 60%它就把对应节点标记为“熔断”后续请求直接走备用路径。这里我用的是一个很简单的滑动窗口计数器from collections import deque failures deque(maxlen10) def record_node_result(success: bool): failures.append(success) if len(failures) failures.maxlen: fail_count sum(1 for x in failures if not x) if fail_count / failures.maxlen 0.6: return circuit_open return circuit_closed熔断不是一个节点内部的私有逻辑它需要被多个分支共享。所以我把这个状态写进了全局FaultManager单例里监控仲裁智能体和业务节点各拿一份引用。每次业务节点执行前调一下检查如果熔断已打开就直接不执行写降级结果。4. 实操过程从零搭一个可自愈的客服 Demo4.1 环境准备与项目骨架首先说明环境。我用的是 Python 3.11安装了langgraph、langchain-openai、sqlite-utils和httpx。这里不指定具体大模型供应商因为 LangGraph 抽象得很好换几个环境变量就能替换成你喜欢的模型。我建议你先把环境搭好至少保证下面的 demo 能跑通再去在意模型效果。项目骨架大致是这样customer_service/ ├── main.py ├── state.py ├── agents/ │ ├── intent_agent.py │ ├── business_agent.py │ ├── service_guard.py │ └── monitor_arbiter.py ├── tools/ │ ├── order.py │ ├── refund.py │ └── faq.py ├── checkpoints/ │ └── chatbot.sqlite └── requirements.txt我按“状态 - 智能体 - 节点 - 图”四层组织代码。state.py 只放 TypedDict 和 reduceragents 负责具体的智能体逻辑tools 放工具函数main.py 里把图构组装起来。4.2 定义四个智能体节点与工具先看intent_agent.py它接收用户消息用轻量模型判断意图。在实际项目里它还会维护一个“意图置信度”字段。from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI def intent_agent_node(state): llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 这里其实应该结合历史但演示里简化成只看最新消息 messages [ SystemMessage(content你是意图识别引擎。只输出JSON{intent:query_order|apply_refund|talk_human|unknown,slots:{}}), HumanMessage(contentf用户消息{state[messages][-1].content}) ] result llm.invoke(messages) # 这里省略解析逻辑假设拿到了 dict return {current_intent: intent, slots: slots}接下来是business_agent.py它根据current_intent执行工具。这里有一个节点内部的自愈分支如果工具执行失败会在返回状态里写入fault_info让图重新路由。self_healing_node(business_agent) def business_agent_node(state): intent state[current_intent] if intent query_order: order_id state[slots].get(order_id) order_info call_order_api(order_id) return {final_answer: f您的订单{order_info[id]}状态是{order_info[status]}预计送达{order_info[eta]}} elif intent apply_refund: return {final_answer: 退款申请已提交预计1-3个工作日原路退回} return {final_answer: 我还没有学会这个操作为您转接人工}注意call_order_api很可能会超时或返回异常但这里不需要写 try-except因为装饰器会统一捕获。这就是自愈逻辑与业务逻辑分离的好处。service_guard.py是服务保障智能体它有三种模式如果目标是“降级”就返回 FAQ 兜底答案如果目标是“重试”就把原状态原样返回等待路由重新走业务节点如果目标是“转人工”就把会话摘要打包。monitor_arbiter.py是监控仲裁智能体它会根据fault_info和全局熔断器的状态决定当前节点是“继续执行”还是“跳过”。它通常不作为条件边返回而是作为图中的一个旁路节点实时监听状态变化。4.3 配置自愈中间件与状态持久化在 main.py 里我把四个节点都挂到图上并定义了条件边。from langgraph.graph import StateGraph builder StateGraph(KumatoAgentState) builder.add_node(intent, intent_agent_node) builder.add_node(business, business_agent_node) builder.add_node(guard, service_guard_node) builder.add_node(monitor, monitor_arbiter_node) builder.set_entry_point(intent) builder.add_edge(intent, business) builder.add_conditional_edges( business, route_by_outcome, { retry: business, fallback: guard, handover: guard, continue: monitor, } ) builder.add_edge(guard, monitor) builder.add_edge(monitor, monitor) # 监控节点可以自查/循环这里我刻意让business和guard之间形成循环。业务节点连续失败后route_by_outcome返回fallback走到guardguard通过修改fault_info把重试计数清零再走回 business 重新尝试。这种方式比直接让业务节点内部死循环要安全因为每次进入节点之间LangGraph 都会生成新的检查点崩溃了也有地方回退。编译时关键步骤是接入检查点与中断点。我在业务节点执行前设置了中断这样如果想完全取消当前节点执行图会在中断点之前停下等待下一次指令。builder.compile(checkpointersaver, interrupt_before[business])4.4 测试与验证注入故障看它怎么活我写了一个本地测试脚本用随机注入故障的方式验证自愈能力。比如把订单接口 URL 改成不存在的地址然后模拟三次正常对话、一次超时对话。def simulate_fault(fault_rate0.3): # 用随机数决定是否让 call_order_api 抛出超时异常 pass graph builder.compile(checkpointersaver) config {configurable: {thread_id: test-user-001}} result graph.invoke({ messages: [{role: user, content: 帮我查订单 12345 到哪了}] }, config) print(result[final_answer])实测下来当订单接口一开始就失败时第一次进入业务节点会抛异常装饰器捕获并记录 retries1条件边返回 retry图回到 business 节点前中断。第二次尝试若恢复成功则业务流程继续最后输出正常答案。如果连续三次失败第四次进入 business 前条件边直接返回 fallback走到 guard 节点输出一条“系统正在升级请稍后重试”的降级话术同时生成一个工单 ID 存储到状态里。这里最重要的观察是自愈不是靠某个节点单打独斗而是靠图的路由 检查点 状态更新三者协作。单靠装饰器重试最多只能让当前节点原地重试无法在全局层面切换路径。有了图每次重试的决策都显式地体现在执行轨迹里运维同学看到图就能明白系统刚才经历了什么。5. 常见问题与排查技巧实录5.1 循环死锁条件边永远回不到终止槽我在开发中遇到的第一个麻烦是业务节点和保障节点之间出现了无限循环。当时我设计的逻辑是业务失败后走到保障节点保障节点尝试“修正”后回到业务。但修正时没有把fault_info里的重试次数清零结果就是业务节点失败 - 保障节点修正 - 业务又失败 - 又修正无限循环。图上的执行路径一直转圈代价是每次循环都会记录一个新的检查点磁盘空间快速膨胀。解决方法是增加“循环次数上限”字段。在业务节点的入口处先检查history_steps里有多少次重复的循环标记如果超过 3 次直接强制返回降级话术并终止图。loop_count sum(1 for step in state[history_steps] if step.startswith(business_retry)) if loop_count 3: return {final_answer: 连续尝试失败已为您创建工单客服会尽快联系您。, handover_required: True}另外还可以在条件边里加一个“没有下一步动作就 End”的判断。如果返回的路由没有对应边或者不满足任何分支LangGraph 会报错但这个报错会被误认为环境问题。我建议每次修改图配置后把图结构打印出来确认从每个节点出发都能经过有限步到达终止节点。5.2 检查点数据冲突并行分支状态覆盖LangGraph 支持图内并行执行多个节点但状态更新是有冲突风险的。比如一个分支在处理退款申请时另一个分支在处理工单生成两者同时更新slots字典默认是直接覆盖掉对方的值。这会导致一个分支看不到另一个分支写入的槽位严重时造成用户数据错乱。我的做法是给所有需要合并的字段定义明确的 reducer。以slots为例不能简单写dict而是要用定义一个合并函数from typing import Annotated from langgraph.graph import StateGraph def merge_slots(existing, new): merged dict(existing or {}) merged.update(new or {}) return merged class KumatoAgentState(TypedDict): slots: Annotated[dict, merge_slots]这样每次写入时新槽位会追加到旧槽位上而不是覆盖整个字典。如果是messages列表直接用内置的add_messages即可。检查点冲突还有一个表现多轮对话中如果同时有多个线程在跑SQLiteSaver 会把检查点分开存储只要每个thread_id唯一不会互相覆盖。容易踩的坑反而是同一个thread_id在并发访问时SQLite 写入会锁库。我在一个压测脚本里踩到过后来改成了AsyncSqliteSaver并正确处理了并发循环。5.3 大语言模型工具调用偶发异常如何兜底LangGraph 环境下大语言模型返回格式不稳定是很常见的。比如让模型输出 JSON 意图识别结果它偶尔会多输出一行 markdown。我的解决方案是在节点内部解析 LLM 输出时做一个“容错解析器”它专门处理 JSON 前后有杂讯的情况。import json import re def parse_json_safely(text): match re.search(r\{.*\}, text, re.DOTALL) if not match: return {intent: unknown, slots: {}} try: data json.loads(match.group()) return data except json.JSONDecodeError: return {intent: unknown, slots: {}}更重要的兜底策略是当模型解析失败时不应该一直重试因为这可能浪费大量 token。我的做法是失败两次就跳过低置信度意图判断直接走一个默认的引导流程请用户重新描述需求同时开启人工接管候选。这个兜底流程也放在服务保障智能体里。5.4 运维技巧通过图谱快速定位卡点LangGraph 提供图可视化能力但我在生产环境不会直接依赖图形界面而是把执行事件打印成文本日志。每个节点开始和结束都会打印事件名和状态摘要。for event in graph.stream(input_data, config, stream_modeupdates): for node_name, updated_state in event.items(): print(f[{node_name}] {updated_state.get(fault_info, )})这个日志有一个好处当系统自愈动作发生时比如发生了重试或降级你可以快速发现是哪一步出的问题。有一次线上客服一直给用户返回“系统繁忙”我看日志发现business节点持续重试了 3 次降级话术触发。再深入看故障日志发现是订单 API 的某证书过期了我把过期证书更新后系统立刻恢复。整个过程不到五分钟这比之前接到用户投诉再排查高效太多了。另外一个有用的运维习惯是监控检查点文件大小。如果发现单个线程的检查点增长特别快说明有可能进入了低效循环。这时候可以限制单个thread_id保留最近 N 个检查点清理更早的快照避免磁盘被撑爆。最后分享一个我个人的实操体会做“可自愈”的客服系统最难的其实不是写重试代码而是明确“什么情况下应该选择放弃并转人工”。自愈并不等于无限重试也不等于永远不让用户失望。花了很多时间调优后我发现最有价值的自愈设计是“优雅地失败”当系统发现自己无法解决时先给出清晰的降级回复再自动带上工单号最后把完整上下文交接给人工。对用户来说“客服解决不了但主动转人工”远比“卡住不回话”体验好得多。LangGraph 的核心优势也正在于此它可以把“尝试、失败、降级、人工接管”这一连串动作编排得干净利落而不是让工程师在乱七八糟的 if-else 里打转。如果你也在做类似的 Agent 编排系统建议先小范围验证这条链路再逐步把自愈策略加到更多节点上。
