5.3 Agent增强的RAG系统

5.3 Agent增强的RAG系统

5.3.1 从单次检索到多步推理的范式转变

传统RAG系统的核心流程是「检索-生成」两步走：用户提出问题，系统从向量库中检索相关文档片段，再将这些片段与原始问题拼接后交给大语言模型生成回答。这种模式在处理简单事实查询时效果良好，但面对复杂问题、需要多轮推理的场景时，往往力不从心。用户的问题可能是模糊的、需要分解的，或者需要结合多个信息源才能回答。在这些场景下，单次检索的局限性就暴露出来了。

Agent增强的RAG系统（Agentic RAG）正是为了解决这一局限而诞生的。其核心思想是将大语言模型的推理能力、工具调用能力与RAG的检索能力深度融合，形成一个能够自主规划、多步检索、动态调整策略的智能系统。与传统的静态检索不同，Agentic RAG赋予了系统「思考」的能力——它可以分析问题复杂度、决定是否需要额外信息、选择合适的检索策略，甚至在检索结果不理想时自动调整方向重新检索。

这种范式转变的意义在于，RAG系统从被动的信息传递管道变成了主动的知识探索者。用户不再需要精心构造查询语句，系统可以自主完成从问题理解到信息获取再到答案合成的全流程。

5.3.2 Agentic RAG的核心架构

Agentic RAG的架构通常包含以下几个核心组件：

规划模块（Planning Module）：这是Agent的「大脑」。当用户提出问题后，规划模块首先分析问题的复杂度和类型。对于简单问题，直接走标准RAG流程；对于复杂问题，则将其分解为多个子问题，并为每个子问题制定检索策略。规划模块还可以根据上下文动态调整计划，例如当某个子问题的检索结果质量不佳时，重新规划该子问题的检索方式。

工具集（Tool Set）：Agent可以调用的工具不仅包括向量检索，还可能包括关键词搜索、SQL查询、API调用、代码执行等。丰富的工具集使得Agent能够根据不同类型的问题选择最合适的信息获取方式。例如，对于结构化数据的查询，直接使用SQL可能比向量检索更精确。

记忆模块（Memory Module）：包括短期记忆和长期记忆。短期记忆保存当前对话的上下文和中间推理结果，长期记忆则存储历史交互中积累的用户偏好和领域知识。记忆模块使得Agent能够在多轮交互中保持连贯性，避免重复检索已经获取过的信息。

执行与反思模块（Execution and Reflection）：负责实际执行检索操作，并对结果进行质量评估。如果检索结果不够好，反思模块会触发重新检索或调整策略。这种自我反思机制是Agentic RAG区别于传统RAG的关键特征之一。

下面是一个简化的Agentic RAG实现框架：

import json
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from enum import Enum


class QueryComplexity(Enum):
    SIMPLE = "simple"       # 单次检索可回答
    COMPOUND = "compound"   # 需要分解为多个子问题
    RESEARCH = "research"   # 需要多轮深入检索


@dataclass
class RetrievalResult:
    content: str
    source: str
    relevance_score: float
    metadata: Dict = field(default_factory=dict)


@dataclass
class SubQuery:
    query_text: str
    search_type: str = "vector"  # vector, keyword, sql, hybrid
    depends_on: List[int] = field(default_factory=list)
    result: Optional[List[RetrievalResult]] = None


class RetrievalPlanner:
    """规划模块：分析问题并制定检索策略"""
    
    def analyze_complexity(self, query: str) -> QueryComplexity:
        # 基于规则和LLM判断的复杂度分析
        compound_keywords = ["对比", "比较", "分别", "各自", "以及", "并且"]
        research_keywords = ["深入分析", "全面梳理", "系统总结", "调研报告"]
        
        query_lower = query.lower()
        if any(kw in query_lower for kw in research_keywords):
            return QueryComplexity.RESEARCH
        if any(kw in query_lower for kw in compound_keywords):
            return QueryComplexity.COMPOUND
        return QueryComplexity.SIMPLE
    
    def decompose_query(self, query: str) -> List[SubQuery]:
        # 将复杂问题分解为子问题
        # 实际实现中会调用LLM进行分解
        prompt = f"""
        请将以下问题分解为需要独立检索的子问题列表。
        每个子问题应该是可以通过单次检索回答的。
        问题：{query}
        
        以JSON格式返回，每个子问题包含 query_text 和 search_type。
        search_type 可选值：vector, keyword, hybrid
        """
        # 调用LLM获取分解结果
        sub_queries = self._call_llm(prompt)
        return [SubQuery(**sq) for sq in sub_queries]


