1. 为什么线性 RAG 会「翻车」

先说结论：线性 RAG 的问题不在于检索不到，而在于它没有「判断」和「回头」的能力。

一个最朴素的 RAG 流水线长这样：

用户问题 → 向量化 → Top-K 检索 → 拼进 Prompt → LLM 生成 → 返回

这条链路是单向的。它默认了三件事：

1. 用户的问题一定需要检索（其实「你好」「谢谢」根本不用查库）；
2. 检索回来的 Top-K 一定和问题相关（其实经常是噪声）；
3. 一次检索就够（其实很多问题需要拆解、改写、多轮查）。

只要其中任何一条不成立，整条链路就会「一本正经地胡说八道」。我在实际项目里踩过的典型翻车场景：

场景	线性 RAG 的表现	根因
用户问「你好，你是谁」	去向量库捞了一堆无关文档，回答变得奇怪	没有「是否需要检索」的判断
用户问「上季度华东区退货率同比变化」	检索到一堆「退货政策」文档，答非所问	问题需要拆解 + 多跳检索
向量库召回的全是低分噪声	直接把噪声塞进 Prompt，模型顺着编	没有「相关性打分」和「重写查询」
知识库里根本没有这个信息	模型硬编一个答案	没有「兜底」出口
答案生成了但和检索内容矛盾	直接返回幻觉答案	没有「生成后校验」

所以我们需要的是一个闭环：能判断要不要检索、能评估检索质量、能改写查询重试、能在实在查不到时老实兜底。这正是 Agentic RAG 要解决的问题。

2. Agentic RAG 的核心思想：把 RAG 变成一张状态图

Agentic RAG 的本质，是把上面那条「直线」改造成一张带条件分支和回边（loop）的状态图：

这张图里有四个关键「智能节点」：

1. 路由判断（Router）：这个问题要不要查库？闲聊、常识、纯计算直接走生成。
2. 检索评估（Grader）：召回文档和问题相关吗？不相关就改写查询重试。
3. 幻觉检测（Hallucination Check）：答案是否被检索内容支撑？不支撑就回炉。
4. 兜底（Fallback）：重试到上限还不行，就老实说「我没找到可靠信息」，而不是编。

为什么用 LangGraph 而不是自己写 while 循环？

• 状态（State）显式声明，节点之间传参清晰，不用靠一堆全局变量；
• 条件边（conditional edge）天然表达「判断 → 分支」，图结构一眼看懂；
• 支持 checkpointer 做持久化、断点续跑、人工介入（human-in-the-loop）；
• 支持流式输出每个节点的中间过程，方便调试和前端展示「AI 正在思考」。

3. 环境准备

本文基于以下版本，建议锁死版本避免踩坑：

pip install "langgraph>=0.2.60" "langchain>=0.3.0" "langchain-openai>=0.2.0" \
            "langchain-community>=0.3.0" "chromadb>=0.5.0" "pydantic>=2.0"

# requirements.txt 参考
langgraph==0.2.60
langchain==0.3.13
langchain-openai==0.2.14
langchain-community==0.3.13
chromadb==0.5.23
pydantic==2.10.3

配置环境变量（这里用 OpenAI 兼容接口，换成 DeepSeek / 通义 / 本地 vLLM 只需改 base_url）：

import os

os.environ["OPENAI_API_KEY"] = "sk-xxxxxxxx"
os.environ["OPENAI_BASE_URL"] = "https://api.openai.com/v1"

4. 第一步：准备一个可检索的知识库

为了让例子能跑起来，我们造一个小型知识库（真实项目里换成你的文档加载 + 切分流程即可）。

