RAG高级模式
概述
本章介绍RAG(Retrieval-Augmented Generation)的高级应用模式,包括多模态RAG、自适应RAG、多步RAG、图RAG等。
多模态RAG
1. 图像RAG
处理图像数据的RAG:
python
# 导入图像描述加载器
from langchain_community.document_loaders import ImageCaptionLoader
# 导入Hugging Face嵌入模型
from langchain_community.embeddings import HuggingFaceEmbeddings
# 导入Chroma向量存储
from langchain_community.vectorstores import Chroma
# 图像描述加载器
# ImageCaptionLoader:加载图像并生成描述
# 参数:
# path_or_url:图像路径或URL
# blip_model:BLIP模型名称,用于生成图像描述
loader = ImageCaptionLoader(
path_or_url="./images", # 图像目录
blip_model="Salesforce/blip-image-captioning-base" # BLIP模型
)
documents = loader.load()
# 使用多模态嵌入
# HuggingFaceEmbeddings:使用CLIP模型进行多模态嵌入
# CLIP模型:可以同时处理图像和文本
embeddings = HuggingFaceEmbeddings(
model_name="clip-ViT-B-32" # CLIP模型
)
# 创建向量存储
vectorstore = Chroma.from_documents(
documents=documents,
embedding=embeddings
)2. 表格RAG
处理表格数据的RAG:
python
# 导入Pandas和DataFrame加载器
import pandas as pd
from langchain_community.document_loaders import DataFrameLoader
# 加载CSV文件
# pd.read_csv():读取CSV文件为DataFrame
df = pd.read_csv("data.csv")
# 转换为文档
# DataFrameLoader:将DataFrame转换为Document对象
# 参数:
# df:DataFrame对象
# page_content_column:用作页面内容的列名
loader = DataFrameLoader(df, page_content_column="description")
documents = loader.load()
# 创建向量存储
vectorstore = Chroma.from_documents(
documents=documents,
embedding=embeddings
)3. 音频RAG
处理音频数据的RAG:
python
# 导入音频加载器和Whisper解析器
from langchain_community.document_loaders import (
AudioLoader, # 音频加载器
YoutubeAudioLoader # YouTube音频加载器
)
from langchain_community.document_loaders.parsers.audio import OpenAIWhisperParser
# 加载音频文件
# AudioLoader:加载本地音频文件
loader = AudioLoader("./audio/")
documents = loader.load()
# 使用Whisper解析
# OpenAIWhisperParser:使用OpenAI Whisper模型解析音频
# Whisper:OpenAI的语音识别模型
parser = OpenAIWhisperParser()
parsed_docs = parser.parse(documents)自适应RAG
1. 查询路由
根据查询类型路由到不同的处理流程:
python
# 导入LangGraph核心组件
from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal
# 定义查询状态
class QueryState(TypedDict):
"""
查询状态定义
属性:
query (str): 用户查询
query_type (str): 查询类型(weather/news/knowledge)
result (str): 查询结果
"""
query: str
query_type: str
result: str
# 定义节点
def classify_query(state: QueryState) -> QueryState:
"""
分类查询节点
参数:
state (QueryState): 当前状态
返回值:
QueryState: 更新后的状态
功能:根据查询内容分类
"""
query = state["query"].lower()
# 根据关键词分类
if "天气" in query:
query_type = "weather"
elif "新闻" in query:
query_type = "news"
else:
query_type = "knowledge"
return {"query_type": query_type}
def handle_weather(state: QueryState) -> QueryState:
"""处理天气查询"""
return {"result": f"天气查询结果: {state['query']}"}
def handle_news(state: QueryState) -> QueryState:
"""处理新闻查询"""
return {"result": f"新闻查询结果: {state['query']}"}
def handle_knowledge(state: QueryState) -> QueryState:
"""处理知识查询"""
# 这里使用RAG
return {"result": f"知识查询结果: {state['query']}"}
# 创建图
graph = StateGraph(QueryState)
# 添加节点
graph.add_node("classify", classify_query)
graph.add_node("weather", handle_weather)
graph.add_node("news", handle_news)
graph.add_node("knowledge", handle_knowledge)
# 设置条件边
def route_function(state: QueryState) -> str:
"""
路由函数
参数:
state (QueryState): 当前状态
返回值:
str: 路由目标
"""
return state["query_type"]
# 添加条件边
graph.add_conditional_edges(
"classify",
route_function,
{
"weather": "weather",
"news": "news",
"knowledge": "knowledge"
}
)
# 设置出口
graph.add_edge("weather", END)
graph.add_edge("news", END)
graph.add_edge("knowledge", END)
# 编译和执行
graph.set_entry_point("classify")
app = graph.compile()2. 自适应检索
根据查询复杂度选择检索策略:
python
# 导入类型提示和OpenAI模型
from typing import List, Dict
from langchain_openai import ChatOpenAI
class AdaptiveRetriever:
"""
自适应检索器
功能:根据查询复杂度选择检索策略
"""
def __init__(self, simple_retriever, complex_retriever):
"""
初始化自适应检索器
参数:
simple_retriever:简单检索器
complex_retriever:复杂检索器
"""
self.simple_retriever = simple_retriever
self.complex_retriever = complex_retriever
self.llm = ChatOpenAI(model="gpt-4o-mini")
def classify_complexity(self, query: str) -> str:
"""
分类查询复杂度
参数:
query (str): 查询文本
返回值:
str: 复杂度(simple/complex)
"""
# 使用LLM分类
response = self.llm.invoke(
f"判断以下查询的复杂度(简单/复杂):{query}"
)
if "简单" in response.content:
return "simple"
return "complex"
def retrieve(self, query: str) -> List[Dict]:
"""
自适应检索
参数:
query (str): 查询文本
返回值:
List[Dict]: 检索结果
"""
# 分类复杂度
complexity = self.classify_complexity(query)
# 根据复杂度选择检索器
if complexity == "simple":
return self.simple_retriever.invoke(query)
else:
return self.complex_retriever.invoke(query)多步RAG
1. 迭代检索
多次迭代检索以提高质量:
python
# 导入OpenAI模型和类型提示
from langchain_openai import ChatOpenAI
from typing import List
class IterativeRAG:
"""
迭代RAG
功能:多次迭代检索以提高质量
"""
def __init__(self, vectorstore, max_iterations=3):
"""
初始化迭代RAG
参数:
vectorstore:向量存储
max_iterations:最大迭代次数
"""
self.vectorstore = vectorstore
self.max_iterations = max_iterations
self.llm = ChatOpenAI(model="gpt-4o-mini")
def retrieve(self, query: str) -> List[str]:
"""
迭代检索
参数:
query (str): 查询文本
返回值:
List[str]: 检索结果
"""
all_docs = []
current_query = query
for i in range(self.max_iterations):
# 检索
docs = self.vectorstore.similarity_search(current_query, k=3)
all_docs.extend(docs)
# 生成新查询
if i < self.max_iterations - 1:
current_query = self.generate_new_query(query, all_docs)
return all_docs
def generate_new_query(self, original_query: str, docs: List[str]) -> str:
"""
生成新查询
参数:
original_query (str): 原始查询
docs (List[str]): 已检索的文档
返回值:
str: 新查询
"""
# 合并文档内容
context = "\n".join([doc.page_content for doc in docs])
# 使用LLM生成新查询
response = self.llm.invoke(
f"基于以下上下文,生成一个更具体的查询:\n"
f"原始查询:{original_query}\n"
f"上下文:{context}"
)
return response.content2. 分层检索
使用分层策略进行检索:
python
# 导入类型提示和Chroma向量存储
from typing import List, Dict
from langchain_community.vectorstores import Chroma
class HierarchicalRetriever:
"""
分层检索器
功能:使用分层策略进行检索
"""
def __init__(self, vectorstores: List[Chroma]):
"""
初始化分层检索器
参数:
vectorstores:向量存储列表,从粗粒度到细粒度
"""
self.vectorstores = vectorstores
def retrieve(self, query: str, k: int = 5) -> List[Dict]:
"""
分层检索
参数:
query (str): 查询文本
k (int): 返回的文档数量
返回值:
List[Dict]: 检索结果
"""
results = []
# 从粗粒度到细粒度检索
for i, vectorstore in enumerate(self.vectorstores):
docs = vectorstore.similarity_search(query, k=k)
for doc in docs:
doc.metadata["level"] = i # 记录检索层级
results.append(doc)
# 去重和排序
unique_results = self.deduplicate(results)
return unique_results[:k]
def deduplicate(self, docs: List[Dict]) -> List[Dict]:
"""
去重
参数:
docs (List[Dict]): 文档列表
返回值:
List[Dict]: 去重后的文档列表
"""
seen = set()
unique_docs = []
for doc in docs:
if doc.page_content not in seen:
seen.add(doc.page_content)
unique_docs.append(doc)
return unique_docs图RAG
1. 知识图谱RAG
结合知识图谱的RAG:
python
# 导入类型提示和NetworkX图库
from typing import List, Dict, Tuple
import networkx as nx
class KnowledgeGraphRAG:
"""
知识图谱RAG
功能:结合知识图谱的RAG
"""
def __init__(self):
"""初始化知识图谱RAG"""
self.graph = nx.DiGraph() # 创建有向图
def add_entity(self, entity: str, properties: Dict):
"""
添加实体
参数:
entity (str): 实体名称
properties (Dict): 实体属性
"""
self.graph.add_node(entity, **properties)
def add_relation(self, source: str, target: str, relation: str):
"""
添加关系
参数:
source (str): 源实体
target (str): 目标实体
relation (str): 关系类型
"""
self.graph.add_edge(source, target, relation=relation)
def retrieve(self, query: str, max_hops: int = 2) -> List[Dict]:
"""
从知识图谱检索
参数:
query (str): 查询文本
max_hops (int): 最大跳数
返回值:
List[Dict]: 检索结果
"""
# 提取查询中的实体
entities = self.extract_entities(query)
# 检索相关实体和关系
results = []
for entity in entities:
if entity in self.graph:
# 获取邻居节点
neighbors = list(self.graph.neighbors(entity))
for neighbor in neighbors[:3]: # 限制数量
edge_data = self.graph[entity][neighbor]
results.append({
"source": entity,
"target": neighbor,
"relation": edge_data.get("relation", "")
})
return results
def extract_entities(self, query: str) -> List[str]:
"""
提取实体
参数:
query (str): 查询文本
返回值:
List[str]: 实体列表
注意:这里可以使用NER模型
"""
# 这里可以使用NER模型
return [query] # 简化实现2. 图嵌入RAG
使用图嵌入的RAG:
python
# 导入类型提示和NumPy
from typing import List, Dict
import numpy as np
class GraphEmbeddingRAG:
"""
图嵌入RAG
功能:使用图嵌入的RAG
"""
def __init__(self, graph, embedding_dim=128):
"""
初始化图嵌入RAG
参数:
graph:图对象
embedding_dim:嵌入维度
"""
self.graph = graph
self.embedding_dim = embedding_dim
self.embeddings = {}
def train_embeddings(self):
"""
训练图嵌入
功能:为图中的每个节点生成嵌入向量
"""
# 这里可以使用Node2Vec等算法
nodes = list(self.graph.nodes())
for i, node in enumerate(nodes):
# 简化的嵌入生成
self.embeddings[node] = np.random.random(self.embedding_dim)
def retrieve(self, query_embedding: np.ndarray, k: int = 5) -> List[Dict]:
"""
基于图嵌入检索
参数:
query_embedding (np.ndarray): 查询嵌入向量
k (int): 返回的结果数量
返回值:
List[Dict]: 检索结果
"""
similarities = []
for node, embedding in self.embeddings.items():
# 计算余弦相似度
similarity = np.dot(query_embedding, embedding) / (
np.linalg.norm(query_embedding) * np.linalg.norm(embedding)
)
similarities.append((node, similarity))
# 排序并返回top-k
similarities.sort(key=lambda x: x[1], reverse=True)
return [{"node": node, "similarity": sim} for node, sim in similarities[:k]]上下文增强RAG
1. 上下文压缩
压缩上下文以减少token使用:
python
# 导入类型提示和OpenAI模型
from typing import List
from langchain_openai import ChatOpenAI
class ContextCompressor:
"""
上下文压缩器
功能:压缩上下文以减少token使用
"""
def __init__(self):
"""初始化上下文压缩器"""
self.llm = ChatOpenAI(model="gpt-4o-mini")
def compress(self, query: str, context: List[str], max_length: int = 1000) -> str:
"""
压缩上下文
参数:
query (str): 查询文本
context (List[str]): 上下文列表
max_length (int): 最大长度
返回值:
str: 压缩后的上下文
"""
# 合并上下文
full_context = "\n".join(context)
# 如果上下文已经足够短,直接返回
if len(full_context) <= max_length:
return full_context
# 使用LLM压缩
response = self.llm.invoke(
f"请将以下上下文压缩到{max_length}字以内,保留与查询相关的关键信息:\n"
f"查询:{query}\n"
f"上下文:{full_context}"
)
return response.content2. 上下文增强
增强上下文以提供更多信息:
python
# 导入类型提示和OpenAI模型
from typing import List, Dict
from langchain_openai import ChatOpenAI
class ContextEnhancer:
"""
上下文增强器
功能:增强上下文以提供更多信息
"""
def __init__(self):
"""初始化上下文增强器"""
self.llm = ChatOpenAI(model="gpt-4o-mini")
def enhance(self, query: str, context: List[str]) -> List[str]:
"""
增强上下文
参数:
query (str): 查询文本
context (List[str]): 上下文列表
返回值:
List[str]: 增强后的上下文
"""
enhanced_context = []
for ctx in context:
# 生成相关问题
questions = self.generate_questions(ctx)
# 检索相关信息
for question in questions:
# 这里可以检索更多信息
enhanced_context.append(f"问题:{question}")
enhanced_context.append(ctx)
return enhanced_context
def generate_questions(self, context: str) -> List[str]:
"""
生成相关问题
参数:
context (str): 上下文
返回值:
List[str]: 问题列表
"""
response = self.llm.invoke(
f"基于以下上下文,生成3个相关问题:\n{context}"
)
return response.content.split("\n")[:3]多智能体RAG
1. 协作式RAG
多个智能体协作完成RAG:
python
# 导入类型提示和OpenAI模型
from typing import List, Dict
from langchain_openai import ChatOpenAI
class CollaborativeRAG:
"""
协作式RAG
功能:多个智能体协作完成RAG
"""
def __init__(self):
"""初始化协作式RAG"""
# 检索智能体:使用GPT-3.5-turbo
self.retriever_llm = ChatOpenAI(model="gpt-4o-mini")
# 分析智能体:使用GPT-4(更强大)
self.analyzer_llm = ChatOpenAI(model="gpt-4")
# 生成智能体:使用GPT-3.5-turbo
self.generator_llm = ChatOpenAI(model="gpt-4o-mini")
def retrieve(self, query: str) -> List[str]:
"""
检索智能体
参数:
query (str): 查询文本
返回值:
List[str]: 检索结果
"""
# 检索文档
docs = vectorstore.similarity_search(query, k=5)
return [doc.page_content for doc in docs]
def analyze(self, query: str, docs: List[str]) -> Dict:
"""
分析智能体
参数:
query (str): 查询文本
docs (List[str]): 文档列表
返回值:
Dict: 分析结果
"""
context = "\n".join(docs)
response = self.analyzer_llm.invoke(
f"分析以下查询和文档的相关性:\n"
f"查询:{query}\n"
f"文档:{context}"
)
return {"analysis": response.content}
def generate(self, query: str, docs: List[str], analysis: Dict) -> str:
"""
生成智能体
参数:
query (str): 查询文本
docs (List[str]): 文档列表
analysis (Dict): 分析结果
返回值:
str: 生成的回答
"""
context = "\n".join(docs)
response = self.generator_llm.invoke(
f"基于以下信息生成回答:\n"
f"查询:{query}\n"
f"文档:{context}\n"
f"分析:{analysis['analysis']}"
)
return response.content
def process(self, query: str) -> str:
"""
多智能体协作处理
参数:
query (str): 查询文本
返回值:
str: 最终回答
"""
# 检索 -> 分析 -> 生成
docs = self.retrieve(query)
analysis = self.analyze(query, docs)
answer = self.generate(query, docs, analysis)
return answer2. 竞争式RAG
多个智能体竞争生成最佳答案:
python
# 导入类型提示和OpenAI模型
from typing import List, Dict
from langchain_openai import ChatOpenAI
class CompetitiveRAG:
"""
竞争式RAG
功能:多个智能体竞争生成最佳答案
"""
def __init__(self, num_agents=3):
"""
初始化竞争式RAG
参数:
num_agents:智能体数量
"""
# 创建多个智能体
self.agents = [ChatOpenAI(model="gpt-4o-mini") for _ in range(num_agents)]
# 评判者:使用GPT-4
self.judge = ChatOpenAI(model="gpt-4")
def generate_candidates(self, query: str, context: str) -> List[str]:
"""
生成候选答案
参数:
query (str): 查询文本
context (str): 上下文
返回值:
List[str]: 候选答案列表
"""
candidates = []
for agent in self.agents:
response = agent.invoke(
f"基于以下上下文回答问题:\n"
f"上下文:{context}\n"
f"问题:{query}"
)
candidates.append(response.content)
return candidates
def judge_answers(self, query: str, candidates: List[str]) -> str:
"""
评判答案
参数:
query (str): 查询文本
candidates (List[str]): 候选答案列表
返回值:
str: 最佳答案
"""
candidates_text = "\n".join([f"答案{i+1}: {c}" for i, c in enumerate(candidates)])
response = self.judge.invoke(
f"请评判以下答案,选择最佳答案:\n"
f"问题:{query}\n"
f"{candidates_text}"
)
return response.content
def process(self, query: str, context: str) -> str:
"""
竞争式处理
参数:
query (str): 查询文本
context (str): 上下文
返回值:
str: 最佳答案
"""
# 生成候选答案
candidates = self.generate_candidates(query, context)
# 评判并选择最佳答案
best_answer = self.judge_answers(query, candidates)
return best_answer评估和优化
1. 自动评估
自动评估RAG系统:
python
from typing import List, Dict
from langchain_openai import ChatOpenAuto评估RAG系统:
```python
# 导入类型提示和OpenAI模型
from typing import List, Dict
from langchain_openai import ChatOpenAI
class RAGEvaluator:
"""
RAG评估器
功能:自动评估RAG系统
"""
def __init__(self):
"""初始化RAG评估器"""
# 使用GPT-4进行评估
self.llm = ChatOpenAI(model="gpt-4")
def evaluate_relevance(self, query: str, context: str) -> float:
"""
评估相关性
参数:
query (str): 查询文本
context (str): 上下文
返回值:
float: 相关性分数(0-1)
"""
response = self.llm.invoke(
f"评估以下上下文与查询的相关性(0-10分):\n"
f"查询:{query}\n"
f"上下文:{context}"
)
# 提取分数
try:
score = float(response.content) / 10
except:
score = 0.5
return score
def evaluate_faithfulness(self, answer: str, context: str) -> float:
"""
评估忠实度
参数:
answer (str): 回答
context (str): 上下文
返回值:
float: 忠实度分数(0-1)
"""
response = self.llm.invoke(
f"评估以下回答是否忠实于上下文(0-10分):\n"
f"回答:{answer}\n"
f"上下文:{context}"
)
try:
score = float(response.content) / 10
except:
score = 0.5
return score
def evaluate_completeness(self, query: str, answer: str) -> float:
"""
评估完整性
参数:
query (str): 查询文本
answer (str): 回答
返回值:
float: 完整性分数(0-1)
"""
response = self.llm.invoke(
f"评估以下回答是否完整地回答了查询(0-10分):\n"
f"查询:{query}\n"
f"回答:{answer}"
)
try:
score = float(response.content) / 10
except:
score = 0.5
return score2. 优化策略
基于评估结果优化RAG系统:
python
# 导入类型提示
from typing import List, Dict
class RAGOptimizer:
"""
RAG优化器
功能:基于评估结果优化RAG系统
"""
def __init__(self, rag_system):
"""
初始化RAG优化器
参数:
rag_system:RAG系统实例
"""
self.rag_system = rag_system
self.evaluator = RAGEvaluator()
def optimize(self, queries: List[str]) -> Dict:
"""
优化RAG系统
参数:
queries (List[str]): 查询列表
返回值:
Dict: 优化结果
"""
results = []
for query in queries:
# 获取当前结果
current_result = self.rag_system.process(query)
# 评估
relevance = self.evaluator.evaluate_relevance(query, current_result["context"])
faithfulness = self.evaluator.evaluate_faithfulness(current_result["answer"], current_result["context"])
completeness = self.evaluator.evaluate_completeness(query, current_result["answer"])
results.append({
"query": query,
"relevance": relevance,
"faithfulness": faithfulness,
"completeness": completeness
})
# 分析结果并提出优化建议
return self.analyze_and_suggest(results)
def analyze_and_suggest(self, results: List[Dict]) -> Dict:
"""
分析结果并提出优化建议
参数:
results (List[Dict]): 评估结果列表
返回值:
Dict: 分析结果和优化建议
"""
# 计算平均分数
avg_relevance = sum(r["relevance"] for r in results) / len(results)
avg_faithfulness = sum(r["faithfulness"] for r in results) / len(results)
avg_completeness = sum(r["completeness"] for r in results) / len(results)
# 生成优化建议
suggestions = []
if avg_relevance < 0.7:
suggestions.append("提高检索相关性:优化嵌入模型或使用重排序")
if avg_faithfulness < 0.7:
suggestions.append("提高忠实度:加强上下文约束或使用引用")
if avg_completeness < 0.7:
suggestions.append("提高完整性:优化提示模板或增加上下文")
return {
"avg_relevance": avg_relevance,
"avg_faithfulness": avg_faithfulness,
"avg_completeness": avg_completeness,
"suggestions": suggestions
}最佳实践
1. 模式选择
- 多模态RAG:处理多种类型的数据
- 自适应RAG:根据查询类型选择策略
- 多步RAG:复杂查询需要多步处理
- 图RAG:处理关系型数据
2. 性能优化
- 缓存机制:缓存常见查询结果
- 异步处理:使用异步提升并发性能
- 批量处理:批量处理多个查询
- 资源管理:合理分配计算资源
3. 质量保证
- 自动评估:使用自动评估监控质量
- 人工审核:关键场景需要人工审核
- 持续优化:基于反馈持续优化
- 版本管理:管理不同版本的RAG系统
常见问题
1. 复杂性问题
- 系统复杂:模块化设计,降低复杂性
- 调试困难:添加详细日志,使用调试工具
- 维护困难:文档完善,代码规范
2. 性能问题
- 响应慢:优化检索和生成,使用缓存
- 资源占用高:优化模型选择,使用量化
- 并发限制:使用异步处理,优化架构
3. 质量问题
- 准确性低:优化检索策略,使用重排序
- 一致性差:加强上下文约束,使用引用
- 可解释性差:添加引用来源,提供推理过程