Metaphor

Agentic RAG（智能体检索增强生成）

21分钟阅读

概述

Agentic RAG（Agentic Retrieval-Augmented Generation）是将自主AI智能体嵌入RAG管道的创新范式，相比传统RAG的静态工作流，Agentic RAG能够进行动态决策、迭代推理和多智能体协作。1

传统RAG系统的局限性：

  • 固定的一次性检索流程
  • 缺乏自适应检索策略
  • 无法处理复杂的多步骤推理
  • 无法根据中间结果调整查询

Agentic RAG通过引入智能体系统，将RAG从”查询-检索-生成”的线性流程转变为动态、迭代、可协作的智能系统。

1. 从传统RAG到Agentic RAG

1.1 演进对比

维度传统RAGAgentic RAG
工作流静态、线性动态、迭代
检索策略固定自适应
决策能力无自主决策
错误处理有限自我纠正
复杂任务困难原生支持
多源检索顺序并行/协作

1.2 核心范式转变

传统RAG：                              Agentic RAG：
                                       
用户查询                                 用户查询
    ↓                                       ↓
固定检索                              智能体分析查询
    ↓                                   ↓
返回Top-K                            动态规划策略
    ↓                                   ↓
一次性生成                            条件触发检索
                                        ↓
                                   检查结果质量
                                        ↓
                                   必要时迭代/补充
                                        ↓
                                   多源协作（如需要）
                                        ↓
                                        生成最终答案

2. Agentic Patterns（智能体模式）

2.1 反思（Reflection）

智能体能够评估自身输出和中间结果，进行自我反思：

class ReflectionAgent:
    """
    反思智能体：评估检索和生成结果的质量
    """
    
    def __init__(self, llm):
        self.llm = llm
    
    def evaluate_retrieval(self, query, retrieved_docs):
        """
        评估检索结果的相关性
        """
        evaluation_prompt = f"""
评估以下检索结果对查询的相关性。
 
查询：{query}
 
检索结果：
{retrieved_docs}
 
请评估：
1. 每个文档与查询的相关性（1-5分）
2. 是否有重要信息被遗漏？
3. 是否需要补充检索？
 
格式：
{{
    "relevance_scores": [分数列表],
    "missing_info": ["遗漏信息列表"],
    "needs_more_retrieval": true/false,
    "reasoning": "评估理由"
}}
"""
        response = self.llm.generate(evaluation_prompt)
        return parse_json(response)
    
    def evaluate_generation(self, query, answer, context):
        """
        评估生成答案的质量
        """
        evaluation_prompt = f"""
评估以下答案的质量。
 
问题：{query}
答案：{answer}
上下文：{context}
 
请评估：
1. 答案是否准确回答了问题？
2. 答案是否基于提供的上下文？
3. 是否有幻觉或错误信息？
4. 答案是否完整？
 
格式：
{{
    "accuracy": 分数(1-5),
    "groundedness": 分数(1-5),
    "completeness": 分数(1-5),
    "issues": ["问题列表"],
    "needs_revision": true/false
}}
"""
        response = self.llm.generate(evaluation_prompt)
        return parse_json(response)

2.2 规划（Planning）

智能体能够将复杂任务分解为子任务并制定执行计划：

class PlanningAgent:
    """
    规划智能体：分解复杂查询为可执行的子任务
    """
    
    def __init__(self, llm):
        self.llm = llm
    
    def decompose_query(self, query):
        """
        将复杂查询分解为子任务
        """
        planning_prompt = f"""
将以下复杂查询分解为可执行的子任务。
 
查询：{query}
 
子任务类型：
- retrieve_specific: 检索特定信息
- retrieve_related: 检索相关背景
- analyze: 分析信息
- synthesize: 综合多个来源
- verify: 验证信息准确性
- clarify: 需要用户澄清
 
输出格式（JSON）：
{{
    "task_type": "查询类型：simple/complex/multi-hop",
    "subtasks": [
        {{
            "id": 1,
            "type": "子任务类型",
            "description": "子任务描述",
            "depends_on": [],
            "priority": 1-5
        }},
        ...
    ],
    "execution_order": [子任务ID列表，按顺序执行]
}}
 
确保：
1. 子任务可以独立或按依赖顺序执行
2. 关键信息不会被遗漏
3. 优先级合理分配
"""
        response = self.llm.generate(planning_prompt)
        return parse_json(response)
    
    def create_execution_plan(self, subtasks):
        """
        创建执行计划
        """
        # 按依赖和优先级排序
        executed = set()
        plan = []
        
        while len(executed) < len(subtasks):
            for task in subtasks:
                if (task["id"] in executed or 
                    not all(dep in executed for dep in task.get("depends_on", []))):
                    continue
                
                plan.append(task)
                executed.add(task["id"])
        
        return plan

