PostgreSQL + pgvector 存 OpenAI / 本地 embeddings 做向量检索

RAG / 语义搜索的标准做法：把文档切成 chunk → 用 embedding model 转向量 →
存向量库 → 查询时 embedding 后 ANN 搜索。

向量库选项：
- 专用：Qdrant / Milvus / Weaviate / Chroma
- 通用 + 向量扩展：PostgreSQL + pgvector

如果你已经在用 PG，pgvector 是最省事的——一套数据库管业务数据 + 向量，
不引入新系统。下面是完整流程。

1. 装扩展

# Debian / Ubuntu
sudo apt install postgresql-16-pgvector

# 或编译：
# git clone https://github.com/pgvector/pgvector
# cd pgvector && make && sudo make install

-- 在目标数据库里执行
CREATE EXTENSION IF NOT EXISTS vector;

2. 建表

CREATE TABLE documents (
    id BIGSERIAL PRIMARY KEY,
    source TEXT NOT NULL,
    chunk TEXT NOT NULL,
    embedding vector(1536),       -- OpenAI text-embedding-3-small 维度
    metadata JSONB,
    created_at TIMESTAMPTZ DEFAULT now()
);

vector(N) 是定长向量类型，N 必须匹配你的 embedding model 输出维度。
常见：

OpenAI text-embedding-3-small: 1536
OpenAI text-embedding-3-large: 3072
Cohere embed-multilingual-v3: 1024
bge-large-zh-v1.5: 1024
bge-m3: 1024
nomic-embed-text: 768

3. 插入 embedding

import psycopg
from openai import OpenAI

client = OpenAI()

def embed(text):
    resp = client.embeddings.create(input=text, model='text-embedding-3-small')
    return resp.data[0].embedding   # List[float]

con = psycopg.connect('postgresql://localhost/mydb')

text = 'PostgreSQL 是开源关系数据库...'
emb = embed(text)

con.execute(
    'INSERT INTO documents (source, chunk, embedding) VALUES (%s, %s, %s)',
    ('manual.md', text, emb)
)
con.commit()

psycopg3 + pgvector-python：

uv add psycopg pgvector

from pgvector.psycopg import register_vector
register_vector(con)   # 现在能直接传 numpy array / list 给 vector 字段

4. ANN 搜索

-- 找最相似的 5 条（距离最小）
SELECT id, source, chunk, embedding <=> $1::vector AS distance
FROM documents
ORDER BY embedding <=> $1::vector
LIMIT 5;

<=> 是余弦距离运算符。pgvector 还支持：

<->：欧氏距离 (L2)
<#>：内积负值（dot product 越大越相似，所以取负）

最常用的是 <=> 余弦距离，对长度归一化的 embedding 等价于内积。

5. 索引：HNSW 或 IVFFlat

无索引时是 brute-force 扫表，100k 行还能用，百万级就慢。
建索引：

-- HNSW（推荐，召回高）
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- IVFFlat（建索引快，召回略低）
CREATE INDEX ON documents
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

索引必须匹配你的距离操作符：

vector_cosine_ops ↔ <=>
vector_l2_ops ↔ <->
vector_ip_ops ↔ <#>

lists 推荐 sqrt(rows)，ef_construction 越大召回越好但建索引越慢。

6. 查询时调召回 / 速度

-- HNSW
SET hnsw.ef_search = 100;  -- 默认 40；越大召回越好越慢
SELECT ... FROM documents ORDER BY embedding <=> $1 LIMIT 10;

-- IVFFlat
SET ivfflat.probes = 10;   -- 默认 1；越大召回越好越慢
SELECT ... FROM documents ORDER BY embedding <=> $1 LIMIT 10;

通常 ef_search 64-256 之间是甜点。

7. 混合搜索（向量 + 全文 + 元数据过滤）

向量搜索的弱点：精确关键词容易丢。最佳实践是混合：

-- 假设 chunk 上建了 to_tsvector('simple', chunk) 的 GIN 索引
WITH vector_results AS (
  SELECT id, chunk, embedding <=> $1::vector AS dist
  FROM documents
  WHERE metadata->>'project' = $2
  ORDER BY embedding <=> $1::vector
  LIMIT 50
),
fts_results AS (
  SELECT id, chunk,
    ts_rank(to_tsvector('simple', chunk), plainto_tsquery('simple', $3)) AS rank
  FROM documents
  WHERE metadata->>'project' = $2
    AND to_tsvector('simple', chunk) @@ plainto_tsquery('simple', $3)
  LIMIT 50
)
SELECT * FROM (
  SELECT id, chunk, 1 - dist AS score FROM vector_results
  UNION ALL
  SELECT id, chunk, rank * 5 AS score FROM fts_results
)
GROUP BY id, chunk
ORDER BY MAX(score) DESC
LIMIT 10;

或者更精致用 RRF (Reciprocal Rank Fusion) 算法。

8. 性能数据（参考）

量级	brute force	HNSW
10k 行	10-50ms	< 5ms
100k	100-500ms	10ms
1M	几秒	20-50ms
10M	60s+	100ms

1000 万向量是 PostgreSQL + pgvector 大致甜点。再大上 Qdrant / Milvus。

9. 批量插入

from pgvector.psycopg import register_vector
register_vector(con)

# 批量
with con.cursor() as cur:
    cur.executemany(
        'INSERT INTO documents (source, chunk, embedding) VALUES (%s, %s, %s)',
        [(s, c, e) for s, c, e in zip(sources, chunks, embeddings)]
    )
con.commit()

千条以上用 COPY ... FROM STDIN，10x 速度。

10. 用 Django

# settings: 装 'pgvector.django' 应用
from pgvector.django import VectorField, HnswIndex

class Document(models.Model):
    source = models.CharField(max_length=200)
    chunk = models.TextField()
    embedding = VectorField(dimensions=1536)

    class Meta:
        indexes = [
            HnswIndex(
                name='doc_emb_hnsw',
                fields=['embedding'],
                m=16, ef_construction=64,
                opclasses=['vector_cosine_ops'],
            ),
        ]

# 查询
from pgvector.django import CosineDistance
Document.objects.alias(d=CosineDistance('embedding', query_emb)).order_by('d')[:10]

踩过的坑

维度不匹配：插 1024 维向量到 vector(1536) 字段会报错。
embedding model 一定要固定，换 model 必须重建索引。
HNSW 索引构建非常慢且耗内存（10M 行可能要几小时 + 10GB+ 内存）。
生产建议在低峰期 CREATE INDEX CONCURRENTLY。
pgvector 不存储原文，只存向量：要返回相关文档需要把原文也存表里。
别在 vector 列上做 WHERE 条件而不带 ORDER BY ... LIMIT：
全表扫的 vector 距离计算极慢。索引只在 ORDER BY 配 LIMIT 时生效。