from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 1. 模拟知识库文档
docs = [
    Document(page_content="LangGraph 是 LangChain 团队推出的用于构建有状态、多参与者应用的框架，核心是状态图（StateGraph）。", metadata={"source": "langgraph_intro"}),
    Document(page_content="LangGraph 的节点是普通 Python 函数，接收 State 返回 State 的部分更新。", metadata={"source": "langgraph_node"}),
    Document(page_content="LangGraph 的条件边通过 add_conditional_edges 添加，根据路由函数的返回值决定下一个节点。", metadata={"source": "langgraph_edge"}),
    Document(page_content="LangGraph 支持 checkpointer 持久化状态，可实现断点续跑和人工介入。", metadata={"source": "langgraph_checkpoint"}),
    Document(page_content="Agentic RAG 指在 RAG 流程中引入 Agent 的决策能力，包括查询改写、相关性评估、多轮检索和自我纠错。", metadata={"source": "agentic_rag"}),
    Document(page_content="RAG 的常见评估指标包括上下文相关性（Context Relevance）、答案忠实度（Faithfulness）和答案相关性（Answer Relevance）。", metadata={"source": "rag_eval"}),
]

# 2. 向量化并入库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=embeddings,
    collection_name="agentic_rag_demo",
    persist_directory="./chroma_db",
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 3. 快速验证
hits = retriever.invoke("LangGraph 的条件边怎么用")
for d in hits:
    print(d.metadata["source"], "|", d.page_content[:40])

5. 第二步：定义全局状态（State）

State 是整个图的「共享内存」，所有节点读写它。用 TypedDict 声明，配合 Annotated 做增量更新。

from typing import Annotated, Literal, TypedDict
from operator import add


class RAGState(TypedDict):
    # 输入
    question: str                      # 用户原始问题
    # 中间产物
    rewritten_question: str            # 改写后的查询
    documents: list[Document]          # 检索到的文档
    generation: str                    # 生成的答案
    # 控制字段
    need_retrieval: bool               # 是否需要检索
    docs_relevant: bool                # 文档是否相关
    is_hallucination: bool             # 答案是否有幻觉
    retry_count: int                   # 重试次数
    # 轨迹（用 add 做列表累加，方便调试）
    trace: Annotated[list[str], add]

几个设计要点：

• trace 用 Annotated[list[str], add]，每个节点往里 append 一条，最后能完整看到「AI 的思考路径」；
• retry_count 是全局共享的计数器，防止死循环；
• 所有判断结果都落成布尔字段，方便条件边读取。

6. 第三步：初始化 LLM 与结构化输出

判断类节点（路由、评估、幻觉检测）最好用结构化输出，避免解析自然语言的脆弱性。

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)


# --- 路由判断 ---
class RouteDecision(BaseModel):
    need_retrieval: bool = Field(description="该问题是否需要检索外部知识库")


# --- 文档相关性评估 ---
class RelevanceScore(BaseModel):
    relevant: bool = Field(description="检索到的文档是否与问题相关")
    reason: str = Field(description="判断理由，一句话")


# --- 幻觉检测 ---
class HallucinationCheck(BaseModel):
    is_hallucination: bool = Field(description="答案是否包含检索文档不支持的内容")
    reason: str = Field(description="判断理由，一句话")


router_llm = llm.with_structured_output(RouteDecision)
grader_llm = llm.with_structured_output(RelevanceScore)
hallucination_llm = llm.with_structured_output(HallucinationCheck)

7. 第四步：逐个实现节点

7.1 路由节点：要不要检索？

def route_question(state: RAGState) -> dict:
    """判断问题是否需要检索知识库"""
    question = state["question"]
    decision = router_llm.invoke(
        f"""你是一个问题分类器。判断下面的问题是否需要查询外部知识库才能回答。

需要检索的情况：涉及具体事实、文档内容、专业术语、内部资料。
不需要检索的情况：打招呼、闲聊、纯逻辑推理、通用常识、数学计算。

问题：{question}"""
    )
    return {
        "need_retrieval": decision.need_retrieval,
        "trace": [f"[route] need_retrieval={decision.need_retrieval}"],
    }

7.2 查询改写节点：把问题变得更「可检索」

def rewrite_query(state: RAGState) -> dict:
    """改写查询，提升召回质量；首次进入时基于原问题，重试时基于反馈"""
    question = state["question"]
    retry = state.get("retry_count", 0)

    if retry == 0:
        prompt = f"把下面的问题改写成更适合向量检索的查询，只输出改写后的查询：\n{question}"
    else:
        prompt = (
            f"原问题：{question}\n"
            f"上一次检索效果不好。请换一个角度改写查询，使用更具体的关键词，只输出改写后的查询。"
        )

    rewritten = llm.invoke(prompt).content.strip()
    return {
        "rewritten_question": rewritten,
        "trace": [f"[rewrite] retry={retry} query={rewritten}"],
    }