2.3 工具使用（Tool Use）

智能体能够调用各种工具扩展能力：

class ToolUsingAgent:
    """
    工具使用智能体：调用外部工具增强RAG
    """
    
    def __init__(self, llm):
        self.llm = llm
        self.tools = {
            "web_search": self.web_search,
            "calculator": self.calculate,
            "code_executor": self.execute_code,
            "api_caller": self.call_api,
            "knowledge_base": self.search_knowledge_base
        }
    
    def select_tools(self, query, context):
        """
        根据查询和上下文选择合适的工具
        """
        selection_prompt = f"""
根据以下查询和当前上下文，选择需要调用的工具。
 
查询：{query}
 
当前上下文：
{context}
 
可用工具：
- web_search: 网络搜索，获取最新信息
- calculator: 数学计算
- code_executor: 执行代码
- api_caller: 调用外部API
- knowledge_base: 搜索知识库
 
输出格式（JSON）：
{{
    "selected_tools": [
        {{
            "tool": "工具名",
            "reason": "选择理由",
            "parameters": {{"参数": "值"}}
        }}
    ],
    "reasoning": "整体选择逻辑"
}}
"""
        response = self.llm.generate(selection_prompt)
        return parse_json(response)
    
    def execute_tools(self, tool_calls):
        """
        执行工具调用
        """
        results = []
        
        for tool_call in tool_calls:
            tool_name = tool_call["tool"]
            params = tool_call.get("parameters", {})
            
            if tool_name in self.tools:
                result = self.tools[tool_name](**params)
                results.append({
                    "tool": tool_name,
                    "result": result,
                    "success": True
                })
            else:
                results.append({
                    "tool": tool_name,
                    "error": "Tool not found",
                    "success": False
                })
        
        return results

2.4 记忆（Memory）

智能体能够维护和使用记忆：

class MemoryAgent:
    """
    记忆智能体：管理短期和长期记忆
    """
    
    def __init__(self):
        # 短期记忆：当前会话上下文
        self.short_term_memory = []
        
        # 长期记忆：历史交互经验
        self.long_term_memory = {}
        
        # 工作内存：当前处理中的信息
        self.working_memory = {}
    
    def add_to_working(self, key, value):
        """添加到工作内存"""
        self.working_memory[key] = {
            "value": value,
            "timestamp": datetime.now(),
            "access_count": 0
        }
    
    def access_working(self, key):
        """访问工作内存"""
        if key in self.working_memory:
            self.working_memory[key]["access_count"] += 1
            return self.working_memory[key]["value"]
        return None
    
    def retrieve_from_long_term(self, query):
        """
        从长期记忆中检索相关信息
        """
        query_embedding = embed_text(query)
        
        # 语义搜索长期记忆
        similarities = {}
        for memory_id, memory in self.long_term_memory.items():
            sim = cosine_similarity(
                query_embedding,
                memory["embedding"]
            )
            if sim > 0.7:
                similarities[memory_id] = sim
        
        # 返回最相关的记忆
        sorted_memories = sorted(
            similarities.items(),
            key=lambda x: x[1],
            reverse=True
        )
        
        return [
            self.long_term_memory[mid]["content"]
            for mid, _ in sorted_memories[:5]
        ]
    
    def consolidate_to_long_term(self, experience):
        """
        将重要经验存入长期记忆
        """
        memory_id = generate_id()
        
        self.long_term_memory[memory_id] = {
            "content": experience,
            "embedding": embed_text(experience),
            "timestamp": datetime.now(),
            "importance": self._assess_importance(experience)
        }
        
        # 如果长期记忆过大，清理不重要记忆
        if len(self.long_term_memory) > 1000:
            self._prune_low_importance()

3. Agentic RAG架构

3.1 多层架构

