wehub-skill-sync
795 lines
21 KiB
Markdown
795 lines
21 KiB
Markdown
# 检索优化
|
||
|
||
---
|
||
|
||
## 优化技术概览
|
||
|
||
| 技术 | 影响 | 复杂度 | 使用时机 |
|
||
|-----------|--------|------------|-------------|
|
||
| **混合搜索** | 高 | 中 | 生产环境始终启用 |
|
||
| **重排序** | 高 | 低 | Top-k 精排 |
|
||
| **查询扩展** | 中 | 中 | 模糊查询 |
|
||
| **HyDE** | 中-高 | 中 | 概念密集型检索 |
|
||
| **元数据过滤** | 高 | 低 | 多租户、分类场景 |
|
||
| **查询分解** | 中 | 高 | 复杂问题 |
|
||
| **上下文压缩** | 中 | 中 | 长检索片段 |
|
||
|
||
---
|
||
|
||
## 混合搜索（向量 + 关键词）
|
||
|
||
### 倒数排序融合（RRF）
|
||
|
||
```python
|
||
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 实现）
|
||
|
||
```python
|
||
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 稀疏-稠密混合
|
||
|
||
```python
|
||
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
|
||
|
||
```python
|
||
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 重排序（开源方案）
|
||
|
||
```python
|
||
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 风格延迟交互
|
||
|
||
```python
|
||
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 的查询扩展
|
||
|
||
```python
|
||
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)
|
||
```
|
||
|
||
### 查询改写
|
||
|
||
```python
|
||
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（假设性文档嵌入）
|
||
|
||
```python
|
||
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（多视角生成）
|
||
|
||
```python
|
||
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)
|
||
```
|
||
|
||
---
|
||
|
||
## 元数据过滤
|
||
|
||
### 多租户过滤
|
||
|
||
```python
|
||
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}
|
||
}
|
||
)
|
||
```
|
||
|
||
### 时间范围过滤
|
||
|
||
```python
|
||
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]
|
||
```
|
||
|
||
---
|
||
|
||
## 查询分解
|
||
|
||
```python
|
||
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)
|
||
```
|
||
|
||
---
|
||
|
||
## 上下文压缩
|
||
|
||
```python
|
||
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 的抽取式压缩
|
||
|
||
```python
|
||
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])
|
||
```
|
||
|
||
---
|
||
|
||
## 完整优化流程
|
||
|
||
```python
|
||
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 管道** — 重排序模型的服务部署
|