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 的同一个字段，默认行为是后写的覆盖先写的。
解决办法是给字段加 reducer：
from 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(checkpointer=MemorySaver()) 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 的“聪明”不在于想得多，而在于知道什么时候该停。给循环设一个合理的上限，比无限追求“更完美”要务实得多。
