多智能体协作这件事我在去年做智能客服系统时踩过不少坑。最开始用单 Agent 硬扛把所有意图识别、知识检索、工单创建、情绪安抚全塞进一个 StateGraph 里结果节点越加越多状态字段膨胀到四十多个调试时根本分不清哪个节点改了哪个字段改一处崩三处。后来拆成多智能体又走了另一个极端——每个 Agent 独立跑互相之间靠字符串传话上下文丢失严重用户问“刚才那个订单”下游 Agent 完全不知道“那个”指的是什么。直到把 LangGraph 的子图模式用对才算真正把架构理顺主图负责路由和全局状态子图负责各自领域的完整推理链路父子图之间通过显式的状态映射通信既隔离又可控。这篇就把这套 AI 客服系统的完整搭建过程拆开讲从为什么选子图、状态怎么设计、路由怎么分流到人工介入和流式输出怎么接全部给到可复现的代码和参数。1. 为什么客服系统非要用子图模式而不是一张大图1.1 单图堆节点的三个致命问题先说清楚痛点不然你不会理解子图的价值。我第一版系统就是一张 StateGraph节点包括意图分类、闲聊回复、订单查询、退款申请、物流追踪、知识库检索、情绪检测、转人工判断、回复生成一共九个节点。跑起来能跑但问题在第二周集中爆发。第一个问题是状态字段污染。LangGraph 的 State 是一个共享的 TypedDict所有节点都能读写。订单查询节点需要order_id、order_status退款节点需要refund_reason、refund_amount这些字段全堆在一个 State 里。当我在退款流程里调试时打印整个 State看到一堆跟退款无关的物流字段排查效率极低。更麻烦的是字段命名冲突——订单查询用status表示订单状态情绪检测也想用status表示情绪状态最后只能改成order_status和emotion_status越改越乱。第二个问题是条件边的复杂度爆炸。九个节点之间的跳转逻辑写在add_conditional_edges里判断函数越来越长。意图分类之后要判断走哪条路退款流程内部还要判断是“用户确认退款”还是“用户取消”物流查询要判断“有物流信息”还是“无物流信息需转人工”。这些判断散落在主图各处改一个业务规则要翻遍整个文件。第三个问题是复用性为零。退款流程在客服系统里要用在售后工单系统里也要用但因为它跟主图的状态强耦合根本抽不出来。每次新业务都要复制粘贴一遍维护成本翻倍。1.2 子图模式到底解决了什么子图模式的本质是把一张大图拆成若干张独立的小图每张小图有自己的 State、自己的节点、自己的边然后作为主图的一个节点被调用。这个“作为节点被调用”是关键——子图对外暴露的接口就是一个普通的节点函数主图不需要知道子图内部有多少节点、怎么跳转。对应到客服系统我拆成了四个子图售前咨询子图处理产品参数、价格、库存、优惠活动内部有知识库检索、参数对比、推荐生成三个节点。订单服务子图处理订单查询、修改地址、取消订单内部有订单校验、状态查询、操作执行三个节点。售后子图处理退款、退货、换货、投诉内部有政策校验、金额计算、工单创建、情绪安抚四个节点。闲聊子图处理打招呼、感谢、无关话题内部就一个回复生成节点。主图只保留三个节点意图路由、子图调度、回复聚合。意图路由判断用户这句话属于哪个领域子图调度调用对应的子图回复聚合把子图的输出整理成最终回复。主图的 State 只保留全局字段user_id、session_id、messages、current_intent、final_response。子图内部的字段全部封装在子图自己的 State 里主图看不见。这样带来的好处很直接调试退款问题时我只需要看售后子图的 State字段不超过十个一目了然新增业务时写一个新的子图挂上去就行主图不用动退款子图可以原封不动搬到工单系统里复用。1.3 子图与主图的状态映射机制这里有个容易踩的坑子图的 State 和主图的 State 不是同一个对象它们之间需要显式映射。LangGraph 的做法是当子图作为节点被调用时你可以定义一个输入转换函数把主图 State 里子图需要的字段提取出来传给子图子图跑完后再定义一个输出转换函数把子图的结果写回主图 State。我一开始没做映射直接让子图读写主图 State结果子图里的临时字段泄漏到主图主图的全局字段被子图意外覆盖。后来改成显式映射代码反而更清晰了。具体写法在第三节展开。2. 客服系统的状态设计主图管全局子图管局部2.1 主图 State 的字段取舍主图 State 我只留了六个字段每个都有明确用途from typing import TypedDict, Annotated, Sequence from langchain_core.messages import BaseMessage import operator class MainState(TypedDict): user_id: str # 用户唯一标识贯穿全流程 session_id: str # 会话标识用于多轮上下文 messages: Annotated[Sequence[BaseMessage], operator.add] # 完整对话历史 current_intent: str # 当前意图由路由节点写入 subgraph_result: dict # 子图返回的结构化结果 final_response: str # 最终回复文本messages用了operator.add作为 reducer这是 LangGraph 的标准做法保证每次追加消息而不是覆盖。subgraph_result是个 dict不同子图返回的结构不同但主图不关心内部结构只负责透传给回复聚合节点。注意不要把子图的内部字段比如refund_amount、order_status加到主图 State 里。一旦加了主图和子图就耦合了子图就没法独立复用。我见过有人图省事把子图字段全加到主图结果主图 State 又变成四十多个字段等于白拆。2.2 子图 State 的独立定义以售后子图为例它的 State 是这样的class AfterSalesState(TypedDict): user_query: str # 用户原始问题 order_info: dict # 订单信息从主图传入 policy_result: dict # 政策校验结果 refund_amount: float # 计算出的退款金额 emotion_level: str # 情绪等级normal/upset/angry action_taken: str # 执行的动作refund/return/exchange response: str # 子图生成的回复注意order_info是从主图传入的其他字段都是子图内部产生的。子图跑完后只把response和action_taken通过输出映射写回主图的subgraph_result其他字段留在子图内部主图看不到。2.3 父子图状态映射的代码实现映射通过两个函数完成。输入映射负责从主图 State 提取子图需要的字段def map_input_to_aftersales(main_state: MainState) - AfterSalesState: return AfterSalesState( user_querymain_state[messages][-1].content, order_infomain_state.get(subgraph_result, {}).get(order_info, {}), policy_result{}, refund_amount0.0, emotion_levelnormal, action_taken, response )输出映射负责把子图结果写回主图def map_output_to_main(aftersales_state: AfterSalesState) - dict: return { subgraph_result: { action_taken: aftersales_state[action_taken], refund_amount: aftersales_state[refund_amount] }, final_response: aftersales_state[response] }然后在主图里用add_node把子图挂上去from langgraph.graph import StateGraph builder StateGraph(MainState) builder.add_node(router, router_node) builder.add_node(aftersales, aftersales_subgraph) # 子图直接作为节点 builder.add_node(aggregator, aggregator_node)LangGraph 会自动处理子图的调用和状态映射前提是子图本身是一个编译好的CompiledGraph。这里有个细节子图编译时用的 State 类型必须是AfterSalesState不能是MainState否则映射会出错。3. 意图路由怎么把用户问题准确分到四个子图3.1 路由节点的三种实现方案对比路由是主图的核心它决定了用户问题走哪个子图。我试过三种方案各有适用场景方案实现方式准确率延迟适用场景关键词匹配正则关键词表70%左右极低意图固定、词汇量小的场景LLM 分类提示词让模型输出意图标签92%左右300-800ms意图复杂、表达多样的场景小模型微调微调一个分类模型95%以上50-100ms高频调用、对延迟敏感的场景我最终选了 LLM 分类因为客服场景用户表达太灵活“我买的东西怎么还没到”和“物流信息查一下”都是物流查询关键词匹配覆盖不全。LLM 分类虽然慢一点但准确率够用而且改意图只需要改提示词不用重新训练。3.2 LLM 路由的提示词设计路由提示词的关键是给模型明确的标签定义和边界示例不然模型会在边界情况上摇摆。我的提示词是这样的ROUTER_PROMPT 你是一个客服意图分类器。根据用户的问题判断它属于以下哪个类别 - presale: 售前咨询包括产品参数、价格、库存、优惠活动、产品对比 - order: 订单服务包括订单查询、修改地址、取消订单、订单状态 - aftersales: 售后服务包括退款、退货、换货、投诉、质量问题 - chitchat: 闲聊包括打招呼、感谢、无关话题、无法归类的 判断规则 1. 如果问题同时涉及多个类别选择最需要立即处理的那个 2. 如果问题包含明确的退款/退货/投诉词汇优先归为 aftersales 3. 如果问题只是打招呼或感谢归为 chitchat 4. 不确定时归为 chitchat 只输出类别标签不要输出其他内容。 用户问题{query} 类别这个提示词我迭代了五版。第一版没有边界示例模型经常把“我要退款”分到 order第二版加了规则好了一些第三版发现“订单有问题要退款”这种复合意图会摇摆加了规则1第四版发现模型偶尔输出“aftersales类别”这种带后缀的加了“只输出类别标签”第五版稳定在92%左右。3.3 路由结果到子图的边定义路由节点跑完后根据current_intent决定走哪个子图。LangGraph 的条件边写法def route_decision(state: MainState) - str: intent state[current_intent] if intent presale: return presale elif intent order: return order elif intent aftersales: return aftersales else: return chitchat builder.add_conditional_edges( router, route_decision, { presale: presale, order: order, aftersales: aftersales, chitchat: chitchat } )这里有个坑add_conditional_edges的第三个参数是映射字典key 是route_decision的返回值value 是节点名。如果 value 写错LangGraph 不会报错而是静默走到END你会看到流程莫名其妙结束了。我因为这个排查了两个小时后来养成习惯每次加条件边都打印一下映射字典确认。4. 售后子图的完整实现从政策校验到情绪安抚4.1 售后子图的节点编排售后子图是四个子图里最复杂的它有四个节点编排逻辑是aftersales_builder StateGraph(AfterSalesState) aftersales_builder.add_node(policy_check, policy_check_node) aftersales_builder.add_node(amount_calc, amount_calc_node) aftersales_builder.add_node(emotion_handle, emotion_handle_node) aftersales_builder.add_node(action_execute, action_execute_node) aftersales_builder.set_entry_point(policy_check) aftersales_builder.add_edge(policy_check, amount_calc) aftersales_builder.add_conditional_edges( amount_calc, emotion_decision, { angry: emotion_handle, normal: action_execute } ) aftersales_builder.add_edge(emotion_handle, action_execute) aftersales_builder.add_edge(action_execute, END) aftersales_subgraph aftersales_builder.compile()流程是先校验退款政策是否在退款期内、商品是否支持退款然后计算退款金额接着判断用户情绪——如果情绪是 angry先走情绪安抚节点再执行动作如果是 normal直接执行动作。这个分支设计是因为愤怒的用户如果直接收到“退款已提交”的机械回复情绪会更差先安抚再执行体验好很多。4.2 政策校验节点的实现细节政策校验节点要查三件事订单是否在退款期内、商品是否属于可退款品类、用户是否已经退过款。代码def policy_check_node(state: AfterSalesState) - dict: order state[order_info] order_date datetime.fromisoformat(order[created_at]) days_since (datetime.now() - order_date).days # 七天无理由退款 if days_since 7: return { policy_result: { passed: False, reason: f订单已超过7天退款期已过{days_since}天 } } # 特殊品类不支持退款 if order[category] in [virtual, customized]: return { policy_result: { passed: False, reason: 虚拟商品和定制商品不支持无理由退款 } } # 已退款订单不能重复退 if order.get(refunded, False): return { policy_result: { passed: False, reason: 该订单已退款不能重复申请 } } return { policy_result: { passed: True, reason: 符合退款政策 } }这里有个实际经验退款期计算要用自然日还是工作日一定要跟业务确认。我第一版用自然日结果用户周五下单下周一申请退款系统算出来过了3天但用户认为只过了1个工作日。后来改成自然日但提示文案里说明“含周末”减少了很多争议。4.3 情绪检测与安抚策略情绪检测我用了一个轻量方案关键词 LLM 二次判断。关键词先做初筛命中“投诉”“差评”“曝光”“12315”这类词直接标记 angry没命中的再用 LLM 判断提示词让模型输出 normal/upset/angry 三档。安抚策略分档处理normal正常回复直接执行动作。upset回复开头加一句“非常抱歉给您带来不便”然后执行动作。angry先走安抚节点安抚节点生成一段共情话术再执行动作且动作执行后追加补偿方案如优惠券。安抚节点的话术生成提示词EMOTION_PROMPT 用户情绪激动请生成一段安抚话术。要求 1. 先共情承认用户的感受 2. 不要辩解不要找借口 3. 给出明确的解决承诺 4. 控制在50字以内 用户问题{query} 用户情绪{emotion_level} 安抚话术实测下来加了安抚节点后angry 用户的二次投诉率从18%降到了7%。这个数据是我自己系统里的不一定通用但方向是对的——情绪处理不能省。5. 人工介入与流式输出生产环境必须补的两块5.1 什么情况下触发人工介入子图跑完后不是所有情况都能自动回复。我设了三个触发条件政策校验不通过且用户情绪为 angry自动回复可能激化矛盾转人工。退款金额超过500元大额退款需要人工审核。连续两轮意图无法识别用户问了两次都没路由成功转人工。触发人工介入的实现是在主图的聚合节点里判断def aggregator_node(state: MainState) - dict: result state[subgraph_result] if result.get(need_human): return { final_response: 您的问题需要人工客服处理正在为您转接..., need_human: True } return {final_response: result.get(response, 抱歉我暂时无法处理)}need_human这个字段我在主图 State 里加了虽然前面说主图只留六个字段但这个字段是必要的因为它是跨子图的全局信号。5.2 流式输出的接入方式客服系统用户对响应速度很敏感等三秒才出完整回复体验很差。LangGraph 支持流式输出用astream_events可以拿到每个节点的中间结果async for event in app.astream_events( {messages: [HumanMessage(contentquery)], user_id: uid, session_id: sid}, versionv2 ): kind event[event] if kind on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield chunk.content这里有个坑子图内部的 LLM 调用也会触发on_chat_model_stream事件如果你不想把子图内部的中间推理暴露给用户需要在事件过滤时判断event[metadata].get(langgraph_node)是不是最终回复节点。我第一版没过滤用户看到了“正在校验退款政策...正在计算金额...”这些内部日志虽然显得透明但正式环境不合适。5.3 多轮对话的上下文管理客服场景经常是多轮的用户第一句问“我的订单到哪了”第二句问“那能改地址吗”。第二句的“那”指代第一句的订单如果子图之间不共享上下文第二句会路由失败。我的做法是在主图 State 的messages里保留完整对话历史路由节点分类时把最近三轮对话一起传给 LLMdef router_node(state: MainState) - dict: recent state[messages][-6:] # 最近三轮 context \n.join([f{m.type}: {m.content} for m in recent]) intent llm.invoke(ROUTER_PROMPT.format(querycontext)) return {current_intent: intent.content.strip()}同时子图内部如果需要历史订单信息从messages里解析。我在订单服务子图里加了一个extract_order_id节点用正则从最近五轮对话里找订单号找到了就复用找不到才追问用户。6. 实测中暴露的三个问题与修复过程6.1 子图编译时的 State 类型不匹配第一个问题出现在子图挂载到主图时。我一开始把子图的 State 定义成了MainState想着“反正字段差不多”结果运行时报KeyError: order_info。原因是主图 State 里没有order_info字段子图访问时找不到。排查过程先打印子图编译后的节点列表确认节点都在然后在子图入口加日志发现state里确实没有order_info最后定位到是 State 类型用错了。修复就是把子图 State 改成独立的AfterSalesState并通过输入映射函数注入order_info。这个坑的教训是子图 State 必须独立定义不能复用主图 State。LangGraph 不会在编译时检查这个只会在运行时暴露。6.2 条件边映射写错导致流程静默结束第二个问题更隐蔽。我在售后子图里加了一个条件边映射字典写成了aftersales_builder.add_conditional_edges( amount_calc, emotion_decision, { angry: emotion_handle, normal: action_execute } )但emotion_decision函数在某些情况下返回了neutral映射字典里没有这个 key。LangGraph 的处理是找不到映射就走到END不报错。结果就是 neutral 情绪的用户走到amount_calc后直接结束没有执行动作也没有回复。排查过程很痛苦因为不报错。我最后是在emotion_decision里加了日志发现返回了neutral才意识到映射字典缺 key。修复方案是给emotion_decision加默认返回值同时映射字典加neutral: action_execute。经验条件边的映射字典一定要覆盖判断函数的所有可能返回值。建议在判断函数里用assert或日志确保返回值在预期范围内。6.3 流式输出时子图内部事件泄漏第三个问题前面提过子图内部的 LLM 调用事件被流式输出到了前端。用户看到了“正在校验退款政策”这种内部日志。修复方式是在事件过滤时加节点判断FINAL_NODES {aggregator, emotion_handle, action_execute} async for event in app.astream_events(...): if event[event] on_chat_model_stream: node event[metadata].get(langgraph_node, ) if node in FINAL_NODES: yield event[data][chunk].content但这里又有个新问题emotion_handle和action_execute都会输出内容用户会看到两段拼接的回复。后来我改成只在aggregator节点输出子图内部的回复先存到 State由 aggregator 统一输出。这样流式输出的内容就是最终回复不会拼接错乱。7. 几个让系统更稳的工程细节7.1 子图的超时与降级子图内部如果有外部 API 调用比如查订单、查物流一定要设超时。我遇到过物流接口挂了子图卡在那里整个会话超时。后来给每个子图加了超时控制import asyncio async def call_subgraph_with_timeout(subgraph, state, timeout5.0): try: return await asyncio.wait_for(subgraph.ainvoke(state), timeouttimeout) except asyncio.TimeoutError: return {response: 系统繁忙请稍后再试, need_human: True}超时后降级到人工比让用户干等好。7.2 状态字段的命名规范多子图系统里字段命名混乱是维护噩梦。我定了一套规范主图字段用snake_case不加前缀。子图字段加子图前缀比如售后子图用as_开头as_refund_amount订单子图用od_开头。布尔字段用is_或has_开头。时间字段统一用 ISO 8601 字符串不用 datetime 对象避免序列化问题。这套规范看起来啰嗦但在调试时能一眼看出字段属于哪个子图省了很多时间。7.3 子图的独立测试子图独立的好处之一是可以单独测试。我给每个子图写了测试用例不经过主图直接调用def test_aftersales_refund_within_7_days(): state AfterSalesState( user_query我要退款, order_info{created_at: 2026-01-10T10:00:00, category: normal, refunded: False}, policy_result{}, refund_amount0.0, emotion_levelnormal, action_taken, response ) result aftersales_subgraph.invoke(state) assert result[policy_result][passed] is True assert result[action_taken] refund这种测试跑得飞快不用启动整个系统改子图逻辑时先跑单测通过了再集成测试效率高很多。7.4 日志与可观测性多智能体系统最怕的是“不知道哪一步出了问题”。我在每个子图的入口和出口都加了结构化日志import structlog logger structlog.get_logger() def policy_check_node(state): logger.info(policy_check_start, user_querystate[user_query][:50]) result do_check(state) logger.info(policy_check_end, passedresult[policy_result][passed]) return result日志里带上session_id排查时按会话过滤能完整还原一次对话的所有节点执行情况。这个投入在出问题时回报巨大。8. 关于子图模式适用边界的一些个人判断子图模式不是银弹。我试过把闲聊也拆成子图结果发现闲聊就一个节点拆成子图反而多了一层映射开销得不偿失。后来把闲聊直接做成主图的一个节点不走子图。我的判断标准是子图内部的节点数超过两个或者子图需要独立复用才值得拆。如果只是一个节点的事直接放主图更简单。另外子图嵌套子图子图里再挂子图我也试过LangGraph 支持但调试复杂度指数上升。除非业务确实需要三层结构否则两层主图子图足够覆盖绝大多数客服场景。还有一点子图之间的通信尽量通过主图中转不要让子图直接调用另一个子图。我见过有人让售后子图直接调订单子图查订单结果两个子图的 State 互相依赖又变成了耦合。正确做法是售后子图需要订单信息时通过主图的subgraph_result传递或者让主图先调订单子图再调售后子图。这套系统上线跑了三个月日均处理两千多轮对话自动解决率78%转人工率22%。转人工的里面一半是大额退款审核一半是情绪激动用户。这个数据不算惊艳但比单图版本稳定太多——单图版本每周至少出两次状态污染导致的 bug子图版本三个月没出过架构层面的问题。如果你也在做多智能体客服建议从子图模式起步别走我单图硬扛的弯路。