1. pgvector 核心原理与架构设计

1.1 什么是 pgvector？

pgvector 是 PostgreSQL 的开源扩展（Extension），为世界上最先进的开源关系型数据库添加向量相似度搜索能力。与 Pinecone（SaaS）和 Milvus（独立系统）不同，pgvector 完全集成在 PostgreSQL 内部，复用其存储引擎、事务机制、复制架构和生态工具。

核心定位：

• 零额外基础设施：现有 PostgreSQL 实例直接启用，无需部署新系统
• ACID 事务保障：向量操作与普通数据共享同一事务边界
• SQL 原生支持：用标准 SQL 进行向量操作，无需学习新查询语言
• 生态复用：PgAdmin、DBeaver、连接池、备份工具全部兼容

版本演进：

• 0.1.x-0.4.x：基础功能，ivfflat 索引
• 0.5.x：HNSW 索引引入，性能大幅提升
• 0.6.x：稀疏向量支持，性能优化
• 0.7.x+：迭代式扫描，量化支持，生产级稳定

1.2 架构原理：PostgreSQL 扩展机制

1.2.1 Postgres 扩展架构

PostgreSQL 通过扩展（Extension）机制在不修改核心代码的情况下添加功能：

┌─────────────────────────────────────────┐ │ PostgreSQL 核心进程 │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ Parser │ │ Planner │ │ Executor│ │ │ │ (SQL解析)│ │ (查询计划)│ │ (执行器) │ │ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ │ │ │ │ └────────────┼────────────┘ │ │ ▼ │ │ ┌─────────────┐ │ │ │ Extension │ │ │ │ Manager │ │ │ │ (加载扩展) │ │ │ └──────┬──────┘ │ └──────────────────┼───────────────────────┘ │ ┌──────────┼──────────┐ ▼ ▼ ▼ ┌────────┐ ┌────────┐ ┌────────┐ │PostGIS │ │pgvector│ │TimescaleDB│ │(地理空间)│ │(向量) │ │(时序) │ └────────┘ └────────┘ └────────┘

pgvector 集成点：

• 新数据类型：vector(dim)，halfvec(dim)，sparsevec(dim)
• 新操作符：<->（欧氏距离），<=>（余弦相似度），<#>（内积）
• 新索引类型：ivfflat，hnsw（使用 Postgres 通用索引接口）
• 新函数：vector_norm，vector_dims，l2_distance等

1.2.2 向量存储格式

pgvector 使用 PostgreSQL 的 TOAST（The Oversized-Attribute Storage Technique）机制存储向量：

内存中（处理时）： [0.23, -0.56, 0.89, ..., 0.12] ← float4 数组，4字节/维度 磁盘存储（压缩后）： ┌─────────────────────────────────────────┐ │ VARHDRA (1-4 bytes) │ 维度 (2 bytes) │ float4[] 数据 │ └─────────────────────────────────────────┘ │ ▼ TOAST 压缩（如果 >2KB） │ ▼ 磁盘页（8KB 默认）

存储优化：

• halfvec：2字节/维度（float16），节省 50% 存储，轻微精度损失
• sparsevec：稀疏向量存储，仅存储非零维度（类似 CSR 格式）

1.3 索引算法详解

pgvector 实现了两种主流 ANN 索引，复用 PostgreSQL 的索引接口：

1.3.1 IVFFlat 索引

-- 创建 IVF 索引 CREATE INDEX ON items USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100); -- 聚类中心数

原理：

训练阶段： 1. 从现有数据采样，K-Means 聚类成 `lists` 个中心 2. 每个向量分配到最近的中心（倒排列表） 3. 为每个列表构建独立的向量数组 查询阶段： 1. 找到查询向量最近的 `probes` 个中心（默认 1） 2. 在这些中心的列表中暴力搜索 3. 合并结果返回 top_k

参数调优：

特点：

