☰
LangGraph子图模式实战:多业务线AI客服系统架构与状态隔离
2026/10/8 4:09:26 网站建设 项目流程

多智能体协作这件事,我在过去一年里反复折腾过不少框架。从最早的纯 Prompt 编排,到后来用状态机硬写流程,再到接触 LangGraph 之后才算真正找到一条能落地的路子。今天要聊的这套“子图模式”,是我在做 AI 客服系统时踩了无数坑之后沉淀下来的方案。它解决的核心问题很具体:当一个客服系统需要同时处理售前咨询、订单查询、售后投诉、技术排查这几类完全不同的任务时,你不可能把所有逻辑塞进一个 Agent 里,那样提示词会膨胀到无法维护,工具调用也会互相干扰。子图模式的价值就在于,把每个业务域拆成一个独立的子图,每个子图有自己的状态、自己的工具集、自己的决策逻辑,最后由一个主图来负责路由和汇总。

这套方案适合谁看?如果你已经写过基础的 LangGraph 单图流程,知道 StateGraph 怎么定义、节点怎么连边,但一到多业务线就感觉状态管理混乱、节点爆炸,那这篇内容就是为你准备的。如果你还没接触过 LangGraph,建议先补一下基础概念,否则后面讲子图的状态隔离和消息传递时会有点吃力。整篇内容我会按照实际项目推进的顺序来讲:先讲清楚为什么选子图而不是其他方案,再拆解状态设计这个最容易翻车的地方,然后逐个模块搭建子图,最后讲主图怎么编排和实际跑起来之后遇到的那些坑。

1. 为什么客服系统非要用子图模式不可

1.1 单图硬扛多业务线的真实后果

我最初的做法很直接,一个 StateGraph 走天下。所有工具函数注册在同一个 ToolNode 里,所有业务逻辑用条件边来分流。刚开始只有售前咨询和订单查询两个场景时,跑得还挺顺。但加到第四个业务域的时候,问题集中爆发了。

第一个问题是提示词冲突。售前咨询需要 Agent 热情、主动推荐、引导下单;售后投诉需要 Agent 克制、共情、先安抚再解决。这两种语气放在同一个系统提示里,模型会表现得精神分裂,该热情的时候端着,该克制的时候又过度推销。第二个问题是工具集污染。订单查询工具需要订单号参数,技术排查工具需要设备型号和错误码,当所有工具都暴露给同一个 Agent 时,模型经常选错工具,把订单号传给技术排查接口,直接报参数错误。第三个问题是状态字段爆炸。每个业务域都有自己的中间状态,售前有推荐商品列表,订单有查询结果缓存,售后有工单号,技术有排查步骤记录。这些字段全塞进一个 State 里,类型定义越来越长,调试时根本看不清哪个字段是哪个环节写入的。

实测下来,单图方案在业务域超过三个之后,维护成本呈指数上升。每次改一个业务域的逻辑,都要回归测试其他三个域有没有被影响。

1.2 子图模式到底解决了什么

子图模式的本质是“分而治之”。每个业务域是一个独立的 StateGraph,拥有自己独立的 State 定义、独立的节点集合、独立的工具绑定。主图只负责一件事:根据用户输入判断该走哪个子图,然后把子图的输出汇总成最终回复。

这样做的好处很实在。状态隔离之后,售前子图的 State 里只有推荐相关字段,售后子图的 State 里只有工单相关字段,互不干扰。工具隔离之后,每个子图只绑定自己需要的工具,模型选错工具的概率大幅下降。提示词隔离之后,每个子图可以用完全不同的系统提示,语气和策略各自独立调优。

还有一个容易被忽略的好处:子图可以独立测试。我可以单独跑售前子图,用一组测试用例验证它的推荐逻辑,不需要启动整个客服系统。这在迭代阶段节省的时间非常可观。

1.3 子图与多 Agent 的区别在哪

这里要澄清一个概念。很多人把子图模式和多 Agent 协作混为一谈,其实两者有本质区别。多 Agent 协作通常指多个独立的 Agent 实例通过消息传递来协商完成任务,每个 Agent 有自己的决策循环。而子图模式是在同一个图执行引擎内,把不同的业务逻辑封装成可复用的子图单元,由主图来调度。

