Agentic RAG 架构实战：让 AI Agent 自主决策检索策略

传统 RAG（Retrieval-Augmented Generation）的"先检索、再生成"流水线在实际生产中频频失效——根据 Vectara 2026 年的 RAG 基准测试，标准 RAG 在多跳推理任务上的准确率仅为 38%，而在需要跨文档聚合信息的场景中，这一数字跌至 22%。根本原因在于：传统 RAG 是一条"傻瓜式"管道，不管用户问什么，都机械地检索→拼接→生成，无法根据问题复杂度动态调整策略。Agentic RAG 正是为解决这个问题而生——它让 AI Agent 自主决定何时检索、检索什么、如何检索、以及检索结果是否足够，将 RAG 从"流水线"升级为"智能体"。

🧠 一、传统 RAG 的致命缺陷与 Agentic RAG 的核心理念

1.1 传统 RAG 的三大痛点

传统 RAG 的基本流程是：用户提问 → Embedding → 向量检索 Top-K → 拼接上下文 → LLM 生成。这个流程看似合理，但在实际生产中暴露了三个根本性问题：

❌ 问题一：无差别检索（Always Retrieve）

并非所有问题都需要检索。"今天星期几？"或"1+1 等于多少？"这类问题，LLM 自身就能回答，强制检索反而引入噪声。反之，"比较 A 公司 2024 年和 2025 年的营收变化"需要多步检索，单次检索无法覆盖。

❌ 问题二：查询与文档的语义鸿沟（Semantic Gap）

用户问"为什么服务器响应慢"，但文档中写的是"P99 延迟升高"。Embedding 模型虽然能捕捉部分语义相似度，但在专业领域的查询-文档匹配上，召回率常常低于 60%。

❌ 问题三：检索结果质量不可控（No Quality Gate）

Top-K 检索回来的文档中，可能有 3 条相关、2 条无关。传统 RAG 把所有文档一股脑塞给 LLM，无关文档不仅浪费 Token，还会误导模型生成错误答案。

1.2 Agentic RAG 的设计哲学

Agentic RAG 的核心思想是：将 RAG 流程中每个固定步骤都交给 Agent 自主决策。Agent 可以根据问题复杂度选择不同的策略：

⚡ **关键结论：**Agentic RAG 不是取代传统 RAG，而是在其基础上增加了"智能路由"和"质量门控"。对于简单查询，它会跳过检索直接回答；对于复杂查询，它会规划多步检索策略。这种自适应能力让 RAG 在真实场景中的准确率提升了 30-50%。

🔧 二、Agentic RAG 核心组件实现

2.1 架构总览

一个完整的 Agentic RAG 系统包含以下核心组件：

用户查询 → 查询分析器 → 策略路由器 → [检索工具集] → 质量评估器 → 生成/重试
                                     ↑                    ↓
                                     └── 查询重写器 ←─────┘（自反思循环）

我们将用 TypeScript + OpenAI Function Calling 实现完整的 Agentic RAG 系统。以下是核心类型定义：

// Agentic RAG 核心类型定义
interface RetrievalResult {
  content: string
  source: string
  score: number
  metadata: Record<string, unknown>
}

interface AgentAction {
  type: 'retrieve_vector' | 'retrieve_keyword' | 'retrieve_sql' | 'rewrite_query' | 'answer_directly'
  params: Record<string, unknown>
}

interface ReflectionResult {
  isSufficient: boolean
  missingInfo: string[]
  suggestedAction: AgentAction
  confidence: number
}

interface AgenticRAGConfig {
  maxIterations: number        // 最大迭代次数，防止无限循环
  confidenceThreshold: number  // 置信度阈值，高于此值直接回答
  retrievalTopK: number        // 每次检索返回的文档数
  model: string                // 使用的 LLM 模型
}

2.2 查询分析与策略路由

查询分析是 Agentic RAG 的"大脑"——它接收用户的原始问题，判断复杂度，并选择最合适的处理策略。

