Agentic RAG 实战:用 LangGraph 构建会自我纠错的智能检索系统

用 LangGraph 1.0 手把手构建自纠错 Agentic RAG 系统。涵盖查询路由、文档评分、查询改写、幻觉检测全流程,含完整 Python 代码、三种进阶架构模式与生产部署关键考量。

为什么传统 RAG 已经不够用了?

如果你在 2024-2025 年搭建过 RAG(检索增强生成)系统,大概率经历过这种场景:用户问了一个稍微复杂点的问题,系统返回一堆不太相关的文档碎片,然后 LLM 硬着头皮"编"出了一个看似合理但其实错得离谱的答案。用户没得到想要的信息,你却为这次调用付了一笔不小的 Token 费。

说实话,这种经历我自己就遇到过不止一次。

问题的根源在于,传统 RAG 本质上是一个单次、线性、无反馈的管道。用户输入查询 → 向量检索 → 拼接上下文 → LLM 生成。检索结果质量差?没关系,整个流程没有任何纠错机制,"垃圾进,垃圾出"就是必然结局。

到了 2026 年,Agentic RAG(智能体驱动的检索增强生成)已经成为 AI 工程领域最火的架构模式之一。它在传统 RAG 基础上引入了 AI 智能体的自主决策能力——检索前先理解意图、检索后自动评估质量、质量不够就自动改写查询重新检索、生成后还能检查幻觉。简单说,就是把 RAG 从一个"死板流水线"升级成了一个"会思考的研究员"。

这篇文章会用 LangGraph(LangChain 的图状态机框架,2026 年已发布 1.0 稳定版)手把手带你构建一个完整的自纠错 Agentic RAG 系统。包含完整可运行的 Python 代码、架构详解、和传统 RAG 的对比,以及在生产环境部署时你需要注意的那些坑。

Agentic RAG 与传统 RAG:核心差异

在动手写代码之前,先把两者的本质区别搞清楚。

传统 RAG:图书管理员模式

传统 RAG 就像一个严格按规矩办事的图书管理员。你说"给我找关于量子计算的书",他就去书架上按标签找,找到什么就给你什么——不管这些书是不是真的跟你的问题相关,也不会追问"你是想了解原理还是实际应用?"

用户查询 → 向量检索(单次) → 拼接上下文 → LLM 生成 → 输出
                    ↑ 没有反馈回路 ↑

Agentic RAG:研究员模式

Agentic RAG 则更像一个经验丰富的研究员。收到问题后,他会先分析"这个问题到底需要什么类型的信息?",然后选择合适的检索策略。拿到结果后还会评估"这些信息够不够?质量怎么样?",不满意就换个关键词再查,甚至去别的数据源找。最后生成答案时还要自查"我有没有编造内容?答案有没有被检索到的文档支持?"

光是这个思路上的转变,就已经是质的飞跃了。

用户查询 → 意图分析/路由
              ↓
        需要检索?──否──→ 直接生成
              ↓ 是
        向量检索 → 文档评分
              ↓
        文档相关?──否──→ 改写查询 → 重新检索 ↑
              ↓ 是
        LLM 生成 → 幻觉检查
              ↓
        答案可靠?──否──→ 补充检索/重新生成 ↑
              ↓ 是
           输出

对比一览表

维度传统 RAGAgentic RAG
执行流程线性单次图状态机,支持循环和分支
检索策略固定(单一向量搜索)动态(向量搜索 + 关键词 + Web 搜索 + API)
自纠错能力查询改写、文档评分、幻觉检测
数据源单一向量库多数据源 + 外部工具
适用场景简单事实性问答多步推理、跨源分析、高准确性要求
延迟与成本低延迟、低成本更高延迟、更多 Token 消耗
准确性依赖检索质量通过自纠错显著提升(研究显示提升约 14%)

环境准备与依赖安装

我们的技术栈基于 2026 年 2 月的最新稳定版本:

  • LangGraph 1.0+:LangChain 的图状态机框架,2025 年底发布 GA 版本
  • LangChain 0.3+:LLM 应用框架
  • ChromaDB:轻量级向量数据库,拿来做教程演示刚好
  • OpenAI GPT-4.1Claude 4 Sonnet:作为推理引擎
  • Python 3.11+