• 构建快，适合批量导入后不再更新的场景
• 查询速度中等，召回率依赖 probes 参数
• 不支持增量更新（需要重建）

1.3.2 HNSW 索引（推荐）

-- 创建 HNSW 索引（pgvector 0.5.0+） CREATE INDEX ON items USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64);

原理：

分层可导航小世界图： - 第 0 层：密集连接的基础图（所有节点） - 第 1 层：随机采样 50% 节点构建上层图 - 第 2 层：再采样 50%，共 log(N) 层 构建阶段： 1. 随机插入节点，计算与现有节点的距离 2. 为每层选择 M 个最近邻居建立连接 3. 动态维护图的连通性 查询阶段： 1. 从顶层随机节点开始贪心搜索 2. 找到局部最优后下沉到下一层 3. 在底层精细搜索，ef 控制搜索宽度

参数调优：

特点：

• 查询极快，召回率高（95%+）
• 支持增量插入（实时更新）
• 构建较慢，内存占用高（约 2 倍原始数据）

1.3.3 索引操作符类

-- 根据距离度量选择操作符类 CREATE INDEX ON items USING hnsw ( embedding vector_cosine_ops -- 余弦相似度（推荐文本） -- embedding vector_l2_ops -- 欧氏距离（推荐图像） -- embedding vector_ip_ops -- 内积（推荐推荐系统） -- embedding vector_hamming_ops -- 汉明距离（二进制向量） );

1.4 查询执行流程

-- 示例查询 SELECT id, content, embedding <=> '[0.1, 0.2, ...]'::vector AS distance FROM documents WHERE category = 'tech' AND published > '2024-01-01' ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector LIMIT 10;

执行计划：

┌─────────────────────────────────────────────────────────┐ │ Limit (cost=100.00..100.10 rows=10) │ │ -> Index Scan using documents_embedding_idx │ │ Index Cond: (embedding <=> '[...]'::vector) │ │ Order By: (embedding <=> '[...]'::vector) │ │ Filter: ((category = 'tech') AND (published > │ │ '2024-01-01'::date)) │ └─────────────────────────────────────────────────────────┘

优化器决策：

• 如果category选择性高，可能先扫 B-Tree 索引再向量排序
• 如果向量过滤后数据少，直接使用 HNSW 索引
• LIMIT推动到索引层，避免全表扫描

2. 部署实战：从开发到生产

2.1 开发环境：Docker 快速启动

# 方式 1：官方镜像（推荐） docker run --name pgvector-dev \ -e POSTGRES_PASSWORD=postgres \ -p 5432:5432 \ -d pgvector/pgvector:pg16 # 进入容器启用扩展 docker exec -it pgvector-dev psql -U postgres -c "CREATE EXTENSION vector;" # 方式 2：Docker Compose（完整开发栈） cat > docker-compose.yml <<'EOF' version: '3.8' services: postgres: image: pgvector/pgvector:pg16 environment: POSTGRES_USER: dev POSTGRES_PASSWORD: devpass POSTGRES_DB: vectordb ports: - "5432:5432" volumes: - pgdata:/var/lib/postgresql/data - ./init.sql:/docker-entrypoint-initdb.d/init.sql command: > postgres -c shared_preload_libraries='pg_stat_statements' -c max_connections=200 pgadmin: image: dpage/pgadmin4:latest environment: PGADMIN_DEFAULT_EMAIL: admin@example.com PGADMIN_DEFAULT_PASSWORD: admin ports: - "8080:80" depends_on: - postgres volumes: pgdata: EOF docker-compose up -d

2.2 生产环境：源码编译安装

场景：需要特定 PostgreSQL 版本，或优化编译参数

# 1. 安装依赖 sudo apt-get update sudo apt-get install -y postgresql-server-dev-16 build-essential git # 2. 下载源码 cd /tmp git clone --branch v0.7.0 https://github.com/pgvector/pgvector.git cd pgvector # 3. 编译安装 make OPTFLAGS="-march=native -O3" # 针对 CPU 优化 sudo make install # 4. 启用扩展 sudo -u postgres psql -c "CREATE EXTENSION vector;" # 5. 验证 psql -c "SELECT * FROM pg_available_extensions WHERE name = 'vector';"

