用 PostgreSQL + pgvector 构建语义、混合、稀疏与量化向量检索系统

> 这是一篇基于 MarkTechPost 原文改写的可复制教程。原文使用 Google Colab 从源码编译 pgvector；本文保留原文核心路线，并补充一个更适合本地验证的 Docker 启动方式。

原文：<https://www.marktechpost.com/2026/05/28/a-coding-guide-to-implement-a-pgvector-powered-semantic-hybrid-sparse-and-quantized-vector-search-system/>
标题：A Coding Guide to Implement a pgvector-Powered Semantic, Hybrid, Sparse, and Quantized Vector Search System
来源：MarkTechPost
发布时间：2026-05-28
作者：Sana Hassan
原文 Notebook / GitHub：原文提到 “Full Codes with Notebook here”，但本次提取结果未提供可验证的 Notebook URL。
提取说明：原站可能存在反爬/挑战页；本教程基于 nested Jina Reader 提取到的正文和代码块整理，已清理 cookie、导航和复制按钮噪声。

1. 你会做出什么

完成后，你会得到一个最小 pgvector 检索实验台，支持：

1. 把文本转成 embedding，并存入 PostgreSQL。 2. 用 HNSW 索引做语义搜索。 3. 按元数据过滤搜索结果。 4. 对比 L2、cosine、inner product、L1 等距离度量。 5. 用 halfvec 做半精度向量存储。 6. 用 binary_quantize 做粗召回，再用原始向量精排。 7. 用 sparsevec 做稀疏向量检索。 8. 把向量检索和 PostgreSQL 全文检索用 RRF 融合。 9. 用 AVG(embedding) 做类别质心分析。

适合的场景：

RAG 文档检索原型。
小到中等规模推荐系统 PoC。
已经使用 PostgreSQL，又不想太早引入独立向量数据库的团队。
学习 pgvector 查询语法和索引机制。

不适合直接宣称的场景：

高 QPS 生产系统。
超大规模向量库。
多租户隔离、分布式扩展、GPU 检索等场景。

原文示例只有 7 条文本，没有给生产压测数据；所以本文把它定位为“可运行实验台”，不是生产架构蓝图。

2. 准备环境

方案 A：沿用原文思路，在 Colab / Linux 中编译 pgvector

原文路线是：安装 PostgreSQL、安装开发依赖、克隆 pgvector 源码、编译安装。

apt-get -qq update
apt-get -qq install -y postgresql postgresql-contrib postgresql-server-dev-all build-essential git
git clone --depth 1 https://github.com/pgvector/pgvector.git /tmp/pgvector
cd /tmp/pgvector && make && make install
service postgresql start
sudo -u postgres psql -c "ALTER USER postgres PASSWORD 'postgres';"
python -m pip install -q pgvector "psycopg[binary]" sentence-transformers numpy

方案 B：本地快速验证，用 pgvector 镜像启动 PostgreSQL（实用扩展）

如果你只是想本地跑通，不想编译扩展，可以直接用 pgvector 官方镜像：

docker run --name pgvector-demo \
  -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=postgres \
  -p 5432:5432 \
  -d pgvector/pgvector:pg16

python -m venv .venv
source .venv/bin/activate
python -m pip install -U pip
python -m pip install pgvector "psycopg[binary]" sentence-transformers numpy

验证数据库可连接：

psql "postgresql://postgres:[email protected]:5432/postgres" -c "CREATE EXTENSION IF NOT EXISTS vector; SELECT extversion FROM pg_extension WHERE extname='vector';"

预期输出形态：能看到 pgvector 的扩展版本，例如 0.x.x。

3. 建表并写入 embedding

新建 pgvector_playground.py：

from __future__ import annotations

import numpy as np
import psycopg
from pgvector.psycopg import register_vector
from sentence_transformers import SentenceTransformer

DSN = "host=127.0.0.1 port=5432 dbname=postgres user=postgres password=postgres"


def connect() -> psycopg.Connection:
    conn = psycopg.connect(DSN, autocommit=True)
    conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
    register_vector(conn)
    version = conn.execute("SELECT extversion FROM pg_extension WHERE extname='vector'").fetchone()[0]
    print(f"pgvector version: {version}")
    return conn


