AI时代首选数据库：PostgreSQL从入门到进阶实战指南

原创

用户12608867

发布于 2026-08-15 13:24:51

1710

AI时代首选数据库：PostgreSQL从入门到进阶实战指南

从关系模型到向量检索，一文掌握PostgreSQL在AI工作流中的核心玩法

一、为什么AI时代，PostgreSQL成了“默认选项”？

在AI应用爆发的今天，数据栈比以往任何时候都更加多元——既有结构化业务数据，又有半结构化JSON日志，还有非结构化的文本、向量嵌入（Embeddings）。传统做法是“拼凑组合”：MySQL存事务数据，Elasticsearch做全文检索，Redis扛缓存，Milvus或Pinecone管向量。这套架构运维复杂、一致性难保、学习成本高。

PostgreSQL 凭借其强大的扩展生态，正在成为AI时代的数据“基座”：

原生JSON/JSONB：支持高效存储与GIN索引查询，替代部分NoSQL场景；
全文检索（tsvector/tsquery）：内置级联词典、权重、排名，无需额外搜索引擎；
pgvector扩展：提供向量数据类型及精确/近似最近邻搜索（IVFFlat、HNSW），直接支撑RAG（检索增强生成）应用；
PL/pgSQL与函数管道：可在数据库内完成数据清洗、特征计算，减少网络往返；
外部数据包装器（FDW）：无缝查询OSS、S3、Parquet等外部数据源，构建湖仓一体；
支持MADlib、pgML等机器学习扩展：在库内运行线性回归、聚类等算法。

本文将从零开始，手把手带你在本地部署PostgreSQL，逐步掌握基础SQL、高级分析、索引调优、全文搜索，最后实战 pgvector + OpenAI Embedding + 本地大模型 的RAG问答系统。所有代码均可直接运行。

二、环境准备：5分钟启动PostgreSQL 16

2.1 使用Docker（推荐）

docker run -d \
  --name pg16 \
  -e POSTGRES_USER=ai_user \
  -e POSTGRES_PASSWORD=ai_pass \
  -e POSTGRES_DB=ai_db \
  -p 5432:5432 \
  postgres:16-alpine

2.2 安装pgvector扩展（容器内执行）

docker exec -it pg16 bash
apk add --no-cache build-base git
git clone https://github.com/pgvector/pgvector.git
cd pgvector
make && make install

或使用官方已集成pgvector的镜像（更简单）：

docker run -d \
  --name pg16-vector \
  -e POSTGRES_USER=ai_user \
  -e POSTGRES_PASSWORD=ai_pass \
  -e POSTGRES_DB=ai_db \
  -p 5432:5432 \
  ankane/pgvector:latest

2.3 连接数据库

docker exec -it pg16-vector psql -U ai_user -d ai_db

创建扩展：

CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS fuzzystrmatch;  -- 用于字符串相似度
CREATE EXTENSION IF NOT EXISTS pg_trgm;        -- 三元组索引

三、基础篇：建表、CRUD与约束

3.1 设计一张“AI模型元数据表”

