PostgreSQL pgvector 向量检索完全教程
适用读者：具备 PostgreSQL 基础、了解向量检索概念的后端/AI 工程师
最后更新：2025 年 5 月
预计阅读时间：35 分钟
目录
- 为什么选择 pgvector
- 安装与配置
- 向量数据类型详解
- 索引类型：IVFFlat 与 HNSW
- 距离度量与相似性计算
- 混合查询：SQL 条件 + 向量检索
- 性能优化与索引调优
- 与专用向量数据库对比
- RAG 系统集成实战
- 分布式方案：Citus + pgvector
- 监控与运维
- 企业级部署最佳实践
- 总结
1. 为什么选择 pgvector
在构建 AI 应用时，向量检索（Vector Search）是核心基础设施之一——无论是 RAG（检索增强生成）、语义搜索、推荐系统还是图像检索，都需要高效地在高维空间中找到最相似的向量。
市面上已有 Milvus、Pinecone、Qdrant、Weaviate 等专用向量数据库，但 pgvector 提供了一个极具吸引力的替代方案：直接在 PostgreSQL 中进行向量检索。
1.1 核心优势
| 优势 | 说明 |
|---|---|
| 零额外基础设施 | 不需要引入新的数据库，复用现有 PostgreSQL |
| 事务一致性 | 向量数据与业务数据在同一事务中，保证 ACID |
| SQL 表达力 | 向量检索与关系查询无缝结合（WHERE、JOIN、GROUP BY） |
| 运维简化 | 备份、复制、监控等全部复用 PostgreSQL 生态 |
| 成本效益 | 不需要为单独的向量数据库付费和运维 |
| 生态成熟 | PostgreSQL 的扩展生态（PostGIS、TimescaleDB 等）可组合使用 |
1.2 适用场景
- 中小规模向量库（<5000 万条向量）
- 需要混合查询（向量 + 结构化过滤）
- 已有 PostgreSQL 基础设施
- 强一致性要求（金融、电商等）
- RAG 系统原型和中等规模生产环境
1.3 何时考虑专用向量数据库
- 向量规模超过 1 亿条，且对延迟要求极高（<10ms）
- 需要分布式向量索引（跨节点分片）
- 纯向量检索场景，不需要关系查询
- 需要 GPU 加速索引构建
2. 安装与配置
2.1 系统要求
| 组件 | 最低版本 | 推荐版本 |
|---|---|---|
| PostgreSQL | 12 | 16 |
| pgvector | 0.5.0 | 0.7.x |
| 操作系统 | Linux / macOS / Windows | Ubuntu 22.04 / Amazon Linux 2023 |
2.2 安装方法
方法一：源码编译安装
# 克隆仓库
git clone --branch v0.7.4 https://github.com/pgvector/pgvector.git
cd pgvector
# 编译（需要 PostgreSQL 开发包）
make
sudo make install
# 在数据库中启用
psql -d your_database -c "CREATE EXTENSION vector;"
方法二：apt 安装（Ubuntu/Debian）
# 添加 PostgreSQL 官方仓库（如果尚未添加）
sudo sh -c 'echo "deb http://apt.postgresql.org/pub/repos/apt $(lsb_release -cs)-pgdg main" > /etc/apt/sources.list.d/pgdg.list'
curl -fsSL https://www.postgresql.org/media/keys/ACCC4CF8.asc | sudo gpg --dearmor -o /etc/apt/trusted.gpg.d/postgresql.gpg
sudo apt-get update
# 安装对应版本的 pgvector
sudo apt-get install postgresql-16-pgvector
# 启用扩展
sudo -u postgres psql -d your_database -c "CREATE EXTENSION vector;"
方法三：Docker 快速启动
# 使用官方 pgvector Docker 镜像
docker run -d \
--name pgvector \
-e POSTGRES_PASSWORD=mysecretpassword \
-p 5432:5432 \
pgvector/pgvector:pg16
# 连接并启用扩展
docker exec -it pgvector psql -U postgres -c "CREATE EXTENSION vector;"
方法四：云服务
主流云数据库已原生支持 pgvector：
| 云服务 | 支持情况 | 说明 |
|---|---|---|
| Amazon RDS | ✅ PostgreSQL 15.4+ | 直接 CREATE EXTENSION vector; |
| Amazon Aurora | ✅ PostgreSQL 15.4+ | 同上 |
| Google Cloud SQL | ✅ PostgreSQL 15+ | 已预装 |
| Azure Database | ✅ PostgreSQL 14+ | 需要 azure.extensions 参数 |
| Supabase | ✅ 所有版本 | 默认可用 |
| Neon | ✅ 所有版本 | 默认可用 |
2.3 验证安装
-- 检查扩展版本
SELECT * FROM pg_extension WHERE extname = 'vector';
-- 或
SELECT extversion FROM pg_extension WHERE extname = 'vector';
-- 快速测试
SELECT '[1,2,3]'::vector <-> '[1,2,4]'::vector AS cosine_distance;
3. 向量数据类型详解
pgvector 提供了多种向量数据类型，适用于不同精度和存储需求。
3.1 支持的数据类型
| 类型 | 元素类型 | 维度限制 | 存储大小（每维） | 典型用途 |
|---|---|---|---|---|
vector |
float32 | 16000 | 4 字节 | 通用嵌入 |
halfvec |
float16 | 16000 | 2 字节 | 大规模存储，节省空间 |
sparsevec |
float32（稀疏） | 16000 | 可变 | 稀疏嵌入（如 SPLADE） |
bit |
二进制 | 8000 | 1/8 字节 | 二值嵌入（如 SimHash） |
3.2 创建向量表
-- 基本向量表
CREATE TABLE documents (
id BIGSERIAL PRIMARY KEY,
content TEXT NOT NULL,
metadata JSONB DEFAULT '{}',
embedding VECTOR(1536), -- OpenAI text-embedding-3-small
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- 使用半精度向量节省存储
CREATE TABLE document_embeddings (
id BIGSERIAL PRIMARY KEY,
document_id BIGINT REFERENCES documents(id),
embedding HALFVEC(1536), -- 存储空间减半
model_name VARCHAR(100) NOT NULL
);
-- 稀疏向量（适用于 SPLADE 等稀疏嵌入模型）
CREATE TABLE sparse_embeddings (
id BIGSERIAL PRIMARY KEY,
content TEXT NOT NULL,
embedding SPARSEVEC(25000) -- 稀疏向量
);
-- 二值向量（适用于 SimHash、MinHash 等）
CREATE TABLE binary_hashes (
id BIGSERIAL PRIMARY KEY,
content TEXT NOT NULL,
hash BIT(256) -- 256 位二值哈希
);
3.3 向量操作与语法
-- 插入向量（多种格式）
INSERT INTO documents (content, embedding) VALUES
('Hello world', '[0.1, 0.2, 0.3, ...]'), -- 字符串格式
('Test', ARRAY[0.1, 0.2, 0.3]::vector), -- 数组转换
('Python', '{0.1, 0.2, 0.3}'::vector); -- PostgreSQL 数组格式
-- 向量运算
SELECT
embedding <-> '[0.1, 0.2, ...]'::vector AS l2_distance,
embedding <=> '[0.1, 0.2, ...]'::vector AS cosine_distance,
embedding <#> '[0.1, 0.2, ...]'::vector AS inner_product
FROM documents
LIMIT 5;
-- 向量维度查询
SELECT vector_dims(embedding) FROM documents LIMIT 1;
-- 向量聚合
SELECT
AVG(embedding)::vector AS centroid,
COUNT(*) AS doc_count
FROM documents
WHERE metadata->>'category' = 'tech';
3.4 类型转换
-- vector ↔ array
SELECT ARRAY[1.0, 2.0, 3.0]::vector;
SELECT '[1,2,3]'::vector::real[];
-- halfvec 转换
SELECT '[1,2,3]'::vector::halfvec;
SELECT '[1,2,3]'::halfvec::vector;
-- 稀疏向量格式: {index:value, index:value}
SELECT '{1:0.5, 3:0.8, 100:0.3}'::sparsevec(1000);
4. 索引类型：IVFFlat 与 HNSW
pgvector 支持两种主要的近似最近邻（ANN）索引类型，各有特点。
4.1 IVFFlat（Inverted File with Flat Quantization）
IVFFlat 将向量空间划分为若干聚类（Voronoi cells），查询时只搜索最近的几个聚类。
-- 创建 IVFFlat 索引
CREATE INDEX ON documents
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);
-- 不同距离度量的索引
CREATE INDEX ON documents USING ivfflat (embedding vector_l2_ops) WITH (lists = 100);
CREATE INDEX ON documents USING ivfflat (embedding vector_ip_ops) WITH (lists = 100);
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
lists 参数选择指南：
| 数据量 | 推荐 lists 值 | 说明 |
|---|---|---|
| <10,000 | rows / 1000 | 数据量小时，每个聚类至少 1000 条 |
| 10,000 - 1,000,000 | sqrt(rows) | 经验公式 |
| 1,000,000 - 10,000,000 | rows / 1000 | 每个聚类 1000 条左右 |
| >10,000,000 | 考虑使用 HNSW | IVFFlat 效率下降 |
IVFFlat 特点：
- ✅ 索引构建速度快
- ✅ 内存占用较低
- ❌ 查询精度不如 HNSW
- ❌ 需要在数据加载完成后才能创建索引
- ❌ 数据更新后需要重建索引
4.2 HNSW（Hierarchical Navigable Small World）
HNSW 基于小世界图结构，是目前最流行的 ANN 索引算法。
-- 创建 HNSW 索引（余弦距离）
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
-- 不同距离度量
CREATE INDEX ON documents USING hnsw (embedding vector_l2_ops);
CREATE INDEX ON documents USING hnsw (embedding vector_ip_ops);
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);
HNSW 参数详解：
| 参数 | 默认值 | 说明 | 调优建议 |
|---|---|---|---|
m |
16 | 每个节点的最大连接数 | 16-64，越大精度越高但越慢 |
ef_construction |
64 | 构建时的搜索宽度 | 100-400，越大索引质量越好 |
ef_search |
40 | 查询时的搜索宽度（运行时设置） | 40-200，越大精度越高但越慢 |
-- 运行时调整查询精度
SET hnsw.ef_search = 100;
-- 查看当前设置
SHOW hnsw.ef_search;
HNSW 特点：
- ✅ 查询精度高（recall > 95%）
- ✅ 支持增量更新（INSERT 后自动维护索引）
- ✅ 查询速度快
- ❌ 索引构建较慢
- ❌ 内存占用较高（需要存储图结构）
- ❌ 索引大小约为 IVFFlat 的 2-3 倍
4.3 索引选择决策树
需要向量索引？
│
├── 数据量 < 50,000
│ └── 不需要索引，顺序扫描就够快
│
├── 数据量 50,000 - 10,000,000
│ ├── 需要高精度？ → HNSW (m=16, ef_construction=200)
│ ├── 索引构建速度优先？ → IVFFlat
│ └── 内存受限？ → IVFFlat
│
└── 数据量 > 10,000,000
├── 可以接受较高内存？ → HNSW (m=32, ef_construction=400)
└── 内存受限？ → IVFFlat (lists = sqrt(rows))
5. 距离度量与相似性计算
5.1 三种距离度量
pgvector 支持三种核心距离度量，适用于不同的相似性语义。
-- 1. L2 距离（欧氏距离）：向量间的直线距离
SELECT embedding <-> '[1,2,3]'::vector AS l2_distance FROM documents;
-- 2. 余弦距离：向量方向的差异（1 - 余弦相似度）
SELECT embedding <=> '[1,2,3]'::vector AS cosine_distance FROM documents;
-- 3. 内积距离：负内积（-dot_product），适用于已归一化的向量
SELECT embedding <#> '[1,2,3]'::vector AS inner_product_distance FROM documents;
5.2 距离度量选择指南
| 度量 | 运算 | 范围 | 适用场景 | 对应索引操作符 |
|---|---|---|---|---|
| L2（欧氏） | ‖a - b‖² | [0, +∞) | 图像特征匹配、推荐系统 | vector_l2_ops |
| 余弦 | 1 - cos(a,b) | [0, 2] | 文本语义搜索（最常用） | vector_cosine_ops |
| 内积 | -(a · b) | (-∞, +∞] | 已归一化向量、排序场景 | vector_ip_ops |
5.3 相似度转换
-- 余弦距离转相似度（0-1，1 为完全相同）
SELECT
1 - (embedding <=> query_vec) AS cosine_similarity
FROM documents, '[0.1, 0.2, ...]'::vector AS query_vec
ORDER BY embedding <=> query_vec
LIMIT 10;
-- L2 距离转相似度（使用高斯核）
SELECT
EXP(-0.5 * (embedding <-> query_vec)) AS l2_similarity
FROM documents, '[0.1, 0.2, ...]'::vector AS query_vec
ORDER BY embedding <-> query_vec
LIMIT 10;
5.4 向量归一化
-- 创建归一化函数
CREATE OR REPLACE FUNCTION normalize_vector(v vector)
RETURNS vector AS $$
SELECT v / sqrt(sum(x * x))::float
FROM unnest(v::real[]) AS x
$$ LANGUAGE sql IMMUTABLE STRICT PARALLEL SAFE;
-- 使用归一化向量进行内积搜索（等价于余弦搜索）
UPDATE documents SET embedding = normalize_vector(embedding);
CREATE INDEX ON documents USING hnsw (embedding vector_ip_ops);
-- 此时 <#> 等价于 <=>
6. 混合查询：SQL 条件 + 向量检索
pgvector 最大的优势在于可以将向量检索与 SQL 查询无缝结合。这是专用向量数据库难以匹敌的能力。
6.1 基本混合查询
-- 在特定分类中搜索相似文档
SELECT
id,
content,
metadata->>'category' AS category,
1 - (embedding <=> $1::vector) AS similarity
FROM documents
WHERE metadata->>'category' = '技术'
ORDER BY embedding <=> $1::vector
LIMIT 10;
-- 时间范围 + 向量检索
SELECT
id,
content,
created_at,
1 - (embedding <=> $1::vector) AS similarity
FROM documents
WHERE created_at >= NOW() - INTERVAL '7 days'
AND metadata->>'language' = 'zh'
ORDER BY embedding <=> $1::vector
LIMIT 20;
6.2 多条件组合查询
-- 复杂业务查询 + 向量检索
SELECT
d.id,
d.content,
d.metadata->>'author' AS author,
c.name AS category_name,
1 - (d.embedding <=> $1::vector) AS similarity
FROM documents d
JOIN categories c ON d.category_id = c.id
WHERE d.status = 'published'
AND c.name IN ('技术', '科学', '工程')
AND d.created_at >= '2024-01-01'
AND d.metadata->>'visibility' = 'public'
ORDER BY d.embedding <=> $1::vector
LIMIT 20;
6.3 带权重重排序
-- 结合文本相关性和向量相似度
WITH vector_results AS (
SELECT
id,
content,
1 - (embedding <=> $1::vector) AS vector_score,
ts_rank(to_tsvector('chinese', content), plainto_tsquery('chinese', $2)) AS text_score
FROM documents
WHERE to_tsvector('chinese', content) @@ plainto_tsquery('chinese', $2)
OR embedding <=> $1::vector < 0.5
)
SELECT
id,
content,
vector_score,
text_score,
-- 混合评分：向量权重 0.7 + 文本权重 0.3
0.7 * vector_score + 0.3 * text_score AS combined_score
FROM vector_results
ORDER BY combined_score DESC
LIMIT 20;
6.4 分页查询
-- 基于游标的分页（推荐方式）
SELECT
id,
content,
1 - (embedding <=> $1::vector) AS similarity
FROM documents
WHERE id > $2 -- 上一页最后一条的 id
ORDER BY embedding <=> $1::vector
LIMIT 20;
-- 使用 OFFSET 分页（简单但不推荐用于大数据集）
SELECT
id,
content,
1 - (embedding <=> $1::vector) AS similarity
FROM documents
ORDER BY embedding <=> $1::vector
LIMIT 20 OFFSET 40;
6.5 子查询与 CTE
-- 使用 CTE 进行多阶段检索
WITH
-- 阶段 1：粗粒度过滤
candidates AS (
SELECT id, content, embedding
FROM documents
WHERE metadata->>'domain' = 'medical'
AND created_at >= NOW() - INTERVAL '1 year'
),
-- 阶段 2：向量检索
ranked AS (
SELECT
id,
content,
1 - (embedding <=> $1::vector) AS similarity,
ROW_NUMBER() OVER (ORDER BY embedding <=> $1::vector) AS rank
FROM candidates
)
-- 阶段 3：输出
SELECT id, content, similarity
FROM ranked
WHERE rank <= 20;
7. 性能优化与索引调优
7.1 EXPLAIN ANALYZE 解读
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT id, content
FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 10;
-- 输出示例：
-- Limit (cost=0.00..1.23 rows=10 width=40) (actual time=0.5..0.8 rows=10 loops=1)
-- -> Index Scan using documents_embedding_idx on documents (cost=0.00..12345.67 rows=100000 width=40) (actual time=0.5..0.8 rows=10 loops=1)
-- Order By: (embedding <=> '[0.1,0.2,...]'::vector)
-- Planning Time: 0.1 ms
-- Execution Time: 0.9 ms
关键指标：
actual time：实际执行时间（毫秒）rows：扫描/返回的行数- 使用了 Index Scan 说明索引生效
7.2 HNSW 参数调优
-- 查看当前 ef_search
SHOW hnsw.ef_search;
-- 临时增大 ef_search 提高精度（会话级别）
SET hnsw.ef_search = 100;
-- 测试不同 ef_search 的精度和速度
CREATE TEMP TABLE ef_benchmark AS
SELECT
unnest(ARRAY[10, 20, 40, 80, 160]) AS ef_search_val;
-- 对每个 ef_search 值测试查询时间
DO $$
DECLARE
ef_val INT;
start_time TIMESTAMPTZ;
end_time TIMESTAMPTZ;
BEGIN
FOR ef_val IN SELECT ef_search_val FROM ef_benchmark LOOP
EXECUTE format('SET hnsw.ef_search = %s', ef_val);
start_time := clock_timestamp();
PERFORM * FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector LIMIT 10;
end_time := clock_timestamp();
RAISE NOTICE 'ef_search=%, time=%ms', ef_val,
EXTRACT(MILLISECONDS FROM (end_time - start_time));
END LOOP;
END $$;
7.3 索引构建优化
-- 增大 maintenance_work_mem 加速索引构建
SET maintenance_work_mem = '2GB';
-- 并行构建（PostgreSQL 11+）
SET max_parallel_maintenance_workers = 4;
-- 创建索引
CREATE INDEX CONCURRENTLY ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
-- CONCURRENTLY 允许在创建索引时不阻塞写入
7.4 查询优化技巧
-- 1. 使用 LIMIT 限制返回数量（索引利用的关键）
-- ❌ 无 LIMIT，全表扫描
SELECT * FROM documents ORDER BY embedding <=> $1;
-- ✅ 有 LIMIT，使用索引
SELECT * FROM documents ORDER BY embedding <=> $1 LIMIT 10;
-- 2. 预过滤减少向量计算范围
SELECT * FROM documents
WHERE metadata->>'type' = 'article' -- 先过滤
ORDER BY embedding <=> $1::vector -- 再排序
LIMIT 10;
-- 3. 使用 prepared statement 减少解析开销
PREPARE vector_search (vector, int) AS
SELECT id, content, embedding <=> $1 AS distance
FROM documents
ORDER BY embedding <=> $1
LIMIT $2;
EXECUTE vector_search('[0.1,0.2,...]'::vector, 10);
-- 4. 批量查询优化
SELECT
q.query_id,
d.id,
d.content,
d.embedding <=> q.query_vec AS distance
FROM (VALUES
(1, '[0.1,0.2,...]'::vector),
(2, '[0.3,0.4,...]'::vector),
(3, '[0.5,0.6,...]'::vector)
) AS q(query_id, query_vec)
CROSS JOIN LATERAL (
SELECT id, content, embedding
FROM documents
ORDER BY embedding <=> q.query_vec
LIMIT 5
) d
ORDER BY q.query_id, d.embedding <=> q.query_vec;
7.5 索引维护
-- 查看索引大小
SELECT
indexname,
pg_size_pretty(pg_relation_size(indexname::regclass)) AS index_size
FROM pg_indexes
WHERE tablename = 'documents'
AND indexname LIKE '%embedding%';
-- 查看索引使用情况
SELECT
indexrelname AS index_name,
idx_scan AS scans,
idx_tup_read AS tuples_read,
idx_tup_fetch AS tuples_fetched
FROM pg_stat_user_indexes
WHERE relname = 'documents';
-- 定期重建索引（数据大幅变更后）
REINDEX INDEX CONCURRENTLY documents_embedding_idx;
-- 分析表统计信息
ANALYZE documents;
8. 与专用向量数据库对比
8.1 功能对比
| 特性 | pgvector | Milvus | Qdrant | Pinecone | Weaviate |
|---|---|---|---|---|---|
| 部署复杂度 | 极低 | 高（依赖 etcd/MinIO） | 中 | SaaS | 中 |
| 向量索引 | IVFFlat, HNSW | IVF, HNSW, DiskANN | HNSW | 专有 | HNSW |
| SQL 支持 | ✅ 完整 SQL | ❌ | ❌ | ❌ | GraphQL |
| 事务 ACID | ✅ | ❌ | ❌ | ❌ | ❌ |
| 混合查询 | ✅ SQL + 向量 | 有限 | 有限 | 有限 | 有限 |
| 最大向量数 | ~5000 万 | 10 亿+ | 10 亿+ | 10 亿+ | 10 亿+ |
| 查询延迟 (10M) | 5-20ms | 2-10ms | 2-10ms | 2-10ms | 3-15ms |
| 内存效率 | 中 | 高 | 高 | N/A | 中 |
| 云原生 | ✅ | ✅ | ✅ | ✅ | ✅ |
| 社区生态 | 极丰富 | 丰富 | 中 | 中 | 中 |
8.2 性能对比（100 万条 1536 维向量）
| 指标 | pgvector HNSW | Milvus HNSW | Qdrant |
|---|---|---|---|
| 索引构建时间 | ~45 分钟 | ~30 分钟 | ~35 分钟 |
| 索引大小 | ~25 GB | ~18 GB | ~20 GB |
| QPS (top-10) | 800 | 2000 | 1800 |
| P99 延迟 | 15ms | 5ms | 6ms |
| Recall@10 | 97% | 98% | 98% |
⚠️ 以上数据为近似参考值，实际性能受硬件、配置、数据分布影响。
8.3 pgvector 胜出的场景
-- 场景 1：电商商品搜索（结构化过滤 + 语义搜索）
SELECT
p.id,
p.name,
p.price,
p.brand,
1 - (p.embedding <=> $1::vector) AS relevance
FROM products p
JOIN categories c ON p.category_id = c.id
WHERE c.name = '手机'
AND p.price BETWEEN 2000 AND 5000
AND p.stock > 0
AND p.status = 'active'
ORDER BY p.embedding <=> $1::vector
LIMIT 20;
-- 场景 2：用户行为 + 向量推荐
WITH user_history AS (
SELECT DISTINCT product_id
FROM user_interactions
WHERE user_id = $1
AND created_at >= NOW() - INTERVAL '30 days'
)
SELECT
p.id,
p.name,
1 - (p.embedding <=> $2::vector) AS similarity
FROM products p
WHERE p.id NOT IN (SELECT product_id FROM user_history)
AND p.status = 'active'
ORDER BY p.embedding <=> $2::vector
LIMIT 10;
这些场景需要 JOIN、子查询、聚合等关系操作，专用向量数据库要么不支持，要么需要额外开发。
9. RAG 系统集成实战
9.1 完整 RAG 表结构
-- 文档表
CREATE TABLE documents (
id BIGSERIAL PRIMARY KEY,
title TEXT NOT NULL,
content TEXT NOT NULL,
source VARCHAR(500),
metadata JSONB DEFAULT '{}',
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
-- 文档块表（RAG 的核心）
CREATE TABLE document_chunks (
id BIGSERIAL PRIMARY KEY,
document_id BIGINT REFERENCES documents(id) ON DELETE CASCADE,
chunk_index INT NOT NULL,
content TEXT NOT NULL,
token_count INT,
embedding VECTOR(1536), -- OpenAI text-embedding-3-small
metadata JSONB DEFAULT '{}',
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- 创建 HNSW 索引
CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
-- 创建其他常用索引
CREATE INDEX idx_chunks_document_id ON document_chunks(document_id);
CREATE INDEX idx_chunks_metadata ON document_chunks USING gin(metadata);
CREATE INDEX idx_documents_created ON documents(created_at);
9.2 Python 实现
import psycopg2
from psycopg2.extras import Json, execute_values
from openai import OpenAI
import numpy as np
class PgVectorRAG:
"""基于 pgvector 的 RAG 系统"""
def __init__(self, db_url: str, embedding_model: str = "text-embedding-3-small"):
self.conn = psycopg2.connect(db_url)
self.conn.autocommit = True
self.client = OpenAI()
self.embedding_model = embedding_model
self._ensure_tables()
def _ensure_tables(self):
"""确保表结构存在"""
with self.conn.cursor() as cur:
cur.execute("CREATE EXTENSION IF NOT EXISTS vector;")
# ... 创建表的 SQL（同上）
def get_embedding(self, text: str) -> list[float]:
"""获取文本嵌入"""
response = self.client.embeddings.create(
model=self.embedding_model,
input=text
)
return response.data[0].embedding
def get_embeddings_batch(self, texts: list[str]) -> list[list[float]]:
"""批量获取嵌入（减少 API 调用）"""
response = self.client.embeddings.create(
model=self.embedding_model,
input=texts
)
return [item.embedding for item in response.data]
def add_document(self, title: str, content: str, chunk_size: int = 500,
overlap: int = 50, metadata: dict = None) -> int:
"""添加文档并自动分块嵌入"""
with self.conn.cursor() as cur:
# 插入文档
cur.execute(
"INSERT INTO documents (title, content, metadata) VALUES (%s, %s, %s) RETURNING id",
(title, content, Json(metadata or {}))
)
doc_id = cur.fetchone()[0]
# 分块
chunks = self._split_text(content, chunk_size, overlap)
# 批量获取嵌入
embeddings = self.get_embeddings_batch(chunks)
# 批量插入块
chunk_data = [
(doc_id, i, chunk, len(chunk.split()), embedding, Json({}))
for i, (chunk, embedding) in enumerate(zip(chunks, embeddings))
]
execute_values(cur, """
INSERT INTO document_chunks
(document_id, chunk_index, content, token_count, embedding, metadata)
VALUES %s
""", chunk_data, template="(%s, %s, %s, %s, %s::vector, %s)")
return doc_id
def _split_text(self, text: str, chunk_size: int, overlap: int) -> list[str]:
"""简单文本分块"""
words = text.split()
chunks = []
i = 0
while i < len(words):
chunk = ' '.join(words[i:i + chunk_size])
chunks.append(chunk)
i += chunk_size - overlap
return chunks
def search(self, query: str, top_k: int = 5,
filters: dict = None) -> list[dict]:
"""语义检索"""
query_embedding = self.get_embedding(query)
where_clause = ""
params = [query_embedding, top_k]
if filters:
conditions = []
for key, value in filters.items():
if key == "document_id":
conditions.append(f"dc.document_id = %s")
params.append(value)
elif key == "metadata":
for mk, mv in value.items():
conditions.append(f"dc.metadata->>%s = %s")
params.extend([mk, mv])
if conditions:
where_clause = "WHERE " + " AND ".join(conditions)
sql = f"""
SELECT
dc.id,
dc.content,
dc.metadata,
d.title AS document_title,
d.source,
1 - (dc.embedding <=> %s::vector) AS similarity
FROM document_chunks dc
JOIN documents d ON dc.document_id = d.id
{where_clause}
ORDER BY dc.embedding <=> %s::vector
LIMIT %s
"""
with self.conn.cursor() as cur:
cur.execute(sql, params)
results = cur.fetchall()
return [
{
"chunk_id": row[0],
"content": row[1],
"metadata": row[2],
"document_title": row[3],
"source": row[4],
"similarity": float(row[5])
}
for row in results
]
def answer(self, question: str, top_k: int = 5,
model: str = "gpt-4o") -> dict:
"""RAG 问答"""
# 1. 检索相关块
results = self.search(question, top_k=top_k)
# 2. 构建上下文
context = "\n\n---\n\n".join([
f"[来源: {r['document_title']}]\n{r['content']}"
for r in results
])
# 3. 生成回答
response = self.client.chat.completions.create(
model=model,
messages=[
{
"role": "system",
"content": "你是一个基于知识库的问答助手。请根据提供的上下文回答问题。如果上下文中没有相关信息，请说明。"
},
{
"role": "user",
"content": f"""上下文信息：
{context}
问题：{question}
请基于以上上下文回答问题，并在回答末尾标注参考来源。"""
}
]
)
return {
"answer": response.choices[0].message.content,
"sources": [
{
"document": r["document_title"],
"similarity": r["similarity"],
"chunk_id": r["chunk_id"]
}
for r in results
]
}
# 使用示例
rag = PgVectorRAG("postgresql://user:pass@localhost:5432/mydb")
# 添加文档
rag.add_document(
title="pgvector 使用指南",
content="pgvector 是 PostgreSQL 的向量检索扩展...",
metadata={"category": "技术文档", "language": "zh"}
)
# 问答
result = rag.answer("pgvector 支持哪些索引类型？")
print(result["answer"])
print("参考来源:", result["sources"])
9.3 增量同步策略
from datetime import datetime
class IncrementalSync:
"""增量同步文档到向量库"""
def __init__(self, rag: PgVectorRAG):
self.rag = rag
def sync_modified_documents(self, since: datetime):
"""同步指定时间后修改的文档"""
with self.rag.conn.cursor() as cur:
cur.execute("""
SELECT id, title, content, metadata
FROM documents
WHERE updated_at >= %s
""", (since,))
documents = cur.fetchall()
for doc_id, title, content, metadata in documents:
# 删除旧的块
cur.execute("DELETE FROM document_chunks WHERE document_id = %s", (doc_id,))
# 重新分块和嵌入
self.rag.add_document(title, content, metadata=metadata)
print(f"已同步文档: {title}")
print(f"共同步 {len(documents)} 个文档")
10. 分布式方案：Citus + pgvector
对于超大规模向量数据，可以通过 Citus 扩展将 pgvector 分布到多个节点。
10.1 架构概述
┌─────────────────────────────────────┐
│ 协调节点 (Coordinator) │
│ ┌─────────────────────────────────┐│
│ │ 元数据 + 查询路由 + 结果合并 ││
│ └─────────────────────────────────┘│
├─────────────────────────────────────┤
│ 工作节点 (Worker Nodes) │
│ ┌──────────┐ ┌──────────┐ │
│ │ Shard 1 │ │ Shard 2 │ ... │
│ │(pgvector)│ │(pgvector)│ │
│ └──────────┘ └──────────┘ │
└─────────────────────────────────────┘
10.2 分布式表配置
-- 在协调节点上执行
-- 1. 创建分布式表
SELECT create_distributed_table('document_chunks', 'document_id');
-- 2. 确保引用完整性
SELECT create_reference_table('documents');
-- 3. 在每个分片上创建索引（自动传播到所有 worker）
CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
-- 4. 分布式向量查询
SELECT
dc.id,
dc.content,
1 - (dc.embedding <=> $1::vector) AS similarity
FROM document_chunks dc
ORDER BY dc.embedding <=> $1::vector
LIMIT 10;
10.3 分布式查询注意事项
-- ⚠️ 分布式向量查询的工作原理：
-- 1. 协调节点将查询广播到所有 worker
-- 2. 每个 worker 在本地执行向量检索
-- 3. 协调节点合并结果并返回 top-K
-- 这意味着：
-- - 查询延迟 ≈ 最慢 worker 的延迟 + 合并时间
-- - 精度可能略低于集中式（因为每个分片只看到部分数据）
-- - 建议每个分片返回更多的候选结果
-- 配置分布式查询参数
SET citus.multi_shard_modify_mode TO 'sequential';
SET hnsw.ef_search TO 100; -- 增大搜索范围
-- 带过滤条件的分布式查询（高效）
SELECT
dc.id,
dc.content,
1 - (dc.embedding <=> $1::vector) AS similarity
FROM document_chunks dc
JOIN documents d ON dc.document_id = d.id
WHERE d.metadata->>'status' = 'published'
ORDER BY dc.embedding <=> $1::vector
LIMIT 20;
11. 监控与运维
11.1 关键监控指标
-- 1. 索引大小监控
SELECT
schemaname,
tablename,
indexname,
pg_size_pretty(pg_relation_size(indexname::regclass)) AS index_size,
pg_size_pretty(pg_total_relation_size(tablename::regclass)) AS table_total_size
FROM pg_indexes
WHERE indexname LIKE '%embedding%'
ORDER BY pg_relation_size(indexname::regclass) DESC;
-- 2. 查询性能监控
SELECT
query,
calls,
mean_exec_time,
max_exec_time,
total_exec_time,
rows
FROM pg_stat_statements
WHERE query LIKE '%embedding%'
AND query LIKE '%<=>%'
ORDER BY mean_exec_time DESC
LIMIT 10;
-- 3. 表膨胀检查
SELECT
schemaname,
tablename,
n_dead_tup,
n_live_tup,
ROUND(n_dead_tup::numeric / NULLIF(n_live_tup, 0) * 100, 2) AS dead_pct,
last_autovacuum
FROM pg_stat_user_tables
WHERE tablename IN ('documents', 'document_chunks');
-- 4. 缓存命中率
SELECT
schemaname,
tablename,
heap_blks_hit,
heap_blks_read,
ROUND(heap_blks_hit::numeric / NULLIF(heap_blks_hit + heap_blks_read, 0) * 100, 2) AS hit_ratio
FROM pg_statio_user_tables
WHERE tablename IN ('documents', 'document_chunks');
11.2 性能监控函数
-- 向量检索性能基准测试函数
CREATE OR REPLACE FUNCTION benchmark_vector_search(
sample_query vector,
num_runs INT DEFAULT 100,
result_limit INT DEFAULT 10
)
RETURNS TABLE (
avg_time_ms NUMERIC,
p50_time_ms NUMERIC,
p95_time_ms NUMERIC,
p99_time_ms NUMERIC,
qps NUMERIC
) AS $$
DECLARE
times NUMERIC[];
start_ts TIMESTAMPTZ;
end_ts TIMESTAMPTZ;
BEGIN
FOR i IN 1..num_runs LOOP
start_ts := clock_timestamp();
PERFORM * FROM document_chunks
ORDER BY embedding <=> sample_query
LIMIT result_limit;
end_ts := clock_timestamp();
times := array_append(times,
EXTRACT(MILLISECOND FROM (end_ts - start_ts)));
END LOOP;
RETURN QUERY
SELECT
ROUND(AVG(t), 2),
ROUND(percentile_cont(0.5) WITHIN GROUP (ORDER BY t), 2),
ROUND(percentile_cont(0.95) WITHIN GROUP (ORDER BY t), 2),
ROUND(percentile_cont(0.99) WITHIN GROUP (ORDER BY t), 2),
ROUND(1000.0 / AVG(t), 0)
FROM unnest(times) AS t;
END;
$$ LANGUAGE plpgsql;
-- 使用
SELECT * FROM benchmark_vector_search(
(SELECT embedding FROM document_chunks LIMIT 1),
100,
10
);
11.3 自动化运维脚本
#!/bin/bash
# pgvector_maintenance.sh - 定期维护脚本
DB_NAME="your_database"
LOG_FILE="/var/log/pgvector_maintenance.log"
echo "$(date): 开始 pgvector 维护" >> $LOG_FILE
# 1. 更新统计信息
psql -d $DB_NAME -c "ANALYZE document_chunks;" 2>&1 >> $LOG_FILE
# 2. 检查索引膨胀
DEAD_RATIO=$(psql -d $DB_NAME -t -c "
SELECT ROUND(n_dead_tup::numeric / NULLIF(n_live_tup, 0) * 100, 2)
FROM pg_stat_user_tables
WHERE tablename = 'document_chunks';
")
if (( $(echo "$DEAD_RATIO > 10" | bc -l) )); then
echo "$(date): 死元组比例 ${DEAD_RATIO}%，执行 VACUUM" >> $LOG_FILE
psql -d $DB_NAME -c "VACUUM ANALYZE document_chunks;" 2>&1 >> $LOG_FILE
fi
# 3. 检查索引大小
INDEX_SIZE=$(psql -d $DB_NAME -t -c "
SELECT pg_relation_size(indexname::regclass)
FROM pg_indexes
WHERE tablename = 'document_chunks'
AND indexname LIKE '%embedding%'
LIMIT 1;
")
echo "$(date): 向量索引大小: $(echo "scale=2; $INDEX_SIZE/1024/1024/1024" | bc) GB" >> $LOG_FILE
# 4. 性能基准测试
psql -d $DB_NAME -c "
SELECT * FROM benchmark_vector_search(
(SELECT embedding FROM document_chunks ORDER BY random() LIMIT 1),
50, 10
);
" 2>&1 >> $LOG_FILE
echo "$(date): 维护完成" >> $LOG_FILE
12. 企业级部署最佳实践
12.1 硬件配置建议
| 规模 | 向量数量 | CPU | 内存 | 存储 | 说明 |
|---|---|---|---|---|---|
| 小型 | <100 万 | 4 核 | 16 GB | SSD 200 GB | 单机部署 |
| 中型 | 100-1000 万 | 8-16 核 | 64-128 GB | NVMe 1 TB | 单机 + 读副本 |
| 大型 | 1000 万-1 亿 | 32+ 核 | 256+ GB | NVMe RAID | 分布式 / 分区表 |
| 超大型 | >1 亿 | 集群 | 集群 | 集群 | Citus 分布式 |
内存公式估算：
所需内存 ≈ 向量数量 × 维度 × 4 字节（float32）× 1.5（索引开销）
示例：1000 万 × 1536 × 4 × 1.5 ≈ 92 GB
12.2 PostgreSQL 配置优化
# postgresql.conf - pgvector 优化配置
# 内存配置
shared_buffers = '16GB' # 总内存的 25-40%
work_mem = '256MB' # 排序和哈希操作内存
maintenance_work_mem = '2GB' # 索引构建内存
effective_cache_size = '48GB' # 预估可用缓存
# 并行配置
max_parallel_workers_per_gather = 4
max_parallel_workers = 8
max_parallel_maintenance_workers = 4
# 查询优化
random_page_cost = 1.1 # SSD 配置
effective_io_concurrency = 200 # SSD 配置
# HNSW 索引查询精度（会话级或全局）
# hnsw.ef_search = 100 # 在 postgresql.conf 中设置
# 自动清理
autovacuum_vacuum_scale_factor = 0.05 # 更频繁的自动清理
autovacuum_analyze_scale_factor = 0.02
# 日志（监控慢查询）
log_min_duration_statement = 100 # 记录超过 100ms 的查询
12.3 备份与恢复
# 使用 pg_dump 备份（包含向量数据）
pg_dump -Fc -Z 6 -f backup.dump your_database
# 恢复
pg_restore -d your_database backup.dump
# 大规模数据的并行备份恢复
pg_dump -Fc -j 4 -f backup.dump your_database
pg_restore -d your_database -j 4 backup.dump
# 注意：恢复后需要重建索引
psql -d your_database -c "
REINDEX INDEX CONCURRENTLY document_chunks_embedding_idx;
ANALYZE document_chunks;
"
12.4 高可用方案
┌─────────────┐
│ Primary │ ──── Streaming Replication ────→ Replica 1 (只读查询)
│ (读写) │ ──── Streaming Replication ────→ Replica 2 (只读查询)
└─────────────┘
│
▼
pgvector 索引
自动复制到副本
-- 在副本上创建额外索引（不影响主库性能）
-- 副本可以有不同于主库的索引配置
-- 例如：在副本上使用更激进的 HNSW 参数以获得更好的查询性能
12.5 安全配置
-- 1. 创建只读用户（用于查询服务）
CREATE ROLE vector_reader LOGIN PASSWORD 'secure_password';
GRANT CONNECT ON DATABASE your_database TO vector_reader;
GRANT USAGE ON SCHEMA public TO vector_reader;
GRANT SELECT ON document_chunks, documents TO vector_reader;
-- 2. 创建写入用户（用于数据同步服务）
CREATE ROLE vector_writer LOGIN PASSWORD 'secure_password';
GRANT CONNECT ON DATABASE your_database TO vector_writer;
GRANT USAGE ON SCHEMA public TO vector_writer;
GRANT INSERT, UPDATE, DELETE ON document_chunks, documents TO vector_writer;
GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO vector_writer;
-- 3. 行级安全策略（可选）
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON documents
USING (metadata->>'tenant_id' = current_setting('app.tenant_id'));
-- 设置当前租户
SET app.tenant_id = 'tenant_123';
SELECT * FROM documents; -- 自动过滤
12.6 生产环境检查清单
| 检查项 | 命令/方法 | 状态 |
|---|---|---|
| pgvector 版本 | SELECT * FROM pg_extension WHERE extname='vector' |
☐ |
| 索引创建完成 | SELECT * FROM pg_indexes WHERE indexname LIKE '%embedding%' |
☐ |
| HNSW 参数合理 | 检查 m 和 ef_construction | ☐ |
| shared_buffers 足够 | 检查 postgresql.conf | ☐ |
| work_mem 足够 | 检查 postgresql.conf | ☐ |
| 监控配置 | pg_stat_statements 已启用 | ☐ |
| 备份策略 | 自动备份已配置 | ☐ |
| 连接池 | PgBouncer / pgcat 已配置 | ☐ |
| 只读用户 | 已创建并授权 | ☐ |
| 慢查询日志 | log_min_duration_statement 已设置 | ☐ |
13. 总结
pgvector 为 PostgreSQL 生态带来了强大的向量检索能力，使得在单一数据库中同时处理结构化数据和向量数据成为可能。
核心要点回顾
- 安装简单：
CREATE EXTENSION vector;即可开始 - 索引选择：大多数场景推荐 HNSW（精度高、支持增量更新）
- 混合查询：SQL + 向量检索是 pgvector 最大的竞争优势
- RAG 集成：Python + psycopg2 + OpenAI Embeddings 即可构建完整 RAG
- 性能调优：HNSW 参数（m、ef_construction、ef_search）是关键
- 企业级：Citus 分布式 + 读写分离 + 安全配置
推荐技术栈
| 层 | 推荐方案 |
|---|---|
| 嵌入模型 | OpenAI text-embedding-3-small / BGE-large-zh |
| 向量数据库 | pgvector (HNSW) |
| 连接池 | PgBouncer / pgcat |
| ORM | SQLAlchemy + pgvector-python |
| 监控 | pg_stat_statements + Grafana |
| 高可用 | Patroni + Streaming Replication |
学习路径建议
- 入门：Docker 启动 pgvector，跑通基本的 CRUD 和相似性搜索
- 进阶：实现完整的 RAG 系统，掌握混合查询技巧
- 优化：学习 HNSW 参数调优，理解查询计划
- 生产：配置监控、备份、高可用，部署到云数据库
- 扩展：学习 Citus 分布式方案，处理超大规模数据
参考资料