// 查询分析器：判断问题类型并路由到合适策略
import OpenAI from 'openai'

const openai = new OpenAI()

async function analyzeQuery(query: string): Promise<AgentAction> {
  const response = await openai.chat.completions.create({
    model: 'gpt-4o',
    temperature: 0,
    messages: [
      {
        role: 'system',
        content: `你是一个查询分析专家。分析用户问题，选择最合适的处理策略：
- answer_directly: 简单事实性问题，LLM 自身可以回答
- retrieve_vector: 需要语义检索的开放性问题
- retrieve_keyword: 包含精确术语的专业问题
- retrieve_sql: 需要结构化数据查询的问题
- rewrite_query: 问题模糊或需要先分解再检索

返回 JSON 格式: { "type": "...", "params": { "reason": "...", "subQueries": [...] } }`
      },
      { role: 'user', content: query }
    ],
    response_format: { type: 'json_object' }
  })

  return JSON.parse(response.choices[0].message.content!) as AgentAction
}

💡 **提示：**查询分析器的关键在于 temperature: 0，确保路由决策的一致性。在生产环境中，建议用 few-shot examples 覆盖你业务领域的典型查询模式，而不仅仅依赖 system prompt。

2.3 多策略检索工具集

Agentic RAG 的核心优势在于它能调用多种检索工具。以下是三种主要检索策略的实现：

// 向量检索：适合语义相似度匹配
async function vectorSearch(query: string, topK: number): Promise<RetrievalResult[]> {
  const embedding = await openai.embeddings.create({
    model: 'text-embedding-3-small',
    input: query
  })

  // 调用向量数据库（以 Pinecone 为例）
  const results = await pineconeIndex.query({
    vector: embedding.data[0].embedding,
    topK,
    includeMetadata: true
  })

  return results.matches.map(match => ({
    content: match.metadata?.content as string,
    source: match.metadata?.source as string,
    score: match.score ?? 0,
    metadata: match.metadata ?? {}
  }))
}

// 关键词检索（BM25）：适合精确术语匹配
async function keywordSearch(query: string, topK: number): Promise<RetrievalResult[]> {
  // 使用 Elasticsearch 的 match_query
  const response = await esClient.search({
    index: 'documents',
    body: {
      query: {
        multi_match: {
          query,
          fields: ['title^3', 'content', 'tags^2'],
          type: 'best_fields',
          fuzziness: 'AUTO'
        }
      },
      size: topK
    }
  })

  return response.hits.hits.map(hit => ({
    content: hit._source.content,
    source: hit._source.url,
    score: hit._score,
    metadata: hit._source
  }))
}

// SQL 检索：适合结构化数据查询
async function sqlSearch(query: string): Promise<RetrievalResult[]> {
  // 先让 LLM 将自然语言转为 SQL
  const sqlResponse = await openai.chat.completions.create({
    model: 'gpt-4o-mini',
    temperature: 0,
    messages: [
      {
        role: 'system',
        content: `将用户问题转换为 SQL 查询。数据库表结构：
- products(id, name, category, price, description, created_at)
- orders(id, product_id, quantity, total, order_date)
- reviews(id, product_id, rating, content, created_at)
只返回 SQL 语句，不要其他内容。`
      },
      { role: 'user', content: query }
    ]
  })

  const sql = sqlResponse.choices[0].message.content!
  const rows = await db.execute(sql)

  return rows.map(row => ({
    content: JSON.stringify(row),
    source: 'database',
    score: 1.0,
    metadata: { sql, row }
  }))
}

⚠️ **警告：**直接将 LLM 生成的 SQL 发送到生产数据库是极其危险的。在生产环境中，必须对生成的 SQL 进行安全校验：禁止 DELETE/UPDATE/DROP 操作、限制查询表范围、设置查询超时和结果行数上限。建议使用只读数据库连接。

🔄 三、自反思机制与多步检索

3.1 质量评估器：检索结果够不够？

这是 Agentic RAG 最关键的组件——它评估当前检索到的文档是否足以回答用户的问题。如果不够，它会建议下一步行动。

