在前一篇文章中，我们介绍了RAG技术的基本原理和架构设计。在实际应用中，构建一个高性能的RAG系统需要解决许多挑战，包括检索质量、上下文管理、生成质量等方面的问题。本文将深入探讨RAG系统的高级优化策略，帮助你构建更加高效、准确和用户友好的RAG应用。
密集检索增强技术：
class EnhancedDenseRetriever:
def __init__(self, vector_db, query_expansion=True, cross_encoder_rerank=True):
self.vector_db = vector_db
self.query_expansion = query_expansion
self.cross_encoder_rerank = cross_encoder_rerank
if cross_encoder_rerank:
from sentence_transformers import CrossEncoder
self.cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
def expand_query(self, query, num_variations=3):
"""使用LLM进行查询扩展""" import openai
prompt = f""" 将以下查询扩展为{num_variations}个不同的表述，保持核心查询意图不变。 每个变体应从不同角度或使用不同的关键词来表达相同的意思。 查询: {query} 变体列表: """ response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
temperature=0.5 )
expanded_queries = response.choices[0].message.content.strip().split('\n')
# 清理和过滤结果 expanded_queries = [q.strip() for q in expanded_queries if q.strip()]
# 确保不包含原始查询本身 expanded_queries = [q for q in expanded_queries if q.lower() != query.lower()[:len(q)]]
return expanded_queries[:num_variations]
def retrieve_with_expansion(self, query, top_k=5):
"""使用查询扩展进行检索""" # 获取扩展查询 expanded_queries = self.expand_query(query)
all_queries = [query] + expanded_queries
# 对每个查询进行检索 all_results = {}
for q_idx, q in enumerate(all_queries):
results = self.vector_db.query(q, n_results=top_k)
# 合并结果，为来自不同查询的相同文档分配权重 for doc_id, doc, metadata, distance in zip(
results["ids"], results["documents"],
results["metadatas"], results["distances"]
):
relevance_score = 1 - distance
# 原始查询权重更高 if q_idx == 0:
relevance_score *= 1.5 if doc_id not in all_results:
all_results[doc_id] = {
"id": doc_id,
"content": doc,
"metadata": metadata,
"scores": [],
"final_score": 0 }
all_results[doc_id]["scores"].append(relevance_score)
# 计算最终分数（例如取平均值或最大值） for doc_id in all_results:
# 可以使用不同的策略，这里使用最大值 all_results[doc_id]["final_score"] = max(all_results[doc_id]["scores"])
# 按最终分数排序 sorted_results = sorted(
all_results.values(),
key=lambda x: x["final_score"],
reverse=True )
return sorted_results[:top_k]
def rerank_with_cross_encoder(self, query, retrieved_docs, top_k=3):
"""使用交叉编码器重新排序检索结果""" # 准备用于交叉编码器的文本对 pairs = [[query, doc["content"]] for doc in retrieved_docs]
# 计算相关性分数 scores = self.cross_encoder.predict(pairs)
# 更新文档分数 for i, doc in enumerate(retrieved_docs):
doc["cross_score"] = float(scores[i])
# 按交叉编码器分数排序 reranked_docs = sorted(
retrieved_docs,
key=lambda x: x["cross_score"],
reverse=True )
return reranked_docs[:top_k]
def retrieve(self, query, top_k=5):
"""综合检索方法""" if self.query_expansion:
# 使用查询扩展检索 retrieved_docs = self.retrieve_with_expansion(query, top_k=top_k*2) # 多获取一些结果用于重排序 else:
# 基础检索 results = self.vector_db.query(query, n_results=top_k*2)
retrieved_docs = [
{"id": doc_id, "content": doc, "metadata": metadata, "score": 1 - distance}
for doc_id, doc, metadata, distance in zip(
results["ids"], results["documents"],
results["metadatas"], results["distances"]
)
]
# 如果启用了交叉编码器重排序 if self.cross_encoder_rerank and retrieved_docs:
retrieved_docs = self.rerank_with_cross_encoder(query, retrieved_docs, top_k=top_k)
else:
retrieved_docs = retrieved_docs[:top_k]
return retrieved_docs
class AdvancedHybridRetriever:
def __init__(self, vector_db, keyword_engine, fusion_method="rrf"):
self.vector_db = vector_db
self.keyword_engine = keyword_engine # 可以是Elasticsearch或其他关键词搜索引擎 self.fusion_method = fusion_method # rrf: Reciprocal Rank Fusion, linear: 线性加权 def vector_search(self, query, top_k=10):
"""向量检索""" results = self.vector_db.query(query, n_results=top_k)
vector_docs = []
for doc_id, doc, metadata, distance in zip(
results["ids"], results["documents"],
results["metadatas"], results["distances"]
):
vector_docs.append({
"id": doc_id,
"content": doc,
"metadata": metadata,
"vector_score": 1 - distance,
"vector_rank": len(vector_docs) + 1 })
return vector_docs
def keyword_search(self, query, top_k=10):
"""关键词检索""" # 这里是示例，实际实现需要根据具体的关键词引擎调整 # 假设keyword_engine提供了search方法 results = self.keyword_engine.search(query, size=top_k)
keyword_docs = []
for i, result in enumerate(results):
keyword_docs.append({
"id": result["id"],
"content": result["content"],
"metadata": result.get("metadata", {}),
"keyword_score": result["score"],
"keyword_rank": i + 1 })
return keyword_docs
def reciprocal_rank_fusion(self, vector_results, keyword_results, top_k=5):
"""使用倒数排序融合(RRF)合并结果""" # RRF公式: score = sum(1/(k + rank)) 其中k是常数，通常为60 k = 60 fused_scores = {}
# 处理向量检索结果 for doc in vector_results:
doc_id = doc["id"]
if doc_id not in fused_scores:
fused_scores[doc_id] = {"doc": doc, "score": 0}
fused_scores[doc_id]["score"] += 1 / (k + doc["vector_rank"])
# 处理关键词检索结果 for doc in keyword_results:
doc_id = doc["id"]
if doc_id not in fused_scores:
# 如果文档只在关键词结果中，需要合并内容和元数据 fused_scores[doc_id] = {"doc": doc, "score": 0}
else:
# 如果已经存在，合并信息 fused_scores[doc_id]["doc"]["keyword_score"] = doc.get("keyword_score", 0)
fused_scores[doc_id]["doc"]["keyword_rank"] = doc.get("keyword_rank", 0)
fused_scores[doc_id]["score"] += 1 / (k + doc["keyword_rank"])
# 按融合分数排序 sorted_results = sorted(
fused_scores.values(),
key=lambda x: x["score"],
reverse=True )
# 返回前top_k个结果 return [item["doc"] for item in sorted_results[:top_k]]
def linear_fusion(self, vector_results, keyword_results, weights=(0.6, 0.4), top_k=5):
"""线性加权融合结果""" vector_weight, keyword_weight = weights
fused_scores = {}
# 归一化向量分数 max_vector_score = max([doc["vector_score"] for doc in vector_results], default=1)
# 归一化关键词分数（如果有） max_keyword_score = 1 if keyword_results:
max_keyword_score = max([doc["keyword_score"] for doc in keyword_results], default=1)
# 处理向量检索结果 for doc in vector_results:
doc_id = doc["id"]
normalized_vector_score = doc["vector_score"] / max_vector_score
if doc_id not in fused_scores:
fused_scores[doc_id] = {
"doc": doc,
"vector_score": normalized_vector_score,
"keyword_score": 0,
"final_score": 0 }
else:
fused_scores[doc_id]["vector_score"] = normalized_vector_score
# 处理关键词检索结果 for doc in keyword_results:
doc_id = doc["id"]
normalized_keyword_score = doc["keyword_score"] / max_keyword_score
if doc_id not in fused_scores:
# 如果文档只在关键词结果中，需要初始化 fused_scores[doc_id] = {
"doc": doc,
"vector_score": 0,
"keyword_score": normalized_keyword_score,
"final_score": 0 }
else:
# 如果已经存在，更新关键词分数 fused_scores[doc_id]["keyword_score"] = normalized_keyword_score
# 合并文档信息 fused_scores[doc_id]["doc"]["keyword_score"] = doc["keyword_score"]
# 计算最终分数 for doc_id in fused_scores:
fs = fused_scores[doc_id]
fs["final_score"] = (
fs["vector_score"] * vector_weight + fs["keyword_score"] * keyword_weight
)
# 将最终分数添加到文档中 fs["doc"]["final_score"] = fs["final_score"]
# 按最终分数排序 sorted_results = sorted(
fused_scores.values(),
key=lambda x: x["final_score"],
reverse=True )
# 返回前top_k个结果 return [item["doc"] for item in sorted_results[:top_k]]
def retrieve(self, query, top_k=5, vector_weight=0.6, keyword_weight=0.4):
"""执行混合检索""" # 执行向量检索 vector_results = self.vector_search(query, top_k=top_k*3) # 获取更多结果用于融合 # 执行关键词检索 keyword_results = self.keyword_search(query, top_k=top_k*3)
# 融合结果 if self.fusion_method == "rrf":
fused_results = self.reciprocal_rank_fusion(vector_results, keyword_results, top_k=top_k)
else: # linear fused_results = self.linear_fusion(vector_results, keyword_results, (vector_weight, keyword_weight), top_k=top_k)
return fused_results