文章目录
- 一、PGVector 概述
- 1.1 PGVector 简介
- 1.2 基本概念
- 1.3 使用建议
- 二、安装 PGVector
- 2.1 使用 Docker（推荐，快速上手）
- 2.2 在现有 PostgreSQL 中安装（Linux/macOS）
- 三、Python 操作PGVector详解
- 3.1 安装依赖
- 3.2 连接数据库 & 创建表
- 3.3 插入向量数据
- 3.4 向量相似性搜索（核心功能）
- 3.5 使用 SQLAlchemy（ORM 方式）
- 3.6 混合搜索SQL（关键词 + 向量）
- 3.7 完整可执行案例（单文件）
PGVector 是 PostgreSQL 的一个开源扩展，用于高效存储和检索向量（embedding）数据，支持ANN（近似最近邻）搜索，广泛应用于语义搜索、推荐系统、RAG（检索增强生成）等场景。
一、PGVector 概述
1.1 PGVector 简介
- 官方 GitHub：https://github.com/pgvector/pgvector
- 核心功能：
- 存储高维向量（如 768 维的 BERT embedding）
- 支持多种距离度量：L2（欧氏距离）、Inner Product（点积）、Cosine（余弦相似度）
- 提供高效索引（IVFFlat、HNSW）
- 完全集成到 PostgreSQL，SQL 友好
对于新闻检索场景，使用 pgvector 有一个巨大的优势：结构化数据与非结构化数据的统一存储。你不需要单独维护一个向量数据库，新闻的标题、发布时间、作者、原文内容都存在 Postgres 里，而向量作为一个普通的字段存在同一行，查询时可以用 SQL 同时过滤。
1.2 基本概念
| 概念 | 说明 |
|---|---|
vector(n) | 向量类型，n为维度（如vector(768)） |
<-> | L2 距离操作符 |
<#> | 负内积操作符（用于最大内积搜索） |
<=> | 余弦距离操作符（1 - cosine_similarity） |
ivfflat | 基于聚类的 ANN 索引 |
hnsw | 层次化可导航小世界图索引（更快更准，但占内存） |
1.3 使用建议
| 场景 | 建议 |
|---|---|
| 数据量 < 1万 | 无需索引，直接搜索 |
| 1万 ~ 100万 | IVFFlat 索引（lists=100~1000） |
| > 100万 | HNSW 索引（内存充足时） |
| 高写入 | 避免频繁重建索引，批量插入后建索引 |
| 中文支持 | 结合zhparser+tsvector实现混合搜索 |
适用场景：RAG 应用、语义搜索、个性化推荐、图像/文本相似检索。
二、安装 PGVector
2.1 使用 Docker（推荐，快速上手）
# 拉取带 pgvector 的 PostgreSQL 镜像dockerrun -d\--name pgvector-demo\-p5432:5432\-ePOSTGRES_USER=admin\- ePOSTGRES_PASSWORD=123456\-ePOSTGRES_DB=vector_db\ankane/pgvector:latest
ankane/pgvector是官方维护的镜像，已预装 pgvector 扩展。
2.2 在现有 PostgreSQL 中安装（Linux/macOS）
# 克隆源码gitclone https://github.com/pgvector/pgvector.gitcdpgvector# 编译安装（需 postgresql-server-dev）makemakeinstall# 登录 psql，创建扩展CREATE EXTENSION vector;三、Python 操作PGVector详解
推荐使用
psycopg2（轻量）或SQLModel/SQLAlchemy（ORM）
3.1 安装依赖
pipinstallpsycopg2-binary sqlalchemy pgvector注意：
pgvectorPython 包提供类型支持（非必须，但推荐）
3.2 连接数据库 & 创建表
importpsycopg2frompsycopg2.extrasimportRealDictCursorimportnumpyasnp# 数据库连接配置DB_CONFIG={"host":"localhost","port":5432,"database":"vector_db","user":"admin","password":"123456"}conn=psycopg2.connect(**DB_CONFIG)cur=conn.cursor(cursor_factory=RealDictCursor)# 创建扩展（首次运行需要）cur.execute("CREATE EXTENSION IF NOT EXISTS vector;")# 创建新闻表（含向量字段）cur.execute(""" CREATE TABLE IF NOT EXISTS news ( id SERIAL PRIMARY KEY, title TEXT NOT NULL, content TEXT, embedding VECTOR(768) -- 假设使用 768 维 embedding ); """)conn.commit()print(" 表 news 创建成功")3.3 插入向量数据
# 模拟生成 embedding（实际应使用模型如 sentence-transformers）defgenerate_embedding(text:str)->list:# 实际项目替换为：# from sentence_transformers import SentenceTransformer# model = SentenceTransformer('all-MiniLM-L6-v2')# return model.encode(text).tolist()np.random.seed(hash(text)%2**32)returnnp.random.rand(768).tolist()# 插入示例数据news_items=[{"title":"人工智能新突破","content":"科学家开发出新型AI算法..."},{"title":"全球气候峰会召开","content":"各国达成减排新协议..."},{"title":"量子计算进展","content":"中国实现100量子比特操控..."}]foriteminnews_items:embedding=generate_embedding(item["title"]+" "+item["content"])cur.execute("INSERT INTO news (title, content, embedding) VALUES (%s, %s, %s)",(item["title"],item["content"],embedding))conn.commit()print("向量数据插入完成")3.4 向量相似性搜索（核心功能）
1、精确搜索（无索引，小数据量可用）
defsearch_similar_news(query_text:str,limit:int=3):query_embedding=generate_embedding(query_text)# 使用余弦相似度（<=> 返回余弦距离，越小越相似）cur.execute(""" SELECT id, title, content, embedding <=> %s AS distance FROM news ORDER BY embedding <=> %s LIMIT %s; """,(query_embedding,query_embedding,limit))results=cur.fetchall()returnresults# 测试搜索results=search_similar_news("AI 技术发展")forrinresults:print(f"标题:{r['title']}, 距离:{r['distance']:.4f}")💡 距离说明：
- 余弦距离 ∈ [0, 2]，0 表示完全相同
- L2 距离 ≥ 0，0 表示相同
2、创建索引加速（大数据必备）
IVFFlat 索引（推荐入门）
# 创建 IVFFlat 索引（基于余弦距离）cur.execute(""" CREATE INDEX ON news USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100); -- lists ≈ sqrt(数据量) """)conn.commit()print("IVFFlat 索引创建成功")参数建议：
lists = min(100, sqrt(N))，N 为数据量- 支持操作符：
vector_l2_ops,vector_ip_ops,vector_cosine_ops
HNSW 索引（更高性能，PostgreSQL 16+ 或 pgvector v0.7+）
# 需 pgvector >= 0.7.0cur.execute(""" CREATE INDEX ON news USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64); """)HNSW 参数：
m: 每个节点的连接数（默认 16）ef_construction: 构建时的候选集大小（默认 64）
3.5 使用 SQLAlchemy（ORM 方式）
fromsqlalchemyimportcreate_engine,Column,Integer,Textfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmakerfrompgvector.sqlalchemyimportVector Base=declarative_base()classNews(Base):__tablename__='news'id=Column(Integer,primary_key=True)title=Column(Text)content=Column(Text)embedding=Column(Vector(768))# 创建引擎engine=create_engine("postgresql://admin:123456@localhost:5432/vector_db")Base.metadata.create_all(engine)Session=sessionmaker(bind=engine)session=Session()# 插入数据embedding=generate_embedding("测试新闻")news=News(title="测试",content="内容",embedding=embedding)session.add(news)session.commit()# 相似搜索（SQLAlchemy 语法）fromsqlalchemyimporttext query_emb=generate_embedding("查询词")results=session.execute(text(""" SELECT *, embedding <=> :query_emb AS distance FROM news ORDER BY embedding <=> :query_emb LIMIT 3 """),{"query_emb":query_emb}).fetchall()3.6 混合搜索SQL（关键词 + 向量）
SELECT*,ts_rank_cd(to_tsvector('chinese',title),plainto_tsquery('AI'))ASkeyword_score,embedding<=>$1ASvector_distanceFROMnewsWHEREto_tsvector('chinese',title)@@ plainto_tsquery('AI')ORDERBY0.7*vector_distance+0.3*(1-keyword_score)LIMIT5;需要安装
zhparser支持中文分词
3.7 完整可执行案例（单文件）
# pgvector_demo.pyimportpsycopg2importnumpyasnp DB_CONFIG={"host":"localhost","port":5432,"database":"vector_db","user":"admin","password":"123456"}defmain():conn=psycopg2.connect(**DB_CONFIG)cur=conn.cursor()# 初始化cur.execute("CREATE EXTENSION IF NOT EXISTS vector;")cur.execute(""" CREATE TABLE IF NOT EXISTS demo ( id SERIAL PRIMARY KEY, text TEXT, vec VECTOR(4) ); """)# 插入数据data=[("苹果很好吃",[0.1,0.2,0.3,0.4]),("香蕉很香甜",[0.2,0.3,0.4,0.5]),("汽车跑得快",[0.9,0.8,0.7,0.6])]fortext,vecindata:cur.execute("INSERT INTO demo (text, vec) VALUES (%s, %s)",(text,vec))conn.commit()# 搜索query_vec=[0.15,0.25,0.35,0.45]# 类似“水果”cur.execute(""" SELECT text, vec <-> %s AS distance FROM demo ORDER BY vec <-> %s LIMIT 2; """,(query_vec,query_vec))print("最相似的文本:")forrowincur.fetchall():print(f"{row[0]}(距离:{row[1]:.4f})")cur.close()conn.close()if__name__=="__main__":main()运行前确保 Docker 已启动：
dockerstart pgvector-demo python pgvector_demo.py
