Agentic_RAG实战：动态检索策略与自适应推理完整指南

Agentic RAG 实战：动态检索策略与自适应推理完整指南

作者：Crown_22 | AI Agent & RAG 系统开发者 | 技术分享

前言

传统 RAG（Retrieval-Augmented Generation）系统采用固定的\”检索-生成\”流水线，查询进来后先向量检索 top-k 文档，再拼接上下文交给 LLM 生成答案。这种架构简单直接，但在复杂场景下暴露了严重瓶颈：

• 检索粒度单一：用户问\”什么是 Transformer\”和\”Transformer 的注意力机制在长序列上的时间复杂度是多少\”用的是同一套检索策略
• 无法动态调整：检索到的内容不够精准时，系统不会自动重试或换一种检索方式
• 缺乏推理能力：面对需要多步推理的复杂问题，单次检索往往无法提供足够信息

Agentic RAG 的核心思想是：让 AI Agent 来决定如何检索、检索什么、是否需要多次检索，而不是用固定流水线一刀切。

本文将从零构建一个生产级 Agentic RAG 系统，涵盖动态检索策略、自适应推理链、查询路由和质量评估。

一、传统 RAG vs Agentic RAG 架构对比

1.1 传统 RAG 的问题

# 传统 RAG：固定流水线，无决策能力
def traditional_rag(query: str) –> str:
# 步骤固定，无法根据查询复杂度调整
docs = vector_store.similarity_search(query, k=4) # 固定 top-k
context = \”\\n\”.join([doc.page_content for doc in docs])
prompt = f\”根据以下资料回答：\\n{

context}\\n\\n问题：{

query}\”
return llm.invoke(prompt)

实际生产中遇到的问题：

# 问题场景1：查询太模糊，向量检索命中率低
query = \”那个性能问题怎么解决的\”
# 向量检索返回的都是泛泛而谈的性能优化文章，缺少具体上下文

# 问题场景2：需要多步推理的复杂问题
query = \”对比 LangChain 和 LlamaIndex 在多文档推理场景下的实现差异\”
# 单次检索只能获取一个框架的信息，无法同时获取两个框架的对比

# 问题场景3：查询需要不同粒度的检索
query = \”Python asyncio 的事件循环实现原理\”
# 文档级检索太粗，段落级检索太细，需要自适应选择

1.2 Agentic RAG 的核心设计

Agentic RAG 将传统流水线替换为 Agent 驱动的动态决策系统：

用户查询
│
▼
┌─────────────────┐
│ 查询分析 Agent │ ← 分析查询类型、复杂度、所需信息
└────────┬────────┘
│
▼
┌─────────────────┐
│ 查询路由决策 │ ← 选择检索策略（向量/关键词/混合/多跳）
└────────┬────────┘
│
┌────┼────┬────────┐
▼ ▼ ▼ ▼
向量 关键词 SQL Web API
检索 检索 查询 调用
│ │ │ │
└────┴────┴────────┘
│
▼
┌─────────────────┐
│ 结果评估 Agent │ ← 判断信息是否足够，是否需要重新检索
└────────┬────────┘
│ (不够？重试)
▼
┌─────────────────┐
│ 答案生成 Agent │ ← 综合所有信息生成最终答案
└─────────────────┘

二、从零构建 Agentic RAG 系统

2.1 项目结构

agentic-rag/
├── core/
│ ├── agent.py # 主 Agent 逻辑
│ ├── retrievers/ # 多种检索器
│ │ ├── vector.py # 向量检索
│ │ ├── keyword.py # 关键词检索（BM25）
│ │ ├── hybrid.py # 混合检索
│ │ └── multi_hop.py # 多跳检索
│ ├── evaluators/ # 质量评估
│ │ ├── relevance.py # 相关性评估
│ │ └── completeness.py # 完整性评估
│ └── memory/ # 对话记忆
├── config.py
├── main.py
└── requirements.txt

2.2 依赖安装

# requirements.txt
langchain>=0.3.0
langchain–openai>=0.2.0
langchain–community>=0.3.0
faiss–cpu>=1.8.0
rank–bm25>=0.2.2
pydantic>=2.0
tiktoken>=0.7.0

pip install -r requirements.txt

2.3 核心 Agent 实现

# core/agent.py
from typing import Literal
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool

