Adaptive RAG（自适应RAG）

本文介绍 Adaptive RAG（自适应 RAG）范式。

---

核心概念

Adaptive RAG 是一种动态调整检索和生成策略的 RAG 增强范式，旨在根据输入查询的复杂性、上下文需求和数据分布，智能优化检索范围、生成方式和资源分配。

核心思想：

• 动态适应性：不再采用固定的检索-生成流程，而是根据任务需求调整策略
• 资源效率：在简单查询时减少计算开销，在复杂查询时增强检索和推理能力
• 上下文感知：结合用户历史交互、领域知识等优化检索和生成

---

与传统 RAG 的区别

特性	传统 RAG	Adaptive RAG
检索策略	固定	动态调整
复杂度处理	统一处理	区分简单/复杂查询
资源分配	固定分配	按需分配
上下文利用	单次查询	利用历史交互

---

自适应策略

根据问题类型采取不同策略：

问题类型	策略
简单事实性问题	直接生成，无需检索
需要背景知识	单次检索 + 生成
复杂多跳问题	迭代检索 + 推理
领域专业问题	增强检索 + 专家知识

---

实战案例： RAG 工作流

基于 LangGraph 的增强 RAG 系统，支持问题路由、文档检索、生成评估和 Web 搜索。

系统架构

特性	说明
问题路由	自动判断使用 RAG 还是 Web 搜索
文档检索	从向量数据库 Milvus 检索相关文档
文档评分	LLM 评估检索文档与问题的相关性
查询优化	将问题重写为更适合检索的版本
生成评估	幻觉检测 + 答案质量评估
Web 搜索	当知识库无相关内容时使用 Tavily 搜索

流程图

核心模块

模块	功能
route_question	判断问题走 RAG 还是 Web 搜索
retrieve	从 Milvus 检索文档
grade_documents	过滤不相关文档
generate	基于文档生成回答
transform_query	优化用户问题
web_search	执行网络搜索

Chain 功能

Chain	功能
question_router_chain	LLM 判断数据源
retrieval_grader_chain	评估文档相关性
hallucination_grader_chain	检测生成内容是否基于文档
answer_grader_chain	评估答案是否解决用户问题

---

关键代码

1. 状态定义

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

class GraphState(TypedDict):
    question: str              # 用户问题
    transform_count: int      # 查询优化次数
    generation: str           # LLM 生成的回答
    documents: List[Document] # 检索到的文档列表

2. 问题路由 Chain

from typing import Literal
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

class RouteQuery(BaseModel):
    datasource: Literal["vectorstore", "web_search"]

structured_llm_router = llm.with_structured_output(RouteQuery)

system = """向量知识库包含半导体材料、芯片制造、光刻技术相关文档。
其他问题使用网络搜索。"""

route_prompt = ChatPromptTemplate.from_messages([
    ("system", system),
    ("human", "{question}"),
])

question_router_chain = route_prompt | structured_llm_router

3. 文档评分 Chain

from pydantic import BaseModel, Field

class GradeDocuments(BaseModel):
    binary_score: str = Field(description="文档是否与问题相关，yes/no")

structured_llm_grader = llm.with_structured_output(GradeDocuments)

system = """评估检索文档与用户问题的相关性。
包含相关关键词或语义含义则评为相关。"""

grade_prompt = ChatPromptTemplate.from_messages([
    ("system", system),
    ("human", "Retrieved document: \n\n {document} \n\n User question: {question}"),
])

retrieval_grader_chain = grade_prompt | structured_llm_grader

4. 幻觉检测 Chain

class GradeHallucinations(BaseModel):
    binary_score: str = Field(description="回答是否基于事实，yes/no")

structured_llm_grader = llm.with_structured_output(GradeHallucinations)

system = """评估生成内容是否基于检索事实。
yes 表示回答基于给定事实。"""

hallucination_prompt = ChatPromptTemplate.from_messages([
    ("system", system),
    ("human", "事实集: \n\n {documents} \n\n 生成内容: {generation}"),
])

hallucination_grader_chain = hallucination_prompt | structured_llm_grader

5. Web 搜索节点

from langchain_core.documents import Document
from langchain_tavily import TavilySearch

web_search_tool = TavilySearch(max_results=5, topic="general")

def web_search(state):
    question = state["question"]
    docs = web_search_tool.invoke({"query": question})

    # 解析搜索结果
    if isinstance(docs, dict) and "results" in docs:
        web_results = "\n".join([d["content"] for d in docs["results"]])
    else:
        web_results = str(docs)

    web_doc = Document(page_content=web_results)
    return {"documents": web_doc, "question": question}

6. 构建工作流图

from langgraph.constants import START, END
from langgraph.graph import StateGraph

workflow = StateGraph(GraphState)

# 添加节点
workflow.add_node("web_search", web_search)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("transform_query", transform_query)

# 路由判断
workflow.add_conditional_edges(
    START,
    route_question,
    {"web_search": "web_search", "vectorstore": "retrieve"}
)

# Web搜索后直接生成
workflow.add_edge("web_search", "generate")

# 检索后评估文档相关性
workflow.add_edge("retrieve", "grade_documents")

# 文档评估后条件分支
workflow.add_conditional_edges(
    'grade_documents',
    decide_to_generate
)

# 生成结果评估条件分支
workflow.add_conditional_edges(
    "generate",
    grade_generation_v_documents_and_question,
    {
        "not supported": "generate",
        "useful": END,
        "not useful": "transform_query",
    },
)

# 查询优化后重新检索
workflow.add_edge("transform_query", "retrieve")

graph = workflow.compile()

流程决策逻辑

问题输入
    │
    ▼
路由判断 (route_question)
    │
    ├── 半导体/芯片领域 ──► 向量检索 (retrieve)
    │                            │
    │                            ▼
    │                      文档评分 (grade_documents)
    │                            │
    │         ┌──────────────────┼──────────────────┐
    │         ▼                  ▼                  ▼
    │    有相关文档          无相关文档          无相关文档
    │        │              且<2次            且>=2次
    │        ▼                  ▼                  ▼
    │    生成回答          优化查询           Web搜索
    │    (generate)      (transform_query)   (web_search)
    │        │                  │                  │
    │        ▼                  ▼                  ▼
    │    质量评估          重新检索            生成回答
    │        │                  │                  │
    │    ┌──┼──┐               │                   │
    │    ▼  ▼  ▼               │                   │
    │ useful  not  not         │                   │
    │       useful supported    │                   │
    │    (结束)    (重试)       │                   │
    │                      重新检索                │
    │
    └── 其他领域 ──► Web搜索 (web_search)
                      │
                      ▼
                生成回答 (generate)
                      │
                      ▼
                质量评估
                      │
                      ├── useful ──► 结束
                      └── 其他 ──► 重试

运行结果示例

Web 搜索流程：

用户：open claw 是什么
route_question: 路由到 web 搜索
web_search: 执行网络搜索
grade_generation: 幻觉检测通过
grade_generation: 答案质量通过
生成: 'OpenClaw是一个开源的AI执行引擎...'

RAG 检索流程：

用户：EUV 光刻机是什么
route_question: 路由到 RAG 系统
retrieve: 检索文档
grade_documents: 过滤相关文档（保留2/4个）
generate: 生成回答
grade_generation: 幻觉检测通过 + 答案质量通过
生成: 'EUV光刻机（极紫外光刻机）是一种用于半导体制造的...'