企业官网建设流程全解析

做 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_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-wins，messages 如果用普通 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 "langgraph>=0.2" "langchain-openai>=0.1" "langchain-core>=0.2"

如果你不想为每次意图识别调用云端模型，把 ChatOpenAI 换成任意 OpenAI 兼容的本地服务也可以。实测下来，一个 7B 级别的开源模型做意图识别和结果汇总完全够用，售后规则理解反而不依赖模型，因为规则都写在代码里。本地模型只需改两行：

llm = ChatOpenAI( model="qwen2.5:7b", base_url="http://localhost:11434/v1", api_key="not-needed", temperature=0, )

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: str

ORDER_DB 和 FAQ 是两个普通的 Python 字典，我用它们模拟外部服务。状态定义里，子图的 State 字段很少，只保留自己业务需要的字段；父图 State 字段稍多，但每条都对应调度或结果展示的真实需求。

4.3 实现意图识别与三个子图

意图识别节点负责把用户输入转成结构化指令。它会从用户问题里抽取订单号，同时判断是否需要转人工。

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) 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设置为true，reason写清楚原因。" 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 高频报错速查表

我曾遇到一个特别隐蔽的问题：子图 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 就行。这也是多智能体系统最有意思的地方：场景在变，但“父图调度+子图执行”的骨架始终稳定。

最后给个建议：不要一上来就拆一二十个子图。先用“意图识别+一个业务子图+一个兜底转人工”把链路跑通，再逐个增加业务子图。子图模式是给系统减负的，不是给架构添乱的。跑了一个月客服系统之后你会发现，真正让你省心的不是模型多聪明，而是每个子图的边界比上一次又清楚了一分。