AI大模型实战系列（八）：从静态检索到主动决策——Agentic RAG（智能体 RAG）与 LangGraph 工业级编排

在前面的七个章节中，我们从环境搭建起步，一路攻克了数据解析、向量底座、混合检索、Reranker 重排，最终通过 Ragas 自动化评估体系为系统打上了量化质检的钢印。

但这套打磨到极致的 Advanced RAG 系统，依然存在一个底层的结构性缺陷：它是一条单向的、静态的“流水线（Pipeline）”。
无论用户问什么，它永远只能机械地执行“向量搜索 -> 重排 -> 生成”这三步。如果用户问：“请对比我们公司和竞品 A 公司在 2023 年 Q3 的营销策略差异”，传统 RAG 极大概率会失败。因为它一次检索可能只捞到了本公司的资料，它没有能力去“停下来思考”、“发现资料缺失”，然后再“发起第二次检索”。

为了打破这种线性思维的桎梏，本章我们将进入当前大模型应用领域的最前沿：Agentic RAG（智能体 RAG）。我们将深度剖析如何利用 LangChain V0.3 核心力推的 LangGraph 框架，赋予 RAG 系统路由决策、工具调用与自我反思（Self-Reflection）的动态生命力。

---

一、 认知跃迁：单向流水线 vs 智能体工作流

1. 传统 RAG 的“盲人摸象”困境

在传统 RAG 架构中，大语言模型（LLM）被完全架空成了一个“翻译官”。整个检索过程是硬编码的，大模型不参与任何检索策略的制定。
如果向量数据库因为 Query 表达不清而召回了一堆垃圾文档，大模型只能无奈地基于这些垃圾文档回答“我不知道”，或者被迫产生幻觉。它无法主动对系统说：“等一下，这些资料不对，让我换个搜索词重新查一遍。”

2. Agentic RAG 的降维打击

Agentic RAG 彻底颠覆了这一逻辑。它不再把 LLM 当作流水线末端的翻译官，而是将其前置为整个系统的**“中央大脑（大脑路由调度器）”**。

在 Agentic 架构下，RAG 不再是一个死板的流程，而是一个包含了丰富**工具库（Tools）**的武器库。当用户提出复杂问题时：

1. 大模型首先进行意图识别与路由（Routing）：决定是去查内部的 HR 向量库、财务的 SQL 数据库，还是去调用 Google Search 查全网实时信息。
2. 大模型具备自我纠错（Self-Correction）能力：如果查出来的文档它觉得不满意，它会自动触发“查询重写（Query Rewrite）”节点，换一种表达方式再次发起检索，直到凑齐完美回答所需的全部拼图。

---

二、 LangGraph 架构解密：大模型状态机的工程化革命

为了实现这种复杂的循环、条件判断和状态记忆，LangChain 在 V0.3 版本全面拥抱了基于图结构（Graph-based）的编排框架——LangGraph。

1. 为什么弃用旧版 LangChain 的 AgentExecutor？

早期的 LangChain Agent 是一个黑盒，其执行逻辑、死循环拦截极其难以控制，且很难在多步推理中精准保存中间状态。
LangGraph 将大模型应用抽象为有向图（Directed Graph）和状态机（State Machine）：

• State（状态）：一个贯穿整个图运行周期的共享数据结构（通常是一个 Python 字典或 Pydantic 模型），记录了对话历史、当前问题、已召回的文档、当前的重试次数等。
• Nodes（节点）：具体的执行动作，通常是 Python 函数，如 retrieve_documents() 或 grade_documents()。节点负责接收 State，执行逻辑，然后更新 State。
• Edges（边）与 Conditional Edges（条件边）：控制流程走向。例如，一条条件边可以询问 LLM：“召回的文档相关吗？”如果相关，走向 generate 节点；如果不相关，走向 rewrite_query 节点。

---

三、 工业级实战：构建具备“自我反思”能力的 CRAG 架构

