LangChain+LangGraph六模块构建可控智能客服Agent实战

简介基于LangChain与LangGraph的智能客服Agent系统实现代码包面向希望快速搭建智能客服或学习Agent工程化落地的开发者重点解决意图分类、多轮对话状态跟踪、知识库检索、人机转人工决策以及槽位提取等环节可快速跑通完整Demo。资源共27个文件核心为20个Python脚本覆盖意图分类器、对话状态跟踪器、知识检索系统、人机转人工决策模型、LLMbased槽位提取器与API服务层等模块其他文件包括说明文档、知识库JSON、环境配置示例和依赖清单。压缩包整体仅92KB体量小巧便于按模块逐个阅读和调试。目前已有132人浏览学习。项目采用modules、tests、utils分层组织提供多组测试用例与知识库加载脚本并附赠演示项目及说明文件开发者可直接运行Demo也能参考模块划分方式在此基础上扩展自身业务场景。1. LangChain LangGraph 构建智能客服 Agent六个模块拧成一条可控的对话链路客服这个场景真正难的不是让模型“会说话”而是让每一轮对话都知道自己该干什么。这套基于 LangChain 和 LangGraph 的智能客服 Agent 系统把完整链路拆成了六个各司其职的模块意图分类器、对话状态跟踪器、知识检索系统、人机转人工决策模型、LLM 槽位提取器外加一层 API 服务。它不是那种“套壳聊天机器人”而是一个把多轮对话变成可编程决策链路的参考实现适合已经在用 LangChain 写 demo、却卡在“怎么让对话流真正可控”的开发者也适合想搞懂 LangGraph 状态机怎么落到业务场景的工程师。拿到代码先别急着跑把六个模块之间的数据流看清楚后面调参你才知道动哪里。2. 整体架构与 LangGraph 选型为什么状态图比链式调用更适合多轮对话2.1 六个模块怎么串成一条决策链路先看整体数据流。用户一条消息进来先过 API 服务层把请求体翻译成内部结构随后意图分类器判断用户要干什么槽位提取器把话里的关键参数抽出来对话状态跟踪器更新这一会话当前收集到的信息知识检索系统根据意图和参数去查知识库最后人机转人工决策模型看一眼当前状态决定是直接生成回答还是把对话交给人。整个过程不是一条直线中间有反问、有补充、有跳转这才是要用 LangGraph 的根本原因。模块输入输出在链路里的位置意图分类器用户原话意图名 置信度入口节点LLM 槽位提取器用户原话 当前槽位槽位变更操作第二节点对话状态跟踪器槽位变更操作更新后的会话状态第三节点知识检索系统查询语句候选知识片段第四节点人机转人工决策模型完整状态继续回答 / 转人工路由节点API 服务层HTTP 请求结构化响应最外层我见过不少项目一开始用 LangChain 的 SequentialChain 串流程前几轮看着没问题一旦出现“用户先问订单、再补充地址、突然又改口”这种真实场景链式结构的线性弱点立刻暴露。链式调用的前提是每一步的走向在编译期就写死而客服对话的本质是状态驱动的分支循环。LangGraph 把执行流从“链”升级成“图”节点是功能单元边是跳转条件状态是节点间传递的唯一数据载体这让分支、回环、提前终止都变成显式配置。2.2 LangGraph 节点编排StateGraph 与条件路由的最小实现LangGraph 和 LangChain 的分工一句话说清LangChain 提供零件——LLM 封装、Prompt 模板、检索器这些LangGraph 负责装配和调度——哪个零件在什么条件下执行、执行完状态怎么流转。你在代码里会看到 LangChain 的组件被嵌在 LangGraph 的 node 里跑这就是一个系统同时依赖两个框架的原因。如果你只用了 LangChain 的 Chain对话状态只能靠外部变量硬塞而 LangGraph 的 StateGraph 把状态写入图执行的每一次调用里天然支持多轮会话。from typing import TypedDict from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver class AgentState(TypedDict): session_id: str message: str intent: dict # {name: query_order, confidence: 0.92} slots: dict # 已收集槽位如 {order_id: PO2024001} turn_count: int knowledge: list # 检索结果片段 need_human: bool # 是否转人工 reply: str def build_agent_graph(): graph StateGraph(AgentState) graph.add_node(intent_classify, intent_classify_node) graph.add_node(slot_extract, slot_extract_node) graph.add_node(state_update, state_update_node) graph.add_node(knowledge_retrieve, knowledge_retrieve_node) graph.add_node(escalation_decide, escalation_decide_node) graph.add_node(generate_reply, generate_reply_node) graph.set_entry_point(intent_classify) graph.add_edge(intent_classify, slot_extract) graph.add_edge(slot_extract, state_update) graph.add_edge(state_update, knowledge_retrieve) graph.add_edge(knowledge_retrieve, escalation_decide) graph.add_conditional_edges( escalation_decide, route_by_need_human, {answer: generate_reply, human: END} ) graph.add_edge(generate_reply, END) return graph compiled_graph build_agent_graph().compile(checkpointerMemorySaver())这段代码是把整套系统串起来的主干。add_node 注册的是功能模块add_edge 定义必经路径核心在 add_conditional_edgesescalation_decide 节点跑完后会读 need_human 字段决定走 generate_reply 还是直接结束这就是“转人工”的图级实现。compile(checkpointerMemorySaver()) 开启会话级状态持久化调用时通过 config 里的 thread_id 区分不同用户这样状态跟踪器才能跨轮次读到历史数据。还有一个容易忽略的参数是 recursion_limit默认 25 次节点执行上限多轮对话每轮要经过 6 个节点四轮左右就逼近上限了我一般会在 compile 之后显式调高到 100。这里有个血泪经验不要把逻辑写在 add_edge 的静态链接里。所有可能变化的分支都要走条件路由否则每次改需求都要重连整张图。我见过最坑的一次同事把“槽位缺失是否反问”写成固定边加了一个新意图后整条链路全乱。节点只做一件事、状态只通过 state 传递、分支只走条件边这三条约束守住图再大也不乱。最后补一个读这份代码时的边界习惯每个 node 函数只允许 return 它负责的那个字段比如 intent_classify_node 只改 intent不做任何其他写入。原因很直接——LangGraph 的 state 更新是按返回的 key 合并的如果一个节点同时改了 intent 和 slots排查状态问题时你就分不清是谁污染的。有一次状态错乱我查了三个小时最后发现是某个节点多写了一个字段。从那以后我给自己定了个规矩节点的返回 dict 和它名字的职责严格对应。3. 意图分类器 LLM 槽位提取器把用户原话拆成可执行的结构化指令3.1 意图分类器为什么选 LLM few-shot 而不是微调小模型意图分类是整条链路的入口分类错了后面全错。这套系统用的是 LLM 分类器而不是微调模型理由很实际冷启动快意图列表随时能改客服意图通常只有十几个few-shot 足够最关键的是意图分类和槽位提取能复用同一次 LLM 调用省一半延迟和成本。微调小模型的方案适合意图规模大、请求量高的场景但那需要持续的标注和训练管线对大多数业务前期是负担。意图清单我一般按业务动作列这份代码里覆盖六类query_order查订单、after_sales售后/退换、logistics物流咨询、human直接转人工、greeting寒暄、unknown兜底。注意一定要留 unknown 类否则用户随便说一句不相干的话会被硬塞进某个意图后续槽位提取和检索都会跟着跑偏。INTENT_SYSTEM_PROMPT 你是客服意图分类器。只输出 JSON不要多余内容。 意图列表 - query_order: 查询订单状态、订单详情 - after_sales: 退货、换货、退款、售后问题 - logistics: 查询物流、配送进度 - human: 用户明确要求人工、投诉、情绪激烈 - greeting: 问候、开场白 - unknown: 以上都不匹配 规则 1. 每个意图必须返回 confidence范围 0 到 1。 2. 无法确定时返回 unknownconfidence 给 0.5。 3. 示例问「我的订单到哪了」→ query_order。 用户消息{message} def intent_classify_node(state: AgentState) - AgentState: client get_llm_client() resp client.chat.completions.create( modelLLM_MODEL, temperature0, messages[ {role: system, content: INTENT_SYSTEM_PROMPT.format(messagestate[message])}, ], response_format{type: json_object}, ) data json.loads(resp.choices[0].message.content) return {intent: {name: data[intent], confidence: float(data[confidence])}}temperature 必须设 0分类任务里任何随机性都会变成线上不可复现的脏数据。response_format 强制 JSON 输出能省掉大部分解析异常但注意“强制 JSON”依赖模型 API 支持你如果接的是本地部署的模型这个参数可能不生效就得靠后面槽位提取的容错解析兜底。传入的 prompt 里带一条示例是 few-shot 的最简形态实际项目里每个意图给 3 条典型问法置信度会明显稳。落线标准我一般是意图准确率低于 0.9 就先别动模型回去补示例。3.2 LLM 槽位提取器用 schema 约束输出而不是自由文本槽位提取解决的是“把用户话里的参数抠出来”。订单场景通常要维护三个槽位order_id、product_name、issue_type。早期方案是用正则写了三十多条规则还是被各种口语说法打穿后来改成“正则兜底 LLM 主提取”新说法只要 prompt 里有例子就能覆盖这才算消停。SLOT_EXTRACT_PROMPT 你是槽位提取器。根据当前已收集的槽位提取用户这句话里的信息。 已收集槽位{current_slots} 只输出 JSON {operations: [{slot: order_id, action: set, value: PO2024001}]} action 取值 - set: 设置或覆盖槽位 - clear: 用户反悔删除该槽位 槽位定义 - order_id: 订单号格式如 PO 开头 - product_name: 商品名称 - issue_type: 问题类型破损/发错/漏发 用户消息{message} def slot_extract_node(state: AgentState) - AgentState: prompt SLOT_EXTRACT_PROMPT.format( current_slotsjson.dumps(state[slots], ensure_asciiFalse), messagestate[message], ) resp get_llm_client().chat.completions.create( modelLLM_MODEL, temperature0, messages[{role: system, content: prompt}], response_format{type: json_object}, ) raw resp.choices[0].message.content operations robust_json_parse(raw) for op in operations: if op[action] set: state[slots][op[slot]] op[value] elif op[action] clear: state[slots].pop(op[slot], None) return {slots: state[slots]}prompt 里把“当前已收集槽位”传进去非常关键否则用户说“顺便问下退货”时模型不知道 order_id 已经填过可能重复抽取甚至互相覆盖。action 字段是这套设计的亮点之一它把槽位更新从“无脑覆盖”改成“显式操作”用户在上一轮说“不要这个了”时模型可以返回 clear而不是带着旧参数去查知识库。robust_json_parse 是我的兜底函数先 json.loads失败就用正则把花括号里的内容抠出来再解一次再失败就返回空列表让这条消息走 unknown 分支而不是让整张图崩溃。4. 对话状态跟踪与知识检索系统状态机怎么维护、RAG 参数怎么定4.1 对话状态跟踪器状态不只是记录还要能处理改口和指代对话状态跟踪器是整个系统里最容易被低估的模块。很多人以为它就是存一下聊天记录实际上它维护的是“机器当前对用户需求的理解”。在这个系统里状态里至少要有这几样当前意图、已收集槽位、对话轮数、是否已解决、转人工标记。意图和槽位是动态的轮数用于转人工决策resolved 标记让系统知道问题是否已经答完。LangGraph 的 checkpointer 是这里的技术底座。compile 时挂上 MemorySaver每次调用传入 thread_id图执行完会把 state 快照存下来下一轮从快照继续。生产环境我建议把 MemorySaver 换成 Redis 或 Postgres 的 checkpointer 实现否则服务一重启所有会话状态归零——这在客服场景里等于用户白聊。from langgraph.checkpoint.memory import MemorySaver graph build_agent_graph().compile(checkpointerMemorySaver()) def chat_api(session_id: str, user_message: str): config {configurable: {thread_id: session_id}, recursion_limit: 100} result graph.invoke( { session_id: session_id, message: user_message, slots: {}, turn_count: 0, knowledge: [], need_human: False, }, configconfig, ) return result注意 state_update_node 里轮数的自增逻辑每轮调用先读上一轮快照的 turn_count 再 1而不是在外部手动维护。所有状态变化都发生在图内部节点外部只负责传用户消息这样状态就不会出现“两处修改、一处没同步”的经典问题。指代消解也在这里处理用户说“这个不要了”槽位提取器拿到的是“这个”需要结合上一轮快照里的 product_name 才能正确 clear所以状态跟踪器的更新顺序必须是“先读旧快照、再融合新操作、最后写回”顺序错一次整个会话的理解就歪了。提示MemorySaver 只适合本地调试。上线前一定要换持久化 checkpointer否则服务重启后所有 thread_id 对应的会话状态全部丢失用户会觉得机器人失忆了。4.2 知识检索系统chunk、top_k、score 阈值怎么配合知识检索这块就是常说的 RAG用来回答那些不在模型参数里的业务知识——比如售后政策、物流规则。这套系统的检索链路是根据意图和槽位拼出查询语句拿到向量库召回候选片段过滤低分结果后再拼进生成节点的 prompt。查询语句的拼法有个细节不能只拼用户原话。用户说“你们退货运费谁出”槽位 issue_type退货运费查询就应该是“退货运费 承担 规则”把槽位名词抽出来当关键词召回质量明显好过原文。from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS def build_retriever(embedding_model): splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , ], ) chunks splitter.split_documents(raw_docs) vectorstore FAISS.from_documents(chunks, embedding_model) return vectorstore.as_retriever( search_kwargs{k: 4, score_threshold: 0.45} )这份代码里 FAISS 起步是够用的文档量到百万级再考虑向量数据库。embedding 模型我一般先用 bge-large-zh-v1.5中文客服语料上性价比不错换模型时要重跑一遍索引不能只换 embedding 模型不重建向量库这种错我犯过一次。参数上 chunk_size 500 和 chunk_overlap 50 是个通用起点文档里句子长就调大到 800问答对形式的资料调小到 300目的是保证每个块语义完整。top_k4 再加 score_threshold0.45 是配合使用的只调 top_k 会把低分片段也拉进来只调阈值可能召回不足两个一起收敛到“四条里最多两条可用”的体感最佳。检索结果拼进 prompt 的顺序也有讲究按相似度从高到低排来源文档先给标题再给正文每个片段之间用空行隔开最后加一句“只基于以上资料回答资料里没有就说不知道”。我会在知识检索节点输出前做一道去重同一篇文档的不同 chunk 只保留得分最高的那条避免生成回复时反复引用同一段话。检索节点还有一个容易忽视的联动——查询语句要在状态更新之后生成。因为用户可能是第二轮补充了订单号查询必须带完整槽位否则检索结果还是第一轮稀碎的信息。这个顺序写死在图结构里state_update 之后才是 knowledge_retrieve改图的时候别动。5. 避坑/常见问题/排查五个实测故障与定位修复记录这一章直接给结论。下面五条都是这份资源实际跑起来最常见的故障按现象、原因、解决三条写你遇到类似的可以直接照单抓药。5.1 意图与解析类故障坑一现象槽位提取或分类节点报错日志出现 “provider rejected the request schema or tool payload”。原因response_format 或 function calling 的 schema 里定义了非 string 类型比如 integer、array部分模型对 schema 类型支持不完整直接拒绝请求。解决schema 里全部用 string值的内容让 prompt 去约束别指望 JSON Schema 帮你做类型校验同时保留 robust_json_parse 兜底模型返回的内容不规范时自动修复 JSON 而不是抛异常。坑二现象意图置信度普遍在 0.4~0.6 之间打转分类结果经常摇摆。原因few-shot 示例给得太少且没有 negative example模型对一个意图的边界认识模糊自然给不出高分。解决每个意图至少补 3 条典型问法另外给 unknown 类加“用户闲聊、无业务诉求”的示例如果意图数量超过 25 个就别再堆 prompt 了那说明粒度拆得有问题合并同类项更实际。5.2 状态与检索类故障坑三现象用户在多轮里改口“不要这个了”但后面生成回复时还带着旧参数。原因槽位提取是全量覆盖逻辑只认 set 不认 clear改口信息被当作无效输入忽略了。解决按第三章的设计槽位提取输出 operation 列表set 和 clear 都作为显式操作state_update 节点里先执行 clear 再执行 set。这个改动不用动图结构只换内部实现。坑四现象知识检索答非所问甚至两份互相矛盾的资料同时出现在回答里。原因top_k 偏高且没设 score 阈值低相关片段被硬塞进上下文多文档召回时不同来源的说法互相打架模型只能硬圆。解决把 search_kwargs 里的 k 降到 4score_threshold 设为 0.45生成节点的 prompt 里明确“资料冲突时以最新文档为准并提示用户存在多种规则”。我一般还会在检索节点输出前按来源文档去重同一文档只留最高分片段。5.3 执行与服务类故障坑五现象某轮对话突然整体失败日志只有 “agent execution terminated due to error”。原因LangGraph 图里某个 node 抛了未捕获异常图执行被中断整轮对话没有返回。这个报错信息本身就是个黑匣子能看到的只有错误两个字。解决给每个 node 包一层 try-except异常时往 state 里写一条兜底话术“这个问题我暂时没理解帮你转人工”并把异常堆栈打进日志同时把 need_human 置为 True。用户话术可以统一日志必须留原始错误否则线上你只能猜。我排查这类问题就三步第一步看日志里挂的是哪个节点第二步把这个节点的输入 state 原样复制到脚本里单独跑一遍看是不是必现第三步看模型返回的原始文本通常问题出在模型返回了空字符串或超长内容。这三步走完九成的执行失败都能定到具体原因。6. 把转人工决策做成可验证的模块阈值标定、API 封装与回归检查转人工决策模型是整个系统里最需要“售后”的模块。它的输入是完整的会话状态输出只有两个分支继续生成回答或者转接人工。好消息是决策逻辑不复杂三个信号足够覆盖大多数场景意图置信度低于该意图的标定阈值对话轮数达到 3 轮且 resolved 标记为 False必需槽位缺失比如查订单没有订单号反问一次用户仍不给。这三个条件任何一个命中就转人工。def escalation_decide_node(state: AgentState) - AgentState: intent_name state[intent][name] intent_conf state[intent][confidence] if intent_conf INTENT_THRESHOLD.get(intent_name, 0.4): return {need_human: True} if state[turn_count] 3 and not state[state].get(resolved): return {need_human: True} required_slots REQUIRED_SLOTS.get(intent_name, []) for slot_name in required_slots: if not state[slots].get(slot_name): return {need_human: True} return {need_human: False}INTENT_THRESHOLD 是按意图标定的阈值表REQUIRED_SLOTS 是每个意图的必备槽位清单。这两个配置比模型本身更值得反复调因为它们直接决定了人工成本。API 服务层把整个图包在 FastAPI 接口后面对外只暴露 /chat 这一个 POST 接口请求体只需要 session_id 和 user_message响应体里带 reply、need_human、slots 三个字段。我在服务层还做了两道防护LLM 调用超时 30 秒连续 3 次失败就强制转人工宁可让用户找人也不能让用户对着转圈等一个永不返回的回答。验证这件事比写代码更值得花时间。我会准备一组 30 条左右的标注会话样本标好每条的正确意图、期望槽位、是否需要转人工。每次改完 prompt 或决策参数强制跑一遍这组样本看三个指标指标计算方式我用的合格线意图准确率意图预测正确的样本占比 0.9槽位 F1槽位抽取与标注的精确率和召回率的调和平均 0.85转人工精确率转人工样本中确实需要转人工的占比 0.6我有一次改了一版 prompt自测怎么问都对上线后发现简单查询订单的请求大量走转人工就是因为阈值没按意图分开标定。从那以后我每次改完 prompt 或阈值都会强制把这 30 条样本跑一遍三个指标不动就不准动代码。这套样本集比任何 review 都管用。希望帮到你。本文还有配套的精品资源点击获取