多向量 | 提升召回相关性

用户4035096

发布于 2026-07-09 19:30:51

多向量 | 另一种提升召回结果相关性的解决方案

RAG严重依赖向量搜索, 但是有一个漏洞, 可能导致搜索出来的结果并不理想, 原因是相似性!=相关性.

业界的解决办法是混合搜索( 即: 向量(语义)搜索 + 全文检索 + 模糊查询等 + rerank ), 详见这篇文章: 别被“资本炒作”带坑里了！AI搜索!=向量搜索、相关性!=相似性

其实还有一种解决办法: 多向量搜索.

目前向量搜索结果不理想的根本原因是:

query句子/段落 embedding 与库内句子/段落 embedding 的相似性

换句话说, 段落由句子组成, 句子由token(词组)组成, 然而embedding对应的是token(词组)聚合后的向量值.

那么能不能直接使用token(词组)的向量值进行搜索, 减少信息的丢失呢?

能, 但是数据量太大了, 一句话可能包含很多个token, 一段话就更多了. 使用token搜索时会产生笛卡尔积式的搜索.

有没有可能加速呢?

这就是将介绍的 vectorchord maxsim 算子.

以下内容翻译自: https://blog.vectorchord.ai/supercharge-vector-search-with-colbert-rerank-in-postgresql

使用 PostgreSQL 中的 ColBERT rerank 增强向量搜索

传统的向量搜索方法通常使用句子嵌入来定位相似内容。然而，通过池化token嵌入来生成句子嵌入可能会牺牲token级别的细粒度细节。ColBERT通过将文本表示为token级别的多向量（而非单个聚合向量）来克服这个问题。这种方法利用token级别的上下文后期交互，使ColBERT能够保留更细微的信息，并比单纯依赖句子嵌入的方法提高搜索准确率。

ColBERT 结构（摘自原始论文）

如上图所示，ColBERT 将每个文档/查询编码为一个token向量列表(数组)，并在查询期间计算 MaxSim。

Token 级的后期交互计算需要更高的算力和存储空间。这使得在大型数据集中使用 ColBERT 搜索变得极具挑战性，尤其是在低延迟至关重要的情况下。

一种可能的解决方案是将句子级向量搜索与token级后期交互重排序相结合，从而利用近似向量搜索的效率和多向量相似性搜索的高质量。

多向量(即向量数组)方法不仅限于纯文本检索任务，它还可以用于视觉文档理解. 参考此文。对于多模态检索模型，像ColPali和ColQwen这样的先进模型直接将文档图像(pdf,扫描文本等)编码为多向量，并表现出比 OCR 转文本方法更强大的性能。

本博客将演示如何使用 PostgreSQL 扩展插件 VectorChord 和 pgvector 进行 ColBERT 重新排序。

教程

假设我们已经有了文档，让我们创建一个表来存储所有文档：

import psycopg  
from pgvector.psycopg import register_vector  

class PgClient:  
    def __init__(self, url: str, dataset: str, sentence_emb_dim: int, token_emb_dim: int):  
        self.dataset = dataset  
        self.sentence_emb_dim = sentence_emb_dim  
        self.token_emb_dim = token_emb_dim  
        self.conn = psycopg.connect(url, autocommit=True)  
        with self.conn.cursor() as cursor:  
            cursor.execute("CREATE EXTENSION IF NOT EXISTS vchord CASCADE;")  
        register_vector(self.conn)  

    def create(self):  
        with self.conn.cursor() as cursor:  
            cursor.execute(  
                f"CREATE TABLE IF NOT EXISTS {self.dataset}_corpus "
                "(id INT BY DEFAULT AS IDENTITY PRIMARY KEY, text TEXT, "
                f"emb vector({self.sentence_emb_dim}), embs vector({self.token_emb_dim})[]);"
            )

这里我们创建了一个包含句子级嵌入和token级嵌入的表。

有许多嵌入 API 和开源模型。您可以选择适合您用例的模型。

对于标记(token)级嵌入：

from colbert.infra import ColBERTConfig  
from colbert.modeling.checkpoint import Checkpoint  

