ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

基于LangGraph的RAG智能客服系统:从链式调用到状态编排的实践复盘

基于LangGraph的RAG智能客服系统:从链式调用到状态编排的实践复盘 简介大模型应用正从简单的问答走向复杂的业务场景。RAG检索增强生成通过外部知识库提升回答准确性而LangGraph提供的状态图编排模型让流程不再是一条固定的链而是一张可暂停、可回溯、可路由的图。这种理念在智能客服系统中尤为重要多轮对话需要维护会话状态意图识别、知识检索、相关性校验、转人工等环节需要灵活流转。在实际工程中基于LangGraph的节点与条件边设计可以将混合检索、查询改写、人工介入等机制有机整合实现可观测、可干预的客服工作流。本文完整复盘了一个基于LangGraph构建RAG智能客服系统的架构设计、状态管理、检索调优与上线踩坑经验并展望了Agentic RAG等扩展方向。 看到基于LangGraph构建的RAG智能客服系统这个项目名我的第一反应是这不就是把LangChain的RAG示例换个框架重写一遍吗直到我自己把一套线上客服系统的RAG流程从LangChain链式调用迁到LangGraph才意识到这两者的区别不是API形态不同而是整个系统的思考方式变了——从一条流水线变成了一张可编排、可中断、可回溯的状态图。这篇文章完整复盘这个项目为什么放弃LangChain、状态图和节点怎么设计、检索增强怎么嵌进工作流、上线后踩了哪些坑以及后续可以往哪些方向扩展。如果你是第一次接触LangGraph和RAG这篇文章同样适合你。我会把基础概念融入实战讲解不堆术语但该拆的细节一个不落。整套系统我用Python和LangGraph构建向量库用的是开源方案重排模型用的也是社区常见模型所以只要你有一台能跑大模型的机器或者接的API这套方案基本可以原样复现。1. 为什么我放弃了LangChain链式调用改用LangGraph编排RAG客服流程1.1 先搞清楚LangChain和LangGraph的本质区别LangChain的Chain本质上是一个顺序容器你在里面定义好步骤数据就沿着预设路径一路流下去。流程固定的时候比如检索 - 拼接上下文 - 生成答案 - 输出Chain用起来非常舒服代码短、可读性高。但RAG智能客服不是这样的场景。客服对话天然存在大量分支用户可能问产品功能也可能问售后政策还可能只是在吐槽甚至发泄情绪有的问题知识库能回答有的问题检索不到需要转人工用户在追问时模型要记住上文但上下文太长又得做裁剪或摘要。这些诉求放在Chain里你会被迫写一堆条件判断、循环控制、状态传递的胶水代码。LangChain虽然有RouterChain这类工具但本质还是预先定义好路由规则它没有真正意义上的运行时状态管理也没有让流程在执行过程中停下来等人或者调整路径的能力。我之前一版系统就是用LangChain写的到了后期维护成本显著上升原因很简单任何一个分支调整都可能牵动整条链的代码结构改一处冒三处。1.2 LangGraph把流程变成了状态机LangGraph的核心抽象是StateGraph。你定义一个全局共享的状态对象State然后像搭积木一样添加节点Node再通过边Edge定义节点之间的跳转关系。关键区别在于每个节点都能读取和修改全局状态边的定义可以依赖状态字段做条件判断而且框架内置了节点顺序执行、并行执行、循环、中断和恢复的能力。拿客服系统里判断是否需要转人工这个场景举例。在LangChain里我通常的做法是生成答案后再跑一个分类器判断答案是否可靠不可靠就返回特殊标记由外层代码决定是否走另一个流程。这个逻辑能跑但转人工流程与主流程是割裂的状态也是分散的。在LangGraph里这个判断直接在条件边中完成生成节点输出的置信度低就走人工节点分支置信度高就直接返回给用户。整条路径在一张图上可见、可追踪、可测试。1.3 不是所有RAG项目都该上LangGraph这里我得泼一盆冷水如果你的场景就是一个固定知识库的问答不需要多轮上下文不需要转人工不需要兜底逻辑那用LangChain的RetrievalQA或者直接调API反而更简单。LangGraph的引入会带来额外概念负担和调试成本杀鸡用牛刀。那什么情况下值得上我的判断标准很明确一是多轮对话需要维护会话状态和记忆二是流程存在复杂条件分支、循环或人工介入三是需要精细控制每个节点的输入输出比如把检索结果暴露给下游审核四是希望系统运行过程可观测、可回放。智能客服系统几乎全占所以这个项目最终选择了LangGraph。2. RAG客服系统的整体架构与节点设计2.1 架构目标与模块划分先把这个系统的架构目标列明白输入是用户的一个问题输出是一个带有知识依据的回答或者是一个转接人工的提示。系统内部需要完成六件事意图识别、知识检索、相关性校验、答案生成、引用标注、风险兜底。我最终把架构划分为四个模块。入口层负责多轮对话管理把用户当前问题和历史上下文组装成一条带状态的会话意图识别模块负责判断用户诉求是询问、是投诉、还是闲聊RAG核心模块负责从知识库检索和重排生成与校验模块负责组织答案、标注引用来源并对答案是否脱离给定上下文做一次检查。这四个模块在LangGraph里分别对应一组节点和条件边。2.2 节点职责与条件边的设计整个状态图由六个核心节点组成每个节点职责单一不掺杂物classify_intent判断用户意图输出intent字段比如product_query、after_sales、small_talk、escalate_hint。这个节点的输出直接决定后续路由方向。retrieve_docs根据用户查询以及意图标识从向量库检索Top-K文档片段同时跑一遍BM25关键词检索做混合召回。check_relevance对检索结果做相关性打分判定召回片段是否真正能支撑回答。如果分数过低会设置need_fallbackTrue触发兜底逻辑。generate_answer将检索到的文档片段和对话历史拼接成上下文交给大模型生成回答同时要求模型在回答中标注引用来源的文档ID。escalate_to_human转人工节点记录当前会话关键信息触发人工客服接入。fallback当知识库召回不足时返回澄清话术或者引导用户换个说法。条件边主要在两处。一处是classify_intent之后根据intent决定走RAG链路还是走闲聊/转人工分支另一处是check_relevance之后根据相关性分数决定继续生成、重新检索允许一次循环重试还是转人工。这里我要重点说一句不要让检索失败直接落回抱歉我不能回答而是优先给一次重检索机会。用户问题的表述方式经常导致第一次检索结果不理想换个检索策略往往就能解决。2.3 状态对象的数据流设计LangGraph的灵魂在于State。我定义的ChatState大概长这样from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages] query: str intent: str retrieved_docs: list relevance_scores: dict answer: str citations: list need_fallback: bool need_human: bool字段含义不复杂messages是对话历史query是当前问题intent是意图识别结果retrieved_docs是召回片段relevance_scores是相关性打分answer和citations是最终回答和引用来源最后两个布尔值控制流程走向。有一个细节值得特别注意messages字段使用了add_messages这个归约器reducer它保证新消息会被追加而不是覆盖。这是LangGraph处理多轮会话的基础少了它第二轮对话就会把第一轮的消息覆盖掉。设计State的时候最忌讳做成大杂烩有什么字段都往里塞。状态对象会被每个节点读写字段越多越难维护不同节点的读写冲突也容易引入隐蔽bug。我的建议是状态里只放需要跨节点传递的数据节点内部的临时变量不需要进State局部处理完即可。3. 状态管理与工作流编排的核心实现3.1 图结构的搭建与流程路由实现LangGraph把一个复杂的业务逻辑变成了一张图。下面是核心代码骨架也是我项目里最基础的部分from langgraph.graph import StateGraph, START, END builder StateGraph(ChatState) builder.add_node(classify_intent, classify_intent) builder.add_node(retrieve_docs, retrieve_docs) builder.add_node(check_relevance, check_relevance) builder.add_node(generate_answer, generate_answer) builder.add_node(escalate_to_human, escalate_to_human) builder.add_node(fallback, fallback) builder.add_edge(START, classify_intent) builder.add_conditional_edges( classify_intent, route_by_intent, { product_query: retrieve_docs, after_sales: retrieve_docs, small_talk: fallback, escalate_hint: escalate_to_human, }, ) builder.add_edge(retrieve_docs, check_relevance) builder.add_conditional_edges( check_relevance, route_by_relevance, { generate: generate_answer, retry: retrieve_docs, escalate: escalate_to_human, }, ) builder.add_edge(generate_answer, END) builder.add_edge(escalate_to_human, END) builder.add_edge(fallback, END) graph builder.compile()这段代码把一个多分支的客服工作流压缩成了几行配置读起来非常清楚。route_by_intent和route_by_relevance是普通的判断函数输入是State输出是路由目标字符串。比如def route_by_relevance(state: ChatState) - str: if state[need_human]: return escalate if state[need_fallback]: return retry return generate你可能会问这些路由函数看起来就是普通Python函数和LangChain的RouterChain有什么本质区别区别在于LangGraph的路由发生在图执行过程中它的目标节点和当前状态是动态耦合的而且你可以随时在任意位置插入新的条件边。更重要的是LangGraph自带状态回放能力这意味着任何一次路由决策都可以被回溯和分析。3.2 条件路由的阈值设计与重试机制不要小看那个retry分支它让我在线上把转人工率降低了接近三成。设计思路是相关性校验不通过时说明第一次检索可能没有命中用户真实意图但知识库里很可能存在答案。于是我在retry分支里做了两件事一是用改写后的用户问题重新检索这就要求系统在第一次检索引擎返回低分后再让模型基于对话历史重写查询词二是把检索的Top-K值调大一倍让召回范围更宽。这里必须引入重试次数的控制否则一个低质量查询会反复触发重检索拖慢响应时间。我用了一个简单的计数器字段retry_count在条件路由函数里判断如果retry_count 1就直接返回escalate不再继续循环。处理逻辑如下def route_by_relevance(state: ChatState) - str: if state[retry_count] 1: return escalate if state[need_human]: return escalate if state[need_fallback]: state[retry_count] 1 return retry return generate提示重试不是越多越好。每次都重试会让用户等待成倍增长客服场景下响应时间超过8秒用户体验会断崖式下降。我实测下来一次重试是性价比最高的方案超过一次收益就很有限了。3.3 多轮对话中的短期记忆与长期记忆处理客服系统绕不开多轮对话而多轮对话最大的痛点就是上下文怎么带。我采用了分层策略短期记忆放在messages字段里每次请求带最近5轮对话作为上下文长期记忆用户身份、常用地址、历史工单编号等放到外部存储里在会话开始时注入到State中。LangGraph官方文档里有完整的记忆持久化机制可以通过checkpointer把状态保存下来实现会话恢复。我在项目里用了SQLite作为checkpointer的存储后端这样即使服务重启用户还能接着上一轮往下聊。这一点对客服系统非常关键用户可能隔一段时间再回来我们需要让他感觉客服记得我们之前聊过什么。3.4 人工介入机制Human-in-the-Loop这是LangGraph最吸引我的能力之一。智能客服不能完全脱离人工尤其是遇到投诉或者高危场景时。传统的做法是在代码里写一个回调函数通知人工系统。LangGraph则原生支持interrupt机制在节点执行过程中暂停图等待外部输入再恢复执行。我在escalate_to_human节点里就是这么用的当检测到需要转人工时节点会暂停执行把会话摘要发送到人工客服工作台。人工客服处理后把解决方案注入回状态图会从暂停处继续执行并把结果反馈给用户。整个过程都在同一个状态图里不会出现人工处理完但系统不知道结果的尴尬。4. RAG检索增强的实战调优从切块到引用溯源4.1 切块策略怎么选固定窗口、递归切块、还是语义切块RAG的效果一半取决于检索一半取决于知识库预处理。切块策略是老话题但不同方式的适应场景差别很大我用一张表总结下切块方式优点缺点适用场景固定窗口切块实现简单可控性强容易切断语义单元章节结构不明显的长文本递归字符切块按段落边界切分语义完整度较高需要根据文档类型调分隔符大多数通用场景性价比最高语义切块按embedding相似度聚类切分语义最完整计算量大需要额外维护簇结构文档结构复杂、质量要求高的场景我在这个客服项目里用的是递归字符切块separators按层级设置从段落换行符、句号、逗号一路往下拆。同时把标题和章节结构作为元数据一并写入切块。为什么必须保留元数据因为客服回答需要这个是商品详情页的说明还是这个是售后政策的第十条这样的来源信息没有元数据你后面做引用溯源就会很痛苦。4.2 混合检索向量召回 关键词召回很多项目把向量检索当作RAG的唯一检索方式这是一个常见误区。向量检索擅长语义相似但面对专业名词、商品型号这种精确匹配场景经常翻车。比如用户问iPhone 15 Pro Max保修政策如果知识库里写的是iPhone 15 Pro Max提供一年有限保修向量检索大概率能命中但如果用户把型号简写成了iPhone 15 PM向量匹配可能就不如关键词检索精确。我的方案是混合检索向量检索负责召回语义相近的Top-20BM25负责召回关键词精确匹配的Top-20两者合并去重后送入重排序模型精排最终取Top-5作为上下文材料。重排序用一个轻量级cross-encoder模型性能开销在接受范围内。这个流程在LangGraph里就是两个并行节点或者一个节点里并行跑两个检索器在状态里各自产出候选文档最后在check_relevance节点合并处理。4.3 引用溯源与生成结果的可靠性校验智能客服系统最怕两件事一是模型编造知识库里没有的内容二是答非所问。针对前者我给生成环节加了强制约束提示词里明确要求模型只能基于给定的检索片段回答答案必须标注对应片段ID针对后者我在生成完成之后增加了一道校验节点用大模型把回答和检索片段做一次矛盾检测如果发现回答存在明显超出上下文的内容就标记为低可信度走重试或转人工分支。这道校验不能消除所有幻觉但能把高风险回答拦截下来。实测中加入校验节点后人工质检发现的无依据回答数量下降了约40%。代价是增加了额外的大模型调用延迟所以我在校验节点里做了取舍只有当max_relevance_score低于某个阈值时才做完整校验高置信度结果直接放行避免每次回答都多走一遍模型调用。4.4 查询改写一次提升命中率的低成本手段上面提到重试时会改写查询词这里展开讲一下。客服场景里用户提问往往口语化严重比如我手机最近好卡和知识库里的设备运行缓慢表达差异很大。单纯做向量检索这两句话的语义距离可能不够近。我在retrieve_docs节点前加了一个查询改写步骤把对话历史、当前问题、历史检索结果交给模型让它产出若干个更贴合知识库风格的查询词。这个步骤只在首次检索失败时触发不会拖慢正常链路的响应速度。实测下来查询改写让整体命中率提升了约15%到20%成本就是一次额外的小模型调用。5. 上线后的踩坑记录与优化方向5.1 会话状态丢失事故与checkpointer的使用上线初期遇到过一个很隐蔽的问题用户在客服对话中聊了五六个回合之后突然问刚才你说的是什么意思系统完全答不上来。排查后发现我把页面端传过来的历史消息当作唯一记忆源但页面端为了控制请求体大小只传最近两轮消息。这个问题在联调时没暴露因为测试环境消息量小等线上用户真的聊多了才暴露出来。修复方案是把会话状态的重任交给LangGraph的checkpointer而不是依赖前端传参。每次请求进来根据会话ID从checkpointer恢复完整的messages状态页面端只需要传一个session_id。这个改动同时解决了多轮对话上下文失真和消息丢失两个问题。这也是我前面反复强调状态管理是LangGraph核心的原因它不只是一个技术选型而是整个系统的记忆中枢。5.2 召回质量差先看切块再看Embedding模型上线后运营反馈系统经常答非所问。我第一反应是Embedding模型不够强准备换个更大参数的模型但后来通过对比测试发现问题出在切块太碎原本一个完整的售后政策条款被切成好几个小块检索时只召回了其中一段模型拿到残缺的上下文自然答不准。重新调整切块策略加大窗口、保留段落边界之后准确率明显回升。所以如果你也遇到召回质量差不要急着换模型先按这个顺序排查查看实际召回的内容确认是否语义完整检查用户问题和知识库表达方式的差异必要时做查询改写对比不同切块参数下的检索命中率最后再评估是否需要换Embedding模型或者升级到重排模型。5.3 LangGraph Studio与可视化调试调试LangGraph项目时我强烈建议用LangGraph Studio或者网页版的可视化调试工具。它在图结构可视化、节点状态查看、异常路由回放方面比任何打日志的方式都高效。你最直观的感受是整个客服工作流像画了一个流程图点击每个节点能看到当时的State快照、检索结果、路由决策甚至还能在图上直接修改某个节点的输出测试后续流程走什么分支。这个能力在排查为什么这个会话转人工了为什么这一步走了retry分支这类问题时效率极高。我第一次用Studio排查问题时发现有个会话在check_relevance和retrieve_docs之间循环了三次才被拦住原因就是retry_count字段在状态更新时被节点覆盖了而不是累加。这种bug靠日志排查会花很长时间但在Studio里看状态流转路径和字段变化一眼就明白问题出在哪。5.4 后续可扩展的方向这套框架跑通后你会发现图结构带来的扩展性远超预期。我目前还在做两个方向的延伸。一是把用户的长期语义记忆接进状态图让系统在生成回答时能主动带出这位用户上次问过什么。比如用户上次问过怎么退货这次问运费谁承担系统如果能记得上一次的上下文回答会更精准。LangGraph的持久化机制天然支持这个扩展我只需要在恢复状态时加一步长期记忆的注入。二是尝试让检索环节自动根据用户意图动态切换数据源也就是agentic RAG——用一个大模型Agent来规划该查哪个知识库、按什么顺序查、每次查什么。这是LangGraph本来就擅长的领域因为Agent的本质也是一张带循环和工具调用的图。比如用户问我的订单为什么还没发货Agent可以先查订单系统再查物流接口最后把结果汇总成回答。这个方向比固定流程的RAG复杂但LangGraph的图结构让我不担心代码失控——每个工具调用就是一个节点每次决策就是一条条件边复杂度在可控范围内。最后再分享一点个人体会。构建RAG智能客服系统最忌讳的就是一上来就写代码调Prompt。先把状态图想清楚——有哪些节点、有哪些分支、谁触发谁、失败走哪条路——后面所有的实现都是水到渠成的事。LangGraph给我的最大价值不是省了多少代码量而是让我在复杂对话流程里始终能看见整张图而不是迷失在if-else的森林里。如果你也在做一个交互逻辑复杂的RAG系统我建议你先把State设计出来再画一遍节点和边的关系图最后再动手写代码。这个习惯让我少走了很多弯路希望对你也有用。本文还有配套的精品资源点击获取
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进