2.3 云托管服务配置

2.3.1 AWS RDS/Aurora

-- RDS 支持 pgvector（需特定版本） -- 创建参数组启用 -- 通过 RDS 控制台或 CLI 修改 -- 连接后启用 CREATE EXTENSION IF NOT EXISTS vector; -- 查看版本 SELECT * FROM pg_extension WHERE extname = 'vector';

限制：

• 不支持 HNSW 索引的某些高级参数调整
• 无法自定义编译优化
• 依赖 RDS 版本发布节奏

2.3.2 Supabase/Neon 等 Serverless Postgres

-- Supabase 已预装 pgvector -- 直接在 SQL Editor 执行： CREATE EXTENSION vector; -- 向量列限制：Supabase 免费版有 500MB 限制 -- 建议：大向量表使用单独的项目

2.4 连接池与高性能配置

# postgresql.conf 优化（生产环境） # 内存配置（假设 64GB 内存服务器） shared_buffers = 16GB # 25% 内存 effective_cache_size = 48GB # 75% 内存 work_mem = 256MB # 复杂排序/哈希操作 maintenance_work_mem = 2GB # 索引构建 # 并发配置 max_connections = 200 max_parallel_workers_per_gather = 4 max_parallel_workers = 8 # WAL 配置（SSD 优化） wal_buffers = 16MB min_wal_size = 1GB max_wal_size = 4GB checkpoint_completion_target = 0.9 # 查询计划 random_page_cost = 1.1 # SSD 设置 1.1，HDD 默认 4 effective_io_concurrency = 200 # SSD 并发数 # pgvector 特定 # 无特殊 GUC 参数，依赖 Postgres 基础优化

连接池（PgBouncer）：

; pgbouncer.ini [databases] vectordb = host=localhost port=5432 dbname=vectordb [pgbouncer] listen_port = 6432 listen_addr = 0.0.0.0 auth_type = md5 max_client_conn = 10000 default_pool_size = 25 reserve_pool_size = 5 pool_mode = transaction # 推荐事务级池化

3. SQL 操作完全指南

3.1 数据类型与基础操作

-- ==================== 数据类型定义 ==================== -- 稠密向量（最常用） CREATE TABLE documents ( id SERIAL PRIMARY KEY, content TEXT, embedding vector(1536), -- 1536 维 float4 向量 metadata JSONB -- 灵活元数据 ); -- 半精度向量（节省存储） CREATE TABLE images ( id SERIAL PRIMARY KEY, url TEXT, embedding halfvec(512), -- 512 维 float2，省 50% 空间 category VARCHAR(50) ); -- 稀疏向量（关键词权重） CREATE TABLE keywords ( id SERIAL PRIMARY KEY, doc_id INTEGER REFERENCES documents(id), embedding sparsevec(100000), -- 10万维，仅存储非零值 term TEXT ); -- ==================== 向量构造与转换 ==================== -- 从数组构造 SELECT '[1,2,3]'::vector(3); SELECT ARRAY[0.1, 0.2, 0.3]::vector(3); -- 维度检查（严格） SELECT '[1,2]'::vector(3); -- 错误：维度不匹配 -- 类型转换 SELECT embedding::halfvec FROM documents; -- 降精度 SELECT embedding::vector FROM images; -- 升精度 -- ==================== 距离计算 ==================== -- 欧氏距离（L2） SELECT embedding <-> '[0.1, 0.2, ...]'::vector AS distance FROM documents; -- 余弦距离（1 - 余弦相似度，范围 [0, 2]） SELECT embedding <=> '[0.1, 0.2, ...]'::vector AS distance FROM documents; -- 内积（负值表示角度 > 90°） SELECT embedding <#> '[0.1, 0.2, ...]'::vector AS dot_product FROM documents; -- 负内积（用于最大内积搜索） SELECT embedding <+> '[0.1, 0.2, ...]'::vector FROM documents; -- 汉明距离（二进制向量） SELECT embedding <~> '[1,0,1,...]'::bit(10) FROM binary_items; -- ==================== 向量属性 ==================== SELECT vector_dims(embedding) AS dimensions, -- 维度 vector_norm(embedding) AS magnitude, -- 模长 embedding::text AS string_representation -- 文本表示 FROM documents;