class QueryAnalysis(BaseModel):
\”\”\”查询分析结果\”\”\”
query_type: Literal[\”factual\”, \”analytical\”, \”comparative\”, \”procedural\”] = Field(
description=\”查询类型：事实性、分析性、对比性、过程性\”
)
complexity: Literal[\”simple\”, \”moderate\”, \”complex\”] = Field(
description=\”查询复杂度\”
)
sub_queries: list[str] = Field(
default_factory=list,
description=\”拆解后的子查询（复杂查询才需要）\”
)
recommended_strategy: Literal[\”vector\”, \”keyword\”, \”hybrid\”, \”multi_hop\”] = Field(
description=\”推荐的检索策略\”
)
reasoning: str = Field(description=\”策略选择的理由\”)

class RetrievalResult(BaseModel):
\”\”\”检索结果评估\”\”\”
documents: list[dict]
relevance_scores: list[float]
is_sufficient: bool = Field(description=\”信息是否足够回答问题\”)
missing_info: str = Field(default=\”\”, description=\”缺失的信息\”)
suggested_action: Literal[\”answer\”, \”retry_vector\”, \”retry_keyword\”, \”multi_hop\”] = Field(
description=\”建议的下一步操作\”
)

class AgenticRAG:
\”\”\”Agentic RAG 主系统\”\”\”

def __init__(self, vector_store, bm25_index, llm_model=\”gpt-4o\”):
self.vector_store = vector_store
self.bm25_index = bm25_index
self.llm = ChatOpenAI(model=llm_model, temperature=0)
self.structured_llm = self.llm.with_structured_output(QueryAnalysis)

def analyze_query(self, query: str) –> QueryAnalysis:
\”\”\”分析查询，决定检索策略\”\”\”
messages = [
SystemMessage(content=\”\”\”你是一个查询分析专家。分析用户查询并决定最佳检索策略。

查询类型：
– factual：事实性查询，如\”什么是X\”、\”X的定义\”
– analytical：分析性查询，如\”为什么X会导致Y\”
– comparative：对比性查询，如\”X和Y的区别\”
– procedural：过程性查询，如\”如何实现X\”

检索策略选择规则：
– simple + factual → vector（向量检索即可）
– moderate + analytical → hybrid（混合检索）
– complex + comparative → multi_hop（多跳检索）
– simple + procedural → keyword（关键词检索，代码相关）
– complex查询需要拆解为sub_queries\”\”\”),
HumanMessage(content=f\”请分析以下查询：\\n\\n{

query}\”)
]
return self.structured_llm.invoke(messages)

def vector_retrieve(self, query: str, k: int = 4) –> list[dict]:
\”\”\”向量检索\”\”\”
docs = self.vector_store.similarity_search_with_score(query, k=k)
return [{

\”content\”: doc.page_content, \”metadata\”: doc.metadata, \”score\”: score}
for doc, score in docs]

def keyword_retrieve(self, query: str, k: int = 4) –> list[dict]:
\”\”\”BM25 关键词检索\”\”\”
from rank_bm25 import BM25Okapi
tokenized_query = list(query)
scores = self.bm25_index.get_scores(tokenized_query)
top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:k]
return [{

\”content\”: self.corpus[i], \”metadata\”: self.metadata[i], \”score\”: scores[i]}
for i in top_indices]

def hybrid_retrieve(self, query: str, k: int = 6) –> list[dict]:
\”\”\”混合检索：向量 + BM25，RRF 融合\”\”\”
vector_results = self.vector_retrieve(query, k=k)
keyword_results = self.keyword_retrieve(query, k=k)

# Reciprocal Rank Fusion
rrf_scores = {

}
for rank, doc in enumerate(vector_results):
key = doc[\”content\”][:100] # 用内容前100字符作为key
rrf_scores[key] = rrf_scores.get(key, 0) + 1 / (60 + rank)

for rank, doc in enumerate(keyword_results):
key = doc[\”content\”][:100]
rrf_scores[key] = rrf_scores.get(key, 0) + 1 / (60 + rank)

# 合并去重并按RRF分数排序
all_docs = {

doc[\”content\”][:100]: doc for doc in vector_results + keyword_results}
sorted_keys = sorted(rrf_scores.keys(), key=lambda k: rrf_scores[k], reverse=True)

results = []
for key in sorted_keys[:k]:
doc = all_docs[key]
doc[\”rrf_score\”] = rrf_scores[key]
results.append(doc)
return results

def multi_hop_retrieve(self, sub_queries: list[str]) –> list[dict]:
\”\”\”多跳检索：对每个子查询分别检索，去重合并\”\”\”
all_results = []
seen_contents = set()

for sq in sub_queries:
results = self.hybrid_retrieve(sq, k=3)
for doc in results: