hybrid-retrieval

RAG 混合检索

结合向量相似度与关键词匹配，实现更精准的检索。

为什么混合检索在 2026 年不可或缺

仅使用向量搜索会漏掉：

• 精确匹配（产品代码、ID、名称）
• 在嵌入中表示不足的罕见词
• 特定关键词查询（“错误代码 E-5012”）

仅使用关键词搜索会漏掉：

• 语义相似性（“汽车”与“机动车”）
• 上下文和含义
• 改写后的内容

混合搜索结合二者，可将召回率提高 15-25%。

核心模式

模式 1：倒数排名融合（RRF）

这是结合多个检索器排名结果的标准方法：

def reciprocal_rank_fusion(
    results_lists: list[list[dict]],
    k: int = 60
) -> list[dict]:
    """
    Combine multiple ranked result lists using RRF.

    Args:
        results_lists: List of ranked results from different retrievers
        k: Ranking constant (default 60, higher = more weight to lower ranks)

    Returns:
        Fused and re-ranked results
    """
    fused_scores = {}

    for results in results_lists:
        for rank, doc in enumerate(results):
            doc_id = doc["id"]
            if doc_id not in fused_scores:
                fused_scores[doc_id] = {"doc": doc, "score": 0}
            # RRF formula: 1 / (k + rank)
            fused_scores[doc_id]["score"] += 1 / (k + rank + 1)

    # Sort by fused score
    sorted_results = sorted(
        fused_scores.values(),
        key=lambda x: x["score"],
        reverse=True
    )
    return [item["doc"] for item in sorted_results]

模式 2：LangChain Ensemble Retriever

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Chroma

# Create vector retriever
vectorstore = Chroma.from_documents(documents, embeddings)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

# Create BM25 retriever
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 10

# Combine with weights
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # Tune based on your data
)

# Use in RAG chain
results = ensemble_retriever.invoke("your query here")

模式 3：LlamaIndex 混合搜索

from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.retrievers import QueryFusionRetriever
from llama_index.retrievers.bm25 import BM25Retriever

# Build index
index = VectorStoreIndex.from_documents(documents)

# Create retrievers
vector_retriever = index.as_retriever(similarity_top_k=10)
bm25_retriever = BM25Retriever.from_defaults(
    nodes=index.docstore.docs.values(),
    similarity_top_k=10
)

# Fusion retriever with query expansion
retriever = QueryFusionRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    similarity_top_k=10,
    num_queries=4,  # Generate 4 query variations
    mode="reciprocal_rerank",
    use_async=True,
)

模式 4：使用 Qdrant 直接实现

from qdrant_client import QdrantClient
from qdrant_client.models import SparseVector, NamedSparseVector, NamedVector

client = QdrantClient(url="http://localhost:6333")

# Hybrid search with both dense and sparse vectors
results = client.query_points(
    collection_name="documents",
    prefetch=[
        # Dense vector search
        models.Prefetch(
            query=dense_embedding,  # [0.1, 0.2, ...]
            using="dense",
            limit=20
        ),
        # Sparse vector search (BM25-style)
        models.Prefetch(
            query=SparseVector(
                indices=[1, 42, 123],  # Token IDs
                values=[0.5, 0.8, 0.3]  # Token weights
            ),
            using="sparse",
            limit=20
        ),
    ],
    query=models.FusionQuery(fusion=models.Fusion.RRF),
    limit=10
)

通过重排序提高精确率

混合检索后，对结果重新排序以确定最终顺序：

交叉编码器重排序

from sentence_transformers import CrossEncoder

# Load reranker model
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank_results(query: str, documents: list[str], top_k: int = 5):
    """Rerank documents using cross-encoder."""
    # Create query-document pairs
    pairs = [[query, doc] for doc in documents]

    # Score all pairs
    scores = reranker.predict(pairs)

    # Sort by score
    scored_docs = list(zip(documents, scores))
    scored_docs.sort(key=lambda x: x[1], reverse=True)

    return [doc for doc, score in scored_docs[:top_k]]

Cohere Rerank API

import cohere

co = cohere.Client("your-api-key")

def cohere_rerank(query: str, documents: list[str], top_k: int = 5):
    """Rerank using Cohere's rerank endpoint."""
    response = co.rerank(
        model="rerank-english-v3.0",
        query=query,
        documents=documents,
        top_n=top_k,
        return_documents=True
    )
    return [result.document.text for result in response.results]

权重调优指南

数据类型	向量权重	关键词权重
技术文档	0.5	0.5
法律/合规	0.4	0.6
创意内容	0.7	0.3
产品目录	0.3	0.7
代码仓库	0.4	0.6

最佳实践

1. 始终进行基准测试 - 使用你的数据测试仅向量、仅关键词和混合检索
2. 根据实测调优权重 - 从 0.5/0.5 开始，根据评估结果进行调整
3. 使用重排序 - 混合检索 + 重排序 = 最佳结果
4. 考虑查询类型 - 将精确匹配查询路由到关键词搜索，将语义查询路由到向量搜索
5. 监控延迟 - 混合检索会增加开销；尽可能使用缓存

快速决策树

Is the query an exact match (ID, code, name)?
├─ Yes → Keyword-heavy (0.3 vector / 0.7 keyword)
└─ No → Is it conceptual/semantic?
         ├─ Yes → Vector-heavy (0.7 vector / 0.3 keyword)
         └─ Mixed → Balanced (0.5 / 0.5) + reranking