3.2 表设计与索引策略

-- ==================== 生产级表设计 ==================== -- 分区表（按时间分区，适合日志/时序数据） CREATE TABLE document_chunks ( id BIGSERIAL, doc_id INTEGER NOT NULL, chunk_index INTEGER NOT NULL, content TEXT NOT NULL, embedding vector(1536) NOT NULL, created_at TIMESTAMP DEFAULT NOW(), metadata JSONB DEFAULT '{}', PRIMARY KEY (id, created_at) ) PARTITION BY RANGE (created_at); -- 创建分区 CREATE TABLE document_chunks_2024_q1 PARTITION OF document_chunks FOR VALUES FROM ('2024-01-01') TO ('2024-04-01'); CREATE TABLE document_chunks_2024_q2 PARTITION OF document_chunks FOR VALUES FROM ('2024-04-01') TO ('2024-07-01'); -- 索引（每个分区独立创建） CREATE INDEX ON document_chunks_2024_q1 USING hnsw (embedding vector_cosine_ops); CREATE INDEX ON document_chunks_2024_q2 USING hnsw (embedding vector_cosine_ops); -- 普通 B-Tree 索引（元数据过滤） CREATE INDEX ON document_chunks (doc_id); CREATE INDEX ON document_chunks USING gin (metadata); -- JSONB 索引 -- ==================== 索引策略选择 ==================== -- 策略 1：纯 HNSW（小数据量 <100万，高并发查询） CREATE INDEX idx_hnsw ON documents USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64); -- 策略 2：HNSW + 条件索引（分区查询） CREATE INDEX idx_hnsw_tech ON documents USING hnsw (embedding vector_cosine_ops) WHERE category = 'tech'; -- 策略 3：IVFFlat（大数据量，批量导入，少更新） CREATE INDEX idx_ivf ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100); -- 策略 4：混合检索（全文 + 向量） -- 需要额外安装 pg_trgm 或 pg_search CREATE INDEX idx_content_gin ON documents USING gin (content gin_trgm_ops); CREATE INDEX idx_embedding ON documents USING hnsw (embedding vector_cosine_ops); -- ==================== 索引维护 ==================== -- 查看索引大小 SELECT pg_size_pretty(pg_relation_size('idx_hnsw')); -- 重建索引（HNSW 支持并发重建） REINDEX INDEX CONCURRENTLY idx_hnsw; -- 删除并重建（IVF 需要） DROP INDEX idx_ivf; CREATE INDEX idx_ivf ON documents USING ivfflat (embedding vector_cosine_ops); -- 统计信息更新 ANALYZE documents;

3.3 查询操作详解

