为什么你的 RAG 答不准
很多团队做完 RAG 的 demo 后上线，发现效果不稳定：问题稍微换个说法就检索不到。根因通常有三条：
- 切分策略粗糙：按固定字符数硬切，把一句话拆成两半；
- 只用向量检索：语义相似但关键词不匹配，尤其对型号、代码、专有名词效果差；
- 没有重排序：向量召回的 Top-K 里前几条往往不是最相关的。
下面用 PostgreSQL + pgvector 搭一套可直接上线的方案。
一、表结构与索引
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS pg_trgm; -- 模糊匹配
CREATE TABLE chunks (
id bigserial PRIMARY KEY,
doc_id text NOT NULL,
title text,
content text NOT NULL,
tokens int,
-- 假设使用 1024 维嵌入模型
embedding vector(1024) NOT NULL,
tsv tsvector GENERATED ALWAYS AS (to_tsvector('simple', content)) STORED,
created_at timestamptz DEFAULT now()
);
-- HNSW 索引：召回快、构建慢，适合读多写少
CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- 全文索引：关键词兜底
CREATE INDEX ON chunks USING gin (tsv);
-- 元数据索引
CREATE INDEX ON chunks (doc_id);调参提示：m 越大图越密、召回越高但内存越多；查询期用 SET hnsw.ef_search = 100 在召回与延迟间权衡。
二、混合检索：向量 + 全文 + RRF 融合
-- 使用 RRF（Reciprocal Rank Fusion）融合两路召回
WITH vector_search AS (
SELECT id, ROW_NUMBER() OVER (ORDER BY embedding <=> :qvec) AS rank
FROM chunks
ORDER BY embedding <=> :qvec
LIMIT 30
),
keyword_search AS (
SELECT id, ROW_NUMBER() OVER (ORDER BY ts_rank_cd(tsv, plainto_tsquery('simple', :q)) DESC) AS rank
FROM chunks
WHERE tsv @@ plainto_tsquery('simple', :q)
LIMIT 30
)
SELECT c.id, c.title, c.content,
1.0 / (60 + vector_search.rank) * 1.0
+ 1.0 / (60 + keyword_search.rank) * 1.0 AS score
FROM chunks c
LEFT JOIN vector_search ON c.id = vector_search.id
LEFT JOIN keyword_search ON c.id = keyword_search.id
WHERE vector_search.id IS NOT NULL OR keyword_search.id IS NOT NULL
ORDER BY score DESC
LIMIT 12;RRF 的好处是不需要归一化两路的分数，只用排名，鲁棒且实现简单。
三、重排序：把 Top-12 精炼成 Top-4
import httpx
RERANK_URL = "http://localhost:8000/rerank"
async def rerank(query: str, docs: list[str], top_k: int = 4) -> list[int]:
"""返回重排序后的原始下标"""
async with httpx.AsyncClient(timeout=10) as client:
r = await client.post(RERANK_URL, json={
"model": "bge-reranker-v2-m3",
"query": query,
"documents": docs,
"top_k": top_k,
})
r.raise_for_status()
return [item["index"] for item in r.json()["results"]]
async def retrieve(query: str) -> list[dict]:
qvec = await embed(query)
rows = await db.fetch(HYBRID_SQL, {"qvec": qvec, "q": query})
idx = await rerank(query, [r["content"] for r in rows], top_k=4)
return [rows[i] for i in idx]交叉编码器（cross-encoder）会把 query 与文档拼在一起过一遍模型，精度远高于双塔向量，但代价是慢——所以它只用于重排少量候选，而不是全库检索。
四、切分策略：按语义而非字数
import re
def split_markdown(text: str, max_len: int = 512, overlap: int = 64) -> list[str]:
"""优先按标题切，其次按段落，最后才按长度兜底"""
blocks = re.split(r"\n(?=#{1,6} )", text)
chunks: list[str] = []
for b in blocks:
if len(b) <= max_len:
chunks.append(b.strip())
continue
buf = ""
for para in b.split("\n\n"):
if len(buf) + len(para) > max_len and buf:
chunks.append(buf.strip())
buf = buf[-overlap:] # 保留尾部重叠
buf += para + "\n\n"
if buf.strip():
chunks.append(buf.strip())
return [c for c in chunks if c]经验值：中文场景 300~500 字一块、10%~15% 重叠效果较好；每块额外拼上标题与上级标题，能显著提升召回。
五、评估：别靠感觉
# 用召回率@k 与答案正确率两个指标分别评估
async def evaluate(dataset: list[dict]) -> dict:
hit = 0
for case in dataset:
got = await retrieve(case["question"])
ids = {r["doc_id"] for r in got}
if case["gold_doc_id"] in ids:
hit += 1
return {"recall@4": hit / len(dataset)}准备 50~200 条真实问题 + 标注答案文档，每次改动切分/索引/重排策略后跑一遍，用数字驱动优化，而不是靠“问几句试试”。
六、工程细节
- 嵌入模型版本化：换模型要重建向量列，建议把模型名写进表字段，支持双写灰度。
- 批量写入：先
COPY导入再建索引，比逐条 insert 快一个数量级。 - 连接复用：检索是高频路径，务必使用连接池并预热。
小结
RAG 效果的差距，往往不在“换更大的模型”，而在切分是否合理、召回是否多路、是否重排。pgvector 让这套方案不需要引入额外基础设施，对已有 PostgreSQL 的团队来说是成本最低的选择。
34234