目前业界最主流的 Agentic RAG 范式之一是 CRAG（Corrective RAG，纠错型 RAG）。下面我们将使用 LangGraph 核心代码，从零手搓一个具备文档评估和网页搜索兜底能力的 CRAG 工业级工作流。

1. 定义全局状态（State）

在编写任何节点之前，必须先定义图在流转过程中需要携带哪些核心数据。

from typing import List, TypedDict

class GraphState(TypedDict):
    """
    定义贯穿整个智能体 RAG 生命周期的数据结构
    """
    question: str          # 用户的原始提问
    generation: str        # 最终生成的答案
    documents: List[str]   # 检索到的文档片段缓存
    web_fallback: bool     # 标志位：是否需要触发全网兜底搜索

2. 构建核心执行节点（Nodes）

我们将 RAG 的每一步拆解为独立的纯函数节点。

节点 A：文档检索 (Retrieve)

def retrieve(state: GraphState):
    print("--- 节点：执行向量检索 ---")
    question = state["question"]
    # 调用我们在之前章节封装好的混合检索器
    documents = hybrid_retriever.invoke(question)
    # 将捞出的文档更新到状态字典中
    return {"documents": documents, "question": question}

节点 B：文档相关性硬核评估 (Grade Documents)
这是 Agentic RAG 的灵魂。大模型在此刻充当裁判，评估检索质量。

def grade_documents(state: GraphState):
    print("--- 节点：评估文档相关性 ---")
    question = state["question"]
    documents = state["documents"]
    
    # 此处省略调用 LLM API 进行判断的具体 Prompt 代码
    # 逻辑：让 LLM 逐一判断文档是否包含回答问题所需的知识
    filtered_docs = []
    web_fallback = False
    
    for doc in documents:
        score = llm_grader.invoke({"question": question, "document": doc})
        if score == "yes":
            filtered_docs.append(doc)
        else:
            print("--- 过滤掉无关噪音文档 ---")
            
    # 如果所有文档都被判定为无关垃圾，则激活全网搜索兜底
    if not filtered_docs:
        print("--- 内部知识库检索失败，触发互联网搜索回退机制 ---")
        web_fallback = True
        
    return {"documents": filtered_docs, "web_fallback": web_fallback}

节点 C 与 D：查询重写与生成 (Rewrite & Generate)

def rewrite_query(state: GraphState):
    print("--- 节点：重写并优化查询词 ---")
    # 调用 LLM 将泛泛的 Query 转化为更精准的搜索词
    better_question = llm_rewriter.invoke(state["question"])
    return {"question": better_question}

def generate(state: GraphState):
    print("--- 节点：最终答案生成 ---")
    # 基于经过审核的纯净 documents 生成最终回答
    response = llm_generator.invoke({"context": state["documents"], "question": state["question"]})
    return {"generation": response}

3. 定义条件边（路由决策逻辑）

构建好节点后，我们需要定义决策逻辑来驱动图的流转。

def route_after_grading(state: GraphState):
    """
    路由逻辑：评估完文档后，往哪里走？
    """
    web_fallback = state.get("web_fallback")
    if web_fallback:
        # 如果内部没资料，要求图走向网络搜索重写节点
        return "rewrite"
    else:
        # 资料充足且相关，直接去生成答案
        return "generate"

4. 组装 LangGraph 工作流

最后，我们将散落的节点和边织成一张具有动态循环能力的巨网。

from langgraph.graph import END, StateGraph

# 1. 实例化状态图
workflow = StateGraph(GraphState)

# 2. 注册所有业务节点
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("rewrite_query", rewrite_query)
workflow.add_node("generate", generate)

# 3. 编排边与流程逻辑
workflow.set_entry_point("retrieve") # 入口点：永远先检索
workflow.add_edge("retrieve", "grade_documents") # 检索完必须进行评分过滤