class TokenEncoder:  
    def __init__(self):  
        self.config = ColBERTConfig(doc_maxlen=220, query_maxlen=32)  
        self.checkpoint = Checkpoint(  
            "colbert-ir/colbertv2.0", colbert_config=self.config, verbose=0  
        )  

    def encode_doc(self, doc: str):  
        return self.checkpoint.docFromText([doc], keep_dims=False)[0].numpy()  

    def encode_query(self, query: str):  
        return self.checkpoint.queryFromText([query])[0].numpy()

ColBERT 模型默认生成 128 维向量。

插入数据：

class PgClient:  
    ...  
    def insert(self, documents: list[str]):  
        with self.conn.cursor() as cursor:  
            for doc in tqdm(documents):  
                sentence_emb = sentence_encoder.encode_doc(doc)  
                token_embs = [emb for emb in token_encoder.encode(doc)]  
                cursor.execute(  
                    f"INSERT INTO {self.dataset}_corpus (text, emb, embs) VALUES (%s, %s, %s)"  
                    (doc, sentence_emb, token_embs)  
                )

对于向量搜索部分，我们可以使用VectorChord构建高性能的RaBitQ索引：

class PgClient:  
    ...  
    def index(self, num_doc: int, workers: int):  
        n_cluster = 1 << math.ceil(math.log2(num_doc ** 0.5 * 4))  
        config = f"""  
        residual_quantization = true  
        [build.internal]  
        lists = [{n_cluster}]  
        build_threads = {workers}  
        spherical_centroids = false  
        """  
        with self.conn.cursor() as cursor:  
            cursor.execute(f"SET max_parallel_maintenance_workers TO {workers}")  
            cursor.execute(f"SET max_parallel_workers TO {workers}")  
            cursor.execute(  
                f"CREATE INDEX {self.dataset}_rabitq ON {self.dataset}_corpus USING "  
                f"vchordrq (emb vector_l2_ops) WITH (options = $${config}$$)"  
            )

为了加快索引构建过程，我们可以利用外部质心(即k-means的聚集点)构建。更多详情，请参阅“在 VectorChord 中构建外部质心的优势和步骤” 。

现在，我们可以查询 PostgreSQL：

class PgClient:  
    ...  
    def query(self, doc: str, topk: int):  
        sentence_emb = sentence_encoder.encode_query(doc)  
        with self.conn.cursor() as cursor:  
            cursor.execute(  
                f"SELECT id, text FROM {self.dataset}_corpus ORDER BY emb <-> %s LIMIT {topk}"  
            )  
            res = cursor.fetchall()  
        return res

为了支持MaxSim重新排名(rerank)，我们需要创建一个函数：

class PgClient:  
    def __init__(self, url: str, dataset: str, sentence_emb_dim: int, token_emb_dim: int):  
        ...  
        self.conn.execute("""  
            CREATE OR REPLACE FUNCTION max_sim(document vector[], query vector[]) RETURNS double precision AS $$  
            WITH queries AS (  
                SELECT row_number() OVER () AS query_number, * FROM (SELECT unnest(query) AS query)  
            ),  
            documents AS (  
                SELECT unnest(document) AS document  
            ),  
            similarities AS (  
                SELECT query_number, document <=> query AS similarity FROM queries CROSS JOIN documents  
            ),  
            max_similarities AS (  
                SELECT MAX(similarity) AS max_similarity FROM similarities GROUP BY query_number  
            )  
            SELECT SUM(max_similarity) FROM max_similarities  
            $$ LANGUAGE SQL  
        """)

现在，我们可以对向量搜索检索到的文档进行重新排序：

class PgClient:  
    def rerank(self, query: str, ids: list[int], topk: int):  
        token_embs = [emb for emb in token_encoder.encode_query(query)]  
        with self.conn.cursor() as cursor:  
            cursor.execute(  
                f"SELECT id, text FROM {self.dataset}_corpus WHERE id = ANY(%s) ORDER BY "  
                f"max_sim(embs, %s) DESC LIMIT {topk}"  
                (ids, token_embs)  
            )  
            res = cursor.fetchall()  
        return res