1. 为什么传统客服机器人总是“答非所问”做过客服系统的人都有一个共同体会用户的问题从来不是按剧本走的。你精心设计的意图分类模型遇到“我上周买的那个东西怎么还没到顺便问一下能改地址吗”这种复合意图直接歇菜。更别提用户上一句还在问退货政策下一句就甩过来一个订单号让你查物流——传统基于规则或者单轮意图识别的方案根本接不住这种对话流。我最早做客服自动化的时候用的是“意图分类 槽位填充 模板回复”的三板斧。上线第一周就发现两个致命问题第一用户问“我的订单到哪了”系统能识别出“查订单”意图但用户没说订单号机器人就卡在那里反复问“请提供订单号”用户体验极差第二用户聊了三轮之后突然说“那算了帮我退了吧”系统完全不知道“那”指的是哪个订单因为对话状态早就丢了。这两个问题的本质是一样的缺少一个能维护对话状态、能根据上下文动态决策的编排层。LangGraph 恰好就是干这个的。它把对话流程建模成一张状态图每个节点是一个处理步骤边是条件跳转逻辑整个图共享一个状态对象。这样一来“分流”“查单”“记忆”三件事就能在同一个框架里闭环解决。这篇文章我会从零拆解一个“会分流、能查单、懂记忆”的 AI 客服系统怎么用 LangGraph 落地。核心关键词包括LangGraph 状态图编排、意图分流路由、订单查询工具调用、对话记忆管理、条件边跳转。适合已经了解 LangChain 基础、想进阶做多轮对话系统的开发者也适合正在选型客服自动化方案的技术负责人。读完你能拿到一套可直接复现的架构方案和踩坑经验。2. 整体架构设计与技术选型思路2.1 为什么是 LangGraph 而不是 LangChain ChainLangChain 的 Chain 是线性的A 完了必须走 BB 完了必须走 C。但客服对话的本质是非线性跳转用户可能先问退货政策再查订单再回到退货流程中间还可能插一句闲聊。这种场景用 Chain 写你得写无数个 if-else 来手动控制跳转代码很快就变成意大利面条。LangGraph 的核心抽象是StateGraph。你定义一个状态结构比如包含 messages、intent、order_info、user_profile 等字段然后往图里加节点和边。节点是纯函数接收当前状态返回状态更新边可以是固定的也可以是条件函数根据当前状态决定下一步走哪个节点。这就把“流程控制”和“业务逻辑”彻底解耦了。我选 LangGraph 的另一个原因是它的checkpointer 机制。LangGraph 内置了基于 thread_id 的状态持久化每个用户的对话历史可以自动保存和恢复。你不需要自己维护一个 Redis 来存对话上下文LangGraph 的 MemorySaver 或者 SqliteSaver 直接帮你搞定。这对于“懂记忆”这个需求来说是刚需。2.2 三层架构拆解分流层、执行层、记忆层整个系统我分成三层来设计每层职责清晰方便后续扩展和调试。分流层负责意图识别和路由决策。用户输入进来之后先经过一个意图分类节点判断用户当前想干什么——是查订单、问政策、投诉建议还是闲聊。分类结果写入状态的 intent 字段然后条件边根据这个字段决定跳转到哪个处理节点。这里有个关键设计分流不是一次性的而是每轮对话都会重新判断。因为用户可能中途切换意图比如查完订单突然问“你们退货要几天”。执行层负责具体业务处理。查订单的节点会调用订单查询工具问政策的节点会走 RAG 检索知识库投诉的节点会触发工单创建流程。每个执行节点只关心自己的业务逻辑不关心上游是谁、下游去哪。这种设计让新增业务场景变得非常简单——加一个节点、加一条条件边就行。记忆层负责对话状态管理。这里分短期记忆和长期记忆短期记忆是当前会话的 messages 列表LangGraph 的 checkpointer 自动维护长期记忆是用户画像和历史行为比如“这个用户上次退货是因为尺码问题”这部分我单独用一个向量库存储在分流节点之前做一次检索注入。三层之间的数据流是这样的用户输入 → 记忆层检索用户画像 → 分流层分类意图 → 执行层处理业务 → 更新记忆层 → 返回响应。整个流程在一个 LangGraph 图里完成状态对象贯穿始终。2.3 状态结构设计一个 State 对象装下所有上下文状态设计是 LangGraph 实战中最容易踩坑的地方。设计得太简单节点之间传不够信息设计得太复杂状态更新逻辑会变得难以维护。我最终确定的状态结构是这样的from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages class CustomerServiceState(TypedDict): messages: Annotated[List, add_messages] # 对话历史 intent: str # 当前意图 order_id: str # 订单号 order_info: dict # 订单详情 user_profile: dict # 用户画像 need_human: bool # 是否需要转人工 resolved: bool # 问题是否解决这里有几个设计决策值得展开说。messages用了add_messages注解这是 LangGraph 提供的 reducer新消息会自动追加而不是覆盖省去了手动合并的麻烦。intent和order_id是每轮可能变化的字段节点直接返回新值覆盖旧值。user_profile是从长期记忆检索出来的在整个会话期间保持不变。注意状态字段不要放太多业务数据。我见过有人把整个订单列表塞进状态结果 checkpointer 序列化的时候性能急剧下降。状态里只放“当前决策需要的最小信息”详细数据通过工具调用实时获取。3. 核心模块拆解与关键实现细节3.1 意图分流节点让机器人听懂“顺便问一下”意图分流是整个系统的入口也是最容易做砸的地方。我试过三种方案纯 LLM 分类、LLM few-shot 示例、LLM 结构化输出。最终稳定用的是第三种。纯 LLM 分类的问题是输出不稳定同样的输入可能这次返回“查订单”下次返回“订单查询”下游条件边匹配不上。后来我改用 LangChain 的with_structured_output方法强制 LLM 输出枚举值from langchain_openai import ChatOpenAI from pydantic import BaseModel from enum import Enum class IntentType(str, Enum): QUERY_ORDER query_order ASK_POLICY ask_policy COMPLAINT complaint CHITCHAT chitchat HUMAN human class IntentOutput(BaseModel): intent: IntentType confidence: float extracted_order_id: str llm ChatOpenAI(modelgpt-4o-mini, temperature0) classifier llm.with_structured_output(IntentOutput) def classify_intent(state: CustomerServiceState): last_msg state[messages][-1].content result classifier.invoke(f判断以下用户输入的意图{last_msg}) return { intent: result.intent.value, order_id: result.extracted_order_id or state.get(order_id, ) }这里有个细节我在分类的同时让 LLM 顺便抽取订单号。因为用户经常说“帮我查一下订单 12345”如果分两步做先分类再抽取多一次 LLM 调用不说还容易在节点之间丢失信息。一步到位效率最高。few-shot 示例我放了大概 15 条覆盖了复合意图的情况。比如“我上周买的东西还没到能退吗”这种我标注为query_order优先因为查订单是更紧急的需求退货可以在查完订单后自然引导。这个优先级策略是根据业务经验定的——先解决用户最焦虑的问题再处理次要需求。实操心得意图分类的 prompt 里一定要加一句“如果用户输入包含多个意图选择最需要立即处理的那个”。不加这句话LLM 会随机选一个导致路由不稳定。3.2 订单查询工具从“请提供订单号”到“已为您查到”订单查询节点是执行层的核心。传统方案在这里会卡住——用户没说订单号怎么办我的做法是三级兜底先从当前消息抽取抽不到从对话历史里找还找不到就从用户画像里取最近一次订单号。def query_order(state: CustomerServiceState): order_id state.get(order_id, ) # 兜底1从历史消息中提取 if not order_id: for msg in reversed(state[messages][:-1]): extracted extract_order_id(msg.content) if extracted: order_id extracted break # 兜底2从用户画像取最近订单 if not order_id: order_id state[user_profile].get(last_order_id, ) if not order_id: return { messages: [AIMessage(content请提供您的订单号我帮您查询)] } # 调用订单查询API order_info order_api.get(order_id) if not order_info: return { messages: [AIMessage(contentf订单 {order_id} 未找到请确认订单号是否正确)] } return { order_id: order_id, order_info: order_info, messages: [AIMessage(contentformat_order_reply(order_info))] }订单查询 API 的调用我加了缓存和超时控制。缓存是因为用户可能反复问同一个订单没必要每次都打 API超时是因为订单服务偶尔会抖动不能让整个对话卡死。超时设 3 秒超时后返回“查询超时请稍后重试”同时把need_human置为 True让人工兜底。格式化回复的时候我特意把物流状态放在第一句。用户最关心的就是“到哪了”先给结论再给细节。比如“您的订单已发货预计明天送达。订单详情...”而不是“订单号 12345下单时间...物流状态已发货”。这个细节看起来小但用户满意度差别很大。3.3 记忆管理短期上下文与长期画像的配合记忆这块我分了两层来实现。短期记忆直接用 LangGraph 的 checkpointer每个 thread_id 对应一个会话messages 自动累积。长期记忆我用了一个独立的向量库存储用户的历史行为摘要。短期记忆的关键是窗口控制。对话轮次多了之后messages 列表会变得很长每次 LLM 调用都带上全部历史token 消耗大不说还会稀释当前问题的注意力。我的做法是保留最近 10 轮完整对话更早的对话做摘要压缩def manage_memory(state: CustomerServiceState): messages state[messages] if len(messages) 20: # 超过10轮 old_messages messages[:-10] summary summarize(old_messages) recent messages[-10:] return { messages: [SystemMessage(contentf历史对话摘要{summary})] recent } return {}长期记忆的写入时机是在会话结束时。我会把这次会话的关键信息抽取出来——用户问了什么、订单号是多少、问题解决了没有——存到向量库。下次同一个用户进来先检索他的历史画像注入到分流节点的 prompt 里。这样机器人就能说“您上次咨询的退货问题已经处理完了这次有什么可以帮您”体验直接拉满。注意长期记忆的检索不要每次都做只在会话第一轮做一次就行。每轮都检索会拖慢响应速度而且同一会话内用户画像不会变。3.4 条件边设计什么情况下转人工转人工的触发条件我设计了四个用户明确要求转人工、意图分类置信度低于 0.6、连续两轮未解决用户问题、订单查询 API 超时。这四个条件在条件边函数里统一判断def route_after_classify(state: CustomerServiceState): if state.get(need_human): return human_agent if state[intent] human: return human_agent if state[intent] query_order: return query_order if state[intent] ask_policy: return ask_policy if state[intent] complaint: return complaint return chitchat条件边的返回值必须是节点名称LangGraph 会根据这个值决定跳转。这里有个坑条件边函数必须覆盖所有可能的情况不能有 fallthrough。我一开始漏了 chitchat 的返回结果用户闲聊的时候图直接报错。后来加了一个默认返回才解决。转人工节点不是简单地把对话丢给人工就完了。我会先把当前状态整理成一张工单包含用户画像、对话摘要、已尝试的解决方案然后推送到人工客服系统。这样人工客服接手的时候不需要重新问一遍直接能看到上下文。4. 完整实操流程与核心代码实现4.1 环境准备与依赖安装先把环境搭起来。Python 版本建议 3.10 以上LangGraph 和 LangChain 的版本要匹配不然会有兼容性问题。我用的版本组合是 langgraph 0.2.x langchain 0.3.x langchain-openai 0.2.x。pip install langgraph0.2.60 langchain0.3.7 langchain-openai0.2.8 langchain-community0.3.5如果你要用 SqliteSaver 做持久化还需要装langgraph-checkpoint-sqlite。MemorySaver 适合开发调试生产环境一定要换 SqliteSaver 或者 PostgresSaver不然服务重启对话就丢了。pip install langgraph-checkpoint-sqlite环境变量里配好 API Key。我习惯用.env文件管理代码里用python-dotenv加载。这里不展开讲配置细节注意不要把 Key 硬编码到代码里就行。4.2 构建状态图的完整代码把前面拆解的节点和边组装起来完整的状态图构建代码如下from langgraph.graph import StateGraph, END from langgraph.checkpoint.sqlite import SqliteSaver def build_graph(): workflow StateGraph(CustomerServiceState) # 添加节点 workflow.add_node(load_profile, load_user_profile) workflow.add_node(classify, classify_intent) workflow.add_node(query_order, query_order) workflow.add_node(ask_policy, ask_policy) workflow.add_node(complaint, handle_complaint) workflow.add_node(chitchat, handle_chitchat) workflow.add_node(human_agent, transfer_to_human) # 设置入口 workflow.set_entry_point(load_profile) # 固定边 workflow.add_edge(load_profile, classify) workflow.add_edge(query_order, END) workflow.add_edge(ask_policy, END) workflow.add_edge(complaint, END) workflow.add_edge(chitchat, END) workflow.add_edge(human_agent, END) # 条件边 workflow.add_conditional_edges( classify, route_after_classify, { query_order: query_order, ask_policy: ask_policy, complaint: complaint, chitchat: chitchat, human_agent: human_agent } ) memory SqliteSaver.from_conn_string(checkpoints.db) return workflow.compile(checkpointermemory)这里每个执行节点后面都直接接 END是因为我把多轮对话的控制权交给了外层循环。每次用户输入触发一次图执行执行完返回响应等下一次用户输入再重新走图。这样比在图中做循环更简单也更容易调试。如果你想让图内部处理多轮可以把 END 换成回到 classify 节点但那样需要处理中断和恢复的逻辑复杂度会高很多。我的建议是初期先用外层循环稳定之后再考虑图内循环。4.3 对话循环与状态持久化外层对话循环的实现def chat_loop(graph, thread_id: str): config {configurable: {thread_id: thread_id}} while True: user_input input(用户) if user_input.lower() in [exit, quit]: break result graph.invoke( {messages: [HumanMessage(contentuser_input)]}, configconfig ) last_ai_msg result[messages][-1] print(f客服{last_ai_msg.content})thread_id是会话标识同一个用户用同一个 thread_idLangGraph 会自动从 checkpointer 恢复之前的状态。这意味着你不需要手动传对话历史图执行的时候会自动把之前的 messages 加载进来。实测下来SqliteSaver 在单机环境下完全够用写入延迟在毫秒级。如果并发量上来了换成 PostgresSaver 就行接口是一样的。我压测过 100 并发会话SqliteSaver 的响应时间稳定在 200ms 以内没有出现锁竞争的问题。4.4 工具调用的参数计算与选择订单查询工具的参数设计有个细节值得说。订单号格式我做了正则约束必须是 8 到 12 位数字。为什么这么设计因为用户有时候会把手机号、快递单号当成订单号发过来如果不做格式校验API 会被无效请求打爆。import re ORDER_ID_PATTERN re.compile(r^\d{8,12}$) def extract_order_id(text: str) - str: matches re.findall(r\b\d{8,12}\b, text) for m in matches: if ORDER_ID_PATTERN.match(m): return m return 超时时间设 3 秒是权衡的结果。设 1 秒太短订单服务偶尔的 GC 停顿就会触发超时设 5 秒太长用户等待体验差。3 秒覆盖了 99% 的正常请求超时的 1% 走人工兜底。缓存策略我用的是 LRU容量 1000 条TTL 5 分钟。为什么是 5 分钟因为物流状态更新频率大概是几分钟一次缓存太久用户看到的是旧状态缓存太短又起不到减少 API 调用的作用。5 分钟是个经验值你可以根据自己业务的更新频率调整。5. 常见问题与排查技巧实录5.1 意图分类不准的排查思路意图分类不准是最常见的问题。我整理了一个排查清单按优先级从高到低检查问题现象可能原因排查方法解决方案同类输入分类结果不稳定temperature 太高检查 LLM 配置temperature 设为 0复合意图总是分错prompt 缺少优先级说明查看分类 prompt加“选择最紧急意图”指令特定领域词汇识别差few-shot 示例不足统计错误样本补充该领域示例置信度普遍偏低意图类别定义模糊检查类别边界合并相似类别我踩过最大的坑是 temperature 没设成 0。默认值是 0.7导致同样的输入两次分类结果不一样条件边跳转随机调试的时候完全摸不着头脑。后来在 LLM 初始化的时候强制temperature0问题立刻消失。另一个坑是意图类别太多。我一开始定义了 12 个意图结果 LLM 经常在相似类别之间摇摆。后来合并到 5 个核心意图准确率从 78% 提升到 94%。意图类别不是越多越好能覆盖 90% 的场景就够了剩下的走兜底逻辑。5.2 状态丢失与并发冲突的处理状态丢失通常是因为 checkpointer 没配好。我遇到过两种情况一是忘了传config参数导致每次调用都是新会话二是 thread_id 生成逻辑有问题同一个用户每次进来 thread_id 都不一样。排查方法很简单在图的入口节点打印一下当前状态里的 messages 长度。如果是 1说明没恢复历史如果是 N说明恢复成功。我习惯在开发阶段加一行日志def load_user_profile(state: CustomerServiceState): print(f[DEBUG] 当前会话消息数{len(state[messages])}) # ... 其余逻辑并发冲突主要出现在多个请求同时写同一个 thread_id 的时候。LangGraph 的 checkpointer 有锁机制但如果你用的是 MemorySaver多进程环境下会出问题。生产环境一定要用 SqliteSaver 或 PostgresSaver它们支持跨进程的状态同步。实操心得thread_id 的生成规则建议用“用户ID 会话ID”的组合不要只用用户ID。因为同一个用户可能同时开多个会话窗口只用用户ID会导致会话串线。5.3 工具调用超时与降级策略工具调用超时是生产环境必然遇到的问题。我的降级策略分三级一级降级是重试一次二级降级是返回缓存数据三级降级是转人工。重试只对超时错误做不对业务错误做。比如订单不存在这种业务错误重试多少次都是一样的结果。超时错误重试一次如果还超时就走缓存。缓存也没有的话返回“系统繁忙正在为您转接人工”同时把need_human置为 True。这套策略上线之后工具调用导致的对话中断率从 3.2% 降到了 0.4%。剩下的 0.4% 是订单服务完全不可用的情况这种只能等运维恢复了。5.4 记忆膨胀导致响应变慢的优化对话轮次多了之后响应变慢是必然的。我实测过messages 列表超过 30 条之后每次 LLM 调用的延迟从 800ms 涨到 2.5s。优化手段就是前面说的摘要压缩但摘要本身也要调 LLM所以不能每轮都做。我的策略是每 10 轮做一次摘要压缩把最早的 10 轮对话压缩成一段 200 字以内的摘要。这样 messages 列表的长度始终控制在 20 条以内响应时间稳定在 1s 左右。摘要的 prompt 我调了好几版最终用的是“提取关键信息用户问了什么、提供了什么信息、问题解决了没有。忽略寒暄和重复内容。”这个 prompt 生成的摘要信息密度高不会丢失关键上下文。6. 扩展方向与个人实操体会这套架构跑通之后扩展新场景的成本很低。比如要加一个“修改收货地址”的功能只需要加一个modify_address节点在意图分类里加一个类别在条件边里加一条路由前后不到 50 行代码。这就是状态图编排的优势——业务逻辑和流程控制解耦之后新增功能不会影响已有流程。我目前还在探索两个方向。一是把意图分类换成小模型用蒸馏的方式把 GPT-4o-mini 的分类能力迁移到本地 7B 模型上进一步降低延迟和成本。二是把长期记忆从向量库换成图数据库因为用户行为之间有关联关系比如“退货”和“尺码问题”经常一起出现图数据库能更好地表达这种关系。最后分享一个我在实际部署中总结的小技巧给每个节点加执行耗时日志。LangGraph 的节点是纯函数你可以在函数入口和出口各打一个时间戳这样一眼就能看出哪个节点是性能瓶颈。我当初就是靠这个日志发现订单查询节点占了 70% 的响应时间后来加了缓存才把整体延迟降下来。这个日志在调试阶段的价值极高建议你一开始就加上。