class ReflectionEvaluator:
    """反思模块：评估检索质量并决定是否需要重新检索"""
    
    def __init__(self, relevance_threshold: float = 0.6):
        self.relevance_threshold = relevance_threshold
    
    def evaluate(self, query: str, results: List[RetrievalResult]) -> Dict:
        if not results:
            return {"action": "retry", "reason": "无检索结果", "strategy": "broaden_query"}
        
        avg_score = sum(r.relevance_score for r in results) / len(results)
        max_score = max(r.relevance_score for r in results)
        
        if max_score >= self.relevance_threshold and avg_score >= 0.4:
            return {"action": "proceed", "avg_score": avg_score}
        elif avg_score >= 0.3:
            return {
                "action": "retry",
                "reason": "平均相关性不足",
                "strategy": "expand_results",
                "avg_score": avg_score
            }
        else:
            return {
                "action": "retry",
                "reason": "相关性过低",
                "strategy": "rewrite_query",
                "avg_score": avg_score
            }


class AgenticRAG:
    """Agentic RAG 主控制器"""
    
    def __init__(self, retriever, llm_client, max_retries: int = 3):
        self.retriever = retriever
        self.llm_client = llm_client
        self.planner = RetrievalPlanner()
        self.evaluator = ReflectionEvaluator()
        self.max_retries = max_retries
        self.conversation_history: List[Dict] = []
    
    def query(self, user_query: str) -> str:
        # 1. 分析复杂度
        complexity = self.planner.analyze_complexity(user_query)
        
        if complexity == QueryComplexity.SIMPLE:
            return self._simple_rag(user_query)
        else:
            return self._agentic_rag(user_query, complexity)
    
    def _simple_rag(self, query: str) -> str:
        """标准RAG流程"""
        results = self.retriever.search(query, top_k=5)
        context = self._format_context(results)
        return self._generate_answer(query, context)
    
    def _agentic_rag(self, query: str, complexity: QueryComplexity) -> str:
        """Agent增强的多步RAG流程"""
        # 2. 分解子问题
        sub_queries = self.planner.decompose_query(query)
        
        # 3. 逐个执行子查询（带反思重试）
        all_results = []
        for sq in sub_queries:
            results = self._execute_with_reflection(sq)
            sq.result = results
            all_results.extend(results)
        
        # 4. 综合生成答案
        context = self._format_context(all_results)
        answer = self._generate_answer(query, context)
        return answer
    
    def _execute_with_reflection(self, sq: SubQuery) -> List[RetrievalResult]:
        """带反思机制的检索执行"""
        for attempt in range(self.max_retries):
            results = self.retriever.search(sq.query_text, top_k=5)
            evaluation = self.evaluator.evaluate(sq.query_text, results)
            
            if evaluation["action"] == "proceed":
                return results
            
            # 根据评估结果调整策略
            strategy = evaluation.get("strategy", "rewrite_query")
            if strategy == "broaden_query":
                sq.query_text = self._broaden_query(sq.query_text)
            elif strategy == "expand_results":
                results = self.retriever.search(sq.query_text, top_k=15)
                return results
            elif strategy == "rewrite_query":
                sq.query_text = self._rewrite_query(sq.query_text)
        
        # 超过重试次数，返回已有结果
        return results
​

5.3.3 多步检索策略详解

多步检索是Agentic RAG的核心能力之一。根据不同的应用场景，可以设计多种多步检索策略。

串行检索策略：适用于需要逐步深入的场景。例如，用户问「请分析某公司的技术栈选型理由」，系统首先检索该公司的基本信息，然后根据获取到的技术栈列表，逐一检索每种技术的优缺点，最终综合生成分析报告。这种策略的特点是每一步的检索都依赖前一步的结果，形成一条推理链。

