做知识库问答项目做了快半年我最大的感触是RAG 的上限取决于它会不会自我纠错。刚上手那会儿我用的是教科书级别的标准流程——用户提问向量检索拼 Prompt丢给 LLM 生成。简单问题还好一旦遇到“帮我对比一下 A 和 B 两套方案的优劣并结合我们今年的预算给个倾向性建议”这种复合问题系统就开始答非所问要么检索不全要么拿到的文档和问题八竿子打不着要么模型对着几段相似文本自信地编了一个答案。最后朋友一句话点醒我你的 RAG 只会执行不会反思。于是我把目光投向 LangGraph开始构建一个真正“会反思”的 Agentic RAG 问答 Agent。这篇文章就把我从 0 到 1 的实现过程、代码细节和踩坑记录完整展开适合已经跑通过基础 RAG、想进一步做智能体化改造的开发者参考。1. 先把问题看透传统 RAG 到底卡在哪里1.1 你用的“伪 RAG”其实只是一条流水线很多教程里教的 RAG本质是一条固定流水线query - 向量检索 - 拼上下文 - 生成。它的问题不在于“检索”和“生成”这两个动作本身而在于没有任何一个环节会审视自己的输出质量。检索结果的排名是固定的哪怕排第一的那段文本和问题根本不相关流水线也会把它喂给大模型生成的答案只要格式对就没人检查它是不是答非所问。我拿一个真实场景说明。我的知识库里有一批产品手册和售后工单。用户问“上次那个登录报错的问题后来有没有在 2.3.1 版本里修掉”传统 RAG 的做法是把“登录报错”“2.3.1 版本”拆成语义向量去检索。但售后的工单里描述问题时用的词可能是“认证失败”“session 过期”版本号在文档里又经常被写成v2.3.1、2_3_1各种格式。向量检索很可能召回一堆“相关但不对路”的片段LLM 只能基于这些错误上下文硬答。结果就是用户拿到一个看似流畅、实则没解决任何问题的答案。这类失败有一个共同特征系统缺少“知道自己不知道”的能力。它不评判当前检索结果是否足够支撑回答也不追问“还缺什么信息”。这就是传统 RAG 的天花板。1.2 从“检索增强生成”到“Agentic RAG”差别在哪里Agentic RAG 不是把检索流程包一层 Agent 壳子它的核心变化是把“检索—评估—再检索—生成”的控制权交给智能体让每一步都由模型根据当前状态做决策。整个流程变成了一个闭环回路而不是单向管道。我整理过一张对比表虽然博客里画不了图但表格更直观维度传统 RAGAgentic RAG决策方式固定管道检索后直接生成模型自主决定是否检索、检索什么、是否重查查询处理用户原句直接向量化支持条件路由、查询改写、拆解子问题质量反馈无反馈一次到底生成后进行自我评估失败则触发补偿机制适用问题简单事实型问答多跳推理、对比类、需要实时知识的问题可解释性黑盒无法追踪每一步状态可见可回溯、可干预工程复杂度低几行代码跑完高需要图编排、状态管理有一件事得说清楚Agentic RAG 不是万能药。如果你只做“某产品的退货政策是什么”这类单跳检索传统 RAG 就够了引入反思回路反而增加延迟和 token 开销。但做知识库问答 Agent用户的问题一定五花八门所以我很早就决定要么不做要做就做成“会反思”的。1.3 顺便把三类知识库的边界理清楚热搜词里有人问“kg 知识库、rag 知识库和结构知识库怎么区分”这里顺带说一嘴。知识库只是数据的组织形式和上层是不是 Agentic 没有必然关系结构知识库指强 schema 约束的数据源比如关系型数据库里的订单表、配置表适合精确查询SQL 或参数化查询。RAG 知识库通常指非结构化文本切块 向量化后组成的集合适合模糊语义检索但无法做精准条件过滤。知识图谱KG用实体和关系组织三元组适合回答“谁和谁之间有什么关系”这类多跳问题。做 Agentic RAG 时我推荐你把它们混着用。比如用户问“最近一个月哪些工单提到了登录异常”普通向量库回答不了“最近一个月”这个时间约束如果我把工单也同步进了结构化表让 Agent 决定走 SQL 还是走向量检索效果会好很多。这篇文章后面的架构里我会预留工具扩展位实际上就是为这种异构数据源准备的。2. 方案选型为什么是 LangGraph而不是自研状态机2.1 LangGraph 解决的核心矛盾状态可见 循环可控一开始我考虑过两条路一是自己用while循环 一堆if-else写状态机二是直接用 LangChain 自带的create_retrieval_chain之类的高层封装。前者写起来自由但一旦节点多了状态没有统一管理很快就变成意大利面条代码后者太死根本没法插入“反思”这个环节。LangGraph 正好卡在中间。它把 Agent 的每一步定义成图节点节点之间的跳转由条件边决定而所有共享数据都放在一个全局状态对象里。这意味着每一步之间不是隐式传递字符串而是结构化更新state调试时可以打印任意时刻的完整快照图允许形成环这是传统 LangChain 链式封装很难做到的而“反思”恰恰需要回路生成 - 评估 - 不通过 - 重写查询 - 再检索可以用interrupt或节点回调在中间步骤插入人工审核这在生产环境里非常关键。2.2 状态、节点、条件边LangGraph 的三要素我用大白话拆一下 LangGraph 的基本概念。一个图有三个东西必须想清楚State状态相当于所有节点共用的“工作台”。所有节点只能通过state读取和写入信息。Node节点一段 Python 函数。输入是当前state输出是一个字典里面是要更新的字段。Edge边定义节点之间的跳转。普通边是无条件跳转条件边根据当前state里某个字段的值动态决定下一步去哪个节点。我的第一版代码里状态结构长这样from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class RAGState(TypedDict): question: str # 原始问题 rewritten_question: str # 改写后的问题 documents: list # 检索到的文档 generation: str # 最终答案 reflections: list # 反思记录 iteration: int # 当前轮次防止死循环 messages: Annotated[list, add_messages] # Agent 的消息历史注意messages字段用了add_messages这个 reducerLangGraph 会用它的语义来做消息追加而不是整体覆盖。如果不用 reducer后写入节点的值会直接把前一个节点写的字段覆盖掉。这个细节很容易踩坑后面调试部分我再展开。2.3 我最终敲定的五节点工作流我的知识库场景比较复杂用户问题跨度很大最终设计了一个五个核心节点 两个条件路由的图。流程可以这样描述入口判断agent 节点大模型基于当前问题决定下一步动作——需要检索还是已有足够上下文可以回答。检索retrieve 节点把当前查询向量化从知识库召回 top_k 文档。生成generate 节点基于检索到的文档生成带引用的答案。反思reflect 节点把问题、答案、文档一起喂给模型让模型判断答案是否“完整回答”了问题输出pass或fail并给出缺失信息。查询改写rewrite 节点如果反思失败模型根据缺失信息改写查询回到检索节点再来一轮。条件路由的核心逻辑是反思后判断pass则直接返回最终答案判断fail则判断迭代次数是否超限没超限就进入改写节点超限就用当前已有答案兜底返回。这样一个回路既给了 Agent 自我纠错的机会又不会无限循环烧钱。3. 代码实战构建会反思的 Agentic RAG3.1 环境准备与项目结构我建议用uv或pip新建一个干净环境Python 版本 3.10 以上。核心依赖如下pip install langgraph langchain langchain-openai langchain-community langchain-text-splitters faiss-cpu我用的是 FAISS 做向量库因为本地跑、小规模知识库场景 FAISS 完全够用不用额外起服务。如果你检索量在百万级以上再考虑 ES、Milvus 或者 pgvector 不迟。嵌入模型我用的是本地化的 BGE 系列BAAI/bge-small-zh-v1.5好处是不走外部 API响应稳定适合国内网络环境。项目结构大致是agentic_rag/ ├── main.py # 主图组装与调用入口 ├── nodes.py # 五个节点的具体实现 ├── state.py # 状态定义 ├── retriever.py # 向量库与检索器初始化 └── prompts.py # 各节点的 prompt 模板3.2 定义状态和检索器先看状态定义这一部分对应前面设计里的RAGState不再重复。重点是检索器初始化。我封装了一个函数把文档切分、向量化、构建检索器一步做完# retriever.py from langchain_community.vectorstores import FAISS from langchain_huggingface import HuggingFaceEmbeddings from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.documents import Document def build_retriever(docs, chunk_size512, chunk_overlap80, top_k6): # 使用递归字符切分器优先按段落切再按句号兜底 splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , ., ], keep_separatorend, ) chunks splitter.split_documents(docs) embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-small-zh-v1.5, encode_kwargs{normalize_embeddings: True}, ) vectorstore FAISS.from_documents(chunks, embeddings) return vectorstore.as_retriever(search_kwargs{k: top_k})这里有个值得说道的点separators里我把中文句号、感叹号、问号都加进去了而且设了keep_separatorend意思是切分时把标点符号保留在上一块结尾。默认的RecursiveCharacterTextSplitter对中文支持不太行经常把一个完整句子从中间切断导致语义碎片化。这个问题在后续检索质量上影响很大。3.3 五个节点逐一实现接下来是最关键的部分。先看prompts.py里三个核心模板# prompts.py FROM_SYSTEM 你是一个知识库问答助手请基于【上下文】内容回答问题。 如果上下文中没有足够信息明确说“知识库中没有相关资料”不要编造。 回答结尾附上引用的来源编号格式[来源1][来源2]。 REWRITE_SYSTEM 你是一个查询优化专家。下面是用户问题、已有上下文片段以及当前答案缺失的信息。 请重写一个更精确的搜索查询帮助在知识库中找回缺失信息。 只输出重写后的查询本身不要任何解释。 REFLECT_SYSTEM 你是答案质量评估专家。请判断下面的回答是否完整回答了用户问题。 判断标准 1. 是否直接回答了问题核心而不是答偏或回避 2. 是否基于给定上下文没有编造 3. 是否有关键信息缺失导致用户无法据此行动。 如果回答满足要求输出PASS 如果不满足输出FAIL并在下一行用一句话说明缺失什么。注意这些 prompt 里我特意加了“不要编造”“缺失就承认缺失”的约束。反思节点尤其重要如果这个 prompt 写得含糊模型很容易变成“全票通过”的复读机整个反思回路就没有意义了。然后实现节点。nodes.py里五个函数核心逻辑如下# nodes.py from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0.2) # 1. 入口决策 Agent def agent_node(state: RAGState) - dict: if not state.get(documents): return {next_action: retrieve} return {next_action: generate} # 2. 检索节点 def retrieve_node(state: RAGState) - dict: query state.get(rewritten_question) or state[question] docs retriever.invoke(query) return {documents: docs, iteration: state.get(iteration, 0) 1} # 3. 生成节点 def generate_node(state: RAGState) - dict: context \n\n.join( f[来源{i1}] {d.page_content} for i, d in enumerate(state[documents]) ) resp llm.invoke( [ {role: system, content: FROM_SYSTEM}, {role: user, content: f问题{state[question]}\n\n上下文{context}}, ] ) return {generation: resp.content} # 4. 反思节点 def reflect_node(state: RAGState) - dict: pret llm.invoke( [ {role: system, content: REFLECT_SYSTEM}, {role: user, content: f用户问题{state[question]}\n\n当前回答{state[generation]}}, ] ) result pret.content.strip() if result.startswith(PASS): return {reflections: [result], should_continue: pass} missing result.split(\n)[1] if \n in result else return {reflections: [result], should_continue: fail, missing_info: missing} # 5. 查询改写节点 def rewrite_node(state: RAGState) - dict: resp llm.invoke( [ {role: system, content: REWRITE_SYSTEM}, {role: user, content: ( f原始问题{state[question]}\n f缺失信息{state.get(missing_info, )}\n f已有部分答案{state[generation]} )}, ] ) return {rewritten_question: resp.content.strip()}这个实现里我做了一个重要决定入口的agent_node目前逻辑很简化只是判断“有没有文档”。真正复杂场景下这里应该是一个完整的 ReAct 风格的 LLM 决策可以决定“调用检索工具”还是“直接回答”。我为了演示反思回路先保持简单但你完全可以把agent_node升级成带工具调用的 LLM 节点。3.4 用 StateGraph 组装反思闭环最后是主图组装。LangGraph 的核心 API 是StateGraph我先创建图注册节点然后按条件边把它们连起来# main.py from langgraph.graph import StateGraph, END def route_to_next(state: RAGState) - str: if state.get(should_continue) pass: return end if state.get(iteration, 0) 3: return end return rewrite builder StateGraph(RAGState) builder.add_node(agent, agent_node) builder.add_node(retrieve, retrieve_node) builder.add_node(generate, generate_node) builder.add_node(reflect, reflect_node) builder.add_node(rewrite, rewrite_node) builder.set_entry_point(agent) builder.add_edge(agent, retrieve) builder.add_edge(retrieve, generate) builder.add_edge(generate, reflect) builder.add_conditional_edges(reflect, route_to_next, { rewrite: rewrite, end: END, }) builder.add_edge(rewrite, retrieve) app builder.compile() # 调用 if __name__ __main__: result app.invoke({ question: 登录报错的问题在 2.3.1 版本里修复了吗, iteration: 0, documents: [], }) print(result[generation]) print(反思记录, result[reflections])这里的关键点是条件边的映射表。route_to_next返回的是字符串add_conditional_edges里的字典负责把这个字符串映射到目标节点名。我留着iteration 3的兜底是为了防止反思失败时无限循环“连续改写了三次还找不到资料就老老实实用最后一次生成的答案返回同时把反思记录抛给调用方让上层知道这次回答可能不可靠。”这种“答不上来也明说”的设计在我看来是生产级 RAG 和玩具级 RAG 的分水岭。用户宁可看到一句“知识库中暂无相关资料”也不愿意看到模型编一个看似专业的假答案。4. 调试与踩坑实录这些坑我替你踩过了4.1 调参经验chunk_size 和 top_k 怎么定讲真RAG 项目里决定检索质量的第一因素不是模型而是切块参数。我一开始用默认的chunk_size1000结果复合问题经常只被分在同一个大块里导致检索召回的内容要么互相重复、要么漏掉关键分支。后来我做了个小实验把 chunk_size 分别设成 256、512、768对 50 个测试问题做了对比chunk_size平均检索命中率生成答案完整率备注25678%65%上下文碎片化需要 top_k 调大51286%82%最均衡我最终选了这个76883%77%块太大细粒度信息容易丢规律是中文文档在 512 左右是个甜点。因为 bge-small-zh 这类中文嵌入模型对 256 到 512 token 的文本段建模效果最好太长会稀释语义太短又会丢掉上下文。top_k不要盲目调大我的知识库文档质量参差top_k 从 6 改到 10 之后检索到的段落相关性明显下降反而拉低了生成质量。最终定在top_k6。我还发现一个细节chunk_overlap80对中文特别重要。中文没有天然空格分词切块时经常把一个概念切两半overlap 能让相邻块共享一些边界文本避免关键术语被截断。4.2 “反思”容易写成假反思prompt 得硬核第一次跑通反思回路时我挺失望的输出结果里reflections几乎都是PASS哪怕答案明显有遗漏。查了几轮才发现问题出在两个地方一是反思 prompt 缺少具体标准。“是否完整回答”这种表述太模糊模型倾向给好评。我后来在 prompt 里加了三段硬性判断标准直接回答、基于上下文、关键信息缺失并且要求输出必须是PASS或FAIL 一句话说明逼模型表态。二是模型自评默认不愿“自我否定”。后来我在反思节点里把“当前答案”和“检索到的所有文档”都喂给了评估模型让评估模型不是凭空挑错而是拿着原文对照检查。这个改动让 fail 率从不到 5% 提升到了 30% 左右反思回路终于开始发挥真实作用。4.3 token 成本与上下文长度控制反思回路的效果上来了账单问题也跟着来了。一次不通过的查询实际要跑agent 决策 检索 生成 反思 改写 再检索 再生成 再反思最多可以到三轮迭代token 消耗是传统 RAG 的 5 到 8 倍。我做了三个优化把反思和生成拆成两个模型调用反思用更小的gpt-4o-mini生成用gpt-4o。这样反思虽然多了一轮调用但 token 单价低。改写节点只传缺失信息摘要不把全部历史消息塞进去。上面代码里我只传了“原始问题 缺失信息 已有答案摘要”这就是刻意控制的。限制最大迭代次数为 3从概率上讲三轮检索还补不上信息再多也是如此。如果你做的是垂直领域的专业知识库可以考虑在反思节点前加一层“置信度判断”当模型对答案本身置信度很高时连反思都跳过。这个 trick 能省下大约 30% 的 token 成本。4.4 本地搭建与并发瓶颈热词里有人问“怎么在 Mac 上搭建 RAG 知识库”我本地的开发环境就是一台 MacBook。LangGraph FAISS BGE 嵌入模型这套组合在 Mac 上跑得很顺不需要 GPU。嵌入模型用的是 CPU 推理2000 篇文档切块后入库大概 3 分钟完全可接受。但要注意FAISS 默认不支持增量添加我每次更新文档库都是重建索引。如果是频繁更新的场景建议用FAISS.add_documents()配合持久化或者直接换 pgvector。并发这块我也踩了坑。LangGraph 的图状态对象不是线程安全的多个请求同时invoke同一个编译后的图时state会被互相污染——我在测试接口时发现一个用户的查询偶尔会带上另一个用户的检索结果。解决方案是每个请求创建独立的图实例或者用app.ainvoke在异步事件循环里隔离。异步版本适合 FastAPI 集成# 在 FastAPI 路由中 app.post(/ask) async def ask(question: str): async with async_app.astream({question: question, iteration: 0, documents: []}): last_event None async for event in async_app.astream(...): last_event event return last_event更稳妥的方式是维护一个“图实例池”每个请求从池里取一个干净的图对象用完归还。这样可以规避 LangGraph 内部状态被并发请求覆盖的问题。4.5 常见问题速查表症状可能原因解决方向检索结果和问题完全不相关嵌入模型选型差或文档切块太粗暴换中文专用嵌入模型调 chunk_size 和 overlap反思节点全部 PASS不触发重查反思 prompt 太模糊加三条硬性判断标准把上下文文档也喂给评估模型一模一样的问题回答不稳定LLM 温度太高生成节点温度降到 0.2 以下反思节点用 0多次改写后还是答不上来知识库里真的没有相关信息迭代上限兜底明确告知用户缺失资料接口并发后答案串了共享图实例的 state 被污染每个请求独立图实例或用 async 图本地内存占用持续上涨FAISS 索引常驻 大文档列表流式处理 chunks用完释放避免把全部文档塞进 state我最后悔的一件事是前期没做“检索—生成”两阶段联调。单独看检索质量、单独看生成质量都没问题一旦接上反思回路很多问题才暴露。建议你从一开始就给测试集打好标签每条用例标注“期望检索到的来源”“期望答案要点”这样调参时有个客观标尺。5. 从“会反思”到“能干活”扩展思路与适用边界5.1 知识库到底能不能存图片、怎么处理多模态热词里有人问“RAG 知识库能存储图片嘛”这里统一回答普通的文本向量检索链路存不了图片但可以存“图片的描述”。我试过的常见做法是把图片先用多模态模型比如 GPT-4o、Qwen-VL生成一段详细描述再把描述文本切块进入向量库。用户问“文档里那张架构图讲了什么”检索到的是图片描述文本生成答案时仍然可以把图片以 Markdown 形式附在上下文里前端展示给用户。如果图片本身包含大量图表信息只靠描述会损失细节。更重量的方案是引入多模态向量模型比如 CLIP、SigLIP 做图文联合嵌入然后把图片和文本统一进同一个向量空间。但工程复杂度和资源占用会直线上升。我的建议是90% 的文档型知识库场景图片转描述就够用没必要一开始就上多模态检索。5.2 Agentic RAG 和纯 Agent 的边界在哪热搜词里有“agent框架与编排”“harness 和 agent 区别”我借这个地方把边界说透。Agentic RAG 本质上是“带检索工具的 Agent”它和通用 Agent 的区别在于RAG Agent 的目标是回答知识库问题动作空间收敛在“检索、改写、生成、反思”这几个节点上而通用 Agent 可以自主调用任意工具、操作外部系统、执行多步任务。实践里我更喜欢一种分层方式底层是 LangGraph 这个编排“骨架”上层定义节点的行为策略。RAG Agent 只是骨架上的一个专用 harness——把检索、生成、反思这些工具按固定拓扑串起来同时保留 Agent 的自主决策能力。至于“Multi-Agent 架构”“Agent 记忆”这些话题属于同一套图编排方法里的其他玩法等你的 RAG Agent 稳定跑通了再扩展不迟。5.3 什么时候你根本不需要反思回路说实话不是每个 RAG 项目都值得上 Agentic 这套方案。我在下面这些场景里会把反思回路关掉FAQ 问答问题很固定直接做“问题相似度匹配 标准答案返回”就行让 LLM 生成反而是浪费低延迟场景比如在线客服首响必须低于 2 秒一次检索 一次生成的链路已经极限了知识库很干净且问题大多是单跳查询反思和改写带来的提升不大成本却翻倍。一个比较务实的做法是在图里加一个“快速通道”路由让入口 agent 节点判断如果问题明显是简单事实类直接走retrieve - generate - END只有遇到多跳、对比、追问类问题时才进入反思回路。这样既保住了复杂问题的质量又不会让简单问题变慢变贵。个人体验调试反思回路时最让我惊喜的一个瞬间项目收尾时我拿了一个“刁钻”问题测试用户问“2.3.1 版本发布说明里有没有提到登录认证超时的修复跟 2.3.0 的改动有什么区别”。第一轮检索答案只提到了“认证超时修复”但没说和 2.3.0 的对比反思节点给出 FAIL理由是“缺少版本间差异信息”。查询改写节点把问题改写成“2.3.1 vs 2.3.0 登录认证超时改动对比”第二轮检索真就把两个版本的发布说明片段都拉了出来生成了一版带对比和引用的回答。看到这个结果的时候我还是挺感慨的——它不再是一个机械的检索管道而是真的像一个项目助理发现自己漏了关键信息会主动补充后再提交答案。当然这套系统离“完美”还差得远比如反思节点偶尔会误判、改写节点可能把问题改得面目全非、多模态扩展还没做深。但如果你也在做知识库问答我建议不要只停在“能检索、能生成”这一步给系统加上反思闭环你会看到完全不一样的回答质量。最后提醒一句先把测试集建好再做 Agentic 化改造否则你会被各种“薛定谔的检索质量”搞崩溃。