区别体现在三个地方。第一,子图共享同一个执行上下文,状态传递是显式的、类型安全的;多 Agent 之间的消息传递往往是自然语言,容易丢失结构化信息。第二,子图的执行是确定性的图遍历,你可以精确控制每一步走哪个节点;多 Agent 的协商过程带有随机性,调试困难。第三,子图模式更容易做权限和边界控制,因为每个子图的输入输出都是明确定义的。

对于客服系统这种需要强流程控制的场景,子图模式比多 Agent 协商更合适。当然,如果你要做的是开放式创意协作,多 Agent 可能更灵活。

2. 状态设计:子图模式最容易翻车的地方

2.1 主图 State 和子图 State 的字段划分原则

状态设计是子图模式的地基,地基没打好,后面全是坑。我总结了一条核心原则:主图 State 只放跨子图共享的字段,子图 State 只放本业务域内部使用的字段。

具体来说,主图 State 需要包含这些字段:用户原始输入、对话历史、路由决策结果、当前活跃子图标识、最终回复。这些字段是所有子图都可能读写的,或者主图路由逻辑必须访问的。

子图 State 则包含各自业务域特有的字段。比如订单子图需要订单查询结果、订单状态、物流信息;售后子图需要工单号、投诉类型、处理进度;技术子图需要设备信息、错误码、排查步骤。

这里有个关键细节:子图 State 必须继承主图 State 的字段定义。LangGraph 的子图机制要求子图的 State schema 是主图 State 的超集或者至少兼容。我通常的做法是定义一个 BaseState 包含共享字段,然后每个子图的 State 继承 BaseState 并追加自己的字段。

from typing import TypedDict, Annotated, Literal from langgraph.graph import MessagesState class BaseState(MessagesState): user_input: str route_decision: str final_response: str class OrderState(BaseState): order_id: str order_status: str logistics_info: str class AfterSalesState(BaseState): ticket_id: str complaint_type: str progress: str

这样定义之后,子图在执行时既能访问共享字段,又能操作自己的专属字段,类型检查也能通过。

2.2 消息历史在子图间的传递陷阱

消息历史的处理是另一个高频翻车点。LangGraph 的 MessagesState 用 add_messages 注解来管理消息列表,但在子图场景下,消息的追加和读取需要特别注意。

我遇到的问题是:主图把用户输入传给子图时,如果直接把完整的对话历史塞进子图的 messages,子图会看到其他业务域的历史消息,导致上下文污染。比如用户在售前聊了半天商品推荐,然后转到售后投诉,售后子图如果看到之前的推荐记录,可能会在回复里莫名其妙地提到商品。

解决方案是在子图入口做消息裁剪。只保留最近若干轮对话和当前用户输入,把其他业务域的历史消息过滤掉。具体实现可以在子图的第一个节点里做预处理:

def prepare_subgraph_input(state: BaseState): recent_messages = state["messages"][-4:] return {"messages": recent_messages}

这个裁剪窗口的大小需要根据业务调整。售前咨询可能需要更多历史来理解用户偏好,售后投诉可能只需要当前这一条。我一般设成 4 到 6 条,实测下来够用且不会引入太多噪声。

2.3 子图输出如何干净地回写主图

子图执行完之后,输出怎么回写到主图 State,这里也有讲究。最直接的做法是把子图的整个 State 返回给主图,但这样会把子图的专属字段也带回来,污染主图 State。

正确的做法是在子图末尾加一个汇总节点,只提取需要回写的字段。比如订单子图执行完查询后,只把 order_status 和 logistics_info 拼成一段自然语言回复,写入 final_response 字段,其他中间字段不往外传。

def summarize_order_result(state: OrderState): summary = f"订单{state['order_id']}当前状态:{state['order_status']},物流信息:{state['logistics_info']}" return {"final_response": summary}

主图拿到 final_response 之后直接返回给用户,不需要关心子图内部经历了什么。这种“黑盒”式的接口设计让主图和子图的耦合度降到最低,后续替换子图实现时主图完全不用改。

3. 逐个搭建业务子图:从售前到技术排查

3.1 售前咨询子图:意图识别与商品推荐链路

售前咨询子图的核心任务是理解用户想买什么,然后给出推荐。这个子图的节点链路我设计成四步:意图澄清、需求提取、商品检索、推荐生成。