// 自反思评估器：判断检索结果是否充分
async function reflectOnResults(
  query: string,
  retrievedDocs: RetrievalResult[]
): Promise<ReflectionResult> {
  const docsContext = retrievedDocs
    .map((doc, i) => `[文档${i + 1}] (来源: ${doc.source}, 相关度: ${doc.score.toFixed(2)})\n${doc.content}`)
    .join('\n\n')

  const response = await openai.chat.completions.create({
    model: 'gpt-4o',
    temperature: 0,
    messages: [
      {
        role: 'system',
        content: `你是一个检索质量评估专家。评估以下检索结果是否足够回答用户问题。

评估标准：
1. 相关性：检索到的文档是否与问题相关？
2. 完整性：文档是否包含回答问题所需的全部信息？
3. 可信度：信息来源是否可靠？

返回 JSON：
{
  "isSufficient": boolean,
  "missingInfo": ["缺失的信息1", "缺失的信息2"],
  "suggestedAction": {
    "type": "retrieve_vector|retrieve_keyword|rewrite_query|answer_directly",
    "params": { "reason": "建议原因", "newQuery": "重写后的查询（如果需要）" }
  },
  "confidence": 0.0-1.0
}`
      },
      {
        role: 'user',
        content: `用户问题：${query}\n\n检索结果：\n${docsContext}`
      }
    ],
    response_format: { type: 'json_object' }
  })

  return JSON.parse(response.choices[0].message.content!) as ReflectionResult
}

3.2 查询重写器：让检索更精准

当检索结果不理想时，Agent 会重写查询以获得更好的检索效果。查询重写有多种策略：

// 查询重写：支持多种重写策略
async function rewriteQuery(
  originalQuery: string,
  missingInfo: string[],
  strategy: 'expand' | 'decompose' | 'hyde'
): Promise<string[]> {
  const prompts: Record<string, string> = {
    // 扩展：添加同义词和相关概念，提高召回率
    expand: `将以下查询扩展为 3 个不同表述方式，保持语义一致但使用不同的关键词。
原始查询: ${originalQuery}
缺失信息: ${missingInfo.join(', ')}`,

    // 分解：将复杂问题拆解为子问题
    decompose: `将以下复杂问题分解为 2-4 个独立的子问题，每个子问题可以单独检索回答。
原始查询: ${originalQuery}
缺失信息: ${missingInfo.join(', ')}`,

    // HyDE（Hypothetical Document Embeddings）：生成假设性文档来缩小语义差距
    hyde: `为以下问题生成一段假设性的回答文档（200字左右），这段文档应该像真实文档一样包含专业术语。
问题: ${originalQuery}`
  }

  const response = await openai.chat.completions.create({
    model: 'gpt-4o-mini',
    temperature: 0.3,
    messages: [
      { role: 'system', content: prompts[strategy] }
    ]
  })

  const content = response.choices[0].message.content!
  // 按编号分割为独立查询
  return content.split(/\d+\.\s+/).filter(q => q.trim().length > 0)
}

📌 **记住：**HyDE（Hypothetical Document Embeddings）是一种特别有效的查询重写策略。它的原理是：不直接用问题去检索，而是先生成一段"假设性答案文档"，再用这篇文档去检索。因为假设文档与真实文档在表述方式上更接近，检索召回率通常能提升 15-25%。

3.3 完整的 Agentic RAG 编排循环

将所有组件串联起来，形成完整的 Agentic RAG 执行循环：

