简介面向AI应用开发者和智能客服系统设计人员这套基于LangChain与LangGraph框架构建的智能客服智能体是一份完整实现方案。系统集成了意图分类器、对话状态跟踪器、知识检索系统、人机转人工决策模型、大模型槽位提取器及接口服务层覆盖从用户意图识别、状态维护、知识查询到人工接管决策的全链路流程。压缩包共27个文件以20个Python源码文件为主包含各核心模块的独立实现与测试脚本另有配置示例、说明文档和附赠资料便于按模块理解。整体仅92KB轻量易部署。目前已有132人学习下载演示工程内置知识库加载、状态跟踪、检索测试等可运行代码可帮助开发者快速掌握智能客服智能体的工程结构并按需扩展成生产级系统。1. 智能客服Agent系统从规则脚本到LangGraph状态编排过去十年做智能客服基本是“意图菜单 关键词规则 人工兜底”的三件套用户问一句“我的订单什么时候到”脚本命中“订单”关键词推一条物流模板命中不了就转人工。这套东西上线快但改一次话术要动逻辑多轮对话靠状态机硬编码维护成本逐年走高。基于LangChain和LangGraph框架构建的智能客服Agent系统解决的就是这个痛点把意图分类、对话状态跟踪、知识检索、人机转人工决策这些模块交给LLM做语义理解和槽位提取用LangGraph做状态编排。这套方案适合手里已有知识库或工单体系、想用大模型提升自助解决率的团队。它不是银弹要承担LLM的API调用量要准备一份能召回的业务文档还要有人愿意在prompt和边界阈值上反复调。2. 用LangGraph编排客服对话状态机LangChain与LangGraph的分工边界2.1 LangChain与LangGraph的区别链式调用与状态图的本质差异LangChain擅长把“调用大模型”“调用工具”“取回结果”串成一条链适合请求-响应模式。但智能客服Agent的问题在于多轮对话的状态不是线性的。用户上一秒在查物流下一秒问“那退款呢”意图变了槽位要重新收集槽位没齐要不要追问知识库没命中要不要转人工。每一步的下一步取决于当前状态不是一个固定链条能表达的。LangGraph把流程建模成状态图——节点做计算边决定流转全局状态对象在节点间传递。这也是“LangChain与LangGraph的区别”在Agent开发里最直观的答案前者是执行流后者是状态机。选择LangGraph还有一个现实原因可回滚性好。客服对话里最容易出现“改口”场景——用户先说“查订单”后说“等下我是要退款”。在LangChain链式结构里之前的步骤已经执行完很难回溯LangGraph的状态里保留了步骤历史可以让条件边重新走一遍槽位收集节点。这一点在真实客服场景里非常重要第3章会展开讲。2.2 定义State与节点把意图分类、槽位提取、检索做成可回滚的图节点写一个LangGraph客服Agent第一步是定义全局State。常见做法是定义一个TypedDict或Pydantic类字段包括对话历史、当前意图、槽位字典、检索结果、是否需要转人工、最终回复。这里最容易被忽略的是“增量合并”和“整体覆盖”的区别。from typing import TypedDict, Annotated, Optional import operator def merge_dict(left: dict, right: dict) - dict: # 只做一层浅合并不递归 merged dict(left) merged.update(right) return merged class AgentState(TypedDict): # 多轮对话历史按轮次追加 messages: Annotated[list, operator.add] # 当前轮用户的原始输入 user_input: str # 意图分类结果 intent: Optional[str] # 槽位字典例如 {order_id: SO2024001, action: refund} slots: Annotated[dict, merge_dict] # 知识检索命中的文档列表 retrieved_docs: list # 是否转人工 need_human: bool # 给到API层或前端的最终回复 final_response: str逻辑说明messages用Annotated[list, operator.add]告诉LangGraph这个字段是增量追加——每执行一个节点新消息就追加到列表尾部而不是覆盖旧列表。slots用一个自定义的浅合并函数让每个节点只写自己新增或修改的槽位不影响其他节点写入的字段。比如意图分类节点写intent槽位提取节点写slots里的order_id两个节点互不覆盖。如果某个节点想主动清空槽位比如用户重新发起一个新意图那就先清空字典再赋值不要依赖默认覆盖行为。参数说明自定义Reducer必须保持纯函数不能修改入参对象本身否则在LangGraph并发执行分支时会出现状态读写不一致表现就是同一个会话里不同用户看到彼此的槽位。2.3 编译与运行条件边如何决定“追问槽位还是直接检索”State定义好之后把节点函数挂到图上。常见做法是把入口节点设为classify_intent然后根据意图走槽位收集或直接检索。这里有一个编译期容易忽略的点add_conditional_edges的第三个参数是映射表key必须是判断函数可能返回的值否则LangGraph启动时会直接报错。from langgraph.graph import StateGraph, END def classify_intent(state: AgentState) - AgentState: state[intent] call_intent_classifier(state[user_input], state[messages]) return state def slot_extract(state: AgentState) - AgentState: state[slots] extract_slots(state[intent], state[user_input]) return state def need_more_slots(state: AgentState) - bool: required get_required_slots(state[intent]) # 当前已有的槽位不满足必填集合时返回True走追问 return not required.issubset(set(state[slots].keys())) def retrieve(state: AgentState) - AgentState: query f{state[intent]} { .join(state[slots].values())} state[retrieved_docs] search_knowledge_base(query, top_k3) return state def decide_human(state: AgentState) - AgentState: # 检索为空或分数过低时标记转人工 if not state[retrieved_docs]: state[need_human] True return state graph StateGraph(AgentState) graph.add_node(classify, classify_intent) graph.add_node(extract, slot_extract) graph.add_node(retrieve, retrieve) graph.add_node(decide_human, decide_human) graph.set_entry_point(classify) graph.add_edge(classify, extract) graph.add_conditional_edges( extract, need_more_slots, {True: ask_again, False: retrieve} ) graph.add_edge(retrieve, decide_human) graph.add_edge(decide_human, END) app graph.compile()逻辑说明graph.add_conditional_edges是LangGraph里最常用的分支写法第二个参数是判断函数第三个参数是“返回值到节点的映射”。示例里need_more_slots返回True就走ask_again节点生成追问话术这里未列出返回False就直接检索。实际工程里一定要补上ask_again节点否则会报“no edge from extract to ask_again”的引用错误。参数说明StateGraph(AgentState)里的State类必须有完整类型注解LangGraph靠注解推断字段的Reducerapp.invoke(initial_state)可以传一个部分字段的dict作为初始状态缺的字段会按类型注解自动补默认值。3. 意图分类器与LLM-based槽位提取器让机器先听懂再填单3.1 意图分类的两种做法硬规则回退与LLM分类器意图分类器的选型决定了系统的基础准确率。常见做法是“LLM分类 规则回退”双轨主路径让LLM从固定意图清单里选一个并给出置信度当LLM返回unknown、置信度低于某个阈值或用户输入短于3个字时回退到关键词规则。为什么不用微调的分类模型因为客服意图清单通常每两周就会加新意图微调一次的成本和验证周期都太长LLM分类只需要改prompt里的意图清单当天能生效。INTENT_LIST [查物流, 退款, 改地址, 开发票, 投诉, 转人工, 其他] intent_prompt f你是客服系统意图分类器。 只从以下意图中选择一个输出JSON {INTENT_LIST} 用户输入{user_input} 对话历史最近3轮 {history} 输出格式{{intent: ..., confidence: 0.0-1.0, reason: ...}} 逻辑说明把意图清单直接放进prompt配合json输出格式约束大部分情况下能稳定返回。要点在于显式声明“只从清单中选择”否则LLM会自行发明新意图名称比如把“开发票”写成“发票申请”导致下游槽位提取节点拿不到匹配的模板。参数说明confidence字段不是LLM校准过的概率只能当作排序信号不要拿它当严格的业务阈值。如果业务上一个输入可能同时命中“退款”和“改地址”可以让LLM输出top3候选再由规则层按优先级仲裁。3.2 LLM-based槽位提取约束输出的三条经验槽位提取比意图分类更难控制——因为你无法枚举所有槽位值尤其订单号、地址这类自由文本。LLM-based槽位提取器的思路是给LLM一份schema让它从用户本轮输入和上下文里提取字段值。三条经验第一schema要带正则校验。LLM提取订单号经常多提一个空格或漏提后缀在prompt里声明“订单号必须匹配验证表达式”同时代码层再校验一次。第二地址槽位要区分“原地址”和“新地址”。用户说“把收货地址改成北京市朝阳区xx路1号”如果只有address一个槽位旧地址会被覆盖应该定义old_address和new_address两个字段让提取器根据动词“改成”决定写入哪个字段。第三不存在的槽位返回null不要返回空字符串。空字符串会在后续拼接检索query时产生一个空token干扰知识检索的召回结果。slot_schema { order_id: {type: string, pattern: ^SO\\d{8}$}, action: {type: enum, values: [refund, re-address, invoice]}, old_address: {type: string}, new_address: {type: string}, reason: {type: string, optional: True} }逻辑说明把schema序列化进prompt后让LLM严格按这个结构输出JSON。代码层做两件事一是用正则二次校验order_id不匹配就让槽位提取节点返回缺失标记触发追问二是将new_address和old_address按业务动词写入对应字段避免地址被覆盖。参数说明action字段的枚举值要和意图分类器的意图清单保持一一对应否则会出现“意图是退款、action是re-address”这种自相矛盾的状态后续转人工决策和话术生成都会出错。3.3 对话状态跟踪器多轮槽位的更新与覆盖策略对话状态跟踪器在三轮以上对话里最容易出问题。用户说“我要退款订单号是SO20240001”系统问“退款原因是什么”用户答“我不想要了”。此时跟踪器要保留order_id槽位只追加reason而不是把整个槽位字典重置。做到这一点靠的是第2章里的增量合并Reducer以及每个节点只写自己负责的字段。还有一个策略问题用户改口时怎么办比如用户先说“查物流”填了order_id马上说“算了退款吧”。此刻意图已经从“查物流”变成“退款”但order_id还在槽位里可以复用只有退款原因这类退款特有的槽位需要重新追问。常见做法是给每个意图定义必填槽位集合意图变化时保留旧槽位与必填集合的交集清掉与新意图无关的字段而不是全部清空。这样用户不用重新报一次订单号。REQUIRED_SLOTS { 查物流: {order_id}, 退款: {order_id, reason}, 改地址: {order_id, new_address}, } def clean_slots_by_intent(old_slots: dict, new_intent: str) - dict: required REQUIRED_SLOTS.get(new_intent, set()) return {k: v for k, v in old_slots.items() if k in required}逻辑说明意图切到“退款”后取旧槽位与REQUIRED_SLOTS[退款]的交集保留order_id清掉与新意图无关的字段比如查物流时可能留下的“物流公司”槽位。参数说明这个清洗逻辑要放在slot_extract节点开头执行也就是“先按新意图清洗旧槽位再做本轮提取”否则新提取的字段会和旧字段叠加成脏数据。如果用户改口后还想保留旧地址作为备选可以把旧字段挪到optional_slots里不要直接丢弃。4. 知识检索系统与人机转人工决策模型RAG召回与转人工边界4.1 知识库检索的召回链路混合检索比纯向量更靠谱知识检索系统的核心不是“部署一个向量库”而是“召回链路的可靠性”。纯向量检索对语义近似的表述效果好但对“订单号”这类精确字符串反而不如BM25。客服知识库里大量条目是“退款条件”“发票类型”“配送范围”这类半结构化文本混合检索BM25向量用分数融合是更稳的默认方案。Embedding模型的选择也很关键建议先用业务语料跑一遍召回测试再决定用通用Embedding还是领域微调的Embedding。from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain_community.vectorstores import FAISS bm25_retriever BM25Retriever.from_documents(docs, k2) vector_retriever FAISS.from_documents(docs, embedding_model).as_retriever( search_kwargs{k: 2} ) ensemble EnsembleRetriever( retrievers[bm25_retriever, vector_retriever], weights[0.4, 0.6] ) docs ensemble.invoke(query)逻辑说明EnsembleRetriever把两路检索结果做分数归一化和加权融合。权重0.4/0.6的意思是精确匹配和语义匹配各取一部分具体比例根据知识库内容调如果知识库以规则条文为主BM25权重可以提到0.6如果以口语化问答对为主向量权重更高。参数说明k表示每路召回条数每路取2融合后最多4条也会去重。别把k设太大——客服场景precision比recall重要召回5条以上时LLM经常会引用错误段落答非所问。4.2 人机转人工决策模型阈值、兜底与LLM评判人机转人工决策模型是整条链路里看起来“最像规则”的模块但规则不能太简单。最低配是“检索无结果就转人工”这会导致大量误转知识库有答案但召回失败时也转人工坐席压力变大。更稳的决策链分三层第一层检索分数低于阈值直接转第二层候选文档进入独立LLM评判让LLM判断“候选文档是否直接回答了当前问题”第三层生成答案后如果答案包含“抱歉”“无法确认”“请咨询人工”等拒答词触发转人工。def decide_human(state: AgentState) - AgentState: if state[need_human]: return state if not state[retrieved_docs]: state[need_human] True return state if max(doc.score for doc in state[retrieved_docs]) 0.55: state[need_human] True return state judge_prompt ( 候选文档是否直接回答了用户问题只回答yes或no。\n f问题{state[user_input]}\n f文档{state[retrieved_docs]} ) judge_result call_llm(judge_prompt).strip().lower() state[need_human] judge_result.startswith(no) return state逻辑说明judge_prompt是第三道门槛防止“检索到了但内容不相关”的漏网之鱼。这里用一个独立LLM调用而不是让生成答案的LLM顺带判断是为了避免“答不上来但硬编答案”的情况——生成模型倾向于自圆其说独立评判更诚实。参数说明0.55这个阈值不是拍脑袋是根据真实对话样本统计的分数分布定的。每接入一个新知识库建议抽样200条真实QA画出命中与未命中的分数分布再定阈值。低于0.45的样本不管文档是否相关生成质量都会明显下降。4.3 转人工前后的状态同步把对话快照交给坐席转人工不是“返回一句转人工提示”就结束坐席需要看到完整上下文。常见做法是在decide_human节点把当前AgentState序列化成一个JSON快照包含对话历史、槽位、检索来源文档写入工单系统或通过WebSocket推给坐席工作台。这一点做得好不好直接决定坐席愿不愿意用这个系统——很多Agent项目翻车不是模型能力不够而是坐席收到转人工之后还要重新问一遍用户发生了什么。def handoff(state: AgentState) - AgentState: snapshot { dialog_history: state[messages][-10:], slots: state[slots], intent: state[intent], source_docs: [d.page_content[:200] for d in state[retrieved_docs]], need_human_reason: ( no_retrieved_doc if not state[retrieved_docs] else low_score ), } push_to_agent_console(snapshot) state[final_response] 正在为您转接人工坐席请稍候。 return state逻辑说明快照里放对话历史最后10轮、槽位、命中的知识来源坐席打开工作台直接看到上下文。source_docs截断到200字符是为了快照体积可控完整段落可以存到工单系统的附件字段里。参数说明转人工原因要区分no_retrieved_doc、low_score、llm_reject几类后续统计人工介入率时按原因分组才知道该优化检索、改阈值还是改prompt。有一个细节快照里的dialog_history尽量按“用户/助手”成对存储不要只存原始消息列表坐席扫一眼就能看懂上下文。5. 四种必踩的坑与排查路径对话状态错乱、工具循环与API超时5.1 对话状态错乱多轮槽位被覆盖用户改口后无法回退现象用户在第3轮把地址从A改成B系统在后续轮次里还是拿A地址去查货甚至把A当作要退款的订单号去调API。原因槽位字典用了默认覆盖Reducer每个节点返回的都是全量新字典后写入的节点把前面保留的字段冲掉了。这种情况在只跑单轮测试时根本发现不了因为单轮只有一个节点写入槽位一到多轮节点执行顺序稍微一变就出问题。解决把State里的slots字段改成自定义merge函数第2章的merge_dict同时给每个槽位加一个status子字段标成confirmed还是guessing。用户改口后把对应槽位的status重置为guessing后续节点看到guessing就优先追问确认而不是直接拿去做检索。5.2 LangGraph节点无限循环工具调用失败后Agent反复重试现象日志里同一个retrieve节点被调用了几十次API账单暴涨线上偶发“请求超时”告警。原因条件边写成了“检索失败就重新检索”失败原因没变自然无限循环。LLM Agent里还有个隐形推手prompt里写了“如果工具调用失败请重试”LLM会把这个指令当成最高优先级失败一次就重试一次直到撞上LangGraph的recursion_limit。解决给工具节点加显式失败计数超过2次直接走decide_human节点转人工不要让它无限循环。同时在调用端设置recursion_limit并捕获超时异常。app.invoke(initial_state, {recursion_limit: 25})逻辑说明recursion_limit是LangGraph的保护机制超过步数直接抛异常。生产中不要把这个异常直接抛给用户要在API层捕获并返回“请稍后再试”。参数说明25步对一条客服对话足够如果业务需要更多轮工具调用优先考虑把工具拆细而不是一味调高limit。同时检查工具描述里有没有“本工具只负责查询订单状态不负责退款操作”这类边界声明很多循环是因为工具职责太宽LLM拿一个工具当万能工具用。5.3 API超时与限流串行LLM调用拖垮服务层现象并发20路对话时部分请求15秒后才返回前端不断重试坐席端收到重复工单。原因每个对话回合里意图分类、槽位提取、检索、生成用了4次串行LLM调用每次2到3秒链条累计远超前端超时上限。高峰期还会触发LLM厂商的限流返回429后重试又叠加延迟。解决一是把可以并行的调用合并——意图分类和槽位提取可以放在一个prompt里同时做一次LLM调用出两个结果二是给LLM客户端配超时0.8秒和指数退避重试重试次数不超过2次三是API服务层把长耗时操作改异步先用202返回受理ID前端轮询结果。最常见的翻车点是前后端超时配置不一致——前端10秒后端链路15秒必炸。5.4 知识检索召回噪声阈值调低后答非所问现象智能客服一本正经地讲了一条“退款政策”实际知识库里根本没有这条用户投诉后才发现。原因转人工阈值设太低检索分数0.4的文档也进了生成环节LLM以为有依据就开始编造。这是RAG系统的通病检索召回噪声比“没召回”更可怕因为没有报错只有一本正经的胡说。解决在prompt里明确“没有检索到相关依据时直接说无法回答并转人工”把检索阈值从0.4抬到0.55并对0.45到0.55的低分区间启动LLM评判环节。另一个排查方向是文档切分粒度——把长文档切成256到512字符的chunk重叠32字符避免把“不适用场景”和“适用场景”切进同一个chunk。切分粒度不对时调阈值治标不治本。6. API服务层部署与验证从单轮接口到压测调优6.1 用FastAPI包装LangGraphAPI服务层用FastAPI封装对外暴露/chat接口。/chat接收用户消息和会话ID内部从Redis取出历史状态invoke LangGraph把新状态写回Redis后返回回复。这里的核心是状态持久化否则多轮对话只能靠前端传历史一旦刷新页面就断。from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): session_id: str message: str app.post(/chat) def chat(req: ChatRequest): state load_state_from_redis(req.session_id) state[user_input] req.message try: result app.invoke(state, {recursion_limit: 25}) except Exception: raise HTTPException(status_code504, detailagent execution timeout) save_state_to_redis(req.session_id, result) return {reply: result[final_response]}逻辑说明load_state_to_redis取出上次状态save_state_to_redis写回让多轮对话不依赖客户端传历史。异常捕获后返回504前端可以提示稍后再试。参数说明Redis里每条状态的TTL设24小时定期清理不活跃会话会话ID用UUID由前端生成并透传。如果并发量再大建议把Redis替换成带持久化的状态服务避免重启丢状态。6.2 调优参数与验证指标上线前跑一组控制变量的调参实验观察两个核心指标人工介入率和槽位命中率。人工介入率太高说明转人工决策过敏感槽位命中率太低说明槽位提取prompt或schema设计有问题。参数建议初始值调优方向temperature0.1客服场景要确定性超过0.3容易偏离槽位格式top_p0.9与temperature配合不要同时调到最高最大输出token256客服回复短太长会拖慢响应检索top_k2-3加大后召回噪声明显上升转人工阈值0.55按知识库样本分数分布调整6.3 验证方法与回归用例三类验证要做扎实一是对话回放把真实历史会话按时间顺序灌进系统对比新系统回复与人工坐席回复的语义相似度二是槽位命中率构造200条带标注的测试集跑一遍槽位提取计算精确率和召回率三是转人工原因分组统计把每条转人工对话打上原因标签每周按原因分组复盘——指向知识库缺口的补文档指向prompt问题的改prompt指向阈值过低的就调阈值。这个习惯帮我救回过两个差点被砍掉的项目一次是知识库文档切分粒度不对导致大量转人工一次是LLM评判环节缺失导致编造答案。每次排查都从“转人工原因标签”入手比盲调参数快得多。希望帮到你。本文还有配套的精品资源点击获取