尧图网页设计 
做 AI 客服系统的时候最让我头疼的不是模型选型而是怎么把多个智能体组织到一起。LangGraph 的“子图模式”是解决这个问题最顺手的方式主图负责调度子图负责具体业务各自维护自己的状态互不干扰。这篇文章把我用 LangGraph 子图模式从零搭建 AI 客服系统的过程完整梳理了一遍包括架构设计、核心代码、踩坑记录目标是让准备把单智能体拆成多智能体、或者已经在用 LangGraph 但没碰过子图的朋友看完就能照着自己的业务复现一套。为什么单独聊“子图”而不是泛泛地聊多智能体因为在真实客服系统里多智能体最难的不是每个 Agent 的模型调用而是分工和状态。子图模式恰好把这两个问题都解决了。这篇文章适合三类读者想给现有客服机器人加能力的后端或算法工程师想做多智能体 POC 但担心过度设计的团队以及单纯想搞清楚 LangGraph 子图状态传递机制的 LangChain 用户。1. 先承认单智能体客服真撑不住真实业务1.1 三个崩溃现场假设我只用一个 Agent 让它既查订单又处理售后还得回答知识库问题第一版勉强能跑但用户只要一句话里同时带“查订单”和“怎么退”模型就开始犯晕。它会拿着售后的语义去调用订单工具或者把知识库回答当成订单状态返回最典型的表现是工具参数被乱填用户问退货模型却把“退货”填进订单号字段最后返回“未查询到该订单”。这不是模型不行是职责没边界。当系统内所有状态都放在同一个 State 里互相可见时A 任务的临时变量就会干扰 B 任务的判断。第二个崩溃现场是提示词越来越长。为了约束模型不乱来我不得不在一个 system prompt 里写下订单状态枚举、退货规则、知识库 FAQ、转人工条件加起来一千多词。提示词一长模型对重点信息的敏感度会下降很多字段开始被选择性忽略。更麻烦的是产品改了一次售后规则我得在那一大段提示词里定位、修改、再回归测试成本越来越高。这里的问题同样是不同业务域的规则混在同一个上下文里相当于把几个部门的工作手册装订成一本员工翻页找规则效率怎么可能高。第三个崩溃现场是 State 设计。单图模式下所有中间变量都堆在一个状态字典里检索到的文档片段、临时订单号、上一次工具调用的返回值甚至还有前端传进来的用户消息。这些字段对当前任务大部分时候是无用的但都会被塞进下一轮模型调用的上下文。我见过最离谱的一次是 RAG 检索出来的历史工单内容被模型当成当前订单的售后记录回复给用户。单智能体不是说完全不能做客服而是当业务超过三个功能域的时候整个系统的复杂度和维护成本会急剧上升。1.2 子图模式把“大图”拆成“部门”子图模式的做法很直接把一张几十个节点的大图拆成一个只有调度逻辑的父图和多个业务子图。父图保留的节点只有意图识别、路由、最终回复真正的订单查询逻辑、售后判断逻辑、知识库检索逻辑全部下沉到各自的子图里。这和真实客服中心的组织结构几乎一模一样前台接待判断来意转给对应班组订单组只管订单售后组只管退款他们各有自己的工作流只通过工单系统交换结果不共享办公桌。子图模式带来的核心价值有三点。第一是职责隔离每个子图的提示词短小专一订单子图不需要知道退货规则售后子图也不需要处理物流查询。第二是状态隔离子图有自己的私有 State内部临时变量不会泄漏回父图父图也不会被一堆无关字段挤爆。第三是可独立测试子图可以单独编译、单独 invoke、单独出问题定位父图对子图内部实现完全无感。2. 系统架构设计父图调度子图干活2.1 父图承担什么不承担什么父图本质是一个路由加汇总层我给它定了三个职责意图识别、子图调度、结果汇总。它不承担任何具体业务逻辑不直接调订单 API不写售后规则不碰知识库索引。父图的 State 只需要定义跟调度有关的字段用户输入、意图、是否需要人工、最终答复再加上几个用于接收子图返回结果的字段。这个约束很重要它强迫你思考哪些状态是整个系统必须知道的哪些只是一个子图内部的“私事”。订单子图内部查到的物流单号父图需要知道吗不需要父图只关心最终给用户说什么。私事留在子图里系统才不会越跑越乱。我见过不少人把父图 State 设计成“所有字段都放这里”结果子图模式和单图模式没有任何区别状态照样互相污染。2.2 四个子图的职责边界这次我按客服业务的典型场景拆了四个子图不是越多越好每个子图必须对应一个明确的业务域。子图名称核心输入核心输出触发意图order_subgraphorder_idorder_resultorder_statusaftercare_subgraphorder_id, user_descaftercare_result, aftercare_need_humanafter_saleknowledge_subgraphquestionanswerknowledgehuman_handoffreasonfinal_answerhumanorder_subgraph 只负责一件事查订单状态和物流信息内部不处理退换货。aftercare_subgraph 只负责售后判断比如破损、质量问题、无理由退货判断完给出一段处理话术如果问题太复杂就标记需要人工。knowledge_subgraph 负责常规问答营业时间、退换货周期、发票这些高频问题用的是很轻量的 FAQ 检索不需要上完整 RAG。human_handoff 是兜底用户发脾气、迷惑提问、模型识别不了意图都走这里。2.3 State 设计子图之间怎么传数据又互不污染先说一个结论父图 State 和子图 State 之间字段名一致是关键。LangGraph 把一个编译好的子图挂到父图节点上时默认行为是“同名映射”子图的 State 里声明了哪些字段它就会尝试从父图 State 里取同名字段作为输入子图返回的字典里有哪些字段也会更新父图 State 中的同名字段。所以设计时我让子图只声明自己需要的输入和输出例如 OrderState 只有 order_id 和 order_result它不需要知道 messages、final_answer 这些父图字段也接触不到。这一点和函数参数的设计非常像子图的 State 就是函数签名签名越精简调用方越不容易传错。class OrderState(TypedDict): order_id: str order_result: str class AftercareState(TypedDict): order_id: str user_desc: str aftercare_result: str aftercare_need_human: bool class KnowledgeState(TypedDict): question: str answer: str class MainState(TypedDict): messages: Annotated[list, add_messages] user_input: str intent: str order_id: str user_desc: str need_human: bool reason: str order_result: str aftercare_result: str aftercare_need_human: bool answer: str final_answer: str为什么 messages 用Annotated[list, add_messages]因为 LangGraph 默认对普通字段做 last-write-winsmessages 如果用普通 list每轮返回都会直接覆盖上一轮对话历史加 add_messages reducer返回的新消息就会追加到现有列表而不是覆盖。业务字段如 order_result、aftercare_result 不需要 reducer因为子图返回的就是最终结果覆盖是我们要的行为。3. 核心细节拆解LangGraph 子图模式的几个关键机制3.1 子图就是“能当节点的图”挂载与状态传递把一个子图挂到父图上最直观的写法是这样构建子图compile然后 add_node。order_subgraph build_order_subgraph() main.add_node(order_subgraph, order_subgraph)这里有个容易踩坑的地方如果把未编译的 StateGraph 直接塞进 add_node虽然某些版本可能不报错但后续行为不可控排查问题时会非常痛苦。编译后的子图会完成 schema 校验、channel 初始化、执行计划优化也方便我们单独测试它。实测下来子图内部的所有节点对父图完全不可见父图只能看到子图 State 中声明过的字段这是 LangGraph 内置的一种“信息防火墙”。子图内部的执行路径由 entry_point 和 finish_point 定义。以订单子图为例g.add_edge(START, lookup) g.add_edge(lookup, format) g.add_edge(format, END)这意味着进入子图后先跑 lookup_order_node 查状态再跑 format_order_node 整理话术最后结束返回。子图在结束时会把它最终 State 中声明的字段返回给父图父图对应同名字段就会更新。整个过程很像调用一个函数传入参数返回结果外部不关心函数内部怎么实现。3.2 条件路由意图识别节点怎么把流量分给子图父图启动后第一步永远是意图识别。我用了一个结构化输出模型让 LLM 直接返回一个包含 intent、order_id、need_human、reason 的对象而不是让它输出 JSON 字符串再自己解析。用 with_structured_output 可以绕开正则解析 JSON 的各种脏数据问题模型返回的东西天然是字典。class IntentResult(TypedDict): intent: Literal[order_status, after_sale, knowledge, human] order_id: str need_human: bool reason: str intent_chain llm.with_structured_output(IntentResult) def intent_node(state: MainState) - dict: res intent_chain.invoke( 你是客服系统的意图识别器。 order_status表示查订单after_sale表示售后/退换货knowledge表示咨询常规问题。 如果用户情绪激烈、提到投诉/人工、或意图无法判断need_human设置为true。 f用户问题{state[user_input]} ) return { intent: res[intent], order_id: res[order_id], need_human: res[need_human], reason: res[reason], user_desc: state[user_input], }条件边函数拿到当前 State根据 intent 返回下一个节点名def route_after_intent(state: MainState) - str: if state[need_human] or state[intent] human: return human_handoff return { order_status: order_subgraph, after_sale: aftercare_subgraph, knowledge: knowledge_subgraph, }.get(state[intent], human_handoff)注意最后那个兜底如果 LLM 返回了一个不在枚举里的意图或者干脆乱填就一律转人工。这个兜底特别重要。真实用户会问“你叫什么名字”“你是不是机器人”这类问题模型很可能返回一个未知意图如果没有兜底图就会因为没有可路由的边而直接报错。3.3 reducer 与字段覆盖最容易翻车的状态合并LangGraph 每个字段都是一个 channel默认的 channel 行为是“最后写入生效”。如果子图返回的字段和父图某个普通字段同名就会直接覆盖成子图返回的值。大多数情况下这是我们想要的但有翻车风险。我踩过的一个典型坑是知识库子图内部有一个 question 字段父图 MainState 为了调试方便也放了一个 question 字段保存用户原始问题。结果知识库子图每次运行结束都会把 question 改成它内部的规范化版本导致 final_answer 节点拿到的不是用户原话而是被截断或改写过的字符串。排查了一下午才发现是同名覆盖。解决办法很简单把父图字段改名为 user_input或者不让子图 State 返回 question。所以设计 State 时有一条原则子图返回的字段名尽量不要和父图中语义不同的字段重名。如果要处理列表类的状态比如多轮对话消息必须用Annotated[list, add_messages]这类 reducer 来追加而不是覆盖。reducer 就是一个函数LangGraph 每次写入该字段时会调用它来做合并而不是简单替换。3.4 转人工链路中断与恢复怎么设计客服系统逃不开人工接管。这里有两种方案。简单方案是“假转人工”父图路由到 human_handoff 节点生成一段话术告诉用户正在转接然后结束本次会话。可靠方案是 LangGraph 原生的 human-in-the-loop在子图内部用 interrupt() 挂起图让程序等待外部人工确认后再继续。from langgraph.types import interrupt def human_review_node(state: SomeState): user_request interrupt( { order_id: state[order_id], reason: state[reason], } ) # 恢复后user_request 就是人工写入的审核结果 return {review_result: user_request.get(decision, approve)}interrupt 会抛出一个特殊信号LangGraph 会把图执行状态保存到 checkpointer需要配置内存或数据库存储并把传入的 value 作为 human request 返回给调用方。外部系统拿到这个请求后可以由人工决定同意则调用 update_state() 写入审核结果再 invoke(None, config) 恢复执行拒绝则通过 update_state 改写状态走另一条边。真实生产的工单审批就是这么实现的。为了本地演示方便下面的完整示例里我用 human_handoff 节点模拟转人工不展开 interrupt 的持久化细节。但你要是在真实系统里上线强烈建议把转人工链路换成 interrupt 方案这样用户等待人工期间图的状态不会丢恢复后还能继续走后续流程。4. 实操过程从零搭一个可运行的 AI 客服系统4.1 环境准备与依赖安装我用的版本是 langgraph 0.2 以上、langchain-openai 0.1 以上。安装命令pip install langgraph0.2 langchain-openai0.1 langchain-core0.2如果你不想为每次意图识别调用云端模型把 ChatOpenAI 换成任意 OpenAI 兼容的本地服务也可以。实测下来一个 7B 级别的开源模型做意图识别和结果汇总完全够用售后规则理解反而不依赖模型因为规则都写在代码里。本地模型只需改两行llm ChatOpenAI( modelqwen2.5:7b, base_urlhttp://localhost:11434/v1, api_keynot-needed, temperature0, )4.2 定义状态与工具函数先把最外围的状态定义好。为了代码可运行我模拟了一个订单数据库和一个 FAQ 库真实业务中把字典换成数据库查询或 API 调用即可。from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI ORDER_DB { DD20250101: {status: 已发货, detail: 顺丰SF123456预计3天后送达}, DD20250102: {status: 待付款, detail: 订单尚未完成支付请在24小时内完成}, DD20250103: {status: 已完成, detail: 订单已完成如需售后请描述问题}, } FAQ { 营业时间: 客服工作时间为工作日9:00-18:00周末仅处理紧急问题, 退换货周期: 确认收货后7天内可申请无理由退货15天内可换货, 发票: 订单完成后可在订单详情页申请电子发票, } class IntentResult(TypedDict): intent: Literal[order_status, after_sale, knowledge, human] order_id: str need_human: bool reason: str class OrderState(TypedDict): order_id: str order_result: str class AftercareState(TypedDict): order_id: str user_desc: str aftercare_result: str aftercare_need_human: bool class KnowledgeState(TypedDict): question: str answer: str class MainState(TypedDict): messages: Annotated[list, add_messages] user_input: str intent: str order_id: str user_desc: str need_human: bool reason: str order_result: str aftercare_result: str aftercare_need_human: bool answer: str final_answer: strORDER_DB 和 FAQ 是两个普通的 Python 字典我用它们模拟外部服务。状态定义里子图的 State 字段很少只保留自己业务需要的字段父图 State 字段稍多但每条都对应调度或结果展示的真实需求。4.3 实现意图识别与三个子图意图识别节点负责把用户输入转成结构化指令。它会从用户问题里抽取订单号同时判断是否需要转人工。llm ChatOpenAI(modelgpt-4o-mini, temperature0) intent_chain llm.with_structured_output(IntentResult) def intent_node(state: MainState) - dict: user_text state[user_input] res intent_chain.invoke( 你是客服系统的意图识别器。 order_status表示查订单after_sale表示售后/退换货knowledge表示咨询常规问题。 如果用户情绪激烈、提到投诉/人工、或意图无法判断need_human设置为truereason写清楚原因。 f用户问题{user_text} ) return { intent: res[intent], order_id: res[order_id], need_human: res[need_human], reason: res[reason], user_desc: user_text, }订单子图内部很简单查订单字典然后格式化结果。def get_order_status(order_id: str) - str: info ORDER_DB.get(order_id.upper()) if not info: return f未查询到订单 {order_id}请核对订单号是否填写正确。 return f订单{order_id}当前状态{info[status]}{info[detail]}。 def lookup_order_node(state: OrderState) - dict: return {order_result: get_order_status(state[order_id])} def build_order_subgraph(): g StateGraph(OrderState) g.add_node(lookup, lookup_order_node) g.add_edge(START, lookup) g.add_edge(lookup, END) return g.compile()售后子图要处理两种典型场景商品质量问题走质检通道普通不喜欢走无理由退货其他复杂情况标记人工。def judge_aftercare_node(state: AftercareState) - dict: order_id state[order_id].upper() desc state.get(user_desc, ) if 破损 in desc or 质量问题 in desc: return { aftercare_result: ( f订单{order_id}经商品质量检测后可走‘质量问题退货’通道 需要您上传商品破损照片和订单截图质检通过后运费由商家承担。 ), aftercare_need_human: False, } if 退货 in desc or 不喜欢 in desc: return { aftercare_result: ( f订单{order_id}当前支持7天无理由退货请您在订单详情页点击‘申请售后’ 填写退货原因并发起申请。 ), aftercare_need_human: False, } return { aftercare_result: 你的问题比较复杂我已提交人工客服处理请保持电话畅通。, aftercare_need_human: True, } def build_aftercare_subgraph(): g StateGraph(AftercareState) g.add_node(judge, judge_aftercare_node) g.add_edge(START, judge) g.add_edge(judge, END) return g.compile()知识库子图做的是轻量 FAQ 检索。真实系统里可以换成向量检索或者 RAG但第一版用关键词命中就足够等数据量大了再升级。def search_faq_node(state: KnowledgeState) - dict: question state[question] if 营业时间 in question: ans FAQ[营业时间] elif (退货 in question) or (换货 in question): ans FAQ[退换货周期] elif 发票 in question: ans FAQ[发票] else: ans 这个问题我还在学习建议稍后在帮助中心查看。 return {answer: ans} def build_knowledge_subgraph(): g StateGraph(KnowledgeState) g.add_node(search, search_faq_node) g.add_edge(START, search) g.add_edge(search, END) return g.compile()4.4 组装父图并跑通三条会话父图节点包含:意图识别、三个业务子图、人工兜底、最终回复。我把路由分成两段第一段根据意图选择业务子图第二段根据子图返回结果决定是转人工还是直接回复。def route_after_intent(state: MainState) - str: if state[need_human] or state[intent] human: return human_handoff return { order_status: order_subgraph, after_sale: aftercare_subgraph, knowledge: knowledge_subgraph, }.get(state[intent], human_handoff) def route_after_business(state: MainState) - str: if state.get(need_human) or state.get(aftercare_need_human): return human_handoff return final_answer def human_handoff_node(state: MainState) - dict: return { final_answer: 当前问题已转接人工客服预计排队1-2分钟。工单ID已生成请留意短信通知。 } def final_answer_node(state: MainState) - dict: if state.get(order_result): return {final_answer: state[order_result]} if state.get(aftercare_result): return {final_answer: state[aftercare_result]} if state.get(answer): return {final_answer: state[answer]} if state.get(final_answer): return {final_answer: state[final_answer]} return {final_answer: 感谢咨询再见。} main StateGraph(MainState) main.add_node(intent, intent_node) main.add_node(order_subgraph, build_order_subgraph()) main.add_node(aftercare_subgraph, build_aftercare_subgraph()) main.add_node(knowledge_subgraph, build_knowledge_subgraph()) main.add_node(human_handoff, human_handoff_node) main.add_node(final_answer, final_answer_node) main.add_edge(START, intent) main.add_conditional_edges( intent, route_after_intent, { order_subgraph: order_subgraph, aftercare_subgraph: aftercare_subgraph, knowledge_subgraph: knowledge_subgraph, human_handoff: human_handoff, }, ) main.add_conditional_edges( order_subgraph, route_after_business, {human_handoff: human_handoff, final_answer: final_answer}, ) main.add_conditional_edges( aftercare_subgraph, route_after_business, {human_handoff: human_handoff, final_answer: final_answer}, ) main.add_conditional_edges( knowledge_subgraph, route_after_business, {human_handoff: human_handoff, final_answer: final_answer}, ) main.add_edge(human_handoff, final_answer) main.add_edge(final_answer, END) app main.compile()执行测试时我写了三个典型的用户问题查订单、退换货、问营业时间外加一个转人工请求。def run_case(user_input: str) - str: result app.invoke( { user_input: user_input, messages: [], }, {configurable: {thread_id: case-001}}, ) return result[final_answer] print(run_case(帮我查一下DD20250101的物流)) print(run_case(DD20250102这个订单不要了想退货)) print(run_case(你们的营业时间是)) print(run_case(我不满意叫人工))预期输出查询物流会返回“订单DD20250101当前状态已发货顺丰SF123456预计3天后送达。”退货售后子图会判断为无理由退货返回 7 天无理由退货的操作指引。营业时间知识库子图返回工作时间和紧急处理说明。叫人工意图识别把 need_human 置为 true父图直接路由到 human_handoff返回转接话术。4.5 执行结果与数据流复盘第一次 invoke 时数据流是这样的intent 节点返回 intent、order_id、need_human 等字段到父图 State条件边根据 intent 路由到对应子图。以“帮我查一下DD20250101的物流”为例order_subgraph 运行时它的 OrderState 从父图 State 中拿到 order_id查完订单后返回 order_result。因为 MainState 和 OrderState 里有同名字段order_result 自动合并回父图。最终 final_answer 节点从 State 中读出 order_result作为最终回复输出。这里有个容易被忽略的细节order_subgraph 作为节点运行时它只会读取父图 State 中 OrderState 声明过的字段也就是 order_id。MainState 里的 user_input、messages、need_human 等字段订单子图完全看不到。这就是子图模式的状态隔离也是排查问题时最需要记住的一点如果子图没有拿到预期输入先检查子图 State 到底声明了哪些字段。5. 常见问题与排查技巧实录5.1 高频报错速查表报错或现象原因解决方案KeyError: order_id 或 Missing required keys父图 State 没有 order_id子图 State 声明了但它取不到在父图 State 中增加同名字段或用状态映射显式传值RecursionError: Maximum recursion depth exceeded条件边设计成环或模型反复返回同一意图导致无限循环在路由状态中加入次数计数器或设计有限步流程子图没有任何输出entry_point/finish_point 配错或传入了未编译的 StateGraph确保子图已 compile且 START 和 END 有完整边父图字段被莫名覆盖子图返回字段与父图普通字段同名触发 last-write-wins把父图同名字段改名或使用 Annotated reducerwith_structured_output 返回空或报错模型对 schema 理解不够或温度设置过高换更强的模型temperature 设为 0必要时重试两次历史对话丢失每轮 invoke 都传入空 messages或字段没用 add_messagesmessages 用 add_messages reducer并检查 checkpointer 配置我曾遇到一个特别隐蔽的问题子图 State 里定义的字段在父图 State 中不存在运行时不会立刻报错而是等到该子图真正执行时才抛 KeyError。排查这类问题最快的方法是直接单独 invoke 子图传入一份构造好的输入看它是否正常返回。5.2 调试与观测技巧让子图的行为“透明化”子图模式虽然隔离了职责但也带来一个副作用子图内部发生的事情对父图不可见。所以调试时不要只盯着最终输出要把子图单独拉出来跑。# 单独测试订单子图 order_graph build_order_subgraph() result order_graph.invoke({order_id: DD20250101}) print(result[order_result])这样能快速定位问题是出在子图内部还是出在父图的路由和状态传递上。如果子图单独跑没问题重点检查父图 State 里有没有对应字段如果子图单独跑就报错问题大概率在子图的节点函数或 State 定义里。第二个建议是善用 LangSmith 的追踪能力。开启 LangSmith 后每次 invoke 都能看到完整调用链父图节点、子图内部节点、每个节点的输入输出、每一步模型调用。子图模式下的调用链层级会比较深但恰恰因为层级清晰出问题时一眼就能看到是哪个子图的哪个节点断了。第三个建议是玩熟 get_state 和 update_state。配置了 checkpointer 之后每次执行中间节点时都可以调用app.get_state(config)查看当前 State 快照。转人工中断后人工处理完毕用app.update_state(config, {need_human: False})修改状态再恢复执行。这套组合拳是排查 human-in-the-loop 流程的必备工具。6. 写在最后子图模式真正教会我的事做了这个客服系统我最大的体会是子图模式真正的价值不是“代码复用”而是“边界清晰”。一个系统里如果有多个智能体最难的不是让每个智能体变聪明而是让它们各司其职、不越界。子图模式下每个子图只需要在乎自己的输入输出契约父图不用理会子图内部有多少临时变量团队也可以并行维护不同子图谁负责订单谁负责售后互不耽误。我还想分享一个扩展方向。客服场景里用户经常直接发图反馈“商品破损”这时候让售后子图去理解图片就不太合适了。这两年多模态模型迭代非常快已经可以做到“拍照即得问题描述是否支持退货”的判断。你可以把多模态模型封装成另一个子图节点挂进来老的售后子图逻辑完全不用动只要把图片字段加进子图 State 就行。这也是多智能体系统最有意思的地方场景在变但“父图调度子图执行”的骨架始终稳定。最后给个建议不要一上来就拆一二十个子图。先用“意图识别一个业务子图一个兜底转人工”把链路跑通再逐个增加业务子图。子图模式是给系统减负的不是给架构添乱的。跑了一个月客服系统之后你会发现真正让你省心的不是模型多聪明而是每个子图的边界比上一次又清楚了一分。 LangGraph子图模式实战：从零搭建可运行的AI客服系统
做 AI 客服系统的时候最让我头疼的不是模型选型而是怎么把多个智能体组织到一起。LangGraph 的“子图模式”是解决这个问题最顺手的方式主图负责调度子图负责具体业务各自维护自己的状态互不干扰。这篇文章把我用 LangGraph 子图模式从零搭建 AI 客服系统的过程完整梳理了一遍包括架构设计、核心代码、踩坑记录目标是让准备把单智能体拆成多智能体、或者已经在用 LangGraph 但没碰过子图的朋友看完就能照着自己的业务复现一套。为什么单独聊“子图”而不是泛泛地聊多智能体因为在真实客服系统里多智能体最难的不是每个 Agent 的模型调用而是分工和状态。子图模式恰好把这两个问题都解决了。这篇文章适合三类读者想给现有客服机器人加能力的后端或算法工程师想做多智能体 POC 但担心过度设计的团队以及单纯想搞清楚 LangGraph 子图状态传递机制的 LangChain 用户。1. 先承认单智能体客服真撑不住真实业务1.1 三个崩溃现场假设我只用一个 Agent 让它既查订单又处理售后还得回答知识库问题第一版勉强能跑但用户只要一句话里同时带“查订单”和“怎么退”模型就开始犯晕。它会拿着售后的语义去调用订单工具或者把知识库回答当成订单状态返回最典型的表现是工具参数被乱填用户问退货模型却把“退货”填进订单号字段最后返回“未查询到该订单”。这不是模型不行是职责没边界。当系统内所有状态都放在同一个 State 里互相可见时A 任务的临时变量就会干扰 B 任务的判断。第二个崩溃现场是提示词越来越长。为了约束模型不乱来我不得不在一个 system prompt 里写下订单状态枚举、退货规则、知识库 FAQ、转人工条件加起来一千多词。提示词一长模型对重点信息的敏感度会下降很多字段开始被选择性忽略。更麻烦的是产品改了一次售后规则我得在那一大段提示词里定位、修改、再回归测试成本越来越高。这里的问题同样是不同业务域的规则混在同一个上下文里相当于把几个部门的工作手册装订成一本员工翻页找规则效率怎么可能高。第三个崩溃现场是 State 设计。单图模式下所有中间变量都堆在一个状态字典里检索到的文档片段、临时订单号、上一次工具调用的返回值甚至还有前端传进来的用户消息。这些字段对当前任务大部分时候是无用的但都会被塞进下一轮模型调用的上下文。我见过最离谱的一次是 RAG 检索出来的历史工单内容被模型当成当前订单的售后记录回复给用户。单智能体不是说完全不能做客服而是当业务超过三个功能域的时候整个系统的复杂度和维护成本会急剧上升。1.2 子图模式把“大图”拆成“部门”子图模式的做法很直接把一张几十个节点的大图拆成一个只有调度逻辑的父图和多个业务子图。父图保留的节点只有意图识别、路由、最终回复真正的订单查询逻辑、售后判断逻辑、知识库检索逻辑全部下沉到各自的子图里。这和真实客服中心的组织结构几乎一模一样前台接待判断来意转给对应班组订单组只管订单售后组只管退款他们各有自己的工作流只通过工单系统交换结果不共享办公桌。子图模式带来的核心价值有三点。第一是职责隔离每个子图的提示词短小专一订单子图不需要知道退货规则售后子图也不需要处理物流查询。第二是状态隔离子图有自己的私有 State内部临时变量不会泄漏回父图父图也不会被一堆无关字段挤爆。第三是可独立测试子图可以单独编译、单独 invoke、单独出问题定位父图对子图内部实现完全无感。2. 系统架构设计父图调度子图干活2.1 父图承担什么不承担什么父图本质是一个路由加汇总层我给它定了三个职责意图识别、子图调度、结果汇总。它不承担任何具体业务逻辑不直接调订单 API不写售后规则不碰知识库索引。父图的 State 只需要定义跟调度有关的字段用户输入、意图、是否需要人工、最终答复再加上几个用于接收子图返回结果的字段。这个约束很重要它强迫你思考哪些状态是整个系统必须知道的哪些只是一个子图内部的“私事”。订单子图内部查到的物流单号父图需要知道吗不需要父图只关心最终给用户说什么。私事留在子图里系统才不会越跑越乱。我见过不少人把父图 State 设计成“所有字段都放这里”结果子图模式和单图模式没有任何区别状态照样互相污染。2.2 四个子图的职责边界这次我按客服业务的典型场景拆了四个子图不是越多越好每个子图必须对应一个明确的业务域。子图名称核心输入核心输出触发意图order_subgraphorder_idorder_resultorder_statusaftercare_subgraphorder_id, user_descaftercare_result, aftercare_need_humanafter_saleknowledge_subgraphquestionanswerknowledgehuman_handoffreasonfinal_answerhumanorder_subgraph 只负责一件事查订单状态和物流信息内部不处理退换货。aftercare_subgraph 只负责售后判断比如破损、质量问题、无理由退货判断完给出一段处理话术如果问题太复杂就标记需要人工。knowledge_subgraph 负责常规问答营业时间、退换货周期、发票这些高频问题用的是很轻量的 FAQ 检索不需要上完整 RAG。human_handoff 是兜底用户发脾气、迷惑提问、模型识别不了意图都走这里。2.3 State 设计子图之间怎么传数据又互不污染先说一个结论父图 State 和子图 State 之间字段名一致是关键。LangGraph 把一个编译好的子图挂到父图节点上时默认行为是“同名映射”子图的 State 里声明了哪些字段它就会尝试从父图 State 里取同名字段作为输入子图返回的字典里有哪些字段也会更新父图 State 中的同名字段。所以设计时我让子图只声明自己需要的输入和输出例如 OrderState 只有 order_id 和 order_result它不需要知道 messages、final_answer 这些父图字段也接触不到。这一点和函数参数的设计非常像子图的 State 就是函数签名签名越精简调用方越不容易传错。class OrderState(TypedDict): order_id: str order_result: str class AftercareState(TypedDict): order_id: str user_desc: str aftercare_result: str aftercare_need_human: bool class KnowledgeState(TypedDict): question: str answer: str class MainState(TypedDict): messages: Annotated[list, add_messages] user_input: str intent: str order_id: str user_desc: str need_human: bool reason: str order_result: str aftercare_result: str aftercare_need_human: bool answer: str final_answer: str为什么 messages 用Annotated[list, add_messages]因为 LangGraph 默认对普通字段做 last-write-winsmessages 如果用普通 list每轮返回都会直接覆盖上一轮对话历史加 add_messages reducer返回的新消息就会追加到现有列表而不是覆盖。业务字段如 order_result、aftercare_result 不需要 reducer因为子图返回的就是最终结果覆盖是我们要的行为。3. 核心细节拆解LangGraph 子图模式的几个关键机制3.1 子图就是“能当节点的图”挂载与状态传递把一个子图挂到父图上最直观的写法是这样构建子图compile然后 add_node。order_subgraph build_order_subgraph() main.add_node(order_subgraph, order_subgraph)这里有个容易踩坑的地方如果把未编译的 StateGraph 直接塞进 add_node虽然某些版本可能不报错但后续行为不可控排查问题时会非常痛苦。编译后的子图会完成 schema 校验、channel 初始化、执行计划优化也方便我们单独测试它。实测下来子图内部的所有节点对父图完全不可见父图只能看到子图 State 中声明过的字段这是 LangGraph 内置的一种“信息防火墙”。子图内部的执行路径由 entry_point 和 finish_point 定义。以订单子图为例g.add_edge(START, lookup) g.add_edge(lookup, format) g.add_edge(format, END)这意味着进入子图后先跑 lookup_order_node 查状态再跑 format_order_node 整理话术最后结束返回。子图在结束时会把它最终 State 中声明的字段返回给父图父图对应同名字段就会更新。整个过程很像调用一个函数传入参数返回结果外部不关心函数内部怎么实现。3.2 条件路由意图识别节点怎么把流量分给子图父图启动后第一步永远是意图识别。我用了一个结构化输出模型让 LLM 直接返回一个包含 intent、order_id、need_human、reason 的对象而不是让它输出 JSON 字符串再自己解析。用 with_structured_output 可以绕开正则解析 JSON 的各种脏数据问题模型返回的东西天然是字典。class IntentResult(TypedDict): intent: Literal[order_status, after_sale, knowledge, human] order_id: str need_human: bool reason: str intent_chain llm.with_structured_output(IntentResult) def intent_node(state: MainState) - dict: res intent_chain.invoke( 你是客服系统的意图识别器。 order_status表示查订单after_sale表示售后/退换货knowledge表示咨询常规问题。 如果用户情绪激烈、提到投诉/人工、或意图无法判断need_human设置为true。 f用户问题{state[user_input]} ) return { intent: res[intent], order_id: res[order_id], need_human: res[need_human], reason: res[reason], user_desc: state[user_input], }条件边函数拿到当前 State根据 intent 返回下一个节点名def route_after_intent(state: MainState) - str: if state[need_human] or state[intent] human: return human_handoff return { order_status: order_subgraph, after_sale: aftercare_subgraph, knowledge: knowledge_subgraph, }.get(state[intent], human_handoff)注意最后那个兜底如果 LLM 返回了一个不在枚举里的意图或者干脆乱填就一律转人工。这个兜底特别重要。真实用户会问“你叫什么名字”“你是不是机器人”这类问题模型很可能返回一个未知意图如果没有兜底图就会因为没有可路由的边而直接报错。3.3 reducer 与字段覆盖最容易翻车的状态合并LangGraph 每个字段都是一个 channel默认的 channel 行为是“最后写入生效”。如果子图返回的字段和父图某个普通字段同名就会直接覆盖成子图返回的值。大多数情况下这是我们想要的但有翻车风险。我踩过的一个典型坑是知识库子图内部有一个 question 字段父图 MainState 为了调试方便也放了一个 question 字段保存用户原始问题。结果知识库子图每次运行结束都会把 question 改成它内部的规范化版本导致 final_answer 节点拿到的不是用户原话而是被截断或改写过的字符串。排查了一下午才发现是同名覆盖。解决办法很简单把父图字段改名为 user_input或者不让子图 State 返回 question。所以设计 State 时有一条原则子图返回的字段名尽量不要和父图中语义不同的字段重名。如果要处理列表类的状态比如多轮对话消息必须用Annotated[list, add_messages]这类 reducer 来追加而不是覆盖。reducer 就是一个函数LangGraph 每次写入该字段时会调用它来做合并而不是简单替换。3.4 转人工链路中断与恢复怎么设计客服系统逃不开人工接管。这里有两种方案。简单方案是“假转人工”父图路由到 human_handoff 节点生成一段话术告诉用户正在转接然后结束本次会话。可靠方案是 LangGraph 原生的 human-in-the-loop在子图内部用 interrupt() 挂起图让程序等待外部人工确认后再继续。from langgraph.types import interrupt def human_review_node(state: SomeState): user_request interrupt( { order_id: state[order_id], reason: state[reason], } ) # 恢复后user_request 就是人工写入的审核结果 return {review_result: user_request.get(decision, approve)}interrupt 会抛出一个特殊信号LangGraph 会把图执行状态保存到 checkpointer需要配置内存或数据库存储并把传入的 value 作为 human request 返回给调用方。外部系统拿到这个请求后可以由人工决定同意则调用 update_state() 写入审核结果再 invoke(None, config) 恢复执行拒绝则通过 update_state 改写状态走另一条边。真实生产的工单审批就是这么实现的。为了本地演示方便下面的完整示例里我用 human_handoff 节点模拟转人工不展开 interrupt 的持久化细节。但你要是在真实系统里上线强烈建议把转人工链路换成 interrupt 方案这样用户等待人工期间图的状态不会丢恢复后还能继续走后续流程。4. 实操过程从零搭一个可运行的 AI 客服系统4.1 环境准备与依赖安装我用的版本是 langgraph 0.2 以上、langchain-openai 0.1 以上。安装命令pip install langgraph0.2 langchain-openai0.1 langchain-core0.2如果你不想为每次意图识别调用云端模型把 ChatOpenAI 换成任意 OpenAI 兼容的本地服务也可以。实测下来一个 7B 级别的开源模型做意图识别和结果汇总完全够用售后规则理解反而不依赖模型因为规则都写在代码里。本地模型只需改两行llm ChatOpenAI( modelqwen2.5:7b, base_urlhttp://localhost:11434/v1, api_keynot-needed, temperature0, )4.2 定义状态与工具函数先把最外围的状态定义好。为了代码可运行我模拟了一个订单数据库和一个 FAQ 库真实业务中把字典换成数据库查询或 API 调用即可。from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI ORDER_DB { DD20250101: {status: 已发货, detail: 顺丰SF123456预计3天后送达}, DD20250102: {status: 待付款, detail: 订单尚未完成支付请在24小时内完成}, DD20250103: {status: 已完成, detail: 订单已完成如需售后请描述问题}, } FAQ { 营业时间: 客服工作时间为工作日9:00-18:00周末仅处理紧急问题, 退换货周期: 确认收货后7天内可申请无理由退货15天内可换货, 发票: 订单完成后可在订单详情页申请电子发票, } class IntentResult(TypedDict): intent: Literal[order_status, after_sale, knowledge, human] order_id: str need_human: bool reason: str class OrderState(TypedDict): order_id: str order_result: str class AftercareState(TypedDict): order_id: str user_desc: str aftercare_result: str aftercare_need_human: bool class KnowledgeState(TypedDict): question: str answer: str class MainState(TypedDict): messages: Annotated[list, add_messages] user_input: str intent: str order_id: str user_desc: str need_human: bool reason: str order_result: str aftercare_result: str aftercare_need_human: bool answer: str final_answer: strORDER_DB 和 FAQ 是两个普通的 Python 字典我用它们模拟外部服务。状态定义里子图的 State 字段很少只保留自己业务需要的字段父图 State 字段稍多但每条都对应调度或结果展示的真实需求。4.3 实现意图识别与三个子图意图识别节点负责把用户输入转成结构化指令。它会从用户问题里抽取订单号同时判断是否需要转人工。llm ChatOpenAI(modelgpt-4o-mini, temperature0) intent_chain llm.with_structured_output(IntentResult) def intent_node(state: MainState) - dict: user_text state[user_input] res intent_chain.invoke( 你是客服系统的意图识别器。 order_status表示查订单after_sale表示售后/退换货knowledge表示咨询常规问题。 如果用户情绪激烈、提到投诉/人工、或意图无法判断need_human设置为truereason写清楚原因。 f用户问题{user_text} ) return { intent: res[intent], order_id: res[order_id], need_human: res[need_human], reason: res[reason], user_desc: user_text, }订单子图内部很简单查订单字典然后格式化结果。def get_order_status(order_id: str) - str: info ORDER_DB.get(order_id.upper()) if not info: return f未查询到订单 {order_id}请核对订单号是否填写正确。 return f订单{order_id}当前状态{info[status]}{info[detail]}。 def lookup_order_node(state: OrderState) - dict: return {order_result: get_order_status(state[order_id])} def build_order_subgraph(): g StateGraph(OrderState) g.add_node(lookup, lookup_order_node) g.add_edge(START, lookup) g.add_edge(lookup, END) return g.compile()售后子图要处理两种典型场景商品质量问题走质检通道普通不喜欢走无理由退货其他复杂情况标记人工。def judge_aftercare_node(state: AftercareState) - dict: order_id state[order_id].upper() desc state.get(user_desc, ) if 破损 in desc or 质量问题 in desc: return { aftercare_result: ( f订单{order_id}经商品质量检测后可走‘质量问题退货’通道 需要您上传商品破损照片和订单截图质检通过后运费由商家承担。 ), aftercare_need_human: False, } if 退货 in desc or 不喜欢 in desc: return { aftercare_result: ( f订单{order_id}当前支持7天无理由退货请您在订单详情页点击‘申请售后’ 填写退货原因并发起申请。 ), aftercare_need_human: False, } return { aftercare_result: 你的问题比较复杂我已提交人工客服处理请保持电话畅通。, aftercare_need_human: True, } def build_aftercare_subgraph(): g StateGraph(AftercareState) g.add_node(judge, judge_aftercare_node) g.add_edge(START, judge) g.add_edge(judge, END) return g.compile()知识库子图做的是轻量 FAQ 检索。真实系统里可以换成向量检索或者 RAG但第一版用关键词命中就足够等数据量大了再升级。def search_faq_node(state: KnowledgeState) - dict: question state[question] if 营业时间 in question: ans FAQ[营业时间] elif (退货 in question) or (换货 in question): ans FAQ[退换货周期] elif 发票 in question: ans FAQ[发票] else: ans 这个问题我还在学习建议稍后在帮助中心查看。 return {answer: ans} def build_knowledge_subgraph(): g StateGraph(KnowledgeState) g.add_node(search, search_faq_node) g.add_edge(START, search) g.add_edge(search, END) return g.compile()4.4 组装父图并跑通三条会话父图节点包含:意图识别、三个业务子图、人工兜底、最终回复。我把路由分成两段第一段根据意图选择业务子图第二段根据子图返回结果决定是转人工还是直接回复。def route_after_intent(state: MainState) - str: if state[need_human] or state[intent] human: return human_handoff return { order_status: order_subgraph, after_sale: aftercare_subgraph, knowledge: knowledge_subgraph, }.get(state[intent], human_handoff) def route_after_business(state: MainState) - str: if state.get(need_human) or state.get(aftercare_need_human): return human_handoff return final_answer def human_handoff_node(state: MainState) - dict: return { final_answer: 当前问题已转接人工客服预计排队1-2分钟。工单ID已生成请留意短信通知。 } def final_answer_node(state: MainState) - dict: if state.get(order_result): return {final_answer: state[order_result]} if state.get(aftercare_result): return {final_answer: state[aftercare_result]} if state.get(answer): return {final_answer: state[answer]} if state.get(final_answer): return {final_answer: state[final_answer]} return {final_answer: 感谢咨询再见。} main StateGraph(MainState) main.add_node(intent, intent_node) main.add_node(order_subgraph, build_order_subgraph()) main.add_node(aftercare_subgraph, build_aftercare_subgraph()) main.add_node(knowledge_subgraph, build_knowledge_subgraph()) main.add_node(human_handoff, human_handoff_node) main.add_node(final_answer, final_answer_node) main.add_edge(START, intent) main.add_conditional_edges( intent, route_after_intent, { order_subgraph: order_subgraph, aftercare_subgraph: aftercare_subgraph, knowledge_subgraph: knowledge_subgraph, human_handoff: human_handoff, }, ) main.add_conditional_edges( order_subgraph, route_after_business, {human_handoff: human_handoff, final_answer: final_answer}, ) main.add_conditional_edges( aftercare_subgraph, route_after_business, {human_handoff: human_handoff, final_answer: final_answer}, ) main.add_conditional_edges( knowledge_subgraph, route_after_business, {human_handoff: human_handoff, final_answer: final_answer}, ) main.add_edge(human_handoff, final_answer) main.add_edge(final_answer, END) app main.compile()执行测试时我写了三个典型的用户问题查订单、退换货、问营业时间外加一个转人工请求。def run_case(user_input: str) - str: result app.invoke( { user_input: user_input, messages: [], }, {configurable: {thread_id: case-001}}, ) return result[final_answer] print(run_case(帮我查一下DD20250101的物流)) print(run_case(DD20250102这个订单不要了想退货)) print(run_case(你们的营业时间是)) print(run_case(我不满意叫人工))预期输出查询物流会返回“订单DD20250101当前状态已发货顺丰SF123456预计3天后送达。”退货售后子图会判断为无理由退货返回 7 天无理由退货的操作指引。营业时间知识库子图返回工作时间和紧急处理说明。叫人工意图识别把 need_human 置为 true父图直接路由到 human_handoff返回转接话术。4.5 执行结果与数据流复盘第一次 invoke 时数据流是这样的intent 节点返回 intent、order_id、need_human 等字段到父图 State条件边根据 intent 路由到对应子图。以“帮我查一下DD20250101的物流”为例order_subgraph 运行时它的 OrderState 从父图 State 中拿到 order_id查完订单后返回 order_result。因为 MainState 和 OrderState 里有同名字段order_result 自动合并回父图。最终 final_answer 节点从 State 中读出 order_result作为最终回复输出。这里有个容易被忽略的细节order_subgraph 作为节点运行时它只会读取父图 State 中 OrderState 声明过的字段也就是 order_id。MainState 里的 user_input、messages、need_human 等字段订单子图完全看不到。这就是子图模式的状态隔离也是排查问题时最需要记住的一点如果子图没有拿到预期输入先检查子图 State 到底声明了哪些字段。5. 常见问题与排查技巧实录5.1 高频报错速查表报错或现象原因解决方案KeyError: order_id 或 Missing required keys父图 State 没有 order_id子图 State 声明了但它取不到在父图 State 中增加同名字段或用状态映射显式传值RecursionError: Maximum recursion depth exceeded条件边设计成环或模型反复返回同一意图导致无限循环在路由状态中加入次数计数器或设计有限步流程子图没有任何输出entry_point/finish_point 配错或传入了未编译的 StateGraph确保子图已 compile且 START 和 END 有完整边父图字段被莫名覆盖子图返回字段与父图普通字段同名触发 last-write-wins把父图同名字段改名或使用 Annotated reducerwith_structured_output 返回空或报错模型对 schema 理解不够或温度设置过高换更强的模型temperature 设为 0必要时重试两次历史对话丢失每轮 invoke 都传入空 messages或字段没用 add_messagesmessages 用 add_messages reducer并检查 checkpointer 配置我曾遇到一个特别隐蔽的问题子图 State 里定义的字段在父图 State 中不存在运行时不会立刻报错而是等到该子图真正执行时才抛 KeyError。排查这类问题最快的方法是直接单独 invoke 子图传入一份构造好的输入看它是否正常返回。5.2 调试与观测技巧让子图的行为“透明化”子图模式虽然隔离了职责但也带来一个副作用子图内部发生的事情对父图不可见。所以调试时不要只盯着最终输出要把子图单独拉出来跑。# 单独测试订单子图 order_graph build_order_subgraph() result order_graph.invoke({order_id: DD20250101}) print(result[order_result])这样能快速定位问题是出在子图内部还是出在父图的路由和状态传递上。如果子图单独跑没问题重点检查父图 State 里有没有对应字段如果子图单独跑就报错问题大概率在子图的节点函数或 State 定义里。第二个建议是善用 LangSmith 的追踪能力。开启 LangSmith 后每次 invoke 都能看到完整调用链父图节点、子图内部节点、每个节点的输入输出、每一步模型调用。子图模式下的调用链层级会比较深但恰恰因为层级清晰出问题时一眼就能看到是哪个子图的哪个节点断了。第三个建议是玩熟 get_state 和 update_state。配置了 checkpointer 之后每次执行中间节点时都可以调用app.get_state(config)查看当前 State 快照。转人工中断后人工处理完毕用app.update_state(config, {need_human: False})修改状态再恢复执行。这套组合拳是排查 human-in-the-loop 流程的必备工具。6. 写在最后子图模式真正教会我的事做了这个客服系统我最大的体会是子图模式真正的价值不是“代码复用”而是“边界清晰”。一个系统里如果有多个智能体最难的不是让每个智能体变聪明而是让它们各司其职、不越界。子图模式下每个子图只需要在乎自己的输入输出契约父图不用理会子图内部有多少临时变量团队也可以并行维护不同子图谁负责订单谁负责售后互不耽误。我还想分享一个扩展方向。客服场景里用户经常直接发图反馈“商品破损”这时候让售后子图去理解图片就不太合适了。这两年多模态模型迭代非常快已经可以做到“拍照即得问题描述是否支持退货”的判断。你可以把多模态模型封装成另一个子图节点挂进来老的售后子图逻辑完全不用动只要把图片字段加进子图 State 就行。这也是多智能体系统最有意思的地方场景在变但“父图调度子图执行”的骨架始终稳定。最后给个建议不要一上来就拆一二十个子图。先用“意图识别一个业务子图一个兜底转人工”把链路跑通再逐个增加业务子图。子图模式是给系统减负的不是给架构添乱的。跑了一个月客服系统之后你会发现真正让你省心的不是模型多聪明而是每个子图的边界比上一次又清楚了一分。