-- ==================== 基础相似度搜索 ==================== -- 最近邻搜索（KNN） SELECT id, content, embedding <=> '[0.023, -0.456, ...]'::vector AS distance FROM documents ORDER BY embedding <=> '[0.023, -0.456, ...]'::vector LIMIT 10; -- 使用绑定参数（应用开发） PREPARE knn_query (vector) AS SELECT id, content, embedding <=> $1 AS distance FROM documents ORDER BY embedding <=> $1 LIMIT 10; EXECUTE knn_query ('[0.1, 0.2, ...]'::vector); -- ==================== 混合查询（向量 + 元数据） ==================== -- 方式 1：先元数据过滤，再向量排序（选择性高时） SELECT id, content, embedding <=> '[...]'::vector AS distance FROM documents WHERE category = 'tech' AND published > '2024-01-01' AND metadata->>'author' = 'Alice' ORDER BY embedding <=> '[...]'::vector LIMIT 10; -- 方式 2：向量索引扫描 + 过滤（向量选择性强时） -- 优化器自动选择，或强制使用索引 SET enable_seqscan = off; SELECT id, content, embedding <=> '[...]'::vector AS distance FROM documents WHERE embedding <=> '[...]'::vector < 0.3 -- 距离阈值 AND category = 'tech' ORDER BY embedding <=> '[...]'::vector LIMIT 10; -- ==================== 范围查询 ==================== -- 距离阈值搜索（返回所有相似度 > 0.8 的） SELECT id, 1 - (embedding <=> '[...]'::vector) / 2 AS cosine_similarity FROM documents WHERE embedding <=> '[...]'::vector < 0.4 -- 余弦距离 < 0.4 即相似度 > 0.8 ORDER BY embedding <=> '[...]'::vector; -- ==================== 批量查询 ==================== -- 使用 LATERAL 进行多查询批量处理 WITH queries AS ( SELECT 1 AS qid, '[0.1, ...]'::vector AS vec UNION ALL SELECT 2, '[0.2, ...]'::vector ) SELECT q.qid, d.id, d.content, d.embedding <=> q.vec AS distance FROM queries q LEFT JOIN LATERAL ( SELECT * FROM documents ORDER BY embedding <=> q.vec LIMIT 5 ) d ON true; -- ==================== 聚合与分组 ==================== -- 按类别统计平均相似度 SELECT category, AVG(embedding <=> '[...]'::vector) AS avg_distance, COUNT(*) AS count FROM documents GROUP BY category; -- 找出每个类别最接近的文档 SELECT DISTINCT ON (category) category, id, content, embedding <=> '[...]'::vector AS distance FROM documents ORDER BY category, embedding <=> '[...]'::vector; -- ==================== 向量运算 ==================== -- 向量加法（语义组合："国王" - "男人" + "女人" ≈ "女王"） SELECT d1.embedding - d2.embedding + d3.embedding AS combined_vector FROM documents d1, documents d2, documents d3 WHERE d1.content = 'king' AND d2.content = 'man' AND d3.content = 'woman'; -- 使用结果进行搜索 WITH combined AS ( SELECT embedding - '[...]'::vector + '[...]'::vector AS vec FROM documents WHERE id = 1 ) SELECT d.id, d.content FROM combined c CROSS JOIN LATERAL ( SELECT * FROM documents ORDER BY embedding <=> c.vec LIMIT 5 ) d;

3.4 事务与并发控制

-- ==================== ACID 事务示例 ==================== BEGIN; -- 插入文档和向量 INSERT INTO documents (content, embedding, metadata) VALUES ( 'PostgreSQL 向量数据库', '[0.1, 0.2, ...]'::vector, '{"author": "Bob", "tags": ["database", "ai"]}' ) RETURNING id; -- 同时更新相关统计表 UPDATE doc_stats SET count = count + 1 WHERE category = 'database'; -- 向量搜索在同一事务中立即可见 SELECT * FROM documents ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector LIMIT 5; COMMIT; -- ==================== 并发插入优化 ==================== -- 高并发插入时使用批量提交 -- 应用层代码逻辑： -- 1. 积累 1000 条记录 -- 2. 单条 INSERT 多值 -- 3. 提交事务 INSERT INTO documents (content, embedding) VALUES ('doc1', '[...]'::vector), ('doc2', '[...]'::vector), ...; -- ==================== 悲观锁与向量更新 ==================== -- 更新向量（先删后插的替代方案，HNSW 支持原地更新） BEGIN; SELECT * FROM documents WHERE id = 100 FOR UPDATE; UPDATE documents SET embedding = '[new vector]'::vector, updated_at = NOW() WHERE id = 100; COMMIT;

