Agentic RAG 检索增强智能体完全教程
零基础到实战：系统掌握 Agentic RAG 架构设计、自适应检索策略与智能体融合模式
适用人群
- AI 架构师：需要设计企业级 RAG 系统的技术负责人
- 高级开发者：已有 LLM 应用开发经验，希望深入 Agentic RAG
- 算法工程师：关注检索质量与推理能力的技术人员
- 产品经理：需要理解 Agentic RAG 技术边界与应用前景
前置知识
- Python 基础（函数、类、异步编程）
- 大语言模型基本概念（Prompt、Token、Embedding）
- 基本的向量检索知识（相似度搜索、向量数据库）
目录
- 第一章：Agentic RAG 概述与核心理念
- 第二章：Agentic RAG 架构设计
- 第三章：混合检索管线
- 第四章：重排序与压缩
- 第五章：查询路由与分发
- 第六章：自适应检索策略
- 第七章：多步推理与检索循环
- 第八章：Self-RAG 与 Corrective RAG
- 第九章：Agent-RAG 融合模式
- 第十章：LangGraph 实现 Agentic RAG
- 第十一章：评估体系
- 第十二章：实战项目一 —— 智能研究助手
- 第十三章：实战项目二 —— 多源知识问答系统
- 第十四章：常见问题与最佳实践
第一章：Agentic RAG 概述与核心理念
1.1 什么是 RAG
RAG（Retrieval-Augmented Generation，检索增强生成）是一种将外部知识检索与大语言模型生成能力相结合的技术范式。其核心思想是：在 LLM 生成回答之前，先从外部知识库中检索相关信息，将检索结果作为上下文注入到 Prompt 中，从而让模型基于真实、最新的知识生成更准确的回答。
RAG 解决了 LLM 的三大核心痛点：
- 知识截止：LLM 训练数据有时间截止，无法获取最新信息
- 幻觉问题：LLM 在缺乏知识时容易生成看似合理但错误的内容
- 领域知识不足：通用 LLM 对特定行业或企业的内部知识了解有限
一个最基本的 RAG 流程如下：
用户提问 → 查询编码 → 向量检索 → 获取相关文档 → 拼接 Prompt → LLM 生成回答
1.2 从 Naive RAG 到 Agentic RAG 的演进
RAG 技术经历了三个发展阶段：
第一代：Naive RAG（朴素 RAG）
最简单的"检索-生成"管线。用户查询直接用于检索，检索结果直接拼接后送入 LLM。优点是实现简单，缺点是检索质量完全取决于原始查询的质量，无法处理复杂问题。
第二代：Advanced RAG（进阶 RAG）
在 Naive RAG 基础上增加了查询改写、重排序、上下文压缩等优化技术。显著提升了检索质量，但整体流程仍然是线性的，缺乏自主决策能力。
第三代：Agentic RAG（智能体 RAG）
将 Agent 的自主决策能力引入 RAG 系统。系统不再是一个固定的管线，而是一个能够自主规划检索策略、动态选择数据源、迭代优化检索结果、判断何时停止检索的智能体。这是当前 RAG 技术的最前沿。
演进路径：
Naive RAG: 查询 → 检索 → 生成（线性管线）
Advanced RAG: 查询 → 改写 → 检索 → 重排序 → 压缩 → 生成（优化管线）
Agentic RAG: 查询 → 规划 → [检索 ↔ 推理 ↔ 决策] → 生成（自主循环）
1.3 Agentic RAG 的核心特征
Agentic RAG 与传统 RAG 的本质区别在于自主性。具体表现为以下五个核心特征：
1. 自主规划（Autonomous Planning） 系统能够根据查询的复杂度自动规划检索策略。简单查询直接检索，复杂查询先分解再逐步检索。
2. 动态路由（Dynamic Routing） 根据查询意图，自动选择最合适的数据源和检索方式。例如，技术问题走文档库，实时问题走搜索引擎，结构化数据走 SQL 查询。
3. 迭代优化（Iterative Refinement） 系统能够评估检索结果的质量，如果不够好，自动调整查询重新检索，直到获得满意的结果。
4. 多步推理（Multi-step Reasoning） 对于需要综合多个信息源的复杂问题，系统能够进行多步推理，逐步积累证据，最终形成完整答案。
5. 自我反思（Self-Reflection） 系统能够评估自身生成的答案质量，判断是否需要补充检索或修正答案。
1.4 适用场景与技术选型
适合使用 Agentic RAG 的场景：
| 场景 | 特征 | 示例 |
|---|---|---|
| 复杂问答 | 需要多步推理或多源信息 | "比较 A 公司和 B 公司近三年的营收趋势并分析原因" |
| 多源知识 | 需要从不同类型数据源获取信息 | 结合文档、数据库、API 的综合查询 |
| 研究分析 | 需要深入挖掘和迭代检索 | 学术文献综述、市场调研报告 |
| 动态决策 | 需要根据中间结果调整策略 | 故障诊断、法律案例分析 |
不适合的场景：
- 简单的事实性问答（直接用 Naive RAG 即可）
- 实时性要求极高（Agent 循环会增加延迟）
- 对延迟极度敏感的在线服务（考虑预计算方案）
技术选型建议：
| 技术栈 | 适用场景 | 特点 |
|---|---|---|
| LangGraph | 复杂 Agentic RAG 工作流 | 图状态机，灵活的条件路由 |
| LlamaIndex | 快速构建 RAG 应用 | 丰富的组件生态 |
| Haystack | 企业级 RAG 管线 | 模块化，生产就绪 |
| 自研框架 | 极致定制需求 | 完全可控，学习成本高 |
第二章：Agentic RAG 架构设计
2.1 整体架构分层
Agentic RAG 系统可以分为五个层次：
┌─────────────────────────────────────────┐
│ 应用层 (Application) │
│ 用户界面、API 接口、对话管理 │
├─────────────────────────────────────────┤
│ 编排层 (Orchestration) │
│ Agent 推理引擎、任务规划、流程控制 │
├─────────────────────────────────────────┤
│ 检索层 (Retrieval) │
│ 混合检索、重排序、压缩、路由 │
├─────────────────────────────────────────┤
│ 存储层 (Storage) │
│ 向量数据库、文档存储、知识图谱 │
├─────────────────────────────────────────┤
│ 数据层 (Data) │
│ 数据采集、清洗、分块、索引 │
└─────────────────────────────────────────┘
2.2 核心组件详解
Agent 推理引擎：系统的"大脑"，负责理解查询意图、规划检索策略、评估结果质量、决定下一步行动。
检索路由器：根据查询类型，将请求分发到不同的检索器（向量检索、关键词检索、SQL 查询、API 调用等）。
混合检索器：结合稠密检索和稀疏检索的优势，提供更全面的召回结果。
重排序器：对检索结果进行精细化排序，将最相关的内容排在前面。
上下文压缩器：去除检索结果中的冗余信息，保留核心内容，节省 Token 用量。
状态管理器：维护 Agent 的推理状态，包括已检索的信息、中间结论、检索历史等。
2.3 数据流与控制流设计
在 Agentic RAG 中，数据流和控制流是分离的：
数据流（信息的流动方向）：
原始查询 → 查询理解 → [多路检索] → 结果聚合 → 重排序 → 压缩 → 上下文构建 → LLM 生成
控制流（决策的流动方向）：
查询分析 → 策略规划 → 执行检索 → 结果评估 → [满意? → 生成回答 / 不满意 → 调整策略 → 重新检索]
控制流的关键在于循环机制。Agent 在每一步都可以做出决策：继续检索、调整策略、或者停止并生成答案。
2.4 状态管理与持久化
Agentic RAG 的状态管理是架构设计的关键。核心状态包括：
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from enum import Enum
class AgentState(Enum):
ANALYZING = "analyzing" # 分析查询
PLANNING = "planning" # 规划策略
RETRIEVING = "retrieving" # 执行检索
EVALUATING = "evaluating" # 评估结果
REASONING = "reasoning" # 推理整合
GENERATING = "generating" # 生成回答
REFLECTING = "reflecting" # 自我反思
COMPLETED = "completed" # 完成
@dataclass
class RetrievalResult:
content: str # 检索内容
source: str # 来源标识
score: float # 相关性分数
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class AgenticRAGState:
query: str # 原始查询
current_state: AgentState = AgentState.ANALYZING
rewritten_queries: List[str] = field(default_factory=list)
sub_queries: List[str] = field(default_factory=list)
retrieval_results: List[RetrievalResult] = field(default_factory=list)
reasoning_steps: List[str] = field(default_factory=list)
intermediate_answer: Optional[str] = None
final_answer: Optional[str] = None
iteration_count: int = 0
max_iterations: int = 5
confidence_score: float = 0.0
def should_continue(self) -> bool:
"""判断是否需要继续检索"""
if self.iteration_count >= self.max_iterations:
return False
if self.confidence_score >= 0.85:
return False
return True
第三章：混合检索管线
3.1 稠密检索（Dense Retrieval）
稠密检索使用 Embedding 模型将查询和文档映射到同一个向量空间，通过向量相似度进行检索。优点是能捕获语义信息，缺点是对精确关键词匹配不够敏感。
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Tuple
class DenseRetriever:
def __init__(self, model_name: str = "BAAI/bge-large-zh-v1.5"):
self.model = SentenceTransformer(model_name)
self.documents: List[str] = []
self.embeddings: np.ndarray = None
def index(self, documents: List[str]):
"""构建稠密索引"""
self.documents = documents
self.embeddings = self.model.encode(
documents,
normalize_embeddings=True,
show_progress_bar=True
)
def search(self, query: str, top_k: int = 10) -> List[Tuple[int, float]]:
"""稠密检索"""
query_embedding = self.model.encode(
[query], normalize_embeddings=True
)
# 余弦相似度计算
scores = np.dot(self.embeddings, query_embedding.T).flatten()
top_indices = np.argsort(scores)[::-1][:top_k]
return [(idx, scores[idx]) for idx in top_indices]
3.2 稀疏检索（Sparse Retrieval）
稀疏检索基于关键词匹配，典型代表是 BM25 算法。优点是对精确匹配效果好，缺点是无法理解语义相似性。
from rank_bm25 import BM25Okapi
import jieba
from typing import List, Tuple
class SparseRetriever:
def __init__(self):
self.bm25 = None
self.documents: List[str] = []
def _tokenize(self, text: str) -> List[str]:
"""中文分词"""
return list(jieba.cut(text))
def index(self, documents: List[str]):
"""构建稀疏索引"""
self.documents = documents
tokenized_docs = [self._tokenize(doc) for doc in documents]
self.bm25 = BM25Okapi(tokenized_docs)
def search(self, query: str, top_k: int = 10) -> List[Tuple[int, float]]:
"""稀疏检索"""
tokenized_query = self._tokenize(query)
scores = self.bm25.get_scores(tokenized_query)
top_indices = np.argsort(scores)[::-1][:top_k]
return [(idx, scores[idx]) for idx in top_indices]
3.3 混合检索融合策略
混合检索将稠密检索和稀疏检索的结果进行融合，常用的融合策略有：
1. RRF（Reciprocal Rank Fusion）融合
RRF 是最常用的融合策略，基于排名而非分数进行融合，对不同检索器的分数尺度不敏感。
def reciprocal_rank_fusion(
results_list: List[List[Tuple[int, float]]],
k: int = 60
) -> List[Tuple[int, float]]:
"""
RRF 融合算法
k: 平滑参数，通常取 60
"""
fused_scores: Dict[int, float] = {}
for results in results_list:
for rank, (doc_id, _) in enumerate(results):
if doc_id not in fused_scores:
fused_scores[doc_id] = 0.0
fused_scores[doc_id] += 1.0 / (k + rank + 1)
# 按融合分数降序排列
sorted_results = sorted(
fused_scores.items(),
key=lambda x: x[1],
reverse=True
)
return sorted_results
class HybridRetriever:
def __init__(self, dense_weight: float = 0.6, sparse_weight: float = 0.4):
self.dense_retriever = DenseRetriever()
self.sparse_retriever = SparseRetriever()
self.dense_weight = dense_weight
self.sparse_weight = sparse_weight
def index(self, documents: List[str]):
self.dense_retriever.index(documents)
self.sparse_retriever.index(documents)
def search(self, query: str, top_k: int = 10) -> List[Tuple[int, float, str]]:
"""混合检索"""
dense_results = self.dense_retriever.search(query, top_k=top_k * 2)
sparse_results = self.sparse_retriever.search(query, top_k=top_k * 2)
# RRF 融合
fused = reciprocal_rank_fusion([dense_results, sparse_results], k=60)
results = []
for doc_id, score in fused[:top_k]:
results.append((doc_id, score, self.dense_retriever.documents[doc_id]))
return results
2. 加权线性融合
def weighted_linear_fusion(
dense_results: List[Tuple[int, float]],
sparse_results: List[Tuple[int, float]],
dense_weight: float = 0.6,
sparse_weight: float = 0.4
) -> List[Tuple[int, float]]:
"""加权线性融合，需要先对分数做归一化"""
def normalize(results):
if not results:
return results
scores = [s for _, s in results]
min_s, max_s = min(scores), max(scores)
if max_s == min_s:
return [(idx, 1.0) for idx, _ in results]
return [(idx, (s - min_s) / (max_s - min_s)) for idx, s in results]
dense_norm = normalize(dense_results)
sparse_norm = normalize(sparse_results)
fused_scores: Dict[int, float] = {}
for idx, score in dense_norm:
fused_scores[idx] = fused_scores.get(idx, 0) + dense_weight * score
for idx, score in sparse_norm:
fused_scores[idx] = fused_scores.get(idx, 0) + sparse_weight * score
return sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
3.4 多模态检索扩展
当知识库包含图片、表格等非文本内容时，需要扩展为多模态检索：
class MultiModalRetriever:
"""多模态检索器示例框架"""
def __init__(self):
self.text_retriever = HybridRetriever()
self.image_index = {} # 图片向量索引
self.table_index = {} # 表格索引
def search(self, query: str, modality: str = "auto", top_k: int = 10):
if modality == "auto":
modality = self._detect_modality(query)
if modality == "text":
return self.text_retriever.search(query, top_k)
elif modality == "image":
return self._search_images(query, top_k)
elif modality == "table":
return self._search_tables(query, top_k)
elif modality == "all":
# 跨模态检索并融合
text_results = self.text_retriever.search(query, top_k)
image_results = self._search_images(query, top_k)
table_results = self._search_tables(query, top_k)
return reciprocal_rank_fusion([text_results, image_results, table_results])
def _detect_modality(self, query: str) -> str:
"""自动检测查询需要的模态"""
image_keywords = ["图片", "图像", "截图", "照片", "图中", "示意"]
table_keywords = ["表格", "数据", "统计", "对比", "排名"]
if any(kw in query for kw in image_keywords):
return "image"
elif any(kw in query for kw in table_keywords):
return "table"
return "text"
第四章：重排序与压缩
4.1 重排序模型（Reranker）
重排序是对初步检索结果进行精细化排序的过程。与 Embedding 模型不同，Reranker 使用交叉编码器（Cross-Encoder）直接计算查询和文档的相关性，精度更高但速度更慢。
from sentence_transformers import CrossEncoder
from typing import List, Tuple
class Reranker:
def __init__(self, model_name: str = "BAAI/bge-reranker-large"):
self.model = CrossEncoder(model_name)
def rerank(
self,
query: str,
documents: List[str],
top_k: int = 5
) -> List[Tuple[int, float, str]]:
"""重排序检索结果"""
# 构造查询-文档对
pairs = [(query, doc) for doc in documents]
# 计算相关性分数
scores = self.model.predict(pairs)
# 按分数降序排列
scored_docs = list(enumerate(scores))
scored_docs.sort(key=lambda x: x[1], reverse=True)
results = []
for idx, score in scored_docs[:top_k]:
results.append((idx, float(score), documents[idx]))
return results
4.2 上下文压缩（Context Compression）
上下文压缩的目的是去除检索结果中的冗余信息，保留与查询最相关的内容，从而节省 Token 用量并提高生成质量。
基于 LLM 的提取式压缩：
class LLMCompressor:
"""使用 LLM 进行上下文压缩"""
def __init__(self, llm):
self.llm = llm
def compress(self, query: str, context: str) -> str:
"""提取与查询相关的句子"""
prompt = f"""请从以下文档中提取与问题直接相关的句子或段落。
只保留回答问题所必需的信息，去除无关内容。
问题：{query}
文档：
{context}
提取的相关内容："""
return self.llm.invoke(prompt)
class SentenceCompressor:
"""基于句子级别的压缩"""
def __init__(self, similarity_threshold: float = 0.5):
self.similarity_threshold = similarity_threshold
self.embedder = SentenceTransformer("BAAI/bge-large-zh-v1.5")
def compress(self, query: str, document: str, top_k: int = 5) -> str:
"""提取与查询最相关的句子"""
# 将文档拆分为句子
sentences = self._split_sentences(document)
if not sentences:
return document
# 计算查询和每个句子的相似度
query_emb = self.embedder.encode([query], normalize_embeddings=True)
sent_embs = self.embedder.encode(sentences, normalize_embeddings=True)
scores = np.dot(sent_embs, query_emb.T).flatten()
# 选择最相关的句子并保持原始顺序
top_indices = np.argsort(scores)[::-1][:top_k]
selected = sorted(top_indices) # 恢复原始顺序
return "。".join([sentences[i] for i in selected]) + "。"
def _split_sentences(self, text: str) -> List[str]:
"""中文句子分割"""
import re
sentences = re.split(r'[。！？；\n]+', text)
return [s.strip() for s in sentences if s.strip()]
4.3 智能截断与摘要策略
当检索到的文档过长时，需要进行智能截断：
class SmartTruncator:
"""智能截断器"""
def __init__(self, max_tokens: int = 4000):
self.max_tokens = max_tokens
def truncate(self, documents: List[str], query: str) -> str:
"""
智能截断策略：
1. 优先保留高相关性文档
2. 在 Token 预算内尽量保留更多信息
3. 保持文档完整性（不在句子中间截断）
"""
result_parts = []
current_tokens = 0
for doc in documents:
doc_tokens = self._count_tokens(doc)
if current_tokens + doc_tokens <= self.max_tokens:
result_parts.append(doc)
current_tokens += doc_tokens
else:
# 剩余空间内截取
remaining = self.max_tokens - current_tokens
if remaining > 100: # 至少保留 100 token 的空间
truncated = self._truncate_to_tokens(doc, remaining)
result_parts.append(truncated)
break
return "\n\n---\n\n".join(result_parts)
def _count_tokens(self, text: str) -> int:
"""估算 Token 数（简化版）"""
return len(text) // 2 # 粗略估算，中文约 1 字 ≈ 1-2 token
def _truncate_to_tokens(self, text: str, max_tokens: int) -> str:
"""在 Token 限制内截断文本，保持句子完整"""
# 按句子截断
sentences = text.split("。")
result = []
current_tokens = 0
for sent in sentences:
sent_tokens = self._count_tokens(sent)
if current_tokens + sent_tokens > max_tokens:
break
result.append(sent)
current_tokens += sent_tokens
return "。".join(result) + "。"
第五章：查询路由与分发
5.1 查询分类与意图识别
查询路由的第一步是理解用户查询的意图，然后将其分发到最合适的处理管道。
from enum import Enum
from typing import Tuple
class QueryIntent(Enum):
FACTUAL = "factual" # 事实性查询
ANALYTICAL = "analytical" # 分析性查询
COMPARATIVE = "comparative" # 比较性查询
PROCEDURAL = "procedural" # 步骤性查询
CREATIVE = "creative" # 创造性查询
CONVERSATIONAL = "conversational" # 对话性查询
OUT_OF_SCOPE = "out_of_scope" # 超出范围
class IntentClassifier:
"""基于 LLM 的意图分类器"""
def __init__(self, llm):
self.llm = llm
def classify(self, query: str) -> Tuple[QueryIntent, float]:
"""分类查询意图"""
prompt = f"""请分析以下用户查询的意图类型，并返回 JSON 格式的结果。
查询：{query}
意图类型定义：
- factual: 询问具体事实，如"什么是量子计算"
- analytical: 需要分析推理，如"分析 AI 对就业市场的影响"
- comparative: 比较两个或多个事物，如"Python 和 Java 哪个更适合初学者"
- procedural: 询问操作步骤，如"如何配置 Docker 环境"
- creative: 需要创造性回答，如"帮我写一个产品介绍"
- conversational: 日常对话，如"你好"、"谢谢"
- out_of_scope: 超出知识范围
请返回：{{"intent": "类型", "confidence": 0.0-1.0}}"""
response = self.llm.invoke(prompt)
# 解析 JSON 响应（简化处理）
import json
try:
result = json.loads(response)
intent = QueryIntent(result["intent"])
confidence = float(result["confidence"])
return intent, confidence
except:
return QueryIntent.FACTUAL, 0.5
5.2 路由策略设计
根据查询意图，将请求路由到不同的处理管道：
class QueryRouter:
"""查询路由器"""
def __init__(self, llm):
self.classifier = IntentClassifier(llm)
self.routes = {}
def register_route(self, intent: QueryIntent, handler):
"""注册路由处理器"""
self.routes[intent] = handler
def route(self, query: str) -> dict:
"""路由查询到合适的处理器"""
intent, confidence = self.classifier.classify(query)
handler = self.routes.get(intent)
if handler is None:
handler = self.routes.get(QueryIntent.FACTUAL) # 默认路由
return {
"intent": intent,
"confidence": confidence,
"handler": handler,
"query": query
}
# 路由处理器示例
class FactualHandler:
"""事实性查询处理器 - 使用向量检索"""
def __init__(self, retriever, llm):
self.retriever = retriever
self.llm = llm
def handle(self, query: str) -> str:
results = self.retriever.search(query, top_k=3)
context = "\n".join([r[2] for r in results])
return self.llm.invoke(f"基于以下信息回答问题：\n{context}\n\n问题：{query}")
class AnalyticalHandler:
"""分析性查询处理器 - 使用多步检索和推理"""
def __init__(self, retriever, llm):
self.retriever = retriever
self.llm = llm
def handle(self, query: str) -> str:
# 第一步：分解查询
sub_queries = self._decompose(query)
# 第二步：分别检索
all_contexts = []
for sq in sub_queries:
results = self.retriever.search(sq, top_k=3)
all_contexts.extend([r[2] for r in results])
# 第三步：综合分析
context = "\n".join(all_contexts)
return self.llm.invoke(f"请基于以下信息进行深入分析：\n{context}\n\n分析要求：{query}")
def _decompose(self, query: str) -> list:
prompt = f"请将以下复杂问题分解为 2-4 个子问题：\n{query}"
response = self.llm.invoke(prompt)
return [q.strip() for q in response.split("\n") if q.strip()]
5.3 多源分发与聚合
当查询需要从多个数据源获取信息时，需要并行分发并聚合结果：
import asyncio
from typing import List, Dict, Any
class MultiSourceDispatcher:
"""多源分发器"""
def __init__(self):
self.sources: Dict[str, Any] = {}
def register_source(self, name: str, source):
"""注册数据源"""
self.sources[name] = source
async def dispatch(
self,
query: str,
target_sources: List[str] = None
) -> Dict[str, List]:
"""并行分发查询到多个数据源"""
if target_sources is None:
target_sources = list(self.sources.keys())
tasks = []
for source_name in target_sources:
source = self.sources.get(source_name)
if source:
tasks.append(self._query_source(source_name, source, query))
results = await asyncio.gather(*tasks, return_exceptions=True)
# 聚合结果
aggregated = {}
for source_name, result in zip(target_sources, results):
if isinstance(result, Exception):
aggregated[source_name] = {"error": str(result)}
else:
aggregated[source_name] = result
return aggregated
async def _query_source(self, name: str, source, query: str):
"""查询单个数据源"""
if hasattr(source, 'search_async'):
return await source.search_async(query)
else:
return source.search(query)
第六章：自适应检索策略
6.1 查询改写（Query Rewriting）
查询改写是提升检索质量的重要手段。通过 LLM 将用户的原始查询改写为更适合检索的形式。
class QueryRewriter:
"""查询改写器"""
def __init__(self, llm):
self.llm = llm
def rewrite(self, query: str, strategy: str = "hyde") -> List[str]:
"""
查询改写策略：
- hyde: HyDE（假设性文档嵌入）
- expand: 查询扩展
- clarify: 查询澄清
- multi: 多角度改写
"""
if strategy == "hyde":
return self._hyde_rewrite(query)
elif strategy == "expand":
return self._expand_rewrite(query)
elif strategy == "clarify":
return self._clarify_rewrite(query)
elif strategy == "multi":
return self._multi_angle_rewrite(query)
return [query]
def _hyde_rewrite(self, query: str) -> List[str]:
"""HyDE: 生成假设性答案作为查询"""
prompt = f"""请根据以下问题，写一段可能的参考答案（不需要完全准确，只需要包含相关关键词和概念）。
问题：{query}
参考答案："""
hypothetical_doc = self.llm.invoke(prompt)
return [query, hypothetical_doc]
def _expand_rewrite(self, query: str) -> List[str]:
"""查询扩展：生成同义或相关的查询"""
prompt = f"""请为以下查询生成 3 个不同表述方式，保持含义相同但使用不同的关键词。
原始查询：{query}
请每行返回一个改写版本："""
response = self.llm.invoke(prompt)
rewrites = [q.strip() for q in response.split("\n") if q.strip()]
return [query] + rewrites[:3]
def _multi_angle_rewrite(self, query: str) -> List[str]:
"""多角度改写：从不同角度重新表述查询"""
prompt = f"""请从以下 3 个不同角度改写查询，每个角度一行：
1. 更具体的角度
2. 更宏观的角度
3. 换一个相关的切入点
原始查询：{query}"""
response = self.llm.invoke(prompt)
rewrites = [q.strip() for q in response.split("\n") if q.strip()]
return [query] + rewrites[:3]
6.2 查询分解（Query Decomposition）
对于复杂查询，将其分解为多个子查询分别处理：
class QueryDecomposer:
"""查询分解器"""
def __init__(self, llm):
self.llm = llm
def decompose(self, query: str) -> List[str]:
"""将复杂查询分解为子查询"""
prompt = f"""请将以下复杂问题分解为多个独立的子问题，每个子问题应该能够单独检索和回答。
原始问题：{query}
要求：
1. 每个子问题应该是独立的、可检索的
2. 子问题的答案组合起来应该能回答原始问题
3. 返回 2-5 个子问题，每行一个
子问题列表："""
response = self.llm.invoke(prompt)
sub_queries = [q.strip() for q in response.split("\n") if q.strip()]
# 过滤和清理
sub_queries = [q.lstrip("0123456789.-、") for q in sub_queries]
sub_queries = [q for q in sub_queries if len(q) > 5]
return sub_queries if sub_queries else [query]
def decompose_with_dependency(self, query: str) -> List[Dict]:
"""带依赖关系的查询分解"""
prompt = f"""请将以下问题分解为子问题，并标注依赖关系。
问题：{query}
请以 JSON 格式返回：
[
{{"id": 1, "question": "子问题1", "depends_on": []}},
{{"id": 2, "question": "子问题2", "depends_on": [1]}}
]"""
response = self.llm.invoke(prompt)
import json
try:
return json.loads(response)
except:
return [{"id": 1, "question": query, "depends_on": []}]
6.3 自适应检索深度控制
根据查询的复杂度和检索结果的质量，动态调整检索深度：
class AdaptiveDepthController:
"""自适应检索深度控制器"""
def __init__(self, llm, min_confidence: float = 0.7, max_iterations: int = 5):
self.llm = llm
self.min_confidence = min_confidence
self.max_iterations = max_iterations
def assess_complexity(self, query: str) -> float:
"""评估查询复杂度 (0-1)"""
prompt = f"""请评估以下查询的复杂度，返回 0 到 1 之间的分数。
0 = 非常简单（单步检索即可回答）
1 = 非常复杂（需要多步推理和多次检索）
查询：{query}
复杂度分数："""
response = self.llm.invoke(prompt)
try:
return float(response.strip())
except:
return 0.5
def evaluate_results(self, query: str, results: List[str]) -> float:
"""评估检索结果的质量 (0-1)"""
context = "\n---\n".join(results[:3])
prompt = f"""请评估以下检索结果能否充分回答问题，返回 0 到 1 之间的分数。
0 = 完全无关或不足
1 = 非常充分，可以直接回答
问题：{query}
检索结果：
{context}
质量分数："""
response = self.llm.invoke(prompt)
try:
return float(response.strip())
except:
return 0.5
def decide_action(
self,
query: str,
results: List[str],
iteration: int
) -> str:
"""决定下一步行动"""
if iteration >= self.max_iterations:
return "generate"
quality = self.evaluate_results(query, results)
if quality >= self.min_confidence:
return "generate"
elif quality >= 0.4:
return "refine_query" # 改写查询重新检索
else:
return "expand_search" # 扩大检索范围
第七章：多步推理与检索循环
7.1 推理-检索交互模式
Agentic RAG 的核心是推理与检索的交替循环。Agent 在推理过程中发现信息缺口，然后进行有针对性的检索。
class ReasoningRetrievalLoop:
"""推理-检索循环引擎"""
def __init__(self, llm, retriever, max_steps: int = 5):
self.llm = llm
self.retriever = retriever
self.max_steps = max_steps
def run(self, query: str) -> Dict[str, Any]:
"""执行推理-检索循环"""
state = {
"query": query,
"reasoning_chain": [],
"evidence": [],
"current_answer": None,
"needs_more_info": True,
"step": 0
}
while state["needs_more_info"] and state["step"] < self.max_steps:
state["step"] += 1
# 推理步骤：分析当前信息，判断是否足够
reasoning_result = self._reason(state)
state["reasoning_chain"].append(reasoning_result)
if reasoning_result["sufficient"]:
state["current_answer"] = reasoning_result["answer"]
state["needs_more_info"] = False
else:
# 检索步骤：获取更多信息
search_queries = reasoning_result["search_queries"]
for sq in search_queries:
results = self.retriever.search(sq, top_k=3)
state["evidence"].extend(results)
return state
def _reason(self, state: Dict) -> Dict:
"""推理步骤"""
evidence_text = "\n".join([e[2] for e in state["evidence"]])
prompt = f"""你是一个研究助手，正在回答以下问题。
原始问题：{state['query']}
已收集的证据：
{evidence_text if evidence_text else "暂无"}
当前推理步骤：{state['step']}/{self.max_steps}
请分析：
1. 已有信息是否足够回答问题？
2. 如果足够，请给出完整答案。
3. 如果不足，还需要检索哪些信息？请给出具体的搜索查询。
请以 JSON 格式返回：
{{
"sufficient": true/false,
"answer": "如果足够则填写答案",
"search_queries": ["如果不足则填写需要搜索的查询"],
"reasoning": "你的推理过程"
}}"""
response = self.llm.invoke(prompt)
import json
try:
return json.loads(response)
except:
return {
"sufficient": True,
"answer": response,
"search_queries": [],
"reasoning": "无法解析结构化输出"
}
7.2 迭代检索循环设计
一个完整的迭代检索循环包含以下阶段：
class IterativeRetrievalEngine:
"""迭代检索引擎"""
def __init__(self, llm, hybrid_retriever, reranker, compressor):
self.llm = llm
self.retriever = hybrid_retriever
self.reranker = reranker
self.compressor = compressor
self.rewriter = QueryRewriter(llm)
self.decomposer = QueryDecomposer(llm)
self.depth_controller = AdaptiveDepthController(llm)
def process(self, query: str) -> Dict[str, Any]:
"""处理查询的完整流程"""
# 1. 评估查询复杂度
complexity = self.depth_controller.assess_complexity(query)
# 2. 根据复杂度选择策略
if complexity < 0.3:
return self._simple_retrieval(query)
elif complexity < 0.7:
return self._multi_query_retrieval(query)
else:
return self._decomposed_retrieval(query)
def _simple_retrieval(self, query: str) -> Dict:
"""简单检索：单次检索 + 重排序"""
# 查询改写
queries = self.rewriter.rewrite(query, strategy="hyde")
# 检索
all_results = []
for q in queries:
results = self.retriever.search(q, top_k=5)
all_results.extend(results)
# 去重和重排序
unique_docs = list(set([r[2] for r in all_results]))
reranked = self.reranker.rerank(query, unique_docs, top_k=3)
# 压缩
context = self.compressor.compress(query, "\n".join([r[2] for r in reranked]))
# 生成
answer = self._generate(query, context)
return {
"answer": answer,
"strategy": "simple",
"sources": [r[2][:100] for r in reranked]
}
def _multi_query_retrieval(self, query: str) -> Dict:
"""多查询检索：多角度改写 + 并行检索"""
queries = self.rewriter.rewrite(query, strategy="multi")
all_results = []
for q in queries:
results = self.retriever.search(q, top_k=5)
all_results.extend(results)
# RRF 融合
unique_docs = list(set([r[2] for r in all_results]))
reranked = self.reranker.rerank(query, unique_docs, top_k=5)
context = "\n".join([r[2] for r in reranked])
answer = self._generate(query, context)
return {
"answer": answer,
"strategy": "multi_query",
"sub_queries": queries,
"sources": [r[2][:100] for r in reranked]
}
def _decomposed_retrieval(self, query: str) -> Dict:
"""分解检索：查询分解 + 分别检索 + 综合"""
sub_queries = self.decomposer.decompose(query)
sub_answers = []
all_sources = []
for sq in sub_queries:
results = self.retriever.search(sq, top_k=3)
context = "\n".join([r[2] for r in results])
sub_answer = self._generate(sq, context)
sub_answers.append({"question": sq, "answer": sub_answer})
all_sources.extend(results)
# 综合所有子答案
synthesis_prompt = f"""请综合以下子问题的答案，形成对原始问题的完整回答。
原始问题：{query}
子问题及答案：
{chr(10).join([f"Q: {sa['question']}\\nA: {sa['answer']}" for sa in sub_answers])}
综合回答："""
final_answer = self.llm.invoke(synthesis_prompt)
return {
"answer": final_answer,
"strategy": "decomposed",
"sub_answers": sub_answers,
"sources": [r[2][:100] for r in all_sources[:5]]
}
def _generate(self, query: str, context: str) -> str:
"""生成回答"""
prompt = f"""请基于以下参考资料回答问题。如果资料不足，请说明。
参考资料：
{context}
问题：{query}
回答："""
return self.llm.invoke(prompt)
7.3 推理链追踪与可观测性
在生产环境中，追踪 Agent 的推理过程至关重要：
import time
import logging
from dataclasses import dataclass, field
from typing import List, Dict, Any
from datetime import datetime
@dataclass
class TraceStep:
step_id: int
action: str
input_data: Dict[str, Any]
output_data: Dict[str, Any]
duration_ms: float
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
class TracingEngine:
"""推理链追踪引擎"""
def __init__(self):
self.traces: List[TraceStep] = []
self.step_counter = 0
self.logger = logging.getLogger("agentic_rag_trace")
def trace(self, action: str):
"""追踪装饰器"""
def decorator(func):
def wrapper(*args, **kwargs):
self.step_counter += 1
start = time.time()
result = func(*args, **kwargs)
duration = (time.time() - start) * 1000
step = TraceStep(
step_id=self.step_counter,
action=action,
input_data={"args": str(args)[:200], "kwargs": str(kwargs)[:200]},
output_data={"result": str(result)[:500]},
duration_ms=duration
)
self.traces.append(step)
self.logger.info(f"[Step {step.step_id}] {action}: {duration:.1f}ms")
return result
return wrapper
return decorator
def get_trace_summary(self) -> Dict:
"""获取追踪摘要"""
total_duration = sum(t.duration_ms for t in self.traces)
return {
"total_steps": len(self.traces),
"total_duration_ms": total_duration,
"steps": [
{
"step": t.step_id,
"action": t.action,
"duration_ms": t.duration_ms
}
for t in self.traces
]
}
第八章：Self-RAG 与 Corrective RAG
8.1 Self-RAG 原理与实现
Self-RAG 是一种让 LLM 自主决定何时检索、如何使用检索结果、以及评估生成质量的框架。其核心是通过"反思标记"（Reflection Tokens）来控制检索和生成过程。
Self-RAG 的四种反思标记：
- [Retrieve]：是否需要检索
- [ISREL]：检索结果是否相关
- [ISSUP]：生成内容是否有检索结果支持
- [ISUSE]：生成内容是否对用户有用
class SelfRAG:
"""Self-RAG 实现"""
def __init__(self, llm, retriever):
self.llm = llm
self.retriever = retriever
def generate(self, query: str) -> Dict[str, Any]:
"""Self-RAG 生成流程"""
# 步骤 1：判断是否需要检索
retrieve_decision = self._decide_retrieve(query)
if retrieve_decision["need_retrieve"]:
# 步骤 2：执行检索
passages = self.retriever.search(query, top_k=5)
# 步骤 3：评估每个段落的相关性
relevant_passages = []
for passage in passages:
relevance = self._assess_relevance(query, passage[2])
if relevance["is_relevant"]:
relevant_passages.append({
"content": passage[2],
"relevance_score": relevance["score"]
})
# 步骤 4：基于相关段落生成回答
if relevant_passages:
generation = self._generate_with_passages(query, relevant_passages)
# 步骤 5：评估生成质量
support_check = self._check_support(generation["answer"], relevant_passages)
utility_check = self._check_utility(query, generation["answer"])
return {
"answer": generation["answer"],
"used_retrieval": True,
"relevant_passages": len(relevant_passages),
"is_supported": support_check["is_supported"],
"utility_score": utility_check["score"],
"confidence": min(
support_check["score"],
utility_check["score"]
)
}
# 不需要检索或无相关段落，直接生成
direct_answer = self._generate_direct(query)
return {
"answer": direct_answer,
"used_retrieval": False,
"confidence": 0.5
}
def _decide_retrieve(self, query: str) -> Dict:
"""判断是否需要检索"""
prompt = f"""请判断以下问题是否需要从外部知识库检索信息来回答。
问题：{query}
判断标准：
- 需要检索：涉及专业知识、具体事实、最新信息、数据统计
- 不需要检索：简单问候、常识问题、创意写作、观点表达
请返回 JSON：{{"need_retrieve": true/false, "reason": "原因"}}"""
response = self.llm.invoke(prompt)
import json
try:
return json.loads(response)
except:
return {"need_retrieve": True, "reason": "默认检索"}
def _assess_relevance(self, query: str, passage: str) -> Dict:
"""评估段落与查询的相关性"""
prompt = f"""请判断以下段落是否与问题相关。
问题：{query}
段落：{passage[:500]}
请返回 JSON：{{"is_relevant": true/false, "score": 0.0-1.0, "reason": "原因"}}"""
response = self.llm.invoke(prompt)
import json
try:
return json.loads(response)
except:
return {"is_relevant": True, "score": 0.5}
def _check_support(self, answer: str, passages: List[Dict]) -> Dict:
"""检查生成内容是否有检索结果支持"""
context = "\n".join([p["content"][:200] for p in passages])
prompt = f"""请判断以下回答是否被参考资料所支持。
回答：{answer}
参考资料：{context}
请返回 JSON：{{"is_supported": true/false, "score": 0.0-1.0}}"""
response = self.llm.invoke(prompt)
import json
try:
return json.loads(response)
except:
return {"is_supported": True, "score": 0.5}
def _check_utility(self, query: str, answer: str) -> Dict:
"""检查回答对用户的有用性"""
prompt = f"""请评估以下回答对用户的有用程度。
问题：{query}
回答：{answer}
请返回 JSON：{{"score": 0.0-1.0, "reason": "原因"}}"""
response = self.llm.invoke(prompt)
import json
try:
return json.loads(response)
except:
return {"score": 0.5}
def _generate_with_passages(self, query: str, passages: List[Dict]) -> Dict:
"""基于检索结果生成回答"""
context = "\n\n".join([p["content"] for p in passages])
prompt = f"""请基于以下参考资料回答问题。
参考资料：
{context}
问题：{query}
回答："""
answer = self.llm.invoke(prompt)
return {"answer": answer}
def _generate_direct(self, query: str) -> str:
"""直接生成回答（不使用检索）"""
return self.llm.invoke(f"请回答以下问题：\n{query}")
8.2 Corrective RAG（CRAG）原理与实现
Corrective RAG 的核心思想是：在检索结果不理想时，自动触发纠正机制，通过网络搜索或知识精炼来改善结果质量。
class CorrectiveRAG:
"""Corrective RAG 实现"""
def __init__(self, llm, retriever, web_searcher=None):
self.llm = llm
self.retriever = retriever
self.web_searcher = web_searcher
def generate(self, query: str) -> Dict[str, Any]:
"""CRAG 生成流程"""
# 步骤 1：检索
retrieved_docs = self.retriever.search(query, top_k=5)
# 步骤 2：评估检索质量
quality_assessment = self._assess_retrieval_quality(query, retrieved_docs)
# 步骤 3：根据质量决定纠正策略
if quality_assessment["status"] == "correct":
# 检索结果良好，直接使用
refined_docs = self._knowledge_refinement(query, retrieved_docs)
answer = self._generate_answer(query, refined_docs)
strategy = "direct_use"
elif quality_assessment["status"] == "ambiguous":
# 检索结果模糊，需要精炼
refined_docs = self._knowledge_refinement(query, retrieved_docs)
answer = self._generate_answer(query, refined_docs)
strategy = "refined"
elif quality_assessment["status"] == "incorrect":
# 检索结果不佳，触发纠正
if self.web_searcher:
# 使用网络搜索补充
web_results = self.web_searcher.search(query)
all_docs = retrieved_docs + web_results
refined_docs = self._knowledge_refinement(query, all_docs)
answer = self._generate_answer(query, refined_docs)
strategy = "web_corrected"
else:
# 无网络搜索，尽力而为
refined_docs = self._knowledge_refinement(query, retrieved_docs)
answer = self._generate_answer(query, refined_docs)
strategy = "best_effort"
else:
answer = self._generate_answer(query, retrieved_docs)
strategy = "fallback"
return {
"answer": answer,
"strategy": strategy,
"retrieval_quality": quality_assessment,
"sources_count": len(retrieved_docs)
}
def _assess_retrieval_quality(self, query: str, docs: List) -> Dict:
"""评估检索结果的质量"""
docs_text = "\n---\n".join([d[2][:300] for d in docs[:3]])
prompt = f"""请评估以下检索结果对回答问题的质量。
问题：{query}
检索结果：
{docs_text}
评估标准：
- correct: 检索结果高度相关且充分
- ambiguous: 检索结果相关但不够充分或有歧义
- incorrect: 检索结果不相关或质量很差
请返回 JSON：{{"status": "correct/ambiguous/incorrect", "score": 0.0-1.0, "reason": "原因"}}"""
response = self.llm.invoke(prompt)
import json
try:
return json.loads(response)
except:
return {"status": "ambiguous", "score": 0.5}
def _knowledge_refinement(self, query: str, docs: List) -> List[str]:
"""知识精炼：提取关键信息片段"""
refined = []
for doc in docs:
content = doc[2] if isinstance(doc, tuple) else doc
prompt = f"""请从以下文本中提取与问题相关的关键信息片段。
只保留直接相关的句子，去除无关内容。
问题：{query}
文本：{content[:500]}
关键信息："""
key_info = self.llm.invoke(prompt)
if key_info.strip():
refined.append(key_info.strip())
return refined
def _generate_answer(self, query: str, context_items: List[str]) -> str:
"""生成最终答案"""
context = "\n\n".join(context_items)
prompt = f"""请基于以下参考资料回答问题。请确保答案准确、完整。
参考资料：
{context}
问题：{query}
回答："""
return self.llm.invoke(prompt)
8.3 两种范式的对比与融合
| 维度 | Self-RAG | Corrective RAG |
|---|---|---|
| 核心思想 | 自主决定何时检索和如何使用结果 | 评估检索质量并触发纠正 |
| 检索决策 | 由 LLM 动态决定是否检索 | 总是检索，但评估结果质量 |
| 纠正机制 | 通过反思标记评估生成质量 | 通过知识精炼和网络搜索纠正 |
| 适用场景 | 简单到中等复杂度的查询 | 检索质量不稳定的数据源 |
| 实现复杂度 | 中等 | 较高 |
融合方案：
class HybridSelfCorrectiveRAG:
"""融合 Self-RAG 和 CRAG 的混合方案"""
def __init__(self, llm, retriever, web_searcher=None):
self.self_rag = SelfRAG(llm, retriever)
self.crag = CorrectiveRAG(llm, retriever, web_searcher)
self.llm = llm
def generate(self, query: str) -> Dict[str, Any]:
# 先用 Self-RAG 的检索决策
decision = self.self_rag._decide_retrieve(query)
if not decision["need_retrieve"]:
# 不需要检索，直接生成
answer = self.llm.invoke(f"请回答：{query}")
return {"answer": answer, "strategy": "direct", "used_retrieval": False}
# 需要检索，使用 CRAG 的质量评估和纠正流程
result = self.crag.generate(query)
# Self-RAG 的质量检查
if result.get("retrieval_quality", {}).get("score", 0) < 0.6:
# 质量不佳，尝试 Self-RAG 的反思机制
self_rag_result = self.self_rag.generate(query)
if self_rag_result.get("confidence", 0) > result.get("retrieval_quality", {}).get("score", 0):
return self_rag_result
return result
第九章：Agent-RAG 融合模式
9.1 Agent 作为检索器
在这种模式中，Agent 本身就是检索过程的控制者，能够自主决定如何查询、查询什么、以及何时停止。
class AgentAsRetriever:
"""Agent 作为检索器的实现"""
def __init__(self, llm, tools: Dict[str, Any]):
self.llm = llm
self.tools = tools
self.max_iterations = 5
def run(self, query: str) -> Dict[str, Any]:
"""Agent 自主检索流程"""
context = {"query": query, "collected_info": [], "iterations": 0}
while context["iterations"] < self.max_iterations:
# Agent 决定下一步行动
action = self._decide_action(context)
if action["type"] == "finish":
break
# 执行行动
result = self._execute_action(action)
context["collected_info"].append({
"action": action,
"result": result
})
context["iterations"] += 1
# 生成最终答案
answer = self._synthesize(query, context["collected_info"])
return {
"answer": answer,
"iterations": context["iterations"],
"actions_taken": [c["action"]["type"] for c in context["collected_info"]]
}
def _decide_action(self, context: Dict) -> Dict:
"""Agent 决策：下一步做什么"""
tools_desc = "\n".join([
f"- {name}: {tool.description}"
for name, tool in self.tools.items()
])
info_summary = "\n".join([
f"[{c['action']['type']}] {str(c['result'])[:200]}"
for c in context["collected_info"]
]) or "暂无"
prompt = f"""你是一个智能研究助手。请根据当前状态决定下一步行动。
原始问题：{context['query']}
已收集信息：
{info_summary}
当前迭代：{context['iterations']}/{self.max_iterations}
可用工具：
{tools_desc}
请以 JSON 格式返回你的决策：
{{"type": "工具名或finish", "params": {{}}, "reasoning": "推理过程"}}"""
response = self.llm.invoke(prompt)
import json
try:
return json.loads(response)
except:
return {"type": "finish", "params": {}, "reasoning": "解析失败，结束"}
def _execute_action(self, action: Dict) -> Any:
"""执行工具调用"""
tool_name = action["type"]
params = action.get("params", {})
if tool_name in self.tools:
return self.tools[tool_name].run(**params)
return None
def _synthesize(self, query: str, collected_info: List[Dict]) -> str:
"""综合所有信息生成答案"""
info_text = "\n\n".join([
f"来源：{c['action']['type']}\n内容：{str(c['result'])[:500]}"
for c in collected_info
])
prompt = f"""请综合以下收集到的信息，回答用户的问题。
问题：{query}
收集到的信息：
{info_text}
请提供一个完整、准确的回答："""
return self.llm.invoke(prompt)
9.2 检索器作为 Agent 工具
在这种模式中，RAG 系统被封装为 Agent 可以调用的工具之一：
class RAGTool:
"""将 RAG 封装为 Agent 工具"""
def __init__(self, name: str, description: str, retriever, llm):
self.name = name
self.description = description
self.retriever = retriever
self.llm = llm
def run(self, query: str) -> str:
"""执行 RAG 检索和生成"""
results = self.retriever.search(query, top_k=5)
context = "\n".join([r[2] for r in results])
prompt = f"""基于以下参考资料回答问题。
如果资料不足，请如实说明。
参考资料：
{context}
问题：{query}
回答："""
return self.llm.invoke(prompt)
class MultiToolAgent:
"""多工具 Agent"""
def __init__(self, llm):
self.llm = llm
self.tools: Dict[str, Any] = {}
def register_tool(self, tool):
"""注册工具"""
self.tools[tool.name] = tool
def run(self, query: str) -> str:
"""Agent 执行流程"""
tools_desc = "\n".join([
f"- {name}: {tool.description}"
for name, tool in self.tools.items()
])
# ReAct 模式
prompt = f"""你是一个智能助手，可以使用以下工具：
{tools_desc}
请使用 ReAct 模式回答问题：
Thought: 思考下一步
Action: 工具名称
Action Input: 工具输入
Observation: 工具输出
... (重复直到可以给出最终答案)
Final Answer: 最终答案
问题：{query}"""
return self.llm.invoke(prompt)
# 使用示例
def create_multi_tool_agent(llm):
agent = MultiToolAgent(llm)
# 注册 RAG 工具
rag_tool = RAGTool(
name="knowledge_base",
description="从内部知识库检索信息",
retriever=HybridRetriever(),
llm=llm
)
agent.register_tool(rag_tool)
# 可以注册更多工具
# agent.register_tool(web_search_tool)
# agent.register_tool(calculator_tool)
# agent.register_tool(sql_query_tool)
return agent
9.3 多 Agent 协作 RAG
在复杂场景中，多个专业化 Agent 协作完成 RAG 任务：
class SpecializedAgent:
"""专业化 Agent 基类"""
def __init__(self, name: str, role: str, llm):
self.name = name
self.role = role
self.llm = llm
def process(self, task: str, context: Dict = None) -> Dict:
raise NotImplementedError
class ResearchAgent(SpecializedAgent):
"""研究 Agent：负责信息检索"""
def __init__(self, llm, retrievers: Dict[str, Any]):
super().__init__("researcher", "信息检索专家", llm)
self.retrievers = retrievers
def process(self, task: str, context: Dict = None) -> Dict:
all_results = {}
for name, retriever in self.retrievers.items():
results = retriever.search(task, top_k=3)
all_results[name] = results
return {
"agent": self.name,
"findings": all_results,
"summary": f"从 {len(all_results)} 个数据源检索到信息"
}
class AnalysisAgent(SpecializedAgent):
"""分析 Agent：负责推理分析"""
def __init__(self, llm):
super().__init__("analyst", "推理分析专家", llm)
def process(self, task: str, context: Dict = None) -> Dict:
findings = context.get("findings", {}) if context else {}
findings_text = "\n\n".join([
f"来源: {source}\n内容: {str(results)[:500]}"
for source, results in findings.items()
])
prompt = f"""请基于以下研究发现进行分析。
任务：{task}
研究发现：
{findings_text}
请提供：1. 关键发现 2. 深入分析 3. 潜在问题"""
analysis = self.llm.invoke(prompt)
return {
"agent": self.name,
"analysis": analysis
}
class WritingAgent(SpecializedAgent):
"""写作 Agent：负责生成最终回答"""
def __init__(self, llm):
super().__init__("writer", "内容写作专家", llm)
def process(self, task: str, context: Dict = None) -> Dict:
analysis = context.get("analysis", "") if context else ""
prompt = f"""请基于以下分析结果，撰写一个清晰、专业的回答。
原始问题：{task}
分析结果：
{analysis}
请撰写回答："""
answer = self.llm.invoke(prompt)
return {
"agent": self.name,
"final_answer": answer
}
class MultiAgentOrchestrator:
"""多 Agent 编排器"""
def __init__(self, agents: List[SpecializedAgent]):
self.agents = agents
def run(self, query: str) -> Dict[str, Any]:
"""编排多个 Agent 协作完成任务"""
context = {"query": query}
for agent in self.agents:
result = agent.process(query, context)
context.update(result)
return context
第十章：LangGraph 实现 Agentic RAG
10.1 LangGraph 核心概念
LangGraph 是 LangChain 生态中用于构建有状态、多步骤 Agent 应用的框架。其核心概念包括：
- StateGraph：有状态的图，定义了 Agent 的状态和状态转换
- Node：图中的节点，每个节点执行一个操作
- Edge：节点之间的边，定义了状态转换的条件
- State：在节点之间传递的状态对象
10.2 用 LangGraph 构建 Agentic RAG 流程
from typing import TypedDict, Annotated, List, Dict, Any, Literal
from langgraph.graph import StateGraph, END
import operator
# 定义状态类型
class RAGState(TypedDict):
query: str
query_type: str # simple, complex, conversational
sub_queries: List[str]
retrieved_docs: Annotated[List[str], operator.add]
reranked_docs: List[str]
answer: str
confidence: float
iteration: int
needs_retrieval: bool
correction_needed: bool
def create_agentic_rag_graph(llm, retriever, reranker):
"""创建 Agentic RAG 的 LangGraph 图"""
# 定义节点函数
def classify_query(state: RAGState) -> RAGState:
"""分类查询类型"""
query = state["query"]
prompt = f"""请判断查询类型，只返回一个词：simple/complex/conversational
查询：{query}"""
query_type = llm.invoke(prompt).strip().lower()
return {
"query_type": query_type,
"iteration": state.get("iteration", 0) + 1,
"needs_retrieval": query_type != "conversational"
}
def decompose_query(state: RAGState) -> RAGState:
"""分解复杂查询"""
query = state["query"]
prompt = f"""请将以下问题分解为 2-3 个子问题，每行一个：
{query}"""
response = llm.invoke(prompt)
sub_queries = [q.strip() for q in response.split("\n") if q.strip()]
return {"sub_queries": sub_queries if sub_queries else [query]}
def retrieve(state: RAGState) -> RAGState:
"""执行检索"""
queries = state.get("sub_queries", [state["query"]])
all_docs = []
for q in queries:
results = retriever.search(q, top_k=5)
all_docs.extend([r[2] for r in results])
# 去重
unique_docs = list(set(all_docs))
return {"retrieved_docs": unique_docs}
def rerank(state: RAGState) -> RAGState:
"""重排序"""
docs = state["retrieved_docs"]
reranked = reranker.rerank(state["query"], docs, top_k=5)
return {"reranked_docs": [r[2] for r in reranked]}
def generate(state: RAGState) -> RAGState:
"""生成回答"""
context = "\n\n".join(state["reranked_docs"][:3])
prompt = f"""基于以下参考资料回答问题。
参考资料：
{context}
问题：{state['query']}
回答："""
answer = llm.invoke(prompt)
return {"answer": answer, "confidence": 0.8}
def evaluate_and_correct(state: RAGState) -> RAGState:
"""评估并纠正"""
prompt = f"""请评估以下回答的质量（0-1分数）。
问题：{state['query']}
回答：{state['answer']}
只返回分数："""
try:
score = float(llm.invoke(prompt).strip())
except:
score = 0.8
return {
"confidence": score,
"correction_needed": score < 0.7
}
# 定义条件路由
def should_retrieve(state: RAGState) -> Literal["retrieve", "direct_answer"]:
if state.get("needs_retrieval", True):
return "retrieve"
return "direct_answer"
def should_decompose(state: RAGState) -> Literal["decompose", "retrieve"]:
if state.get("query_type") == "complex":
return "decompose"
return "retrieve"
def should_correct(state: RAGState) -> Literal["retrieve", "end"]:
if state.get("correction_needed", False) and state.get("iteration", 0) < 3:
return "retrieve"
return "end"
# 构建图
graph = StateGraph(RAGState)
# 添加节点
graph.add_node("classify", classify_query)
graph.add_node("decompose", decompose_query)
graph.add_node("retrieve", retrieve)
graph.add_node("rerank", rerank)
graph.add_node("generate", generate)
graph.add_node("evaluate", evaluate_and_correct)
graph.add_node("direct_answer", lambda s: {"answer": llm.invoke(s["query"]), "confidence": 1.0})
# 添加边
graph.set_entry_point("classify")
graph.add_conditional_edges("classify", should_retrieve, {
"retrieve": "decompose",
"direct_answer": "direct_answer"
})
graph.add_conditional_edges("decompose", should_decompose, {
"decompose": "retrieve",
"retrieve": "retrieve"
})
graph.add_edge("retrieve", "rerank")
graph.add_edge("rerank", "generate")
graph.add_edge("generate", "evaluate")
graph.add_conditional_edges("evaluate", should_correct, {
"retrieve": "retrieve",
"end": END
})
graph.add_edge("direct_answer", END)
return graph.compile()
10.3 条件路由与状态机
LangGraph 的强大之处在于其条件路由能力，可以实现复杂的状态机逻辑：
def add_advanced_routing(graph: StateGraph):
"""添加高级路由逻辑"""
def route_after_evaluation(state: RAGState) -> str:
"""评估后的路由决策"""
confidence = state.get("confidence", 0)
iteration = state.get("iteration", 0)
if confidence >= 0.85:
return "high_confidence"
elif confidence >= 0.6:
if iteration < 3:
return "refine_and_retry"
else:
return "accept_and_output"
else:
if iteration < 2:
return "try_alternative_strategy"
else:
return "low_confidence_output"
# 注册路由
graph.add_conditional_edges("evaluate", route_after_evaluation, {
"high_confidence": END,
"refine_and_retry": "retrieve",
"accept_and_output": END,
"try_alternative_strategy": "decompose",
"low_confidence_output": "generate"
})
10.4 人机协作节点
在某些场景下，需要人类介入审核或提供额外信息：
class HumanInTheLoopNode:
"""人机协作节点"""
def __init__(self, llm):
self.llm = llm
self.pending_reviews = {}
def request_review(self, state: RAGState) -> Dict:
"""请求人类审核"""
review_request = {
"query": state["query"],
"draft_answer": state.get("answer", ""),
"confidence": state.get("confidence", 0),
"question": "请审核此回答是否需要修改？",
"options": ["approve", "reject", "modify"]
}
return {
"type": "human_review",
"request": review_request,
"status": "pending"
}
def process_feedback(self, feedback: Dict, state: RAGState) -> RAGState:
"""处理人类反馈"""
if feedback["action"] == "approve":
return {**state, "human_approved": True}
elif feedback["action"] == "modify":
# 使用人类的修改意见重新生成
modified_answer = self._incorporate_feedback(
state["answer"],
feedback["comments"]
)
return {**state, "answer": modified_answer, "human_approved": True}
else:
# 拒绝，重新检索
return {**state, "iteration": state.get("iteration", 0) + 1}
def _incorporate_feedback(self, answer: str, feedback: str) -> str:
"""将人类反馈融入答案"""
prompt = f"""请根据反馈修改以下答案。
原始答案：{answer}
反馈意见：{feedback}
修改后的答案："""
return self.llm.invoke(prompt)
第十一章：评估体系
11.1 检索质量评估
检索质量是 RAG 系统的基础。主要评估指标包括：
from typing import List, Dict
import numpy as np
class RetrievalEvaluator:
"""检索质量评估器"""
def __init__(self):
self.metrics_history = []
def evaluate(
self,
retrieved_docs: List[str],
relevant_docs: List[str],
k_values: List[int] = [1, 3, 5, 10]
) -> Dict[str, float]:
"""计算检索质量指标"""
results = {}
# 计算每个 k 值的指标
for k in k_values:
retrieved_at_k = retrieved_docs[:k]
# Precision@K
precision = self._precision_at_k(retrieved_at_k, relevant_docs)
results[f"precision@{k}"] = precision
# Recall@K
recall = self._recall_at_k(retrieved_at_k, relevant_docs)
results[f"recall@{k}"] = recall
# MRR (Mean Reciprocal Rank)
mrr = self._mrr(retrieved_at_k, relevant_docs)
results[f"mrr@{k}"] = mrr
# NDCG@K
for k in k_values:
ndcg = self._ndcg_at_k(retrieved_docs[:k], relevant_docs, k)
results[f"ndcg@{k}"] = ndcg
self.metrics_history.append(results)
return results
def _precision_at_k(self, retrieved: List[str], relevant: List[str]) -> float:
"""Precision@K"""
if not retrieved:
return 0.0
relevant_set = set(relevant)
hits = sum(1 for doc in retrieved if doc in relevant_set)
return hits / len(retrieved)
def _recall_at_k(self, retrieved: List[str], relevant: List[str]) -> float:
"""Recall@K"""
if not relevant:
return 0.0
relevant_set = set(relevant)
hits = sum(1 for doc in retrieved if doc in relevant_set)
return hits / len(relevant_set)
def _mrr(self, retrieved: List[str], relevant: List[str]) -> float:
"""Mean Reciprocal Rank"""
relevant_set = set(relevant)
for i, doc in enumerate(retrieved):
if doc in relevant_set:
return 1.0 / (i + 1)
return 0.0
def _ndcg_at_k(self, retrieved: List[str], relevant: List[str], k: int) -> float:
"""Normalized Discounted Cumulative Gain"""
relevant_set = set(relevant)
# DCG
dcg = 0.0
for i, doc in enumerate(retrieved[:k]):
rel = 1.0 if doc in relevant_set else 0.0
dcg += rel / np.log2(i + 2)
# Ideal DCG
ideal_rels = sorted(
[1.0 if doc in relevant_set else 0.0 for doc in retrieved[:k]],
reverse=True
)
idcg = sum(rel / np.log2(i + 2) for i, rel in enumerate(ideal_rels))
return dcg / idcg if idcg > 0 else 0.0
def get_average_metrics(self) -> Dict[str, float]:
"""获取平均指标"""
if not self.metrics_history:
return {}
avg_metrics = {}
for key in self.metrics_history[0].keys():
values = [m[key] for m in self.metrics_history]
avg_metrics[key] = np.mean(values)
return avg_metrics
11.2 生成质量评估
class GenerationEvaluator:
"""生成质量评估器"""
def __init__(self, llm=None):
self.llm = llm
def evaluate(
self,
query: str,
answer: str,
reference: str = None,
context: str = None
) -> Dict[str, float]:
"""评估生成质量"""
results = {}
# 相关性评估
results["relevance"] = self._evaluate_relevance(query, answer)
# 忠实度评估（如果有上下文）
if context:
results["faithfulness"] = self._evaluate_faithfulness(answer, context)
# 完整性评估
results["completeness"] = self._evaluate_completeness(query, answer)
# 如果有参考答案，计算相似度
if reference:
results["reference_similarity"] = self._evaluate_similarity(answer, reference)
return results
def _evaluate_relevance(self, query: str, answer: str) -> float:
"""评估答案与问题的相关性"""
if not self.llm:
return 0.5
prompt = f"""请评估以下回答与问题的相关性，返回 0-1 的分数。
问题：{query}
回答：{answer}
相关性分数："""
try:
return float(self.llm.invoke(prompt).strip())
except:
return 0.5
def _evaluate_faithfulness(self, answer: str, context: str) -> float:
"""评估答案对上下文的忠实度"""
if not self.llm:
return 0.5
prompt = f"""请评估以下回答是否忠实于提供的参考资料。
只评估事实准确性，不评估完整性。
回答：{answer}
参考资料：{context[:1000]}
忠实度分数（0-1）："""
try:
return float(self.llm.invoke(prompt).strip())
except:
return 0.5
def _evaluate_completeness(self, query: str, answer: str) -> float:
"""评估答案的完整性"""
if not self.llm:
return 0.5
prompt = f"""请评估以下回答是否完整地回答了问题。
问题：{query}
回答：{answer}
完整性分数（0-1）："""
try:
return float(self.llm.invoke(prompt).strip())
except:
return 0.5
def _evaluate_similarity(self, answer: str, reference: str) -> float:
"""评估答案与参考答案的相似度"""
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("BAAI/bge-large-zh-v1.5")
answer_emb = model.encode([answer], normalize_embeddings=True)
ref_emb = model.encode([reference], normalize_embeddings=True)
return float(np.dot(answer_emb, ref_emb.T)[0][0])
11.3 端到端评估框架
class EndToEndEvaluator:
"""端到端评估框架"""
def __init__(self, llm=None):
self.retrieval_evaluator = RetrievalEvaluator()
self.generation_evaluator = GenerationEvaluator(llm)
self.results = []
def evaluate_single(
self,
query: str,
retrieved_docs: List[str],
relevant_docs: List[str],
generated_answer: str,
reference_answer: str = None,
context: str = None
) -> Dict:
"""评估单个查询"""
result = {
"query": query,
"retrieval": self.retrieval_evaluator.evaluate(retrieved_docs, relevant_docs),
"generation": self.generation_evaluator.evaluate(
query, generated_answer, reference_answer, context
)
}
# 计算综合分数
result["overall"] = self._calculate_overall_score(result)
self.results.append(result)
return result
def evaluate_batch(self, test_cases: List[Dict]) -> Dict:
"""批量评估"""
for case in test_cases:
self.evaluate_single(**case)
return self.get_summary()
def _calculate_overall_score(self, result: Dict) -> float:
"""计算综合分数"""
retrieval_score = np.mean([
v for k, v in result["retrieval"].items()
if "ndcg@5" in k or "mrr@5" in k
]) if result["retrieval"] else 0.5
generation_score = np.mean([
v for v in result["generation"].values()
]) if result["generation"] else 0.5
return 0.4 * retrieval_score + 0.6 * generation_score
def get_summary(self) -> Dict:
"""获取评估摘要"""
if not self.results:
return {}
summary = {
"total_queries": len(self.results),
"avg_retrieval_metrics": self._average_metrics("retrieval"),
"avg_generation_metrics": self._average_metrics("generation"),
"avg_overall_score": np.mean([r["overall"] for r in self.results])
}
return summary
def _average_metrics(self, category: str) -> Dict[str, float]:
"""计算某类指标的平均值"""
all_metrics = {}
for result in self.results:
for key, value in result[category].items():
if key not in all_metrics:
all_metrics[key] = []
all_metrics[key].append(value)
return {key: np.mean(values) for key, values in all_metrics.items()}
11.4 在线评估与 A/B 测试
class OnlineEvaluator:
"""在线评估器"""
def __init__(self):
self.feedback_log = []
def log_interaction(
self,
query: str,
answer: str,
user_feedback: Dict = None
):
"""记录交互数据"""
entry = {
"query": query,
"answer": answer,
"timestamp": datetime.now().isoformat(),
"feedback": user_feedback
}
self.feedback_log.append(entry)
def calculate_satisfaction_rate(self, time_window_hours: int = 24) -> float:
"""计算用户满意度"""
cutoff = datetime.now().timestamp() - time_window_hours * 3600
recent = [
e for e in self.feedback_log
if datetime.fromisoformat(e["timestamp"]).timestamp() > cutoff
]
if not recent:
return 0.0
positive = sum(
1 for e in recent
if e.get("feedback", {}).get("rating", 0) >= 4
)
return positive / len(recent)
def identify_failure_cases(self, threshold: float = 2.0) -> List[Dict]:
"""识别失败案例（低评分）"""
return [
e for e in self.feedback_log
if e.get("feedback", {}).get("rating", 5) <= threshold
]
第十二章：实战项目一 —— 智能研究助手
12.1 项目需求与架构设计
项目目标：构建一个能够自主进行深度研究的智能助手，支持多轮检索、信息综合、自动生成研究报告。
核心功能：
- 理解研究主题并自动规划研究策略
- 从多个数据源检索相关信息
- 迭代深化研究，直到信息充分
- 综合所有发现，生成结构化的研究报告
系统架构：
┌─────────────────────────────────────────────┐
│ 智能研究助手 │
├──────────┬──────────┬──────────┬─────────────┤
│ 研究规划 │ 信息检索 │ 内容分析 │ 报告生成 │
│ Agent │ Pipeline │ Agent │ Pipeline │
├──────────┴──────────┴──────────┴─────────────┤
│ 混合检索引擎 │
│ 向量检索 | 关键词检索 | 网络搜索 │
├──────────────────────────────────────────────┤
│ 知识存储层 │
│ 向量数据库 | 文档存储 | 研究笔记 │
└──────────────────────────────────────────────┘
12.2 核心代码实现
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import json
# ========== 数据模型 ==========
class ResearchPhase(Enum):
PLANNING = "planning"
GATHERING = "gathering"
ANALYZING = "analyzing"
SYNTHESIZING = "synthesizing"
REPORTING = "reporting"
COMPLETED = "completed"
@dataclass
class ResearchNote:
content: str
source: str
relevance: float
category: str
timestamp: str = ""
@dataclass
class ResearchPlan:
topic: str
sub_topics: List[str]
search_queries: List[str]
current_phase: ResearchPhase
notes: List[ResearchNote] = field(default_factory=list)
iteration: int = 0
max_iterations: int = 5
is_sufficient: bool = False
# ========== 核心组件 ==========
class ResearchPlanner:
"""研究规划器"""
def __init__(self, llm):
self.llm = llm
def create_plan(self, topic: str) -> ResearchPlan:
"""创建研究计划"""
prompt = f"""请为以下研究主题制定详细的研究计划。
研究主题：{topic}
请以 JSON 格式返回：
{{
"sub_topics": ["子主题1", "子主题2", ...],
"search_queries": ["搜索查询1", "搜索查询2", ...],
"research_questions": ["研究问题1", "研究问题2", ...]
}}"""
response = self.llm.invoke(prompt)
try:
plan_data = json.loads(response)
except:
plan_data = {
"sub_topics": [topic],
"search_queries": [topic],
"research_questions": [f"什么是{topic}？"]
}
return ResearchPlan(
topic=topic,
sub_topics=plan_data.get("sub_topics", []),
search_queries=plan_data.get("search_queries", []),
current_phase=ResearchPhase.GATHERING
)
def evaluate_sufficiency(self, plan: ResearchPlan) -> Dict:
"""评估已收集的信息是否充分"""
notes_summary = "\n".join([
f"- [{n.category}] {n.content[:100]}"
for n in plan.notes[:20]
])
prompt = f"""请评估以下研究收集的信息是否足以撰写研究报告。
研究主题：{plan.topic}
已收集的笔记：
{notes_summary}
请返回 JSON：
{{"is_sufficient": true/false, "missing_aspects": ["缺失方面1", ...], "confidence": 0.0-1.0}}"""
response = self.llm.invoke(prompt)
try:
return json.loads(response)
except:
return {"is_sufficient": len(plan.notes) >= 10, "missing_aspects": [], "confidence": 0.5}
class InformationGatherer:
"""信息收集器"""
def __init__(self, llm, retrievers: Dict[str, Any]):
self.llm = llm
self.retrievers = retrievers
def gather(self, queries: List[str]) -> List[ResearchNote]:
"""收集信息"""
notes = []
for query in queries:
for source_name, retriever in self.retrievers.items():
try:
results = retriever.search(query, top_k=3)
for result in results:
note = ResearchNote(
content=result[2],
source=source_name,
relevance=result[1],
category=self._categorize(query, result[2])
)
notes.append(note)
except Exception as e:
print(f"检索 {source_name} 失败: {e}")
return notes
def _categorize(self, query: str, content: str) -> str:
"""对内容进行分类"""
prompt = f"""请为以下内容分配一个类别标签。
查询：{query}
内容：{content[:200]}
类别选项：背景信息、核心概念、技术细节、案例分析、数据统计、专家观点、其他
只返回类别名称："""
return self.llm.invoke(prompt).strip()
class ReportGenerator:
"""报告生成器"""
def __init__(self, llm):
self.llm = llm
def generate_report(self, plan: ResearchPlan) -> str:
"""生成研究报告"""
# 按类别组织笔记
categorized_notes = {}
for note in plan.notes:
if note.category not in categorized_notes:
categorized_notes[note.category] = []
categorized_notes[note.category].append(note)
# 构建报告大纲
outline = self._create_outline(plan.topic, categorized_notes)
# 生成各章节
sections = []
for section_title in outline["sections"]:
section_content = self._generate_section(
plan.topic, section_title, categorized_notes
)
sections.append(f"## {section_title}\n\n{section_content}")
# 组装报告
report = f"""# {plan.topic} - 研究报告
## 摘要
{outline.get('abstract', '')}
{chr(10).join(sections)}
## 参考来源
{self._format_sources(plan.notes)}
"""
return report
def _create_outline(self, topic: str, categorized_notes: Dict) -> Dict:
"""创建报告大纲"""
categories = list(categorized_notes.keys())
prompt = f"""请为以下研究主题创建报告大纲。
主题：{topic}
已有信息类别：{', '.join(categories)}
请以 JSON 格式返回：
{{"abstract": "摘要内容", "sections": ["章节1标题", "章节2标题", ...]}}"""
response = self.llm.invoke(prompt)
try:
return json.loads(response)
except:
return {
"abstract": f"本报告对{topic}进行了系统研究。",
"sections": ["背景介绍", "核心内容", "分析与讨论", "结论与展望"]
}
def _generate_section(
self, topic: str, section_title: str, categorized_notes: Dict
) -> str:
"""生成报告章节"""
# 收集相关笔记
relevant_notes = []
for category, notes in categorized_notes.items():
for note in notes:
relevant_notes.append(note.content[:300])
context = "\n---\n".join(relevant_notes[:10])
prompt = f"""请基于以下参考资料，撰写报告的一个章节。
研究报告主题：{topic}
章节标题：{section_title}
参考资料：
{context}
请撰写该章节的内容（约 500 字）："""
return self.llm.invoke(prompt)
def _format_sources(self, notes: List[ResearchNote]) -> str:
"""格式化参考来源"""
sources = list(set([f"- {n.source}" for n in notes]))
return "\n".join(sources[:20])
# ========== 主系统 ==========
class SmartResearchAssistant:
"""智能研究助手主系统"""
def __init__(self, llm, retrievers: Dict[str, Any]):
self.llm = llm
self.planner = ResearchPlanner(llm)
self.gatherer = InformationGatherer(llm, retrievers)
self.reporter = ReportGenerator(llm)
def research(self, topic: str) -> str:
"""执行完整的研究流程"""
print(f"📚 开始研究: {topic}")
# 阶段 1：创建研究计划
plan = self.planner.create_plan(topic)
print(f"📋 研究计划: {len(plan.sub_topics)} 个子主题, {len(plan.search_queries)} 个搜索查询")
# 阶段 2：迭代收集信息
while plan.iteration < plan.max_iterations and not plan.is_sufficient:
plan.iteration += 1
print(f"🔍 第 {plan.iteration} 轮信息收集...")
# 收集信息
new_notes = self.gatherer.gather(plan.search_queries)
plan.notes.extend(new_notes)
print(f" 收集到 {len(new_notes)} 条新信息")
# 评估信息充分性
sufficiency = self.planner.evaluate_sufficiency(plan)
if sufficiency.get("is_sufficient", False):
plan.is_sufficient = True
print("✅ 信息收集充分")
else:
# 根据缺失方面调整搜索策略
missing = sufficiency.get("missing_aspects", [])
if missing:
plan.search_queries = missing
print(f" 需要补充: {', '.join(missing[:3])}")
# 阶段 3：生成研究报告
print("📝 生成研究报告...")
report = self.reporter.generate_report(plan)
print(f"✅ 研究完成！共收集 {len(plan.notes)} 条信息，迭代 {plan.iteration} 轮")
return report
12.3 部署与测试
# 使用示例
def test_research_assistant():
"""测试智能研究助手"""
# 初始化组件（需要替换为实际的 LLM 和检索器）
# llm = YourLLM()
# vector_retriever = DenseRetriever()
# keyword_retriever = SparseRetriever()
# 创建助手
# assistant = SmartResearchAssistant(
# llm=llm,
# retrievers={
# "vector_db": vector_retriever,
# "keyword_search": keyword_retriever
# }
# )
# 执行研究
# report = assistant.research("大语言模型的最新发展趋势")
# print(report)
pass
# FastAPI 接口
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class ResearchRequest(BaseModel):
topic: str
max_iterations: int = 5
@app.post("/research")
async def create_research(request: ResearchRequest):
# assistant = get_assistant() # 从依赖注入获取
# report = assistant.research(request.topic)
# return {"report": report}
pass
第十三章：实战项目二 —— 多源知识问答系统
13.1 项目需求与架构设计
项目目标：构建一个能够从多个异构数据源（文档、数据库、API）获取信息并回答用户问题的系统。
核心功能：
- 智能查询路由：根据问题类型选择合适的数据源
- 多源并行检索：同时从多个数据源获取信息
- 结果融合与去重：合并来自不同源的结果
- 答案综合与验证：综合多源信息生成可靠答案
系统架构：
┌─────────────────────────────────────────────────┐
│ 用户查询接口 │
├─────────────────────────────────────────────────┤
│ 查询理解与路由 │
│ 意图识别 → 数据源选择 → 查询改写 │
├────────┬──────────┬──────────┬──────────────────┤
│ 文档库 │ 知识图谱 │ SQL数据库 │ 外部API │
│ 检索器 │ 查询器 │ 查询器 │ 调用器 │
├────────┴──────────┴──────────┴──────────────────┤
│ 结果融合引擎 │
│ 去重 → 排序 → 冲突检测 → 一致性处理 │
├─────────────────────────────────────────────────┤
│ 答案生成引擎 │
│ 上下文构建 → LLM 生成 → 答案验证 → 引用标注 │
└─────────────────────────────────────────────────┘
13.2 核心代码实现
from typing import List, Dict, Any, Optional
from abc import ABC, abstractmethod
from dataclasses import dataclass
import asyncio
# ========== 数据源抽象 ==========
@dataclass
class SourceResult:
content: str
source_type: str # document, knowledge_graph, database, api
source_name: str
confidence: float
metadata: Dict[str, Any] = None
def __post_init__(self):
if self.metadata is None:
self.metadata = {}
class DataSource(ABC):
"""数据源抽象基类"""
@property
@abstractmethod
def name(self) -> str:
pass
@property
@abstractmethod
def source_type(self) -> str:
pass
@property
@abstractmethod
def capabilities(self) -> List[str]:
"""该数据源能处理的查询类型"""
pass
@abstractmethod
async def query(self, query: str, params: Dict = None) -> List[SourceResult]:
pass
# ========== 具体数据源实现 ==========
class DocumentSource(DataSource):
"""文档数据源"""
def __init__(self, retriever):
self.retriever = retriever
@property
def name(self) -> str:
return "document_store"
@property
def source_type(self) -> str:
return "document"
@property
def capabilities(self) -> List[str]:
return ["factual", "conceptual", "procedural"]
async def query(self, query: str, params: Dict = None) -> List[SourceResult]:
results = self.retriever.search(query, top_k=5)
return [
SourceResult(
content=r[2],
source_type="document",
source_name=self.name,
confidence=r[1],
metadata={"index": r[0]}
)
for r in results
]
class KnowledgeGraphSource(DataSource):
"""知识图谱数据源"""
def __init__(self, graph_client):
self.client = graph_client
@property
def name(self) -> str:
return "knowledge_graph"
@property
def source_type(self) -> str:
return "knowledge_graph"
@property
def capabilities(self) -> List[str]:
return ["relational", "factual", "causal"]
async def query(self, query: str, params: Dict = None) -> List[SourceResult]:
# 从知识图谱查询
# 这里是示例，实际需要对接具体的图数据库
try:
entities = self._extract_entities(query)
triples = []
for entity in entities:
results = self.client.query(f"MATCH (n)-[r]->(m) WHERE n.name = '{entity}' RETURN n, r, m LIMIT 5")
triples.extend(results)
return [
SourceResult(
content=self._format_triple(t),
source_type="knowledge_graph",
source_name=self.name,
confidence=0.8
)
for t in triples
]
except Exception as e:
return []
def _extract_entities(self, query: str) -> List[str]:
# 简化的实体抽取
return [query.split("的")[0]] if "的" in query else [query[:10]]
def _format_triple(self, triple: Dict) -> str:
return f"{triple.get('subject', '')} --{triple.get('relation', '')}--> {triple.get('object', '')}"
class DatabaseSource(DataSource):
"""SQL 数据库数据源"""
def __init__(self, db_connection, llm):
self.db = db_connection
self.llm = llm
@property
def name(self) -> str:
return "sql_database"
@property
def source_type(self) -> str:
return "database"
@property
def capabilities(self) -> List[str]:
return ["statistical", "quantitative", "aggregation"]
async def query(self, query: str, params: Dict = None) -> List[SourceResult]:
try:
# 自然语言转 SQL
sql = self._nl_to_sql(query)
# 执行 SQL
results = self.db.execute(sql)
# 格式化结果
formatted = self._format_results(results)
return [
SourceResult(
content=formatted,
source_type="database",
source_name=self.name,
confidence=0.9,
metadata={"sql": sql}
)
]
except Exception as e:
return []
def _nl_to_sql(self, query: str) -> str:
prompt = f"""请将以下自然语言问题转换为 SQL 查询。
问题：{query}
只返回 SQL 语句："""
return self.llm.invoke(prompt).strip()
def _format_results(self, results) -> str:
return str(results)[:500]
class APISource(DataSource):
"""外部 API 数据源"""
def __init__(self, api_client, api_name: str):
self.client = api_client
self._name = api_name
@property
def name(self) -> str:
return self._name
@property
def source_type(self) -> str:
return "api"
@property
def capabilities(self) -> List[str]:
return ["realtime", "external"]
async def query(self, query: str, params: Dict = None) -> List[SourceResult]:
try:
response = await self.client.search(query)
return [
SourceResult(
content=item.get("content", ""),
source_type="api",
source_name=self.name,
confidence=item.get("score", 0.5),
metadata=item
)
for item in response.get("results", [])[:5]
]
except Exception as e:
return []
# ========== 查询路由器 ==========
class QueryRouterV2:
"""增强版查询路由器"""
def __init__(self, llm, sources: List[DataSource]):
self.llm = llm
self.sources = {s.name: s for s in sources}
self.capability_map = {}
# 构建能力映射
for source in sources:
for cap in source.capabilities:
if cap not in self.capability_map:
self.capability_map[cap] = []
self.capability_map[cap].append(source.name)
async def route_and_execute(self, query: str) -> Dict[str, List[SourceResult]]:
"""路由查询并执行"""
# 分析查询需要的能力
required_capabilities = self._analyze_capabilities(query)
# 选择数据源
selected_sources = self._select_sources(required_capabilities)
# 并行执行查询
tasks = []
source_names = []
for source_name in selected_sources:
source = self.sources[source_name]
tasks.append(source.query(query))
source_names.append(source_name)
results_list = await asyncio.gather(*tasks, return_exceptions=True)
# 组织结果
results = {}
for source_name, results_data in zip(source_names, results_list):
if isinstance(results_data, Exception):
results[source_name] = []
else:
results[source_name] = results_data
return results
def _analyze_capabilities(self, query: str) -> List[str]:
"""分析查询需要的能力"""
prompt = f"""请分析以下查询需要从哪些类型的数据源获取信息。
查询：{query}
能力选项：
- factual: 事实性信息
- conceptual: 概念解释
- procedural: 操作步骤
- relational: 实体关系
- causal: 因果关系
- statistical: 统计数据
- quantitative: 量化数据
- aggregation: 聚合分析
- realtime: 实时信息
- external: 外部信息
请返回 JSON 数组：["能力1", "能力2", ...]"""
response = self.llm.invoke(prompt)
try:
return json.loads(response)
except:
return ["factual"]
def _select_sources(self, capabilities: List[str]) -> List[str]:
"""根据能力选择数据源"""
selected = set()
for cap in capabilities:
if cap in self.capability_map:
selected.update(self.capability_map[cap])
return list(selected)
# ========== 结果融合器 ==========
class ResultFusionEngine:
"""结果融合引擎"""
def __init__(self, llm):
self.llm = llm
def fuse(self, query: str, results: Dict[str, List[SourceResult]]) -> List[SourceResult]:
"""融合多源结果"""
# 收集所有结果
all_results = []
for source_name, source_results in results.items():
all_results.extend(source_results)
if not all_results:
return []
# 去重
unique_results = self._deduplicate(all_results)
# 排序
sorted_results = self._rank(query, unique_results)
# 冲突检测
conflicts = self._detect_conflicts(sorted_results)
return sorted_results
def _deduplicate(self, results: List[SourceResult]) -> List[SourceResult]:
"""去重"""
seen_contents = set()
unique = []
for result in results:
# 使用内容的前 100 字符作为去重键
key = result.content[:100]
if key not in seen_contents:
seen_contents.add(key)
unique.append(result)
return unique
def _rank(self, query: str, results: List[SourceResult]) -> List[SourceResult]:
"""排序"""
# 按置信度排序
return sorted(results, key=lambda r: r.confidence, reverse=True)
def _detect_conflicts(self, results: List[SourceResult]) -> List[Dict]:
"""检测结果间的冲突"""
# 简化实现：实际应该用 LLM 检测语义冲突
return []
# ========== 答案生成器 ==========
class AnswerGenerator:
"""答案生成器"""
def __init__(self, llm):
self.llm = llm
def generate(
self,
query: str,
results: List[SourceResult],
include_citations: bool = True
) -> Dict[str, Any]:
"""生成答案"""
# 构建上下文
context_parts = []
citations = []
for i, result in enumerate(results[:10]):
context_parts.append(f"[来源{i+1}] {result.content}")
citations.append({
"id": i + 1,
"source": result.source_name,
"type": result.source_type,
"confidence": result.confidence
})
context = "\n\n".join(context_parts)
# 生成答案
prompt = f"""请基于以下多个来源的信息，回答用户的问题。
要求：
1. 综合多个来源的信息，给出全面、准确的回答
2. 如果不同来源的信息存在冲突，请指出并说明
3. 在回答中使用 [来源X] 标注信息来源
4. 如果信息不足，请如实说明
参考资料：
{context}
问题：{query}
回答："""
answer = self.llm.invoke(prompt)
# 验证答案
verification = self._verify_answer(query, answer, context)
return {
"answer": answer,
"citations": citations if include_citations else [],
"sources_used": len(results),
"confidence": verification.get("confidence", 0.7),
"verification": verification
}
def _verify_answer(self, query: str, answer: str, context: str) -> Dict:
"""验证答案质量"""
prompt = f"""请验证以下答案的质量。
问题：{query}
答案：{answer}
参考信息：{context[:500]}
请返回 JSON：
{{"confidence": 0.0-1.0, "is_faithful": true/false, "issues": ["问题1", ...]}}"""
response = self.llm.invoke(prompt)
try:
return json.loads(response)
except:
return {"confidence": 0.7, "is_faithful": True, "issues": []}
# ========== 主系统 ==========
class MultiSourceQA:
"""多源知识问答系统"""
def __init__(self, llm, sources: List[DataSource]):
self.llm = llm
self.router = QueryRouterV2(llm, sources)
self.fusion = ResultFusionEngine(llm)
self.generator = AnswerGenerator(llm)
async def answer(self, query: str) -> Dict[str, Any]:
"""回答问题"""
# 1. 路由并执行多源检索
raw_results = await self.router.route_and_execute(query)
# 2. 融合结果
fused_results = self.fusion.fuse(query, raw_results)
# 3. 生成答案
answer_result = self.generator.generate(query, fused_results)
# 4. 添加元数据
answer_result["query"] = query
answer_result["sources_consulted"] = list(raw_results.keys())
answer_result["total_results_retrieved"] = sum(
len(v) for v in raw_results.values()
)
return answer_result
13.3 部署与测试
# FastAPI 部署
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI(title="多源知识问答系统")
class QARequest(BaseModel):
question: str
include_citations: bool = True
max_sources: int = 5
class QAResponse(BaseModel):
answer: str
citations: List[Dict]
sources_consulted: List[str]
confidence: float
# 全局系统实例
# qa_system = None
@app.on_event("startup")
async def startup():
"""初始化系统"""
# global qa_system
# llm = init_llm()
# sources = [
# DocumentSource(init_retriever()),
# KnowledgeGraphSource(init_graph_client()),
# DatabaseSource(init_db(), llm),
# ]
# qa_system = MultiSourceQA(llm, sources)
pass
@app.post("/ask", response_model=QAResponse)
async def ask_question(request: QARequest):
"""问答接口"""
# if qa_system is None:
# raise HTTPException(status_code=503, detail="系统未初始化")
#
# result = await qa_system.answer(request.question)
# return QAResponse(**result)
pass
@app.get("/health")
async def health_check():
return {"status": "healthy"}
# 测试用例
async def test_multi_source_qa():
"""测试多源问答系统"""
test_questions = [
"2023年中国GDP是多少？同比增长了多少？", # 需要数据库 + 文档
"量子计算和经典计算的核心区别是什么？", # 需要文档 + 知识图谱
"如何配置 Kubernetes 集群的自动扩缩容？", # 需要文档
"最近一周的热门 AI 论文有哪些？", # 需要外部 API
]
# for question in test_questions:
# result = await qa_system.answer(question)
# print(f"\n问题: {question}")
# print(f"答案: {result['answer'][:200]}...")
# print(f"来源: {result['sources_consulted']}")
# print(f"置信度: {result['confidence']}")
pass
第十四章：常见问题与最佳实践
14.1 常见问题排查
Q1: 检索结果不相关怎么办？
诊断步骤：
1. 检查分词质量 → 中文分词是否准确
2. 检查 Embedding 模型 → 是否适合你的领域
3. 检查索引质量 → 文档分块是否合理
4. 尝试查询改写 → HyDE、扩展查询
解决方案：
- 使用领域微调的 Embedding 模型
- 优化文档分块策略（按语义分块）
- 增加重排序步骤
- 实现查询改写（HyDE 效果通常最好）
Q2: Agent 陷入无限循环怎么办？
常见原因：
1. 没有设置最大迭代次数
2. 评估函数总是返回"不充分"
3. 检索结果质量波动大
解决方案：
- 设置硬性最大迭代次数（建议 3-5 次）
- 实现渐进式置信度阈值
- 添加循环检测（相同查询不重复检索）
- 设置超时机制
Q3: 响应延迟太高怎么办？
优化策略：
1. 并行检索：多数据源同时查询
2. 缓存机制：缓存常见查询的结果
3. 流式输出：先返回部分结果
4. 异步处理：非关键步骤异步执行
5. 模型选择：简单任务用小模型
6. 减少 LLM 调用：规则判断替代部分 LLM 决策
Q4: 多源结果冲突怎么办？
处理策略：
1. 优先级机制：官方数据 > 第三方数据
2. 时效性机制：新数据 > 旧数据
3. 多数投票：多个来源一致的结果更可信
4. 透明展示：向用户展示不同来源的说法
5. LLM 判断：用 LLM 分析冲突并给出判断
14.2 性能优化建议
# 1. 查询缓存
from functools import lru_cache
from hashlib import md5
class QueryCache:
"""查询结果缓存"""
def __init__(self, max_size: int = 1000, ttl_seconds: int = 3600):
self.cache = {}
self.max_size = max_size
self.ttl = ttl_seconds
def get(self, query: str) -> Optional[Dict]:
key = md5(query.encode()).hexdigest()
if key in self.cache:
entry = self.cache[key]
if time.time() - entry["timestamp"] < self.ttl:
return entry["data"]
else:
del self.cache[key]
return None
def set(self, query: str, data: Dict):
if len(self.cache) >= self.max_size:
# 删除最旧的条目
oldest_key = min(self.cache, key=lambda k: self.cache[k]["timestamp"])
del self.cache[oldest_key]
key = md5(query.encode()).hexdigest()
self.cache[key] = {"data": data, "timestamp": time.time()}
# 2. 批量处理
class BatchProcessor:
"""批量处理优化"""
def __init__(self, batch_size: int = 10):
self.batch_size = batch_size
self.queue = []
async def add_and_process(self, item: Any) -> Any:
self.queue.append(item)
if len(self.queue) >= self.batch_size:
return await self._process_batch()
# 返回 Future
return asyncio.Future()
async def _process_batch(self) -> List[Any]:
batch = self.queue[:self.batch_size]
self.queue = self.queue[self.batch_size:]
# 批量调用 LLM
results = await asyncio.gather(*[self._process_item(item) for item in batch])
return results
# 3. 连接池管理
class ConnectionPool:
"""数据库连接池"""
def __init__(self, create_connection, max_size: int = 10):
self.create_connection = create_connection
self.pool = asyncio.Queue(maxsize=max_size)
self.max_size = max_size
async def get_connection(self):
try:
return self.pool.get_nowait()
except asyncio.QueueEmpty:
return self.create_connection()
async def release_connection(self, conn):
try:
self.pool.put_nowait(conn)
except asyncio.QueueFull:
pass # 连接池已满，丢弃连接
14.3 安全与合规
class SecurityFilter:
"""安全过滤器"""
def __init__(self):
self.sensitive_patterns = [
r'\b\d{18}\b', # 身份证号
r'\b\d{11}\b', # 手机号
r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', # 邮箱
]
def filter_query(self, query: str) -> str:
"""过滤查询中的敏感信息"""
import re
filtered = query
for pattern in self.sensitive_patterns:
filtered = re.sub(pattern, '[已过滤]', filtered)
return filtered
def filter_response(self, response: str) -> str:
"""过滤响应中的敏感信息"""
import re
filtered = response
for pattern in self.sensitive_patterns:
filtered = re.sub(pattern, '[已过滤]', filtered)
return filtered
def validate_query(self, query: str) -> bool:
"""验证查询是否合规"""
# 检查是否包含恶意内容
forbidden_keywords = ["注入", "攻击", "破解"]
return not any(kw in query for kw in forbidden_keywords)
14.4 未来发展方向
1. 更智能的 Agent 架构
未来的 Agentic RAG 将更加智能化，Agent 能够：
- 自动学习最优检索策略
- 根据用户反馈持续改进
- 跨会话记忆和知识积累
2. 多模态融合
随着多模态模型的发展，Agentic RAG 将支持：
- 图片理解和检索
- 视频内容分析
- 音频信息处理
3. 端侧部署
随着模型压缩技术的发展，Agentic RAG 将能够在边缘设备上运行：
- 手机端 RAG 助手
- IoT 设备知识问答
- 离线环境下的智能检索
4. 协作式 RAG
多个 RAG 系统之间的协作：
- 知识共享与同步
- 分布式检索与聚合
- 跨组织知识协作
总结
Agentic RAG 代表了 RAG 技术的最前沿，它将 Agent 的自主决策能力引入检索增强生成，使系统能够：
- 自主规划：根据查询复杂度自动选择策略
- 动态路由：智能选择最合适的数据源
- 迭代优化：不断改进检索结果直到满意
- 自我反思：评估并提升生成质量
本教程从基础概念到高级架构，从核心算法到实战项目，系统地覆盖了 Agentic RAG 的各个方面。希望读者能够：
- 理解 Agentic RAG 的核心理念和设计原则
- 掌握混合检索、重排序、查询路由等关键技术
- 学会使用 LangGraph 构建复杂的 Agentic RAG 流程
- 能够独立设计和实现生产级的 Agentic RAG 系统
技术在不断演进，保持学习和实践是掌握 Agentic RAG 的关键。祝你在 Agentic RAG 的探索之路上取得成功！
参考资源
- LangChain 官方文档
- LangGraph 官方文档
- LlamaIndex 文档
- 相关学术论文：Self-RAG、CRAG、HyDE 等
- 开源项目：RAGFlow、Dify、FastGPT 等
本教程内容仅供学习参考，代码示例为教学目的简化版本，生产环境使用请根据实际需求进行调整和优化。
