PostgreSQL + pgvector向量搜尋：2026年不用向量資料庫也能做RAG檢索
编程语言
PostgreSQL + pgvector向量搜尋：2026年不用向量資料庫也能做RAG檢索
你是不是一提到RAG就想到Pinecone、Milvus、Weaviate這些向量資料庫？為了一個簡單的語意搜尋功能就要引入一套全新的資料庫系統，維運成本翻倍，資料同步頭痛不已？2026年，PostgreSQL的pgvector擴充套件已經足夠成熟——你現有的PostgreSQL就能做向量搜尋，不需要額外的向量資料庫。
背景知識
向量搜尋基礎
向量搜尋的核心是將文字、圖像等非結構化資料轉化為高維向量（Embedding），然後透過計算向量間的距離（餘弦相似度、歐氏距離等）找到語意最相近的結果。
| 概念 | 說明 | 範例 |
|---|---|---|
| Embedding | 將資料映射為向量 | text-embedding-3-large輸出3072維向量 |
| 餘弦相似度 | 向量夾角餘弦值，範圍[-1,1] | 1表示完全相同 |
| 歐氏距離 | 向量間的直線距離 | 0表示完全相同 |
| 內積 | 向量點乘 | 適合歸一化向量 |
| ANN搜尋 | 近似最近鄰，犧牲精度換速度 | HNSW、IVFFlat |
pgvector支援的索引型別
| 索引 | 演算法 | 建構速度 | 查詢速度 | 召回率 | 記憶體佔用 | 適用場景 |
|---|---|---|---|---|---|---|
| HNSW | 圖索引 | 慢 | 快 | 高 | 高 | 中小規模，高召回 |
| IVFFlat | 倒排索引 | 快 | 中 | 中 | 低 | 大規模，可接受精度損失 |
問題分析
為什麼不用專門的向量資料庫？
- 維運成本：新增一套資料庫意味著備份、監控、升級、高可用全都要額外處理
- 資料一致性：業務資料在PostgreSQL，向量資料在向量資料庫，雙寫一致性難以保證
- 查詢能力：向量資料庫的過濾能力有限，無法做複雜的關聯查詢
- 遷移風險：向量資料庫市場碎片化，Pinecone/Milvus/Weaviate API不相容
pgvector讓你在PostgreSQL內完成向量搜尋+業務過濾+關聯查詢，一站式解決。
分步實操
步驟1：安裝pgvector擴充套件
-- 連線PostgreSQL
psql -U postgres -d mydb
-- 安裝擴充套件
CREATE EXTENSION IF NOT EXISTS vector;
-- 驗證版本
SELECT extversion FROM pg_extension WHERE extname = 'vector';
-- 0.8.0+
Docker方式：
# docker-compose.yml
services:
postgres:
image: pgvector/pgvector:pg16
environment:
POSTGRES_DB: mydb
POSTGRES_USER: postgres
POSTGRES_PASSWORD: password
ports:
- "5432:5432"
volumes:
- pgdata:/var/lib/postgresql/data
volumes:
pgdata:
步驟2：建立向量表
CREATE TABLE documents (
id SERIAL PRIMARY KEY,
title TEXT NOT NULL,
content TEXT NOT NULL,
source TEXT,
category TEXT,
created_at TIMESTAMPTZ DEFAULT NOW(),
embedding vector(1536)
);
-- 建立HNSW索引（推薦用於高召回場景）
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- 建立IVFFlat索引（推薦用於大規模資料）
-- CREATE INDEX ON documents
-- USING ivfflat (embedding vector_cosine_ops)
-- WITH (lists = 100);
步驟3：生成和儲存向量
import psycopg2
from openai import OpenAI
client = OpenAI()
conn = psycopg2.connect("dbname=mydb user=postgres password=password")
cur = conn.cursor()
def generate_embedding(text: str) -> list[float]:
response = client.embeddings.create(
model="text-embedding-3-small",
input=text,
)
return response.data[0].embedding
def insert_document(title: str, content: str, source: str, category: str):
full_text = f"{title}\n{content}"
embedding = generate_embedding(full_text)
cur.execute(
"INSERT INTO documents (title, content, source, category, embedding) VALUES (%s, %s, %s, %s, %s)",
(title, content, source, category, str(embedding)),
)
conn.commit()
insert_document(
"K8s Gateway API指南",
"Gateway API是Kubernetes新一代流量管理標準...",
"toolsku.dev/blog",
"DevOps",
)
步驟4：向量搜尋查詢
-- 餘弦相似度搜尋
SELECT id, title, content, 1 - (embedding <=> $1) AS similarity
FROM documents
WHERE 1 - (embedding <=> $1) > 0.7
ORDER BY embedding <=> $1
LIMIT 10;
-- 帶過濾條件的向量搜尋
SELECT id, title, 1 - (embedding <=> $1) AS similarity
FROM documents
WHERE category = 'DevOps'
AND 1 - (embedding <=> $1) > 0.7
ORDER BY embedding <=> $1
LIMIT 10;
步驟5：混合查詢（向量+全文）
-- 建立全文搜尋索引
ALTER TABLE documents ADD COLUMN tsv tsvector
GENERATED ALWAYS AS (to_tsvector('simple', coalesce(title, '') || ' ' || coalesce(content, ''))) STORED;
CREATE INDEX ON documents USING gin(tsv);
-- 混合查詢：向量搜尋 + 全文搜尋 + RRF融合
WITH vector_results AS (
SELECT id, title,
ROW_NUMBER() OVER (ORDER BY embedding <=> $1) AS vector_rank,
1 - (embedding <=> $1) AS vector_score
FROM documents
ORDER BY embedding <=> $1
LIMIT 50
),
text_results AS (
SELECT id, title,
ROW_NUMBER() OVER (ORDER BY ts_rank(tsv, plainto_tsquery('simple', $2)) DESC) AS text_rank,
ts_rank(tsv, plainto_tsquery('simple', $2)) AS text_score
FROM documents
WHERE tsv @@ plainto_tsquery('simple', $2)
LIMIT 50
),
combined AS (
SELECT
COALESCE(v.id, t.id) AS id,
COALESCE(v.title, t.title) AS title,
COALESCE(1.0 / (60 + v.vector_rank), 0) +
COALESCE(1.0 / (60 + t.text_rank), 0) AS rrf_score
FROM vector_results v
FULL OUTER JOIN text_results t ON v.id = t.id
)
SELECT id, title, rrf_score
FROM combined
ORDER BY rrf_score DESC
LIMIT 10;
完整程式碼：RAG檢索系統
import psycopg2
from openai import OpenAI
from dataclasses import dataclass
@dataclass
class SearchResult:
id: int
title: str
content: str
score: float
source: str
class PgVectorRAG:
def __init__(self, db_url: str, openai_api_key: str):
self.conn = psycopg2.connect(db_url)
self.client = OpenAI(api_key=openai_api_key)
def embed(self, text: str) -> list[float]:
response = self.client.embeddings.create(
model="text-embedding-3-small",
input=text,
)
return response.data[0].embedding
def search(
self,
query: str,
top_k: int = 10,
threshold: float = 0.6,
category: str | None = None,
use_hybrid: bool = True,
) -> list[SearchResult]:
query_embedding = self.embed(query)
embedding_str = str(query_embedding)
if use_hybrid:
return self._hybrid_search(embedding_str, query, top_k, category)
return self._vector_search(embedding_str, top_k, threshold, category)
def _vector_search(
self,
embedding_str: str,
top_k: int,
threshold: float,
category: str | None,
) -> list[SearchResult]:
sql = """
SELECT id, title, content, source,
1 - (embedding <=> %s::vector) AS similarity
FROM documents
WHERE 1 - (embedding <=> %s::vector) > %s
"""
params: list = [embedding_str, embedding_str, threshold]
if category:
sql += " AND category = %s"
params.append(category)
sql += " ORDER BY embedding <=> %s::vector LIMIT %s"
params.extend([embedding_str, top_k])
cur = self.conn.cursor()
cur.execute(sql, params)
results = cur.fetchall()
return [
SearchResult(id=r[0], title=r[1], content=r[2], source=r[3], score=r[4])
for r in results
]
def _hybrid_search(
self,
embedding_str: str,
query_text: str,
top_k: int,
category: str | None,
) -> list[SearchResult]:
category_filter = "AND category = %s" if category else ""
params: list = [embedding_str, embedding_str]
if category:
params.append(category)
params.extend([embedding_str, query_text])
if category:
params.append(category)
params.extend([embedding_str, top_k])
sql = f"""
WITH vector_results AS (
SELECT id, title, content, source,
ROW_NUMBER() OVER (ORDER BY embedding <=> %s::vector) AS v_rank
FROM documents
WHERE 1 - (embedding <=> %s::vector) > 0.5
{category_filter}
ORDER BY embedding <=> %s::vector
LIMIT 50
),
text_results AS (
SELECT id, title, content, source,
ROW_NUMBER() OVER (ORDER BY ts_rank(tsv, plainto_tsquery('simple', %s)) DESC) AS t_rank
FROM documents
WHERE tsv @@ plainto_tsquery('simple', %s)
{category_filter}
LIMIT 50
),
combined AS (
SELECT
COALESCE(v.id, t.id) AS id,
COALESCE(v.title, t.title) AS title,
COALESCE(v.content, t.content) AS content,
COALESCE(v.source, t.source) AS source,
COALESCE(1.0 / (60 + v.v_rank), 0) +
COALESCE(1.0 / (60 + t.t_rank), 0) AS rrf_score
FROM vector_results v
FULL OUTER JOIN text_results t ON v.id = t.id
)
SELECT id, title, content, source, rrf_score
FROM combined
ORDER BY rrf_score DESC
LIMIT %s
"""
cur = self.conn.cursor()
cur.execute(sql, params)
results = cur.fetchall()
return [
SearchResult(id=r[0], title=r[1], content=r[2], source=r[3], score=r[4])
for r in results
]
def generate_answer(self, query: str, top_k: int = 5) -> str:
results = self.search(query, top_k=top_k, use_hybrid=True)
context = "\n\n".join(f"[{r.title}]\n{r.content}" for r in results)
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "基於以下參考資料回答問題。如果參考資料中沒有相關資訊，請說明。\n\n" + context},
{"role": "user", "content": query},
],
)
return response.choices[0].message.content
if __name__ == "__main__":
rag = PgVectorRAG(
db_url="dbname=mydb user=postgres password=password",
openai_api_key="sk-xxx",
)
answer = rag.generate_answer("如何配置K8s Gateway API？")
print(answer)
避坑指南
坑1：HNSW索引建構時間過長
現象：百萬級資料建立HNSW索引耗時數小時，期間表被鎖定。
解決：使用CONCURRENTLY選項（pgvector 0.7+）或在低峰期建構。降低ef_construction引數（預設64→32）可加速建構但略微降低召回率。分批建構後合併。
坑2：向量維度不匹配導致插入失敗
現象：ERROR: expected 1536 dimensions, not 768。
解決：確保Embedding模型輸出維度與表定義的vector(N)一致。text-embedding-3-small輸出1536維，text-embedding-3-large輸出3072維。修改表定義：ALTER TABLE documents ALTER COLUMN embedding TYPE vector(768)。
坑3：IVFFlat索引在資料更新後召回率驟降
現象：新增大量資料後，搜尋結果明顯不準。
解決：IVFFlat的聚類中心在索引建立時確定，新增資料會導致偏移。需要定期重建索引：REINDEX INDEX documents_embedding_idx。或使用HNSW索引（增量友好）。
坑4：混合查詢效能差
現象：向量+全文混合查詢耗時超過1秒。
解決：確保兩個搜尋分別走索引，使用EXPLAIN ANALYZE確認。限制各自回傳的候選集大小（各50條），RRF融合在少量資料上很快。避免在混合查詢中使用SELECT *。
坑5：pgvector佔用記憶體過大
現象：HNSW索引記憶體佔用是原始資料的2-3倍。
解決：調整m引數（預設16→8）降低圖連線度，減少記憶體佔用但犧牲一些召回率。使用IVFFlat替代HNSW。考慮使用半精度向量：vector(1536) → halfvec(1536)（pgvector 0.7+）。
報錯排查
| 序號 | 報錯資訊 | 原因 | 解決方法 |
|---|---|---|---|
| 1 | extension "vector" not available |
pgvector未安裝 | 使用pgvector/pgvector Docker映像或編譯安裝 |
| 2 | expected 1536 dimensions, not 768 |
向量維度不匹配 | 檢查Embedding模型，修改表定義的維度 |
| 3 | operator does not exist: vector <=> unknown |
缺少型別轉換 | 使用%s::vector顯式轉換 |
| 4 | index row size exceeds maximum |
HNSW索引行過大 | 降低m引數，或使用halfvec |
| 5 | cannot create index concurrently |
pgvector版本不支援 | 升級到0.7+或使用非並行方式 |
| 6 | IVFFlat recall is low after inserts |
新資料偏離聚類中心 | 定期REINDEX或改用HNSW |
| 7 | out of memory during index build |
建構索引記憶體不足 | 增加work_mem，分批建構 |
| 8 | invalid input syntax for type vector |
向量格式錯誤 | 確保使用[0.1, 0.2, ...]格式 |
| 9 | permission denied for schema public |
使用者無建立擴充套件許可權 | 使用超級使用者執行CREATE EXTENSION |
| 10 | query timeout on hybrid search |
混合查詢過慢 | 限制候選集大小，新增索引，檢查EXPLAIN |
進階最佳化
1. 半精度向量節省儲存
-- 使用halfvec（半精度浮點），儲存減半
ALTER TABLE documents ALTER COLUMN embedding TYPE halfvec(1536);
-- 建立HNSW索引
CREATE INDEX ON documents
USING hnsw (embedding halfvec_cosine_ops)
WITH (m = 16, ef_construction = 64);
2. 分割槽加速大規模搜尋
CREATE TABLE documents_partitioned (
LIKE documents INCLUDING DEFAULTS
) PARTITION BY LIST (category);
CREATE TABLE documents_devops PARTITION OF documents_partitioned FOR VALUES IN ('DevOps');
CREATE TABLE documents_frontend PARTITION OF documents_partitioned FOR VALUES IN ('前端工程');
CREATE TABLE documents_ai PARTITION OF documents_partitioned FOR VALUES IN ('AI與大資料');
3. 查詢時調整HNSW搜尋精度
-- 提高ef_search提升召回率（預設40）
SET hnsw.ef_search = 200;
-- 降低ef_search加速查詢
SET hnsw.ef_search = 20;
4. 批量向量生成最佳化
import psycopg2
from openai import OpenAI
import math
client = OpenAI()
def batch_embed(texts: list[str], batch_size: int = 100) -> list[list[float]]:
embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
response = client.embeddings.create(
model="text-embedding-3-small",
input=batch,
)
embeddings.extend([d.embedding for d in response.data])
return embeddings
def bulk_insert_documents(docs: list[dict]):
texts = [f"{d['title']}\n{d['content']}" for d in docs]
embeddings = batch_embed(texts)
cur = conn.cursor()
for doc, emb in zip(docs, embeddings):
cur.execute(
"INSERT INTO documents (title, content, source, category, embedding) VALUES (%s, %s, %s, %s, %s)",
(doc['title'], doc['content'], doc['source'], doc['category'], str(emb)),
)
conn.commit()
對比分析
| 維度 | pgvector | Pinecone | Milvus | Weaviate | Qdrant |
|---|---|---|---|---|---|
| 部署方式 | PostgreSQL擴充套件 | 託管 | 自託管/託管 | 自託管/託管 | 自託管/託管 |
| 維運成本 | 極低（複用PG） | 高（按查詢計費） | 高 | 高 | 中 |
| 混合查詢 | 原生SQL | 有限過濾 | 有限過濾 | GraphQL過濾 | 過濾器 |
| 事務支援 | ACID | 無 | 無 | 無 | 無 |
| 最大維度 | 16000 | 20000 | 32768 | 65535 | 65535 |
| 索引型別 | HNSW/IVFFlat | 自有 | HNSW/IVF/DiskANN | HNSW | HNSW |
| 水平擴充套件 | 需Citus | 自動 | 自動 | 自動 | 自動 |
| 適用規模 | 百萬級 | 十億級 | 十億級 | 億級 | 億級 |
| 資料一致性 | 強一致 | 最終一致 | 最終一致 | 最終一致 | 最終一致 |
總結展望
總結：對於大多數RAG應用場景，PostgreSQL + pgvector已經完全夠用。它的核心優勢是零額外維運成本、原生SQL混合查詢、ACID事務保證。百萬級資料量下HNSW索引效能優秀，混合查詢（向量+全文+RRF）召回率遠超純向量搜尋。只有當資料量達到億級或需要水平擴充套件時，才需要考慮專門的向量資料庫。建議從pgvector起步，按需演進。
線上工具推薦
- JSON格式化（API回應除錯）：/zh-TW/json/format
- Base64編解碼（向量資料處理）：/zh-TW/encode/base64
- curl轉程式碼（API測試）：/zh-TW/dev/curl-to-code
本站提供瀏覽器本地工具，免註冊即可試用 →
#PostgreSQL#pgvector#向量搜索#语义检索#RAG#HNSW#IVFFlat#混合查询