4. 编程语言集成与工具连接

4.1 Python 生态：psycopg2 + pgvector

pip install psycopg2-binary pgvector

import psycopg2 from pgvector.psycopg2 import register_vector import numpy as np from openai import OpenAI # ==================== 连接与初始化 ==================== class PgvectorClient: def __init__(self, dsn="postgresql://dev:devpass@localhost:5432/vectordb"): self.conn = psycopg2.connect(dsn) self.conn.autocommit = False # 手动事务控制 register_vector(self.conn) # 注册 vector 类型适配器 def init_schema(self): """初始化数据库 schema""" with self.conn.cursor() as cur: # 启用扩展 cur.execute("CREATE EXTENSION IF NOT EXISTS vector;") # 创建表 cur.execute(""" CREATE TABLE IF NOT EXISTS documents ( id SERIAL PRIMARY KEY, title VARCHAR(500), content TEXT NOT NULL, embedding vector(1536), metadata JSONB DEFAULT '{}', created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() ); """) # 创建 HNSW 索引（如果不存在） cur.execute(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM pg_indexes WHERE indexname = 'documents_embedding_idx' ) THEN CREATE INDEX documents_embedding_idx ON documents USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64); END IF; END $$; """) # 创建触发器自动更新 updated_at cur.execute(""" CREATE OR REPLACE FUNCTION update_updated_at_column() RETURNS TRIGGER AS $$ BEGIN NEW.updated_at = NOW(); RETURN NEW; END; $$ language 'plpgsql'; DROP TRIGGER IF EXISTS update_documents_updated_at ON documents; CREATE TRIGGER update_documents_updated_at BEFORE UPDATE ON documents FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); """) self.conn.commit() print("Schema 初始化完成") # ==================== 数据操作 ==================== def insert_document(self, title: str, content: str, embedding: list, metadata: dict = None): """插入单条文档""" with self.conn.cursor() as cur: cur.execute(""" INSERT INTO documents (title, content, embedding, metadata) VALUES (%s, %s, %s, %s) RETURNING id; """, (title, content, embedding, metadata)) doc_id = cur.fetchone()[0] self.conn.commit() return doc_id def batch_insert(self, documents: list, batch_size: 1000): """批量插入（高效）""" from psycopg2.extras import execute_values total = len(documents) for i in range(0, total, batch_size): batch = documents[i:i+batch_size] with self.conn.cursor() as cur: execute_values( cur, """ INSERT INTO documents (title, content, embedding, metadata) VALUES %s """, [(d['title'], d['content'], d['embedding'], d.get('metadata', {})) for d in batch], template="(%s, %s, %s::vector, %s::jsonb)" ) self.conn.commit() print(f"已插入 {min(i+batch_size, total)}/{total}") return total # ==================== 向量搜索 ==================== def search_similar(self, query_embedding: list, top_k: int = 5, filters: dict = None, min_similarity: float = None): """相似度搜索""" with self.conn.cursor() as cur: # 构建查询 sql = """ SELECT id, title, content, 1 - (embedding <=> %s::vector) / 2 AS similarity, metadata FROM documents WHERE 1=1 """ params = [query_embedding] # 动态过滤条件 if filters: for key, value in filters.items(): if isinstance(value, list): sql += f" AND metadata->>'{key}' = ANY(%s)" params.append(value) else: sql += f" AND metadata->>'{key}' = %s" params.append(value) # 相似度阈值 if min_similarity: max_distance = 2 * (1 - min_similarity) # 转换余弦相似度到距离 sql += " AND embedding <=> %s::vector < %s" params.extend([query_embedding, max_distance]) sql += " ORDER BY embedding <=> %s::vector LIMIT %s" params.extend([query_embedding, top_k]) cur.execute(sql, params) results = cur.fetchall() return [ { "id": r[0], "title": r[1], "content": r[2][:200] + "..." if len(r[2]) > 200 else r[2], "similarity": round(r[3], 4), "metadata": r[4] } for r in results ] # ==================== 混合全文搜索 ==================== def hybrid_search(self, query_text: str, query_embedding: list, top_k: int = 10, text_weight: float = 0.3): """结合全文和向量搜索（需要 pg_trgm）""" with self.conn.cursor() as cur: cur.execute(""" WITH vector_scores AS ( SELECT id, 1 - (embedding <=> %s::vector) / 2 AS v_score, ROW_NUMBER() OVER (ORDER BY embedding <=> %s::vector) AS v_rank FROM documents ORDER BY embedding <=> %s::vector LIMIT %s * 3 ), text_scores AS ( SELECT id, similarity(content, %s) AS t_score, ROW_NUMBER() OVER (ORDER BY similarity(content, %s) DESC) AS t_rank FROM documents WHERE content %% %s -- 相似度操作符 ORDER BY similarity(content, %s) DESC LIMIT %s * 3 ) SELECT COALESCE(v.id, t.id) AS id, d.title, d.content, COALESCE(v.v_score, 0) * (1 - %s) + COALESCE(t.t_score, 0) * %s AS combined_score FROM vector_scores v FULL OUTER JOIN text_scores t ON v.id = t.id JOIN documents d ON d.id = COALESCE(v.id, t.id) ORDER BY combined_score DESC LIMIT %s; """, ( query_embedding, query_embedding, query_embedding, top_k, query_text, query_text, query_text, query_text, top_k, text_weight, text_weight, top_k )) return cur.fetchall() def close(self): self.conn.close() # ==================== RAG 应用完整示例 ==================== def rag_example(): # 初始化 client = PgvectorClient() client.init_schema() # 嵌入模型 openai = OpenAI() # 准备数据 docs = [ "PostgreSQL 是世界上最先进的开源关系型数据库", "pgvector 为 Postgres 添加了向量相似度搜索功能", "向量数据库是 AI 应用的基础设施", "HNSW 索引提供高效的近似最近邻搜索" ] # 生成嵌入并插入 for doc in docs: response = openai.embeddings.create( model="text-embedding-3-small", input=doc ) embedding = response.data[0].embedding client.insert_document( title=doc[:50], content=doc, embedding=embedding, metadata={"source": "manual", "category": "tech"} ) # 查询 query = "什么是向量数据库？" query_emb = openai.embeddings.create( model="text-embedding-3-small", input=query ).data[0].embedding results = client.search_similar( query_embedding=query_emb, top_k=3, min_similarity=0.7 ) print("搜索结果：") for r in results: print(f" [{r['similarity']:.3f}] {r['content']}") client.close() if __name__ == "__main__": rag_example()

4.2 Node.js 集成：pg + pgvector

npm install pg pgvector

const { Pool } = require('pg'); const pgvector = require('pgvector/pg'); class PgvectorService { constructor() { this.pool = new Pool({ host: 'localhost', port: 5432, database: 'vectordb', user: 'dev', password: 'devpass', max: 20, // 连接池大小 idleTimeoutMillis: 30000 }); // 注册 vector 类型 this.pool.on('connect', async (client) => { await pgvector.registerType(client); }); } async init() { const client = await this.pool.connect(); try { // 启用扩展 await client.query('CREATE EXTENSION IF NOT EXISTS vector'); // 创建表 await client.query(` CREATE TABLE IF NOT EXISTS items ( id SERIAL PRIMARY KEY, name TEXT NOT NULL, embedding vector(1536), attributes JSONB DEFAULT '{}', created_at TIMESTAMP DEFAULT NOW() ) `); // 检查并创建索引 const indexExists = await client.query(` SELECT 1 FROM pg_indexes WHERE indexname = 'items_embedding_idx' `); if (indexExists.rows.length === 0) { await client.query(` CREATE INDEX items_embedding_idx ON items USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64) `); console.log('HNSW 索引创建成功'); } } finally { client.release();