14.08-PGVector实践

要点

  • pgvector 是 PostgreSQL 的向量检索扩展——复用现有数据库，不需要新增组件
  • 安装简单，SQL 语法熟悉，对 PG 用户几乎零学习成本
  • 两种索引：IVFFlat（适合大规模）和 HNSW（查询快、内存占用高）
  • 结合 PG 的事务、JSON、全文检索能力，可以做混合检索
  • 规模限制：百万级以下表现好，更大规模性能下降明显
  • 适合「已经有 PostgreSQL」的团队作为 RAG 起步方案

1. 安装

本地开发

# macOS
brew install pgvector
 
# Ubuntu/Debian
sudo apt install postgresql-16-pgvector
 
# Docker
docker run -d --name pgvector -e POSTGRES_PASSWORD=password -p 5432:5432 pgvector/pgvector:pg16

启用扩展

CREATE EXTENSION IF NOT EXISTS vector;

验证安装：

SELECT * FROM pg_extension WHERE extname = 'vector';

2. 表设计

-- 文档向量表
CREATE TABLE document_chunks (
  id TEXT PRIMARY KEY,                    -- 格式：{docId}-chunk-{index}
  document_id TEXT NOT NULL,
  document_title TEXT NOT NULL,
  chunk_index INTEGER NOT NULL,
  content TEXT NOT NULL,                  -- chunk 原文
  embedding VECTOR(768) NOT NULL,         -- 向量，维度要和 embedding 模型匹配
  metadata JSONB DEFAULT '{}',            -- 额外元数据
  user_id TEXT,                           -- 所属用户（权限控制）
  tenant_id TEXT,                         -- 所属租户
  created_at TIMESTAMP DEFAULT NOW(),
 
  -- 复合索引：按文档查 chunk
  UNIQUE (document_id, chunk_index)
);
 
-- 按文档查 chunk 的索引
CREATE INDEX idx_chunks_document ON document_chunks(document_id);
 
-- 按用户查 chunk 的索引（权限过滤）
CREATE INDEX idx_chunks_user ON document_chunks(user_id);

3. 向量索引

pgvector 支持两种向量索引，各有取舍。

IVFFlat（倒排文件索引）

CREATE INDEX idx_chunks_embedding_ivf
  ON document_chunks
  USING ivfflat (embedding vector_cosine_ops)
  WITH (lists = 100);
  • lists 参数：聚类数，越大检索越精确但越慢
  • 经验值：数据量 < 100 万用 sqrt(n)，> 100 万用 4 * sqrt(n)
  • 优点：内存占用小，构建快
  • 缺点：需要训练（先有数据才能建索引）

HNSW（分层导航小世界）

CREATE INDEX idx_chunks_embedding_hnsw
  ON document_chunks
  USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64);
  • m：每层每个节点的连接数
  • ef_construction：构建时搜索范围
  • 优点：查询快，不需要训练
  • 缺点：内存占用高，构建慢

选择建议

  • 数据量 < 100 万、查询频繁 → HNSW
  • 数据量大、内存有限 → IVFFlat
  • 不确定 → 先用 HNSW，看内存占用再决定

4. 写入向量

// src/services/rag/pgvector-store.ts
import { Pool } from 'pg'
 
const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
})
 
// 格式化向量（pg 要求 [1,2,3] 格式）
function formatVector(vector: number[]): string {
  return `[${vector.join(',')}]`
}
 
export async function upsertChunks(chunks: Array<{
  id: string
  documentId: string
  documentTitle: string
  chunkIndex: number
  content: string
  embedding: number[]
  metadata: Record<string, unknown>
  userId?: string
  tenantId?: string
}>): Promise<void> {
  const client = await pool.connect()
  try {
    await client.query('BEGIN')
 
    for (const chunk of chunks) {
      await client.query(
        `INSERT INTO document_chunks
         (id, document_id, document_title, chunk_index, content, embedding, metadata, user_id, tenant_id)
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
         ON CONFLICT (id) DO UPDATE SET
           content = EXCLUDED.content,
           embedding = EXCLUDED.embedding,
           metadata = EXCLUDED.metadata`,
        [
          chunk.id,
          chunk.documentId,
          chunk.documentTitle,
          chunk.chunkIndex,
          chunk.content,
          formatVector(chunk.embedding),
          JSON.stringify(chunk.metadata),
          chunk.userId,
          chunk.tenantId,
        ]
      )
    }
 
    await client.query('COMMIT')
  } catch (err) {
    await client.query('ROLLBACK')
    throw err
  } finally {
    client.release()
  }
}

