1. 为什么普通 RAG 在真实问答场景里总是“差一口气”做过知识库问答的人大概都有过这种体验demo 阶段效果惊艳一旦丢进真实业务里用户问三个问题就有两个答非所问。你明明把文档切好了、向量库也建了、检索 top-k 也调了可模型给出的答案要么答偏要么把不相关的段落硬凑在一起要么干脆自信地编一段。问题往往不在向量检索本身而在于传统 RAG 是一条单向流水线检索一次拼上下文生成结束。它没有“回头看”的能力。用户问“我们产品的退款政策在海外和国内有什么区别”一次检索可能只召回了国内政策模型就顺着国内政策答完了海外部分直接漏掉。它不会意识到“我好像缺了一半信息”更不会主动发起第二轮检索去补齐。这就是Agentic RAG要解决的核心痛点。它把 RAG 从“一次性管道”升级成“带决策循环的智能体”检索之后先评估“这些资料够不够、相不相关”不够就改写查询再检索够了再生成生成完还能自我检查答案是否被资料支撑。而LangGraph恰好是承载这种循环结构的趁手工具——它用StateGraph把整个流程显式建模成节点和边让“反思”“重试”“分支”这些控制流变得可读、可调试、可持久化。这篇内容适合两类人一是已经跑通过基础 RAG、但被“答不准”折磨的开发者二是想上手 LangGraph、却卡在“知道概念但不知道怎么落地一个真实 Agent”的工程师。我会从状态设计讲到反思循环再到并发和踩坑尽量把每一步的“为什么”说清楚代码可以直接抄改。2. 用 StateGraph 把问答流程拆成可观测的节点2.1 为什么不用 Chain 而用 Graph很多人第一反应是用 LangChain 的SequentialChain或者 LCEL 把流程串起来。串行链的问题在于它只能表达“A 然后 B 然后 C”一旦你要表达“如果检索质量差就回到检索节点重来”“如果答案没被支撑就重新生成”链式结构就开始别扭了——你得在链里塞条件判断逻辑很快变成一团意大利面。LangGraph 的思路完全不同。它把整个 Agent 建模成一张有向图核心三要素是State状态一个贯穿全流程的共享数据结构所有节点读写它。Node节点一个函数接收当前 State返回对 State 的更新。Edge边定义节点之间的流转可以是固定的也可以是条件分支。这样“反思后重试”就变成了一个非常自然的表达从评估节点拉一条条件边判断不通过就指回检索节点。图结构让控制流一目了然也天然支持循环。2.2 状态字段的设计决定了 Agent 的上限State 是整个 Agent 的“记忆总线”字段设计得好不好直接决定后面写起来顺不顺。我踩过的坑是一开始只放了一个question和一个answer结果做反思时发现根本没法判断“检索够不够”因为没地方存检索结果和评估结论。一个能支撑反思循环的 State 至少要有这些字段from typing import TypedDict, List class RAGState(TypedDict): question: str # 原始问题 rewritten_query: str # 改写后的检索查询 documents: List[str] # 当前轮检索到的文档 relevance: str # 检索相关性评估结论 answer: str # 生成的答案 retry_count: int # 已重试次数防止死循环 grounded: str # 答案是否被资料支撑的评估结论这里有几个设计要点值得展开。retry_count是必须的反思循环最大的风险就是无限重试——评估节点永远觉得“还不够好”Agent 就卡死了。给它设一个上限比如 2 次到顶就强制走生成。rewritten_query和question分开存是因为改写后的查询往往和原问题差别很大保留原问题方便最终生成时对齐用户意图。提示State 用TypedDict定义只是最简形式。如果你需要多个节点并发写同一个字段比如并行检索多个数据源要用带 reducer 的Annotated否则后写的会覆盖先写的。这个坑我在第 5 节会详细讲。2.3 节点职责的单一化拆分图建得好不好看节点拆得细不细。我的经验是一个节点只干一件事且这件事要能被单独测试。基于这个原则一个会反思的 RAG Agent 通常拆成五个节点rewrite 节点把用户口语化的问题改写成更适合检索的查询。retrieve 节点拿查询去向量库捞文档。grade 节点评估检索到的文档和问题相关不相关。generate 节点基于文档生成答案。check 节点检查答案是否被文档支撑决定要不要重来。拆这么细的好处是调试时能精确定位。有次我的 Agent 答非所问我把每个节点的输入输出打出来一看发现是 rewrite 节点把“退款要几天”改写成了“退款流程说明”语义偏了检索自然跟着偏。如果全塞在一个大节点里这种问题很难定位。3. 反思循环的两个关键判断点检索够不够、答案稳不稳3.1 检索后评估让 Agent 学会说“这些资料不行”反思循环的第一个判断点是在检索之后、生成之前。传统 RAG 拿到 top-k 就直接喂给模型Agentic RAG 则要先问一句这些文档真的能回答问题吗实现上grade 节点通常用一个 LLM 做二分类或三分类判断。我习惯用三档relevant直接相关、partial部分相关、irrelevant不相关。二分类太粗暴很多真实场景是“沾边但不全”三档能让后续决策更细腻。def grade_documents(state: RAGState): question state[question] docs state[documents] prompt f判断以下文档与问题的相关程度。 问题{question} 文档{docs} 只回答 relevant / partial / irrelevant 三者之一。 result llm.invoke(prompt).content.strip().lower() return {relevance: result}这里有个实操细节评估用的 LLM 最好和生成用的分开配置。评估是个相对简单的判断任务用便宜快的小模型就够生成需要质量用强模型。我一开始图省事全用同一个大模型结果每次反思都烧掉不少 token成本翻倍。分开之后成本降了大概六成效果几乎没差。3.2 条件边把评估结论翻译成流转决策评估出结论之后得靠条件边决定下一步往哪走。这是 LangGraph 最核心的机制之一def decide_after_grade(state: RAGState): if state[relevance] relevant: return generate if state[retry_count] 2: return generate # 重试到顶硬着头皮生成 return rewrite # 回去改写查询再检索 graph.add_conditional_edges( grade, decide_after_grade, {generate: generate, rewrite: rewrite} )这段逻辑里藏着两个经验。第一retry_count 2的兜底分支不能省否则评估节点一旦“较真”整个图就转不出来了。第二重试时回到的是rewrite 节点而不是 retrieve 节点因为如果查询不变检索结果也不会变重试毫无意义。必须改写查询才可能捞到新东西。3.3 生成后校验防止模型“自由发挥”反思循环的第二个判断点在生成之后。模型有时候会无视检索到的资料凭自己的先验知识编答案——这在知识库场景里是致命的因为用户要的是“我们文档里怎么写的”不是“模型觉得应该怎样”。check 节点就是干这个的把生成的答案和检索到的文档一起丢给 LLM问它“这个答案是否完全被文档支撑”。如果答案是“否”就触发重新生成或者在 prompt 里加强“只能依据资料回答”的约束。def check_grounding(state: RAGState): prompt f答案是否完全由以下资料支撑只回答 yes 或 no。 资料{state[documents]} 答案{state[answer]} result llm.invoke(prompt).content.strip().lower() return {grounded: result}注意grounding 检查会增加一次 LLM 调用延迟和成本都会上升。如果你的场景对实时性要求极高可以把它做成“抽样检查”或者只在答案较长时触发不必每轮都跑。3.4 把两个判断点串成完整循环把上面这些拼起来整张图的流转逻辑是这样的rewrite → retrieve → grade →相关则 generate不相关则回 rewrite→ check →被支撑则结束不被支撑则回 generate。两个循环嵌套但各自有重试上限兜底。这种“双反思”结构比单反思更稳。我实测过只做检索反思的版本发现它虽然能补齐信息但偶尔还是会让模型夹带私货加上生成后校验之后答案的“忠实度”明显提升用户反馈“答得更像我们自己的文档了”。4. 从零把这张图跑起来环境、依赖与最小可运行版本4.1 环境准备里最容易忽略的两件事装依赖本身没什么好说的pip install langgraph langchain langchain-openai基本够用。但有两件事新手特别容易忽略。第一是向量库的选型要和数据量匹配。几百到几万条文档用 FAISS 本地跑完全够零运维上到几十万条再考虑 Milvus、Qdrant 这类专门的向量数据库。我见过有人一上来就搭分布式向量库结果数据才两千条纯属给自己找麻烦。第二是embedding 模型要和检索语言对齐。中文知识库用英文为主的 embedding 模型召回质量会明显打折。选型时先拿几十条真实问题做个小测试比看任何 benchmark 都靠谱。4.2 图的组装与编译LangGraph 的图组装分三步建图、加节点、加边最后compile()。from langgraph.graph import StateGraph, START, END builder StateGraph(RAGState) builder.add_node(rewrite, rewrite_query) builder.add_node(retrieve, retrieve_docs) builder.add_node(grade, grade_documents) builder.add_node(generate, generate_answer) builder.add_node(check, check_grounding) builder.add_edge(START, rewrite) builder.add_edge(rewrite, retrieve) builder.add_edge(retrieve, grade) builder.add_conditional_edges(grade, decide_after_grade, {generate: generate, rewrite: rewrite}) builder.add_edge(generate, check) builder.add_conditional_edges(check, decide_after_check, {end: END, regenerate: generate}) app builder.compile()compile()这一步会做图的结构校验比如有没有孤立节点、条件边的目标是否存在。我第一次编译报错就是因为条件边里写了个不存在的节点名LangGraph 直接拦下来了比运行时才崩要好得多。4.3 跑通第一个问题并观察状态流转编译完之后用app.invoke({question: ...})就能跑。但真正有价值的不是看最终答案而是看中间状态怎么流转的。LangGraph 支持流式输出每个节点的更新for event in app.stream({question: 海外退款要几天, retry_count: 0}): print(event)这样你能清楚看到rewrite 把问题改成了什么、retrieve 捞回了哪些文档、grade 判成了什么、有没有触发重试。我强烈建议在开发阶段一直开着这个流式观察它比任何日志都直观。有次我发现 Agent 连续重试了两次一看流式输出原来是 grade 节点把明显相关的文档误判成了 partial调了下评估 prompt 就好了。5. 并发、持久化与那些文档里不会写的坑5.1 多路检索并发时的状态覆盖问题真实知识库往往不止一个数据源——可能有产品文档库、FAQ 库、工单库。一个自然的优化是并行检索多个源然后合并结果。LangGraph 支持从多个节点同时指向一个节点但这里有个大坑如果多个节点同时写 State 的同一个字段默认行为是后写的覆盖先写的。解决办法是给字段加 reducerfrom typing import Annotated import operator class RAGState(TypedDict): documents: Annotated[List[str], operator.add]operator.add表示这个字段的更新是“追加”而不是“覆盖”。这样多个检索节点各自返回的文档列表会被合并到一起。我第一次做并行检索时没加 reducer结果三个库只返回了最后一个库的结果排查了半天才反应过来是状态覆盖。5.2 用 checkpointer 让对话能“记住上一轮”单轮问答跑通之后下一步通常是多轮对话。用户会追问“那国内呢”这时候 Agent 得知道上一轮聊的是退款。LangGraph 的checkpointer就是干这个的它把每一轮的状态按thread_id存下来下一轮自动带上历史。from langgraph.checkpoint.memory import MemorySaver app builder.compile(checkpointerMemorySaver()) config {configurable: {thread_id: user-123}} app.invoke({question: 海外退款要几天}, config) app.invoke({question: 那国内呢}, config) # 自动带上上一轮上下文开发阶段用MemorySaver就够上线要换成数据库版的 checkpointer比如 Postgres否则进程一重启历史就没了。这里要注意多轮场景下 State 里的字段要区分“本轮”和“历史”否则上一轮的documents会污染这一轮的检索评估。5.3 反思循环的成本与延迟控制反思很美好但每一次反思都是一次 LLM 调用都是真金白银和真实延迟。我做过一个粗略统计一个带双反思的 Agent平均每个问题要跑 4 到 6 次 LLM 调用是普通 RAG 的两三倍。如果每个问题都要等五六秒用户体验会很差。几个实用的优化手段优化手段做法适用场景分级模型评估用小模型生成用大模型几乎所有场景条件触发只在检索分数低于阈值时才跑 grade检索质量本身较稳时缓存相同查询的检索结果缓存高频重复问题并行多源检索并行执行多数据源场景我个人最推荐的是分级模型 条件触发的组合。检索分数比如向量相似度本身就能反映一部分相关性分数很高时直接跳过 grade 节点能省下不少调用。5.4 几个我踩过的具体坑第一个坑是评估 prompt 的措辞会显著影响判断。我一开始写“判断文档是否相关”模型倾向于宽松改成“判断文档是否足以回答该问题”判断就严格多了。措辞要贴合你真正想要的语义。第二个坑是重试时如果改写查询失败会陷入无效循环。rewrite 节点偶尔会改出一个更差的查询导致检索更烂、评估更差、又重试。我的做法是在 rewrite 的 prompt 里明确要求“保留原问题的核心实体”避免改写跑偏。第三个坑是grounding 检查对长答案容易误判。答案一长模型判断“是否被支撑”时容易漏看细节。后来我把长答案先做一次要点抽取再逐点校验准确率提升明显。6. 从能跑到好用评估、观测与迭代思路6.1 怎么判断你的 Agent 到底有没有变好“感觉答得不错”不是评估。要迭代得先有可量化的指标。对 RAG Agent 来说我通常盯三个维度检索命中率正确文档有没有被捞进 top-k。答案忠实度答案有没有被检索资料支撑可以用 grounding 检查的通过率近似。端到端正确率人工标注一批问题看最终答案对不对。前两个可以自动化跑第三个需要人工但样本不用多几十条就能看出趋势。我习惯每次改动后跑一遍固定测试集对比这三个指标避免“改了一个地方、坏了另一个地方”。6.2 用 LangSmith 之类的工具看链路LangGraph 的图结构天然适合做链路追踪。每个节点的输入输出、耗时、token 消耗都能被记录下来。接上 LangSmith 之后你能看到一次问答里每个节点花了多久、哪一步是瓶颈。我优化延迟时就是靠它发现 grade 节点意外地慢——原来是我给它配了个大模型换成小模型后整体延迟降了将近一半。即使不接外部工具自己写个简单的回调把每个节点的耗时打出来也比两眼一抹黑强。6.3 迭代的优先级先修检索再调生成新手容易一上来就调生成 prompt觉得答案不好是模型的问题。但根据我的经验八成的问题出在检索。检索没捞对资料生成再强也是巧妇难为无米之炊。所以迭代顺序应该是先看检索命中率命中率不行就调切分策略、embedding 模型、top-k命中率上来了再调生成和反思的 prompt。反思循环本身也是可以迭代的。一开始可以只做检索反思跑稳了再加生成后校验。一次加太多机制出了问题都不知道是哪个环节的锅。6.4 一个关于“度”的经验最后分享一个我反复验证过的体会反思不是越多越好。我试过让 Agent 反思三轮结果发现第三轮基本没有增量收益反而因为过度改写查询把原本能答对的问题改坏了。两轮通常是个甜点超过两轮收益递减、风险递增。Agent 的“聪明”不在于想得多而在于知道什么时候该停。给循环设一个合理的上限比无限追求“更完美”要务实得多。