意图澄清节点负责判断用户的问题是否足够明确。如果用户只说“我想买个耳机”,这个信息太模糊,需要追问使用场景、预算范围、有线还是无线。这个节点用一个小模型做二分类即可,成本低响应快。

需求提取节点从对话中抽取结构化需求,比如品类、预算区间、关键功能点。这里我用的是带 structured output 的模型调用,直接输出 JSON 格式的需求对象。

商品检索节点根据需求对象查询商品库。这里有个经验:不要把商品库查询做成纯向量检索,客服场景下用户需求往往是硬性条件过滤(预算不超过 500、必须支持降噪),纯向量检索会返回不符合硬条件的结果。我的做法是先做结构化过滤,再在过滤结果里做语义排序。

推荐生成节点把检索结果组织成自然语言推荐。这里要注意推荐数量控制在 3 个以内,超过 3 个用户会选择困难。每个推荐要包含商品名、核心卖点、价格,以及一句“为什么适合你”的个性化说明。

def build_presales_subgraph(): graph = StateGraph(PresalesState) graph.add_node("clarify", clarify_intent) graph.add_node("extract", extract_requirements) graph.add_node("search", search_products) graph.add_node("recommend", generate_recommendation) graph.add_edge("clarify", "extract") graph.add_edge("extract", "search") graph.add_edge("search", "recommend") graph.set_entry_point("clarify") return graph.compile()

3.2 订单查询子图:参数校验与多轮补全

订单查询子图看起来简单,实际上参数校验这块很容易出问题。用户可能直接说“查一下我的订单”,但没给订单号。这时候不能直接报错,要做多轮补全。

我的设计是三个节点:参数校验、参数补全、订单查询。参数校验节点检查 order_id 是否存在且格式合法。如果缺失,进入参数补全节点,向用户追问订单号。如果格式不对,提示用户重新输入。

这里有个细节:订单号格式校验要用正则做前置过滤,但不要过于严格。我见过有的实现要求订单号必须是 18 位纯数字,结果用户输入带字母的订单号直接被拒。实际上订单号格式应该从订单系统动态获取规则,或者用宽松匹配加后端二次校验。

订单查询节点调用后端接口获取订单详情。这里必须做超时和异常处理。后端接口挂了的时候,不能让整个客服系统卡死,要返回“查询服务暂时不可用,请稍后再试”这样的兜底回复。

def validate_order_params(state: OrderState): order_id = state.get("order_id", "") if not order_id: return {"next_action": "ask_for_id"} if not re.match(r"^[A-Za-z0-9]{8,24}$", order_id): return {"next_action": "invalid_format"} return {"next_action": "query"}

3.3 售后投诉子图:情绪识别与工单创建

售后投诉子图是最需要小心处理的。用户带着情绪来,如果回复不当,很容易升级成舆情事件。这个子图的节点设计我加了情绪识别环节。

情绪识别节点判断用户当前的情绪状态:平静、不满、愤怒。平静状态走标准处理流程,不满状态先安抚再处理,愤怒状态直接转人工并创建高优先级工单。

工单创建节点把投诉内容、用户信息、情绪等级打包成工单,写入工单系统。这里要注意工单描述要结构化,不能直接把用户原话扔进去。我的做法是提取投诉类型、涉及订单、具体问题描述、用户诉求四个字段。

安抚回复节点根据情绪等级生成不同语气的回复。愤怒状态下,回复要简短、诚恳、明确给出解决时间承诺,不要长篇大论解释原因,那样只会火上浇油。

def detect_emotion(state: AfterSalesState): emotion = emotion_classifier(state["messages"][-1].content) if emotion == "angry": return {"emotion_level": "high", "next_action": "escalate"} elif emotion == "dissatisfied": return {"emotion_level": "medium", "next_action": "soothe"} return {"emotion_level": "low", "next_action": "standard"}

3.4 技术排查子图:分步引导与知识库检索

技术排查子图的特点是交互轮次多,需要一步步引导用户排查问题。这个子图我用的是状态机式的设计,每个排查步骤是一个节点,根据用户反馈决定下一步走哪个节点。