并行检索策略：适用于子问题之间相互独立的情况。例如，用户问「对比React和Vue在大型项目中的表现」，系统可以同时检索React在大型项目中的表现和Vue在大型项目中的表现，然后合并结果进行对比分析。并行策略的优势在于效率高，可以充分利用并发能力缩短响应时间。

迭代精炼策略：适用于初始检索结果不够精确的场景。系统先进行一次粗粒度的广域检索，获取概览性信息，然后根据概览缩小范围进行精确定位。这种策略类似于人类在查阅资料时先看目录再翻到具体章节的行为模式。

class MultiStepRetriever:
    """多步检索执行器"""
    
    def serial_retrieve(self, initial_query: str, steps: int = 3) -> List[RetrievalResult]:
        """串行检索：每步依赖上一步结果"""
        all_results = []
        current_query = initial_query
        
        for i in range(steps):
            results = self.retriever.search(current_query, top_k=5)
            all_results.extend(results)
            
            # 基于当前结果生成下一步查询
            if i < steps - 1:
                context_summary = "\n".join([r.content[:200] for r in results])
                current_query = self._generate_follow_up_query(
                    initial_query, context_summary, step=i+1
                )
        
        return all_results
    
    def parallel_retrieve(self, sub_queries: List[SubQuery]) -> Dict[str, List[RetrievalResult]]:
        """并行检索：独立子问题并发执行"""
        import concurrent.futures
        
        results_map = {}
        with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
            future_to_query = {
                executor.submit(
                    self.retriever.search, sq.query_text, 5
                ): sq.query_text
                for sq in sub_queries
            }
            for future in concurrent.futures.as_completed(future_to_query):
                query_text = future_to_query[future]
                results_map[query_text] = future.result()
        
        return results_map
    
    def iterative_refine(self, query: str, rounds: int = 2) -> List[RetrievalResult]:
        """迭代精炼：先广后精"""
        # 第一轮：广域检索
        broad_results = self.retriever.search(query, top_k=20)
        
        # 提取关键实体和概念
        entities = self._extract_entities(broad_results)
        
        # 第二轮：针对关键实体精确检索
        refined_results = []
        for entity in entities[:5]:
            refined_query = f"{entity} {query}"
            results = self.retriever.search(refined_query, top_k=3)
            refined_results.extend(results)
        
        return broad_results + refined_results
​

5.3.4 工具调用与混合信息获取

Agentic RAG的强大之处在于它不仅能调用向量检索这一种工具，而是可以根据问题类型灵活选择最合适的信息获取方式。一个成熟的Agentic RAG系统通常配备以下工具集：

向量检索（Vector Search）：适用于语义相似度匹配的场景，如概念解释、观点查询等。向量检索的优势在于能够理解查询意图而非仅仅匹配关键词，但缺点是对于精确匹配（如产品型号、特定数值）可能不够精确。

关键词检索（Keyword Search）：适用于精确匹配场景，如查找特定的技术术语、错误代码、版本号等。关键词检索通常基于BM25或Elasticsearch等全文搜索引擎实现。

结构化查询（SQL/GraphQL）：当数据存储在关系型数据库中时，Agent可以直接生成SQL查询语句获取精确数据。这种方式特别适合涉及数值比较、统计分析、关系查询的问题。

网页搜索（Web Search）：对于知识库中不包含的实时信息或外部知识，Agent可以调用网页搜索API获取最新信息，再进行整合分析。

代码执行（Code Execution）：当问题涉及计算、数据分析时，Agent可以编写并执行Python代码，直接对数据进行处理。这种能力在数据分析类RAG应用中尤为重要。

class ToolRegistry:
    """Agent工具注册中心"""
    
    def __init__(self):
        self.tools = {}
    
    def register(self, name: str, func, description: str, parameters: Dict):
        self.tools[name] = {
            "func": func,
            "description": description,
            "parameters": parameters
        }
    
    def get_tool_schemas(self) -> List[Dict]:
        """生成工具描述供LLM选择"""
        return [
            {"name": name, "description": info["description"], "parameters": info["parameters"]}
            for name, info in self.tools.items()
        ]
    
    def call(self, tool_name: str, **kwargs):
        return self.tools[tool_name]["func"](**kwargs)