def build_corpus() -> tuple[list[str], list[str]]:
    rows = [
        ("Octopuses have three hearts and blue blood.", "animals"),
        ("Transformers revolutionized natural language processing.", "technology"),
        ("Quantum computers exploit superposition and entanglement.", "technology"),
        ("GPUs accelerate deep learning by parallelizing matrix math.", "technology"),
        ("Sourdough bread relies on wild yeast and lactobacilli.", "food"),
        ("Dark chocolate contains flavonoid antioxidants.", "food"),
        ("A black hole's gravity is so strong light cannot escape.", "space"),
    ]
    return [content for content, _ in rows], [category for _, category in rows]


def prepare_documents(conn: psycopg.Connection, model: SentenceTransformer) -> int:
    contents, categories = build_corpus()
    embeddings = model.encode(contents, normalize_embeddings=True)
    dim = model.get_sentence_embedding_dimension()

    conn.execute("DROP TABLE IF EXISTS documents")
    conn.execute(
        f"""
        CREATE TABLE documents (
            id bigserial PRIMARY KEY,
            content text NOT NULL,
            category text NOT NULL,
            embedding vector({dim}) NOT NULL
        )
        """
    )

    with conn.cursor() as cur:
        cur.executemany(
            "INSERT INTO documents (content, category, embedding) VALUES (%s, %s, %s)",
            [(content, category, np.asarray(embedding)) for content, category, embedding in zip(contents, categories, embeddings)],
        )

    print(f"inserted {len(contents)} documents with {dim}-d embeddings")
    return dim

这一步做了三件事：

启用 vector 扩展。
用 all-MiniLM-L6-v2 生成 384 维 embedding。
把文本、类别、向量写进 documents 表。

运行时，首次加载模型会下载模型文件，耗时取决于网络。

4. 添加 HNSW 索引并做语义搜索

继续追加代码：

def create_hnsw_index(conn: psycopg.Connection) -> None:
    conn.execute("DROP INDEX IF EXISTS documents_embedding_hnsw_idx")
    conn.execute(
        """
        CREATE INDEX documents_embedding_hnsw_idx
        ON documents
        USING hnsw (embedding vector_cosine_ops)
        WITH (m = 16, ef_construction = 64)
        """
    )
    conn.execute("SET hnsw.ef_search = 100")


def semantic_search(conn: psycopg.Connection, model: SentenceTransformer, query: str, k: int = 4) -> list[tuple[str, str, float]]:
    query_vector = np.asarray(model.encode(query, normalize_embeddings=True))
    rows = conn.execute(
        """
        SELECT content, category, embedding <=> %s AS distance
        FROM documents
        ORDER BY distance
        LIMIT %s
        """,
        (query_vector, k),
    ).fetchall()
    return [(content, category, float(distance)) for content, category, distance in rows]

调用示例：

for content, category, distance in semantic_search(conn, model, "animals that are unusually quick"):
    print(f"{distance:.3f} [{category}] {content}")

说明：

embedding <=> query_vector 是 cosine distance。
距离越小，表示越相似。
m = 16、ef_construction = 64、ef_search = 100 是原文示例参数，不是生产默认答案。

5. 加上元数据过滤

向量库通常不只按相似度查，还要加业务过滤条件，比如类别、时间、租户、权限。

def filtered_search(conn: psycopg.Connection, model: SentenceTransformer, query: str, category: str, k: int = 3) -> list[tuple[str, float]]:
    query_vector = np.asarray(model.encode(query, normalize_embeddings=True))
    rows = conn.execute(
        """
        SELECT content, embedding <=> %s AS distance
        FROM documents
        WHERE category = %s
        ORDER BY distance
        LIMIT %s
        """,
        (query_vector, category, k),
    ).fetchall()
    return [(content, float(distance)) for content, distance in rows]

调用示例：

for content, distance in filtered_search(conn, model, "objects with extreme gravity", "space"):
    print(f"{distance:.3f} {content}")

预期结果：black hole 相关文本应该排在前面。

6. 对比不同距离度量

pgvector 支持多种距离算子：

<->：L2 distance
<=>：cosine distance
<#>：negative inner product
<+>：L1 distance

示例：

