结合pgvector、BM25与交叉编码器重排器的混合RAG检索系统

了解为何纯向量搜索会遗漏部件编号和错误代码，以及如何在 LangChain 中结合 pgvector、BM25 和重排序技术来实现精准的 RAG 检索。

1524 词

基于普通向量搜索构建的检索增强生成（RAG）原型在演示中往往表现出色，但实际使用中却会让用户失望。询问特定泵型的维护计划时，它只会给出通用的泵类保养建议；搜索特定的错误代码时，相关的故障排除步骤也根本找不到。本指南将解释为何密集检索在处理精确标识符时会失败，并展示如何通过混合流程来解决这一问题：利用pgvector进行语义搜索，用BM25进行关键词匹配，再通过重排器决定哪些内容片段能传递给大语言模型。

为何向量搜索无法匹配精确标识符

嵌入模型在处理语义方面表现极为出色。当查询“automobile”时，相关结果会出现在关于“cars”和“vehicles”的文档附近，因为嵌入模型会将相关的概念在高维空间中放置得彼此靠近。但这一特性也正是它的弱点——相似度衡量的是语义上的接近程度，而非完全相同的字符序列。

当用户搜索如TX-99402这样的零件编号或E-404这样的错误代码时，该字符串的嵌入向量可能会与TX-99401或通用的故障排除文本非常接近。检索系统返回的往往是“内容相似”的文档，而非包含用户输入的确切字符串的文档。对于技术手册、产品目录和支持知识库而言，由于标识符承载着大部分信息，这种错误模式尤为常见。

混合检索：密集检索与稀疏检索并行

解决方法是运行两个互补的检索器，并将它们的结果结合起来：

密集检索（向量搜索）能够捕捉上下文与含义。无需单独搭建向量数据库，只需使用pgvector扩展在PostgreSQL中存储嵌入向量即可。许多应用已基于Postgres运行，因此添加向量列既能保持技术栈简洁，又能利用现有的备份、访问控制及事务功能。
稀疏检索（关键词搜索）能够实现精确匹配，适用于处理缩写词和行业术语。标准算法为BM25，这是一种成熟的排序函数，它根据查询词在文档中出现的频率进行评分，同时考虑这些词在整个语料库中的稀有程度，并对文档长度进行标准化处理。

一个有用的思维模型：向量搜索用于找到合适的“社区”，而BM25则能确定具体的“门牌号”。你需要从这两种方法中选取最佳结果并合并它们。关于每种方法的优缺点，可阅读我们关于技术知识混合搜索的文章。

为何合并后的结果需要重新排序

混合搜索立刻带来了一个问题：你现在有了两个评分标准不具可比性的排序列表。BM25的评分取决于词频且没有上限，而向量相似度则是基于完全不同的尺度通过余弦距离计算得出的。0.82的语义评分并不比14.5的BM25评分更高或更低；仅按原始评分对合并后的结果进行排序是毫无意义的。

重排序模型通过忽略原始得分来解决这一问题。它是一个独立的模型，通常是交叉编码器，能够同时读取查询语句和候选文档，并为该对内容输出一个单一的相关性得分。由于它能一次性查看两段文本，因此相比比较两个独立计算出的嵌入向量，能够更精确地判断相关性。

处理流程如下：

从pgvector和BM25两种检索器中各获取10个候选文档。
将它们合并，最多得到20个文本块。
使用重排序模型为每个文本块计算与查询的相关性得分。
保留得分最高的3个文本块，仅将它们传递给大语言模型。

更少但质量更高的文本块意味着提示词更短，模型需要忽略的噪声更少，同时token成本也会降低。

延迟成本

交叉编码器成本较高。为20份文档进行评分会显著增加每次请求的处理时间，在流式聊天API中（例如使用FastAPI构建的API），还会延迟第一个token的生成。请在延迟监测中单独统计这一环节的时间。虽然准确率提升通常值得付出这样的代价，但需根据预算调整候选项数量；我们关于为何重新排序需要付出延迟代价的文章进一步探讨了这一权衡问题。

使用LangChain实现整个流程

LangChain为每个环节提供了相应的构建模块，因此整个流程仅需两个简短的Python函数即可实现。以下示例仅为概念性说明，请根据实际环境调整连接字符串、模型及文件路径。

数据摄取：分块处理、嵌入向量并两次建立索引

该加载函数会读取一个文本文件，将其分割成每段1,000个字符且相互重叠100个字符的片段，然后以两种方式对这些片段建立索引。首先，使用OpenAI的text-embedding-3-small模型对它们进行嵌入，并通过PGVector.from_documents将其存储在pgvector集合中；其次，为这些片段适配BM25Retriever模型，并使用pickle将其序列化到磁盘上，因为BM25会在内存中构建索引。

import pickle
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_postgres.vectorstores import PGVector
from langchain_community.retrievers import BM25Retriever

CONNECTION_STRING = "postgresql+psycopg://user:password@localhost:5432/mydb"
COLLECTION_NAME = "hybrid_docs"