┌──────────────────────────────────────────────────────────────────┐
│                        应用层                                      │
│            (用户界面、任务提交、结果展示)                          │
├──────────────────────────────────────────────────────────────────┤
│                       协调层                                      │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐ │
│  │  任务规划   │  │   流程控制  │  │    异常处理与恢复       │ │
│  │  Agent     │  │   Agent     │  │    Agent                │ │
│  └─────────────┘  └─────────────┘  └─────────────────────────┘ │
├──────────────────────────────────────────────────────────────────┤
│                       执行层                                      │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐ │
│  │  检索       │  │   生成      │  │    工具使用             │ │
│  │  Agent      │  │   Agent     │  │    Agent                │ │
│  └─────────────┘  └─────────────┘  └─────────────────────────┘ │
├──────────────────────────────────────────────────────────────────┤
│                       记忆层                                      │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐ │
│  │  短期记忆   │  │   长期记忆  │  │    工作内存             │ │
│  └─────────────┘  └─────────────┘  └─────────────────────────┘ │
├──────────────────────────────────────────────────────────────────┤
│                       数据层                                      │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐ │
│  │  向量数据库 │  │  知识图谱   │  │    结构化数据库         │ │
│  └─────────────┘  └─────────────┘  └─────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘

3.2 核心执行流程

class AgenticRAGSystem:
    """
    Agentic RAG系统核心实现
    """
    
    def __init__(self, llm, vector_db, tools=None):
        self.llm = llm
        self.vector_db = vector_db
        
        # 初始化各类型Agent
        self.planner = PlanningAgent(llm)
        self.retriever = RetrievalAgent(llm, vector_db)
        self.generator = GeneratorAgent(llm)
        self.reflector = ReflectionAgent(llm)
        self.memory = MemoryAgent()
        
        # 可用工具
        self.tools = tools or {}
    
    def process_query(self, query, max_iterations=5):
        """
        处理用户查询的完整流程
        """
        # 步骤1：查询理解与规划
        query_plan = self.planner.decompose_query(query)
        
        if query_plan["task_type"] == "simple":
            # 简单查询：单次检索-生成
            return self._simple_rag(query)
        else:
            # 复杂查询：多步骤Agentic处理
            return self._complex_rag(query, query_plan, max_iterations)
    
    def _simple_rag(self, query):
        """简单RAG流程"""
        # 检索
        docs = self.retriever.search(query, top_k=5)
        
        # 生成
        answer = self.generator.generate(query, docs)
        
        # 评估
        evaluation = self.reflector.evaluate_generation(
            query, answer, docs
        )
        
        if evaluation.get("needs_revision"):
            # 如果需要修订，增加更多上下文
            docs = self.retriever.search(
                query, top_k=10, include_subqueries=True
            )
            answer = self.generator.generate(query, docs)
        
        return answer
    
    def _complex_rag(self, query, plan, max_iterations):
        """复杂Agentic RAG流程"""
        # 创建执行计划
        execution_plan = self.planner.create_execution_plan(plan["subtasks"])
        
        # 迭代执行
        iteration = 0
        all_context = []
        
        while iteration < max_iterations:
            iteration += 1
            
            # 获取当前待执行任务
            current_task = self._get_next_task(execution_plan, all_context)
            
            if current_task is None:
                break  # 所有任务完成
            
            # 执行任务
            result = self._execute_task(current_task, all_context, query)
            
            # 添加到上下文
            all_context.append({
                "task": current_task,
                "result": result,
                "iteration": iteration
            })
            
            # 评估中间结果
            evaluation = self.reflector.evaluate_retrieval(
                current_task["description"],
                result
            )
            
            # 如果需要补充检索
            if evaluation.get("needs_more_retrieval"):
                additional_context = self.retriever.search(
                    evaluation.get("missing_info", []),
                    top_k=5
                )
                all_context.append({
                    "type": "additional_retrieval",
                    "result": additional_context
                })
        
        # 最终生成
        final_answer = self.generator.generate(
            query,
            self._consolidate_context(all_context)
        )
        
        # 评估最终答案
        final_evaluation = self.reflector.evaluate_generation(
            query, final_answer, all_context
        )
        
        if final_evaluation.get("needs_revision"):
            # 修订策略
            final_answer = self._revise_answer(
                query, final_answer, all_context, final_evaluation
            )
        
        # 存储重要经验到长期记忆
        self._store_experience(query, all_context, final_answer)
        
        return final_answer
    
    def _execute_task(self, task, context, original_query):
        """执行单个任务"""
        task_type = task["type"]
        
        if task_type == "retrieve_specific":
            # 检索特定信息
            return self.retriever.search(
                task["description"],
                top_k=5,
                filters=self._extract_filters(task)
            )
        
        elif task_type == "retrieve_related":
            # 检索相关背景
            return self.retriever.search(
                task["description"],
                top_k=8,
                include_expanded=True
            )
        
        elif task_type == "analyze":
            # 分析信息
            return self.generator.analyze(
                task["description"],
                context
            )
        
        elif task_type == "verify":
            # 验证信息
            return self._verify_information(
                task["description"],
                context
            )
        
        elif task_type == "use_tool":
            # 使用工具
            tool_name = task.get("tool")
            if tool_name in self.tools:
                return self.tools[tool_name](**task.get("parameters", {}))
        
        return None