def compare_distance_metrics(conn: psycopg.Connection, model: SentenceTransformer, query: str) -> list[tuple[str, float, str]]:
    query_vector = np.asarray(model.encode(query, normalize_embeddings=True))
    metrics = [
        ("<->", "L2"),
        ("<=>", "cosine"),
        ("<#>", "negative_inner_product"),
        ("<+>", "L1"),
    ]
    results: list[tuple[str, float, str]] = []
    for operator, label in metrics:
        content, score = conn.execute(
            f"SELECT content, embedding {operator} %s AS score FROM documents ORDER BY score LIMIT 1",
            (query_vector,),
        ).fetchone()
        results.append((label, float(score), content))
    return results

调用：

for label, score, content in compare_distance_metrics(conn, model, "brewing a hot caffeinated drink"):
    print(f"{label}: {score:+.3f} {content}")

注意：这里使用 f-string 拼接 SQL 运算符是安全的，因为运算符来自固定白名单，不来自用户输入。真实服务里不要把用户输入直接拼进 SQL。

7. 用 halfvec 降低存储成本

halfvec 使用半精度浮点存储向量，目标是降低存储和索引成本。

def add_halfvec_search(conn: psycopg.Connection, model: SentenceTransformer, dim: int) -> list[tuple[str, float]]:
    conn.execute(f"ALTER TABLE documents ADD COLUMN IF NOT EXISTS embedding_half halfvec({dim})")
    conn.execute("UPDATE documents SET embedding_half = embedding::halfvec")
    conn.execute("DROP INDEX IF EXISTS documents_embedding_half_hnsw_idx")
    conn.execute(
        """
        CREATE INDEX documents_embedding_half_hnsw_idx
        ON documents
        USING hnsw (embedding_half halfvec_cosine_ops)
        """
    )

    query_vector = np.asarray(model.encode("the galaxy we live in", normalize_embeddings=True))
    rows = conn.execute(
        """
        SELECT content, embedding_half <=> %s::halfvec AS distance
        FROM documents
        ORDER BY distance
        LIMIT 2
        """,
        (query_vector,),
    ).fetchall()
    return [(content, float(distance)) for content, distance in rows]

使用建议：

PoC 阶段可以先用全精度 vector。
数据量变大后，再比较 vector 和 halfvec 的召回质量、索引大小、查询延迟。
不要只看“能不能跑”，要看召回结果是否变差。

8. 二进制量化：先粗召回，再精排

二进制量化的常见思路：

1. 把向量压成 bit 表示。 2. 用汉明距离快速找候选。 3. 对候选再用原始向量做精确重排。

def binary_quantized_rerank(conn: psycopg.Connection, model: SentenceTransformer, dim: int) -> list[tuple[str, float]]:
    conn.execute("DROP INDEX IF EXISTS documents_embedding_binary_hnsw_idx")
    conn.execute(
        f"""
        CREATE INDEX documents_embedding_binary_hnsw_idx
        ON documents
        USING hnsw ((binary_quantize(embedding)::bit({dim})) bit_hamming_ops)
        """
    )

    query_vector = np.asarray(model.encode("parallel hardware for AI training", normalize_embeddings=True))
    sql = f"""
        SELECT content, candidates.embedding <=> %(query_vector)s AS exact_distance
        FROM (
            SELECT content, embedding
            FROM documents
            ORDER BY binary_quantize(embedding)::bit({dim}) <~> binary_quantize(%(query_vector)s)::bit({dim})
            LIMIT 8
        ) AS candidates
        ORDER BY exact_distance
        LIMIT 3
    """
    rows = conn.execute(sql, {"query_vector": query_vector}).fetchall()
    return [(content, float(distance)) for content, distance in rows]

这里的关键不是“量化后直接当最终答案”，而是“两阶段检索”：

第一阶段：快，召回候选。
第二阶段：准，用原始向量重排。

生产化前需要验证：候选数从 8 改成 20、50、100 时，召回质量和延迟如何变化。

9. sparsevec：用稀疏向量表达关键词权重

稀疏向量适合表达关键词权重、SPLADE 风格稀疏表示，或者你自己构造的特征权重。

from pgvector import SparseVector


