开头我先说个现象。这两年很多人做客服系统上来就堆多智能体一个意图识别、一个订单查询、一个售后处理再加个知识库检索看着分工明确跑起来却经常“一坏全坏”。我参与过一套基于LangGraph开发的多智能体客服架构一开始也这样直到把“自愈”能力真正设计进图里系统才算从demo变成了能用的产品。这篇就用这套架构当例子讲讲怎么用LangGraph构建一个既能多智能体协作、又能从错误里自己恢复的客服系统。如果你正准备用LangGraph做多智能体应用或者在做的过程中发现Agent互相踢皮球、出错后整个对话流程就废了那这篇文章应该能帮你少走一段弯路。我不打算讲LangGraph的API手册重点放在架构设计的取舍、自愈机制的实现细节以及我实测中踩过的坑。1. 为什么客服系统需要“自愈”而不是“高可用”先说清楚一个问题我们常说的“高可用”是运维层面的概念节点挂了就重启、流量大了就扩容这解决的是服务宕机的问题。但大模型客服系统里真正频繁发生的故障不是服务挂了而是“决策错了”。举个例子用户说“我上周买的东西还没到帮我看看”资深客服会先查订单号、再查物流状态。但多智能体系统里可能发生什么意图识别把“查物流”误判成“查订单”订单Agent查完说“订单正常”用户接着追问物流物流Agent又要重新理解上下文。一圈下来用户的问题没解决几个Agent还把状态改乱了。更麻烦的是如果中间某一步工具调用异常比如物流API超时整条图直接中断用户只能从头再来。这类问题靠重启服务解决不了必须让系统具备“自愈”能力。我对自愈的定义是四层错误检测快速识别出当前状态是否异常比如意图置信度过低、工具返回错误、Agent输出格式不对。重试与修正针对可恢复的异常调整策略后重新执行而不是原样重跑。降级当某个Agent或工具不可用时走一条功能弱化但依然能用的路径比如从“多智能体精确查询”降级为“FAQ关键词匹配”。恢复从持久化的状态里回到最近的健康检查点或者重置局部状态而不是销毁整个对话。也就是说自愈的目标不是让系统“不出错”而是让系统“出错后还能把这件事办完”。这个思路直接影响了我后面的架构设计。客服场景和写代码助手、内容生成不太一样它对“完成率”要求很高用户不会容忍Agent聊到一半撂挑子。所以我把自愈能力当成一等公民来设计而不是事后加一层异常补丁。2. 图编排的核心三个角色怎么分工才不会乱这套系统的基础结构可以抽象成三个角色编排者Supervisor、执行者Worker Agent、护栏Safety Guard。我用一个生活化的类比Supervisor是带班班长只负责分活和汇总Worker是业务专员只负责把自己那摊事干完护栏是质检员兼保险丝管着哪些情况可以重试、哪些情况必须降级或转人工。为什么要这样拆因为客服场景天然适合“集中式编排”而不是“Agent群聊”。群聊模式看起来灵活每个Agent都能说话但其实很容易出现上下文污染A Agent被B Agent的表述带偏C Agent看到一段不完整的对话就做了错误判断。我在实测中体会很深——多Agent直接互相通信语义漂移速度快得惊人经常第三轮就不知道在聊什么了。所以在LangGraph里我用了Supervisor模式所有用户消息先到Supervisor它判断意图后把任务分发给具体的执行Agent执行Agent只返回结果不直接和其他Agent对话。下面是我定义的状态结构和图骨架。from typing import Annotated, Literal, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import AnyMessage class AgentState(TypedDict): # 对外暴露的消息流最终答案从这里取 messages: Annotated[list[AnyMessage], add_messages] # 编排者解析出的用户意图 current_intent: str # 各Agent返回的结果互不污染 agent_outputs: dict # 自愈用的计数器 retry_count: int max_retries: int # 是否已降级 degraded: bool这里有个关键点我要强调执行Agent的工作成果不塞进messages而是放进agent_outputs这个独立字典。为什么因为messages加了add_messages这个reducer后只增不改如果把中间步骤全塞进去上下文会越滚越大几轮对话后token消耗直接失控。更致命的是一旦一个Agent的错误输出进了消息流后面的Agent很难判断这是“正确结论”还是“错误尝试”自愈逻辑根本无从下手。图里注册了四个核心节点from langgraph.graph import StateGraph def supervisor_node(state: AgentState) - dict: # 用LLM判断意图返回 current_intent ... def order_agent_node(state: AgentState) - dict: # 处理订单相关问题调用订单API ... def logistics_agent_node(state: AgentState) - dict: # 处理物流相关问题 ... def refund_agent_node(state: AgentState) - dict: # 处理退款/售后问题 ... def safety_guard_node(state: AgentState) - dict: # 统一异常处理分类、重试、降级 ... builder StateGraph(AgentState) builder.add_node(supervisor, supervisor_node) builder.add_node(order_agent, order_agent_node) builder.add_node(logistics_agent, logistics_agent_node) builder.add_node(refund_agent, refund_agent_node) builder.add_node(safety_guard, safety_guard_node)路由逻辑是一个条件边函数。这个函数决定了Supervisor判断完意图后下一步进哪个Agent以及要不要先进自愈护栏。def route_after_supervisor(state: AgentState) - Literal[order_agent, logistics_agent, refund_agent, safety_guard, human_handoff]: # 如果之前已经重试过太多次直接转人工 if state.get(retry_count, 0) state.get(max_retries, 3): return human_handoff # 意图置信度低先进护栏做消歧义确认 if state.get(intent_confidence, 1.0) 0.6: return safety_guard intent state[current_intent] if intent order: return order_agent if intent logistics: return logistics_agent if intent refund: return refund_agent # 兜底绝不能返回一个不存在分支 return safety_guard builder.add_conditional_edges( supervisor, route_after_supervisor, { order_agent: order_agent, logistics_agent: logistics_agent, refund_agent: refund_agent, safety_guard: safety_guard, human_handoff: human_handoff, }, )注意最后那个human_handoff分支我特别强调这个是因为条件路由必须覆盖所有可能返回值否则LangGraph在运行时直接报错。很多第一次写条件边的人容易漏掉兜底分支一旦LLM返回了预料外的意图词Graph就找不到对应的边整个请求直接失败。这种失败其实也是一种“非自愈”的表现而我们希望系统能优雅处理所有未知情况。我走的边是这样闭环的supervisor - 某个agent - safety_guard - supervisor。每个Agent跑完必须过一遍护栏护栏判断这次执行“是不是健康”再决定是回Supervisor继续下一步还是重试当前Agent还是直接降级。这样设计的好处是自愈逻辑集中在护栏节点里各Agent只需要专注于业务处理不用自己关心错误恢复。3. 自愈机制的关键实现错误分类、重试与降级护栏节点是整个架构的灵魂。我在里面实现了一套错误分类表把客服系统常见异常分成五类每类对应不同的处理策略。这个分类是我跑了一段时间后才稳定下来的最初只分“能重试”和“不能重试”两种太粗糙了。错误类型典型例子处理策略输出解析异常Agent返回的JSON残缺、字段缺失解析器修复后重试最多2次工具调用异常订单API超时、物流查询返回502指数退避重试超时时间翻倍业务规则冲突退款金额超过订单金额不重试直接转人工审核知识检索无结果知识库向量检索返回空降级为FAQ关键词匹配或通用话术意图置信度过低用户输入太短、指代不明护栏内追加一轮澄清提问对应的代码结构是这样护栏节点先做异常收集再按分类决定下一个动作。这里我没用LangGraph自带的retry机制因为LangGraph的add_node本身不提供细粒度的重试语义需要在节点内部自行控制状态。def safety_guard_node(state: AgentState) - dict: last_agent_output state[agent_outputs].get(last, {}) error_code last_agent_output.get(error_code) if error_code is None: # 没有异常正常回流给supervisor继续 return {retry_count: 0, degraded: False} # 按错误分类决定策略 if error_code in (parse_error, tool_timeout): if state[retry_count] state[max_retries]: return { retry_count: state[retry_count] 1, next_action: retry_same_agent, error_hint: last_agent_output.get(error_message), } else: # 重试次数耗尽降级 return { degraded: True, next_action: fallback_to_faq, } if error_code in (business_rule_violation,): # 业务冲突必须转人工 return { degraded: True, next_action: human_handoff, handoff_reason: last_agent_output.get(error_message), } # 未知错误保守降级 return {degraded: True, next_action: fallback_to_faq}然后根据护栏返回的next_action再注册一条条件边来决定下一跳。重试不能是简单原样重跑否则大概率还是失败。我在重试时会给Agent注入错误提示让它根据提示修正动作。比如如果是JSON解析失败我会把“解析失败请检查输出严格符合JSON格式不要包含多余说明文字”拼到下一轮的Prompt里。对比过直接重试和带提示重试的效果后者成功率能高出一截这是值得记一笔的实操经验。降级链路我也设计了有层次的方案从高级能力逐步降级到保底能力完整多智能体链路能精确解析意图、能调用工具、能查知识库。单Agent简化链路跳过多智能体协作只让FAQ Bot基于检索回答不调用业务API。纯关键词匹配连向量检索都不用了直接走预设的规则匹配保证至少有个回复。转人工把对话记录、已尝试的动作、错误摘要一起打包给人工坐席让用户不用重复描述问题。这个降级设计里最重要的一个原则是降级不是失败是服务降级而不是服务质量归零。用户问“我的包裹到哪了”系统如果查不到物流API至少可以回一句“物流系统暂时繁忙您可以从订单详情页查看最新状态或者留下订单号我们稍后短信通知您”。这种响应比直接报错强太多。4. 这四点不处理好图会把自己绕死LangGraph本质上是一个有环图节点可以不断重复进入。这个特性让多智能体协作变得灵活但也埋了几个大坑。我一个个说都是实测遇过的。4.1 递归深度限制与死循环风险LangGraph默认有recursion_limit超过就抛异常。多智能体场景里一旦Supervisor判断意图反复摇摆或者护栏一直判定“需要重试”图就可能无限循环。我线上设过一轮10次结果某个会话直接干爆了token预算。解决办法有两层。第一层是系统参数上给足限制比如每条会话最大节点访问次数20次超了就强制打断并转人工。第二层是在状态里显式跟踪循环次数护栏节点里每进一次就给retry_count加一阈值一到不管当前看起来多“有希望”都必须降级。兜底策略的意义就是宁可转人工不能让用户无限等待。我见过一些项目为了“让Agent充分思考”把recursion_limit调成无限这是灾难级的配置。大模型不会被“逼”出更好的推理只会反复横跳消耗成本。4.2 条件路由的兜底分支必须覆盖全前面说过返回值和边映射不一致会直接报错。但还有个更隐蔽的问题LLM返回的意图词和你预定义的不完全一致。比如你定义的意图是order_query模型返回的是order_status在条件路由里匹配不上就会落到默认分支。如果这个默认分支写的是END用户体验就是“系统突然不说话了”。我的做法是在路由函数里显式处理“未知意图”进护栏让LLM重新澄清一次而不是直接结束或报错。另外我会在Supervisor的Prompt里明确给出意图枚举表并要求“只准输出枚举中的一个词不要解释”。低温采样比如temperature降到0.1也能明显减少这种非预期输出。4.3 并行分支对共享状态的写入冲突有一次我给系统加了并行能力订单Agent和物流Agent同时在两个分支上跑想缩短响应时间。结果发现两个Agent往同一个字段里写数据最后谁先谁后全看调度顺序状态完全不可控。在LangGraph里并行节点如果都返回同一个key的更新合并规则如果定义不清晰整个state就会错乱。后面我定了一条铁律并行的Agent只能写入各自的命名空间字段比如order_agent_result和logistics_agent_result公共字段只能由Supervisor节点写入。如果你需要并行请务必为每个并行分支的数据分配独立的key或者用自定义reducer来合并数据千万别共享一个普通字段直接覆盖。4.4 谁对工具副作用负责客服系统必然要调用真实业务API。查询类接口无所谓但退款、改签、补偿这类写操作一旦Agent在重试中就可能重复执行造成严重问题。比如退款请求超时了Agent重试一次用户可能被扣两笔款。这个问题我在设计里用两个手段解决。第一调用写操作API时必须带上由会话ID和安全随机数生成的幂等键idempotency_key服务端靠这个键去重。第二敏感操作不走Agent自动执行而是让图停在中间状态调用interrupt_before机制请求人工确认后再继续。LangGraph的interrupt/Command支持这种暂停恢复模式这也是“自愈”里“恢复”环节的关键用法——状态持久化了人审通过后可以从检查点继续跑不用用户重新开始。def refund_agent_node(state: AgentState) - dict: # 生成幂等键 import uuid idem_key f{state.get(conversation_id)}:{uuid.uuid4().hex} api_result call_refund_api( order_idrefund_request.order_id, amountrefund_request.amount, idempotency_keyidem_key, ) # 返回结果交给护栏判断是否需要人工复核 ...5. 数据一致性的最后一道防线状态持久化与上下文管理LangGraph的MemorySaver或外部Checkpointer是我建议每一个生产级项目都要用上的能力。它的作用不只是“对话历史记忆”更重要的是支持从任意检查点恢复执行。自愈系统如果丢了状态那前面所有恢复逻辑都是空中楼阁。我踩过一个跟add_messages相关的大坑。执行Agent内部如果用了LangChain的AgentExecutor它会往消息流里写入工具调用的中间步骤。因为messages带add_messagesreducer这些中间消息会全部保留到下一次图运行。一开始还好等到第10轮对话Prompt里的历史消息膨胀到一个离谱的长度每次调用都慢且更容易让模型混淆“哪些是给用户的话哪些是内部思考”。后来我改成执行Agent内部用一个独立的局部消息列表来跑推理不断把工具调用、观察结果都放在局部列表里最后只把一句话总结放进agent_outputs不回写公共messages。公共消息流里只保留用户输入和系统最终回复。固定结构大概是async def logistics_agent_node(state: AgentState) - dict: # 直接从 state 提取需要的信息不再读全量 messages user_query state[agent_outputs].get(verified_query, ) local_messages [SystemMessage(contentLOGISTICS_PROMPT), HumanMessage(contentuser_query)] # 内部循环模型提工具、执行工具、再观察循环最多5次 for _ in range(5): ai_msg llm_with_tools.invoke(local_messages) local_messages.append(ai_msg) if not ai_msg.tool_calls: break for tc in ai_msg.tool_calls: tool_result execute_tool(tc) local_messages.append(ToolMessage(contentstr(tool_result), tool_call_idtc[id])) return { agent_outputs: { last: { agent: logistics_agent, summary: final_answer, raw_data: structured_data, } } }这个设计之后token消耗降了30%以上而且Agent之间不再互相猜疑上一轮到底发生了什么事。状态里只有结论和依据没有脏的过程。Checkpointer的thread_id我建议直接用客服会话ID这样用户中途刷新页面再次发起请求时能延续之前的图状态。我有一个小习惯在每次重大状态变更比如降级、转人工时打印一次当前checkpoint的哈希值方便排查时定位“用户到底卡在哪个版本的状态里”。6. 跑通架构之后我踩过的几个深坑理论部分说完分享几个实际运行中暴露的问题。这些问题如果不亲自动手跑文档里几乎看不出来。6.1 工具节点抛异常会直接废掉整张图我用过一版订单Agent工具调用直接写在节点逻辑里没做异常捕获。某个凌晨订单API返回了一个非标准的空响应Python解析直接抛异常LangGraph节点崩溃整个图运行被标记为失败。用户那边的表现就是对话中断。解决办法是在所有Agent节点外层包一层公共的safe_run装饰器任何异常都被捕获转成error_code和error_message放进agent_outputs让护栏节点去决策。一句话总结在LangGraph里节点抛出异常不等于“这个节点失败”而是“整个图失败”。想要自愈就必须让异常在节点内部被消化掉然后以数据的形式交给路由去处理。6.2 “重试成功”比“一次成功”更值得关注我最初只统计成功率后来发现拉低体验的其实是一条链路里某个环节反复重试。比如物流Agent连续重试三次才成功用户感知就是这次对话“顿了好几次”。我后来在护栏里加了重试计数埋点超过阈值就主动走降级不再硬磕。这个策略虽然可能让一些可恢复的问题直接降级了但用户的最终体验其实是更稳的。6.3 降级之后用户满意度反而升了这是我实测里最意外的一个发现。有一个版本某支付平台的接口很慢订单Agent每次都超时重试。我上线了降级策略之后跑FAQ兜底的会话变多了但用户满意度评分反而提高了。原因也合理之前用户等的是“系统一直在转圈”降级后是“立刻有一个明确的回复而且告诉你可以留下订单号稍后处理”。在客服场景里用户对延迟的敏感度远高于对功能完整度的敏感度快速响应比正确但慢更招人喜欢。受这个经验启发我把降级提示文案也优化了不再说“系统繁忙”而是说“正在为您优先处理稍后短信通知结果”。同样是降级用词的差异对用户情绪影响非常明显。6.4 人工转接的信息打包比想象中更重要转人工如果只把当前问题抛给客服客服还得重新问一遍订单号。我在转人工节点专门做了信息汇总用户原始诉求、已尝试的动作、各Agent的输出、错误摘要这些都打包成一段结构化文本放到转人工工单里。后来客服团队反馈这个设计让平均处理时长缩短了一大截。自愈系统转人工不是甩锅而是带着所有上下文去交接。7. 想把这套架构用到生产环境还要补哪些细节如果你看完前面的代码准备照着搭一套那我再补几个容易被忽略的点。首先是流式输出和自愈的冲突。最初我为了让响应更快在Supervisor节点就开始流式输出Token给用户。但问题在于如果后续护栏判断需要重试或者转人工用户已经看到了一部分生成的结果再改口就很难受。后来我把流式输出推迟到“图表决完成后”的阶段只在最终输出时才流式。响应会慢几百毫秒但体验更稳定。其次是知识检索的降级兜底。客服系统离不开知识库但向量检索偶尔会返回空结果。我设计了多路召回第一路向量检索第二路BM25关键词检索两路都空才走FAQ兜底。其中BM25召回意外地能覆盖不少长尾问法成本也低建议结合使用。再就是缓存。同一个用户问同一个问题如果能命中历史检索结果完全没必要再让LLM推理一遍。我在Supervisor前加了一层预处理节点先检查当前query是否和最近几轮意图相同是就直接复用之前的Agent输出只有新意图才进入完整链路。这样处理过的重复咨询响应时间能减少一半以上。最后评估体系不能只盯着准确率。我搭建了一个离线评测集里面既有常规问题也专门加入了“工具超时”“意图不明”“Agent输出残缺”这类异常样本。每次改版都要拿这套评测集跑一遍数一数有多少比例的问题最终被成功解决而不是被降级或转人工。没有这个评测集我不敢轻易改护栏的阈值因为自愈策略稍微调紧一点转人工率就可能飙升。整套架构从第一版到稳定运行我大概迭代了三轮。第一轮把多智能体协作跑通第二轮把自愈护栏加上第三轮才真正解决了状态污染和流式输出的体验问题。如果你现在正准备做类似的项目我的建议是不要急着加第四个Agent或者上复杂工具先把Supervisor加一个执行Agent、一个护栏节点的三角结构跑稳再慢慢扩展。多智能体的复杂度和Agent数量不是线性关系而是指数级的自愈能力跟不上扩展速度系统迟早会因为某个角落的异常而整体崩坏。现在这套系统已经在处理真实用户咨询了每天能自动解决相当比例的问题剩下的转人工工单也都带着结构化的上下文。如果你也在LangGraph里折腾多智能体希望这篇复盘能帮你避开那些我没能避开的坑。有什么更好的自愈思路也欢迎交流。
开头我先说个现象。这两年很多人做客服系统上来就堆多智能体一个意图识别、一个订单查询、一个售后处理再加个知识库检索看着分工明确跑起来却经常“一坏全坏”。我参与过一套基于LangGraph开发的多智能体客服架构一开始也这样直到把“自愈”能力真正设计进图里系统才算从demo变成了能用的产品。这篇就用这套架构当例子讲讲怎么用LangGraph构建一个既能多智能体协作、又能从错误里自己恢复的客服系统。如果你正准备用LangGraph做多智能体应用或者在做的过程中发现Agent互相踢皮球、出错后整个对话流程就废了那这篇文章应该能帮你少走一段弯路。我不打算讲LangGraph的API手册重点放在架构设计的取舍、自愈机制的实现细节以及我实测中踩过的坑。1. 为什么客服系统需要“自愈”而不是“高可用”先说清楚一个问题我们常说的“高可用”是运维层面的概念节点挂了就重启、流量大了就扩容这解决的是服务宕机的问题。但大模型客服系统里真正频繁发生的故障不是服务挂了而是“决策错了”。举个例子用户说“我上周买的东西还没到帮我看看”资深客服会先查订单号、再查物流状态。但多智能体系统里可能发生什么意图识别把“查物流”误判成“查订单”订单Agent查完说“订单正常”用户接着追问物流物流Agent又要重新理解上下文。一圈下来用户的问题没解决几个Agent还把状态改乱了。更麻烦的是如果中间某一步工具调用异常比如物流API超时整条图直接中断用户只能从头再来。这类问题靠重启服务解决不了必须让系统具备“自愈”能力。我对自愈的定义是四层错误检测快速识别出当前状态是否异常比如意图置信度过低、工具返回错误、Agent输出格式不对。重试与修正针对可恢复的异常调整策略后重新执行而不是原样重跑。降级当某个Agent或工具不可用时走一条功能弱化但依然能用的路径比如从“多智能体精确查询”降级为“FAQ关键词匹配”。恢复从持久化的状态里回到最近的健康检查点或者重置局部状态而不是销毁整个对话。也就是说自愈的目标不是让系统“不出错”而是让系统“出错后还能把这件事办完”。这个思路直接影响了我后面的架构设计。客服场景和写代码助手、内容生成不太一样它对“完成率”要求很高用户不会容忍Agent聊到一半撂挑子。所以我把自愈能力当成一等公民来设计而不是事后加一层异常补丁。2. 图编排的核心三个角色怎么分工才不会乱这套系统的基础结构可以抽象成三个角色编排者Supervisor、执行者Worker Agent、护栏Safety Guard。我用一个生活化的类比Supervisor是带班班长只负责分活和汇总Worker是业务专员只负责把自己那摊事干完护栏是质检员兼保险丝管着哪些情况可以重试、哪些情况必须降级或转人工。为什么要这样拆因为客服场景天然适合“集中式编排”而不是“Agent群聊”。群聊模式看起来灵活每个Agent都能说话但其实很容易出现上下文污染A Agent被B Agent的表述带偏C Agent看到一段不完整的对话就做了错误判断。我在实测中体会很深——多Agent直接互相通信语义漂移速度快得惊人经常第三轮就不知道在聊什么了。所以在LangGraph里我用了Supervisor模式所有用户消息先到Supervisor它判断意图后把任务分发给具体的执行Agent执行Agent只返回结果不直接和其他Agent对话。下面是我定义的状态结构和图骨架。from typing import Annotated, Literal, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import AnyMessage class AgentState(TypedDict): # 对外暴露的消息流最终答案从这里取 messages: Annotated[list[AnyMessage], add_messages] # 编排者解析出的用户意图 current_intent: str # 各Agent返回的结果互不污染 agent_outputs: dict # 自愈用的计数器 retry_count: int max_retries: int # 是否已降级 degraded: bool这里有个关键点我要强调执行Agent的工作成果不塞进messages而是放进agent_outputs这个独立字典。为什么因为messages加了add_messages这个reducer后只增不改如果把中间步骤全塞进去上下文会越滚越大几轮对话后token消耗直接失控。更致命的是一旦一个Agent的错误输出进了消息流后面的Agent很难判断这是“正确结论”还是“错误尝试”自愈逻辑根本无从下手。图里注册了四个核心节点from langgraph.graph import StateGraph def supervisor_node(state: AgentState) - dict: # 用LLM判断意图返回 current_intent ... def order_agent_node(state: AgentState) - dict: # 处理订单相关问题调用订单API ... def logistics_agent_node(state: AgentState) - dict: # 处理物流相关问题 ... def refund_agent_node(state: AgentState) - dict: # 处理退款/售后问题 ... def safety_guard_node(state: AgentState) - dict: # 统一异常处理分类、重试、降级 ... builder StateGraph(AgentState) builder.add_node(supervisor, supervisor_node) builder.add_node(order_agent, order_agent_node) builder.add_node(logistics_agent, logistics_agent_node) builder.add_node(refund_agent, refund_agent_node) builder.add_node(safety_guard, safety_guard_node)路由逻辑是一个条件边函数。这个函数决定了Supervisor判断完意图后下一步进哪个Agent以及要不要先进自愈护栏。def route_after_supervisor(state: AgentState) - Literal[order_agent, logistics_agent, refund_agent, safety_guard, human_handoff]: # 如果之前已经重试过太多次直接转人工 if state.get(retry_count, 0) state.get(max_retries, 3): return human_handoff # 意图置信度低先进护栏做消歧义确认 if state.get(intent_confidence, 1.0) 0.6: return safety_guard intent state[current_intent] if intent order: return order_agent if intent logistics: return logistics_agent if intent refund: return refund_agent # 兜底绝不能返回一个不存在分支 return safety_guard builder.add_conditional_edges( supervisor, route_after_supervisor, { order_agent: order_agent, logistics_agent: logistics_agent, refund_agent: refund_agent, safety_guard: safety_guard, human_handoff: human_handoff, }, )注意最后那个human_handoff分支我特别强调这个是因为条件路由必须覆盖所有可能返回值否则LangGraph在运行时直接报错。很多第一次写条件边的人容易漏掉兜底分支一旦LLM返回了预料外的意图词Graph就找不到对应的边整个请求直接失败。这种失败其实也是一种“非自愈”的表现而我们希望系统能优雅处理所有未知情况。我走的边是这样闭环的supervisor - 某个agent - safety_guard - supervisor。每个Agent跑完必须过一遍护栏护栏判断这次执行“是不是健康”再决定是回Supervisor继续下一步还是重试当前Agent还是直接降级。这样设计的好处是自愈逻辑集中在护栏节点里各Agent只需要专注于业务处理不用自己关心错误恢复。3. 自愈机制的关键实现错误分类、重试与降级护栏节点是整个架构的灵魂。我在里面实现了一套错误分类表把客服系统常见异常分成五类每类对应不同的处理策略。这个分类是我跑了一段时间后才稳定下来的最初只分“能重试”和“不能重试”两种太粗糙了。错误类型典型例子处理策略输出解析异常Agent返回的JSON残缺、字段缺失解析器修复后重试最多2次工具调用异常订单API超时、物流查询返回502指数退避重试超时时间翻倍业务规则冲突退款金额超过订单金额不重试直接转人工审核知识检索无结果知识库向量检索返回空降级为FAQ关键词匹配或通用话术意图置信度过低用户输入太短、指代不明护栏内追加一轮澄清提问对应的代码结构是这样护栏节点先做异常收集再按分类决定下一个动作。这里我没用LangGraph自带的retry机制因为LangGraph的add_node本身不提供细粒度的重试语义需要在节点内部自行控制状态。def safety_guard_node(state: AgentState) - dict: last_agent_output state[agent_outputs].get(last, {}) error_code last_agent_output.get(error_code) if error_code is None: # 没有异常正常回流给supervisor继续 return {retry_count: 0, degraded: False} # 按错误分类决定策略 if error_code in (parse_error, tool_timeout): if state[retry_count] state[max_retries]: return { retry_count: state[retry_count] 1, next_action: retry_same_agent, error_hint: last_agent_output.get(error_message), } else: # 重试次数耗尽降级 return { degraded: True, next_action: fallback_to_faq, } if error_code in (business_rule_violation,): # 业务冲突必须转人工 return { degraded: True, next_action: human_handoff, handoff_reason: last_agent_output.get(error_message), } # 未知错误保守降级 return {degraded: True, next_action: fallback_to_faq}然后根据护栏返回的next_action再注册一条条件边来决定下一跳。重试不能是简单原样重跑否则大概率还是失败。我在重试时会给Agent注入错误提示让它根据提示修正动作。比如如果是JSON解析失败我会把“解析失败请检查输出严格符合JSON格式不要包含多余说明文字”拼到下一轮的Prompt里。对比过直接重试和带提示重试的效果后者成功率能高出一截这是值得记一笔的实操经验。降级链路我也设计了有层次的方案从高级能力逐步降级到保底能力完整多智能体链路能精确解析意图、能调用工具、能查知识库。单Agent简化链路跳过多智能体协作只让FAQ Bot基于检索回答不调用业务API。纯关键词匹配连向量检索都不用了直接走预设的规则匹配保证至少有个回复。转人工把对话记录、已尝试的动作、错误摘要一起打包给人工坐席让用户不用重复描述问题。这个降级设计里最重要的一个原则是降级不是失败是服务降级而不是服务质量归零。用户问“我的包裹到哪了”系统如果查不到物流API至少可以回一句“物流系统暂时繁忙您可以从订单详情页查看最新状态或者留下订单号我们稍后短信通知您”。这种响应比直接报错强太多。4. 这四点不处理好图会把自己绕死LangGraph本质上是一个有环图节点可以不断重复进入。这个特性让多智能体协作变得灵活但也埋了几个大坑。我一个个说都是实测遇过的。4.1 递归深度限制与死循环风险LangGraph默认有recursion_limit超过就抛异常。多智能体场景里一旦Supervisor判断意图反复摇摆或者护栏一直判定“需要重试”图就可能无限循环。我线上设过一轮10次结果某个会话直接干爆了token预算。解决办法有两层。第一层是系统参数上给足限制比如每条会话最大节点访问次数20次超了就强制打断并转人工。第二层是在状态里显式跟踪循环次数护栏节点里每进一次就给retry_count加一阈值一到不管当前看起来多“有希望”都必须降级。兜底策略的意义就是宁可转人工不能让用户无限等待。我见过一些项目为了“让Agent充分思考”把recursion_limit调成无限这是灾难级的配置。大模型不会被“逼”出更好的推理只会反复横跳消耗成本。4.2 条件路由的兜底分支必须覆盖全前面说过返回值和边映射不一致会直接报错。但还有个更隐蔽的问题LLM返回的意图词和你预定义的不完全一致。比如你定义的意图是order_query模型返回的是order_status在条件路由里匹配不上就会落到默认分支。如果这个默认分支写的是END用户体验就是“系统突然不说话了”。我的做法是在路由函数里显式处理“未知意图”进护栏让LLM重新澄清一次而不是直接结束或报错。另外我会在Supervisor的Prompt里明确给出意图枚举表并要求“只准输出枚举中的一个词不要解释”。低温采样比如temperature降到0.1也能明显减少这种非预期输出。4.3 并行分支对共享状态的写入冲突有一次我给系统加了并行能力订单Agent和物流Agent同时在两个分支上跑想缩短响应时间。结果发现两个Agent往同一个字段里写数据最后谁先谁后全看调度顺序状态完全不可控。在LangGraph里并行节点如果都返回同一个key的更新合并规则如果定义不清晰整个state就会错乱。后面我定了一条铁律并行的Agent只能写入各自的命名空间字段比如order_agent_result和logistics_agent_result公共字段只能由Supervisor节点写入。如果你需要并行请务必为每个并行分支的数据分配独立的key或者用自定义reducer来合并数据千万别共享一个普通字段直接覆盖。4.4 谁对工具副作用负责客服系统必然要调用真实业务API。查询类接口无所谓但退款、改签、补偿这类写操作一旦Agent在重试中就可能重复执行造成严重问题。比如退款请求超时了Agent重试一次用户可能被扣两笔款。这个问题我在设计里用两个手段解决。第一调用写操作API时必须带上由会话ID和安全随机数生成的幂等键idempotency_key服务端靠这个键去重。第二敏感操作不走Agent自动执行而是让图停在中间状态调用interrupt_before机制请求人工确认后再继续。LangGraph的interrupt/Command支持这种暂停恢复模式这也是“自愈”里“恢复”环节的关键用法——状态持久化了人审通过后可以从检查点继续跑不用用户重新开始。def refund_agent_node(state: AgentState) - dict: # 生成幂等键 import uuid idem_key f{state.get(conversation_id)}:{uuid.uuid4().hex} api_result call_refund_api( order_idrefund_request.order_id, amountrefund_request.amount, idempotency_keyidem_key, ) # 返回结果交给护栏判断是否需要人工复核 ...5. 数据一致性的最后一道防线状态持久化与上下文管理LangGraph的MemorySaver或外部Checkpointer是我建议每一个生产级项目都要用上的能力。它的作用不只是“对话历史记忆”更重要的是支持从任意检查点恢复执行。自愈系统如果丢了状态那前面所有恢复逻辑都是空中楼阁。我踩过一个跟add_messages相关的大坑。执行Agent内部如果用了LangChain的AgentExecutor它会往消息流里写入工具调用的中间步骤。因为messages带add_messagesreducer这些中间消息会全部保留到下一次图运行。一开始还好等到第10轮对话Prompt里的历史消息膨胀到一个离谱的长度每次调用都慢且更容易让模型混淆“哪些是给用户的话哪些是内部思考”。后来我改成执行Agent内部用一个独立的局部消息列表来跑推理不断把工具调用、观察结果都放在局部列表里最后只把一句话总结放进agent_outputs不回写公共messages。公共消息流里只保留用户输入和系统最终回复。固定结构大概是async def logistics_agent_node(state: AgentState) - dict: # 直接从 state 提取需要的信息不再读全量 messages user_query state[agent_outputs].get(verified_query, ) local_messages [SystemMessage(contentLOGISTICS_PROMPT), HumanMessage(contentuser_query)] # 内部循环模型提工具、执行工具、再观察循环最多5次 for _ in range(5): ai_msg llm_with_tools.invoke(local_messages) local_messages.append(ai_msg) if not ai_msg.tool_calls: break for tc in ai_msg.tool_calls: tool_result execute_tool(tc) local_messages.append(ToolMessage(contentstr(tool_result), tool_call_idtc[id])) return { agent_outputs: { last: { agent: logistics_agent, summary: final_answer, raw_data: structured_data, } } }这个设计之后token消耗降了30%以上而且Agent之间不再互相猜疑上一轮到底发生了什么事。状态里只有结论和依据没有脏的过程。Checkpointer的thread_id我建议直接用客服会话ID这样用户中途刷新页面再次发起请求时能延续之前的图状态。我有一个小习惯在每次重大状态变更比如降级、转人工时打印一次当前checkpoint的哈希值方便排查时定位“用户到底卡在哪个版本的状态里”。6. 跑通架构之后我踩过的几个深坑理论部分说完分享几个实际运行中暴露的问题。这些问题如果不亲自动手跑文档里几乎看不出来。6.1 工具节点抛异常会直接废掉整张图我用过一版订单Agent工具调用直接写在节点逻辑里没做异常捕获。某个凌晨订单API返回了一个非标准的空响应Python解析直接抛异常LangGraph节点崩溃整个图运行被标记为失败。用户那边的表现就是对话中断。解决办法是在所有Agent节点外层包一层公共的safe_run装饰器任何异常都被捕获转成error_code和error_message放进agent_outputs让护栏节点去决策。一句话总结在LangGraph里节点抛出异常不等于“这个节点失败”而是“整个图失败”。想要自愈就必须让异常在节点内部被消化掉然后以数据的形式交给路由去处理。6.2 “重试成功”比“一次成功”更值得关注我最初只统计成功率后来发现拉低体验的其实是一条链路里某个环节反复重试。比如物流Agent连续重试三次才成功用户感知就是这次对话“顿了好几次”。我后来在护栏里加了重试计数埋点超过阈值就主动走降级不再硬磕。这个策略虽然可能让一些可恢复的问题直接降级了但用户的最终体验其实是更稳的。6.3 降级之后用户满意度反而升了这是我实测里最意外的一个发现。有一个版本某支付平台的接口很慢订单Agent每次都超时重试。我上线了降级策略之后跑FAQ兜底的会话变多了但用户满意度评分反而提高了。原因也合理之前用户等的是“系统一直在转圈”降级后是“立刻有一个明确的回复而且告诉你可以留下订单号稍后处理”。在客服场景里用户对延迟的敏感度远高于对功能完整度的敏感度快速响应比正确但慢更招人喜欢。受这个经验启发我把降级提示文案也优化了不再说“系统繁忙”而是说“正在为您优先处理稍后短信通知结果”。同样是降级用词的差异对用户情绪影响非常明显。6.4 人工转接的信息打包比想象中更重要转人工如果只把当前问题抛给客服客服还得重新问一遍订单号。我在转人工节点专门做了信息汇总用户原始诉求、已尝试的动作、各Agent的输出、错误摘要这些都打包成一段结构化文本放到转人工工单里。后来客服团队反馈这个设计让平均处理时长缩短了一大截。自愈系统转人工不是甩锅而是带着所有上下文去交接。7. 想把这套架构用到生产环境还要补哪些细节如果你看完前面的代码准备照着搭一套那我再补几个容易被忽略的点。首先是流式输出和自愈的冲突。最初我为了让响应更快在Supervisor节点就开始流式输出Token给用户。但问题在于如果后续护栏判断需要重试或者转人工用户已经看到了一部分生成的结果再改口就很难受。后来我把流式输出推迟到“图表决完成后”的阶段只在最终输出时才流式。响应会慢几百毫秒但体验更稳定。其次是知识检索的降级兜底。客服系统离不开知识库但向量检索偶尔会返回空结果。我设计了多路召回第一路向量检索第二路BM25关键词检索两路都空才走FAQ兜底。其中BM25召回意外地能覆盖不少长尾问法成本也低建议结合使用。再就是缓存。同一个用户问同一个问题如果能命中历史检索结果完全没必要再让LLM推理一遍。我在Supervisor前加了一层预处理节点先检查当前query是否和最近几轮意图相同是就直接复用之前的Agent输出只有新意图才进入完整链路。这样处理过的重复咨询响应时间能减少一半以上。最后评估体系不能只盯着准确率。我搭建了一个离线评测集里面既有常规问题也专门加入了“工具超时”“意图不明”“Agent输出残缺”这类异常样本。每次改版都要拿这套评测集跑一遍数一数有多少比例的问题最终被成功解决而不是被降级或转人工。没有这个评测集我不敢轻易改护栏的阈值因为自愈策略稍微调紧一点转人工率就可能飙升。整套架构从第一版到稳定运行我大概迭代了三轮。第一轮把多智能体协作跑通第二轮把自愈护栏加上第三轮才真正解决了状态污染和流式输出的体验问题。如果你现在正准备做类似的项目我的建议是不要急着加第四个Agent或者上复杂工具先把Supervisor加一个执行Agent、一个护栏节点的三角结构跑稳再慢慢扩展。多智能体的复杂度和Agent数量不是线性关系而是指数级的自愈能力跟不上扩展速度系统迟早会因为某个角落的异常而整体崩坏。现在这套系统已经在处理真实用户咨询了每天能自动解决相当比例的问题剩下的转人工工单也都带着结构化的上下文。如果你也在LangGraph里折腾多智能体希望这篇复盘能帮你避开那些我没能避开的坑。有什么更好的自愈思路也欢迎交流。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