7.3 检索节点：查库

def retrieve(state: RAGState) -> dict:
    """执行向量检索"""
    query = state.get("rewritten_question") or state["question"]
    docs = retriever.invoke(query)
    return {
        "documents": docs,
        "trace": [f"[retrieve] query={query} hits={len(docs)}"],
    }

7.4 相关性评估节点：召回的东西靠谱吗？

def grade_documents(state: RAGState) -> dict:
    """评估检索文档与问题的相关性"""
    question = state["question"]
    docs = state["documents"]

    if not docs:
        return {
            "docs_relevant": False,
            "trace": ["[grade] no documents"],
        }

    # 把文档拼成一段给 LLM 打分（生产环境可逐条打分 + 阈值过滤）
    context = "\n\n".join(
        f"[{i}] {d.page_content}" for i, d in enumerate(docs)
    )
    result = grader_llm.invoke(
        f"""问题：{question}

检索到的文档：
{context}

请判断这些文档是否包含回答该问题所需的信息。"""
    )
    return {
        "docs_relevant": result.relevant,
        "trace": [f"[grade] relevant={result.relevant} reason={result.reason}"],
    }

7.5 生成节点：基于文档作答

def generate(state: RAGState) -> dict:
    """基于检索文档生成答案"""
    question = state["question"]
    docs = state.get("documents", [])

    if docs:
        context = "\n\n".join(d.page_content for d in docs)
        prompt = f"""请严格基于以下资料回答问题，不要编造资料中没有的信息。
如果资料不足以回答，请直接说「根据现有资料无法回答」。

资料：
{context}

问题：{question}
答案："""
    else:
        # 无需检索的路径（闲聊/常识）
        prompt = f"请直接回答下面的问题：\n{question}"

    answer = llm.invoke(prompt).content.strip()
    return {
        "generation": answer,
        "trace": [f"[generate] len={len(answer)}"],
    }

7.6 幻觉检测节点：答案有没有编？

def check_hallucination(state: RAGState) -> dict:
    """检查答案是否被检索文档支撑"""
    question = state["question"]
    answer = state["generation"]
    docs = state.get("documents", [])

    # 无检索路径不做幻觉检测
    if not docs:
        return {
            "is_hallucination": False,
            "trace": ["[hallucination] skip (no docs)"],
        }

    context = "\n\n".join(d.page_content for d in docs)
    result = hallucination_llm.invoke(
        f"""问题：{question}
参考资料：
{context}

模型答案：
{answer}

请判断答案中是否存在参考资料不支持的内容（即幻觉）。"""
    )
    return {
        "is_hallucination": result.is_hallucination,
        "trace": [f"[hallucination] is_hallucination={result.is_hallucination} reason={result.reason}"],
    }

7.7 兜底节点：老实说「不知道」

def fallback(state: RAGState) -> dict:
    """重试耗尽后的兜底回答"""
    question = state["question"]
    answer = (
        f"抱歉，我在知识库中没有找到足够可靠的信息来回答「{question}」。\n"
        f"建议：1）换个说法再问一次；2）补充更具体的背景；3）联系人工客服。"
    )
    return {
        "generation": answer,
        "trace": ["[fallback] retries exhausted"],
    }

8. 第五步：条件边——把「判断」变成「分支」

这是整个 Agentic RAG 的灵魂。三个路由函数：

MAX_RETRY = 2


def route_after_router(state: RAGState) -> Literal["rewrite_query", "generate"]:
    """路由后：需要检索走改写，否则直接生成"""
    return "rewrite_query" if state["need_retrieval"] else "generate"


def route_after_grade(state: RAGState) -> Literal["generate", "rewrite_query", "fallback"]:
    """评估后：相关则生成；不相关则重试；重试耗尽则兜底"""
    if state["docs_relevant"]:
        return "generate"
    if state.get("retry_count", 0) >= MAX_RETRY:
        return "fallback"
    return "rewrite_query"