4. 迭代式推理与检索

4.1 主动检索增强生成（Active RAG）

class ActiveRetrievalRAG:
    """
    主动检索：基于生成内容动态触发额外检索
    """
    
    def __init__(self, llm, retriever):
        self.llm = llm
        self.retriever = retriever
    
    def generate_with_active_retrieval(self, query, max_retrievals=3):
        """
        主动检索增强生成
        
        流程：
        1. 初始检索
        2. 生成答案草稿
        3. 检测知识缺口
        4. 主动补充检索
        5. 修订答案
        """
        # 步骤1：初始检索
        initial_docs = self.retriever.search(query, top_k=5)
        
        # 步骤2：生成草稿
        draft = self._generate_draft(query, initial_docs)
        
        # 步骤3-4：迭代补充检索
        current_docs = initial_docs
        current_draft = draft
        
        for i in range(max_retrievals):
            # 检测知识缺口
            gaps = self._detect_knowledge_gaps(query, current_draft)
            
            if not gaps:
                break  # 没有更多缺口
            
            # 主动检索
            for gap in gaps:
                additional_docs = self.retriever.search(gap, top_k=3)
                current_docs.extend(additional_docs)
            
            # 修订答案
            current_draft = self._revise_draft(
                query, current_draft, current_docs
            )
        
        return current_draft
    
    def _detect_knowledge_gaps(self, query, draft):
        """
        检测答案中的知识缺口
        """
        gap_detection_prompt = f"""
分析以下答案草稿，检测其中存在的知识缺口。
 
问题：{query}
 
答案草稿：
{draft}
 
请识别：
1. 答案中提到的但未解释清楚的概念
2. 答案中暗示但未明确回答的信息需求
3. 可能需要补充的细节
 
格式（JSON）：
{{
    "gaps": [
        {{
            "gap": "缺口描述",
            "importance": "high/medium/low",
            "retrieval_query": "建议的检索查询"
        }}
    ]
}}
"""
        response = self.llm.generate(gap_detection_prompt)
        result = parse_json(response)
        return result.get("gaps", [])
    
    def _generate_draft(self, query, docs):
        """生成答案草稿"""
        context = self._format_context(docs)
        
        prompt = f"""
基于以下上下文，回答问题。如果信息不足，请基于已知信息给出最佳答案。
 
上下文：
{context}
 
问题：{query}
 
答案：
"""
        return self.llm.generate(prompt)
    
    def _revise_draft(self, query, draft, docs):
        """修订答案草稿"""
        context = self._format_context(docs)
        
        prompt = f"""
基于更全面的上下文，修订以下答案。
 
原始问题：{query}
 
原始答案：
{draft}
 
补充上下文：
{context}
 
任务：
1. 整合补充信息
2. 修正可能的错误
3. 使答案更加完整和准确
 
修订后的答案：
"""
        return self.llm.generate(prompt)

4.2 自我纠正机制