排查步骤的设计要遵循“从简到繁”的原则。先让用户检查最简单的可能性,比如设备是否重启过、网络是否正常,再逐步深入到复杂排查。每步只让用户做一个操作,不要一次给三个步骤,用户会漏做。

知识库检索节点在标准排查步骤走完之后触发,用用户描述的错误现象去检索知识库,返回匹配的解决方案。这里检索的 query 构造很关键,要把设备型号、错误码、操作场景都拼进去,提高检索命中率。

def build_tech_support_subgraph(): graph = StateGraph(TechSupportState) graph.add_node("identify_device", identify_device) graph.add_node("basic_check", basic_check) graph.add_node("advanced_check", advanced_check) graph.add_node("kb_search", search_knowledge_base) graph.add_conditional_edges("basic_check", route_after_basic) graph.add_conditional_edges("advanced_check", route_after_advanced) return graph.compile()

4. 主图编排:路由决策与子图调度

4.1 路由节点的分类逻辑怎么设计

主图的路由节点决定了整个系统的分流准确性。我的做法是用一个轻量级分类模型,把用户输入映射到四个业务域之一,外加一个“闲聊/其他”兜底类别。

分类模型的提示词要精心设计。不能只给类别名称,要给每个类别配上典型示例和边界说明。比如“订单查询”和“售后投诉”容易混淆,用户说“我买的东西还没到”到底算查询还是投诉?我的界定是:询问进度算查询,表达不满算投诉。这个边界要在提示词里写清楚。

路由节点还要处理多意图的情况。用户可能一句话里既有查询又有投诉:“我的订单三天了还没发货,你们怎么回事?”这时候应该优先路由到售后投诉子图,因为情绪处理优先级更高。我的做法是在分类时加一个优先级规则:投诉类意图优先于查询类意图。

def route_decision(state: BaseState) -> Literal["presales", "order", "aftersales", "tech", "fallback"]: user_msg = state["messages"][-1].content category = classifier(user_msg) priority_map = {"aftersales": 1, "tech": 2, "order": 3, "presales": 4, "fallback": 5} return category

4.2 子图作为节点的注册方式

LangGraph 支持把编译好的子图直接作为主图的节点添加。这个机制用起来很顺手,但有几个细节要注意。

子图注册时,主图会把子图的 State schema 合并进来。如果子图 State 和主图 State 有同名字段但类型不同,会在编译时报错。所以前面强调的 BaseState 继承机制在这里就体现出价值了,所有子图共享同一套基础字段定义,不会出现类型冲突。

子图节点的返回值只会包含子图 State 中定义的字段。主图拿到返回值后,只会更新主图 State 中存在的字段,子图专属字段会被忽略。这个行为是符合预期的,但初次使用时容易困惑,以为子图输出丢了。

from langgraph.graph import StateGraph, END def build_main_graph(): main_graph = StateGraph(BaseState) main_graph.add_node("router", route_decision) main_graph.add_node("presales", presales_subgraph) main_graph.add_node("order", order_subgraph) main_graph.add_node("aftersales", aftersales_subgraph) main_graph.add_node("tech", tech_subgraph) main_graph.add_node("fallback", fallback_handler) main_graph.set_entry_point("router") main_graph.add_conditional_edges("router", lambda x: x["route_decision"], { "presales": "presales", "order": "order", "aftersales": "aftersales", "tech": "tech", "fallback": "fallback" }) main_graph.add_edge("presales", END) main_graph.add_edge("order", END) main_graph.add_edge("aftersales", END) main_graph.add_edge("tech", END) main_graph.add_edge("fallback", END) return main_graph.compile()

4.3 多轮对话中路由的重新触发

客服场景下,用户经常在一个会话里切换业务域。先问售前,然后查订单,最后投诉物流。这就要求主图支持多轮路由,而不是一次路由定终身。

我的做法是在主图里加一个循环机制。每个子图执行完之后,不直接 END,而是回到路由节点重新判断。但这里要加一个终止条件,否则会无限循环。终止条件有两个:一是子图返回了 final_response 且用户没有新的输入,二是循环次数超过阈值。

实现上可以用一个 turn_count 字段来计数,每次路由加一,超过 5 次强制结束并返回当前汇总结果。这个阈值根据业务调整,一般客服场景 5 轮足够覆盖绝大多数情况。

