Agentic RAG智能检索系统完全教程
本教程全面讲解Agentic RAG智能检索系统的核心架构与开发技术，涵盖从传统RAG到Agentic RAG的演进、自适应检索、自反思RAG、GraphRAG、LangGraph实现等核心内容，帮助开发者构建生产级智能检索系统。
目录
- 概述：从传统RAG到Agentic RAG
- 传统RAG基础回顾
- Agentic RAG核心架构
- 自适应检索路由（Adaptive RAG）
- 自反思RAG（Self-RAG与CRAG）
- 多步推理检索（IRCoT与Step-back）
- 查询规划与分解
- 多工具协同检索
- GraphRAG知识图谱增强
- 检索结果验证与幻觉检测
- Multi-Agent RAG协作
- LangGraph实现Agentic RAG
- 生产部署与优化
- 总结
概述：从传统RAG到Agentic RAG
传统RAG的局限
传统的RAG（Retrieval-Augmented Generation）系统采用"检索-生成"的简单流水线：用户提问→检索相关文档→拼接上下文→LLM生成回答。这种架构虽然有效，但存在明显局限：
传统RAG的核心问题：
- 检索策略固定：无论问题复杂度如何，都使用相同的检索策略
- 单次检索不足：复杂问题需要多步推理和多次检索
- 缺乏自我纠错：检索到无关文档时无法自我修正
- 查询理解薄弱：用户的原始查询可能不适合直接用于检索
- 无推理链路：无法对检索结果进行推理和验证
Agentic RAG的革命性突破
Agentic RAG将AI Agent的自主决策能力引入RAG系统，使其具备：
- 自主判断：判断是否需要检索、何时检索、检索什么
- 自适应策略：根据问题复杂度动态选择检索策略
- 自反思纠错：评估检索质量，必要时重新检索
- 多步推理：将复杂问题分解为子问题，逐步检索和推理
- 工具协同：同时利用多种检索工具和数据源
传统RAG vs Agentic RAG对比：
| 维度 | 传统RAG | Agentic RAG |
|---|---|---|
| 检索策略 | 固定单一 | 动态自适应 |
| 检索次数 | 单次 | 多次（按需） |
| 查询处理 | 直接使用 | 规划、分解、改写 |
| 质量控制 | 无 | 自反思、验证、纠错 |
| 推理能力 | 简单拼接 | 多步推理链 |
| 工具使用 | 单一向量检索 | 多工具协同 |
传统RAG基础回顾
在深入Agentic RAG之前，先回顾传统RAG的核心组件：
基础RAG实现
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class Document:
content: str
metadata: dict
score: float = 0.0
class SimpleRAG:
"""基础RAG系统实现"""
def __init__(self, llm_client, embedding_model, vector_store):
self.llm = llm_client
self.embedder = embedding_model
self.vector_store = vector_store
def add_documents(self, documents: List[Document]):
"""添加文档到知识库"""
texts = [doc.content for doc in documents]
embeddings = self.embedder.encode(texts)
self.vector_store.add(embeddings, documents)
def retrieve(self, query: str, top_k: int = 5) -> List[Document]:
"""检索相关文档"""
query_embedding = self.embedder.encode([query])[0]
results = self.vector_store.search(query_embedding, top_k=top_k)
return results
def generate(self, query: str, context_docs: List[Document]) -> str:
"""基于检索结果生成回答"""
context = "\n\n".join([doc.content for doc in context_docs])
prompt = f"""基于以下参考资料回答用户问题。
参考资料：
{context}
用户问题：{query}
请基于参考资料提供准确、详细的回答。如果参考资料中没有相关信息，请说明。"""
return self.llm.generate(prompt)
def query(self, user_query: str, top_k: int = 5) -> str:
"""完整的RAG查询流程"""
# 1. 检索
docs = self.retrieve(user_query, top_k)
# 2. 生成
return self.generate(user_query, docs)
向量检索优化
import numpy as np
from typing import List, Tuple
class HybridRetriever:
"""混合检索器 - 结合语义检索和关键词检索"""
def __init__(self, embedding_model, bm25_index, vector_store):
self.embedder = embedding_model
self.bm25 = bm25_index
self.vector_store = vector_store
def retrieve(self, query: str, top_k: int = 5,
semantic_weight: float = 0.7,
keyword_weight: float = 0.3) -> List[Tuple[Document, float]]:
"""混合检索"""
# 语义检索
query_emb = self.embedder.encode([query])[0]
semantic_results = self.vector_store.search(query_emb, top_k=top_k * 2)
# 关键词检索（BM25）
keyword_results = self.bm25.search(query, top_k=top_k * 2)
# 融合排序
doc_scores = {}
for doc, score in semantic_results:
doc_scores[doc.content] = {
"doc": doc,
"semantic_score": score,
"keyword_score": 0.0,
}
for doc, score in keyword_results:
if doc.content in doc_scores:
doc_scores[doc.content]["keyword_score"] = score
else:
doc_scores[doc.content] = {
"doc": doc,
"semantic_score": 0.0,
"keyword_score": score,
}
# 加权融合
results = []
for content, scores in doc_scores.items():
final_score = (
scores["semantic_score"] * semantic_weight +
scores["keyword_score"] * keyword_weight
)
results.append((scores["doc"], final_score))
results.sort(key=lambda x: x[1], reverse=True)
return results[:top_k]
class Reranker:
"""重排序器 - 对检索结果进行精排"""
def __init__(self, rerank_model):
self.model = rerank_model
def rerank(self, query: str, documents: List[Document],
top_k: int = 3) -> List[Document]:
"""对文档进行重排序"""
pairs = [(query, doc.content) for doc in documents]
scores = self.model.predict(pairs)
scored_docs = list(zip(documents, scores))
scored_docs.sort(key=lambda x: x[1], reverse=True)
return [doc for doc, _ in scored_docs[:top_k]]
Agentic RAG核心架构
Agentic RAG的核心思想是将Agent的决策能力注入RAG流程的每个环节。
架构概览
用户查询
│
▼
┌─────────────────┐
│ 查询分析器 │ ← 判断查询类型、复杂度
└────────┬────────┘
│
▼
┌─────────────────┐
│ 策略路由器 │ ← 选择检索策略
└────────┬────────┘
│
┌────┴────┐
▼ ▼
┌────────┐ ┌────────┐
│直接回答│ │检索增强│ ← 简单问题直接回答，复杂问题检索
└────────┘ └───┬────┘
│
▼
┌─────────────┐
│ 多步检索循环 │ ← 检索→评估→(重新检索)
└──────┬──────┘
│
▼
┌─────────────┐
│ 结果验证 │ ← 检查检索质量、幻觉检测
└──────┬──────┘
│
▼
┌─────────────┐
│ 答案生成 │ ← 基于验证后的结果生成答案
└─────────────┘
核心Agent类
from typing import List, Dict, Optional, Callable
from enum import Enum
from dataclasses import dataclass, field
import json
class QueryComplexity(Enum):
SIMPLE = "simple" # 简单事实查询
MODERATE = "moderate" # 需要一定推理
COMPLEX = "complex" # 需要多步推理
MULTI_HOP = "multi_hop" # 需要跨文档推理
class RetrievalStrategy(Enum):
DIRECT_ANSWER = "direct_answer" # 直接回答（无需检索）
SINGLE_RETRIEVE = "single_retrieve" # 单次检索
ITERATIVE = "iterative" # 迭代检索
MULTI_QUERY = "multi_query" # 多查询检索
STEP_BACK = "step_back" # 回退检索
@dataclass
class RetrievalResult:
documents: List[Document]
query_used: str
strategy: RetrievalStrategy
quality_score: float
needs_refinement: bool = False
class AgenticRAG:
"""Agentic RAG核心系统"""
def __init__(self, llm_client, retriever, reranker=None):
self.llm = llm_client
self.retriever = retriever
self.reranker = reranker
self.max_iterations = 3
self.quality_threshold = 0.7
def query(self, user_query: str) -> Dict:
"""Agentic RAG查询入口"""
# 1. 查询分析
analysis = self._analyze_query(user_query)
# 2. 策略选择
strategy = self._select_strategy(analysis)
# 3. 执行检索策略
if strategy == RetrievalStrategy.DIRECT_ANSWER:
answer = self._direct_answer(user_query)
return {"answer": answer, "strategy": "direct", "sources": []}
# 4. 检索-评估-优化循环
result = self._execute_retrieval_loop(user_query, strategy, analysis)
# 5. 生成最终答案
answer = self._generate_answer(user_query, result)
return {
"answer": answer,
"strategy": strategy.value,
"sources": [doc.metadata for doc in result.documents],
"iterations": getattr(result, 'iterations', 1),
}
def _analyze_query(self, query: str) -> Dict:
"""分析查询特征"""
analysis_prompt = f"""分析以下查询的特征，返回JSON格式结果：
查询：{query}
请分析：
1. 复杂度（simple/moderate/complex/multi_hop）
2. 是否需要最新信息（true/false）
3. 是否需要多文档综合（true/false）
4. 领域（general/technical/scientific/legal/medical）
5. 预期检索次数（1/2/3）
返回JSON格式。"""
response = self.llm.generate(analysis_prompt)
try:
return json.loads(response)
except:
return {
"complexity": "moderate",
"needs_latest": False,
"needs_multi_doc": True,
"domain": "general",
"expected_retrievals": 2,
}
def _select_strategy(self, analysis: Dict) -> RetrievalStrategy:
"""根据分析结果选择检索策略"""
complexity = analysis.get("complexity", "moderate")
strategy_map = {
"simple": RetrievalStrategy.DIRECT_ANSWER,
"moderate": RetrievalStrategy.SINGLE_RETRIEVE,
"complex": RetrievalStrategy.ITERATIVE,
"multi_hop": RetrievalStrategy.MULTI_QUERY,
}
return strategy_map.get(complexity, RetrievalStrategy.SINGLE_RETRIEVE)
def _execute_retrieval_loop(self, query: str, strategy: RetrievalStrategy,
analysis: Dict) -> RetrievalResult:
"""执行检索-评估-优化循环"""
current_query = query
all_documents = []
iteration = 0
while iteration < self.max_iterations:
iteration += 1
# 检索
if strategy == RetrievalStrategy.MULTI_QUERY:
queries = self._generate_sub_queries(query, analysis)
docs = []
for q in queries:
docs.extend(self.retriever.retrieve(q, top_k=3))
else:
docs = self.retriever.retrieve(current_query, top_k=5)
# 重排序
if self.reranker:
docs = self.reranker.rerank(query, docs, top_k=3)
# 评估检索质量
quality = self._evaluate_retrieval_quality(query, docs)
if quality >= self.quality_threshold:
return RetrievalResult(
documents=docs,
query_used=current_query,
strategy=strategy,
quality_score=quality,
)
# 质量不足，优化查询
current_query = self._refine_query(query, docs, quality)
# 达到最大迭代次数
return RetrievalResult(
documents=all_documents or docs,
query_used=current_query,
strategy=strategy,
quality_score=quality,
needs_refinement=True,
)
def _generate_sub_queries(self, original_query: str, analysis: Dict) -> List[str]:
"""生成子查询（用于多跳推理）"""
prompt = f"""将以下复杂查询分解为2-4个子查询，每个子查询关注问题的不同方面。
原始查询：{original_query}
分析：{json.dumps(analysis, ensure_ascii=False)}
返回JSON数组格式的子查询列表。"""
response = self.llm.generate(prompt)
try:
return json.loads(response)
except:
return [original_query]
def _evaluate_retrieval_quality(self, query: str, documents: List[Document]) -> float:
"""评估检索结果质量"""
if not documents:
return 0.0
context = "\n".join([doc.content[:200] for doc in documents[:3]])
eval_prompt = f"""评估以下检索结果对回答用户问题的相关性。
用户问题：{query}
检索结果：
{context}
请给出0-1之间的相关性分数（0=完全无关，1=高度相关）。只返回数字。"""
response = self.llm.generate(eval_prompt)
try:
return float(response.strip())
except:
return 0.5
def _refine_query(self, original_query: str, documents: List[Document],
quality: float) -> str:
"""优化查询以提高检索质量"""
context = "\n".join([doc.content[:100] for doc in documents[:2]])
prompt = f"""原始查询检索效果不佳（质量分数: {quality:.2f}）。
请优化查询以获得更相关的结果。
原始查询：{original_query]
当前结果片段：
{context}
请返回一个优化后的查询。"""
return self.llm.generate(prompt).strip()
def _direct_answer(self, query: str) -> str:
"""直接回答（无需检索）"""
return self.llm.generate(f"请直接回答以下问题：{query}")
def _generate_answer(self, query: str, result: RetrievalResult) -> str:
"""基于检索结果生成答案"""
context = "\n\n".join([doc.content for doc in result.documents])
prompt = f"""基于以下参考资料回答用户问题。
参考资料：
{context}
用户问题：{query}
请提供准确、详细的回答。如果信息不足，请说明。"""
return self.llm.generate(prompt)
自适应检索路由（Adaptive RAG）
Adaptive RAG的核心思想是根据查询特征动态选择最优的检索策略。
查询分类器
class QueryRouter:
"""查询路由器 - 自适应选择检索策略"""
def __init__(self, llm_client):
self.llm = llm_client
self.strategies = {
"no_retrieval": {
"description": "无需检索，直接用LLM知识回答",
"examples": ["你好", "1+1等于几", "什么是AI"],
},
"single_retrieval": {
"description": "单次检索即可回答",
"examples": ["Python的list和tuple有什么区别", "什么是RAG"],
},
"multi_retrieval": {
"description": "需要多次检索和推理",
"examples": ["比较React和Vue的优缺点", "分析AI对教育的影响"],
},
"iterative_refinement": {
"description": "需要迭代优化检索结果",
"examples": ["最新的GPT-4技术细节", "2024年AI行业趋势"],
},
}
def route(self, query: str) -> str:
"""路由查询到合适的策略"""
strategy_desc = "\n".join([
f"- {name}: {info['description']}"
for name, info in self.strategies.items()
])
prompt = f"""判断以下查询应该使用哪种检索策略。
查询：{query}
可选策略：
{strategy_desc}
只返回策略名称（如 single_retrieval）。"""
response = self.llm.generate(prompt).strip().lower()
if response in self.strategies:
return response
return "single_retrieval" # 默认策略
def get_retrieval_params(self, strategy: str, query: str) -> dict:
"""获取策略对应的检索参数"""
params_map = {
"no_retrieval": {"top_k": 0, "rerank": False},
"single_retrieval": {"top_k": 5, "rerank": True},
"multi_retrieval": {"top_k": 3, "rerank": True, "num_queries": 3},
"iterative_refinement": {"top_k": 5, "rerank": True, "max_iterations": 3},
}
return params_map.get(strategy, {"top_k": 5, "rerank": True})
class AdaptiveRAG:
"""自适应RAG系统"""
def __init__(self, llm_client, retriever, reranker=None):
self.llm = llm_client
self.retriever = retriever
self.reranker = reranker
self.router = QueryRouter(llm_client)
def query(self, user_query: str) -> Dict:
"""自适应查询"""
# 1. 路由决策
strategy = self.router.route(user_query)
params = self.router.get_retrieval_params(strategy, user_query)
# 2. 根据策略执行
if strategy == "no_retrieval":
answer = self.llm.generate(f"回答：{user_query}")
return {"answer": answer, "strategy": strategy, "sources": []}
if strategy == "single_retrieval":
docs = self.retriever.retrieve(user_query, top_k=params["top_k"])
if params.get("rerank") and self.reranker:
docs = self.reranker.rerank(user_query, docs)
answer = self._generate(user_query, docs)
elif strategy == "multi_retrieval":
queries = self._generate_multi_queries(user_query, params.get("num_queries", 3))
all_docs = []
for q in queries:
all_docs.extend(self.retriever.retrieve(q, top_k=params["top_k"]))
if params.get("rerank") and self.reranker:
all_docs = self.reranker.rerank(user_query, all_docs, top_k=5)
answer = self._generate(user_query, all_docs)
docs = all_docs
elif strategy == "iterative_refinement":
docs, answer = self._iterative_retrieval(user_query, params)
else:
docs = self.retriever.retrieve(user_query, top_k=5)
answer = self._generate(user_query, docs)
return {
"answer": answer,
"strategy": strategy,
"sources": [d.metadata for d in (docs if 'docs' in dir() else [])],
}
def _generate_multi_queries(self, query: str, num: int) -> List[str]:
"""生成多个查询变体"""
prompt = f"""为以下查询生成{num}个不同角度的查询变体，以提高检索覆盖率。
原始查询：{query}
返回JSON数组格式。"""
response = self.llm.generate(prompt)
try:
queries = json.loads(response)
return [query] + queries[:num-1]
except:
return [query]
def _iterative_retrieval(self, query: str, params: dict) -> tuple:
"""迭代检索"""
current_query = query
best_docs = []
best_score = 0.0
for i in range(params.get("max_iterations", 3)):
docs = self.retriever.retrieve(current_query, top_k=params["top_k"])
if self.reranker:
docs = self.reranker.rerank(query, docs)
score = self._evaluate_docs(query, docs)
if score > best_score:
best_score = score
best_docs = docs
if score >= 0.8:
break
current_query = self._refine_query(query, docs)
answer = self._generate(query, best_docs)
return best_docs, answer
def _evaluate_docs(self, query: str, docs: List[Document]) -> float:
"""评估文档质量"""
if not docs:
return 0.0
context = "\n".join([d.content[:150] for d in docs[:3]])
prompt = f"查询: {query}\n文档: {context}\n相关性分数(0-1):"
try:
return float(self.llm.generate(prompt).strip())
except:
return 0.5
def _refine_query(self, query: str, docs: List[Document]) -> str:
"""优化查询"""
context = "\n".join([d.content[:100] for d in docs[:2]])
prompt = f"原始查询: {query}\n当前结果不够好。片段: {context}\n优化后的查询:"
return self.llm.generate(prompt).strip()
def _generate(self, query: str, docs: List[Document]) -> str:
"""生成答案"""
context = "\n\n".join([d.content for d in docs])
return self.llm.generate(
f"参考资料:\n{context}\n\n问题: {query}\n\n请基于资料回答:"
)
自反思RAG（Self-RAG与CRAG）
Self-RAG实现
Self-RAG通过特殊的"反思标记"让LLM自我评估检索的必要性和生成质量：
from typing import List, Tuple
from dataclasses import dataclass
@dataclass
class SelfRAGDecision:
"""Self-RAG决策结果"""
need_retrieval: bool # 是否需要检索
relevance_score: float # 文档相关性
support_score: float # 生成内容是否被文档支持
utility_score: float # 回答的实用性
class SelfRAG:
"""Self-RAG: 自反思检索增强生成"""
def __init__(self, llm_client, retriever):
self.llm = llm_client
self.retriever = retriever
def query(self, user_query: str) -> Dict:
"""Self-RAG查询流程"""
# 1. 判断是否需要检索
need_retrieval = self._check_retrieval_need(user_query)
if not need_retrieval:
answer = self._generate_without_retrieval(user_query)
return {"answer": answer, "retrieved": False, "reflections": []}
# 2. 检索文档
documents = self.retriever.retrieve(user_query, top_k=5)
# 3. 对每个文档评估相关性
relevant_docs = []
for doc in documents:
relevance = self._assess_relevance(user_query, doc)
if relevance > 0.5:
relevant_docs.append((doc, relevance))
if not relevant_docs:
# 没有相关文档，尝试优化查询
refined_query = self._refine_query(user_query)
documents = self.retriever.retrieve(refined_query, top_k=5)
for doc in documents:
relevance = self._assess_relevance(user_query, doc)
if relevance > 0.5:
relevant_docs.append((doc, relevance))
# 4. 基于相关文档生成回答
context = "\n\n".join([doc.content for doc, _ in relevant_docs])
answer = self._generate_with_context(user_query, context)
# 5. 评估生成质量
support_score = self._assess_support(answer, context)
utility_score = self._assess_utility(user_query, answer)
return {
"answer": answer,
"retrieved": True,
"num_relevant_docs": len(relevant_docs),
"support_score": support_score,
"utility_score": utility_score,
}
def _check_retrieval_need(self, query: str) -> bool:
"""判断是否需要检索"""
prompt = f"""判断以下查询是否需要检索外部知识来回答。
查询：{query}
判断标准：
- 事实性问题、需要最新信息、需要专业知识 → 需要检索
- 通用常识、简单计算、创意写作 → 不需要检索
回答 yes 或 no。"""
response = self.llm.generate(prompt).strip().lower()
return response.startswith("y")
def _assess_relevance(self, query: str, document: Document) -> float:
"""评估文档与查询的相关性"""
prompt = f"""评估以下文档对回答用户问题的相关性。
用户问题：{query}
文档内容：{document.content[:300]}
评分标准：
- 0.0-0.3: 完全无关
- 0.3-0.5: 部分相关
- 0.5-0.7: 相关
- 0.7-1.0: 高度相关
只返回0-1之间的数字。"""
try:
return float(self.llm.generate(prompt).strip())
except:
return 0.5
def _assess_support(self, answer: str, context: str) -> float:
"""评估回答是否被上下文支持"""
prompt = f"""评估以下回答是否完全基于提供的参考资料。
参考资料：{context[:500]}
回答：{answer[:500]}
评分：
- 0.0-0.3: 回答包含大量未被支持的信息
- 0.3-0.7: 部分信息有支持，部分没有
- 0.7-1.0: 回答完全被参考资料支持
只返回数字。"""
try:
return float(self.llm.generate(prompt).strip())
except:
return 0.5
def _assess_utility(self, query: str, answer: str) -> float:
"""评估回答的实用性"""
prompt = f"""评估以下回答对用户问题的实用性。
问题：{query}
回答：{answer[:500]}
评分(0-1)，只返回数字。"""
try:
return float(self.llm.generate(prompt).strip())
except:
return 0.5
def _generate_without_retrieval(self, query: str) -> str:
"""不使用检索直接生成"""
return self.llm.generate(f"请回答：{query}")
def _generate_with_context(self, query: str, context: str) -> str:
"""基于上下文生成"""
return self.llm.generate(
f"参考资料:\n{context}\n\n问题: {query}\n\n基于资料回答:"
)
def _refine_query(self, query: str) -> str:
"""优化查询"""
return self.llm.generate(
f"以下查询检索效果不佳，请优化：{query}\n优化后的查询:"
).strip()
### Corrective RAG (CRAG)
CRAG在Self-RAG基础上增加了"纠正"机制，当检索结果不理想时自动触发纠正流程：
class CorrectiveRAG:
"""Corrective RAG - 纠正式检索增强生成"""
def __init__(self, llm_client, retriever, web_search_fn=None):
self.llm = llm_client
self.retriever = retriever
self.web_search = web_search_fn
def query(self, user_query: str) -> Dict:
"""CRAG查询流程"""
# 1. 检索文档
documents = self.retriever.retrieve(user_query, top_k=5)
# 2. 评估检索质量
quality_assessment = self._assess_retrieval_quality(user_query, documents)
# 3. 根据质量决定纠正策略
if quality_assessment["status"] == "correct":
# 检索质量好，直接使用
refined_docs = self._extract_relevant_info(user_query, documents)
answer = self._generate(user_query, refined_docs)
elif quality_assessment["status"] == "ambiguous":
# 检索质量一般，进行知识精炼
refined_docs = self._knowledge_refinement(user_query, documents)
answer = self._generate(user_query, refined_docs)
elif quality_assessment["status"] == "incorrect":
# 检索质量差，触发纠正
if self.web_search:
# 使用网络搜索补充
web_results = self.web_search(user_query)
answer = self._generate_with_web(user_query, documents, web_results)
else:
# 重新生成查询并检索
new_query = self._reformulate_query(user_query, documents)
new_docs = self.retriever.retrieve(new_query, top_k=5)
answer = self._generate(user_query, new_docs)
else:
answer = self._generate(user_query, documents)
return {
"answer": answer,
"retrieval_quality": quality_assessment,
}
def _assess_retrieval_quality(self, query: str, documents: List[Document]) -> Dict:
"""评估检索质量（correct/ambiguous/incorrect）"""
context = "\n".join([doc.content[:200] for doc in documents[:3]])
prompt = f"""评估检索结果的质量。
查询：{query}
检索结果：
{context}
判断检索结果属于以下哪种情况：
1. correct: 检索结果高度相关，可以直接使用
2. ambiguous: 检索结果部分相关，需要精炼
3. incorrect: 检索结果不相关或错误
返回JSON: {{"status": "correct/ambiguous/incorrect", "confidence": 0.0-1.0, "reason": "..."}}"""
response = self.llm.generate(prompt)
try:
return json.loads(response)
except:
return {"status": "ambiguous", "confidence": 0.5, "reason": "无法判断"}
def _extract_relevant_info(self, query: str, documents: List[Document]) -> str:
"""从文档中提取相关信息"""
context = "\n\n".join([doc.content for doc in documents])
prompt = f"""从以下文档中提取与查询相关的关键信息。
查询：{query}
文档：{context}
提取关键信息（保持原文，不要改写）："""
return self.llm.generate(prompt)
def _knowledge_refinement(self, query: str, documents: List[Document]) -> str:
"""知识精炼"""
context = "\n\n".join([doc.content for doc in documents])
prompt = f"""以下检索结果部分相关但不完全匹配查询。
请精炼这些信息，保留相关内容，去除无关内容。
查询：{query}
原始信息：{context}
精炼后的信息："""
return self.llm.generate(prompt)
def _reformulate_query(self, query: str, bad_docs: List[Document]) -> str:
"""重新生成查询"""
context = "\n".join([doc.content[:100] for doc in bad_docs[:2]])
prompt = f"""之前的检索结果不理想，请重新生成一个更好的查询。
原始查询：{query}
不理想的结果：{context}
新查询："""
return self.llm.generate(prompt).strip()
def _generate(self, query: str, context: str) -> str:
return self.llm.generate(f"参考资料:\n{context}\n\n问题: {query}\n\n回答:")
def _generate_with_web(self, query: str, local_docs: List[Document],
web_results: str) -> str:
local_context = "\n".join([doc.content[:200] for doc in local_docs[:2]])
return self.llm.generate(
f"本地资料:\n{local_context}\n\n网络搜索结果:\n{web_results}\n\n"
f"问题: {query}\n\n综合回答:"
)
多步推理检索（IRCoT与Step-back）
IRCoT（Interleaving Retrieval with Chain-of-Thought）
IRCoT将检索与思维链推理交替进行，每一步推理后检索补充信息：
class IRCoT:
"""IRCoT: 交替检索与思维链推理"""
def __init__(self, llm_client, retriever, max_steps: int = 5):
self.llm = llm_client
self.retriever = retriever
self.max_steps = max_steps
def query(self, user_query: str) -> Dict:
"""IRCoT查询流程"""
reasoning_chain = []
all_documents = []
# 初始思维链
thought = self._generate_initial_thought(user_query)
reasoning_chain.append(thought)
for step in range(self.max_steps):
# 从当前思维链中提取检索需求
retrieval_need = self._extract_retrieval_need(
user_query, reasoning_chain
)
if retrieval_need.get("needs_retrieval", False):
# 检索补充信息
query = retrieval_need.get("query", user_query)
docs = self.retriever.retrieve(query, top_k=3)
all_documents.extend(docs)
# 基于新信息继续推理
context = "\n".join([doc.content[:200] for doc in docs])
thought = self._continue_reasoning(
user_query, reasoning_chain, context
)
else:
# 无需继续检索，生成最终推理
thought = self._final_reasoning(user_query, reasoning_chain)
reasoning_chain.append(thought)
# 检查是否完成推理
if self._is_reasoning_complete(user_query, reasoning_chain):
break
# 生成最终答案
answer = self._generate_answer(user_query, reasoning_chain, all_documents)
return {
"answer": answer,
"reasoning_chain": reasoning_chain,
"num_retrievals": len(all_documents),
"num_steps": len(reasoning_chain),
}
def _generate_initial_thought(self, query: str) -> str:
"""生成初始推理"""
prompt = f"""你是一个善于逐步推理的AI。对于以下问题，请开始你的推理过程。
问题：{query}
开始推理（说明你需要知道什么信息来回答这个问题）："""
return self.llm.generate(prompt)
def _extract_retrieval_need(self, query: str, chain: List[str]) -> Dict:
"""从推理链中提取检索需求"""
chain_text = "\n".join(chain)
prompt = f"""基于当前推理过程，判断是否需要检索更多信息。
问题：{query}
当前推理：
{chain_text}
如果需要检索，返回JSON:
{{"needs_retrieval": true, "query": "需要检索的内容"}}
如果推理已足够回答，返回:
{{"needs_retrieval": false}}"""
response = self.llm.generate(prompt)
try:
return json.loads(response)
except:
return {"needs_retrieval": False}
def _continue_reasoning(self, query: str, chain: List[str],
new_context: str) -> str:
"""基于新信息继续推理"""
chain_text = "\n".join(chain)
prompt = f"""基于新的信息继续推理。
问题：{query}
已有推理：
{chain_text}
新获取的信息：
{new_context}
继续推理："""
return self.llm.generate(prompt)
def _final_reasoning(self, query: str, chain: List[str]) -> str:
"""生成最终推理"""
chain_text = "\n".join(chain)
return self.llm.generate(
f"问题: {query}\n推理过程:\n{chain_text}\n\n最终推理总结:"
)
def _is_reasoning_complete(self, query: str, chain: List[str]) -> bool:
"""判断推理是否完成"""
chain_text = "\n".join(chain[-2:])
prompt = f"""判断以下推理是否已经足够回答问题。
问题：{query}
推理：{chain_text}
回答 yes 或 no。"""
return "yes" in self.llm.generate(prompt).strip().lower()
def _generate_answer(self, query: str, chain: List[str],
docs: List[Document]) -> str:
"""生成最终答案"""
reasoning = "\n".join(chain)
context = "\n".join([doc.content[:200] for doc in docs[:5]])
return self.llm.generate(
f"参考资料:\n{context}\n\n推理过程:\n{reasoning}\n\n"
f"问题: {query}\n\n基于推理和资料的最终回答:"
)
### Step-back Prompting
Step-back通过"退一步"提问，先获取背景知识，再回答原始问题：
class StepBackRAG:
"""Step-back RAG: 退一步检索增强"""
def __init__(self, llm_client, retriever):
self.llm = llm_client
self.retriever = retriever
def query(self, user_query: str) -> Dict:
"""Step-back查询流程"""
# 1. 生成"退一步"问题
step_back_query = self._generate_step_back_question(user_query)
# 2. 检索背景知识
background_docs = self.retriever.retrieve(step_back_query, top_k=5)
background = "\n".join([doc.content for doc in background_docs])
# 3. 同时检索原始问题
direct_docs = self.retriever.retrieve(user_query, top_k=3)
direct_context = "\n".join([doc.content for doc in direct_docs])
# 4. 综合生成答案
answer = self._generate(user_query, step_back_query, background, direct_context)
return {
"answer": answer,
"step_back_query": step_back_query,
"background_docs": len(background_docs),
"direct_docs": len(direct_docs),
}
def _generate_step_back_question(self, query: str) -> str:
"""生成退一步问题"""
prompt = f"""将以下具体问题抽象为一个更高层次的背景知识问题。
原始问题：{query}
例如：
- 原始: "Python 3.12的新特性有哪些？"
- 退一步: "Python语言的版本演进历史是什么？"
- 原始: "Transformer的注意力机制如何工作？"
- 退一步: "什么是注意力机制？"
退一步问题："""
return self.llm.generate(prompt).strip()
def _generate(self, query: str, step_back_query: str,
background: str, direct: str) -> str:
return self.llm.generate(f"""背景知识（来自: {step_back_query}）:
{background}
直接相关信息:
{direct}
问题: {query}
请综合背景知识和直接信息，提供全面的回答:""")
查询规划与分解
多步查询规划器
class QueryPlanner:
"""查询规划器 - 将复杂查询分解为可执行的子任务"""
def __init__(self, llm_client):
self.llm = llm_client
def plan(self, query: str) -> Dict:
"""为复杂查询生成执行计划"""
prompt = f"""分析以下查询，生成一个检索和推理的执行计划。
查询：{query}
请返回JSON格式的执行计划：
{{
"query_type": "factual/comparative/analytical/procedural",
"complexity": "simple/moderate/complex",
"sub_queries": [
{{"id": 1, "query": "子查询1", "purpose": "目的说明", "depends_on": []}},
{{"id": 2, "query": "子查询2", "purpose": "目的说明", "depends_on": [1]}}
],
"synthesis_strategy": "如何综合各子查询结果"
}}"""
response = self.llm.generate(prompt)
try:
return json.loads(response)
except:
return {
"query_type": "factual",
"complexity": "moderate",
"sub_queries": [{"id": 1, "query": query, "purpose": "直接查询", "depends_on": []}],
"synthesis_strategy": "直接使用结果",
}
def execute_plan(self, plan: Dict, retriever, llm_client) -> Dict:
"""执行查询计划"""
results = {}
# 按依赖关系排序执行
sorted_queries = sorted(plan["sub_queries"], key=lambda q: len(q.get("depends_on", [])))
for sub_query in sorted_queries:
qid = sub_query["id"]
query_text = sub_query["query"]
# 如果有依赖，将依赖结果加入查询
if sub_query.get("depends_on"):
dep_context = "\n".join([
f"前置信息{i}: {results.get(dep_id, '')[:200]}"
for dep_id in sub_query["depends_on"]
])
query_text = f"{query_text}\n\n背景：{dep_context}"
# 检索
docs = retriever.retrieve(query_text, top_k=3)
context = "\n".join([doc.content for doc in docs])
# 生成子查询答案
answer = llm_client.generate(
f"参考资料:\n{context}\n\n问题: {query_text}\n\n回答:"
)
results[qid] = answer
# 综合所有结果
synthesis = self._synthesize(plan, results, llm_client)
return {
"sub_results": results,
"synthesis": synthesis,
}
def _synthesize(self, plan: Dict, results: Dict, llm_client) -> str:
"""综合所有子查询结果"""
results_text = "\n".join([
f"子问题{i}: {answer}"
for i, answer in results.items()
])
return llm_client.generate(
f"综合以下各子问题的回答，生成完整答案：\n\n{results_text}\n\n"
f"综合策略: {plan.get('synthesis_strategy', '')}\n\n完整答案:"
)
多工具协同检索
Agentic RAG可以同时利用多种检索工具和数据源：
from typing import List, Dict, Any, Callable
from dataclasses import dataclass
@dataclass
class ToolResult:
tool_name: str
content: str
confidence: float
metadata: dict = None
class MultiToolRetriever:
"""多工具协同检索器"""
def __init__(self, llm_client):
self.llm = llm_client
self.tools: Dict[str, Callable] = {}
self.tool_descriptions: Dict[str, str] = {}
def register_tool(self, name: str, func: Callable, description: str):
"""注册检索工具"""
self.tools[name] = func
self.tool_descriptions[name] = description
def retrieve(self, query: str) -> List[ToolResult]:
"""使用多工具协同检索"""
# 1. 选择合适的工具
selected_tools = self._select_tools(query)
# 2. 并行执行检索
results = []
for tool_name in selected_tools:
try:
tool_func = self.tools[tool_name]
content = tool_func(query)
results.append(ToolResult(
tool_name=tool_name,
content=content,
confidence=self._assess_confidence(query, content),
))
except Exception as e:
results.append(ToolResult(
tool_name=tool_name,
content=f"工具调用失败: {str(e)}",
confidence=0.0,
))
# 3. 融合结果
return self._fuse_results(query, results)
def _select_tools(self, query: str) -> List[str]:
"""智能选择工具"""
tool_desc = "\n".join([
f"- {name}: {desc}"
for name, desc in self.tool_descriptions.items()
])
prompt = f"""为以下查询选择最合适的检索工具（1-3个）。
查询：{query}
可用工具：
{tool_desc}
返回JSON数组格式的工具名称列表。"""
response = self.llm.generate(prompt)
try:
selected = json.loads(response)
return [t for t in selected if t in self.tools]
except:
return list(self.tools.keys())[:2]
def _assess_confidence(self, query: str, content: str) -> float:
"""评估检索结果置信度"""
prompt = f"查询: {query}\n结果: {content[:200]}\n置信度(0-1):"
try:
return float(self.llm.generate(prompt).strip())
except:
return 0.5
def _fuse_results(self, query: str, results: List[ToolResult]) -> List[ToolResult]:
"""融合多工具结果"""
# 按置信度排序
results.sort(key=lambda r: r.confidence, reverse=True)
return results
# 使用示例
class VectorSearchTool:
def __init__(self, vector_store, embedder):
self.store = vector_store
self.embedder = embedder
def search(self, query: str) -> str:
emb = self.embedder.encode([query])[0]
docs = self.store.search(emb, top_k=3)
return "\n".join([doc.content for doc in docs])
class WebSearchTool:
def search(self, query: str) -> str:
# 集成实际的Web搜索API
return f"Web search results for: {query}"
class KnowledgeGraphTool:
def query(self, query: str) -> str:
# 集成知识图谱查询
return f"KG results for: {query}"
# 创建多工具检索器
multi_retriever = MultiToolRetriever(llm_client=None)
multi_retriever.register_tool("vector_search", VectorSearchTool(None, None).search,
"本地文档向量搜索，适合已有知识库查询")
multi_retriever.register_tool("web_search", WebSearchTool().search,
"互联网搜索，适合最新信息和外部知识")
multi_retriever.register_tool("knowledge_graph", KnowledgeGraphTool().query,
"知识图谱查询，适合实体关系和结构化知识")
GraphRAG知识图谱增强
GraphRAG将知识图谱与RAG结合，利用图结构增强检索和推理能力：
from typing import List, Dict, Set, Tuple
from dataclasses import dataclass, field
@dataclass
class KGEntity:
name: str
entity_type: str
properties: dict = field(default_factory=dict)
@dataclass
class KGRelation:
source: str
relation: str
target: str
properties: dict = field(default_factory=dict)
class KnowledgeGraph:
"""轻量级知识图谱"""
def __init__(self):
self.entities: Dict[str, KGEntity] = {}
self.relations: List[KGRelation] = []
self.adjacency: Dict[str, List[Tuple[str, str]]] = {} # entity -> [(relation, entity)]
def add_entity(self, name: str, entity_type: str, **properties):
"""添加实体"""
self.entities[name] = KGEntity(name, entity_type, properties)
if name not in self.adjacency:
self.adjacency[name] = []
def add_relation(self, source: str, relation: str, target: str, **properties):
"""添加关系"""
self.relations.append(KGRelation(source, relation, target, properties))
if source not in self.adjacency:
self.adjacency[source] = []
self.adjacency[source].append((relation, target))
def get_neighbors(self, entity: str, depth: int = 1) -> Dict:
"""获取实体的邻居"""
visited = set()
result = {"entities": {}, "relations": []}
def dfs(current, current_depth):
if current_depth > depth or current in visited:
return
visited.add(current)
if current in self.entities:
result["entities"][current] = self.entities[current]
for rel, neighbor in self.adjacency.get(current, []):
result["relations"].append((current, rel, neighbor))
dfs(neighbor, current_depth + 1)
dfs(entity, 0)
return result
def find_path(self, source: str, target: str, max_depth: int = 3) -> List[List[str]]:
"""查找两个实体之间的路径"""
paths = []
def dfs(current, path, visited):
if len(path) > max_depth:
return
if current == target:
paths.append(list(path))
return
for rel, neighbor in self.adjacency.get(current, []):
if neighbor not in visited:
visited.add(neighbor)
path.append((rel, neighbor))
dfs(neighbor, path, visited)
path.pop()
visited.discard(neighbor)
dfs(source, [(None, source)], {source})
return paths
class GraphRAG:
"""GraphRAG: 知识图谱增强的RAG"""
def __init__(self, llm_client, kg: KnowledgeGraph, vector_retriever=None):
self.llm = llm_client
self.kg = kg
self.vector_retriever = vector_retriever
def query(self, user_query: str) -> Dict:
"""GraphRAG查询流程"""
# 1. 从查询中提取实体
entities = self._extract_entities(user_query)
# 2. 从知识图谱获取相关信息
kg_context = ""
for entity in entities:
if entity in self.kg.entities:
neighbors = self.kg.get_neighbors(entity, depth=2)
kg_context += self._format_kg_context(entity, neighbors)
# 3. 从向量库获取补充信息
vector_context = ""
if self.vector_retriever:
docs = self.vector_retriever.retrieve(user_query, top_k=3)
vector_context = "\n".join([doc.content for doc in docs])
# 4. 融合生成答案
answer = self._generate(user_query, kg_context, vector_context)
return {
"answer": answer,
"entities_found": entities,
"has_kg_context": bool(kg_context),
"has_vector_context": bool(vector_context),
}
def _extract_entities(self, query: str) -> List[str]:
"""从查询中提取实体"""
entity_list = list(self.kg.entities.keys())[:50] # 限制长度
prompt = f"""从以下查询中识别实体名称。
查询：{query}
候选实体列表：
{', '.join(entity_list)}
返回JSON数组格式的实体名称列表（只返回在候选列表中存在的实体）。"""
response = self.llm.generate(prompt)
try:
return json.loads(response)
except:
return []
def _format_kg_context(self, entity: str, neighbors: Dict) -> str:
"""格式化知识图谱上下文"""
lines = [f"## 实体: {entity}"]
if entity in neighbors.get("entities", {}):
ent = neighbors["entities"][entity]
lines.append(f"类型: {ent.entity_type}")
for k, v in ent.properties.items():
lines.append(f"{k}: {v}")
lines.append("\n关系:")
for src, rel, tgt in neighbors.get("relations", []):
lines.append(f" {src} --[{rel}]--> {tgt}")
return "\n".join(lines) + "\n\n"
def _generate(self, query: str, kg_context: str, vector_context: str) -> str:
"""融合知识图谱和向量检索结果生成答案"""
prompt = f"""基于以下信息回答问题。
知识图谱信息：
{kg_context if kg_context else "无相关信息"}
文档信息：
{vector_context if vector_context else "无相关信息"}
问题：{query}
请综合以上信息，提供准确、全面的回答。如果某些信息缺失，请基于已有信息作答并说明不足。"""
return self.llm.generate(prompt)
# 使用示例
kg = KnowledgeGraph()
# 构建知识图谱
kg.add_entity("Python", "programming_language", version="3.12", creator="Guido van Rossum")
kg.add_entity("Guido van Rossum", "person", nationality="Dutch")
kg.add_entity("Django", "framework", language="Python")
kg.add_entity("FastAPI", "framework", language="Python")
kg.add_relation("Python", "created_by", "Guido van Rossum")
kg.add_relation("Django", "built_with", "Python")
kg.add_relation("FastAPI", "built_with", "Python")
kg.add_relation("FastAPI", "alternative_to", "Django")
# 查询
# graph_rag = GraphRAG(llm_client, kg, vector_retriever)
# result = graph_rag.query("Python的创始人是谁？他还创建了什么？")
检索结果验证与幻觉检测
幻觉检测器
class HallucinationDetector:
"""幻觉检测器 - 检测生成内容中的幻觉"""
def __init__(self, llm_client):
self.llm = llm_client
def detect(self, generated_text: str, source_documents: List[str]) -> Dict:
"""检测生成文本中的幻觉"""
source_text = "\n".join(source_documents)
# 1. 逐句检查
sentences = self._split_sentences(generated_text)
sentence_results = []
for sentence in sentences:
is_supported = self._check_support(sentence, source_text)
sentence_results.append({
"sentence": sentence,
"is_supported": is_supported,
})
# 2. 计算整体幻觉分数
total = len(sentence_results)
supported = sum(1 for r in sentence_results if r["is_supported"])
hallucination_rate = 1 - (supported / total) if total > 0 else 0
return {
"hallucination_rate": hallucination_rate,
"total_sentences": total,
"supported_sentences": supported,
"hallucinated_sentences": total - supported,
"details": sentence_results,
"is_hallucinated": hallucination_rate > 0.3,
}
def _split_sentences(self, text: str) -> List[str]:
"""分句"""
import re
sentences = re.split(r'[。！？.!?]', text)
return [s.strip() for s in sentences if s.strip()]
def _check_support(self, sentence: str, source: str) -> bool:
"""检查句子是否被源文档支持"""
prompt = f"""判断以下句子是否被参考资料完全支持。
句子：{sentence}
参考资料：
{source[:1000]}
如果句子完全被参考资料支持，回答 yes。
如果句子包含参考资料中没有的信息，回答 no。
只回答 yes 或 no。"""
response = self.llm.generate(prompt).strip().lower()
return "yes" in response
class RetrievalValidator:
"""检索结果验证器"""
def __init__(self, llm_client):
self.llm = llm_client
def validate(self, query: str, documents: List[Document]) -> Dict:
"""验证检索结果的质量"""
if not documents:
return {"is_valid": False, "reason": "没有检索到文档"}
# 1. 相关性检查
relevance_scores = []
for doc in documents:
score = self._check_relevance(query, doc.content)
relevance_scores.append(score)
avg_relevance = sum(relevance_scores) / len(relevance_scores)
# 2. 信息充分性检查
context = "\n".join([doc.content[:200] for doc in documents])
sufficiency = self._check_sufficiency(query, context)
# 3. 一致性检查（文档之间是否矛盾）
consistency = self._check_consistency(documents)
return {
"is_valid": avg_relevance > 0.5 and sufficiency > 0.5,
"avg_relevance": avg_relevance,
"sufficiency": sufficiency,
"consistency": consistency,
"document_scores": list(zip(range(len(documents)), relevance_scores)),
}
def _check_relevance(self, query: str, content: str) -> float:
"""检查文档相关性"""
prompt = f"查询: {query}\n文档: {content[:300]}\n相关性(0-1):"
try:
return float(self.llm.generate(prompt).strip())
except:
return 0.5
def _check_sufficiency(self, query: str, context: str) -> float:
"""检查信息充分性"""
prompt = f"""评估以下参考资料是否足够回答问题。
问题：{query}
资料：{context[:500]}
充分性分数(0-1)，只返回数字："""
try:
return float(self.llm.generate(prompt).strip())
except:
return 0.5
def _check_consistency(self, documents: List[Document]) -> float:
"""检查文档间一致性"""
if len(documents) < 2:
return 1.0
docs_text = "\n---\n".join([doc.content[:200] for doc in documents[:3]])
prompt = f"""评估以下文档之间是否存在矛盾信息。
文档：
{docs_text}
一致性分数(0-1, 1=完全一致)，只返回数字："""
try:
return float(self.llm.generate(prompt).strip())
except:
return 0.5
Multi-Agent RAG协作
多个Agent协作完成复杂检索任务：
from typing import List, Dict
from enum import Enum
class AgentRole(Enum):
PLANNER = "planner" # 规划Agent
RETRIEVER = "retriever" # 检索Agent
REASONER = "reasoner" # 推理Agent
VALIDATOR = "validator" # 验证Agent
SYNTHESIZER = "synthesizer" # 综合Agent
class RAGAgent:
"""RAG协作Agent"""
def __init__(self, role: AgentRole, llm_client):
self.role = role
self.llm = llm_client
def process(self, task: str, context: dict = None) -> str:
"""处理任务"""
prompts = {
AgentRole.PLANNER: f"作为规划专家，分析以下查询并制定检索计划：\n{task}",
AgentRole.RETRIEVER: f"作为检索专家，根据以下计划执行检索：\n{task}",
AgentRole.REASONER: f"作为推理专家，基于以下信息进行推理：\n{task}",
AgentRole.VALIDATOR: f"作为验证专家，验证以下内容的准确性：\n{task}",
AgentRole.SYNTHESIZER: f"作为综合专家，整合以下信息生成最终答案：\n{task}",
}
prompt = prompts.get(self.role, task)
if context:
prompt += f"\n\n上下文信息：{json.dumps(context, ensure_ascii=False)[:1000]}"
return self.llm.generate(prompt)
class MultiAgentRAG:
"""Multi-Agent RAG协作系统"""
def __init__(self, llm_client, retriever):
self.llm = llm_client
self.retriever = retriever
self.agents = {
role: RAGAgent(role, llm_client)
for role in AgentRole
}
def query(self, user_query: str) -> Dict:
"""Multi-Agent协作查询"""
# 1. 规划Agent制定计划
plan = self.agents[AgentRole.PLANNER].process(user_query)
plan = self._parse_plan(plan)
# 2. 检索Agent执行检索
retrieval_results = []
for sub_task in plan.get("tasks", [user_query]):
docs = self.retriever.retrieve(sub_task, top_k=3)
retrieval_results.append({
"task": sub_task,
"documents": [doc.content[:300] for doc in docs],
})
# 3. 推理Agent进行推理
reasoning = self.agents[AgentRole.REASONER].process(
user_query,
{"retrieval_results": retrieval_results}
)
# 4. 验证Agent验证结果
validation = self.agents[AgentRole.VALIDATOR].process(
f"查询: {user_query}\n推理结果: {reasoning}",
{"source_documents": retrieval_results}
)
# 5. 综合Agent生成最终答案
final_answer = self.agents[AgentRole.SYNTHESIZER].process(
user_query,
{
"reasoning": reasoning,
"validation": validation,
"sources": retrieval_results,
}
)
return {
"answer": final_answer,
"plan": plan,
"reasoning": reasoning,
"validation": validation,
"num_agents_involved": 5,
}
def _parse_plan(self, plan_text: str) -> Dict:
"""解析规划结果"""
try:
return json.loads(plan_text)
except:
return {"tasks": [plan_text]}
# 使用示例
# multi_agent_rag = MultiAgentRAG(llm_client, retriever)
# result = multi_agent_rag.query("比较RAG和Fine-tuning在不同场景下的优劣")
LangGraph实现Agentic RAG
LangGraph是构建有状态、多步骤Agent应用的理想框架。以下是用LangGraph实现完整Agentic RAG的示例：
"""
LangGraph Agentic RAG实现
安装依赖:
pip install langgraph langchain langchain-openai
"""
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.documents import Document
import operator
# 定义状态
class RAGState(TypedDict):
query: str
query_type: str # "simple", "complex", "multi_hop"
sub_queries: List[str]
retrieved_docs: Annotated[List[Document], operator.add]
current_iteration: int
max_iterations: int
reasoning: str
answer: str
quality_score: float
needs_refinement: bool
# 节点函数
def analyze_query(state: RAGState) -> dict:
"""分析查询"""
llm = ChatOpenAI(model="gpt-4o-mini")
query = state["query"]
response = llm.invoke(f"""分析查询类型：
查询：{query}
类型选项：simple, complex, multi_hop
只返回类型名称。""")
query_type = response.content.strip().lower()
if query_type not in ["simple", "complex", "multi_hop"]:
query_type = "complex"
return {"query_type": query_type}
def generate_sub_queries(state: RAGState) -> dict:
"""生成子查询"""
llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke(f"""将以下查询分解为2-3个子查询：
查询：{state['query']}
返回JSON数组格式。""")
try:
sub_queries = eval(response.content)
except:
sub_queries = [state["query"]]
return {"sub_queries": sub_queries}
def retrieve_documents(state: RAGState) -> dict:
"""检索文档（需要接入实际检索器）"""
# 这里是示例，实际需要接入向量检索器
queries = state.get("sub_queries", [state["query"]])
docs = []
for q in queries:
# retrieved = retriever.retrieve(q, top_k=3)
# docs.extend(retrieved)
docs.append(Document(page_content=f"检索结果 for: {q}", metadata={"query": q}))
return {"retrieved_docs": docs, "current_iteration": state.get("current_iteration", 0) + 1}
def evaluate_quality(state: RAGState) -> dict:
"""评估检索质量"""
llm = ChatOpenAI(model="gpt-4o-mini")
docs_text = "\n".join([doc.page_content[:200] for doc in state["retrieved_docs"]])
response = llm.invoke(f"""评估检索结果质量(0-1)：
查询：{state['query']}
文档：{docs_text}
只返回数字。""")
try:
score = float(response.content.strip())
except:
score = 0.5
return {
"quality_score": score,
"needs_refinement": score < 0.7 and state.get("current_iteration", 0) < state.get("max_iterations", 3),
}
def refine_query(state: RAGState) -> dict:
"""优化查询"""
llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke(f"""优化以下查询以获得更好的检索结果：
原始查询：{state['query']}
当前质量分数：{state['quality_score']}
返回优化后的查询。""")
return {"query": response.content.strip()}
def generate_reasoning(state: RAGState) -> dict:
"""生成推理链"""
llm = ChatOpenAI(model="gpt-4o-mini")
docs_text = "\n".join([doc.page_content for doc in state["retrieved_docs"]])
response = llm.invoke(f"""基于以下信息进行推理：
查询：{state['query']}
文档：{docs_text}
请进行逐步推理：""")
return {"reasoning": response.content}
def generate_answer(state: RAGState) -> dict:
"""生成最终答案"""
llm = ChatOpenAI(model="gpt-4o-mini")
docs_text = "\n".join([doc.page_content for doc in state["retrieved_docs"]])
response = llm.invoke(f"""基于以下信息回答问题：
参考资料：{docs_text}
推理过程：{state.get('reasoning', '')}
问题：{state['query']}
请提供准确、详细的回答：""")
return {"answer": response.content}
# 路由函数
def route_after_analysis(state: RAGState) -> str:
"""查询分析后的路由"""
if state["query_type"] == "simple":
return "generate_answer"
return "generate_sub_queries"
def route_after_evaluation(state: RAGState) -> str:
"""质量评估后的路由"""
if state.get("needs_refinement", False):
return "refine_query"
return "generate_reasoning"
# 构建LangGraph
def build_agentic_rag_graph():
"""构建Agentic RAG图"""
workflow = StateGraph(RAGState)
# 添加节点
workflow.add_node("analyze_query", analyze_query)
workflow.add_node("generate_sub_queries", generate_sub_queries)
workflow.add_node("retrieve_documents", retrieve_documents)
workflow.add_node("evaluate_quality", evaluate_quality)
workflow.add_node("refine_query", refine_query)
workflow.add_node("generate_reasoning", generate_reasoning)
workflow.add_node("generate_answer", generate_answer)
# 设置入口
workflow.set_entry_point("analyze_query")
# 添加边
workflow.add_conditional_edges(
"analyze_query",
route_after_analysis,
{
"generate_answer": "generate_answer",
"generate_sub_queries": "generate_sub_queries",
}
)
workflow.add_edge("generate_sub_queries", "retrieve_documents")
workflow.add_edge("retrieve_documents", "evaluate_quality")
workflow.add_conditional_edges(
"evaluate_quality",
route_after_evaluation,
{
"refine_query": "refine_query",
"generate_reasoning": "generate_reasoning",
}
)
workflow.add_edge("refine_query", "retrieve_documents")
workflow.add_edge("generate_reasoning", "generate_answer")
workflow.add_edge("generate_answer", END)
return workflow.compile()
# 使用示例
"""
graph = build_agentic_rag_graph()
result = graph.invoke({
"query": "比较Transformer和RNN在序列建模任务中的优劣",
"query_type": "",
"sub_queries": [],
"retrieved_docs": [],
"current_iteration": 0,
"max_iterations": 3,
"reasoning": "",
"answer": "",
"quality_score": 0.0,
"needs_refinement": False,
})
print(f"答案: {result['answer']}")
print(f"迭代次数: {result['current_iteration']}")
print(f"质量分数: {result['quality_score']}")
"""
生产部署与优化
性能优化策略
class RAGOptimizer:
"""RAG系统优化器"""
@staticmethod
def optimize_retrieval(retriever, query_cache_size: int = 1000):
"""优化检索性能"""
from functools import lru_cache
# 1. 查询缓存
@lru_cache(maxsize=query_cache_size)
def cached_retrieve(query_hash: str, top_k: int):
return retriever.retrieve(query_hash, top_k)
# 2. 批量编码优化
# 3. 向量索引优化（使用HNSW、IVF等）
return cached_retrieve
@staticmethod
def optimize_generation(llm_client, max_context_length: int = 4000):
"""优化生成性能"""
def optimized_generate(query: str, context: str) -> str:
# 截断过长的上下文
if len(context) > max_context_length:
context = context[:max_context_length] + "..."
# 使用流式生成
return llm_client.generate(query, context)
return optimized_generate
@staticmethod
def implement_caching():
"""实现多级缓存"""
# L1: 查询结果缓存
# L2: 嵌入向量缓存
# L3: 文档分块缓存
pass
class RAGMetrics:
"""RAG系统监控指标"""
def __init__(self):
self.metrics = {
"total_queries": 0,
"avg_latency": 0.0,
"avg_retrieval_quality": 0.0,
"avg_answer_quality": 0.0,
"cache_hit_rate": 0.0,
"error_rate": 0.0,
}
self.query_log: List[Dict] = []
def log_query(self, query: str, result: Dict, latency: float):
"""记录查询"""
self.metrics["total_queries"] += 1
self.query_log.append({
"query": query,
"strategy": result.get("strategy"),
"latency": latency,
"quality": result.get("quality_score", 0),
})
# 更新平均值
n = self.metrics["total_queries"]
self.metrics["avg_latency"] = (
self.metrics["avg_latency"] * (n-1) + latency
) / n
def get_report(self) -> str:
"""生成监控报告"""
return f"""RAG系统监控报告
==================
总查询数: {self.metrics['total_queries']}
平均延迟: {self.metrics['avg_latency']:.2f}秒
平均检索质量: {self.metrics['avg_retrieval_quality']:.2f}
缓存命中率: {self.metrics['cache_hit_rate']:.1%}
错误率: {self.metrics['error_rate']:.1%}
"""
生产部署检查清单
class ProductionChecklist:
"""生产部署检查清单"""
CHECKS = {
"retrieval": [
"向量索引已构建并优化",
"混合检索已配置（语义+关键词）",
"重排序模型已部署",
"检索缓存已启用",
],
"generation": [
"LLM调用有重试机制",
"输出长度已限制",
"流式响应已支持",
"幻觉检测已启用",
],
"agentic": [
"查询分析器已测试",
"策略路由器已验证",
"最大迭代次数已设置",
"超时机制已配置",
],
"monitoring": [
"日志系统已部署",
"指标采集已配置",
"告警规则已设置",
"用户反馈渠道已建立",
],
"security": [
"输入验证已启用",
"输出过滤已配置",
"访问控制已实施",
"敏感信息保护已启用",
],
}
def run_check(self) -> Dict:
"""运行检查"""
results = {}
for category, checks in self.CHECKS.items():
results[category] = {
"total": len(checks),
"checks": checks,
}
return results
def generate_report(self) -> str:
"""生成检查报告"""
lines = ["# Agentic RAG 生产部署检查清单\n"]
for category, info in self.run_check().items():
lines.append(f"## {category.upper()}")
for check in info["checks"]:
lines.append(f"- [ ] {check}")
lines.append("")
return "\n".join(lines)
总结
Agentic RAG技术全景
本教程涵盖了Agentic RAG的核心技术栈：
| 技术 | 核心思想 | 适用场景 |
|---|---|---|
| Adaptive RAG | 根据查询复杂度选择策略 | 通用场景 |
| Self-RAG | 自我评估检索必要性和质量 | 需要高质量输出 |
| CRAG | 检索质量差时自动纠正 | 检索质量不稳定 |
| IRCoT | 推理与检索交替进行 | 多步推理问题 |
| Step-back | 先获取背景知识再回答 | 需要专业知识 |
| GraphRAG | 知识图谱增强 | 实体关系密集 |
| Multi-Agent | 多Agent协作 | 复杂综合任务 |
选型建议
- 简单问答：传统RAG即可，无需Agentic
- 复杂推理：IRCoT或Step-back RAG
- 质量要求高：Self-RAG或CRAG
- 知识图谱丰富：GraphRAG
- 多源数据：Multi-Agent RAG + 多工具协同
- 生产环境：Adaptive RAG + LangGraph编排
推荐技术栈
- 框架：LangGraph、LlamaIndex、Haystack
- 向量数据库：Milvus、Qdrant、Weaviate、Chroma
- 嵌入模型：text-embedding-3-small、BGE、E5
- 重排序：Cohere Rerank、BGE Reranker、Cross-encoder
- LLM：GPT-4o、Claude 3.5、DeepSeek-V3
通过本教程的技术和方法，开发者可以构建从简单到复杂的各层级Agentic RAG系统，满足不同场景的智能检索需求。
参考资源：
- Self-RAG论文: "Self-RAG: Learning to Retrieve, Generate, and Critique"
- CRAG论文: "Corrective Retrieval Augmented Generation"
- IRCoT论文: "Enhancing Chain-of-Thoughts Prompting with Iterative Retrieval Augmentation"
- GraphRAG论文: "From Local to Global: A Graph RAG Approach"
- LangGraph文档: langchain-ai.github.io/langgraph