class SelfCorrectionAgent:
    """
    自我纠正智能体
    """
    
    def __init__(self, llm):
        self.llm = llm
    
    def detect_errors(self, answer, context, query):
        """
        检测答案中的错误
        """
        error_detection_prompt = f"""
仔细审查以下答案，检测其中的错误。
 
问题：{query}
答案：{answer}
上下文：{context}
 
请检测：
1. 事实性错误（与上下文不符）
2. 逻辑错误（推理过程有问题）
3. 幻觉（上下文未支持的信息）
4. 不完整的部分
 
格式（JSON）：
{{
    "factual_errors": [
        {{"text": "错误文本", "reason": "错误原因"}}
    ],
    "logical_errors": [
        {{"text": "错误文本", "reason": "错误原因"}}
    ],
    "hallucinations": [
        {{"text": "幻觉内容", "reason": "为何是幻觉"}}
    ],
    "incomplete_parts": [
        {{"topic": "不完整的主题", "what_missing": "缺失什么"}}
    ],
    "overall_quality": "excellent/good/poor"
}}
"""
        response = self.llm.generate(error_detection_prompt)
        return parse_json(response)
    
    def correct_errors(self, answer, errors, context, query):
        """
        纠正检测到的错误
        """
        correction_prompt = f"""
请修正以下答案中的错误。
 
问题：{query}
 
原始答案：{answer}
 
检测到的错误：
{errors}
 
上下文：
{context}
 
请生成修正后的答案，确保：
1. 修正所有事实性错误
2. 修正所有逻辑错误
3. 移除幻觉内容
4. 补充不完整部分
 
修正后的答案：
"""
        return self.llm.generate(correction_prompt)
    
    def verify_correction(self, corrected_answer, context, query):
        """
        验证修正后的答案
        """
        verification_prompt = f"""
验证以下修正后的答案是否正确。
 
问题：{query}
修正后答案：{corrected_answer}
上下文：{context}
 
请确认：
1. 答案是否准确回答了问题？
2. 答案是否与上下文一致？
3. 是否还存在其他问题？
 
回答：yes/no，并简述理由。
"""
        response = self.llm.generate(verification_prompt)
        return "yes" in response.lower()

5. 多智能体协作

5.1 协作框架

class MultiAgentCollaboration:
    """
    多智能体协作框架
    """
    
    def __init__(self, llm, retriever):
        self.llm = llm
        self.retriever = retriever
        
        # 定义专业智能体
        self.agents = {
            "researcher": ResearcherAgent(llm, retriever),
            "analyst": AnalystAgent(llm),
            "critic": CriticAgent(llm),
            "synthesizer": SynthesizerAgent(llm)
        }
    
    def collaborative_query(self, query):
        """
        多智能体协作处理查询
        """
        # 阶段1：研究阶段
        research_result = self.agents["researcher"].research(query)
        
        # 阶段2：分析阶段
        analysis_result = self.agents["analyst"].analyze(
            query, research_result
        )
        
        # 阶段3：批评阶段
        critique = self.agents["critic"].critique(
            query, research_result, analysis_result
        )
        
        # 阶段4：综合阶段
        if critique.get("needs_revision"):
            # 如果有批评，需要修订
            research_result = self.agents["researcher"].research(
                query, focus_areas=critique.get("focus_areas")
            )
            analysis_result = self.agents["analyst"].analyze(
                query, research_result
            )
        
        # 最终综合
        final_answer = self.agents["synthesizer"].synthesize(
            query, research_result, analysis_result
        )
        
        return final_answer
 
 
class ResearcherAgent:
    """研究智能体：负责信息检索"""
    
    def __init__(self, llm, retriever):
        self.llm = llm
        self.retriever = retriever
    
    def research(self, query, focus_areas=None):
        """
        执行研究任务
        """
        # 分解研究主题
        research_plan = self._create_research_plan(query, focus_areas)
        
        results = []
        for aspect in research_plan:
            docs = self.retriever.search(
                aspect,
                top_k=5,
                include_metadata=True
            )
            results.append({
                "aspect": aspect,
                "documents": docs,
                "key_findings": self._extract_findings(docs)
            })
        
        return {
            "query": query,
            "aspects": results,
            "overall_summary": self._summarize(results)
        }
    
    def _create_research_plan(self, query, focus_areas=None):
        """创建研究计划"""
        # ...
 
 
class CriticAgent:
    """批评智能体：评估研究质量"""
    
    def __init__(self, llm):
        self.llm = llm
    
    def critique(self, query, research, analysis):
        """
        批评研究结果
        """
        critique_prompt = f"""
评估以下研究和分析的质量和完整性。
 
查询：{query}
 
研究结果：
{research}
 
分析结果：
{analysis}
 
请评估：
1. 研究是否覆盖了查询的所有重要方面？
2. 分析是否有逻辑问题？
3. 是否有重要遗漏？
4. 是否需要进一步调查特定领域？
 
格式（JSON）：
{{
    "coverage_score": 1-10,
    "logic_score": 1-10,
    "completeness_score": 1-10,
    "needs_revision": true/false,
    "focus_areas": ["需要重点关注的领域"],
    "main_critiques": ["主要批评意见"]
}}
"""
        response = self.llm.generate(critique_prompt)
        return parse_json(response)

