企业官网建设流程全解析

1. 为什么传统客服机器人总是“答非所问”

做过客服系统的人都有一个共同体会：用户的问题从来不是按剧本走的。你精心设计的意图分类模型，遇到“我上周买的那个东西怎么还没到，顺便问一下能改地址吗”这种复合意图，直接歇菜。更别提用户上一句还在问退货政策，下一句就甩过来一个订单号让你查物流——传统基于规则或者单轮意图识别的方案，根本接不住这种对话流。

我最早做客服自动化的时候，用的是“意图分类 + 槽位填充 + 模板回复”的三板斧。上线第一周就发现两个致命问题：第一，用户问“我的订单到哪了”，系统能识别出“查订单”意图，但用户没说订单号，机器人就卡在那里反复问“请提供订单号”，用户体验极差；第二，用户聊了三轮之后突然说“那算了，帮我退了吧”，系统完全不知道“那”指的是哪个订单，因为对话状态早就丢了。

这两个问题的本质是一样的：缺少一个能维护对话状态、能根据上下文动态决策的编排层。LangGraph 恰好就是干这个的。它把对话流程建模成一张状态图，每个节点是一个处理步骤，边是条件跳转逻辑，整个图共享一个状态对象。这样一来，“分流”“查单”“记忆”三件事就能在同一个框架里闭环解决。

这篇文章我会从零拆解一个“会分流、能查单、懂记忆”的 AI 客服系统怎么用 LangGraph 落地。核心关键词包括LangGraph 状态图编排、意图分流路由、订单查询工具调用、对话记忆管理、条件边跳转。适合已经了解 LangChain 基础、想进阶做多轮对话系统的开发者，也适合正在选型客服自动化方案的技术负责人。读完你能拿到一套可直接复现的架构方案和踩坑经验。

2. 整体架构设计与技术选型思路

2.1 为什么是 LangGraph 而不是 LangChain Chain

LangChain 的 Chain 是线性的，A 完了必须走 B，B 完了必须走 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(model="gpt-4o-mini", temperature=0) 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(content=f"订单 {order_id} 未找到，请确认订单号是否正确")] } return { "order_id": order_id, "order_info": order_info, "messages": [AIMessage(content=format_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(content=f"历史对话摘要：{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 langgraph==0.2.60 langchain==0.3.7 langchain-openai==0.2.8 langchain-community==0.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(checkpointer=memory)

这里每个执行节点后面都直接接 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(content=user_input)]}, config=config ) last_ai_msg = result["messages"][-1] print(f"客服：{last_ai_msg.content}")

thread_id是会话标识，同一个用户用同一个 thread_id，LangGraph 会自动从 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 没设成 0。默认值是 0.7，导致同样的输入两次分类结果不一样，条件边跳转随机，调试的时候完全摸不着头脑。后来在 LLM 初始化的时候强制temperature=0，问题立刻消失。

另一个坑是意图类别太多。我一开始定义了 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% 的响应时间，后来加了缓存才把整体延迟降下来。这个日志在调试阶段的价值极高，建议你一开始就加上。