# 注入动态条件路由：评估完之后，动态决定是重写还是生成
workflow.add_conditional_edges(
    "grade_documents",  # 路由的起点节点
    route_after_grading, # 路由判断函数
    {
        "rewrite": "rewrite_query", # 如果函数返回 rewrite，则跳到重写节点
        "generate": "generate"      # 如果函数返回 generate，则跳到生成节点
    }
)

# 形成闭环逻辑：重写完之后，重新回到起点再次检索 (这里构成了一个强大的自循环纠错机制！)
workflow.add_edge("rewrite_query", "retrieve") 

# 生成完毕后，流程结束
workflow.add_edge("generate", END)

# 4. 编译这个极其强大的 Agentic 工作流
app = workflow.compile()

---

四、 商业化终局：RAG 系统的工程化与服务端部署

当这样一个具备思考、纠错、甚至能够调用外部 API 的 Agentic RAG 在本地跑通后，如何将其无缝部署到企业生产环境中交付给前端或业务线？

在商业实战中，我们绝对不会让业务方直接运行 Python 脚本。标准的企业级部署方案是将其封装为高并发的 RESTful API 或流式（Streaming）接口。

1. LangServe 的极速封装

基于 FastAPI 构建的 LangServe 是目前部署 LangChain/LangGraph 应用的绝对利器。它自动为你的应用生成具备并发保护、流式输出以及 OpenAPI 文档的生产级接口。

from fastapi import FastAPI
from langserve import add_routes

# 创建 FastAPI 基础应用
app = FastAPI(title="企业级 Agentic RAG 核心路由服务")

# 一行代码，将我们刚才编译好的 LangGraph 智能体挂载为标准的 HTTP 接口
add_routes(
    app,
    workflow_app, # 刚才编译好的 app
    path="/agentic-rag"
)

if __name__ == "__main__":
    import uvicorn
    # 启动生产级 Uvicorn 服务器
    uvicorn.run(app, host="0.0.0.0", port=8000)

2. 流式输出（Streaming）与用户体验

现代 AI 产品（如 ChatGPT）的体验核心在于“打字机效果”。大模型的推理可能需要 5 秒，如果不使用流式传输，用户会以为系统死机了。
通过 LangServe，前端开发者可以直接调用 /agentic-rag/stream 接口。结合 Server-Sent Events (SSE) 技术，前端可以在大模型生成出第一个 Token 的瞬间就开始在界面上渲染，将用户的可感知等待时间从 5 秒压缩到了 300 毫秒以内。

3. 可观测性监控（LangSmith 深度集成）

当 RAG 系统演变为复杂的图结构时，一旦回答出错，排查极其困难（到底是哪一次重写的 Query 导致了检索跑偏？）。
企业生产环境必须开启全链路追踪。通过配置系统环境变量引入 LangSmith，你可以获得一个极其华丽的控制台：清晰地回放每一次用户请求在图结构中游走的完整轨迹，精确到每一个 Node 消耗了多少 Token 和毫秒级的延迟，实现“所见即所得”的架构级 Debug。

---

本章总结

从依赖硬编码指令的线性 RAG，到引入 LangGraph 赋予大模型自我反思、主动规划路线的 Agentic RAG，我们完成了 AI 应用从“检索工具”到“智能中枢”的质变。我们见证了系统如何在内部知识库匮乏时，通过条件边（Conditional Edge）优雅地重写查询并触发互联网兜底检索。

同时，借助 LangServe 与流式并发部署方案，这套在实验室里打磨的黄金架构，终于插上了商业化落地的翅膀，具备了应对高并发企业级生产环境的硬实力。

至此，《AI大模型实战系列》圆满收官！从第一章的环境搭建，到第八章的智能体工业部署，我们走过了一段惊心动魄且极具价值的技术攀峰之旅。掌握了这套完整的底层逻辑与代码工程心法，你已经彻底脱离了 API 调用者的范畴，拥有了主导任何顶级企业 AI 大模型项目的架构师底蕴。未来的智能时代，这套基建，将成为你最强硬的技术底牌！