混合搜索RAG系统
向量与关键词的完美融合

一、为什么需要混合搜索？

构建 RAG 系统时，很多开发者面临一个两难困境：

以一个真实场景为例：用户在企业知识库中搜索 "合同第3.2条的违约金条款"。纯向量搜索可能返回泛泛的"违约相关内容"，而纯关键词搜索可能漏掉表述不同的同义段落。混合搜索则能同时捕捉语义相关性和关键词精确匹配，显著提升召回质量。

二、系统架构与核心原理

混合搜索 RAG 系统由以下核心模块组成：

1. 文档切分器（Chunker）：将原始文档分割为语义完整的片段
2. 双通道索引：同时建立 BM25 倒排索引和向量嵌入索引
3. 并行检索器：同一查询同时触发两套检索系统
4. RRF 融合器：使用递归排名融合算法合并两路结果
5. LLM 生成器：将融合后的上下文送入语言模型生成回答

递归排名融合（RRF）算法原理

RRF 是混合搜索的核心。它不依赖分数绝对值（不同系统分数量纲不统一），而是利用排名位置计算综合得分：

RRF_score(doc) = Σ 1 / (k + rank_i(doc))

其中：
  k     = 60（平滑常数，防止排名靠前的文档过度主导）
  rank_i = 文档在第 i 个检索系统中的排名（从1开始）

示例：
  文档A：向量搜索排名=1，BM25排名=5
  RRF(A) = 1/(60+1) + 1/(60+5) = 0.0164 + 0.0154 = 0.0318

  文档B：向量搜索排名=3，BM25排名=2
  RRF(B) = 1/(60+3) + 1/(60+2) = 0.0159 + 0.0161 = 0.0320

  → 文档B最终排名更靠前（两路均表现良好）

三、环境准备与依赖安装

conda create -n hybrid-rag python=3.11
conda activate hybrid-rag

pip install langchain langchain-openai langchain-community \
    faiss-cpu rank-bm25 sentence-transformers \
    openai python-dotenv tiktoken pypdf

OPENAI_API_KEY=sk-your-api-key-here
# 如使用本地模型（如 Ollama），可跳过上面这行
# OLLAMA_BASE_URL=http://localhost:11434

四、实现 BM25 关键词检索

BM25（Best Matching 25）是信息检索领域最经典的关键词算法，Elasticsearch 默认使用它作为全文搜索引擎。

加载文档并切分

from langchain.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

def load_and_split_documents(file_paths: list[str]) -> list:
    """加载多个文档并切分为语义片段"""
    all_docs = []

    for path in file_paths:
        if path.endswith('.pdf'):
            loader = PyPDFLoader(path)
        else:
            loader = TextLoader(path, encoding='utf-8')
        all_docs.extend(loader.load())

    # 切分策略：chunk_size=512，overlap=64 保证语义连贯
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=512,
        chunk_overlap=64,
        separators=["\n\n", "\n", "。", "！", "？", " ", ""],
    )
    chunks = splitter.split_documents(all_docs)
    print(f"文档切分完成：{len(all_docs)} 个文件 → {len(chunks)} 个片段")
    return chunks

构建 BM25 检索器

from langchain_community.retrievers import BM25Retriever

def build_bm25_retriever(chunks: list, top_k: int = 10) -> BM25Retriever:
    """基于文档片段构建 BM25 关键词检索器"""
    retriever = BM25Retriever.from_documents(
        chunks,
        k=top_k  # 返回 top-k 结果
    )
    return retriever

# 测试 BM25 检索
bm25 = build_bm25_retriever(chunks, top_k=10)
results = bm25.invoke("API v2.3.1 接口说明")
for i, doc in enumerate(results[:3], 1):
    print(f"[BM25 #{i}] {doc.page_content[:100]}...")

五、实现向量语义检索

向量检索通过将文本编码为高维向量，计算语义相似度来召回语义相关的文档。我们使用 FAISS 作为向量存储，支持高效的近似最近邻搜索。

初始化嵌入模型

import os
from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings

load_dotenv()