# 注册常用工具
registry = ToolRegistry()
registry.register(
    "vector_search",
    lambda query, top_k=5: retriever.search(query, top_k),
    "语义向量检索，适用于概念性问题和模糊查询",
    {"query": "str", "top_k": "int, optional, default=5"}
)
registry.register(
    "keyword_search",
    lambda query, top_k=10: keyword_engine.search(query, top_k),
    "关键词精确匹配检索，适用于特定术语和代码",
    {"query": "str", "top_k": "int, optional, default=10"}
)
registry.register(
    "sql_query",
    lambda sql: db_engine.execute(sql),
    "执行SQL查询，适用于结构化数据的精确查询",
    {"sql": "str"}
)
registry.register(
    "web_search",
    lambda query: web_search_api.search(query),
    "网页搜索，适用于获取外部最新信息",
    {"query": "str"}
)
​

5.3.5 检索质量自评估与自适应优化

Agentic RAG的一个关键特征是系统能够对自身的检索结果进行质量评估。这种自评估能力使得系统不再盲目地信任检索结果，而是能够在发现问题时主动采取纠正措施。

检索质量评估可以从以下几个维度进行：

相关性评分：评估检索结果与查询问题的语义相关程度。可以通过计算查询向量与结果向量的余弦相似度来获得一个基线评分，也可以使用交叉编码器（Cross-Encoder）进行更精确的相关性打分。

覆盖度评估：评估检索结果是否充分覆盖了问题的各个方面。对于一个复合问题，如果只检索到了其中一部分子问题的相关信息，则覆盖度不足。

一致性检查：当多个检索片段之间存在矛盾信息时，系统需要识别这些矛盾并进行处理，而非简单地全部传递给生成模型。

时效性验证：对于涉及时间敏感信息的问题，系统需要检查检索结果的时效性，判断信息是否可能已经过时。

class QualityEvaluator:
    """检索质量多维度评估器"""
    
    def comprehensive_evaluate(
        self, query: str, results: List[RetrievalResult],
        expected_aspects: List[str] = None
    ) -> Dict:
        # 维度一：相关性
        relevance = self._evaluate_relevance(query, results)
        
        # 维度二：覆盖度
        coverage = self._evaluate_coverage(results, expected_aspects)
        
        # 维度三：一致性
        consistency = self._evaluate_consistency(results)
        
        # 维度四：信息密度
        density = self._evaluate_density(results)
        
        overall_score = (
            relevance * 0.35 +
            coverage * 0.30 +
            consistency * 0.20 +
            density * 0.15
        )
        
        return {
            "overall_score": overall_score,
            "relevance": relevance,
            "coverage": coverage,
            "consistency": consistency,
            "density": density,
            "pass": overall_score >= 0.6
        }
    
    def _evaluate_relevance(self, query: str, results: List[RetrievalResult]) -> float:
        """使用Cross-Encoder精确评估相关性"""
        if not results:
            return 0.0
        scores = [self.cross_encoder.predict(query, r.content) for r in results]
        return sum(scores) / len(scores)
    
    def _evaluate_coverage(self, results: List[RetrievalResult], aspects: List[str]) -> float:
        """评估结果是否覆盖了问题的所有方面"""
        if not aspects:
            return 1.0
        all_content = " ".join([r.content for r in results])
        covered = sum(1 for aspect in aspects if aspect in all_content)
        return covered / len(aspects)
    
    def _evaluate_consistency(self, results: List[RetrievalResult]) -> float:
        """检查结果之间是否存在矛盾"""
        if len(results) < 2:
            return 1.0
        # 简化实现：检查高相似度片段之间的关键信息是否一致
        pairs_checked = 0
        consistent_pairs = 0
        for i in range(len(results)):
            for j in range(i+1, min(i+3, len(results))):
                pairs_checked += 1
                if self._check_pair_consistency(results[i], results[j]):
                    consistent_pairs += 1
        return consistent_pairs / max(pairs_checked, 1)