批量写入优化：

// 用 COPY 批量写入（更快）
import { format } from 'pg-copy-streams'
 
export async function bulkUpsertChunks(chunks: Chunk[]): Promise<void> {
  const client = await pool.connect()
  try {
    // 创建临时表
    await client.query(`
      CREATE TEMP TABLE temp_chunks (LIKE document_chunks) ON COMMIT DROP
    `)
 
    // COPY 写入临时表
    const stream = client.query(copy.from(`
      COPY temp_chunks (id, document_id, document_title, chunk_index, content, embedding, metadata)
      FROM STDIN CSV
    `))
 
    for (const chunk of chunks) {
      stream.write([
        chunk.id,
        chunk.documentId,
        chunk.documentTitle,
        chunk.chunkIndex,
        chunk.content.replace(/"/g, '""'),  // CSV 转义
        formatVector(chunk.embedding),
        JSON.stringify(chunk.metadata),
      ].join(',') + '\n')
    }
    stream.end()
 
    // 合并到主表
    await client.query(`
      INSERT INTO document_chunks
      SELECT * FROM temp_chunks
      ON CONFLICT (id) DO UPDATE SET
        content = EXCLUDED.content,
        embedding = EXCLUDED.embedding,
        metadata = EXCLUDED.metadata
    `)
  } finally {
    client.release()
  }
}

5. 相似度查询

export async function searchChunks(
  queryVector: number[],
  options: {
    topK?: number
    threshold?: number       // 最低相似度阈值
    userId?: string          // 权限过滤
    tenantId?: string        // 租户过滤
    documentId?: string      // 文档过滤
  }
): Promise<SearchResult[]> {
  const { topK = 5, threshold = 0.7, userId, tenantId, documentId } = options
 
  const conditions: string[] = []
  const params: unknown[] = [formatVector(queryVector)]
  let paramIndex = 2
 
  if (userId) {
    conditions.push(`user_id = $${paramIndex++}`)
    params.push(userId)
  }
  if (tenantId) {
    conditions.push(`tenant_id = $${paramIndex++}`)
    params.push(tenantId)
  }
  if (documentId) {
    conditions.push(`document_id = $${paramIndex++}`)
    params.push(documentId)
  }
 
  const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : ''
 
  const query = `
    SELECT
      id,
      document_id,
      document_title,
      chunk_index,
      content,
      metadata,
      1 - (embedding <=> $1) AS similarity
    FROM document_chunks
    ${whereClause}
    ${whereClause ? 'AND' : 'WHERE'} 1 - (embedding <=> $1) >= $${paramIndex++}
    ORDER BY embedding <=> $1
    LIMIT $${paramIndex++}
  `
  params.push(threshold, topK)
 
  const result = await pool.query(query, params)
 
  return result.rows.map((row) => ({
    id: row.id,
    documentId: row.document_id,
    documentTitle: row.document_title,
    chunkIndex: row.chunk_index,
    content: row.content,
    metadata: row.metadata,
    similarity: row.similarity,
  }))
}

&lt;=&gt; 是余弦距离操作符（1 - cosine similarity）。1 - (embedding &lt;=&gt; $1) 得到余弦相似度。

6. 混合检索

pgvector 可以和 PostgreSQL 的全文检索结合，做混合检索。

-- 添加 tsvector 列
ALTER TABLE document_chunks ADD COLUMN content_tsv TSVECTOR;
 
-- 自动生成 tsvector
CREATE INDEX idx_chunks_tsv ON document_chunks USING gin(content_tsv);
 
-- 更新触发器：内容变化时自动更新 tsvector
CREATE TRIGGER update_content_tsv
  BEFORE INSERT OR UPDATE OF content ON document_chunks
  FOR EACH ROW
  EXECUTE FUNCTION tsvector_update_trigger(content_tsv, 'simple', content);
 
-- 混合查询：向量相似度 + 关键词匹配
SELECT
  id,
  content,
  1 - (embedding <=> $1) AS vector_score,
  ts_rank(content_tsv, plainto_tsquery('simple', $2)) AS keyword_score,
  -- 融合分数
  0.7 * (1 - (embedding <=> $1)) + 0.3 * ts_rank(content_tsv, plainto_tsquery('simple', $2)) AS combined_score
FROM document_chunks
WHERE content_tsv @@ plainto_tsquery('simple', $2)  -- 必须有关键词匹配
   OR 1 - (embedding <=> $1) >= 0.8                 -- 或者向量相似度够高
ORDER BY combined_score DESC
LIMIT 5

混合检索的好处：

  • 关键词检索精确匹配专有名词（产品名、人名、订单号）
  • 向量检索捕捉语义相似（「退款」和「返款」）

7. 删除文档

删除文档时需要清理所有相关 chunk。

export async function deleteDocument(documentId: string): Promise<void> {
  await pool.query('DELETE FROM document_chunks WHERE document_id = $1', [documentId])
}

8. 性能调优

查询参数

-- HNSW 查询时的搜索范围
SET hnsw.ef_search = 100;  -- 默认 40，越大越精确但越慢
 
-- IVFFlat 查询时的探测列表数
SET ivfflat.probes = 10;  -- 默认 1，越大越精确但越慢

可以在查询前动态设置：

await client.query('SET LOCAL hnsw.ef_search = 100')
const result = await client.query(searchQuery, params)

VACUUM 和 ANALYZE

向量索引需要定期 VACUUM 和 ANALYZE：

VACUUM ANALYZE document_chunks;

删除大量数据后尤其需要。

内存配置

-- PostgreSQL 配置
shared_buffers = '4GB'         -- 系统内存的 25%
effective_cache_size = '12GB'  -- 系统内存的 75%
maintenance_work_mem = '2GB'   -- 索引构建用

9. 扩展性限制

pgvector 的规模限制：

向量数性能建议
< 10 万优秀正常使用
10 万 - 100 万良好HNSW 索引，足够内存
100 万 - 1000 万一般考虑分区或迁移
> 1000 万较差迁移到专用向量数据库

突破限制的方案：

  • 表分区：按 tenant_id 或 document 分区
  • 读写分离：查询走只读副本
  • 迁移：当数据量超过阈值，迁移到 Qdrant / Milvus

10. 完整示例

// src/services/rag/pgvector-store.ts（完整）
import { Pool } from 'pg'
 
const pool = new Pool({ connectionString: process.env.DATABASE_URL })
 
export class PGVectorStore {
  async upsertChunks(chunks: ChunkWithEmbedding[]): Promise<void> {
    // ... 见第 4 节
  }
 
  async search(queryVector: number[], options: SearchOptions): Promise<SearchResult[]> {
    // ... 见第 5 节
  }
 
  async deleteDocument(documentId: string): Promise<void> {
    await pool.query('DELETE FROM document_chunks WHERE document_id = $1', [documentId])
  }
 
  async getStats(): Promise<{ totalChunks: number; totalDocuments: number }> {
    const result = await pool.query(`
      SELECT
        COUNT(*) as total_chunks,
        COUNT(DISTINCT document_id) as total_documents
      FROM document_chunks
    `)
    return result.rows[0]
  }
}

总结

回顾这一节的要点：

  • pgvector 复用 PostgreSQL，零新增组件
  • 两种索引：IVFFlat（省内存）和 HNSW（查询快）
  • 可以和 PG 全文检索结合做混合检索
  • 规模限制：百万级以下表现好，更大规模需要迁移
  • 性能调优：ef_search、probes、VACUUM ANALYZE、内存配置
  • 适合已有 PostgreSQL 的团队作为 RAG 起步方案

下一篇讲 Qdrant 实践——功能更全面的专用向量数据库。