def get_embeddings(use_local: bool = False):
    """
    use_local=False: 使用 OpenAI text-embedding-3-small（API调用）
    use_local=True:  使用本地 sentence-transformers 模型（免费）
    """
    if use_local:
        # 推荐中文模型：BAAI/bge-m3 或 shibing624/text2vec-base-chinese
        return HuggingFaceEmbeddings(
            model_name="BAAI/bge-m3",
            model_kwargs={"device": "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )
    else:
        return OpenAIEmbeddings(
            model="text-embedding-3-small",
            openai_api_key=os.getenv("OPENAI_API_KEY"),
        )

构建 FAISS 向量索引

from langchain_community.vectorstores import FAISS

def build_vector_store(chunks: list, embeddings, index_path: str = "faiss_index"):
    """构建并持久化 FAISS 向量索引"""
    import os

    if os.path.exists(index_path):
        # 如果索引已存在，直接加载（避免重复计算）
        print(f"加载已有索引：{index_path}")
        vector_store = FAISS.load_local(
            index_path, embeddings,
            allow_dangerous_deserialization=True
        )
    else:
        print(f"构建新索引（共 {len(chunks)} 个片段）...")
        vector_store = FAISS.from_documents(chunks, embeddings)
        vector_store.save_local(index_path)
        print(f"索引已保存至：{index_path}")

    return vector_store

def build_vector_retriever(vector_store, top_k: int = 10):
    """从向量存储创建检索器"""
    return vector_store.as_retriever(
        search_type="similarity",
        search_kwargs={"k": top_k}
    )

# 使用示例
embeddings = get_embeddings(use_local=True)  # 本地免费模型
vector_store = build_vector_store(chunks, embeddings)
vector_retriever = build_vector_retriever(vector_store, top_k=10)

六、递归排名融合（RRF）合并结果

现在我们将两路检索结果通过 RRF 算法融合，实现真正的混合搜索。

实现 RRF 融合函数

from langchain.schema import Document

def reciprocal_rank_fusion(
    results_list: list[list[Document]],
    k: int = 60,
    top_n: int = 5
) -> list[Document]:
    """
    递归排名融合算法

    Args:
        results_list: 多个检索器的结果列表，每个元素是一个 Document 列表
        k: RRF 平滑常数（默认60，平衡精确匹配和语义相关性）
        top_n: 最终返回的文档数量

    Returns:
        按 RRF 分数排序的 Document 列表
    """
    # 用 page_content 作为文档唯一标识
    doc_scores: dict[str, float] = {}
    doc_map: dict[str, Document] = {}

    for results in results_list:
        for rank, doc in enumerate(results, start=1):
            doc_id = doc.page_content
            score = 1.0 / (k + rank)

            if doc_id in doc_scores:
                doc_scores[doc_id] += score
            else:
                doc_scores[doc_id] = score
                doc_map[doc_id] = doc

    # 按 RRF 分数降序排列
    sorted_docs = sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)

    return [doc_map[doc_id] for doc_id, _ in sorted_docs[:top_n]]

混合检索主函数

def hybrid_search(
    query: str,
    bm25_retriever,
    vector_retriever,
    top_k_each: int = 10,
    final_top_n: int = 5
) -> list[Document]:
    """
    执行混合搜索：同时调用 BM25 和向量检索，用 RRF 融合结果
    """
    # 并行获取两路结果
    bm25_results = bm25_retriever.invoke(query)[:top_k_each]
    vector_results = vector_retriever.invoke(query)[:top_k_each]

    print(f"BM25 召回: {len(bm25_results)} 条")
    print(f"向量召回: {len(vector_results)} 条")

    # RRF 融合
    fused = reciprocal_rank_fusion(
        [bm25_results, vector_results],
        k=60,
        top_n=final_top_n
    )
    print(f"RRF 融合后保留: {len(fused)} 条")
    return fused

# 测试混合检索
results = hybrid_search(
    query="如何处理合同违约情况",
    bm25_retriever=bm25,
    vector_retriever=vector_retriever,
    top_k_each=10,
    final_top_n=5
)

for i, doc in enumerate(results, 1):
    print(f"\n[#{i}] {doc.page_content[:150]}...")

七、接入 LLM 完成 RAG 闭环

将混合搜索结果作为上下文，传递给 LLM 生成最终回答，完成完整的 RAG 闭环。

构建完整 RAG 链

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser

# 初始化 LLM
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.1,  # 低温度保证回答稳定
    max_tokens=1024,
)

