基于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. 自愈机制的关键实现：错误分类、重试与降级

护栏节点是整个架构的灵魂。我在里面实现了一套错误分类表，把客服系统常见异常分成五类，每类对应不同的处理策略。这个分类是我跑了一段时间后才稳定下来的，最初只分“能重试”和“不能重试”两种，太粗糙了。

对应的代码结构是这样：护栏节点先做异常收集，再按分类决定下一个动作。这里我没用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里。对比过直接重试和带提示重试的效果，后者成功率能高出一截，这是值得记一笔的实操经验。

降级链路我也设计了有层次的方案，从高级能力逐步降级到保底能力：

1. 完整多智能体链路：能精确解析意图、能调用工具、能查知识库。
2. 单Agent简化链路：跳过多智能体协作，只让FAQ Bot基于检索回答，不调用业务API。
3. 纯关键词匹配：连向量检索都不用了，直接走预设的规则匹配，保证至少有个回复。
4. 转人工：把对话记录、已尝试的动作、错误摘要一起打包给人工坐席，让用户不用重复描述问题。

这个降级设计里，最重要的一个原则是：降级不是失败，是服务降级而不是服务质量归零。用户问“我的包裹到哪了”，系统如果查不到物流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_id=refund_request.order_id, amount=refund_request.amount, idempotency_key=idem_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(content=LOGISTICS_PROMPT), HumanMessage(content=user_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(content=str(tool_result), tool_call_id=tc["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里折腾多智能体，希望这篇复盘能帮你避开那些我没能避开的坑。有什么更好的自愈思路，也欢迎交流。