def sparse_vector_search(conn: psycopg.Connection) -> list[tuple[int, float, list[int]]]:
    conn.execute("DROP TABLE IF EXISTS sparse_items")
    conn.execute("CREATE TABLE sparse_items (id bigserial PRIMARY KEY, embedding sparsevec(10))")

    sparse_data = [
        SparseVector({0: 1.0, 3: 2.0, 7: 1.5}, 10),
        SparseVector({1: 0.5, 3: 1.0, 9: 3.0}, 10),
        SparseVector({0: 0.2, 4: 2.5, 7: 0.8}, 10),
    ]

    with conn.cursor() as cur:
        cur.executemany("INSERT INTO sparse_items (embedding) VALUES (%s)", [(item,) for item in sparse_data])

    query = SparseVector({0: 1.0, 7: 1.0}, 10)
    rows = conn.execute(
        """
        SELECT id, embedding, embedding <#> %s AS negative_inner_product
        FROM sparse_items
        ORDER BY negative_inner_product
        LIMIT 3
        """,
        (query,),
    ).fetchall()
    return [(int(row_id), -float(negative_inner_product), embedding.indices()) for row_id, embedding, negative_inner_product in rows]

<#> 返回的是 negative inner product，所以示例里用 -negative_inner_product 转回正向内积，方便阅读。

10. 混合检索：向量召回 + 全文检索 + RRF

纯向量检索擅长语义相似，但可能漏掉精确关键词；全文检索擅长关键词命中，但不理解语义。混合检索把两者结合起来。

原文使用 Reciprocal Rank Fusion（RRF）：

score = 1 / (60 + rank)

SQL 示例：

def hybrid_search(conn: psycopg.Connection, model: SentenceTransformer, user_query: str) -> list[tuple[str, float]]:
    query_vector = np.asarray(model.encode(user_query, normalize_embeddings=True))
    sql = """
    WITH semantic AS (
        SELECT id, RANK() OVER (ORDER BY embedding <=> %(query_vector)s) AS rank
        FROM documents
        ORDER BY embedding <=> %(query_vector)s
        LIMIT 20
    ),
    keyword AS (
        SELECT d.id,
               RANK() OVER (ORDER BY ts_rank_cd(to_tsvector('english', d.content), q) DESC) AS rank
        FROM documents d, plainto_tsquery('english', %(query_text)s) AS q
        WHERE to_tsvector('english', d.content) @@ q
    )
    SELECT d.content,
           COALESCE(1.0 / (60 + semantic.rank), 0.0)
         + COALESCE(1.0 / (60 + keyword.rank), 0.0) AS rrf_score
    FROM documents d
    LEFT JOIN semantic ON d.id = semantic.id
    LEFT JOIN keyword ON d.id = keyword.id
    WHERE semantic.id IS NOT NULL OR keyword.id IS NOT NULL
    ORDER BY rrf_score DESC
    LIMIT 4
    """
    rows = conn.execute(sql, {"query_vector": query_vector, "query_text": user_query}).fetchall()
    return [(content, float(score)) for content, score in rows]

调用：

for content, score in hybrid_search(conn, model, "fast animal"):
    print(f"{score:.5f} {content}")

实践建议：

RRF 常数 60 是经验值，不要当成固定真理。
如果业务更看重关键词精确命中，可以给全文检索一侧加权。
如果业务更看重语义泛化，可以给向量检索一侧加权。
最终应使用人工标注样本或线上点击/转化数据评估排序质量。

11. 向量聚合：找类别质心和代表样本

pgvector 支持对向量做聚合，例如计算某个类别的平均 embedding。

def category_centroid(conn: psycopg.Connection, category: str) -> tuple[int, str, float]:
    centroid = conn.execute("SELECT AVG(embedding) FROM documents WHERE category = %s", (category,)).fetchone()[0]
    content, distance = conn.execute(
        """
        SELECT content, embedding <=> %s AS distance
        FROM documents
        WHERE category = %s
        ORDER BY distance
        LIMIT 1
        """,
        (np.asarray(centroid), category),
    ).fetchone()
    return len(centroid), content, float(distance)

用途：

找某个类别最有代表性的文档。
给用户兴趣、商品类别、知识主题构造中心向量。
做聚类结果解释。

局限：平均向量不一定能表达复杂类别边界，尤其当类别内部本身包含多个子主题时。

