wehub-skill-sync
21 KiB
21 KiB
检索优化
优化技术概览
| 技术 | 影响 | 复杂度 | 使用时机 |
|---|---|---|---|
| 混合搜索 | 高 | 中 | 生产环境始终启用 |
| 重排序 | 高 | 低 | Top-k 精排 |
| 查询扩展 | 中 | 中 | 模糊查询 |
| HyDE | 中-高 | 中 | 概念密集型检索 |
| 元数据过滤 | 高 | 低 | 多租户、分类场景 |
| 查询分解 | 中 | 高 | 复杂问题 |
| 上下文压缩 | 中 | 中 | 长检索片段 |
混合搜索（向量 + 关键词）
倒数排序融合（RRF）
from dataclasses import dataclass
from typing import Callable
@dataclass
class SearchResult:
id: str
text: str
score: float
source: str # "vector" or "keyword"
def reciprocal_rank_fusion(
vector_results: list[SearchResult],
keyword_results: list[SearchResult],
k: int = 60,
vector_weight: float = 0.5
) -> list[SearchResult]:
"""
使用 RRF 融合向量和关键词结果。
k 是用于降低高排名影响的常数（通常为 60）。
"""
scores: dict[str, float] = {}
docs: dict[str, SearchResult] = {}
# 对向量结果打分
for rank, result in enumerate(vector_results, 1):
rrf_score = vector_weight * (1 / (k + rank))
scores[result.id] = scores.get(result.id, 0) + rrf_score
docs[result.id] = result
# 对关键词结果打分
keyword_weight = 1 - vector_weight
for rank, result in enumerate(keyword_results, 1):
rrf_score = keyword_weight * (1 / (k + rank))
scores[result.id] = scores.get(result.id, 0) + rrf_score
if result.id not in docs:
docs[result.id] = result
# 按融合得分排序
sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
return [
SearchResult(
id=doc_id,
text=docs[doc_id].text,
score=scores[doc_id],
source="hybrid"
)
for doc_id in sorted_ids
]
# 使用示例
hybrid_results = reciprocal_rank_fusion(
vector_results=vector_search(query_embedding, top_k=20),
keyword_results=bm25_search(query_text, top_k=20),
vector_weight=0.6 # 偏向语义相似度
)
BM25 + 向量（Weaviate 实现）
from weaviate.classes.query import HybridFusion
collection = client.collections.get("Documents")
# 可配置融合方式的混合搜索
results = collection.query.hybrid(
query="how to configure authentication",
alpha=0.5, # 0 = 纯 BM25，1 = 纯向量
fusion_type=HybridFusion.RELATIVE_SCORE, # 或 RANKED
limit=10,
return_metadata=["score", "explain_score"]
)
# 遍历结果
for obj in results.objects:
print(f"Score: {obj.metadata.score}")
print(f"Explanation: {obj.metadata.explain_score}")
print(f"Text: {obj.properties['content'][:200]}")
Pinecone 稀疏-稠密混合
from pinecone_text.sparse import BM25Encoder
# 在语料上训练 BM25 编码器
bm25 = BM25Encoder()
bm25.fit(corpus_documents)
# 为混合搜索编码查询
sparse_vector = bm25.encode_queries(query_text)
dense_vector = get_embedding(query_text)
# 使用两个向量进行搜索
results = index.query(
vector=dense_vector,
sparse_vector=sparse_vector,
top_k=10,
include_metadata=True
)
重排序
Cohere Rerank
import cohere
co = cohere.Client(api_key="your-api-key")
def rerank_results(
query: str,
documents: list[str],
top_n: int = 5,
model: str = "rerank-english-v3.0"
) -> list[dict]:
"""使用 Cohere 对文档进行重排序。"""
response = co.rerank(
query=query,
documents=documents,
top_n=top_n,
model=model,
return_documents=True
)
return [
{
"text": result.document.text,
"relevance_score": result.relevance_score,
"original_index": result.index
}
for result in response.results
]
# 流程：先多检索，再少重排
initial_results = vector_search(query_embedding, top_k=50)
documents = [r.text for r in initial_results]
reranked = rerank_results(
query="how to configure OAuth2 authentication",
documents=documents,
top_n=5
)
# 使用重排后的前 5 条文档作为 LLM 上下文
context = "\n\n".join([r["text"] for r in reranked])
Cross-Encoder 重排序（开源方案）
from sentence_transformers import CrossEncoder
class Reranker:
"""使用 cross-encoder 模型进行重排序。"""
def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
self.model = CrossEncoder(model_name)
def rerank(
self,
query: str,
documents: list[str],
top_k: int = 5
) -> list[tuple[str, float]]:
"""根据与查询的相关性对文档进行重排序。"""
# 创建查询-文档对
pairs = [[query, doc] for doc in documents]
# 获取相关性得分
scores = self.model.predict(pairs)
# 按得分排序
doc_scores = list(zip(documents, scores))
doc_scores.sort(key=lambda x: x[1], reverse=True)
return doc_scores[:top_k]
# 使用示例
reranker = Reranker()
top_docs = reranker.rerank(
query="OAuth2 setup guide",
documents=retrieved_documents,
top_k=5
)
ColBERT 风格延迟交互
from colbert import Searcher
from colbert.infra import Run, RunConfig
# 设置 ColBERT 索引（一次性）
with Run().context(RunConfig(nranks=1)):
searcher = Searcher(index="path/to/colbert_index")
# 使用延迟交互评分进行搜索
results = searcher.search(
query="how to configure authentication",
k=10
)
# 结果包含 token 级别的匹配得分
for passage_id, rank, score in zip(*results):
print(f"Rank {rank}: Doc {passage_id}, Score: {score}")
查询扩展
基于 LLM 的查询扩展
from openai import OpenAI
client = OpenAI()
def expand_query(query: str, num_expansions: int = 3) -> list[str]:
"""使用 LLM 生成查询变体。"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": f"""生成 {num_expansions} 个替代搜索查询，
这些查询应有助于找到用户问题的相关文档。
包括：
- 同义词变体
- 更具体的版本
- 更通用的版本
以 JSON 字符串数组格式返回。"""
},
{
"role": "user",
"content": query
}
],
response_format={"type": "json_object"}
)
import json
result = json.loads(response.choices[0].message.content)
return [query] + result.get("queries", [])
# 使用示例
original_query = "how to fix memory leak"
expanded_queries = expand_query(original_query)
# ["how to fix memory leak", "debug memory issues", "memory leak detection",
# "troubleshoot high memory usage"]
# 使用所有查询进行搜索并合并结果
all_results = []
for q in expanded_queries:
results = vector_search(get_embedding(q), top_k=10)
all_results.extend(results)
# 去重并按出现频率排序
deduped = deduplicate_by_id(all_results)
查询改写
def rewrite_query_for_retrieval(
conversational_query: str,
chat_history: list[dict]
) -> str:
"""将会话查询改写为独立的搜索查询。"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": """将用户的问题改写为独立的搜索查询。
包含聊天历史中的相关上下文。
只输出改写后的查询，不附带其他内容。"""
},
{
"role": "user",
"content": f"""聊天历史：
{format_chat_history(chat_history)}
用户的问题：{conversational_query}
改写后的搜索查询："""
}
],
max_tokens=100
)
return response.choices[0].message.content.strip()
# 示例
history = [
{"role": "user", "content": "Tell me about Python web frameworks"},
{"role": "assistant", "content": "Popular Python web frameworks include Django, Flask, and FastAPI..."}
]
query = "Which one is best for APIs?"
rewritten = rewrite_query_for_retrieval(query, history)
# 输出："Best Python web framework for building REST APIs: Django vs Flask vs FastAPI"
HyDE（假设性文档嵌入）
def hyde_search(
query: str,
vector_store,
embedding_model,
top_k: int = 10
) -> list[SearchResult]:
"""
生成假设性答案，对其进行嵌入，然后进行搜索。
使查询嵌入空间与文档嵌入空间对齐。
"""
# 生成假设性文档
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": """撰写一段能够回答用户问题的短文。
以专业文档作者的身份来写。
内容需具体且偏技术性，约 100-200 词。"""
},
{
"role": "user",
"content": query
}
],
max_tokens=300
)
hypothetical_doc = response.choices[0].message.content
# 嵌入假设性文档
hyde_embedding = embedding_model.encode(hypothetical_doc)
# 使用假设性文档的嵌入进行搜索
results = vector_store.search(
vector=hyde_embedding,
top_k=top_k
)
return results
# 使用示例
results = hyde_search(
query="How do I handle rate limiting in my API?",
vector_store=qdrant_client,
embedding_model=sentence_transformer
)
Multi-HyDE（多视角生成）
def multi_hyde_search(
query: str,
vector_store,
embedding_model,
num_hypotheticals: int = 3,
top_k: int = 10
) -> list[SearchResult]:
"""生成多个假设性文档，实现多样化检索。"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": f"""生成 {num_hypotheticals} 篇不同的段落，
从不同角度回答该问题：
1. 技术深度剖析
2. 初学者友好解释
3. 最佳实践总结
以 JSON 格式返回，包含 "passages" 数组。"""
},
{
"role": "user",
"content": query
}
],
response_format={"type": "json_object"}
)
import json
passages = json.loads(response.choices[0].message.content)["passages"]
# 嵌入所有假设性文档
all_results = []
for passage in passages:
embedding = embedding_model.encode(passage)
results = vector_store.search(vector=embedding, top_k=top_k)
all_results.extend(results)
# 去重并合并得分
return deduplicate_and_merge(all_results)
元数据过滤
多租户过滤
class MultiTenantRetriever:
"""带强制租户隔离的检索器。"""
def __init__(self, vector_store):
self.vector_store = vector_store
def search(
self,
query_embedding: list[float],
tenant_id: str,
top_k: int = 10,
additional_filters: dict | None = None
) -> list[SearchResult]:
"""使用强制租户过滤器进行搜索。"""
# 构建过滤器 —— 租户始终为必填项
filters = {"tenant_id": {"$eq": tenant_id}}
if additional_filters:
filters = {"$and": [filters, additional_filters]}
return self.vector_store.search(
vector=query_embedding,
filter=filters,
top_k=top_k
)
# 使用示例
retriever = MultiTenantRetriever(pinecone_index)
results = retriever.search(
query_embedding=embedding,
tenant_id="acme-corp",
additional_filters={
"doc_type": {"$in": ["manual", "faq"]},
"published": {"$eq": True}
}
)
时间范围过滤
from datetime import datetime, timedelta
def search_recent_documents(
query_embedding: list[float],
vector_store,
days_back: int = 30,
top_k: int = 10
) -> list[SearchResult]:
"""搜索指定时间窗口内更新过的文档。"""
cutoff_date = datetime.utcnow() - timedelta(days=days_back)
return vector_store.search(
vector=query_embedding,
filter={
"updated_at": {"$gte": cutoff_date.isoformat()}
},
top_k=top_k
)
def search_with_recency_boost(
query_embedding: list[float],
vector_store,
recency_weight: float = 0.2,
top_k: int = 10
) -> list[SearchResult]:
"""在排序中提升近期文档的权重。"""
# 获取更多结果以便进行后置过滤
results = vector_store.search(
vector=query_embedding,
top_k=top_k * 3
)
now = datetime.utcnow()
def compute_boosted_score(result):
doc_date = datetime.fromisoformat(result.metadata["updated_at"])
days_old = (now - doc_date).days
recency_score = max(0, 1 - (days_old / 365)) # 一年内衰减
return result.score * (1 - recency_weight) + recency_score * recency_weight
# 使用时效性加分进行重排序
for result in results:
result.boosted_score = compute_boosted_score(result)
results.sort(key=lambda x: x.boosted_score, reverse=True)
return results[:top_k]
查询分解
def decompose_complex_query(query: str) -> list[str]:
"""将复杂查询分解为多个子问题。"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": """将这个复杂问题拆分为更简单的子问题，
每个子问题应能独立回答。每个子问题应具备可搜索性。
以 JSON 格式返回，包含 "questions" 数组。"""
},
{
"role": "user",
"content": query
}
],
response_format={"type": "json_object"}
)
import json
result = json.loads(response.choices[0].message.content)
return result.get("questions", [query])
def search_with_decomposition(
complex_query: str,
vector_store,
embedding_model,
top_k_per_subquery: int = 5
) -> dict:
"""对每个子问题进行搜索并汇总结果。"""
sub_questions = decompose_complex_query(complex_query)
aggregated_results = {
"sub_questions": [],
"all_documents": []
}
seen_doc_ids = set()
for sub_q in sub_questions:
embedding = embedding_model.encode(sub_q)
results = vector_store.search(vector=embedding, top_k=top_k_per_subquery)
sub_q_results = []
for r in results:
if r.id not in seen_doc_ids:
seen_doc_ids.add(r.id)
sub_q_results.append(r)
aggregated_results["all_documents"].append(r)
aggregated_results["sub_questions"].append({
"question": sub_q,
"results": sub_q_results
})
return aggregated_results
# 使用示例
complex_q = "Compare the security features of OAuth2 and API keys, and explain when to use each"
results = search_with_decomposition(complex_q, vector_store, embedding_model)
上下文压缩
def compress_retrieved_context(
query: str,
documents: list[str],
max_tokens: int = 2000
) -> str:
"""仅从文档中提取与查询相关的部分。"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": f"""仅提取这些文档中与回答用户问题相关的部分。
删除无关信息。
将提取内容控制在 {max_tokens} 个 token 以内。
保留来源归属信息。"""
},
{
"role": "user",
"content": f"""问题：{query}
文档：
{chr(10).join([f'[Doc {i+1}]: {doc}' for i, doc in enumerate(documents)])}
提取的相关内容："""
}
],
max_tokens=max_tokens
)
return response.choices[0].message.content
基于 Cross-Encoder 的抽取式压缩
from sentence_transformers import CrossEncoder
def extractive_compress(
query: str,
document: str,
cross_encoder: CrossEncoder,
top_k_sentences: int = 5
) -> str:
"""从文档中提取最相关的句子。"""
import re
sentences = re.split(r'(?<=[.!?])\s+', document)
if len(sentences) <= top_k_sentences:
return document
# 对每个句子打分
pairs = [[query, sent] for sent in sentences]
scores = cross_encoder.predict(pairs)
# 按原始顺序获取 top 句子
scored_sentences = list(zip(range(len(sentences)), sentences, scores))
top_sentences = sorted(scored_sentences, key=lambda x: x[2], reverse=True)[:top_k_sentences]
top_sentences = sorted(top_sentences, key=lambda x: x[0]) # 恢复原始顺序
return " ".join([s[1] for s in top_sentences])
完整优化流程
class OptimizedRetriever:
"""包含所有优化策略的生产级检索流程。"""
def __init__(
self,
vector_store,
embedding_model,
reranker,
bm25_index
):
self.vector_store = vector_store
self.embedding_model = embedding_model
self.reranker = reranker
self.bm25_index = bm25_index
async def retrieve(
self,
query: str,
tenant_id: str,
top_k: int = 5,
use_hyde: bool = False,
use_query_expansion: bool = True
) -> list[dict]:
"""完整的优化检索流程。"""
# 第 1 步：查询预处理
processed_query = self._preprocess_query(query)
# 第 2 步：可选的 HyDE
if use_hyde:
query_embedding = await self._hyde_embed(processed_query)
else:
query_embedding = self.embedding_model.encode(processed_query)
# 第 3 步：混合搜索（向量 + BM25）
vector_results = self.vector_store.search(
vector=query_embedding,
filter={"tenant_id": tenant_id},
top_k=50
)
bm25_results = self.bm25_index.search(processed_query, top_k=50)
# 第 4 步：使用 RRF 合并
merged = reciprocal_rank_fusion(
vector_results,
bm25_results,
vector_weight=0.6
)[:30]
# 第 5 步：可选的查询扩展
if use_query_expansion:
expanded_queries = await self._expand_query(processed_query)
for exp_query in expanded_queries[1:]: # 跳过原始查询
exp_embedding = self.embedding_model.encode(exp_query)
exp_results = self.vector_store.search(
vector=exp_embedding,
filter={"tenant_id": tenant_id},
top_k=10
)
merged.extend(exp_results)
merged = deduplicate_by_id(merged)[:30]
# 第 6 步：重排序
documents = [r.text for r in merged]
reranked = self.reranker.rerank(
query=processed_query,
documents=documents,
top_k=top_k
)
return [
{
"text": doc,
"score": score,
"metadata": merged[i].metadata
}
for i, (doc, score) in enumerate(reranked)
]
def _preprocess_query(self, query: str) -> str:
"""清洗和标准化查询。"""
import re
query = re.sub(r'\s+', ' ', query).strip()
return query
async def _hyde_embed(self, query: str) -> list[float]:
"""生成假设性文档并进行嵌入。"""
# 实现参考 HyDE 章节
pass
async def _expand_query(self, query: str) -> list[str]:
"""用变体扩展查询。"""
# 实现参考查询扩展章节
pass
性能基准
| 技术 | 延迟影响 | 质量影响 | 成本影响 |
|---|---|---|---|
| 纯向量 | 基准 | 基准 | 基准 |
| + BM25 混合 | +10-20ms | +5-15% 精确率 | 极小 |
| + 重排序 | +50-100ms | +10-20% 精确率 | +$0.001/查询 |
| + 查询扩展 | +100-200ms | +5-10% 召回率 | +$0.002/查询 |
| + HyDE | +200-500ms | +10-25% 精确率 | +$0.003/查询 |
快速参考
| 目标 | 技术 | 实现方式 |
|---|---|---|
| 提升精确率 | 重排序 | Cross-encoder 或 Cohere |
| 提升召回率 | 查询扩展 | LLM 生成的变体 |
| 处理同义词 | 混合搜索 | BM25 + 向量 + RRF |
| 概念搜索 | HyDE | 假设性文档嵌入 |
| 多租户 | 元数据过滤 | 强制 tenant_id |
| 最新内容 | 时间范围过滤 | 日期范围查询 |
| 复杂问题 | 分解 | 子问题检索 |
相关技能
- RAG 架构师 — 系统设计与架构
- NLP 工程师 — 查询理解
- Python 高手 — 异步实现
- ML 管道 — 重排序模型的服务部署