def route_after_hallucination(state: RAGState) -> Literal["end", "rewrite_query", "fallback"]:
    """幻觉检测后：无幻觉结束；有幻觉重试；重试耗尽兜底"""
    if not state["is_hallucination"]:
        return "end"
    if state.get("retry_count", 0) >= MAX_RETRY:
        return "fallback"
    return "rewrite_query"

注意 retry_count 的自增位置：它应该在「决定重试」的那一刻 +1，而不是在检索节点里。我们把它放在改写节点里最自然——每次改写都意味着一次新的尝试。

def rewrite_query(state: RAGState) -> dict:
    question = state["question"]
    retry = state.get("retry_count", 0)

    if retry == 0:
        prompt = f"把下面的问题改写成更适合向量检索的查询，只输出改写后的查询：\n{question}"
    else:
        prompt = (
            f"原问题：{question}\n"
            f"上一次检索效果不好。请换一个角度改写查询，使用更具体的关键词，只输出改写后的查询。"
        )

    rewritten = llm.invoke(prompt).content.strip()
    return {
        "rewritten_question": rewritten,
        "retry_count": retry + 1,   # 关键：在这里自增
        "trace": [f"[rewrite] retry={retry + 1} query={rewritten}"],
    }

9. 第六步：组装图

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(RAGState)

# 注册节点
workflow.add_node("route_question", route_question)
workflow.add_node("rewrite_query", rewrite_query)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("check_hallucination", check_hallucination)
workflow.add_node("fallback", fallback)

# 入口
workflow.add_edge(START, "route_question")

# 路由分支
workflow.add_conditional_edges(
    "route_question",
    route_after_router,
    {"rewrite_query": "rewrite_query", "generate": "generate"},
)

# 检索链路
workflow.add_edge("rewrite_query", "retrieve")
workflow.add_edge("retrieve", "grade_documents")

# 评估分支
workflow.add_conditional_edges(
    "grade_documents",
    route_after_grade,
    {"generate": "generate", "rewrite_query": "rewrite_query", "fallback": "fallback"},
)

# 生成后校验
workflow.add_edge("generate", "check_hallucination")
workflow.add_conditional_edges(
    "check_hallucination",
    route_after_hallucination,
    {"end": END, "rewrite_query": "rewrite_query", "fallback": "fallback"},
)

# 兜底结束
workflow.add_edge("fallback", END)

app = workflow.compile()

编译后可以打印图结构确认：

print(app.get_graph().draw_mermaid())

10. 第七步：跑起来看看

def run(question: str):
    print(f"\n{'='*60}\n问题：{question}\n{'='*60}")
    result = app.invoke({
        "question": question,
        "retry_count": 0,
        "trace": [],
    })
    print("答案：", result["generation"])
    print("\n执行轨迹：")
    for t in result["trace"]:
        print("  ", t)
    return result


# 场景 1：闲聊，不需要检索
run("你好，你是谁？")

# 场景 2：正常知识问答
run("LangGraph 的条件边怎么用？")

# 场景 3：知识库里没有的问题，触发兜底
run("LangGraph 支持哪些数据库的分布式事务？")

预期输出（节选）：

============================================================
问题：你好，你是谁？
============================================================
答案： 我是 AI 助手，可以帮你解答问题。
执行轨迹：
   [route] need_retrieval=False
   [generate] len=18
   [hallucination] skip (no docs)

============================================================
问题：LangGraph 的条件边怎么用？
============================================================
答案： 通过 add_conditional_edges 添加，根据路由函数的返回值决定下一个节点。
执行轨迹：
   [route] need_retrieval=True
   [rewrite] retry=1 query=LangGraph add_conditional_edges 使用方法
   [retrieve] query=LangGraph add_conditional_edges 使用方法 hits=3
   [grade] relevant=True reason=文档直接说明了条件边的用法
   [generate] len=42
   [hallucination] is_hallucination=False reason=答案完全来自文档

11. 进阶：流式输出每个节点的思考过程

Agentic RAG 的中间过程对用户很有价值。用 stream 模式可以实时拿到每个节点的输出：

