Agentic RAG：当 RAG 遇到 Agent，检索增强的下一步

用户提问
    ↓
┌─────────────────────────────────────┐
│         Agent Controller            │
│  ┌───────────────────────────────┐  │
│  │  1. 意图理解 & 任务规划       │  │
│  │  2. 工具选择 & 执行           │  │
│  │  3. 结果评估 & 决策           │  │
│  │  4. 迭代或终止                │  │
│  └───────────────────────────────┘  │
└─────────────────────────────────────┘
    ↓              ↓             ↓
┌────────┐   ┌──────────┐  ┌──────────┐
│向量检索 │   │知识图谱  │  │Web搜索   │
│工具    │   │查询工具  │  │工具      │
└────────┘   └──────────┘  └──────────┘
    ↓              ↓             ↓
┌─────────────────────────────────────┐
│         结果整合 & 验证              │
│  交叉验证 → 去重 → 排序 → 生成      │
└─────────────────────────────────────┘

from enum import Enum
from typing import List, Dict, Any

class AgentAction(Enum):
    RETRIEVE_VECTOR = "retrieve_vector"
    RETRIEVE_GRAPH = "retrieve_graph"
    WEB_SEARCH = "web_search"
    CALCULATE = "calculate"
    SYNTHESIZE = "synthesize"
    REFINE_QUERY = "refine_query"
    FINISH = "finish"

class AgenticRAG:
    def __init__(self, llm, vector_store, graph_store, web_searcher):
        self.llm = llm
        self.tools = {
            AgentAction.RETRIEVE_VECTOR: self._retrieve_vector,
            AgentAction.RETRIEVE_GRAPH: self._retrieve_graph,
            AgentAction.WEB_SEARCH: self._web_search,
            AgentAction.CALCULATE: self._calculate,
        }
        self.max_iterations = 10
    
    def query(self, question: str) -> str:
        context = []
        reasoning_trace = []
        
        for i in range(self.max_iterations):
            # Agent 决策：下一步做什么？
            decision = self._decide_action(question, context, reasoning_trace)
            
            reasoning_trace.append({
                "iteration": i,
                "thought": decision.thought,
                "action": decision.action.value,
                "action_input": decision.action_input
            })
            
            if decision.action == AgentAction.FINISH:
                # 生成最终答案
                return self._generate_answer(question, context, reasoning_trace)
            
            if decision.action == AgentAction.REFINE_QUERY:
                question = decision.action_input["refined_query"]
                continue
            
            # 执行工具
            tool_result = self.tools[decision.action](**decision.action_input)
            context.append(tool_result)
            
            # 评估结果质量
            evaluation = self._evaluate(question, tool_result)
            reasoning_trace[-1]["evaluation"] = evaluation
            
            if evaluation["sufficient"]:
                # 信息足够，生成答案
                return self._generate_answer(question, context, reasoning_trace)
        
        return self._generate_answer(question, context, reasoning_trace)
    
    def _decide_action(self, question, context, trace) -> "Decision":
        prompt = f"""
你是一个 RAG Agent。根据当前状态决定下一步行动。

问题：{question}

已收集的上下文：
{self._format_context(context)}

推理历史：
{self._format_trace(trace)}

可选行动：
1. retrieve_vector - 向量检索（语义相关信息）
2. retrieve_graph - 图谱查询（实体关系）
3. web_search - 网络搜索（实时信息）
4. calculate - 计算（数值处理）
5. refine_query - 优化查询
6. finish - 信息足够，生成答案

请输出 JSON：
{{"thought": "思考过程", "action": "行动名称", "action_input": {{...}}}}
"""
        result = self.llm.generate(prompt, response_format="json")
        return Decision(**result)

def iterative_retrieve(self, question: str, max_rounds: int = 3):
    """多轮迭代检索：每轮基于上一轮的结果深化检索"""
    
    all_context = []
    current_query = question
    
    for round_idx in range(max_rounds):
        # 检索
        results = self.vector_store.search(
            embed_model.encode(current_query),
            top_k=10
        )
        
        # 评估检索结果
        relevance = self._assess_relevance(question, results)
        
        if relevance["score"] > 0.8:
            all_context.extend(results)
            break
        
        # 生成 follow-up query
        current_query = self._generate_followup_query(
            question, results, all_context
        )
        
        all_context.extend(results)
    
    return all_context

def _generate_followup_query(self, original_query, current_results, past_context):
    prompt = f"""
原始问题：{original_query}

已检索到的信息：
{self._format_results(current_results)}

已有上下文：
{self._format_results(past_context)}

信息缺口分析：还有哪些信息需要检索？
请生成一个更精确的 follow-up 查询。
"""
    return self.llm.generate(prompt).strip()

def self_evaluate(self, question: str, answer: str, context: list) -> dict:
    prompt = f"""
评估以下回答的质量：

问题：{question}
回答：{answer}
参考上下文：{self._format_context(context)}

请从以下维度评估（0-1分）：
1. 准确性：回答是否与上下文一致？
2. 完整性：是否回答了问题的所有方面？
3. 引用性：关键论断是否有引用支撑？
4. 幻觉率：回答中是否有上下文不支持的内容？

输出 JSON：{{"accuracy": 0.0, "completeness": 0.0, "citation": 0.0, "hallucination": 0.0, "needs_retry": false, "reason": "..."}}
"""
    return self.llm.generate(prompt, response_format="json")

# 混合路由：简单问题走传统RAG，复杂问题走Agentic RAG
class HybridRAGRouter:
    def __init__(self):
        self.classifier = QueryClassifier()
        self.traditional_rag = TraditionalRAG(...)
        self.agentic_rag = AgenticRAG(...)
    
    def query(self, question: str):
        complexity = self.classifier.predict(question)
        
        if complexity < 0.3:
            # 简单问题
            return self.traditional_rag.query(question)
        else:
            # 复杂问题
            return self.agentic_rag.query(question)