12. 完整主函数

把前面的函数串起来：

def main() -> None:
    conn = connect()
    model = SentenceTransformer("all-MiniLM-L6-v2")
    dim = prepare_documents(conn, model)

    create_hnsw_index(conn)

    print("\nsemantic search")
    for content, category, distance in semantic_search(conn, model, "animals that are unusually quick"):
        print(f"{distance:.3f} [{category}] {content}")

    print("\nfiltered search")
    for content, distance in filtered_search(conn, model, "objects with extreme gravity", "space"):
        print(f"{distance:.3f} {content}")

    print("\ndistance metrics")
    for label, score, content in compare_distance_metrics(conn, model, "brewing a hot caffeinated drink"):
        print(f"{label}: {score:+.3f} {content}")

    print("\nhalfvec search")
    for content, distance in add_halfvec_search(conn, model, dim):
        print(f"{distance:.3f} {content}")

    print("\nbinary quantized rerank")
    for content, distance in binary_quantized_rerank(conn, model, dim):
        print(f"{distance:.3f} {content}")

    print("\nsparse vector search")
    for row_id, inner_product, indices in sparse_vector_search(conn):
        print(f"id={row_id} inner_product={inner_product:.2f} indices={indices}")

    print("\nhybrid search")
    for content, score in hybrid_search(conn, model, "fast animal"):
        print(f"{score:.5f} {content}")

    print("\ncategory centroid")
    centroid_dim, content, distance = category_centroid(conn, "food")
    print(f"centroid_dim={centroid_dim} representative={content} distance={distance:.3f}")

    conn.close()


if __name__ == "__main__":
    main()

运行：

python pgvector_playground.py

预期输出形态：

pgvector version: 0.x.x
inserted 7 documents with 384-d embeddings

semantic search
0.xxx [animals] Octopuses have three hearts and blue blood.
...

hybrid search
0.0xxxx Octopuses have three hearts and blue blood.
...

注意：embedding 模型版本、pgvector 版本、硬件和索引参数会影响具体分数；教程只要求输出形态和相对语义合理，不要求数字完全一致。

13. 从实验台走向真实项目，需要补哪些东西

原文重点是教学演示，不覆盖生产问题。真正落地前至少补以下检查：

13.1 数据规模压测

分别测试：

1 万条文档。
10 万条文档。
100 万条文档。

记录：

索引构建时间。
索引大小。
查询 p50 / p95 / p99 延迟。
内存占用。
插入和更新成本。

13.2 召回质量评估

准备一批查询和人工标注答案，比较：

纯全文检索。
纯向量检索。
向量 + 全文混合检索。
vector vs halfvec。
二进制量化候选数 8 / 20 / 50 / 100。

13.3 权限与业务过滤

真实 RAG 系统通常不能只做相似度排序，还要过滤：

用户权限。
租户 ID。
文档状态。
发布时间。
业务类型。

如果过滤条件很强，索引和查询计划可能变化，需要单独验证。

13.4 数据更新策略

需要明确：

新文档何时生成 embedding。
旧文档修改后如何重算 embedding。
删除文档是否软删除。
索引膨胀如何处理。
embedding 模型升级后是否全量重建。

14. 推荐的最小落地路线

如果你要把这套方法用于自己的项目，不建议一开始就追求所有高级特性。按下面顺序推进更稳：

1. 先跑通 vector + cosine + HNSW。 2. 加上业务过滤字段，例如 tenant_id、category、created_at。 3. 做一版全文检索 + 向量检索的 RRF 混合召回。 4. 用人工样本评估召回质量。 5. 数据量变大后，再测试 halfvec 和二进制量化。 6. 只有当 PostgreSQL 无法满足容量、吞吐、隔离或运维目标时，再评估专用向量数据库。

15. 一句话总结

这篇原文最有价值的地方，不是证明 PostgreSQL 可以替代所有向量数据库，而是给了一个很完整的 pgvector 检索实验路线：从普通语义检索开始，逐步扩展到过滤、索引、距离度量、半精度、量化、稀疏向量、混合检索和聚合分析。照这个顺序做，你可以用很低的系统复杂度验证一个 RAG / 推荐 / 相似度搜索原型。