def run_stream(question: str):
    print(f"\n问题：{question}")
    for event in app.stream(
        {"question": question, "retry_count": 0, "trace": []},
        stream_mode="updates",
    ):
        for node_name, node_output in event.items():
            print(f"[{node_name}]", node_output.get("trace", [""])[-1])


run_stream("LangGraph 的 checkpointer 有什么用？")

输出：

问题：LangGraph 的 checkpointer 有什么用？
[route_question] [route] need_retrieval=True
[rewrite_query] [rewrite] retry=1 query=LangGraph checkpointer 持久化状态 用途
[retrieve] [retrieve] query=... hits=3
[grade_documents] [grade] relevant=True reason=文档说明了 checkpointer 的作用
[generate] [generate] len=56
[check_hallucination] [hallucination] is_hallucination=False reason=答案来自文档

12. 进阶：加持久化与人工介入

LangGraph 的 checkpointer 让整个图具备「记忆」能力，可以按 thread_id 恢复会话：

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app_with_memory = workflow.compile(
    checkpointer=memory,
    interrupt_before=["fallback"],   # 兜底前暂停，等待人工确认
)

config = {"configurable": {"thread_id": "user-001"}}

# 第一次调用
result = app_with_memory.invoke(
    {"question": "LangGraph 支持分布式事务吗？", "retry_count": 0, "trace": []},
    config=config,
)

# 如果停在 fallback 前，可以人工注入答案后继续
state = app_with_memory.get_state(config)
if state.next:
    print("图暂停在：", state.next)
    # 人工修改状态后继续
    app_with_memory.update_state(config, {"generation": "人工补充的答案"})
    result = app_with_memory.invoke(None, config=config)

生产环境把 MemorySaver 换成 SqliteSaver 或 PostgresSaver 即可持久化到数据库。

13. 工程化建议与踩坑记录

13.1 成本控制

每个问题最多会调用 LLM 5 次（路由 + 改写 + 评估 + 生成 + 幻觉检测），重试时更多。优化手段：

• 路由和评估用小模型（如 gpt-4o-mini），生成用大模型；
• 评估节点批量打分：一次调用评估所有文档，而不是逐条调用；
• 缓存：相同问题命中缓存直接返回，LangGraph 的 checkpointer 天然支持；
• 降级：MAX_RETRY 设为 1~2，别设太大。

13.2 死循环防护

一定要有 retry_count 上限，并且在条件边里判断，而不是在节点里 raise。否则图会一直转下去。

13.3 评估节点的准确性

LLM 做相关性判断会有误判。生产环境建议：

• 用小模型 + few-shot 提升稳定性；
• 结合向量相似度阈值做双重过滤（相似度 < 0.5 直接判不相关，不用调 LLM）；
• 记录评估结果，定期人工抽检校准。

13.4 兜底不是失败

兜底回答要有用，而不是干巴巴一句「不知道」。给出「换个说法」「补充背景」「联系人工」的具体建议，用户体验会好很多。

13.5 可观测性

trace 字段是调试神器。生产环境把它接到 LangSmith 或自建日志，能看到每个问题的完整决策路径，出问题能快速定位是「路由错了」还是「评估误判」。

14. 总结

从线性 RAG 到 Agentic RAG，核心变化是把「一次性流水线」变成「带判断和回边的状态机」：

维度	线性 RAG	Agentic RAG
流程	单向直线	带条件分支和回边的图
检索决策	总是检索	按需检索
查询质量	原样查询	改写 + 重试
结果校验	无	相关性评估 + 幻觉检测
失败处理	硬编答案	兜底 + 人工介入
可观测性	黑盒	全链路 trace

用 LangGraph 实现的关键就三件事：

1. State 设计好：把判断结果、重试计数、轨迹都放进 State；
2. 节点职责单一：每个节点只做一件事，方便替换和测试；
3. 条件边写清楚：所有分支和重试上限都在条件边里表达，图结构就是文档。

代码可以直接跑，把知识库换成你自己的文档、把 LLM 换成你用的模型，就是一个能上生产的 Agentic RAG 骨架。下一步可以尝试的方向：多跳检索（问题拆解成子问题）、多路召回（向量 + BM25 混合）、以及用 LangGraph 的 Send API 做并行检索。