6. 应用场景

6.1 企业知识管理

场景：企业知识库问答系统

Agentic RAG优势：
├── 理解复杂业务逻辑
├── 多文档关联分析
├── 动态选择检索策略
├── 自我纠正错误答案
└── 持续学习和改进

6.2 研究助手

场景：科研文献分析与综合

Agentic RAG能力：
├── 跨论文关系发现
├── 研究趋势分析
├── 假设验证
├── 文献对比评估
└── 生成研究综述

6.3 法律文档分析

场景：法律案例研究

Agentic RAG能力：
├── 相关判例检索
├── 法律条文关联
├── 论证逻辑分析
├── 风险评估
└── 生成法律意见书

7. 评估框架

7.1 Agentic RAG评估指标

AGENTIC_RAG_METRICS = {
    "retrieval_quality": {
        "precision": "检索结果精确度",
        "recall": "检索结果召回率",
        "diversity": "检索结果多样性",
        "relevance": "检索结果相关性"
    },
    "reasoning_quality": {
        "task_decomposition": "任务分解质量",
        "execution_order": "执行顺序合理性",
        "error_detection": "错误检测能力",
        "self_correction": "自我纠正有效性"
    },
    "generation_quality": {
        "faithfulness": "答案忠于检索内容",
        "completeness": "答案完整性",
        "coherence": "答案连贯性",
        "accuracy": "答案准确性"
    },
    "agent_capabilities": {
        "tool_usage": "工具使用正确性",
        "memory_utilization": "记忆利用效率",
        "collaboration": "多智能体协作效果",
        "adaptability": "策略适应能力"
    }
}

7.2 评估实现

def evaluate_agentic_rag(system, test_queries, ground_truth):
    """
    评估Agentic RAG系统
    """
    results = []
    
    for query, gt in zip(test_queries, ground_truth):
        # 执行查询
        answer = system.process_query(query)
        
        # 计算各维度指标
        retrieval_metrics = evaluate_retrieval(
            query, system.last_retrieved_docs
        )
        
        agent_metrics = evaluate_agent_behavior(
            query, system.last_agent_actions
        )
        
        generation_metrics = evaluate_generation(
            query, answer, gt
        )
        
        results.append({
            "query": query,
            "answer": answer,
            "retrieval": retrieval_metrics,
            "agent": agent_metrics,
            "generation": generation_metrics,
            "overall": compute_overall_score(
                retrieval_metrics,
                agent_metrics,
                generation_metrics
            )
        })
    
    return aggregate_results(results)

8. 最佳实践

8.1 系统设计原则

  1. 模块化设计：各智能体功能独立，便于调试和优化
  2. 容错机制：关键步骤添加验证和回退策略
  3. 可观测性：记录详细的执行日志，便于分析问题
  4. 资源管理：设置最大迭代次数，防止无限循环
  5. 用户控制：提供用户干预接口，允许手动修正方向

8.2 性能优化建议

# 性能优化配置
OPTIMIZATION_CONFIG = {
    # 缓存策略
    "cache": {
        "enable": True,
        "ttl": 3600,  # 1小时
        "similarity_threshold": 0.95
    },
    
    # 并行化
    "parallel": {
        "independent_retrieval": True,
        "max_parallel_tools": 3
    },
    
    # 早停策略
    "early_stopping": {
        "enable": True,
        "min_improvement": 0.05,
        "patience": 2
    },
    
    # 资源限制
    "limits": {
        "max_iterations": 5,
        "max_retrieval_docs": 50,
        "max_context_tokens": 100000
    }
}

9. 总结

Agentic RAG代表了RAG技术的重要演进方向：

特性价值
动态决策自适应检索策略，匹配查询需求
迭代推理多轮检索-生成循环，提高答案质量
自我纠正自动检测和修正错误
工具扩展集成外部能力，增强应用范围
记忆系统利用历史经验，提升长期性能
多智能体协作专业分工，处理复杂任务

Agentic RAG的核心理念是将RAG从被动工具转变为主动助手，使其能够像人类专家一样思考、检索、验证和改进。

参考资料

Footnotes

  1. Gao, Y., et al. “Agentic Retrieval-Augmented Generation: A Survey on Agentic RAG.” arXiv:2501.09136, 2025. ↩

关系图谱

反向链接