Agentic RAG：当RAG遇到智能体的架构革命

Naive RAG (2023)  →  Advanced RAG (2024)  →  Agentic RAG (2025-2026)
  朴素检索            增强检索              智能体检索
  单次查询            多次优化              自主决策
  固定管道            模块化                动态规划

┌─────────────────────────────────────────────────────┐
│                  Agentic RAG Architecture            │
├─────────────────────────────────────────────────────┤
│                                                      │
│   ┌──────────┐    ┌──────────────────────────────┐  │
│   │  User     │───▶│      Agent Controller        │  │
│   │  Query    │    │  (LLM + Planning + Memory)   │  │
│   └──────────┘    └──────────┬───────────────────┘  │
│                              │                       │
│                    ┌─────────┼─────────┐             │
│                    ▼         ▼         ▼             │
│              ┌─────────┐ ┌───────┐ ┌─────────┐      │
│              │ Router   │ │ Query │ │ Evaluator│      │
│              │ (路由)   │ │Rewrite│ │ (评估器) │      │
│              └────┬────┘ └───┬───┘ └────┬────┘      │
│                   │          │          │            │
│         ┌─────────┼──────────┼──────────┘            │
│         ▼         ▼          ▼                       │
│   ┌──────────┐ ┌────────┐ ┌──────────┐ ┌─────────┐  │
│   │ Vector   │ │  Web   │ │  SQL     │ │  Graph  │  │
│   │ Store    │ │ Search │ │  DB      │ │  Store  │  │
│   └──────────┘ └────────┘ └──────────┘ └─────────┘  │
│                                                      │
└─────────────────────────────────────────────────────┘

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
import operator

class AgentState(TypedDict):
    question: str
    retrieved_docs: Annotated[List[str], operator.add]
    answer: str
    iteration: int
    sufficient: bool

def retrieve_node(state: AgentState) -> dict:
    """基于当前问题+已有信息生成查询并检索"""
    query = query_rewriter(state["question"], state["retrieved_docs"])
    docs = vector_store.similarity_search(query, k=5)
    return {"retrieved_docs": docs, "iteration": state["iteration"] + 1}

def evaluate_node(state: AgentState) -> dict:
    """评估检索结果是否足以回答问题"""
    if state["iteration"] >= 3:
        return {"sufficient": True}
    
    prompt = f"""基于以下检索到的文档，能否完整回答用户问题？
    
    问题: {state['question']}
    文档: {state['retrieved_docs']}
    
    如果信息充分回答"SUFFICIENT"，否则回答"INSUFFICIENT"并说明缺什么。"""
    
    response = llm.invoke(prompt)
    sufficient = "SUFFICIENT" in response
    return {"sufficient": sufficient}

def answer_node(state: AgentState) -> dict:
    """基于所有检索文档生成最终回答"""
    prompt = f"问题: {state['question']}\n\n参考资料:\n{state['retrieved_docs']}"
    return {"answer": llm.invoke(prompt)}

# 构建工作流
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("evaluate", evaluate_node)
workflow.add_node("answer", answer_node)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "evaluate")
workflow.add_conditional_edges(
    "evaluate",
    lambda state: "answer" if state["sufficient"] else "retrieve"
)
workflow.add_edge("answer", END)

agent = workflow.compile()

def route_query(state: AgentState) -> str:
    """根据问题类型选择检索路径"""
    routing_prompt = f"""
    分析以下问题，选择最合适的数据源：
    
    问题: {state['question']}
    
    选项:
    - "vector": 适合从内部文档库检索事实性信息
    - "web": 适合需要最新信息的时效性问题
    - "sql": 适合结构化数据查询
    - "graph": 适合关系推理和多方关联
    
    返回JSON: {{"route": "...", "reason": "..."}}
    """
    decision = llm.invoke(routing_prompt)
    return json.loads(decision)["route"]

def grade_documents(state: AgentState) -> dict:
    """对检索文档打分，过滤低质量结果"""
    graded_docs = []
    for doc in state["retrieved_docs"]:
        score_prompt = f"""评估文档与问题的相关性(0-10):
        问题: {state['question']}
        文档: {doc.page_content[:500]}
        只返回数字。"""
        score = int(llm.invoke(score_prompt).strip())
        if score >= 6:
            graded_docs.append(doc)
    
    # 如果所有文档都不相关，触发查询改写
    if not graded_docs:
        new_query = llm.invoke(f"改写这个查询以获得更好的检索结果: {state['question']}")
        return {"retrieved_docs": [], "question": new_query}
    
    return {"retrieved_docs": graded_docs}

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# 定义工具
@tool
def vector_search(query: str) -> str:
    """从内部知识库检索相关文档"""
    docs = vector_store.similarity_search(query, k=5)
    return "\n\n".join([d.page_content for d in docs])

@tool
def web_search(query: str) -> str:
    """从网络搜索最新信息"""
    results = search_api.search(query, num_results=5)
    return "\n\n".join([r["content"] for r in results])

@tool
def sql_query(question: str) -> str:
    """查询业务数据库"""
    sql = text2sql(question)
    return db.execute(sql)

# Agent决策循环
tools = [vector_search, web_search, sql_query]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)

def agent_decision(state):
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def should_continue(state):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# 构建Graph
graph = StateGraph(dict)
graph.add_node("agent", agent_decision)
graph.add_node("tools", ToolNode(tools))

graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")  # 工具执行后回到agent做下一轮决策

app = graph.compile()

# 执行
result = app.invoke({
    "messages": [HumanMessage(content="对比2026年Q1三大云厂商的AI服务收入")]
})

MAX_ITERATIONS = 5

def should_continue(state):
    if state.get("iteration", 0) >= MAX_ITERATIONS:
        return "finalize"
    # ... 正常决策

# 简单决策用小模型，复杂推理用大模型
def route_model(complexity: str):
    if complexity == "simple":
        return ChatOpenAI(model="gpt-4o-mini")
    return ChatOpenAI(model="gpt-4o")

from langchain_core.caches import InMemoryCache
from langchain_core.globals import set_llm_cache

set_llm_cache(InMemoryCache())
# 相同查询不会重复调用LLM