# 创建虚拟环境
python -m venv agentic-rag-env
source agentic-rag-env/bin/activate

# 安装核心依赖
pip install langgraph>=1.0.0 langchain>=0.3.0 langchain-openai>=0.3.0
pip install chromadb>=0.6.0 langchain-chroma>=0.2.0
pip install langchain-community>=0.3.0

# 设置 API Key
export OPENAI_API_KEY="your-api-key-here"

第一步:构建向量知识库

先搞一个示例知识库。在实际项目中,这些可以是你的企业文档、产品手册、技术文档——什么都行。这里为了演示方便,我们直接手写几个 Document 对象。

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 准备示例文档(实际项目中用文档加载器替代)
documents = [
    Document(
        page_content="LangGraph 是 LangChain 生态中的图状态机框架,"
                     "专门用于构建多步骤、有状态的 AI 智能体工作流。"
                     "它在 2025 年底发布了 1.0 GA 版本,"
                     "已被 Uber、LinkedIn、Klarna 等企业用于生产环境。"
                     "LangGraph 的核心概念是将工作流定义为节点(Node)"
                     "和边(Edge)组成的有向图。",
        metadata={"source": "langgraph-docs", "topic": "framework"}
    ),
    Document(
        page_content="Agentic RAG 是在传统 RAG 架构之上引入 AI 智能体的"
                     "自主决策能力。核心改进包括:查询改写、文档相关性评分、"
                     "自纠错检索循环、幻觉检测与多源信息整合。"
                     "研究表明 Agentic RAG 相比传统 RAG 可提升约 14% 的准确率。",
        metadata={"source": "research-paper", "topic": "agentic-rag"}
    ),
    Document(
        page_content="向量数据库是 RAG 系统的核心存储组件。"
                     "2026 年主流选择包括 Chroma(轻量级)、Qdrant(高性能)、"
                     "Pinecone(托管服务)和 Weaviate(开源全能型)。"
                     "选择向量数据库时需考虑:索引规模、查询延迟、过滤能力、"
                     "托管方式和成本。对于百万级以下文档,Chroma 足以胜任。",
        metadata={"source": "vector-db-guide", "topic": "infrastructure"}
    ),
    Document(
        page_content="Corrective RAG (CRAG) 是一种自纠错 RAG 技术。"
                     "它的核心思路是在生成之前加入文档质量评估步骤。"
                     "系统为每个检索到的文档计算相关性置信度分数,"
                     "如果分数低于阈值,则触发查询改写并重新检索。"
                     "CRAG 还支持将文档拆分为知识条目(Knowledge Strips),"
                     "对每个条目单独评分并过滤无关内容。",
        metadata={"source": "crag-paper", "topic": "corrective-rag"}
    ),
    Document(
        page_content="上下文工程(Context Engineering)是 2025-2026 年兴起的新范式,"
                     "正在替代传统的提示词工程。"
                     "核心理念是将 Prompt 视为动态构建的软件工件,"
                     "通过 Token 预算管理、滑动窗口、上下文压缩等技术,"
                     "在有限的上下文窗口中最大化信息密度。"
                     "Claude Code 是上下文工程的一个教科书级实践。",
        metadata={"source": "context-engineering", "topic": "prompting"}
    ),
]

# 创建向量数据库
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    collection_name="agentic_rag_demo",
    persist_directory="./chroma_db"
)

# 创建检索器(返回 top-3 相关文档)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

print(f"知识库已创建,共 {len(documents)} 个文档")

第二步:定义图状态与核心节点

LangGraph 的核心思想是把工作流定义为一个有向状态图。每个节点是一个处理函数,边决定数据的流向。状态(State)在整个图执行过程中持续传递和更新。

这个设计其实挺优雅的——如果你熟悉有限状态机的概念,理解起来会很快。

定义状态结构

from typing import List, TypedDict
from langchain_core.documents import Document