​

5.3.6 生产环境中的Agentic RAG实践

将Agentic RAG从概念落地到生产环境，需要考虑多个工程实践问题。

延迟控制：多步检索天然比单次检索慢。在生产环境中，需要设置合理的超时和降级策略。对于实时性要求高的场景，可以设置最大检索步骤数，当步骤用完仍未获得满意结果时，基于已有信息生成答案并标注置信度。同时，可以利用缓存机制，对于相似的历史查询直接复用之前的检索结果。

成本管理：多步检索意味着更多的LLM调用和更多的Token消耗。可以通过以下方式控制成本：使用较小模型进行规划和反思，仅在最终答案生成时使用大模型；对规划结果进行缓存，相似问题可以复用已有的查询分解方案；设置Token预算上限，当消耗达到预算时停止检索。

可观测性：Agentic RAG的决策过程比传统RAG复杂得多，需要完善的日志和追踪机制。每一步的决策原因、检索策略选择、反思结果都应该被记录下来，便于后续分析和优化。可以使用OpenTelemetry等工具实现端到端的追踪。

class ProductionAgenticRAG(AgenticRAG):
    """面向生产环境的Agentic RAG"""
    
    def __init__(self, retriever, llm_client, config: Dict):
        super().__init__(retriever, llm_client)
        self.max_steps = config.get("max_steps", 3)
        self.token_budget = config.get("token_budget", 8000)
        self.timeout_seconds = config.get("timeout_seconds", 30)
        self.cache = QueryCache(max_size=1000, ttl_seconds=3600)
        self.tracer = Tracer(service_name="agentic-rag")
    
    def query(self, user_query: str) -> Dict:
        with self.tracer.start_as_current_span("agentic_query") as span:
            span.set_attribute("query", user_query)
            
            # 检查缓存
            cached = self.cache.get(user_query)
            if cached:
                span.set_attribute("cache_hit", True)
                return {"answer": cached, "source": "cache"}
            
            # 带预算限制的检索
            token_usage = 0
            answer = ""
            steps = 0
            
            while steps < self.max_steps and token_usage < self.token_budget:
                with self.tracer.start_as_current_span(f"step_{steps}"):
                    if steps == 0:
                        answer = super().query(user_query)
                    else:
                        # 基于反思进行补充检索
                        answer = self._supplemental_retrieval(user_query, answer)
                    token_usage += self._estimate_tokens(answer)
                    steps += 1
            
            self.cache.set(user_query, answer)
            span.set_attribute("steps", steps)
            span.set_attribute("token_usage", token_usage)
            
            return {
                "answer": answer,
                "source": "agentic_rag",
                "steps": steps,
                "token_usage": token_usage,
                "confidence": self._estimate_confidence(steps, token_usage)
            }
​

5.3.7 Agentic RAG的局限性与未来方向

尽管Agentic RAG代表了RAG技术的重要发展方向，但它目前仍面临一些挑战。

推理成本：多步检索和多次LLM调用的成本显著高于传统RAG。在商业应用中，需要在效果提升和成本控制之间找到平衡点。未来的优化方向包括更高效的规划算法、更精准的终止条件判断，以及模型蒸馏等技术来降低推理成本。

可靠性问题：Agent的自主决策能力是一把双刃剑。在规划出错或检索策略选择不当的情况下，系统可能产生比传统RAG更差的结果。需要建立完善的防护机制，例如设置安全边界、实现人工审核接口等。

评估标准：传统的RAG评估指标（如召回率、准确率）难以完全衡量Agentic RAG的效果，因为后者涉及多步推理和动态策略调整。业界正在探索新的评估框架，例如基于轨迹的评估（Trace-based Evaluation）和端到端的效果评估。

未来，Agentic RAG有望与更多前沿技术结合。例如，与思维链（Chain of Thought）技术结合增强推理深度，与多模态理解能力结合实现跨模态的知识检索与生成，与联邦学习结合在保护数据隐私的前提下实现跨组织的知识共享。这些方向的探索将进一步拓展RAG系统的能力边界，使其从单纯的问答工具进化为真正的知识工作助手。