// Agentic RAG 主循环：自主决策的检索-生成流程
async function agenticRAG(
  query: string,
  config: AgenticRAGConfig = {
    maxIterations: 3,
    confidenceThreshold: 0.85,
    retrievalTopK: 5,
    model: 'gpt-4o'
  }
): Promise<string> {
  const allDocs: RetrievalResult[] = []
  let currentQuery = query
  let iteration = 0

  while (iteration < config.maxIterations) {
    iteration++
    console.log(`[迭代 ${iteration}] 当前查询: ${currentQuery}`)

    // 步骤 1: 查询分析与策略路由
    const action = await analyzeQuery(currentQuery)
    console.log(`[策略] ${action.type}: ${JSON.stringify(action.params)}`)

    // 步骤 2: 直接回答（无需检索）
    if (action.type === 'answer_directly') {
      return await generateAnswer(query, [])
    }

    // 步骤 3: 执行检索
    let docs: RetrievalResult[] = []
    switch (action.type) {
      case 'retrieve_vector':
        docs = await vectorSearch(currentQuery, config.retrievalTopK)
        break
      case 'retrieve_keyword':
        docs = await keywordSearch(currentQuery, config.retrievalTopK)
        break
      case 'retrieve_sql':
        docs = await sqlSearch(currentQuery)
        break
      case 'rewrite_query':
        const rewrittenQueries = await rewriteQuery(currentQuery, [], 'decompose')
        for (const q of rewrittenQueries) {
          docs.push(...await vectorSearch(q, 3))
        }
        break
    }

    // 去重并合并文档（基于内容 hash）
    const seen = new Set(allDocs.map(d => d.source))
    for (const doc of docs) {
      if (!seen.has(doc.source)) {
        allDocs.push(doc)
        seen.add(doc.source)
      }
    }

    // 步骤 4: 自反思 - 评估检索结果是否充分
    const reflection = await reflectOnResults(query, allDocs)
    console.log(`[反思] 充分性: ${reflection.isSufficient}, 置信度: ${reflection.confidence}`)

    // 步骤 5: 置信度足够高，直接生成答案
    if (reflection.isSufficient && reflection.confidence >= config.confidenceThreshold) {
      return await generateAnswer(query, allDocs)
    }

    // 步骤 6: 检索不足，根据建议重写查询继续
    if (reflection.suggestedAction.type === 'rewrite_query') {
      currentQuery = reflection.suggestedAction.params.newQuery as string || currentQuery
      console.log(`[重写] 新查询: ${currentQuery}`)
    }
  }

  // 达到最大迭代次数，用已有文档生成最佳答案
  console.log(`[警告] 达到最大迭代次数 ${config.maxIterations}，使用当前文档生成答案`)
  return await generateAnswer(query, allDocs)
}

// 答案生成：带来源引用
async function generateAnswer(query: string, docs: RetrievalResult[]): Promise<string> {
  const context = docs.length > 0
    ? docs.map((d, i) => `[${i + 1}] ${d.content} (来源: ${d.source})`).join('\n\n')
    : '无外部文档，请基于自身知识回答。'

  const response = await openai.chat.completions.create({
    model: 'gpt-4o',
    messages: [
      {
        role: 'system',
        content: `基于以下参考资料回答用户问题。要求：
1. 优先使用参考资料中的信息
2. 如果参考资料不足，结合自身知识补充，但需标注"（推测）"
3. 在答案中标注信息来源 [1][2] 等
4. 如果无法确定答案，诚实说明`
      },
      {
        role: 'user',
        content: `参考资料：\n${context}\n\n用户问题：${query}`
      }
    ]
  })

  return response.choices[0].message.content!
}

⚡ 四、生产环境优化与成本控制

4.1 性能对比数据

在真实业务场景中，Agentic RAG 与传统 RAG 的效果差异非常明显：

4.2 成本控制策略

Agentic RAG 的多步检索意味着更高的 Token 消耗。以下是经过生产验证的成本控制方案：

// 语义缓存：相同/相似查询直接返回缓存结果
import { createHash } from 'crypto'

class SemanticCache {
  private cache = new Map<string, { result: string; timestamp: number }>()
  private readonly ttl: number

  constructor(ttlMs: number = 3600_000) {
    this.ttl = ttlMs
  }

