# 检索优化

---

## 优化技术概览

| 技术 | 影响 | 复杂度 | 使用时机 |
|-----------|--------|------------|-------------|
| **混合搜索** | 高 | 中 | 生产环境始终启用 |
| **重排序** | 高 | 低 | 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 管道** — 重排序模型的服务部署