def should_continue(state: BaseState): if state.get("turn_count", 0) >= 5: return "end" if state.get("final_response") and not state.get("needs_followup"): return "end" return "route"

5. 实测中暴露的问题与调优记录

5.1 子图间状态污染的真实案例

上线第一周就遇到一个诡异问题:用户查完订单之后转到售前咨询,售前推荐的商品里居然包含了订单里的商品。排查了半天,发现是订单子图的 order_status 字段被写进了主图 State,售前子图读取 messages 时通过某种路径访问到了这个字段。

根因是子图 State 继承 BaseState 时,如果 BaseState 里定义了可选字段,子图写入后主图会保留。解决方案是在主图 State 里严格区分“共享字段”和“子图专属字段”,子图专属字段不要定义在 BaseState 里,而是定义在各子图自己的 State 扩展里。这样主图 State 里根本不存在这些字段,自然不会污染。

这个坑让我重新审视了状态设计原则:BaseState 只放真正需要跨子图共享的字段,宁少勿多。每多一个字段,就多一份污染风险。

5.2 路由误判的边界情况处理

路由误判主要集中在两个边界:售前和订单的边界、订单和售后的边界。

售前和订单的边界案例:“这个商品有货吗?”这句话既可以理解为售前咨询(询问商品信息),也可以理解为订单相关(库存查询)。我的处理是把这类模糊问题默认路由到售前,因为售前子图可以处理库存查询,而且售前子图的回复更友好。

订单和售后的边界案例:“我的订单怎么还没到?”这句话如果用户语气平静,路由到订单查询;如果带有明显不满,路由到售后。这里情绪识别的准确率直接影响路由效果。我实测下来,纯文本情绪识别的准确率大概在 75% 左右,所以加了一个兜底策略:路由到订单子图后,如果订单子图检测到用户情绪升级,主动转交给售后子图。

def order_subgraph_with_escalation(state: OrderState): result = order_subgraph.invoke(state) if detect_escalation(result["messages"]): return aftersales_subgraph.invoke(result) return result

5.3 子图执行超时的降级方案

子图执行超时是生产环境必须处理的问题。订单查询依赖后端接口,技术排查依赖知识库检索,这些外部依赖都可能超时。

我的降级方案分三层。第一层是子图内部超时,每个外部调用设置 3 秒超时,超时后返回缓存结果或兜底话术。第二层是子图整体超时,主图给每个子图设置 10 秒执行上限,超时后中断子图,返回“正在处理中,请稍后”的提示。第三层是全局超时,整个对话轮次超过 30 秒,强制结束并转人工。

这三层超时机制配合使用之后,系统的可用性从 95% 提升到了 99.5% 以上。关键是每层超时都要有明确的用户提示,不能让用户干等。

5.4 提示词隔离带来的维护便利

最后说一个正面收益。子图模式的提示词隔离,在实际维护中带来的便利超出预期。

以前单图方案时,改一句系统提示要回归测试所有业务域。现在每个子图的提示词独立存放,改售前提示词只影响售前子图,其他子图完全不受影响。我们甚至可以让不同的人负责不同子图的提示词调优,互不干扰。

提示词文件我按子图分目录存放,每个子图目录下有 system_prompt.txt、few_shot_examples.json、tool_descriptions.json 三个文件。这样版本管理清晰,回滚也方便。

prompts/ presales/ system_prompt.txt few_shot_examples.json tool_descriptions.json order/ system_prompt.txt few_shot_examples.json aftersales/ system_prompt.txt few_shot_examples.json tech/ system_prompt.txt few_shot_examples.json

这套结构跑了大半年,子图从 4 个扩展到 7 个,新增业务域只需要复制目录结构、改提示词、注册子图,主图几乎不用动。这种扩展性是我最终选择子图模式而不是其他方案的核心原因。

如果你也在做多业务线的客服系统,我的建议是不要一开始就追求大而全的单图方案,先把业务域拆清楚,每个域用子图独立实现,主图只做路由和汇总。前期多花一点时间设计状态边界,后期能省下大量调试和维护成本。子图之间的状态隔离和消息传递是最容易出问题的地方,建议在开发阶段就写好单元测试,每个子图单独跑通再接入主图。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询