class AgenticRAGState(TypedDict):
    """Agentic RAG 工作流的状态定义"""
    question: str            # 用户原始问题
    rewritten_question: str  # 改写后的问题(用于重新检索)
    documents: List[Document]  # 检索到的文档
    generation: str          # LLM 生成的答案
    retry_count: int         # 重试计数器(防止无限循环)
    route_decision: str      # 路由决策:retrieve / direct_answer

定义各个处理节点

接下来是重头戏——我们要定义 7 个核心节点。每个节点负责流程中的一个环节,从查询路由到幻觉检测,各司其职。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4.1", temperature=0)


# === 节点 1:查询路由 ===
def route_question(state: AgenticRAGState) -> AgenticRAGState:
    """分析查询意图,决定是否需要检索"""
    router_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个查询路由器。根据用户问题判断是否需要从知识库检索信息。

规则:
- 如果问题涉及特定技术细节、框架、工具或需要引用数据,输出 "retrieve"
- 如果是闲聊、通用知识或模型本身就能准确回答的问题,输出 "direct_answer"

仅输出 "retrieve" 或 "direct_answer",不要输出其他内容。"""),
        ("human", "{question}")
    ])

    chain = router_prompt | llm | StrOutputParser()
    decision = chain.invoke({"question": state["question"]}).strip().lower()

    return {**state, "route_decision": decision}


# === 节点 2:文档检索 ===
def retrieve_documents(state: AgenticRAGState) -> AgenticRAGState:
    """从向量数据库检索相关文档"""
    query = state.get("rewritten_question") or state["question"]
    docs = retriever.invoke(query)
    return {**state, "documents": docs}


# === 节点 3:文档相关性评分 ===
def grade_documents(state: AgenticRAGState) -> AgenticRAGState:
    """评估检索到的文档与问题的相关性,过滤低质量文档"""
    grader_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个文档相关性评分器。判断给定文档是否与用户问题相关。

评分标准:
- 文档包含与问题直接相关的信息 → "relevant"
- 文档与问题无关或仅有微弱关联 → "irrelevant"

仅输出 "relevant" 或 "irrelevant"。"""),
        ("human", "问题:{question}\n\n文档内容:{document}")
    ])

    chain = grader_prompt | llm | StrOutputParser()

    relevant_docs = []
    for doc in state["documents"]:
        score = chain.invoke({
            "question": state["question"],
            "document": doc.page_content
        }).strip().lower()
        if score == "relevant":
            relevant_docs.append(doc)

    return {**state, "documents": relevant_docs}


# === 节点 4:查询改写 ===
def rewrite_query(state: AgenticRAGState) -> AgenticRAGState:
    """当检索结果不佳时,改写查询以获得更好的检索效果"""
    rewrite_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个查询优化专家。用户的原始查询没有检索到足够相关的文档。
请改写查询,使其更精确、更适合语义搜索。

改写策略:
1. 提取核心关键词
2. 添加同义词或相关术语
3. 分解复合问题为更聚焦的子查询
4. 去除模糊或过于宽泛的表述

仅输出改写后的查询,不要解释。"""),
        ("human", "原始查询:{question}")
    ])

    chain = rewrite_prompt | llm | StrOutputParser()
    new_query = chain.invoke({"question": state["question"]})
    retry = state.get("retry_count", 0) + 1

    return {**state, "rewritten_question": new_query, "retry_count": retry}


# === 节点 5:答案生成 ===
def generate_answer(state: AgenticRAGState) -> AgenticRAGState:
    """基于检索到的文档生成答案"""
    context = "\n\n---\n\n".join(
        [doc.page_content for doc in state["documents"]]
    )

    generate_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个专业的 AI 技术助手。基于提供的参考文档回答用户问题。

规则:
1. 仅基于提供的文档内容回答,不要编造信息
2. 如果文档中的信息不足以完整回答问题,明确说明
3. 回答要结构清晰、准确具体
4. 引用关键数据时标注来源文档"""),
        ("human", "参考文档:\n{context}\n\n问题:{question}")
    ])

    chain = generate_prompt | llm | StrOutputParser()
    answer = chain.invoke({
        "context": context,
        "question": state["question"]
    })

    return {**state, "generation": answer}


