ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

如何用LangGraph构建带反思循环的Agentic RAG

如何用LangGraph构建带反思循环的Agentic RAG 1. 项目概述给 RAG 装上“会反思”的脑子先说我为什么做这个项目。在公司内部知识库问答系统跑了小半年后我发现传统 RAG 的命门根本不在“检索不到”而在“检索到一堆毫不相关的内容却半点不自知”。用户问一句“报销流程里电子发票的金额限制是多少”向量检索很可能捞回一堆“发票粘贴要求”“冲账时限”然后 LLM 硬着头皮把不相干的材料组织成一篇看似流畅的答复。这种体验用一句话总结就是错得很有礼貌。所以我用 LangGraph 重做了一套 Agentic RAG。核心差异在于把“查询 → 检索 → 生成”这条线性流水线改成了一条带反馈回路的图。系统拿到用户问题后先做查询改写再去多路检索然后用一个独立的评分节点判断每个文档片段与问题的相关性。如果整体相关性不达标就生成一条反思反馈驱动下一轮“改写查询 → 重新检索”直到达到阈值或触发兜底。整个过程像人一样查资料、判断资料值不值得看、发现跑偏了就换个思路再查而不是一次性梭哈。这套方案解决的具体问题是传统 RAG 的检索噪声、多义词和口语化查询导致的召回失败、以及“生成即终局”导致的不可纠正。它适合三类人来参考一是被 RAG 幻觉和误召回折磨的开发者二是想入门 LangGraph 状态编排的人三是需要把知识库问答落到生产环境、但又担心效果不稳定的小团队。下文我会把状态设计、节点实现、反思循环、生产化部署和踩坑记录全部展开直接提供可复现代码和参数选择理由。1.1 传统 RAG 的三个死穴先用一个生活化类比。传统 RAG 就像一个只会用百度查一次关键词的实习生你问“母公司对子公司的担保额度审批需要哪些材料”他搜索“担保额度审批材料”贴出前三页链接就完事。而 Agentic RAG 是那个会先拆解问题、搜完之后自己判断“这篇压根没说母公司子公司关系”、然后换个问法再查的老手。这里的差距不是模型能力而是流程结构。第一个死穴是查询与文档之间的语义鸿沟。企业内部知识库里用户口语化描述和文档正式术语往往对不上。比如员工问“公司笔记本坏了找谁”而知识库里写的是“IT资产报修流程”。如果不做查询改写向量召回的结果基本靠缘分。第二个死穴是检索后缺少质量闸门。传统 RAG 把 top-k 结果直接塞给 LLM哪怕五个片段里只有两段相关模型也会被迫组织答案导致伪相关内容和幻觉混在一起输出。很多团队把召回率指标调得再高实际生成质量依然不行问题就出在这个环节没有校验。第三个死穴是一次检索定生死没有试错机制。现实中的知识库经常有术语缺失、文档切分碎片化、查询意图模糊等状况单凭一次检索很难稳定命中。传统 RAG 没有回路设计检索失败了就是把错误答案输出给用户而在 Agent 化之后系统至少有机会“知道自己不知道”。1.2 Agentic RAG 与传统 RAG 的本质区别用一个表格对比最直观设计维度传统 RAGAgentic RAG流程结构线性query → retrieve → generate图状带条件分支与循环回路查询处理直接用用户原始问题检索先改写、拆解、意图识别检索策略单一向量检索top-k 直接透传多路召回 动态调整检索参数质量校验无检索结果直接进生成独立评分节点过滤低相关片段容错机制无重试和反思路径反思反馈驱动重新改写与再检索可控性LLM 一次性输出不可干预每步状态可观测、可中断、可回退我在实际项目中感受最深的一点是Agentic RAG 的“Agent”不是指系统更聪明而是指系统拥有了决策能力。它决定了下一步调用哪个工具、是否重试、是否终止。LangGraph 之所以成为我最终选择是因为它在 Python 里用“图”这种数据结构天然表达决策流节点的返回值就是图的状态更新条件边就是决策本身比手写一堆 if/else 加 while 循环维护起来清晰太多。1.3 这个项目适合谁如果你只是想在本地跑一个简单的文档问答那不用上 Agentic RAGChroma 加 LlamaIndex 十分钟就能搞定。但如果你遇到过下面几种情况就说明该上这套方案了用户问法千奇百怪导致频繁检索失败知识库内容垂直术语和口语差异巨大领导要求在问答中给出可追溯的引用来源或者你已经在用 LangGraph但想知道“反思”这类循环结构怎么真正落地。2. 整体架构与核心设计拆解2.1 LangGraph 为什么适合编排 AgentLangGraph 的定位是“给语言模型应用加上可编排的有向图运行时”。它的核心抽象只有三个节点Node、边Edge和状态State。每次调用节点就是一个函数函数接收当前状态、返回状态的部分更新边决定节点执行后的下一步走向可以是无条件边也可以是条件边。这套抽象和传统 Agent 框架最大的不同是它不把控制权完全交给 LLM而是由开发者显式定义图的拓扑结构。我选择它的理由主要有四个。第一循环天然支持而循环是“反思”的前提没有循环就只能做一次前向传播。第二状态是显式的 TypedDict执行到哪一步、检索到哪些文档、评分是多少都能完整观察对排查问题非常友好。第三官方提供持久化Checkpoint机制可以把执行状态存到数据库断点续跑、人工介入、审计回放都方便。第四它只是编排层检索、Embedding、评分用的都是普通 Python 函数不会绑架你现有的工具链。这里有个容易踩的坑LangGraph 跟 LangChain 是不同层的东西。LangGraph 负责“流程怎么走”LangChain 负责“LLM 和工具的封装”两者可以配合也可以只用 LangGraph 加原生 OpenAI SDK。我项目里就是用 LangGraph 做编排检索器是自己封装的多路召回LLM 调用直接用 LangChain 的 ChatOpenAI图完全不受框架限制。2.2 状态的定义与流转把“流水线”变成“回路”把 RAG 从流水线改成回路最关键的是状态字段的设计。我在项目中定义了下面这些字段每一个都有存在理由from typing import TypedDict, List, Annotated from langgraph.graph import StateGraph, END class RagState(TypedDict): question: str # 用户原始问题永不覆盖 rewritten_question: str # 当前轮次的改写后查询 documents: List[dict] # 当前轮次检索到的文档片段 graded_docs: List[dict] # 评分后的文档过滤掉低相关片段 reflection_feedback: str # 反思反馈驱动下一轮改写 attempt: int # 当前是第几轮检索 max_attempts: int # 最大尝试轮次 generation: str # 最终生成答案 used_documents: List[dict] # 最终实际引用的文档用于溯源设计这些状态有两个原则值得留意。第一原始问题与派生问题分离。改写后的查询会变化但原始问题必须保留否则几轮反思之后你可能都搞不清系统在回答什么。第二把“本轮检索结果”和“最终采纳结果”分开。反思循环中每轮都会产生文档但只有最后一轮通过评分的文档才进入生成这样既保留了过程信息又不会让历史噪声污染最终答案。关于状态更新LangGraph 默认是“覆盖式”也就是节点返回什么键就覆盖什么键。如果需要累加比如把每轮检索到的文档都追加到一个大列表里做审计就要用 Annotated 加上 reducer像documents: Annotated[List[dict], operator.add]。我建议不要滥用累加因为状态太大既浪费序列化开销又会让 Prompt 上下文膨胀除非有审计需求否则每轮覆盖即可。2.3 从单路检索改到多路召回的设计这个项目的编排设计与检索方案是同步确定的多路召回 交叉评分。所谓多路召回就是对同一个查询同时用多种检索策略取回候选文档再做去重和融合。我的项目里用了四路向量检索通过 Embedding 语义相似度取回 top-8关键词检索ES 或 BM25 精确匹配取回 top-4标题检索先用 LLM 抽取查询中的关键实体匹配文档标题取回 top-2知识图谱兜底如果知识库有实体关系用实体链接取回相关实体附带的文档。为什么要多路因为不同查询类型适合不同检索方式。用户问“2024年第三季度营收是多少”时BM25 对数字和季度的匹配比向量可靠得多用户问“怎么申请了报销一直没到账”这种口语则只能靠向量语义。多路召回之后每路的候选集合并去重再交给评分节点统一判断大幅降低单路检索漏检的概率。这里想提醒一个细节多路召回会显著增加检索引擎的消耗如果知识库只有几千个文档完全没有必要上多路。我的经验是文档量低于 5 万片且查询类型单一单路向量加 BM25 双路融合就足够。真正需要上多路的是垂直领域、错误答案代价高、用户表述跨度大的场景。3. 核心流程实现改写、检索、评分、反思与生成3.1 查询改写节点先把“人话”翻译成“文档话”查询改写是整个反思回路的第一环也是最容易被低估的一环。传统做法是把用户原始 query 直接塞给向量检索但企业内部查询往往伴随习惯性简称、口语表达、指代歧义直接检索的召回率很低。我的做法是让 LLM 做一次“翻译”把用户问题转写成一个更贴近知识库表述的检索式。def rewrite_node(state: RagState) - dict: prompt f你是一名信息检索专家。请把用户的问题改写成更适合知识库检索的查询。 要求 1. 保留原始意图补充同义术语和上下文 2. 如果是复合性问题拆成最多两个子查询用换行分隔 3. 不要编造事实只做语言层面的改写。 用户问题{state[question]} 请直接输出改写后的查询 resp llm.invoke(prompt) new_query resp.content.strip() return {rewritten_question: new_query, attempt: state[attempt] 1}这里有几个实操心得。第一改写不是每次都该发生如果第一轮已经检索到了高相关文档再改写反而可能引入噪声所以需要配合后面的评分节点决定是否需要进入下一轮改写。第二拆分子查询时要用换行分隔然后分别检索、合并结果不要试图用一条查询覆盖所有子问题否则任何一路都匹配不好。第三把 LLM 温度调到 0检索式改写是确定性任务不是创意任务。3.2 多路检索节点把候选集拉回来检索节点负责执行真正的召回。我给每个子查询做多路检索然后把结果合并def retrieve_node(state: RagState) - dict: query state[rewritten_question] if state[rewritten_question] else state[question] sub_queries [q.strip() for q in query.split(\n) if q.strip()] docs [] seen_ids set() for q in sub_queries: for d in vector_store.similarity_search(q, k8): if d.id not in seen_ids: docs.append({id: d.id, text: d.text, score: d.score}) seen_ids.add(d.id) for d in bm25_search(q, top_k4): if d.id not in seen_ids: docs.append({id: d.id, text: d.text, score: d.score}) seen_ids.add(d.id) for d in title_search(q, top_k2): if d.id not in seen_ids: docs.append({id: d.id, text: d.text, score: d.score}) seen_ids.add(d.id) if len(docs) 3: for d in vector_store.similarity_search(query, k20, score_threshold0.7): if d.id not in seen_ids: docs.append({id: d.id, text: d.text, score: d.score}) seen_ids.add(d.id) return {documents: docs}这个节点里最容易被忽视的是“候选集太少的兜底策略”。我曾经遇到过一个查询在所有路里都只召回一两篇文档评分节点再怎么反思也很难翻盘后来加了分数阈值放宽的兜底逻辑把向量检索的相似度阈值从 0.85 降到 0.7召回率显著提升。阈值怎么定我建议抽样一批真实查询画出相似度分布再选一个能覆盖 80% 正常查询的阈值不要拍脑袋。3.3 相关性评分节点给检索结果当“守门员”评分节点是整个“会反思”机制的心脏。它的工作很简单对送入节点的每篇文档判断它是否与当前用户问题相关。判断方式不是算向量相似度而是让 LLM 做二分类判断并给出理由。这是反思反馈的第一手信息来源。def grade_node(state: RagState) - dict: question state[question] # 用原始问题不要用改写后的 graded [] for doc in state[documents][:12]: prompt f你是一名检索质量评审员。以下是用户问题和一段候选文档片段。 用户问题{question} 候选文档片段{doc[text][:800]} 请判断该片段是否包含任何回答该问题所需的信息。 只输出 JSON{{relevant: yes 或 no, reason: 一句话理由}} resp llm.invoke(prompt) data parse_json(resp.content) if data.get(relevant) yes: graded.append({**doc, reason: data.get(reason, )}) relevant_ratio len(graded) / max(len(state[documents]), 1) feedback if relevant_ratio 0.5 and state[documents]: prompt2 f给定用户问题和当前检索结果分析为什么检索效果不佳并给出具体的查询改写建议。 用户问题{question} 检索到的文档片段{[ .join(d[text][:200]) for d in state[documents][:5]]} 请输出一段不超过100字的改进建议 feedback llm.invoke(prompt2).content.strip() return {graded_docs: graded, reflection_feedback: feedback}这里有个关键细节评分用的问题是原始用户问题而不是改写后的查询。我之前犯过错误用改写后的 query 去评分结果把“改写后的语义偏差”也当成评分噪声导致反思反馈失真。原因很简单改写的目标是提升召回但评分的标准永远是“这段文档是否在回答用户真正想问的问题”必须回到原点和用户对齐。另一个问题是成本。如果每轮检索 50 个文档让 LLM 逐一评分费用会非常高。我的做法是两级筛选先用 embedding 余弦相似度做粗筛只保留相似度前 15 的候选再对这 15 个做 LLM 精评。粗筛会漏掉一部分语义差异大的相关文档但那些文档本来在向量检索阶段就大概率通不过总体性价比非常划算。3.4 反思与重检索循环让系统“知道自己跑偏了”反思反馈的作用是让系统在检索失败时不要盲目重试而是有方向地调整。我设计的逻辑是这样的评分节点发现相关文档占比低于 50% 时生成一段不超过 100 字的改进建议并作为reflection_feedback写入状态然后条件路由就把执行权交给下一轮查询改写改写节点把第一轮失败的原因合并进去生成新的检索式。def route_after_grade(state: RagState) - str: if len(state[graded_docs]) 3: return generate if state[attempt] state[max_attempts]: return fallback return rewrite路由逻辑要注意一个平衡评分通过的文档数量阈值设太低比如只有 1 篇生成质量波动很大设太高比如 5 篇大量查询会反复循环延迟和成本都不可控。我最终用的是“至少 3 篇通过 或 循环次数上限 3 次”并且为路由加了判断如果第一轮只通过了 1 篇但这篇文档本身就完整覆盖了用户问题的核心实体那么直接生成是更优选择。为了处理这种情况我在评分节点里额外判断了“通过文档是否覆盖问题核心实体”如果覆盖则提前走向生成。反思的质量直接影响下一轮检索的效果。我提供一个改进反思内容的技巧给反思 LLM 提供的不只是检索失败的文档列表还要提供“用户真实意图”分析否则反思容易变成瞎猜测。我的反思 Prompt 完整版包含三部分原始问题、用户可能的意图、失败文档列表并要求输出“下一轮查询中应该补充什么术语、排除什么噪声”。3.5 最终答案生成带着引用与溯源生成通过评分后的文档进入生成节点。这里有两个要点一是让 LLM 知道哪些文档是可信的、哪些是要忽略的二是让生成结果带上引用来源。我把文档材料按“检索系统提供的参考材料”标识传进 Prompt同时要求生成时在每个事实后标注来源 ID最后从生成文本中抽取出引用文档 ID 列表用于前端展示。def generate_node(state: RagState) - dict: context \n\n.join( f[文档{d[id]}]\n{d[text]} for d in state[graded_docs] ) prompt f请基于以下参考材料回答用户问题。如果材料不足以回答问题请明确说知识库中没有找到相关信息。 用户问题{state[question]} 参考材料 {context} 要求 1. 回答要准确、简洁优先使用材料中的原话 2. 在每个关键事实后标注来源ID 3. 不要编造参考材料之外的信息。 resp llm.invoke(prompt) return {generation: resp.content}生成节点的 Prompt 设计里我特别加了“材料不足以回答时必须明说”的指令。这是因为反思循环结束后仍可能有部分查询最终没有通过评分如果模型硬答就会产生幻觉。这个指令本质上是给系统留了一条“诚实退出”的路径。实际效果非常明显上线后用户对“抱歉知识库暂无该信息”的接受度远高于“一本正经地胡说八道”。4. 核心代码实现与实操解析4.1 图的编排代码前面我们设计了状态和节点函数现在把它们组装成图。这里可以直接贴一个完整可运行的 LangGraph 编排from langgraph.graph import StateGraph, END def build_graph(): graph StateGraph(RagState) graph.add_node(rewrite, rewrite_node) graph.add_node(retrieve, retrieve_node) graph.add_node(grade, grade_node) graph.add_node(generate, generate_node) graph.add_node(fallback, fallback_node) graph.set_entry_point(rewrite) graph.add_edge(rewrite, retrieve) graph.add_edge(retrieve, grade) graph.add_conditional_edges( grade, route_after_grade, { generate: generate, rewrite: rewrite, fallback: fallback, }, ) graph.add_edge(generate, END) graph.add_edge(fallback, END) return graph.compile()这段代码里有几个值得说清楚的地方。set_entry_point定义起点起点必须是整个流程图最先执行的节点在反思场景下起点是“改写”而不是“检索”因为我们要让第一轮查询就经过语义扩展。add_conditional_edges是核心它从评分节点出发根据route_after_grade的返回值选择下一步这样就实现了回路和分支。compile()返回一个可执行的 Runnable后续接.invoke()或.stream()都行。4.2 节点实现中的 LangChain 细节由于我们用 LangChain 的 ChatOpenAI 调用 LLM有几个封装细节要说明。关于 JSON 输出解析我见过太多人在这里踩坑模型经常在 JSON 前后加说明文字直接json.loads会抛异常。我的做法是给 ChatOpenAI 指定response_format{type: json_object}这样模型会输出严格 JSON再用json.loads和异常 fallback 双层解析。实际项目里parse_json函数要先试json.loads失败则用正则抽取大括号内容再试还不成就当作“no”处理宁可漏过不可错放。from langchain_openai import ChatOpenAI import json, re llm ChatOpenAI(modelgpt-4o-mini, temperature0, response_format{type: json_object}) def parse_json(text: str) - dict: try: return json.loads(text) except json.JSONDecodeError: m re.search(r\{.*\}, text, re.DOTALL) if m: try: return json.loads(m.group()) except Exception: return {} return {}再提醒一点所有 LLM 调用都建议显式设置超时和重试。生产中经常遇到 LLM 服务超时如果节点函数抛异常整个图就会中断。我封装了一个llm_call_with_retry对瞬时异常重试两次间隔 1 秒、2 秒超过三次直接返回一个空结果让路由走进 fallback 分支而不是挂死整个进程。4.3 编译与运行的观察方式编译完成后开发期可以用graph.invoke(initial_state)直接跑但更重要的是利用 stream 模式观察每一步状态。LangGraph 支持graph.stream(initial_state, config, stream_modeupdates)会逐步输出每个节点的状态变更。我在开发时把结果打印出来逐行看每一轮的改写是什么、检索到哪些文档、评分结论是什么、反思反馈是什么找出“哪一步开始跑偏”就会非常直观。initial_state { question: 公司笔记本坏了找谁报修, rewritten_question: , documents: [], graded_docs: [], reflection_feedback: , attempt: 0, max_attempts: 3, generation: , used_documents: [], } for chunk in graph.stream(initial_state, stream_modeupdates): for node_name, state_update in chunk.items(): print(f节点: {node_name}, 更新: {state_update})这个观察习惯非常重要。曾经有一次用户反馈“回答总是缺漏”我光看最终输出根本看不出问题用 stream 逐步看才发现评分节点把很多“背景型”文档判定为相关但生成节点却因为 Prompt 结构问题把它们当成主要依据导致遗漏核心数据。这个问题就是靠观察每步状态发现的。5. 生产化落地并发、知识库选型、记忆与安全5.1 RAG 知识库到底能不能存图片这个热搜词问得挺有意思很多人把“RAG 知识库”理解为某种特殊的数据库问能不能放图片。其实 RAG 不是一个存储引擎而是一套检索增强流程。如果你要支持的图片是“文档中的图表、截图”那思路是把图片转成文字描述或用多模态模型抽取图表要点跟正文一起切成片段进向量库如果用户后续问的是图片内容检索到的片段里就带有图片描述和原图路径生成回答时可以把原图路径一起返回给前端展示。如果你的“图片”是指用户上传的纯图片文件那需要多模态 embedding 模型和图像检索能力普通文本向量库确实存不了图像本身的语义。所以更准确的回答是知识库可以“接住”图片但前提是拆解和索引设计得当不能简单地把图片二进制塞进向量库。我在实际项目里用的是“图文双通道”方案对每张图先做 OCR 和视觉理解生成结构化描述字符串再作为文本分段入库原图 URL 存在文档元数据字段里。用户问“报销流程图里第二步是什么”向量检索命中的其实是那张图的文字描述片段但生成回答时我会附上原图 URL前端可以弹图。这个方案成本低效果也很直观。5.2 向量库 vs 知识图谱 vs 结构化知识库怎么选很多人把“知识库”当成一个笼统概念其实按数据结构可以分成三类对应的检索方式完全不同。我做个表格类型适合的数据检索方式典型场景在我这套 Agent 里的角色向量知识库非结构化文本、PDF、网页语义相似度规章制度、FAQ、操作手册主检索通道支撑大多数问答知识图谱 KG实体和关系图遍历 / 多跳查询组织架构、产品关系、因果关系补充检索处理多跳问题结构化知识库表格、SQL、CSV精确查询 / 聚合销售额、库存、人员数据精确数字查询的首选选型不是非此即彼。我强烈建议把三者叠加起来主 RAG 用向量实体验证和关联查询用知识图谱数字聚合查询接 SQL。这套项目里我在多路检索中已经埋了标题检索和实体匹配背后的实现就是一张小知识图谱。如果你一开始就想全量建设知识图谱成本很高建议先从“标题实体”起步把常用 500 个实体映射建好效果提升就很明显。5.3 Agent 记忆从短期到长期搜索热词里“agent记忆”出现频率很高。在 Agentic RAG 里记忆分为三个层面。对话内记忆把用户最近几轮的问题和系统回答打包进系统 Prompt让 Agent 理解指代比如“刚才说的那个流程”会话摘要记忆当对话很长时用 LLM 把历史对话总结成摘要避免上下文爆炸长期用户记忆记住该用户喜欢的表达方式、常用偏好这需要把用户画像写入向量库或配置中心。我的项目第一期只做了对话内记忆在每次用户请求前把最近 4 轮对话作为上下文放进改写和生成节点效果提升显著。LangGraph 对记忆有官方支持通过 checkpoint 机制可以持久化图的状态在并发场景下还能实现多用户会话隔离。我的建议是短期用内存态即可但生产环境一定要接 Redis 或 Postgres 的 checkpoint saver否则服务重启后所有会话状态都会丢失Agent 的对话连续性就断了。5.4 AI Agent 怎么扛并发“AI Agent 怎么扛并发”是完全值得单开一篇的话题。首先明确一个事实LangGraph 本身不解决并发它只负责编排和状态管理。要把这套东西扛住生产流量我总结出四条关键路径。第一条接口层用异步套接。用 FastAPI 包装图的执行接口用async def调用await graph.ainvoke(...)这样才能让 FastAPI 的事件循环在 LLM 和检索等待期间处理其他请求避免每个请求独占一个线程。第二条检索层并发执行。多路召回里向量检索、BM25、标题检索彼此独立在 retrieve_node 内部用asyncio.gather并发执行能把单次检索耗时从三路串行的 200 毫秒降到 80 毫秒左右。这是代码层面最直接的优化。第三条模型调用是最大的瓶颈一定要加缓存和限流。我把 embedding 结果和 LLM 的评分结果都做了语义缓存按查询的 embedding 相似度 0.98 命中缓存。实际线上 40% 的高频问题都会命中缓存LLM 调用量直接减半。另外越高的并发越要关注模型服务限流我用十次重试的指数退避策略处理 429 错误并设置并发信号量控制同一时刻的 LLM 请求数。第四条水平扩容与状态外置。图是有状态的如果要水平扩容到多个副本必须把 checkpoint 外置到 Redis 或 Postgres否则副本之间状态不同步。加上 checkpoint 外置后部署到 K8s 按流量 HPA 扩容配合语义缓存单副本 QPS 从 2 提到 8扩容到 10 副本后基本能应对日常峰值。这还是没考虑本地推理服务的情况如果用了本地模型并发瓶颈会更明显建议把推理服务独立部署并通过队列削峰。5.5 Agent 安全必须提前考虑的事Agent 化的 RAG 比普通 RAG 更强大也更需要安全护栏尤其是多轮反思循环放大了用户输入的引导能力。我在上线前重点做了四件事。第一是输入过滤。对用户问题做正则和 LLM 双重检测识别提示注入模式比如“忽略以上指令”“你现在是系统管理员”这类话术。第二是权限隔离。知识库里不同文档属于不同部门检索前通过用户身份过滤可见文档集合这一步必须在向量检索前做否则检索结果会泄露不可见内容。第三是输出审核。生成阶段用独立的审核模型对答案做合规检测简单场景用关键词黑名单加敏感词库复杂场景再加一个 LLM 审核节点。第四是引用溯源。最终回答必须携带来源文档 ID方便用户点开原文核对这对降低幻觉投诉非常有效。6. 常见问题与排查技巧实录6.1 问题速查表我把真实使用中碰到的高频问题整理成一张速查表遇到问题先对号入座现象可能原因排查手段解决方案检索总是不命中查询语义鸿沟大打印 rewrite 节点输出增强改写 Prompt补充同义词表评分全不过评分标准太严抽样评审各文档人工打标放宽 Prompt 措辞调低通过阈值反思循环不收敛改写结果没变化观察两轮 rewritten_question 对比给改写节点传入上一轮反馈强制变化回答来源无法追溯生成 Prompt 未要求标注引用检查最终文本中来源 ID 数量重写生成 Prompt 结构逐句标注LLM 超时报错模型服务限流或网络抖动看节点异常栈加重试和超时设置 fallback 分支并发高峰期响应慢LLM 串行调用看请求链路耗时asyncio.gather 并发多路加语义缓存6.2 容易忽略的坑和避坑经验第一个坑循环里状态无限膨胀。反思循环如果让检索节点在每轮都追加新文档到同一个列表状态会越来越大Prompt 会越塞越满最后超出上下文长度。一定要限制状态大小我设置文档最多保留 12 个超过就由评分节点先剪枝再说。第二个坑把原始问题弄丢了。我在评分节点里强调要用原始问题其实还有更隐蔽的场景生成节点如果用了改写后的问题模型会回答“改写后的问法”而不是用户真正的问题。所以每个使用 question 的地方要统一来自state[question]我甚至在状态里加了注释来防止后续接手的人改错。第三个坑评分 LLM 的成本失控。如果用最强的模型逐文档评分一天几十万次调用费用会非常可观。我的经验是评分任务用 mini 级别模型足够但生成任务用大模型保证质量。另外把评分结果做缓存同一文档加同一问题的评分只算一次。第四个坑反思反馈太过笼统。如果反思反馈只是“增加检索关键词”下一轮改写也不知道怎么改。我的做法是把反馈结构化必须说明“补充哪个领域的哪类术语、排除哪个歧义词义”这样下一轮改写才真正有抓手。6.3 评估与迭代方法论最后说下怎么评估这套 Agentic RAG 的效果。传统 RAG 用召回率和答案准确性指标但 Agentic RAG 引入了循环和决策评估维度必须加宽。我维护了一个 300 条真实客服问题的评测集按四个维度打分答案正确性人工打分、引用准确性核对引用是否真实存在且支撑结论、幻觉出现率是否出现材料外信息、效率平均轮次、平均延迟。上线前每周跑一轮全量评测整体准确率基线提升到 86%幻觉率从基线的 12% 降到 4%平均轮次 1.8有 30% 的查询会经历至少一次反思循环。迭代方向上我接下来会重点做三项工作一是把反思反馈从“文本建议”升级为“结构化行动计划”比如指定换检索器、换语言模型、换分块策略让循环更可控二是引入评估节点反向优化改写 Prompt自动形成“失败查询、改进反馈、改写效果”的闭环日志三是把长期用户记忆接上让系统在多次交互后记住用户偏好这会让回答质量再上一个台阶。我实际跑下来的体会是LangGraph 这套图编排的思路真正改变的不是代码写法而是调试和迭代的方式。以前调 RAG 只能看最终输出的好坏现在每一步的决策都能看到、能干预、能回放这比任何“智能”的噱头都实在。最后分享一个实用技巧评估 Agentic RAG 时别只看成功率一定记录“反思循环到底多少次挽救了失败检索”这个数字会告诉你系统的反思机制是不是真的在起作用。如果反思率一直很低说明你的检索主链路已经足够好反思循环反而是不必要的复杂度这时候简化架构比堆功能更值得。
返回列表