别一提 RAG 就默认是向量检索 拼 Prompt 丢给大模型这条流水线。我见过太多团队把知识库问答做成查得到就答、查不到就编最后上线后被业务方追着骂。问题不在 RAG 这个思路而在实现方式太死板查询不会改写、检索结果没人把关、遇到实时性问题直接哑火。这篇文章我直接用 LangGraph 带你实现一个不那么呆的 RAG 系统——它会先判断该走哪条路检索结果差会自动重写查询再搜一次本地知识库实在搞不定时还能自己去联网找答案。全程带可运行的代码不是概念演示是能直接改吧改吧丢到项目里的那种。1. 传统 RAG 的死板到底死在哪三个最常见的失败现场1.1 第一个现场切块边界把答案切没了RAG 第一道工序就是把文档切成小块再向量化很多团队图省事按固定字符数硬切token 数设个 500 就完事。结果就是一段完整的业务规则刚好被从中间劈开前半段在 chunk_001后半段在 chunk_002用户问的问题涉及的知识点横跨两个 chunk。检索时 TopK 取回来的是两个残缺片段模型拼来拼去也拼不出完整答案最后给你一本正经地胡说八道。我当时处理过的一个真实案例客户问产品 A 在华东区域的退货政策里关于包装破损的条款是什么原文段落里刚好先是华东区域退货政策接着是包装破损必须拍照留证中间被一个\n\n隔开固定切块就把它们分到了两个向量里。最终检索只命中了前半段模型回答说华东区域有退货政策但未提及包装破损这就是典型的检索相关但信息不全。1.2 第二个现场多跳问题单次检索根本答不了传统 RAG 是一次检索、一次生成但真实业务问题往往是多跳的。比如小王上个月负责的那个项目的验收报告里提到的主要风险有哪些这个问题表面上是问报告内容实际上先要确定小王上个月负责了哪个项目再去查那个项目的验收报告。你指望一次向量检索同时命中员工信息和项目文档两个库里的内容基本不现实。更麻烦的是即使用户问题本身不复杂比如对比一下 A 方案和 B 方案在成本上的差异单次 TopK 检索经常只召回 A 方案相关的段落B 方案的内容因为相似度排序被挤到 TopK 之外。模型拿到不完整信息生成的对比自然也是片面的。1.3 第三个现场时效性问题把知识库变成过期罐头知识库再勤快更新也赶不上外部信息的实时变化。用户问XX 公司昨天发布的财报里毛利率是多少你的向量库索引的是上个月的数据检索结果再准也是过期的。传统 RAG 没有任何机制感知我的知识不够新它只会老老实实把过时信息拿出来再配上模型一本正经的语气杀伤力极大。这三个现场暴露的是同一个本质问题传统 RAG 是一个单向的、没有反馈回路的管道。查询发出去就进了黑盒检索结果没有质量评估生成答案之前没有二次确认。整个流程缺的不是更好的 Embedding 模型也不是更大的向量库而是判断和决策。2. Agentic RAG 的核心思路把检索问答改成自主决策环路2.1 从流水线到带反馈的循环先理解一个最关键的转变传统 RAG 是流水线Query 进去Answer 出来中间每一步都是定死的Agentic RAG 是环路系统会根据中间结果动态决定下一步做什么。我用一个生活化的类比传统 RAG 就像你打电话给客服不等你说完客服直接按流程念了一段 FAQ你追问这不是我要问的客服还是重复那段 FAQ。Agentic RAG 则像一个有经验的业务专员他会先听你描述问题判断这个问题是常见问题还是复杂问题去查资料如果查到的资料对不上他会换个说法再查一遍还是查不到就直接去问外部专家最后给你一个确认过的答案。这个判断—行动—评估—再行动的循环就是 Agentic RAG 与传统 RAG 的本质区别。2.2 三个自主能力规划、评估、反射具体到技术实现Agentic RAG 比传统 RAG 多了三个核心能力规划Planning在检索之前先对用户 Query 做意图识别。这个问题适合走本地知识库还是更适合搜索公开网页或者干脆就是一个可以闲聊的话题系统要能自主决定路线。评估Grading检索结果不是直接丢给大模型生成答案而是先做一个可达性判断——这批文档和用户问题到底相不相关相关度够不够支撑生成不够就拒绝生成或标记为低置信度。反射Reflection评估不通过时系统要能反思哪里出了问题。是查询的关键词偏了还是切块粒度不对反思的结果是重写查询、换一种检索方式甚至放弃本地检索、转向 Web 搜索。这三个能力不是孤立存在的它们通过一个决策循环串起来先规划再检索再评估评估不通过就反射并重试重试达到上限或评估通过才进入生成阶段。2.3 为什么要用 LangGraph 而不是 LangChain很多人会问LangChain 也能调 LLM、也能接向量库为什么非要用 LangGraph我当时的判断依据很简单LangChain 表达不了循环和条件分支。LangChain 的链式调用Chain是静态的先做什么、后做什么在定义链的时候就写死了。虽然你可以用LLMRouterChain做路由用各种 hack 的方式实现反馈但代码会变得非常绕状态共享和中断恢复更是麻烦。LangGraph 是专门为有状态、可循环、可中断的 Agent 应用设计的它把流程建模成一张图节点Node是执行单元边Edge是流转路径状态State是全局共享的数据容器。你可以在任意节点之间建立循环也可以根据状态内容决定下一步走向哪一个节点。我换句话总结如果你只是想调一次接口、拼一个 Prompt用 LangChain 足够如果你想做一个会思考的问答系统状态要在多个环节之间反复流转、决策要依赖上一步的结果、可能要循环重试那就上 LangGraph。3. LangGraph 实战前置环境、概念与最小可运行示例3.1 环境安装与模型选型我这边用的是 Python 3.10 LangGraph 最新稳定版安装命令很简单pip install langgraph langchain-openai langchain-community langchain-text-splitters faiss-cpu tavily-python模型方面我建议分三个角色来选角色用途建议模型规划/路由判断 Query 意图是简单问题还是复杂问题GPT-4o-mini / Qwen-plus / GLM-4-Flash生成/评估生成最终答案、给检索结果打分GPT-4o / DeepSeek-V3 / Qwen-MaxEmbedding本地知识库向量化text-embedding-3-small / bge-m3规划和路由对模型能力要求不高用便宜快速的小模型就行生成和评估是质量关键用更强的模型Embedding 模型里 bge-m3 是国产开源里性价比很高的选择支持中文效果好也能本地部署。3.2 核心概念速通State、Node、Edge、Conditional EdgeLangGraph 里有四个概念你必须在动手前搞清楚State状态全局共享的数据容器用TypedDict定义。每个节点都可以读取和更新 State更新的方式由Annotated指定默认是覆盖也可以自定义成追加。Node节点一个 Python 函数或可调用对象接收当前 State返回一个字典字典里的键值会被合并进 State。节点是执行单元可以调用 LLM、查向量库、执行任意 Python 逻辑。Edge边两个节点之间的连接表示上一个节点执行完后无条件进入下一个节点。Conditional Edge条件边根据 State 中的某些字段动态决定走哪条边。这是实现路由和循环的关键。还有一个特殊节点叫END表示整个流程终止。图编译之后用invoke()或stream()驱动执行。3.3 一个 30 行的最小示例先建立整体手感在接入 RAG 之前我先用一个无检索的最小示例让你感受 Graph 的执行逻辑。这个例子模拟一个简单的判断 分支流程from typing import TypedDict, Literal from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI # 1. 定义状态 class DemoState(TypedDict): question: str route: str answer: str # 2. 初始化模型 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 3. 定义节点 def analyze(state: DemoState) - dict: 判断问题是简单问题还是复杂问题 prompt f判断下面这个问题属于哪一类只回答simple或complex。 问题{state[question]} result llm.invoke(prompt) route result.content.strip().lower() return {route: route} def answer_simple(state: DemoState) - dict: return {answer: f【简单问题】使用通用知识回答{state[question]}} def answer_complex(state: DemoState) - dict: return {answer: f【复杂问题】需要检索知识库先记录问题{state[question]}} # 4. 定义条件路由 def route_decision(state: DemoState) - Literal[simple, complex]: if state[route] simple: return simple return complex # 5. 构建图 graph StateGraph(DemoState) graph.add_node(analyze, analyze) graph.add_node(answer_simple, answer_simple) graph.add_node(answer_complex, answer_complex) graph.set_entry_point(analyze) graph.add_conditional_edges(analyze, route_decision, { simple: answer_simple, complex: answer_complex, }) graph.add_edge(answer_simple, END) graph.add_edge(answer_complex, END) app graph.compile() result app.invoke({question: 中国的首都是什么}) print(result[answer])你跑一遍就会发现LangGraph 的核心心智模型就是State 在节点间流动边的选择由上一节点的输出决定。后面所有 Agentic RAG 的复杂度都是在这个最小模型上叠加出来的。4. 会思考的实现查询分析节点与条件路由4.1 为什么要先对问题做分类很多人做 RAG 时把所有问题一股脑送进向量检索从来不问这个问题是否适合用向量检索解决。我见过最典型的情况是用户问你好在吗系统也去检索一堆文档然后模型东拼西凑回一句。这是巨大的浪费也是糟糕的体验。所以在 Agentic RAG 里检索之前先加一个查询分析节点让 LLM 判断这个 Query 属于哪一类再决定后续走哪条路。我一般把问题分成四类vector需要查本地知识库的具体业务问题比如咱们公司的年假制度是什么web实时性很强、本地库无法覆盖的问题比如今天黄金价格是多少chat闲聊、问候不需要检索complex多跳或跨文档综合类问题需要走多步检索与验证4.2 节点实现让 LLM 扮演指挥官查询分析节点的代码非常直观本质上是一个带结构化输出的 LLM 调用from typing import TypedDict, Literal from pydantic import BaseModel, Field from langchain_openai import ChatOpenAI class QueryIntent(BaseModel): 查询意图分类结果 search_type: Literal[vector, web, chat, complex] Field( ..., description查询类型vector-本地知识库web-联网检索chat-闲聊complex-多跳复杂问题 ) reasoning: str Field(..., description分类原因) rewritten_query: str Field(..., description优化后的查询语句) def analyze_query_node(state: AgenticRAGState) - dict: llm ChatOpenAI(modelgpt-4o-mini, temperature0) structured_llm llm.with_structured_output(QueryIntent) system_prompt 你是一个查询分析器。根据用户的问题判断最适合的检索方式。 规则 1. 如果问题涉及企业内部知识、文档、流程、政策等选择 vector。 2. 如果问题涉及实时信息新闻、股价、天气、最新事件等选择 web。 3. 如果问题只是打招呼或闲聊选择 chat。 4. 如果问题需要结合多个文档或多个步骤才能回答选择 complex。 5. 如果你认为原问题表述不够清晰可以在 rewritten_query 中改写一个更利于检索的版本。 intent structured_llm.invoke([ (system, system_prompt), (human, f用户问题{state[question]}), ]) return { search_type: intent.search_type, reasoning: intent.reasoning, rewritten_query: intent.rewritten_query or state[question], }这里有个关键细节我用with_structured_output(QueryIntent)让 LLM 直接输出结构化 JSON而不是让模型自由发挥文本再去正则解析。这种方式解析稳定、容错率高强烈建议你在生产环境这么干。4.3 条件路由四种意图四张地图分类完之后用条件边把不同意图映射到不同节点路径def route_by_intent(state: AgenticRAGState) - Literal[vector, web, chat, complex]: return state[search_type] # 在构建图时 graph.add_conditional_edges(analyze_query, route_by_intent, { vector: retrieve, web: web_search, chat: chat_answer, complex: retrieve, })这样系统就有了最基础的思考能力。举个例子用户问帮我看看公司附近有什么好吃的这个 Query 会被分类为 web直接走联网搜索用户问咱们的报销流程是啥走 vector 查询本地知识库用户问你好走 chat 节点直接返回问候语根本不会触发检索省了向量库的调用成本。4.4 效果验证一个简单的对照实验我当时用同一组 50 条测试问题跑了对照实验。传统 RAG 对闲聊类问题也会强制走检索Agentic RAG 会拦截对XX 公司最新股价传统 RAG 从知识库捞出一堆不相关文档然后胡编Agentic RAG 会正确转去联网。分类准确率在 GPT-4o-mini 上能做到 95% 以上加上 rewritten_query 字段相当于每个问题都先经过一次整理措辞再进入检索相关性提升非常明显。5. 会纠错的实现检索质量评估、查询重写与循环机制5.1 检索结果为什么要打分这是整个 Agentic RAG 里最容易被忽略、但价值最高的一环。向量检索返回的 TopK 文档很可能看起来相似但语义不相关。比如你问公司对加班是怎么规定的检索回来的文档里可能有一篇讲考勤制度的定义、一篇讲加班费计算的法律依据这些文档在向量空间里距离确实近但能不能支撑一个准确答案需要二次确认。我在实现里加了一个grade_documents节点让 LLM 对每一篇检索到的文档做一个相关性判定输出yes或no。如果所有文档都判定为no说明这次检索失败系统进入纠错流程。5.2 纠错的第一种手段查询重写检索失败最常见的原因是查询词和文档里的措辞不一致。用户问员工离职怎么处理文档里写的是劳动关系终止流程Embedding 模型再怎么厉害跨这种同义不同形的鸿沟也经常翻车。这时候不能死磕同一句话再检索一遍而要换一个表达重新搜。查询重写节点让 LLM 根据原问题以及当前检索结果不相关这个反馈生成一个新的检索语句def rewrite_query_node(state: AgenticRAGState) - dict: llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt f你是在为知识库检索优化查询语句的专家。 用户原始问题{state[question]} 之前的查询语句是{state[rewritten_query]} 但检索结果与问题不相关请重新生成一个查询语句要求 1. 保留原始意图 2. 使用文档中更可能出现的措辞 3. 适度增加同义词扩展 只返回改写后的查询语句。 result llm.invoke(prompt) return {rewritten_query: result.content.strip()}这里有一个我踩过的坑不要把rewritten_query和question混在一起用。有些实现会把两个句子拼接后一起检索效果往往更差因为检索器会同时匹配两段信息反而稀释了核心语义。我的做法是rewritten_query始终覆盖state[rewritten_query]检索器只用改写后的版本原问题只用于生成答案时的上下文。5.3 纠错的第二种手段可控的循环上限有了重写机制就可以在 Graph 里构建一个检索—评估—不满则重写—再检索的循环。但循环必须有一个上限否则一个永远检索不到的问题会无限烧 token。我在 State 里维护一个retry_countclass AgenticRAGState(TypedDict): question: str rewritten_query: str search_type: str documents: list generation: str retry_count: int # 记录重试次数 loop_limit: int # 循环上限条件边写成这样def decide_after_grade(state: AgenticRAGState) - str: if state[retry_count] state[loop_limit]: return max_retry if any(doc[grade] yes for doc in state[documents]): return generate return rewrite graph.add_conditional_edges(grade_documents, decide_after_grade, { max_retry: web_search, # 本地库实在不行转联网搜索兜底 generate: generate_answer, rewrite: rewrite_query, })注意max_retry我直接导到了web_search这是一个很实用的兜底策略本地知识库检索两轮都失败说明这个问题大概率超出了知识库的覆盖范围与其第三次重试硬耗不如转移到 Web 搜索碰碰运气。5.4 纠错机制的整体运转效果我不建议只依赖查询重写这一招。实际生产里向量库检索失败的原因五花八门有的是 query 措辞问题有的是切块质量问题有的是 TopK 太小把相关文档挤出去了。所以我在grade_documents之外还做了两件事如果单条文档的相关性分数低于阈值但整体文档集里存在部分相关文档我会在生成阶段用 Prompt 明确告诉模型以下文档仅部分相关请标注不确定性不要强行作答。如果 TopK 文档的分数普遍偏低我会把 TopK 从 4 动态调到 8 重新检索一次而不是只改查询词。这段逻辑你在实现时可以灵活处理关键是评估结果要能驱动策略变化。加了纠错机制之后我用同一批测试集对比过传统 RAG 的正确率大概是 68%加入评估和重写循环后提升到 86%而且检索不出来硬编答案的情况大幅减少。这才是 Agentic RAG 比传统 RAG 值的核心原因——它会承认自己没找到而不是装懂。6. 会联网的实现把 Web 搜索作为 Agent 的 Tool 接入6.1 什么时候应该触发联网我在第 4 节说过了查询分析阶段就会把明显实时性的问题直接导入web_search节点。但还有另一种触发场景更考验系统设计检索评估后仍然无法回答。用户问的可能是知识库里没有沉淀过的冷门问题重写两次也找不到与其让模型硬编不如让它知难而退转向检索全网。我把这两个触发点总结成一个决策表触发时机触发条件处理方式查询分析阶段问题涉及实时信息、外部事件、个性化数据直接走 web_search检索评估阶段本地知识库经过两轮重试仍无法命中转向 web_search 兜底6.2 用 Tavily 实现联网搜索Web 搜索这块我选的是 Tavily专为大模型场景设计的搜索 API返回的是干净的正文摘要不用自己解析 HTML也能控制搜索结果的 token 量。关键代码from langchain_community.tools import TavilySearchResults def web_search_node(state: AgenticRAGState) - dict: 联网搜索用 Tavily 搜索并提取相关段落 tavily_tool TavilySearchResults( max_results5, search_depthadvanced, include_answerTrue, ) results tavily_tool.invoke({query: state[rewritten_query]}) docs [] for item in results: if isinstance(item, dict) and content in item: docs.append({ content: item[content], source: item.get(url, web), }) elif hasattr(item, content): docs.append({ content: item.content, source: getattr(item, url, web), }) return { documents: docs, search_type: web, }Tavily 的include_answerTrue会直接返回一个针对查询的摘要答案有些场景下这个摘要已经可以直接用连生成节点都能省。但为了答案质量我还是会把它作为一个候选来源放进documents交给生成节点综合判断。6.3 把 RAG 文档和 Web 文档混在一起Prompt 要区分来源一个容易踩坑的地方web_search返回的文档和向量库返回的文档结构不同、可信度也不同。如果你把它们一股脑塞进同一个documents列表生成节点根本分不清哪个是内部知识库的、哪个是互联网的——万一联网搜索到的结果和内部制度冲突模型可能把外网信息当成公司政策回答出来后果很严重。我给每篇文档加了一个source_type字段生成阶段在 Prompt 里明确标注来源权重def generate_answer_node(state: AgenticRAGState) - dict: llm ChatOpenAI(modelgpt-4o, temperature0.3) context_parts [] for doc in state[documents]: source_tag 【内部知识库】 if doc.get(source_type) internal else 【公开网络】 context_parts.append(f{source_tag}\n来源{doc.get(source, unknown)}\n内容{doc[content]}) context \n\n.join(context_parts) prompt f你是企业知识助手请基于以下上下文回答用户问题。 上下文 {context} 用户问题{state[question]} 要求 1. 优先采用【内部知识库】的信息回答。 2. 当【内部知识库】信息不足时可以引用【公开网络】信息但必须在回答末尾标注以上内容部分来自公开网络仅供参考。 3. 如果两类来源信息冲突以【内部知识库】为准并提示用户存在外部信息不一致。 4. 不要强行作答如果信息不足请直接说知识库中未找到相关信息。 response llm.invoke(prompt) return {generation: response.content}这个来源敏感的设计我强烈建议你保留。它不仅是技术问题也是内容合规风险问题——把外网信息伪装成企业内部制度在业务上是不可接受的。7. 端到端整合完整 Agentic RAG 代码与效果验证7.1 先看整体状态定义与图结构前面几节我把各个节点的实现拆开讲了这一节把它们拼成一个完整可运行的系统。先定义一个覆盖全流程的状态from typing import TypedDict, Literal, Annotated, List import operator from langgraph.graph import StateGraph, END from langgraph.checkpoint import MemorySaver class AgenticRAGState(TypedDict): question: str rewritten_query: str search_type: Literal[vector, web, chat, complex] documents: Annotated[List[dict], operator.add] generation: str retry_count: int loop_limit: int注意documents字段我用了Annotated[List[dict], operator.add]这意味着每次节点返回的新文档会追加到已有列表而不是覆盖。如果你不希望历史检索结果残留记得在重试前清空——我一般在rewrite_query节点里返回{documents: []}之外还会在构建图时给retrieve节点加一个清空逻辑。7.2 完整的图构建代码from langchain_openai import ChatOpenAI from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings # 假设你已经用知识库构建好了 vectorstore vectorstore FAISS.load_local( knowledge_base_index, OpenAIEmbeddings(modeltext-embedding-3-small), allow_dangerous_deserializationTrue, ) retriever vectorstore.as_retriever(search_kwargs{k: 4}) # ---------- 节点 1: 查询分析 ---------- def analyze_query_node(state: AgenticRAGState) - dict: # 完整实现见第 4 节 pass # ---------- 节点 2: 本地检索 ---------- def retrieve_node(state: AgenticRAGState) - dict: query state[rewritten_query] or state[question] docs retriever.invoke(query) return { documents: [{ content: doc.page_content, source: doc.metadata.get(source, local), source_type: internal, grade: None, } for doc in docs] } # ---------- 节点 3: 检索评估 ---------- def grade_documents_node(state: AgenticRAGState) - dict: llm ChatOpenAI(modelgpt-4o-mini, temperature0) graded_docs [] for doc in state[documents]: prompt f判断下面这篇文档是否与问题相关只回答 yes 或 no。 问题{state[question]} 文档内容{doc[content][:500]} result llm.invoke(prompt).content.strip().lower() graded_docs.append({**doc, grade: result}) # 判断是否存在相关文档 has_relevant any(d[grade] yes for d in graded_docs) return { documents: graded_docs, retry_count: state[retry_count] 1, } # ---------- 节点 4: 查询重写 ---------- def rewrite_query_node(state: AgenticRAGState) - dict: # 完整实现见第 5.2 节同时清空 documents pass # ---------- 节点 5: Web 搜索 ---------- def web_search_node(state: AgenticRAGState) - dict: # 完整实现见第 6 节 pass # ---------- 节点 6: 生成答案 ---------- def generate_answer_node(state: AgenticRAGState) - dict: # 完整实现见第 6.3 节 pass # ---------- 节点 7: 闲聊 ---------- def chat_answer_node(state: AgenticRAGState) - dict: return { generation: 你好我是企业知识助手可以问我关于公司制度、项目文档、产品使用等方面的问题。 } # ---------- 条件边 ---------- def route_by_intent(state: AgenticRAGState) - str: return state[search_type] def decide_after_grade(state: AgenticRAGState) - str: if state[retry_count] state[loop_limit]: return max_retry if any(d.get(grade) yes for d in state[documents]): return generate return rewrite # ---------- 构建图 ---------- graph StateGraph(AgenticRAGState) graph.add_node(analyze_query, analyze_query_node) graph.add_node(retrieve, retrieve_node) graph.add_node(grade_documents, grade_documents_node) graph.add_node(rewrite_query, rewrite_query_node) graph.add_node(web_search, web_search_node) graph.add_node(generate_answer, generate_answer_node) graph.add_node(chat_answer, chat_answer_node) graph.set_entry_point(analyze_query) graph.add_conditional_edges(analyze_query, route_by_intent, { vector: retrieve, web: web_search, chat: chat_answer, complex: retrieve, }) graph.add_edge(retrieve, grade_documents) graph.add_conditional_edges(grade_documents, decide_after_grade, { max_retry: web_search, generate: generate_answer, rewrite: rewrite_query, }) graph.add_edge(rewrite_query, retrieve) graph.add_edge(web_search, generate_answer) graph.add_edge(generate_answer, END) graph.add_edge(chat_answer, END) # 用 MemorySaver 启用状态记忆后续可以支持多轮对话 app graph.compile(checkpointerMemorySaver()) # ---------- 调用 ---------- result app.invoke( {question: 公司年假制度是怎么规定的, retry_count: 0, loop_limit: 2}, config{configurable: {thread_id: test-001}}, ) print(result[generation])7.3 完整状态流转逻辑整个系统跑起来后的决策路径是这样的analyze_query先把问题分类。普通政策问题走vector实时性问题直接走web_search闲聊走chat_answer。走vector路径时先retrieve从 FAISS 取回 TopK然后进入grade_documents打分。打分结果有相关文档进入generate_answer生成答案。打分全不相关且retry_count loop_limit进入rewrite_query改写查询词改写完成后重新retrieve。重试到上限仍然失败转向web_search兜底把网络结果交给生成节点。最终答案统一通过generate_answer输出来源经过 Prompt 层面的约束。这套流程跑下来一次问答最多可能经历本地检索 两轮重写 联网兜底四次信息获取虽然比传统 RAG 慢一些但答案的可靠性和覆盖面完全不在一个量级。7.4 多轮对话怎么扩展热搜词里很多人问RAG 多轮对话怎么设计我在这个图里已经埋了伏笔——MemorySaver。LangGraph 的 checkpointer 会把每次调用后的 State 保存下来你下一次 invoke 时带上同一个thread_id就能在节点里读取历史 State。最简单的做法是在analyze_query_node里把历史对话拼接成上下文让 LLM 判断当前问题是否依赖前文。比如用户先问咱们的年假制度是啥接着问那哺乳假呢第二句的完整语义是咱们的哺乳假制度是啥。我在查询分析阶段就把历史消息拼进去让 LLM 补全当前问题后再做分类和改写。def _build_conversation_context(state): # 从 checkpointer 或外部存储读取历史消息 # 这里简化处理实际可以从 MemorySaver 的 State 快照里恢复 return \n.join(history_messages[-4:]) # 保留最近两轮更精细的多轮处理还可以把rewritten_query改成补全后的完整问题而不是只改措辞。这个方向你可以继续深入这里先给一个思路。8. 踩坑记录与生产化建议8.1 State 更新的坑覆盖和追加别搞混LangGraph 默认每个节点返回的字典键值会覆盖 State 里的旧值。我在早期调试时犯过一个错retrieve节点返回的documents覆盖了上一轮检索结果但grade_documents节点里我只给最新一份文档打了分之前的文档全丢了。后来把documents设计成operator.add追加模式又遇到新问题——重写查询后重新检索老文档还留在列表里生成节点看到一堆积压的旧上下文。最终解决方案很土但有效在rewrite_query节点里主动清空documents每次重新检索前保证状态是干净的。文档字段建议加一个round_id标记检索轮次这样既能追查问题又能在生成时灵活过滤。8.2 条件边的键必须和映射表严格一致LangGraph 的条件边有个隐形要求条件函数返回的字符串必须是映射字典的键之一否则运行时直接报错。我在第一个版本里写过return retry映射表里键是rewrite结果一跑就崩。建议你在写条件路由函数时用字面量类型Literal[max_retry, generate, rewrite]做函数返回注解这样 IDE 能帮你提前检查比靠记忆力强得多。8.3 token 成本控制评估节点也别用大模型前面我的示例里grade_documents用的是gpt-4o-mini不是gpt-4o这是有意为之。一个查询要评估 4 到 8 篇文档每篇都是一次 LLM 调用如果用顶级模型成本直接翻好几倍。实测下来gpt-4o-mini做相关性判断的准确率在 90% 以上完全够用。但这里有一个前提给评估模型的 Prompt 要写清楚宁可错杀不可漏判还是宁可放过不可误报。我发现对文档是否与问题相关的判断评估模型容易出现偏差。后来在 Prompt 里加了一句如果文档包含问题的部分答案或关键线索也算相关准确率提升明显。8.4 并发与流式输出从 CLI 到 Web 服务示例代码里用的是同步invoke放到 Web 服务里会有性能问题。LangGraph 的图和节点函数支持异步实现实际部署时建议用async版本的节点函数配合ainvoke/astream。如果有多个检索策略可以并行比如同时查本地库和 Web用 LangGraph 的SendAPI 或parallel节点让它们并发执行整体延迟能降一半。前端展示用astream流式输出generation字段用户不用等全部生成完才看到文字。8.5 Human-in-the-loop把最后确认权交给人工LangGraph 有一个interrupt_before参数可以在执行到指定节点前暂停图等人工确认后再恢复。这在高风险场景里很有用。比如专利侵权风险分析这种问题即使 RAG 检索到了相关资料我也不希望系统自动给出结论而是先把检索到的证据列出来让业务专家确认后再生成最终答复。app graph.compile( checkpointerMemorySaver(), interrupt_before[generate_answer], # 生成前暂停 )这种机器检索、人来做主的模式特别适合合规审查、合同分析、医疗建议等场景。LangGraph 把这个功能做成了配置项而不是 hack这也是我推荐它的重要原因。8.6 和 MCP 的关系一句话说清最近很多人问RAG 和 MCP 有什么区别。我的理解是MCP 是模型与外部工具之间的标准通信协议解决的是模型怎么调用外部资源的问题Agentic RAG 是检索增强生成的应用架构解决的是怎么自主决定检索策略的问题。两者不在一个层次实际项目里完全可以结合——用 MCP 让 Agent 统一接入企业内部的数据服务和 Web 搜索工具。在我的体验里Agentic RAG 最值钱的地方不是多了一个联网功能而是整个系统从被动执行变成了主动判断。它知道自己在问什么、查到的资料够不够用、查不到的时候该怎么办。这种知进退的能力恰恰是传统 RAG 最大的短板。最后分享一个我自己养成的习惯不要一上来就上 LangGraph先用传统 RAG 把基线跑通用一批真实问题的失败案例来驱动设计。你会发现需要加评估节点还是重写节点不是拍脑袋决定的而是被那些失败案例逼出来的。等基线暴露了问题再上 Agentic RAG你的图每一步都是对症下药而不是为了炫技堆功能。