# === 节点 6:幻觉检测 ===
def check_hallucination(state: AgenticRAGState) -> AgenticRAGState:
    """检查生成的答案是否被文档内容支持"""
    context = "\n\n".join(
        [doc.page_content for doc in state["documents"]]
    )

    hallucination_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个事实核查员。判断 AI 生成的答案是否完全被参考文档支持。

评判标准:
- 答案中的所有事实性声明都能在文档中找到依据 → "supported"
- 答案中包含文档未提及的事实性声明(即幻觉) → "hallucination"

仅输出 "supported" 或 "hallucination"。"""),
        ("human", "参考文档:\n{context}\n\n生成的答案:\n{generation}")
    ])

    chain = hallucination_prompt | llm | StrOutputParser()
    result = chain.invoke({
        "context": context,
        "generation": state["generation"]
    }).strip().lower()

    return {**state, "hallucination_check": result}


# === 节点 7:直接回答(无需检索) ===
def direct_answer(state: AgenticRAGState) -> AgenticRAGState:
    """对于不需要检索的问题,直接用 LLM 回答"""
    direct_prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个专业的 AI 技术助手。请直接回答用户的问题。"),
        ("human", "{question}")
    ])

    chain = direct_prompt | llm | StrOutputParser()
    answer = chain.invoke({"question": state["question"]})

    return {**state, "generation": answer}

第三步:组装 LangGraph 状态图

现在把所有节点用边连接起来,形成完整的工作流图。坦白说,这一步是整个系统最核心的部分——条件边(Conditional Edges)定义了智能体的全部决策逻辑,搞清楚这里基本就理解了 Agentic RAG 的精髓。

from langgraph.graph import StateGraph, START, END

# 创建状态图
workflow = StateGraph(AgenticRAGState)

# 添加所有节点
workflow.add_node("route_question", route_question)
workflow.add_node("retrieve_documents", retrieve_documents)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("rewrite_query", rewrite_query)
workflow.add_node("generate_answer", generate_answer)
workflow.add_node("check_hallucination", check_hallucination)
workflow.add_node("direct_answer", direct_answer)


# === 定义条件路由函数 ===
def route_after_classification(state: AgenticRAGState) -> str:
    """路由节点后的条件分支"""
    if state.get("route_decision") == "direct_answer":
        return "direct_answer"
    return "retrieve_documents"


def route_after_grading(state: AgenticRAGState) -> str:
    """文档评分后的条件分支"""
    if len(state["documents"]) == 0:
        # 没有相关文档,需要改写查询
        if state.get("retry_count", 0) >= 2:
            # 已达最大重试次数,用现有信息生成
            return "generate_answer"
        return "rewrite_query"
    return "generate_answer"


def route_after_hallucination_check(state: AgenticRAGState) -> str:
    """幻觉检测后的条件分支"""
    if state.get("hallucination_check") == "hallucination":
        if state.get("retry_count", 0) >= 2:
            return END
        return "generate_answer"  # 重新生成
    return END


# === 连接边 ===
# 入口 → 路由
workflow.add_edge(START, "route_question")

# 路由 → 检索/直接回答
workflow.add_conditional_edges(
    "route_question",
    route_after_classification,
    {
        "retrieve_documents": "retrieve_documents",
        "direct_answer": "direct_answer"
    }
)

# 检索 → 评分
workflow.add_edge("retrieve_documents", "grade_documents")

# 评分 → 改写/生成
workflow.add_conditional_edges(
    "grade_documents",
    route_after_grading,
    {
        "rewrite_query": "rewrite_query",
        "generate_answer": "generate_answer"
    }
)

# 改写 → 重新检索
workflow.add_edge("rewrite_query", "retrieve_documents")

# 生成 → 幻觉检测
workflow.add_edge("generate_answer", "check_hallucination")

# 幻觉检测 → 结束/重新生成
workflow.add_conditional_edges(
    "check_hallucination",
    route_after_hallucination_check,
    {
        "generate_answer": "generate_answer",
        END: END
    }
)

# 直接回答 → 结束
workflow.add_edge("direct_answer", END)

# 编译图
app = workflow.compile()
print("Agentic RAG 工作流已编译完成")

第四步:运行与测试

好了,系统搭建完毕。来看看它在面对不同类型查询时的实际表现。

测试 1:需要检索的技术问题

# 需要检索的问题
result = app.invoke({
    "question": "LangGraph 的核心架构是什么?有哪些企业在生产环境使用?",
    "rewritten_question": "",
    "documents": [],
    "generation": "",
    "retry_count": 0,
    "route_decision": ""
})

print("=" * 60)
print(f"问题:{result['question']}")
print(f"路由决策:{result['route_decision']}")
print(f"检索到 {len(result['documents'])} 个相关文档")
print(f"重试次数:{result['retry_count']}")
print(f"\n答案:\n{result['generation']}")

测试 2:不需要检索的通用问题

# 不需要检索的通用问题
result = app.invoke({
    "question": "什么是 Python 的列表推导式?",
    "rewritten_question": "",
    "documents": [],
    "generation": "",
    "retry_count": 0,
    "route_decision": ""
})

print(f"路由决策:{result['route_decision']}")
# 预期:direct_answer,跳过检索直接回答

测试 3:触发自纠错的模糊查询

这个测试比较有意思——我们故意用一个很含糊的问题来看系统能不能自己"想明白"用户在问什么。

# 故意用模糊查询测试自纠错
result = app.invoke({
    "question": "那个可以修复自己错误的AI检索技术叫什么?",
    "rewritten_question": "",
    "documents": [],
    "generation": "",
    "retry_count": 0,
    "route_decision": ""
})

print(f"改写后的查询:{result['rewritten_question']}")
print(f"重试次数:{result['retry_count']}")
print(f"最终答案:\n{result['generation']}")
# 预期:系统会改写查询为更精确的表述,如"自纠错 RAG 技术 CRAG"

第五步:可视化工作流执行轨迹

调试 Agentic RAG 的时候,了解每一步到底走了哪条路非常重要。(相信我,如果你不做这个,debug 的时候会很痛苦。)LangGraph 支持流式输出每个节点的执行状态:

def run_with_trace(question: str):
    """运行 Agentic RAG 并输出详细执行轨迹"""
    initial_state = {
        "question": question,
        "rewritten_question": "",
        "documents": [],
        "generation": "",
        "retry_count": 0,
        "route_decision": ""
    }

    print(f"\n{'=' * 60}")
    print(f"查询:{question}")
    print(f"{'=' * 60}")

    for step_output in app.stream(initial_state):
        for node_name, node_state in step_output.items():
            print(f"\n→ 节点 [{node_name}]")
            if node_name == "route_question":
                print(f"  路由决策:{node_state.get('route_decision', 'N/A')}")
            elif node_name == "retrieve_documents":
                print(f"  检索到 {len(node_state.get('documents', []))} 个文档")
            elif node_name == "grade_documents":
                docs = node_state.get('documents', [])
                print(f"  筛选后剩余 {len(docs)} 个相关文档")
            elif node_name == "rewrite_query":
                rq = node_state.get('rewritten_question', '')
                print(f"  改写查询:{rq}")
            elif node_name == "generate_answer":
                gen = node_state.get("generation", "")
                print(f"  生成答案(前100字):{gen[:100]}...")
            elif node_name == "check_hallucination":
                hc = node_state.get('hallucination_check', 'N/A')
                print(f"  幻觉检测:{hc}")

    print(f"\n{'-' * 60}")
    print("执行完毕")


# 运行测试
run_with_trace("Agentic RAG 和 Corrective RAG 有什么关系?")

进阶:三种 Agentic RAG 架构模式

上面我们实现的是最基础的自纠错 Agentic RAG。但在实际生产环境中,根据不同需求,还有几种更高级的玩法值得了解。

模式一:Adaptive RAG(自适应 RAG)

根据查询的复杂度动态选择检索策略。简单问题直接用 LLM 内部知识回答;中等复杂度的用向量搜索;需要最新信息的走 Web 搜索;高复杂度的同时调用多个数据源。

我个人觉得这是最实用的模式——大部分查询其实不需要完整的 Agentic 流程,用自适应策略可以在质量和成本之间取得很好的平衡。

def adaptive_router(state: AgenticRAGState) -> str:
    """自适应路由:根据查询复杂度选择检索策略"""
    analysis_prompt = ChatPromptTemplate.from_messages([
        ("system", """分析查询并选择最佳检索策略:
