☰
LangGraph实战:用状态图构建可控客服Agent,解决答非所问
2026/10/13 3:21:30 网站建设 项目流程

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(checkpointer=checkpointer) # 调用时传入 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(checkpointer=checkpointer)

编译完可以用graph.get_graph().draw_mermaid()看结构(这里只是说明有可视化能力,实际部署时按需使用)。图能画出来,说明结构没写错。

6.3 一次完整对话的流转记录

跑一次真实对话,看状态怎么流:

用户输入:“我上周买的耳机到哪了”

  1. intent_router识别为order_query,但没提取到订单号。
  2. 路由到order_lookup,返回missing_order_id。
  3. 路由到response_gen,生成“请提供订单号”。
  4. memory_update记录“等待订单号”状态。

用户输入:“订单号 A12345”

  1. intent_router检测到等待状态,直接填订单号,跳过意图识别。
  2. order_lookup查到真实数据。
  3. response_gen生成“您的耳机正在运输中,预计后天到达”。
  4. 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% 的延迟,后来换成更小的模型,整体响应时间直接砍半。

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

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

立即咨询