Agentic RAG：让 Agent 驱动检索增强生成的架构演进与工程实践

┌────────────────────────────────────────────────────────────────┐
│                  Agentic RAG 架构分类                           │
│                                                                │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐ │
│  │ 单 Agent     │  │ 多 Agent     │  │ 层级式 Agent         │ │
│  │ Router 架构  │  │ 协作架构      │  │ 层级架构             │ │
│  ├──────────────┤  ├──────────────┤  ├──────────────────────┤ │
│  │ • 路由决策    │  │ • 并行检索    │  │ • 总控规划 Agent     │ │
│  │ • 策略选择    │  │ • 领域分工    │  │ • 子任务 Agent       │ │
│  │ • 单源查询    │  │ • 结果聚合    │  │ • 执行层 Agent       │ │
│  └──────────────┘  └──────────────┘  └──────────────────────┘ │
│                                                                │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐ │
│  │ Corrective   │  │ Adaptive     │  │ Graph-Based          │ │
│  │ RAG 架构     │  │ RAG 架构     │  │ RAG 架构             │ │
│  ├──────────────┤  ├──────────────┤  ├──────────────────────┤ │
│  │ • 质量评估    │  │ • 查询复杂度 │  │ • 知识图谱遍历       │ │
│  │ • 自动重检索  │  │   动态判断    │  │ • Agent-G 框架       │ │
│  │ • 纠错循环    │  │ • 策略自适应  │  │ • GeAR 增强          │ │
│  └──────────────┘  └──────────────┘  └──────────────────────┘ │
└────────────────────────────────────────────────────────────────┘

用户查询 → Agent 分析意图 → 选择检索策略 → 执行检索 → 评估结果 → 生成/重检索

用户查询 → 协调 Agent 分解子任务 → 分发给领域 Agent → 各自检索 → 结果聚合 → 综合生成

检索结果 → 质量评估 Agent → 评分 ≥ 阈值 → 生成答案
                          → 评分 < 阈值 → 查询改写 → 重新检索 → 再次评估

# 伪代码：Prompt Chaining 模式
def agentic_rag_chain(query: str) -> str:
    sub_queries = planner_agent.decompose(query)
    context_parts = []
    for sq in sub_queries:
        docs = retriever.search(sq)
        refined = reflector_agent.evaluate(docs, sq)
        context_parts.append(refined)
    return generator_agent.synthesize(query, context_parts)

                    ┌→ 技术文档检索 Agent ─┐
用户查询 → 路由 Agent ├→ 财务数据检索 Agent ─┤ → 结果聚合 → 生成
                    └→ 外部 API 调用 Agent ─┘

生成结果 → 评估 Agent 评判 → 合格 → 输出
                          → 不合格 → 反馈优化 → 重新生成 → 再次评估

class AgenticPlanner:
    def plan(self, query: str) -> RetrievalPlan:
        query_type = self.classify_query(query)
        if query_type == "simple_fact":
            return RetrievalPlan(steps=[SingleRetrieval(query)])
        elif query_type == "multi_hop":
            sub_queries = self.decompose(query)
            return RetrievalPlan(
                steps=[IterativeRetrieval(sq) for sq in sub_queries],
                strategy="sequential_with_reflection"
            )
        elif query_type == "comparative":
            return RetrievalPlan(
                steps=[ParallelRetrieval(query)],
                strategy="parallel_then_merge"
            )

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from typing import TypedDict, Annotated, List
import operator

class AgentState(TypedDict):
    query: str
    search_strategy: str
    retrieved_docs: List[str]
    quality_score: float
    iteration: int
    answer: str

llm = ChatOpenAI(model="gpt-4o", temperature=0)
vectorstore = Chroma(embedding_function=OpenAIEmbeddings())

def classify_query(state: AgentState) -> AgentState:
    response = llm.invoke(f"""
    分析以下查询的复杂度，输出分类结果：
    查询: {state["query"]}
    分类: simple / moderate / complex
    """)
    strategy = "direct" if "simple" in response.content.lower() else \
               "rewrite" if "moderate" in response.content.lower() else \
               "decompose"
    return {**state, "search_strategy": strategy, "iteration": 0}

def retrieve(state: AgentState) -> AgentState:
    query = state["query"]
    if state["search_strategy"] == "rewrite":
        response = llm.invoke(f"改写以下查询以提高检索质量：{query}")
        query = response.content
    docs = vectorstore.similarity_search(query, k=5)
    return {**state, "retrieved_docs": [d.page_content for d in docs]}

def evaluate(state: AgentState) -> AgentState:
    docs_text = "\n".join(state["retrieved_docs"])
    response = llm.invoke(f"""
    评估以下检索结果对查询 '{state["query"]}' 的回答质量。
    输出 1-10 的分数和简要理由。
    检索结果: {docs_text}
    """)
    score = float(response.content.split()[0])
    return {**state, "quality_score": score, "iteration": state["iteration"] + 1}

def should_retry(state: AgentState) -> str:
    if state["quality_score"] >= 7 or state["iteration"] >= 3:
        return "generate"
    return "retrieve"

def generate(state: AgentState) -> AgentState:
    docs_text = "\n".join(state["retrieved_docs"])
    response = llm.invoke(f"""
    基于以下参考资料回答用户问题。
    问题: {state["query"]}
    参考资料: {docs_text}
    """)
    return {**state, "answer": response.content}

graph = StateGraph(AgentState)
graph.add_node("classify", classify_query)
graph.add_node("retrieve", retrieve)
graph.add_node("evaluate", evaluate)
graph.add_node("generate", generate)

graph.set_entry_point("classify")
graph.add_edge("classify", "retrieve")
graph.add_edge("retrieve", "evaluate")
graph.add_conditional_edges("evaluate", should_retry, {
    "retrieve": "retrieve",
    "generate": "generate"
})
graph.add_edge("generate", END)

app = graph.compile()
result = app.invoke({"query": "对比分析向量数据库和图数据库在 RAG 场景下的性能差异"})