- "parametric":LLM 内部知识足以回答(通用知识、简单概念)
- "vector_search":需要从知识库检索(特定技术细节、内部文档)
- "web_search":需要最新信息(时事新闻、最新版本信息)
- "hybrid":需要多源综合(复杂分析、对比研究)

仅输出策略名称。"""),
        ("human", "{question}")
    ])

    chain = analysis_prompt | llm | StrOutputParser()
    return chain.invoke({"question": state["question"]}).strip()

模式二:Corrective RAG (CRAG)

在文档评分阶段做得更精细——不仅判断整篇文档的相关性,还把文档拆分为知识条目(Knowledge Strips),对每个条目单独评分。这样可以从一个部分相关的文档中提取出真正有用的段落,同时丢弃那些无关的部分。

这个思路相当巧妙。

模式三:Self-RAG(自反思 RAG)

在生成阶段引入多维度的自我检查——不仅看有没有幻觉,还评估答案的完整性、准确性和有用性。如果任何一个维度不达标,就触发补充检索或重新生成。

def multi_dimension_check(state: AgenticRAGState) -> dict:
    """多维度答案质量评估"""
    eval_prompt = ChatPromptTemplate.from_messages([
        ("system", """评估生成答案的质量。对以下三个维度分别评分(yes/no):