# RAG Prompt 模板
RAG_PROMPT = ChatPromptTemplate.from_template("""
你是一个专业的知识库助手。请根据以下检索到的上下文内容，准确回答用户问题。

**注意事项：**
- 只基于提供的上下文内容回答
- 如果上下文中没有相关信息，明确告知用户
- 引用具体段落时，使用 [来源N] 标注

---

**检索到的上下文：**
{context}

---

**用户问题：** {question}

**回答：**
""")

def format_context(docs: list[Document]) -> str:
    """将文档列表格式化为 LLM 可读的上下文字符串"""
    formatted = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get('source', '未知来源')
        page = doc.metadata.get('page', '')
        page_info = f"第{page}页" if page else ""
        formatted.append(f"[来源{i}] {source} {page_info}\n{doc.page_content}")
    return "\n\n---\n\n".join(formatted)

def hybrid_rag_answer(question: str) -> str:
    """
    完整的混合搜索 RAG 问答流程
    1. 混合检索相关文档
    2. 格式化上下文
    3. 调用 LLM 生成回答
    """
    # Step 1: 混合检索
    relevant_docs = hybrid_search(
        query=question,
        bm25_retriever=bm25,
        vector_retriever=vector_retriever,
        top_k_each=10,
        final_top_n=5
    )

    # Step 2: 格式化上下文
    context = format_context(relevant_docs)

    # Step 3: LLM 生成
    chain = RAG_PROMPT | llm | StrOutputParser()
    answer = chain.invoke({
        "context": context,
        "question": question
    })
    return answer

# 运行完整 RAG 系统
if __name__ == "__main__":
    # 先加载和准备文档（实际使用中替换为你的文档路径）
    chunks = load_and_split_documents(["./docs/manual.pdf"])

    embeddings = get_embeddings(use_local=True)
    vector_store = build_vector_store(chunks, embeddings)

    bm25 = build_bm25_retriever(chunks, top_k=10)
    vector_retriever = build_vector_retriever(vector_store, top_k=10)

    # 开始问答
    answer = hybrid_rag_answer("合同第3.2条的违约金如何计算？")
    print("\n=== RAG 回答 ===")
    print(answer)

八、性能调优与最佳实践

1. 文档切分策略

切分质量直接影响检索效果，以下是不同场景的推荐参数：

2. 向量索引 HNSW 参数调优

import faiss
import numpy as np

def build_optimized_faiss_index(embeddings_matrix: np.ndarray, dimension: int):
    """
    构建 HNSW 索引（比 Flat L2 快约10倍，精度损失可接受）

    M=32: 每个节点连接数（越大精度越高，内存占用越多）
    efConstruction=200: 构建时搜索范围（越大构建越慢但质量更好）
    """
    index = faiss.IndexHNSWFlat(dimension, 32)
    index.hnsw.efConstruction = 200
    index.hnsw.efSearch = 128  # 查询时的搜索范围
    index.add(embeddings_matrix)
    return index

3. 语义缓存加速重复查询

import hashlib
from functools import lru_cache

# 简单的查询缓存（生产环境建议使用 Redis）
_search_cache: dict[str, list] = {}

def cached_hybrid_search(query: str, **kwargs) -> list:
    """带缓存的混合搜索，相似查询直接返回缓存结果"""
    cache_key = hashlib.md5(query.encode()).hexdigest()

    if cache_key in _search_cache:
        print(f"命中缓存: {query[:30]}...")
        return _search_cache[cache_key]

    results = hybrid_search(query, **kwargs)
    _search_cache[cache_key] = results
    return results

4. 异步并行检索（生产级优化）

import asyncio

async def async_hybrid_search(query: str, bm25_retriever, vector_retriever, top_k=10):
    """异步并行执行两路检索，减少总延迟"""
    # 同时发起两个检索任务
    bm25_task = asyncio.to_thread(bm25_retriever.invoke, query)
    vector_task = asyncio.to_thread(vector_retriever.invoke, query)

    bm25_results, vector_results = await asyncio.gather(bm25_task, vector_task)

    return reciprocal_rank_fusion(
        [bm25_results[:top_k], vector_results[:top_k]],
        final_top_n=5
    )

# 使用方式
# results = await async_hybrid_search(query, bm25, vector_retriever)

常见问题 FAQ