1. 为什么传统客服机器人总是“答非所问”做过客服系统的人大概都有这种体会用户问“我上周买的那个东西到哪了”传统关键词匹配的机器人要么回一句“请提供订单号”要么直接甩一个帮助中心链接。问题出在哪传统方案把“理解意图”“查询数据”“记住上下文”这三件事揉在一个大模型调用里模型既要判断用户想干嘛又要凭空编出一个订单状态结果就是幻觉频发。LangGraph 这套东西解决的正是这个痛点。它把一次客服对话拆成一张状态图入口节点负责分流意图查单节点专门对接订单接口记忆节点负责维护多轮上下文最后再由一个生成节点把结构化数据翻译成人话。每个节点只干一件事职责边界清晰模型不需要“猜”订单状态因为状态是查出来的不是编出来的。这篇文章适合两类人看一是已经用过大模型 API、想进一步做可控 Agent 的开发者二是正在被客服机器人准确率折磨、想找一套可落地架构的技术负责人。我会把整套图的节点设计、状态字段、路由条件、记忆策略全部拆开讲代码可以直接抄。整套方案的核心关键词就三个分流、查单、记忆下面逐个击破。先说清楚一个前提LangGraph 不是替代大模型而是给大模型套上一个“流程骨架”。你可以把它理解成给一个很聪明但容易走神的实习生配了一份 SOP 手册他每做一步都要对照手册确认下一步去哪而不是凭感觉自由发挥。这个类比贯穿全文理解了它后面所有设计都顺了。2. 整体架构设计一张图讲清分流、查单、记忆怎么协作2.1 状态图的核心思路与节点划分LangGraph 的核心抽象是StateGraph你定义一个状态结构State然后往里塞节点Node和边Edge。节点是处理单元边决定流转方向。整个客服图我设计成五个节点intent_router意图分流判断用户这句话属于哪类诉求是查订单、问售后、还是闲聊。order_lookup订单查询拿着订单号去查真实数据返回结构化结果。memory_update记忆更新把本轮对话的关键信息写进长期记忆。response_gen回复生成把查询结果和上下文组装成自然语言。fallback兜底意图不明或查询失败时的兜底处理。为什么要把分流单独拎出来因为如果让主模型同时判断意图和生成回复它会在“我该查单还是该闲聊”这件事上摇摆。实测下来单独一个分流节点用轻量模型甚至规则小模型就能做到 95% 以上的准确率成本还低。这就是关注点分离的价值。2.2 状态字段设计让数据在图里流动状态是整个图的“共享内存”每个节点都能读写。我的状态定义大致长这样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 # 是否转人工这里有个关键点messages用了add_messages这个 reducer意味着每次节点返回新消息时会追加而不是覆盖。这是 LangGraph 处理多轮对话的标准做法。而order_info这类字段是覆盖式的因为每次查询都应该拿最新结果。注意状态字段不要设计得太宽泛。我见过有人把所有东西塞进一个context字典结果节点之间互相污染调试时根本不知道谁改了啥。字段粒度要跟节点职责对齐。2.3 边的设计条件路由是灵魂普通边是固定的 A→B条件边conditional edge才是让图“会分流”的关键。分流节点执行完后用一个路由函数决定下一步def route_by_intent(state: CustomerServiceState) - str: if state[intent] order_query: return order_lookup elif state[intent] after_sale: return response_gen else: return fallback graph.add_conditional_edges( intent_router, route_by_intent, { order_lookup: order_lookup, response_gen: response_gen, fallback: fallback, } )这个路由函数就是整张图的“交通警察”。它的判断依据来自分流节点写入的intent字段。为什么要用字典显式映射因为这样图的结构一目了然可视化出来清清楚楚出问题也好定位。3. 分流节点意图识别怎么做才准3.1 意图分类的三种方案对比分流是整条链路的第一道关分错了后面全错。我试过三种方案直接上对比方案准确率成本维护难度适用场景纯关键词规则70%左右极低低意图极少的简单场景小模型分类90%左右低中意图固定、样本充足大模型结构化输出95%中低意图复杂、快速迭代我最终选的是大模型 结构化输出但不是让大模型自由发挥而是强制它输出 JSON。用 LangChain 的with_structured_output或者直接约束输出格式都行。核心是给模型一个封闭的意图集合别让它自己发明新意图。3.2 结构化输出的提示词设计分流节点的提示词我打磨了很多版最终稳定下来的结构是这样的INTENT_PROMPT 你是一个客服意图分类器。请判断用户消息属于以下哪一类 - order_query: 查询订单状态、物流、发货时间 - after_sale: 退换货、维修、投诉 - product_qa: 产品功能、使用方法咨询 - chitchat: 闲聊、问候 - unknown: 无法判断 只输出JSON格式{{intent: 类别, order_id: 提取到的订单号或空字符串}} 用户消息{user_message} 注意这里我让模型顺便把订单号也提取出来。为什么因为查单节点需要订单号如果分流时能一起拿到就省了一次模型调用。这是典型的“一次调用多产出”优化。订单号提取用正则兜底模型没提取到就用正则扫一遍双保险。3.3 分流节点的实操心得踩过的坑分享几个。第一别让模型输出中文意图名用英文枚举值因为中文容易有同义词漂移今天输出“查订单”明天输出“订单查询”路由函数就崩了。第二给 unknown 留位置不要强行把每句话都归到某个意图识别不了就进兜底比乱猜强。第三分流节点用便宜模型这个任务不需要 GPT-4 级别的推理小模型完全够用成本能降一个数量级。实操技巧在分流节点加一个置信度字段。如果模型对意图判断没把握直接走兜底或转人工别硬着头皮往下走。我一般让模型输出confidence低于 0.7 就转人工。4. 查单节点把“编数据”变成“查数据”4.1 订单查询的接口对接与参数校验查单节点是整个方案里最能体现价值的地方——它彻底消灭了订单幻觉。这个节点不调用大模型纯代码逻辑拿order_id去查数据库或订单服务接口把结果写进order_info。def order_lookup_node(state: CustomerServiceState) - dict: order_id state.get(order_id, ).strip() if not order_id: return {order_info: {error: missing_order_id}} try: result order_service.query(order_id) return {order_info: result} except OrderNotFound: return {order_info: {error: not_found}} except Exception as e: return {order_info: {error: system_error, detail: str(e)}}看到没这里没有任何模型参与。订单状态是查出来的模型只负责最后把它翻译成人话。这就是数据与语言分离的原则。模型可以胡说八道但它没有机会胡说订单状态因为订单状态根本不经它手。4.2 订单号缺失时的多轮补全真实场景里用户经常第一句不带订单号比如“我的快递怎么还没到”。这时候分流节点提取不到订单号查单节点会返回missing_order_id。怎么办不能直接报错要触发多轮补全。我的做法是在查单节点后加一个条件边如果order_info里是missing_order_id就回到response_gen生成一句“请提供订单号”同时把状态标记为“等待订单号”。下一轮用户输入订单号后分流节点会识别到当前处于等待状态直接把订单号填进去跳过意图识别。def route_after_lookup(state): info state.get(order_info, {}) if info.get(error) missing_order_id: return ask_order_id return response_gen这个“等待状态”是记忆的一部分下面讲记忆节点时会串起来。4.3 查询结果的标准化处理不同订单接口返回的字段名五花八门有的叫status有的叫order_status有的用数字编码。查单节点要负责归一化把结果统一成一套内部字段NORMALIZED { status_text: , # 人类可读状态 logistics: , # 物流信息 estimated_arrival: ,# 预计到达 raw: {} # 原始数据留底 }为什么要留raw因为出问题时你要能回溯原始数据不然排查起来两眼一抹黑。归一化后的字段直接喂给回复生成节点模型不用再去理解各种奇怪的字段名。注意订单接口一定要设超时和重试。我见过因为订单服务偶发慢查询整个客服图卡死的情况。给查询加个 3 秒超时超时就走兜底话术用户体验比干等强。5. 记忆节点让机器人记住“刚才说过什么”5.1 短期记忆与长期记忆的分层记忆这块最容易做糊。我的分法是两层短期记忆就是messages里的对话历史LangGraph 自带的 checkpointer 能自动维护长期记忆是跨会话的用户画像比如“这个用户偏好短信通知”“这个用户是 VIP”存在外部存储里。短期记忆解决“刚才说了啥”长期记忆解决“这个用户是谁”。两者不能混。我见过把用户画像塞进 messages 的做法结果对话历史越来越长token 爆炸模型还容易被历史里的无关信息干扰。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) # 调用时传入 thread_id 区分会话 config {configurable: {thread_id: user_123_session_1}}thread_id是短期记忆的隔离键同一个 thread 内的对话历史自动累积不同 thread 互不干扰。5.2 关键信息抽取与持久化长期记忆需要主动抽取。我在每轮对话结束后跑一个轻量的抽取节点把值得记住的信息写进用户画像def memory_update_node(state): profile state.get(user_profile, {}) # 从本轮对话抽取关键信息 extracted extract_profile_info(state[messages][-2:]) profile.update(extracted) save_to_store(state[user_id], profile) return {user_profile: profile}抽取什么订单号、偏好、投诉记录、VIP 标识。别什么都抽只抽对后续服务有用的。抽取同样可以用小模型做成本可控。5.3 记忆的读取与注入时机记忆写进去还要会用。在回复生成节点我会把用户画像的关键字段拼进提示词def build_response_prompt(state): profile state.get(user_profile, {}) vip_hint 该用户是VIP语气要更客气。 if profile.get(is_vip) else return f{vip_hint} 订单信息{state.get(order_info)} 请用自然语言回复用户不要编造订单信息里没有的内容。 注意最后那句“不要编造”这是给模型的硬约束。虽然订单信息是查出来的但模型仍可能在措辞时添油加醋比如订单明明“运输中”它说成“明天就到”。加一句约束能显著降低这类问题。6. 完整实操从零跑通这张图6.1 环境准备与依赖安装先把环境搭起来。Python 3.10 以上装这几个包pip install langgraph langchain langchain-openai如果你用别的模型服务把langchain-openai换成对应的包即可。LangGraph 本身不绑定模型这点很友好。6.2 图的组装与编译把前面所有节点串起来from langgraph.graph import StateGraph, START, END builder StateGraph(CustomerServiceState) builder.add_node(intent_router, intent_router_node) builder.add_node(order_lookup, order_lookup_node) builder.add_node(memory_update, memory_update_node) builder.add_node(response_gen, response_gen_node) builder.add_node(fallback, fallback_node) builder.add_edge(START, intent_router) builder.add_conditional_edges(intent_router, route_by_intent, { order_lookup: order_lookup, response_gen: response_gen, fallback: fallback, }) builder.add_conditional_edges(order_lookup, route_after_lookup, { ask_order_id: response_gen, response_gen: response_gen, }) builder.add_edge(response_gen, memory_update) builder.add_edge(memory_update, END) builder.add_edge(fallback, END) graph builder.compile(checkpointercheckpointer)编译完可以用graph.get_graph().draw_mermaid()看结构这里只是说明有可视化能力实际部署时按需使用。图能画出来说明结构没写错。6.3 一次完整对话的流转记录跑一次真实对话看状态怎么流用户输入“我上周买的耳机到哪了”intent_router识别为order_query但没提取到订单号。路由到order_lookup返回missing_order_id。路由到response_gen生成“请提供订单号”。memory_update记录“等待订单号”状态。用户输入“订单号 A12345”intent_router检测到等待状态直接填订单号跳过意图识别。order_lookup查到真实数据。response_gen生成“您的耳机正在运输中预计后天到达”。memory_update记录本次查询。整个流程清晰可控每一步都能在日志里看到。这就是图结构的好处——可观测。7. 常见问题与排查技巧实录7.1 分流不准的排查思路分流错了先看模型输出的原始 JSON。我遇到最多的情况是模型把“退货”识别成“查订单”因为两者都涉及订单。解决办法是在提示词里给每个意图加反例比如“退货属于 after_sale不是 order_query”。加了反例后准确率明显提升。7.2 记忆串会话的问题有次测试发现 A 用户的对话历史跑到了 B 用户那里。排查发现是thread_id生成逻辑有 bug用了固定值。记住thread_id必须包含用户标识和会话标识两者缺一不可。7.3 订单查询超时导致图卡死前面提过给查询加超时。补充一点超时后不要重试太多次一次重试足够否则用户等待时间翻倍。超时直接走兜底话术“系统繁忙请稍后再试”同时记录日志告警。问题现象可能原因排查动作分流总进兜底提示词意图集合不匹配检查模型原始输出订单号提取失败用户表述太口语化加正则兜底回复编造物流提示词约束不足加“不得编造”约束记忆不生效thread_id 配置错误检查会话隔离键图执行报错状态字段类型不符检查 TypedDict 定义7.4 成本控制的几个实操技巧分流和抽取用小模型只有回复生成用大模型这一条就能省一半成本。另外短期记忆不要无限累积超过 20 轮就做摘要压缩把早期对话浓缩成一段话既省 token 又不丢关键信息。8. 后续可以怎么扩展这套骨架搭好后扩展性很强。想加“售后工单”流程就加一个after_sale节点和对应的条件边想接人工客服就在兜底节点里加转接逻辑想做多语言就在回复生成节点前加一个语言检测节点。我个人在实际操作中的体会是LangGraph 最大的价值不是让你少写代码而是让复杂对话流程变得可调试、可观测、可迭代。传统那种“一个大模型包打天下”的做法出了问题你根本不知道是哪一步错了。拆成图之后每个节点的输入输出都清清楚楚改哪一步、为什么改心里有数。最后分享一个小技巧给每个节点加执行耗时日志。跑一段时间后你会发现瓶颈在哪优化就有了方向。我这边实测下来分流节点占了 60% 的延迟后来换成更小的模型整体响应时间直接砍半。
