pgvector教程：为PostgreSQL添加向量搜索，实现RAG和语义搜索

在构建RAG管道或需要语义搜索的应用时，首先需要决定的是将Embedding存储在哪里。Pinecone、Qdrant、Weaviate等专用向量数据库是一种选择，但对于已经运行PostgreSQL的团队而言，pgvector是更快、更经济、运维更简单的方案。

pgvector是PostgreSQL的开源Extension，为现有Postgres数据库增加了vector数据类型、相似度搜索运算符以及HNSW/IVFFlat索引。Embedding与应用数据存储在同一数据库中，使用标准SQL查询，享有Postgres完整的ACID保障。

本教程从安装到生产级RAG查询，覆盖全流程并提供完整代码。

最终成果： 在PostgreSQL中存储文档Embedding，并为指定查询检索最相关Chunk的语义搜索系统——即RAG管道的Retrieval层。

技术栈：

• PostgreSQL 15+（pgvector 0.7+）
• Python 3.11
• Embedding生成：openai或ollama
• 数据库连接：psycopg2 / asyncpg

---

pgvector是什么，为什么要用它？

Vector Embedding是以浮点数列表形式（通常768至3,072维）编码文本、图像等数据语义内容的表示。意义相近的两段文本在该高维空间中生成位置相近的向量。

pgvector使Postgres能够存储这些向量并高效执行距离计算：

运算	SQL
余弦相似度	embedding <=> query_vector
L2距离	embedding <-> query_vector
内积	embedding <#> query_vector

适合使用pgvector的场景：

• 已有PostgreSQL运维经验
• 向量数量在500万条以下
• 需要将Embedding与关系型数据JOIN（用户ID、文档元数据、权限控制）
• 需要跨应用数据和Embedding的ACID事务
• 不想维护额外的基础设施组件

应选择专用向量数据库的场景：

• 超过1,000万向量，且需要低于10ms的延迟
• 需要大规模高级过滤
• 尚无Postgres基础设施

对于中国企业的RAG项目——内部文档检索、合规文档管理（等保2.0 / PIPL / 数据安全法）、OA知识库——pgvector通常足够，且与用友、金蝶等国产ERP的PostgreSQL集成更为顺畅，无需引入额外的基础设施组件。

---

架构概览

flowchart TD
    A["文档\n(PDF、文本、数据库记录)"] --> B["分块处理\n(约500 Token每段)"]
    B --> C["Embedding模型\n(OpenAI / Ollama / 本地)"]
    C --> D["pgvector\n(PostgreSQL)"]
    E["用户查询"] --> F["查询转Embedding"]
    F --> G["相似度搜索\n(余弦 / L2)"]
    D --> G
    G --> H["Top-k相关Chunk"]
    H --> I["大语言模型\n(GPT / Claude / 本地)"]
    I --> J["含引用来源的回答"]

---

第一步 — 安装pgvector

选项A：Docker（推荐用于开发环境）

docker run -d \
  --name pgvector-dev \
  -e POSTGRES_PASSWORD=secret \
  -e POSTGRES_DB=ragdb \
  -p 5432:5432 \
  pgvector/pgvector:pg16

此镜像已预装pgvector，无需手动编译Extension。

选项B：在现有PostgreSQL上安装

Ubuntu / Debian：

sudo apt install postgresql-16-pgvector

macOS（Homebrew）：

brew install pgvector

在数据库中启用Extension

CREATE EXTENSION IF NOT EXISTS vector;

-- 验证安装
SELECT * FROM pg_extension WHERE extname = 'vector';

---

第二步 — 创建Schema