1. grounded:答案的事实声明是否都被文档支持?
2. complete:答案是否完整回答了用户的问题?
3. useful:答案对用户是否真正有帮助?

以 JSON 格式输出:
{"grounded": "yes/no", "complete": "yes/no", "useful": "yes/no"}"""),
        ("human", "文档:{context}\n问题:{question}\n答案:{generation}")
    ])

    chain = eval_prompt | llm | StrOutputParser()
    context = "\n".join([d.page_content for d in state["documents"]])
    result = chain.invoke({
        "context": context,
        "question": state["question"],
        "generation": state["generation"]
    })

    return result

生产部署关键考量

从教程代码到生产系统,中间还有不少坑要踩。以下是几个最值得注意的点。

1. 防止无限循环

Agentic RAG 最大的风险就是自纠错循环失控——系统不停地改写查询、重新检索,但就是找不到满意的结果。我们在代码中用 retry_count 做了硬限制(最多 2 次重试),但生产环境还应该加全局超时:

import asyncio

async def run_with_timeout(question: str, timeout_seconds: int = 30):
    """带超时保护的执行"""
    try:
        result = await asyncio.wait_for(
            app.ainvoke({
                "question": question,
                "rewritten_question": "",
                "documents": [],
                "generation": "",
                "retry_count": 0,
                "route_decision": ""
            }),
            timeout=timeout_seconds
        )
        return result
    except asyncio.TimeoutError:
        return {"generation": "抱歉,处理超时,请简化您的问题后重试。"}

2. Token 成本控制

每一轮自纠错循环都要额外花 Token。长期跑下来这笔费用可不小。建议这样做:

  • 用轻量模型(如 GPT-4.1-mini 或 Claude Haiku)做路由和评分这类"判断类"任务
  • 用更强的模型(如 GPT-4.1 或 Claude Sonnet)做最终生成
  • 设置单次查询的 Token 预算上限
  • 记录每次查询的 Token 消耗,方便后续做成本分析

3. 可观测性与日志

Agentic RAG 的执行路径是动态的,每次查询走的路线可能都不一样,调试难度远超传统 RAG。强烈建议集成 LangSmith 或其他 LLM 可观测性平台,记录每个节点的输入、输出和延迟。不然出了问题你根本不知道是哪个环节掉链子了。

4. 向量数据库选型

教程中用 ChromaDB 做演示,但到了生产环境你得认真想想用什么:

  • 百万级以下文档:Chroma 或 Qdrant(自托管)完全够用
  • 百万至亿级文档:推荐 Qdrant 集群或 Pinecone 托管服务
  • 需要混合检索(语义 + 关键词):Weaviate 或 Elasticsearch + 向量插件

5. 嵌入模型一致性

这是一个特别容易踩的坑:索引时用了模型 A 生成嵌入向量,查询时却换成了模型 B。不同模型的向量空间是不兼容的,混用会导致检索质量断崖式下降。始终确保文档嵌入和查询嵌入使用完全相同的模型。

常见问题 FAQ

Q1:Agentic RAG 会不会因为多次检索和评估而太慢?

确实会比传统 RAG 慢,因为它可能要跑多轮检索和评估。但你得换个角度想——这不是"快 vs 慢"的问题,而是"低质量快速回答 vs 高质量可靠回答"的权衡。实际应用中可以通过几种方式优化延迟:用轻量模型做评分和路由、设置合理的最大重试次数、对子查询做并行检索。如果你的场景对延迟特别敏感(比如实时聊天),可以用 Adaptive RAG 模式,只让复杂查询走完整的 Agentic 流程。

Q2:什么时候该用 Agentic RAG 而不是传统 RAG?

如果你的场景只是简单的事实性问答("产品 X 的价格是多少?"),传统 RAG 完全够用,没必要杀鸡用牛刀。但碰到以下情况就该考虑升级了:需要跨多个数据源整合信息;用户提问经常措辞模糊、需要意图解析;对回答准确性要求极高(金融、医疗、法律等领域);需要多步推理才能得出答案。

一个很实用的判断标准:如果传统 RAG 系统超过 20% 的回答质量不达标,就是时候升级了。

Q3:LangGraph 和 LangChain 是什么关系?必须一起用吗?

LangGraph 是 LangChain 生态的一部分,专门用来构建有状态的多步骤 AI 工作流。它依赖 LangChain 的核心组件(LLM 接口、Prompt 模板、输出解析器等),但核心价值在于图状态机架构——支持条件分支、循环和持久状态,这些是 LangChain 原生的链式调用做不到的。

当然,你不一定非得用 LangGraph。LlamaIndex 的 Agent 模块或者自己用 Python 手写状态机也行。不过 LangGraph 在 2026 年已经是事实上的主流选择了,1.0 稳定版在不少企业中都有部署。

Q4:Agentic RAG 如何防止幻觉?效果比传统 RAG 好多少?

传统 RAG 的幻觉来自两个阶段:检索阶段拿到不相关的文档,生成阶段 LLM 编造文档里没有的信息。Agentic RAG 靠两道防线来应对——第一道是文档评分节点,在生成前就把不相关的文档过滤掉;第二道是幻觉检测节点,在生成后检查答案是否有文档支持。

研究数据显示,这种双重验证能把准确率提升约 14%。但需要注意的是,幻觉检测本身也依赖 LLM 的判断力,不是 100% 可靠的——在高风险场景中还是建议加入人工审核。

Q5:超大上下文窗口(100万 Token)会让 RAG 过时吗?

这大概是 2025-2026 年 AI 圈子里争论最多的话题之一了。

简短回答:不会。哪怕模型能一次处理 100 万 Token,把所有文档一股脑塞进去仍然有三个核心问题。一是注意力稀释——研究表明,当相关信息淹没在大量无关内容中时,模型准确性会明显下降。二是成本——处理百万 Token 的推理费用远高于只检索少量相关文档。三是实时性——上下文窗口只能处理已经放进去的信息,而 RAG 可以在查询的那一刻实时检索最新数据。

实际上,2026 年最靠谱的生产系统恰恰是将 RAG 和大上下文窗口结合起来用——用 RAG 精准定位相关内容,用大上下文窗口处理那些需要长文理解的任务。两者是互补的,不是替代关系。

关于作者 Editorial Team

Our team of expert writers and editors.