def ingest_documents(file_path: str):
    # 1. Load and chunk the document
    loader = TextLoader(file_path)
    docs = loader.load()

    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
    chunks = text_splitter.split_documents(docs)

    # 2. Store dense embeddings in pgvector
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    PGVector.from_documents(
        embedding=embeddings,
        documents=chunks,
        collection_name=COLLECTION_NAME,
        connection=CONNECTION_STRING,
    )

    # 3. Fit and save the BM25 sparse retriever
    bm25_retriever = BM25Retriever.from_documents(chunks)
    with open("bm25_retriever.pkl", "wb") as f:
        pickle.dump(bm25_retriever, f)

    print(f"Successfully ingested {len(chunks)} chunks.")

# Example usage:
# ingest_documents("technical_manual.txt")

需要注意的事项：

BM25索引只是快照。当文档内容发生变化时，必须重新构建并保存索引，否则它就会与向量存储失去同步。
仅可反序列化自己创建的pickle文件。加载pickle文件会执行其中的代码，因此被篡改的文件存在安全风险。

连接字符串中包含凭据；应从配置文件中加载，而非直接硬编码。

如果希望也在数据库内部进行关键词搜索，PostgreSQL内置的全文搜索功能可作为进程内BM25索引的替代方案，其排序机制有所不同。

检索：集成后再重新排序

检索函数会重建多个检索器并将它们串联起来。pgvector检索器会返回前10个语义匹配项（k=10），而解压后的BM25检索器也设置为返回10个结果。EnsembleRetriever以各0.5的权重将它们合并。带有top_n=3参数的压缩器会将这个集成体封装在ContextualCompressionRetriever中，这样每个查询只需一次invoke调用即可完成检索、合并和重新排序。

import pickle
from langchain_openai import OpenAIEmbeddings
from langchain_postgres.vectorstores import PGVector
from langchain.retrievers import EnsembleRetriever, ContextualCompressionRetriever
from langchain_cohere import CohereRerank

CONNECTION_STRING = "postgresql+psycopg://user:password@localhost:5432/mydb"
COLLECTION_NAME = "hybrid_docs"

def setup_hybrid_retriever():
    # 1. Initialize Vector Retriever
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = PGVector(
        connection=CONNECTION_STRING,
        embeddings=embeddings,
        collection_name=COLLECTION_NAME,
    )
    # Fetch top 10 semantic matches
    pgvector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

    # 2. Load Keyword Retriever (BM25)
    with open("bm25_retriever.pkl", "rb") as f:
        bm25_retriever = pickle.load(f)
    # Fetch top 10 exact keyword matches
    bm25_retriever.k = 10

    # 3. Merge pools with EnsembleRetriever (50/50 weighting)
    hybrid_retriever = EnsembleRetriever(
        retrievers=[bm25_retriever, pgvector_retriever],
        weights=[0.5, 0.5]
    )

    # 4. Rerank the combined 20 chunks to output the absolute top 3
    reranker = CohereRerank(cohere_api_key="YOUR_COHERE_API_KEY", top_n=3)
    advanced_retriever = ContextualCompressionRetriever(
        base_compressor=reranker,
        base_retriever=hybrid_retriever
    )

    return advanced_retriever

def query_system(query: str):
    retriever = setup_hybrid_retriever()
    best_docs = retriever.invoke(query)

    for i, doc in enumerate(best_docs):
        print(f"\n--- Result {i+1} ---")
        print(doc.page_content)

# Example usage:
# query_system("What is the warranty period for the TX-99402 sensor?")

一些容易被忽略的细节：

EnsembleRetriever并不会直接使用原始分数，而是通过加权倒数排名融合算法按排名顺序合并列表，从而避免上述的尺度不匹配问题。此外它还会去除重复项，因此当两个检索器找到相同的内容时，重新排序器接收到的数据块数量可能会少于20个。
绝不要在源代码中直接包含API密钥，应从环境变量或密钥管理工具中读取Cohere的密钥。
这里的setup_hybrid_retriever()函数会在每次查询时都被调用，每次都会重新连接Postgres并反序列化BM25模型。在真正的服务中，应在启动时仅构建一次检索器并重复使用。

LangChain 在不同版本中重新组织了其包结构，因此 EnsembleRetriever 和 ContextualCompressionRetriever 等类在您使用的版本中可能位于不同的包中。如果导入失败，请查看当前的 LangChain 文档。

关键要点

纯向量搜索在处理零件编号、SKU 和错误代码等精确字符串时表现不佳，BM25 能弥补这一缺陷。
pgvector 允许您在现有的 PostgreSQL 环境中添加密集检索功能，而无需单独的向量数据库。
不同检索器生成的分数无法直接比较，因此应先按排名合并结果，再由交叉编码器进行最终排序。
重新排序可以提高精确度，但会增加延迟；需有意识地控制候选项数量并持续监控。
应将 BM25 索引视为需随数据同步更新的构建产物，并避免将凭证信息写入代码中。

混合检索虽不能保证得到完美答案，但能消除生产环境中的RAG系统返回看似合理实则错误上下文的最常见原因。