  // 用查询的 embedding 的 hash 作为缓存 key（近似语义匹配）
  async get(query: string): Promise<string | null> {
    const key = await this.computeKey(query)
    const entry = this.cache.get(key)

    if (!entry) return null
    if (Date.now() - entry.timestamp > this.ttl) {
      this.cache.delete(key)
      return null
    }

    console.log(`[缓存命中] ${query.substring(0, 50)}...`)
    return entry.result
  }

  async set(query: string, result: string): Promise<void> {
    const key = await this.computeKey(query)
    this.cache.set(key, { result, timestamp: Date.now() })
  }

  private async computeKey(query: string): Promise<string> {
    // 使用 Embedding 量化后的 hash 作为 key
    // 相似查询会得到相近的量化向量，从而命中同一缓存
    const embedding = await openai.embeddings.create({
      model: 'text-embedding-3-small',
      input: query.toLowerCase().trim()
    })

    // 量化 embedding 到 4-bit 以实现模糊匹配
    const quantized = embedding.data[0].embedding
      .map(v => v > 0 ? 1 : 0)
      .join('')

    return createHash('sha256').update(quantized).digest('hex').substring(0, 16)
  }
}

// 使用示例
const cache = new SemanticCache()

async function agenticRAGWithCache(query: string, config: AgenticRAGConfig): Promise<string> {
  const cached = await cache.get(query)
  if (cached) return cached

  const result = await agenticRAG(query, config)
  await cache.set(query, result)
  return result
}

4.3 成本预算表

在不同使用场景下，Agentic RAG 的 Token 消耗估算：

💡 **提示：**通过语义缓存可以将重复查询的成本降低 60-80%。在客服场景中，"如何重置密码"这类问题占总查询量的 30% 以上，缓存命中率非常高。

4.4 最佳实践与避坑指南

✅ 推荐做法：

• 设置 maxIterations 上限（建议 3-5），防止 Agent 陷入无限检索循环
• 用 gpt-4o-mini 做查询分析和质量评估，用 gpt-4o 做最终生成，节省 60%+ 成本
• 对检索结果做去重和相关度过滤，避免上下文窗口被无关文档填满
• 在自反思中引入"信心衰减"——如果连续 2 次重写查询都未改善结果，果断用当前文档生成答案

❌ 避免做法：

• ❌ 不要在生产环境中开启无限迭代，这会导致成本失控
• ❌ 不要将所有检索结果无差别拼接，超过 8000 Token 的上下文会稀释关键信息的注意力
• ❌ 不要只用向量检索，在精确术语查询场景中 BM25 的召回率比 Embedding 高 20-30%
• ❌ 不要忽略 LLM 生成 SQL 的安全校验，这是严重的安全漏洞

⚠️ 关键注意事项：

• Agentic RAG 的延迟高于传统 RAG（平均多 2-5 秒），在延迟敏感的场景中需要权衡
• 建议在日志中记录每一步的检索结果和反思决策，便于调试和优化
• 定期对 Agent 的决策进行人工审计，确保路由策略符合业务预期

📊 总结

Agentic RAG 代表了 RAG 技术的下一代演进方向。它用 Agent 的自主决策能力取代了传统 RAG 的固定流水线，在复杂推理任务上的准确率提升可达 30-50%，但代价是更高的复杂度和成本。

明确的选型建议：

• ✅ **用传统 RAG：**简单 QA、FAQ 机器人、文档搜索等单步检索场景
• ✅ **用 Agentic RAG：**多跳推理、跨文档分析、复杂业务查询、需要精确答案的专业领域
• ⚠️ **混合方案：**在实际生产中，最佳实践是用查询分析器做路由——简单查询走传统 RAG（低延迟低成本），复杂查询走 Agentic RAG（高准确率）

相关工具推荐：

• **向量数据库：**Pinecone、Qdrant、Weaviate、pgvector
• **Embedding 模型：**text-embedding-3-small（OpenAI）、BGE-M3（开源多语言）
• **Agent 框架：**Vercel AI SDK、LangChain.js、LlamaIndex.TS
• **监控：**Langfuse（LLM 可观测性平台，支持 Agent 调试和 Token 成本追踪）