如何用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 的本质区别

用一个表格对比最直观：

我在实际项目中感受最深的一点是：Agentic RAG 的“Agent”不是指系统更聪明，而是指系统拥有了决策能力。它决定了下一步调用哪个工具、是否重试、是否终止。LangGraph 之所以成为我最终选择，是因为它在 Python 里用“图”这种数据结构天然表达决策流，节点的返回值就是图的状态更新，条件边就是决策本身，比手写一堆 if/else 加 while 循环维护起来清晰太多。

1.3 这个项目适合谁

如果你只是想在本地跑一个简单的文档问答，那不用上 Agentic RAG，Chroma 加 LlamaIndex 十分钟就能搞定。但如果你遇到过下面几种情况，就说明该上这套方案了：用户问法千奇百怪导致频繁检索失败；知识库内容垂直，术语和口语差异巨大；领导要求在问答中给出可追溯的引用来源；或者你已经在用 LangGraph，但想知道“反思”这类循环结构怎么真正落地。

2. 整体架构与核心设计拆解

2.1 LangGraph 为什么适合编排 Agent

LangGraph 的定位是“给语言模型应用加上可编排的有向图运行时”。它的核心抽象只有三个：节点（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, k=8): 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_k=4): 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_k=2): 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, k=20, score_threshold=0.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(model="gpt-4o-mini", temperature=0, 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_mode="updates")，会逐步输出每个节点的状态变更。我在开发时把结果打印出来，逐行看每一轮的改写是什么、检索到哪些文档、评分结论是什么、反思反馈是什么，找出“哪一步开始跑偏”就会非常直观。

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_mode="updates"): 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 结构化知识库怎么选

很多人把“知识库”当成一个笼统概念，其实按数据结构可以分成三类，对应的检索方式完全不同。我做个表格：

选型不是非此即彼。我强烈建议把三者叠加起来：主 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 问题速查表

我把真实使用中碰到的高频问题整理成一张速查表，遇到问题先对号入座：

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 时，别只看成功率，一定记录“反思循环到底多少次挽救了失败检索”，这个数字会告诉你系统的反思机制是不是真的在起作用。如果反思率一直很低，说明你的检索主链路已经足够好，反思循环反而是不必要的复杂度，这时候简化架构比堆功能更值得。