CREATE TABLE models (
    id          SERIAL PRIMARY KEY,
    name        TEXT NOT NULL,
    version     TEXT,
    framework   TEXT CHECK (framework IN ('PyTorch', 'TensorFlow', 'ONNX')),
    params      JSONB,                      -- 超参数
    created_at  TIMESTAMPTZ DEFAULT now(),
    updated_at  TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE model_embeddings (
    model_id    INTEGER REFERENCES models(id) ON DELETE CASCADE,
    chunk_text  TEXT,
    embedding   vector(1536),               -- OpenAI text-embedding-ada-002 维度
    metadata    JSONB,
    PRIMARY KEY (model_id, chunk_text)
);

3.2 插入与查询

INSERT INTO models (name, version, framework, params)
VALUES ('ResNet50', 'v1.0', 'PyTorch', '{"batch_size": 32, "lr": 0.001}'),
       ('BERT-base', '2.1', 'TensorFlow', '{"max_len": 512, "dropout": 0.1}');

-- 更新 updated_at 自动触发器
CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at = now();
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_models_updated_at
BEFORE UPDATE ON models
FOR EACH ROW EXECUTE FUNCTION update_updated_at();

3.3 JSONB查询实战

-- 查找 lr 参数大于 0.0005 的模型
SELECT name, params->>'lr' AS lr
FROM models
WHERE params @> '{"lr": 0.001}';

-- 使用 GIN 索引加速 JSONB 查询
CREATE INDEX idx_models_params ON models USING GIN (params);

四、进阶篇：窗口函数、CTE与递归查询

4.1 窗口函数：计算实验运行平均值

假设我们有一张模型训练日志表：

CREATE TABLE training_logs (
    model_id    INTEGER REFERENCES models(id),
    epoch       INTEGER,
    loss        NUMERIC,
    accuracy    NUMERIC,
    log_time    TIMESTAMPTZ DEFAULT now()
);

-- 插入样例数据（略）
-- 求每个模型最近3次epoch的滑动平均损失
SELECT model_id, epoch, loss,
       AVG(loss) OVER (PARTITION BY model_id ORDER BY epoch ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS moving_avg_loss
FROM training_logs
ORDER BY model_id, epoch;

4.2 递归CTE：解析依赖关系（用于模型DAG）

CREATE TABLE model_dependencies (
    parent_id   INTEGER REFERENCES models(id),
    child_id    INTEGER REFERENCES models(id),
    PRIMARY KEY (parent_id, child_id)
);

-- 递归查询找出所有依赖链条
WITH RECURSIVE dep_tree AS (
    SELECT parent_id, child_id, 1 AS depth
    FROM model_dependencies
    WHERE parent_id = 1
    UNION ALL
    SELECT md.parent_id, md.child_id, dt.depth + 1
    FROM model_dependencies md
    JOIN dep_tree dt ON md.parent_id = dt.child_id
)
SELECT * FROM dep_tree ORDER BY depth;

五、索引优化：为AI工作负载加速

5.1 B-tree vs Hash vs BRIN

-- 适合等值查询的哈希索引
CREATE INDEX idx_models_name_hash ON models USING HASH (name);

-- 适合超大表的块级索引（时序数据）
CREATE INDEX idx_training_logs_time_brin ON training_logs USING BRIN (log_time);

5.2 部分索引：只索引活跃数据

-- 假设有 status 字段，只关心 'active' 状态
CREATE INDEX idx_models_active ON models (name) WHERE status = 'active';

5.3 覆盖索引（INCLUDE）

-- 查询只需 name 和 params，无需回表
CREATE INDEX idx_models_cover ON models (name) INCLUDE (params);

5.4 使用 pg_trgm 进行模糊搜索加速

CREATE INDEX idx_models_name_trgm ON models USING GIN (name gin_trgm_ops);

-- 查询相似度 > 0.3 的模型
SELECT name, similarity(name, 'ResNet101') AS sim
FROM models
WHERE name % 'ResNet101'   -- % 运算符依赖 pg_trgm
ORDER BY sim DESC;

六、全文搜索：让数据库变成搜索引擎

6.1 创建 tsvector 与 tsquery

-- 给 model 添加 description 字段
ALTER TABLE models ADD COLUMN description TEXT;

UPDATE models SET description = 'ResNet50 is a deep residual network for image classification' WHERE id=1;
UPDATE models SET description = 'BERT is a transformer-based model for NLP tasks' WHERE id=2;

-- 生成词向量（支持中文需配置zhparser）
CREATE INDEX idx_models_fts ON models USING GIN (to_tsvector('english', description));

-- 搜索包含 'residual' 和 'image' 的记录
SELECT id, name, description,
       ts_rank(to_tsvector('english', description), plainto_tsquery('english', 'residual image')) AS rank
FROM models
WHERE to_tsvector('english', description) @@ plainto_tsquery('english', 'residual image')
ORDER BY rank DESC;

6.2 自定义词典与权重

-- 设置不同字段权重 (A/B/C/D)
ALTER TABLE models ADD COLUMN fts_document tsvector
GENERATED ALWAYS AS (
    setweight(to_tsvector('english', coalesce(name,'')), 'A') ||
    setweight(to_tsvector('english', coalesce(description,'')), 'B')
) STORED;

CREATE INDEX idx_models_fts_weighted ON models USING GIN (fts_document);

-- 查询时使用权重
SELECT name, ts_rank(fts_document, websearch_to_tsquery('english', 'transformer NLP')) AS rank
FROM models
WHERE fts_document @@ websearch_to_tsquery('english', 'transformer NLP')
ORDER BY rank DESC;

七、向量检索：pgvector 核心实战（RAG 核心）

7.1 生成并插入向量

假设我们已通过 Python 调用 OpenAI Embedding API 得到向量（1536维），这里用随机向量模拟：

INSERT INTO model_embeddings (model_id, chunk_text, embedding)
VALUES (1, 'ResNet uses skip connections', '[' || array_to_string(array(select random()::numeric(10,8) from generate_series(1,1536)), ',') || ']'::vector);

实际生产中用 Python 插入：

import psycopg2
import openai

conn = psycopg2.connect("dbname=ai_db user=ai_user password=ai_pass host=localhost")
cur = conn.cursor()

text = "ResNet uses skip connections to avoid vanishing gradients"
response = openai.Embedding.create(input=text, model="text-embedding-ada-002")
embedding = response['data'][0]['embedding']  # list of 1536 floats

cur.execute("""
    INSERT INTO model_embeddings (model_id, chunk_text, embedding)
    VALUES (%s, %s, %s::vector)
""", (1, text, embedding))
conn.commit()

7.2 精确最近邻搜索（L2距离）

SELECT model_id, chunk_text,
       embedding <-> '[0.123, 0.456, ...]'::vector AS distance
FROM model_embeddings
ORDER BY distance
LIMIT 5;

7.3 索引加速：IVFFlat 与 HNSW

-- 构建 IVFFlat 索引（需先有数据）
CREATE INDEX idx_embeddings_ivfflat ON model_embeddings
USING ivfflat (embedding vector_l2_ops) WITH (lists = 100);

-- 或 HNSW（构建更慢但查询更快）
CREATE INDEX idx_embeddings_hnsw ON model_embeddings
USING hnsw (embedding vector_l2_ops) WITH (m = 16, ef_construction = 200);

-- 查询时指定 probes（仅 IVFFlat）
SET ivfflat.probes = 10;  -- 增大召回率

7.4 混合检索：结合全文搜索与向量相似度

WITH semantic AS (
    SELECT model_id, chunk_text,
           1 - (embedding <=> query_embedding) AS similarity
    FROM model_embeddings, (SELECT '[0.1, ...]'::vector AS query_embedding) q
    ORDER BY similarity DESC
    LIMIT 20
),
keyword AS (
    SELECT id, ts_rank(fts_document, query_ts) AS rank
    FROM models, plainto_tsquery('english', 'skip connections') AS query_ts
    WHERE fts_document @@ query_ts
    ORDER BY rank DESC
    LIMIT 20
)
SELECT COALESCE(s.model_id, k.id) AS id,
       COALESCE(s.similarity, 0) * 0.6 + COALESCE(k.rank, 0) * 0.4 AS hybrid_score
FROM semantic s
FULL OUTER JOIN keyword k ON s.model_id = k.id
ORDER BY hybrid_score DESC;

八、AI 集成：在数据库内调用外部 API

8.1 使用 plpython3u 调用 OpenAI（需安装 Python 扩展）

CREATE EXTENSION plpython3u;

CREATE OR REPLACE FUNCTION openai_embed(text_input TEXT)
RETURNS vector(1536) AS $$
    import openai
    openai.api_key = 'your-key'
    resp = openai.Embedding.create(input=text_input, model="text-embedding-ada-002")
    return resp['data'][0]['embedding']
$$ LANGUAGE plpython3u VOLATILE;

-- 查询时直接生成向量并比较（生产环境慎用，易超时）
SELECT chunk_text
FROM model_embeddings
ORDER BY embedding <-> openai_embed('new query text')
LIMIT 5;

8.2 使用 pg_http 调用本地 LLM（如 Ollama）

-- 安装 pg_http (需编译)
CREATE EXTENSION http;

CREATE OR REPLACE FUNCTION ollama_generate(prompt TEXT)
RETURNS TEXT AS $$
DECLARE
    response_json JSONB;
BEGIN
    response_json := http_post(
        'http://localhost:11434/api/generate',
        jsonb_build_object('model', 'llama2', 'prompt', prompt, 'stream', false)::text,
        'application/json'
    )::jsonb;
    RETURN response_json->>'response';
END;
$$ LANGUAGE plpgsql;

-- 结合检索结果生成答案
SELECT ollama_generate('基于以下上下文回答：' || string_agg(chunk_text, ' '))
FROM model_embeddings
WHERE embedding <-> '[query_vector]'::vector < 0.5;

九、性能调优：让PostgreSQL跑得更快

9.1 调整共享内存与缓存

在 postgresql.conf 中根据内存调整：

shared_buffers = 4GB                # 系统内存的 25%
effective_cache_size = 12GB         # 系统内存的 75%
work_mem = 64MB                     # 排序/哈希操作内存
maintenance_work_mem = 1GB          # VACUUM/索引构建
max_parallel_workers_per_gather = 4
wal_buffers = 64MB
checkpoint_completion_target = 0.9

9.2 监控慢查询与执行计划

-- 开启 pg_stat_statements
CREATE EXTENSION pg_stat_statements;

-- 查看最耗时的 TOP 5 查询
SELECT query, calls, total_time, mean_time
FROM pg_stat_statements
ORDER BY total_time DESC
LIMIT 5;

-- 分析执行计划
EXPLAIN (ANALYZE, BUFFERS, COSTS)
SELECT * FROM model_embeddings
ORDER BY embedding <-> '[0.1, ...]'::vector
LIMIT 10;

9.3 分区表：处理海量向量数据

CREATE TABLE embeddings_partitioned (
    model_id INTEGER,
    chunk_text TEXT,
    embedding vector(1536),
    created_date DATE DEFAULT CURRENT_DATE
) PARTITION BY RANGE (created_date);

CREATE TABLE embeddings_2025_q1 PARTITION OF embeddings_partitioned
FOR VALUES FROM ('2025-01-01') TO ('2025-04-01');

-- 查询时自动分区剪裁
EXPLAIN SELECT * FROM embeddings_partitioned WHERE created_date >= '2025-02-01';

十、完整实战：构建 RAG 问答系统（端到端）

10.1 架构图（文字描述）

文档切片 → 将 PDF/Markdown 切为 512 token 块
向量化 → 调用 OpenAI Embedding 存入 model_embeddings
查询处理 → 用户问题向量化，通过 pgvector 检索 top-k 相关块
上下文组装 → 将检索结果拼接成 Prompt
LLM 生成 → 调用 OpenAI ChatCompletion 或本地 Ollama

10.2 核心代码 (Python + psycopg2)

import os
import openai
import psycopg2
from psycopg2.extras import Json

openai.api_key = os.getenv("OPENAI_API_KEY")
conn = psycopg2.connect("dbname=ai_db user=ai_user password=ai_pass host=localhost")

def embed(text):
    resp = openai.Embedding.create(input=text, model="text-embedding-ada-002")
    return resp['data'][0]['embedding']

def retrieve(query, top_k=5):
    vec = embed(query)
    cur = conn.cursor()
    cur.execute("""
        SELECT chunk_text, 1 - (embedding <=> %s::vector) AS similarity
        FROM model_embeddings
        ORDER BY embedding <=> %s::vector
        LIMIT %s
    """, (vec, vec, top_k))
    return [row[0] for row in cur.fetchall()]

def generate_answer(query):
    chunks = retrieve(query)
    context = "\n".join(chunks)
    prompt = f"基于以下信息回答问题：\n{context}\n\n问题：{query}"
    resp = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3
    )
    return resp.choices[0].message.content

# 测试
print(generate_answer("ResNet如何解决梯度消失？"))

10.3 在数据库内部实现召回+生成（PL/pgSQL + Python）

为了减少网络开销，可将检索逻辑封装为函数，返回 JSON 结果集：

CREATE OR REPLACE FUNCTION rag_search(query_text TEXT, k INT DEFAULT 5)
RETURNS TABLE(chunk TEXT, score FLOAT) AS $$
BEGIN
    RETURN QUERY
    SELECT chunk_text, 1 - (embedding <=> openai_embed(query_text)) AS sim
    FROM model_embeddings
    ORDER BY embedding <=> openai_embed(query_text)
    LIMIT k;
END;
$$ LANGUAGE plpgsql VOLATILE;

十一、监控与运维：生产环境必备

11.1 查看索引使用情况

SELECT schemaname, tablename, indexname, idx_scan, idx_tup_read
FROM pg_stat_user_indexes
ORDER BY idx_scan;

11.2 自动清理（VACUUM）调优

-- 调整自动清理阈值
ALTER TABLE model_embeddings SET (autovacuum_vacuum_scale_factor = 0.1);
ALTER TABLE model_embeddings SET (autovacuum_analyze_scale_factor = 0.05);

11.3 使用 pgcli 或 pgAdmin 进行可视化

十二、总结与展望

本文从零开始，完整展示了 PostgreSQL 在 AI 时代的全方位能力：

✅ 基础 CRUD 与 JSONB 灵活性
✅ 窗口函数、CTE 解决复杂分析
✅ 多类型索引（B-tree, GIN, BRIN, 部分索引）
✅ 全文搜索与 pg_trgm 模糊匹配
✅ pgvector 向量索引及混合检索
✅ 数据库内调用 AI API（plpython3u / http）
✅ 生产级调优与监控

PostgreSQL 不再只是“关系型数据库”，而是 AI 数据操作系统。它统一了结构化、半结构化和向量数据，让 RAG、推荐系统、特征存储等 AI 应用得以在单一可靠平台上构建。随着 pgai、pgml 等扩展的成熟，数据库内直接运行轻量级模型推理将成为常态。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

AIGC

AI时代首选数据库：PostgreSQL从入门到进阶实战指南

AI时代首选数据库：PostgreSQL从入门到进阶实战指南

AI时代首选数据库：PostgreSQL从入门到进阶实战指南

一、为什么AI时代，PostgreSQL成了“默认选项”？

二、环境准备：5分钟启动PostgreSQL 16

2.1 使用Docker（推荐）

2.2 安装pgvector扩展（容器内执行）

2.3 连接数据库

三、基础篇：建表、CRUD与约束

3.1 设计一张“AI模型元数据表”

3.2 插入与查询

3.3 JSONB查询实战

四、进阶篇：窗口函数、CTE与递归查询

4.1 窗口函数：计算实验运行平均值

4.2 递归CTE：解析依赖关系（用于模型DAG）

五、索引优化：为AI工作负载加速

5.1 B-tree vs Hash vs BRIN

5.2 部分索引：只索引活跃数据

5.3 覆盖索引（INCLUDE）

5.4 使用 pg_trgm 进行模糊搜索加速

六、全文搜索：让数据库变成搜索引擎

6.1 创建 tsvector 与 tsquery

6.2 自定义词典与权重

七、向量检索：pgvector 核心实战（RAG 核心）

7.1 生成并插入向量

7.2 精确最近邻搜索（L2距离）

7.3 索引加速：IVFFlat 与 HNSW

7.4 混合检索：结合全文搜索与向量相似度

八、AI 集成：在数据库内调用外部 API

8.1 使用 plpython3u 调用 OpenAI（需安装 Python 扩展）

8.2 使用 pg_http 调用本地 LLM（如 Ollama）

九、性能调优：让PostgreSQL跑得更快

9.1 调整共享内存与缓存

9.2 监控慢查询与执行计划

9.3 分区表：处理海量向量数据

十、完整实战：构建 RAG 问答系统（端到端）

10.1 架构图（文字描述）

10.2 核心代码 (Python + psycopg2)

10.3 在数据库内部实现召回+生成（PL/pgSQL + Python）

十一、监控与运维：生产环境必备

11.1 查看索引使用情况

11.2 自动清理（VACUUM）调优

11.3 使用 pgcli 或 pgAdmin 进行可视化

十二、总结与展望

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