-- documents表：存储源文本和元数据
CREATE TABLE documents (
    id          BIGSERIAL PRIMARY KEY,
    title       TEXT NOT NULL,
    source_url  TEXT,
    lang        VARCHAR(5) DEFAULT 'zh',
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- document_chunks表：存储文本片段和Embedding
CREATE TABLE document_chunks (
    id          BIGSERIAL PRIMARY KEY,
    document_id BIGINT REFERENCES documents(id) ON DELETE CASCADE,
    chunk_index INTEGER NOT NULL,
    content     TEXT NOT NULL,
    token_count INTEGER,
    embedding   vector(1536),  -- 维度必须与Embedding模型匹配
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

各模型Embedding维度：

模型	维度
OpenAI text-embedding-3-small	1536
OpenAI text-embedding-3-large	3072
Ollama nomic-embed-text	768
Ollama mxbai-embed-large	1024
BGE-M3（多语言，中文效果优秀）	1024

vector(1536)列类型必须完全匹配，同一列内不能混用不同维度。

---

第三步 — 文档摄入与Embedding生成

import os
import psycopg2
from openai import OpenAI
import tiktoken

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
DB_URL = "postgresql://postgres:secret@localhost:5432/ragdb"

def chunk_text(text: str, max_tokens: int = 500, overlap: int = 50) -> list[str]:
    """将文本按Token数量分割为带重叠的Chunk"""
    enc = tiktoken.get_encoding("cl100k_base")
    tokens = enc.encode(text)
    chunks = []
    start = 0
    while start < len(tokens):
        end = min(start + max_tokens, len(tokens))
        chunks.append(enc.decode(tokens[start:end]))
        start += max_tokens - overlap
    return chunks

def get_embedding(text: str) -> list[float]:
    """从OpenAI获取Embedding向量"""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text.replace("\n", " ")
    )
    return response.data[0].embedding

def ingest_document(title: str, content: str, source_url: str = None, lang: str = "zh"):
    """对文档分块、生成Embedding并存入PostgreSQL"""
    conn = psycopg2.connect(DB_URL)
    cur = conn.cursor()
    try:
        cur.execute(
            "INSERT INTO documents (title, source_url, lang) VALUES (%s, %s, %s) RETURNING id",
            (title, source_url, lang)
        )
        doc_id = cur.fetchone()[0]
        chunks = chunk_text(content)
        print(f"正在摄入《{title}》：{len(chunks)}个Chunk")
        for i, chunk in enumerate(chunks):
            embedding = get_embedding(chunk)
            cur.execute(
                """INSERT INTO document_chunks
                   (document_id, chunk_index, content, token_count, embedding)
                   VALUES (%s, %s, %s, %s, %s)""",
                (doc_id, i, chunk, len(chunk.split()), embedding)
            )
        conn.commit()
        print(f"完成：document_id={doc_id}")
        return doc_id
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        cur.close()
        conn.close()

# 使用示例
ingest_document(
    title="个人信息保护制度 v3.2（PIPL合规版）",
    content=open("pipl_policy.txt").read(),
    source_url="https://internal.company.cn/policies/pipl",
    lang="zh"
)

使用Ollama进行本地Embedding（无需API密钥）

import httpx

def get_embedding_ollama(text: str, model: str = "nomic-embed-text") -> list[float]:
    """从本地Ollama实例获取Embedding"""
    response = httpx.post(
        "http://localhost:11434/api/embeddings",
        json={"model": model, "prompt": text}
    )
    return response.json()["embedding"]

将get_embedding()替换为get_embedding_ollama()，使用nomic-embed-text时需将Schema中的向量维度改为768。本地Embedding方案特别适合等保2.0要求数据不出境的场景。

---

第四步 — 创建高速相似搜索索引

HNSW（大多数场景推荐）

CREATE INDEX ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

参数	默认值	效果
m	16	每节点连接数。越高召回率越好，内存占用越大
ef_construction	64	构建时搜索宽度。越高召回率越好，构建越慢

IVFFlat（超大规模数据集）

CREATE INDEX ON document_chunks
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

对比：

	HNSW	IVFFlat
召回率	更高	较低（可调）
构建时间	较慢	较快
内存占用	更多	较少
适用场景	200万向量以下	200万向量以上

---

第五步 — 运行相似度搜索

def semantic_search(
    query: str,
    top_k: int = 5,
    lang_filter: str = None,
    min_similarity: float = 0.7
) -> list[dict]:
    """查找与查询语义最相近的Chunk"""
    query_embedding = get_embedding(query)
    conn = psycopg2.connect(DB_URL)
    cur = conn.cursor()
    try:
        cur.execute("SET hnsw.ef_search = 100")
        sql = """
            SELECT dc.id, dc.content, dc.chunk_index,
                   d.title, d.source_url, d.lang,
                   1 - (dc.embedding <=> %s::vector) AS similarity
            FROM document_chunks dc
            JOIN documents d ON d.id = dc.document_id
            WHERE 1 - (dc.embedding <=> %s::vector) >= %s
            {lang_clause}
            ORDER BY dc.embedding <=> %s::vector
            LIMIT %s
        """.format(lang_clause="AND d.lang = %s" if lang_filter else "")
        params = [query_embedding, query_embedding, min_similarity]
        if lang_filter:
            params.append(lang_filter)
        params.extend([query_embedding, top_k])
        cur.execute(sql, params)
        return [
            {"id": r[0], "content": r[1], "chunk_index": r[2],
             "title": r[3], "source_url": r[4], "lang": r[5],
             "similarity": float(r[6])}
            for r in cur.fetchall()
        ]
    finally:
        cur.close()
        conn.close()

# 使用示例
results = semantic_search(
    query="个人信息跨境传输的合规要求是什么？",
    top_k=5,
    lang_filter="zh",
    min_similarity=0.75
)

---

第六步 — 构建完整RAG管道

def rag_query(question: str, lang: str = "zh", top_k: int = 5) -> dict:
    """完整RAG管道：检索相关Chunk并生成有引用的回答"""
    chunks = semantic_search(question, top_k=top_k, lang_filter=lang)
    if not chunks:
        return {"answer": "未找到与此问题相关的文档。", "sources": []}

    context = "\n\n".join([f"[{i+1}] {c['content']}" for i, c in enumerate(chunks)])
    sources = [{"title": c["title"], "url": c["source_url"], "similarity": c["similarity"]}
               for c in chunks]

    system_prompt = """你是一个有帮助的助手。请仅使用提供的Context回答问题。
如果Context中没有答案，请回答"在现有文档中未找到此信息"。
引用信息时请务必标注来源编号[1]、[2]等。
请用中文回答。"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Context:\n{context}\n\n问题：{question}\n\n回答："}
        ],
        temperature=0.1
    )
    return {"answer": response.choices[0].message.content, "sources": sources}

# 使用示例
result = rag_query("等保2.0对工控系统的日志审计有哪些具体要求？", lang="zh")
print(result["answer"])

---

第七步 — 混合搜索（向量 + 全文检索）

纯向量搜索有时会遗漏精确的关键词匹配。混合搜索结合向量相似度与PostgreSQL全文检索，实现更高召回率：

ALTER TABLE document_chunks ADD COLUMN content_tsv tsvector
    GENERATED ALWAYS AS (to_tsvector('simple', content)) STORED;

CREATE INDEX ON document_chunks USING gin(content_tsv);

对于中文全文检索，建议使用zhparser或pg_jiebaExtension：

-- 安装中文分词支持后
CREATE INDEX ON document_chunks
USING gin(to_tsvector('zhparser', content));

混合搜索在领域专用语料库上可提升15–30%的召回率，特别适合包含产品编码、法规条款编号（如GB/T 22240-2008等保标准）或专有名词的文档。

---

性能调优

-- 增加内存加速索引构建
SET maintenance_work_mem = '1GB';
CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops);

-- 提高ef_search以获得更好的召回率（默认值40）
SET hnsw.ef_search = 100;

-- 按语言创建部分索引（适合多语言场景）
CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops)
WHERE document_id IN (SELECT id FROM documents WHERE lang = 'zh');

-- 确认查询计划是否使用了索引
EXPLAIN (ANALYZE, BUFFERS)
SELECT id, content, embedding <=> '[0.1, 0.2, ...]'::vector AS distance
FROM document_chunks ORDER BY distance LIMIT 5;

---

常见问题

pgvector能处理多少向量？

在标准硬件（16GB内存、SSD）上，pgvector可流畅处理100至500万条向量。超过这个范围查询延迟会开始上升。超过1,000万条时，建议考虑Qdrant或Weaviate。

应该使用余弦相似度还是L2距离？

文本Embedding应使用余弦相似度（<=>），它衡量向量间的角度而忽略大小。RAG的文本场景请始终使用余弦相似度。

等保2.0合规场景下使用pgvector有哪些注意事项？

pgvector数据存储在PostgreSQL中，可直接应用标准数据库安全措施。数据加密可通过PostgreSQL的pgcrypto或透明磁盘加密实现；访问日志通过pgaudit记录，满足等保2.0的日志审计要求（第三级及以上）。对于OT/工控系统（ICS）场景，建议Embedding数据库与生产控制网络逻辑隔离，符合等保2.0对工控系统的网络安全要求。PIPL要求个人信息不得出境时，使用Ollama本地Embedding模型可完全避免数据离开本地环境。

Chunk大小设置多少合适？

以50 Token重叠的500 Token为起点。过小（低于100 Token）会丢失上下文；过大（超过1,000 Token）会降低相关性。对于结构化文档（法律合同、技术规范），建议按章节边界分块而非固定Token数。

pgvector与Pinecone如何选择？

如果已有Postgres：直接用pgvector。可以使用SQL JOIN、事务，无需维护额外组件，也没有按查询计费。对于500万向量以下的企业RAG项目，pgvector是最具性价比的选择。在私有化部署或等保合规要求下，pgvector比托管云服务更容易通过安全评审。

---

上线前检查清单

• [ ] 已在Embedding列上创建HNSW或IVFFlat索引
• [ ] 已将maintenance_work_mem设置为至少512MB用于索引构建
• [ ] 已调整hnsw.ef_search（从100开始，对召回率与延迟进行基准测试）
• [ ] 过滤频繁时已按语言或租户创建部分索引
• [ ] 已在PostgreSQL前部署连接池（PgBouncer）
• [ ] Embedding生成为异步队列处理，不与用户请求同步执行
• [ ] 已设置min_similarity阈值（0.7–0.8）过滤低质量匹配
• [ ] 已对慢速向量查询监控pg_stat_activity
• [ ] 批量摄入后定期执行VACUUM ANALYZE
• [ ] 已确认pgaudit日志记录满足等保2.0审计要求

---

下一步

本教程覆盖了Retrieval层。完善RAG技术栈：

• 中文和多语言文档摄入与Embedding详解 — 参见：LlamaIndex + pgvector: Production RAG for Thai and Japanese Business Documents
• 本地运行Embedding模型的硬件选型（无需OpenAI） — 参见：Choosing Hardware for Local LLMs in 2026
• RAG在私有AI整体架构中的定位 — 参见：Private AI vs ChatGPT

正在为内部文档、等保2.0合规工作流或客服知识库构建RAG系统？欢迎联系 hello@simplico.net — 我们为泰国和日本的企业客户提供生产级RAG系统部署服务。