© 2026 志趣 ZhiQu AI 开发者知识沉淀社区 · 记录每一次踩坑，沉淀每一个解决方案
RAG 实战指南：从零搭建知识库（pgvector + 嵌入模型 + 混合搜索） — 志趣 ZhiQu
RAG 实战指南：从零搭建知识库（pgvector + 嵌入模型 + 混合搜索） RAG（Retrieval-Augmented Generation）是 2025-2026 年 AI 应用落地最广泛的架构模式。它的核心思想很简单：让 AI 先查资料，再回答问题 。但真正搭建一个生产可用的 RAG 系统，涉及向量数据库选型、嵌入模型对比、分块策略、混合搜索、重排序等至少 8 个关键决策。本文以 pgvector + PostgreSQL 为底座，带你从零搭建一个完整的 RAG 知识库——包括「为什么选它」「怎么写代码」「怎么调优」。
目录：
RAG 核心原理与架构 pgvector：PostgreSQL 的向量能力 嵌入模型选型指南 文档分块策略 混合搜索架构 从零实现 RAG 知识库 检索质量优化 生产环境考量 常见坑点排查
1. RAG 核心原理与架构
1.1 为什么需要 RAG
大语言模型有两个天然局限：
知识截止日期 ：训练数据停在某个时间点，不知道最新信息幻觉问题 ：被问到不知道的事，模型可能「编造」答案
RAG 解决这两个问题的方式非常直觉化：给模型一本参考书，让它先查后答 。
┌──────────────────────────────────────────────────────┐
│ RAG 工作流程 │
│ │
│ 用户提问 ──→ 嵌入向量化 ──→ 向量相似度搜索 │
│ │ │
│ ↓ │
│ 返回结果 ←── LLM 生成 ←── Top-K 相关文档 │
│ │
│ 1. Embedding: 把问题转成向量 │
│ 2. Retrieval: 在知识库中找到最相似的文档片段 │
│ 3. Augmented: 把检索结果拼入 Prompt │
│ 4. Generation: LLM 基于上下文生成答案 │
└──────────────────────────────────────────────────────┘
1.2 RAG vs 微调 vs 长上下文
这是最常见的选型困惑。三者的适用场景完全不同：
版权声明
本文原创发布于 zhiqu.ac ，未经书面许可禁止全文转载、采集、商用；转载必须完整标注原文链接、作者。
知识更新 即时（改数据库即可） 需重新训练 需重新输入 可解释性 高（能看到引用的文档） 低（黑盒） 中（可追溯） 成本 嵌入+向量存储 训练 GPU 成本高 Token 成本高 适用规模 百万级文档 特定领域风格适配 单文档深度分析 幻觉控制 强（有明确来源） 一般 弱（长文可能忽略细节） 最佳场景 企业知识库、客服 领域术语/格式适配 代码库审查、论文分析
结论 ：大多数企业场景选 RAG，特定风格需求补充微调，单文档深度分析用长上下文。本文专注 RAG。
1.3 生产级 RAG 架构全景 ┌──────────────┐
│ 用户提问 │
└──────┬───────┘
│
┌────────────▼────────────┐
│ Query 预处理 │
│ - 改写/扩展 │
│ - 意图识别 │
└────────────┬────────────┘
│
┌──────────────────┼──────────────────┐
│ │ │
┌────────▼───────┐ ┌──────▼──────┐ ┌───────▼────────┐
│ 向量检索 │ │ 关键词检索 │ │ 元数据过滤 │
│ (语义相似度) │ │ (BM25) │ │ (日期/标签等) │
│ pgvector ◄───→ │ │ ts_vector │ │ WHERE 子句 │
└────────┬───────┘ └──────┬──────┘ └───────┬────────┘
│ │ │
└──────────────────┼──────────────────┘
│
┌────────────▼────────────┐
│ 结果融合 & 重排序 │
│ - RRF (Reciprocal Rank) │
│ - Cross-encoder Rerank │
└────────────┬────────────┘
│
┌────────────▼────────────┐
│ LLM 生成 │
│ - Prompt 组装 │
│ - 引用标注 │
└────────────┬────────────┘
│
┌──────▼───────┐
│ 返回答案 │
│ + 引用来源 │
└──────────────┘
2. pgvector：PostgreSQL 的向量能力
2.1 为什么选 pgvector 向量数据库选型是 RAG 的第一个关键决策。主流选项：
方案 优势 劣势 适用 pgvector 与 PostgreSQL 一体、ACID、已有 DBA 熟悉 超大集群不如专用方案 ⭐ 大多数项目 Pinecone 全托管、高性能 贵、数据不在自己手里 不想管运维的团队 Milvus 分布式、十亿级 运维复杂、资源消耗大 超大规模 Qdrant Rust 实现、高性能 生态较新 追求极致性能 Weaviate 混合搜索原生支持 Go 实现、团队需适应 需要开箱即用的混合搜索 Chroma 轻量、Python 原生 不适合生产 原型验证
零额外运维 ：你已经有 PostgreSQL 了，加个扩展即可，不需要再维护一套数据库ACID 保障 ：向量和业务数据在同一个事务中，不会出现「向量写入了但元数据丢失」SQL 表达力 ：用熟悉的 SQL 做向量搜索，配合 WHERE、JOIN、子查询成本低 ：不需要额外付费的向量数据库服务pgvector 已成熟 ：0.5.0+ 支持 HNSW 索引、并行构建、半精度向量
2.2 安装 pgvector # docker-compose.yml
services:
postgres:
image: pgvector/pgvector:pg16
# 这个镜像已经内置 pgvector，无需手动安装
environment:
POSTGRES_USER: zhiqu
POSTGRES_PASSWORD: your_password
POSTGRES_DB: zhiqu
ports:
- "5432:5432"
volumes:
- pgdata:/var/lib/postgresql/data
command: |
-c shared_preload_libraries=vector
-c max_connections=100
-c shared_buffers=256MB
volumes:
pgdata:
-- 安装扩展（需要超级用户权限）
CREATE EXTENSION IF NOT EXISTS vector;
-- 验证安装
SELECT extname, extversion FROM pg_extension WHERE extname = 'vector';
-- extname | extversion
-- ---------+------------
-- vector | 0.8.0
# 安装编译依赖
sudo apt-get install postgresql-server-dev-16 git make gcc
# 克隆并编译
cd /tmp
git clone --branch v0.8.0 https://github.com/pgvector/pgvector.git
cd pgvector
make
sudo make install
2.3 创建向量表 -- 文档块表：存储分块后的文档片段及其向量
CREATE TABLE document_chunks (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
document_id UUID NOT NULL, -- 所属文档 ID
content TEXT NOT NULL, -- 分块文本内容
chunk_index INT NOT NULL, -- 分块序号（第几块）
embedding vector(1536), -- OpenAI text-embedding-ada-002 为 1536 维
-- 或者用 vector(768) 对应 BGE-M3 / text-embedding-3-small
-- 或者用 vector(1024) 对应 Jina embeddings v3
-- 元数据字段（用于过滤）
source VARCHAR(500), -- 来源 URL / 文件名
title VARCHAR(500), -- 文档标题
tags TEXT[], -- 标签数组
created_at TIMESTAMPTZ DEFAULT NOW(),
-- 全文搜索字段（混合搜索用）
tsv TSVECTOR -- PostgreSQL 全文搜索向量
);
-- 向量索引（HNSW）
CREATE INDEX ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
-- 全文搜索索引
CREATE INDEX ON document_chunks USING GIN (tsv);
2.4 向量索引对比 特性 IVFFlat HNSW 建索引速度 快 慢（但支持并行） 查询速度 中等 快（1-5ms） 召回率 ~90%（近似） ~99%（接近精确） 内存占用 低 较高 构建参数 listsm, ef_construction查询参数 probesef_search推荐场景 原型/百万级以下 生产环境 ⭐
-- m: 每层每个节点的最大连接数（默认 16，范围 2-100）
-- 越大 → 召回率越高 → 索引越大 → 构建越慢
-- 推荐：1536 维用 16，768 维用 24，更小维度用 32
-- ef_construction: 构建时动态候选列表大小（默认 64，范围 4-1000）
-- 越大 → 索引质量越高 → 构建越慢
-- 推荐：100-200（平衡），500-1000（追求极致召回率）
CREATE INDEX ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
-- ef_search 控制查询时的搜索广度（默认 40）
-- 越大 → 召回率越高 → 查询越慢
SET hnsw.ef_search = 100;
-- 然后执行查询
SELECT id, content, 1 - (embedding <=> '[0.1, 0.2, ...]') AS similarity
FROM document_chunks
ORDER BY embedding <=> '[0.1, 0.2, ...]'
LIMIT 10;
2.5 三种距离度量 -- 1. 余弦距离 (Cosine Distance)：最常用，推荐 ⭐
-- 范围 [0, 2]，0 = 完全相同方向，2 = 完全相反
SELECT embedding <=> query_vector AS cosine_distance
FROM document_chunks
ORDER BY embedding <=> query_vector
LIMIT 10;
-- 2. 欧几里得距离 (L2 Distance)：
-- 范围 [0, ∞)，对向量长度敏感
SELECT embedding <-> query_vector AS l2_distance
FROM document_chunks
ORDER BY embedding <-> query_vector
LIMIT 10;
-- 3. 内积 (Inner Product)：适用于归一化向量
-- 范围 (-∞, ∞)，值越大越相似
SELECT embedding <#> query_vector AS neg_inner_product
FROM document_chunks
ORDER BY embedding <#> query_vector
LIMIT 10;
选型建议 ：嵌入模型输出的向量通常已经归一化（长度=1），此时余弦距离和内积等价。默认选余弦距离 ，除非你的嵌入模型特别声明了其他度量方式。
3. 嵌入模型选型指南
3.1 主流嵌入模型对比 模型 维度 最大输入 中文效果 价格（每 1M token） 部署方式 OpenAI text-embedding-3-small 512/1536 8191 ⭐⭐⭐ $0.02 API OpenAI text-embedding-3-large 256/1024/3072 8191 ⭐⭐⭐⭐ $0.13 API BGE-M3 (BAAI) 1024 8192 ⭐⭐⭐⭐⭐ 免费 本地 / HF BGE-large-zh-v1.5 1024 512 ⭐⭐⭐⭐⭐ 免费 本地 Jina embeddings v3 1024 8192 ⭐⭐⭐⭐ $0.02 API / 本地 Cohere embed-v3 1024 512 ⭐⭐⭐ $0.10 API mGTE (Alibaba) 768 8192 ⭐⭐⭐⭐ 免费 本地 stella-base-zh-v3-1792d 1792 512 ⭐⭐⭐⭐ 免费 本地
3.2 选型决策树 需要多语言混合检索？
├── YES → BGE-M3 ⭐（支持中英日韩等 100+ 语言，且支持稀疏+稠密双向量）
└── NO → 只用中文？
├── YES → 有 GPU？
│ ├── YES → BGE-large-zh-v1.5 或 stella-base-zh-v3-1792d
│ └── NO → 不想管运维 → OpenAI text-embedding-3-small
│ └── 预算充足 → text-embedding-3-large
└── NO → 只用英文 → OpenAI text-embedding-3-small（性价比最高）
3.3 在生产环境中使用 BGE-M3 # embedding_service.py
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List
class EmbeddingService:
"""嵌入服务：统一接口，支持本地模型和 API 切换"""
def __init__(self, model_name: str = "BAAI/bge-m3"):
self.model_name = model_name
self.model = None
self.dimension = 1024 # BGE-M3 输出 1024 维
def _lazy_load(self):
"""延迟加载模型（首次调用时才加载到 GPU/CPU）"""
if self.model is None:
self.model = SentenceTransformer(
self.model_name,
device="cuda" if self._has_gpu() else "cpu"
)
return self.model
def _has_gpu(self) -> bool:
try:
import torch
return torch.cuda.is_available()
except ImportError:
return False
def encode(self, texts: List[str]) -> List[List[float]]:
"""将文本列表转换为向量列表"""
model = self._lazy_load()
# BGE-M3 特殊处理：查询时加前缀可以提升效果
# 参考 https://huggingface.co/BAAI/bge-m3
embeddings = model.encode(
texts,
normalize_embeddings=True, # L2 归一化，确保余弦距离=内积
show_progress_bar=False,
batch_size=32, # 根据 GPU 显存调整
)
return embeddings.tolist()
def encode_query(self, query: str) -> List[float]:
"""对查询文本做嵌入（可加 query 前缀）"""
# BGE 系列模型：query 和 passage 用不同前缀
return self.encode([f"为这个查询找到相关文档：{query}"])[0]
def encode_document(self, text: str) -> List[float]:
"""对文档文本做嵌入"""
return self.encode([text])[0]
def encode_batch(self, texts: List[str], batch_size: int = 64) -> List[List[float]]:
"""批量嵌入，适合文档入库场景"""
model = self._lazy_load()
embeddings = model.encode(
texts,
normalize_embeddings=True,
show_progress_bar=True,
batch_size=batch_size,
)
return embeddings.tolist()
3.4 API 嵌入服务（OpenAI 兼容） // openai-embedding.service.ts
import OpenAI from 'openai';
export class OpenAIEmbeddingService {
private client: OpenAI;
constructor() {
this.client = new OpenAI({
apiKey: process.env.OPENAI_API_KEY,
baseURL: process.env.OPENAI_BASE_URL || 'https://api.openai.com/v1',
});
}
async embedTexts(
texts: string[],
model: string = 'text-embedding-3-small',
dimensions?: number, // 支持 Matryoshka 降维
): Promise {
const response = await this.client.embeddings.create({
model,
input: texts,
dimensions, // 如 512（从 1536 降到 512，省存储保精度）
});
return response.data
.sort((a, b) => a.index - b.index)
.map(d => d.embedding);
}
async embedQuery(query: string): Promise {
return (await this.embedTexts([query]))[0];
}
}
4. 文档分块策略
4.1 为什么分块这么重要 分块（Chunking）是 RAG 系统中最被低估的环节。分块太大，检索精度低（噪声多）；分块太小，缺少上下文（语义不完整）。好的分块策略 = RAG 系统一半的成功。
4.2 五种分块策略 ┌─────────────────────────────────────────────────────────┐
│ 分块策略全景 │
│ │
│ 1. 固定长度分块 │
│ │──512 tokens──│──512 tokens──│──512 tokens──│ │
│ 简单但粗暴，常把句子拦腰截断 │
│ │
│ 2. 句子级分块 │
│ │──句子1──│──句子2+3──│──句子4──│──句子5+6──│ │
│ 在句号/换行处断开，保持句子完整 │
│ │
│ 3. 段落级分块 │
│ │────── 段落1 ──────│────── 段落2 ──────│ │
│ 自然边界，但段落长度不均匀 │
│ │
│ 4. 语义分块（推荐 ⭐） │
│ │── 概念A ──│── 概念B ──│──── 概念C ────│ │
│ 基于语义相似度切分，相似句子聚在一起 │
│ │
│ 5. 递归分块 │
│ 先用段落→如果太长用句子→如果太长用固定长度 │
│ 兼顾完整性和长度控制 │
│ │
│ 6. 父子文档（推荐 ⭐） │
│ 父文档：完整段落（大块，用于上下文补充） │
│ 子文档：精细分块（小块，用于检索命中） │
│ 检索时返回子块 → 补充父块上下文 │
└─────────────────────────────────────────────────────────┘
4.3 实战分块器实现 # chunker.py
import re
from typing import List, Tuple
from dataclasses import dataclass
@dataclass
class Chunk:
content: str
chunk_index: int
start_char: int
end_char: int
title: str = ""
class RecursiveChunker:
"""递归分块器：自上而下尝试最佳分界点"""
def __init__(
self,
chunk_size: int = 512, # 目标分块大小（tokens，约等于中文字数×1.5）
chunk_overlap: int = 64, # 相邻分块之间的重叠量
separators: List[str] = None,
):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
# 优先级从高到低：段落 → 换行 → 句子 → 逗号 → 字符
self.separators = separators or [
"\n\n", # 段落分隔
"\n", # 换行
"。", # 中文句号
"；", # 中文分号
"？", # 中文问号
"！", # 中文感叹号
". ", # 英文句号
", ", # 英文逗号
" ", # 空格（英文词边界）
"", # 逐字分割（最后手段）
]
def split_text(self, text: str, title: str = "") -> List[Chunk]:
"""将文本拆分为多个分块"""
chunks = self._recursive_split(text, self.separators)
return [
Chunk(
content=chunk,
chunk_index=i,
start_char=text.find(chunk),
end_char=text.find(chunk) + len(chunk),
title=title,
)
for i, chunk in enumerate(chunks)
]
def _recursive_split(self, text: str, separators: List[str]) -> List[str]:
"""递归分割核心逻辑"""
if len(text) <= self.chunk_size:
return [text] if text.strip() else []
# 尝试用当前优先级最高的分隔符
separator = separators[0]
remaining_separators = separators[1:]
if separator == "":
# 最后手段：逐字切分
return self._split_by_chars(text)
splits = text.split(separator)
# 合并短片段，避免产生大量琐碎分块
merged = self._merge_splits(splits, separator)
chunks = []
for segment in merged:
if len(segment) <= self.chunk_size:
if segment.strip():
chunks.append(segment)
else:
# 该片段仍太长，用下一级分隔符递归处理
chunks.extend(
self._recursive_split(segment, remaining_separators)
)
return chunks
def _merge_splits(self, splits: List[str], separator: str) -> List[str]:
"""合并短片段并加入重叠"""
merged = []
current = ""
for split in splits:
candidate = current + (separator if current else "") + split
if len(candidate) <= self.chunk_size:
current = candidate
else:
if current:
merged.append(current)
# 重叠策略：取出 current 的最后 chunk_overlap 个字符作为新片段的开头
if self.chunk_overlap > 0 and len(current) > self.chunk_overlap:
current = current[-self.chunk_overlap:] + separator + split
else:
current = split
else:
# 单个片段就超过限制，必须硬切
merged.append(split[:self.chunk_size])
current = split[self.chunk_size - self.chunk_overlap:]
if current:
merged.append(current)
return merged
def _split_by_chars(self, text: str) -> List[str]:
"""按字符硬切（最后手段）"""
chunks = []
for i in range(0, len(text), self.chunk_size - self.chunk_overlap):
chunk = text[i : i + self.chunk_size]
if chunk.strip():
chunks.append(chunk)
return chunks
4.4 分块大小实验数据 文档类型 推荐分块大小 推荐重叠 原因 FAQ / 短问答 128-256 tokens 0-32 问题和答案通常很短 技术文档 512-768 tokens 64-128 需要足够的上下文理解技术概念 长文章/论文 768-1024 tokens 128-256 段落长，上下文重要 代码 512-1024 tokens 64-128 函数可能很长 法律/合同 256-512 tokens 64 条款边界清晰，不宜跨条款合并
经验法则 ：从 512 tokens + 64 重叠开始，跑 50 条真实查询，看检索命中率，再迭代调整。
5. 混合搜索架构
5.1 为什么混合搜索 用户搜索：「Claude Code 如何配置 MCP？」
向量搜索结果：
1. ✅ "MCP 配置指南"（语义匹配）
2. ❌ "VS Code MCP 插件推荐"（MCP 相关但不是 Claude Code）
3. ❌ "MCP 协议详解"（太宽泛）
4. ✅ "Claude Code MCP 踩坑记录"（好结果但排在第 4）
关键词搜索结果（BM25）：
1. ✅ "Claude Code MCP 踩坑记录"（精确匹配关键词）
2. ✅ "Claude Code 配置指南"（匹配「配置」）
3. ✅ "MCP 配置指南"（匹配「MCP」「配置」）
向量搜索擅长语义，但会漏掉精确关键词匹配。关键词搜索精确，但不懂同义词。 混合搜索 = 两个都要。
5.2 混合搜索架构实现 // hybrid-search.service.ts
import { PrismaService } from '../prisma/prisma.service';
import { EmbeddingService } from './embedding.service';
interface SearchResult {
id: string;
content: string;
title: string;
score: number; // 融合后的最终分数
vectorScore: number; // 向量相似度分数
textScore: number; // 全文搜索分数
}
export class HybridSearchService {
constructor(
private prisma: PrismaService,
private embedding: EmbeddingService,
) {}
async search(
query: string,
topK: number = 10,
filters?: {
tags?: string[];
source?: string;
dateRange?: { start: Date; end: Date };
},
): Promise {
// Step 1: 生成查询向量
const queryVector = await this.embedding.embedQuery(query);
// Step 2: 并行执行向量搜索和全文搜索
const [vectorResults, textResults] = await Promise.all([
this.vectorSearch(queryVector, topK * 2, filters), // 多取一些，给融合留空间
this.textSearch(query, topK * 2, filters),
]);
// Step 3: RRF 融合
return this.reciprocalRankFusion(vectorResults, textResults, topK);
}
private async vectorSearch(
queryVector: number[],
limit: number,
filters?: any,
): Promise {
// 构建向量搜索 SQL
const vectorStr = `[${queryVector.join(',')}]`;
const results = await this.prisma.$queryRawUnsafe>(
`SELECT
id, content, title,
1 - (embedding <=> $1::vector) AS vector_score
FROM document_chunks
WHERE 1=1
${filters?.tags?.length ? `AND tags && $2::text[]` : ''}
ORDER BY embedding <=> $1::vector
LIMIT $3`,
vectorStr,
filters?.tags || [],
limit,
);
return results.map(r => ({
id: r.id,
content: r.content,
title: r.title,
score: r.vector_score,
vectorScore: r.vector_score,
textScore: 0,
}));
}
private async textSearch(
query: string,
limit: number,
filters?: any,
): Promise {
// 使用 PostgreSQL 全文搜索
const results = await this.prisma.$queryRawUnsafe>(
`SELECT
id, content, title,
ts_rank(tsv, plainto_tsquery('chinese', $1)) AS text_score
FROM document_chunks
WHERE tsv @@ plainto_tsquery('chinese', $1)
${filters?.tags?.length ? `AND tags && $2::text[]` : ''}
ORDER BY text_score DESC
LIMIT $3`,
query,
filters?.tags || [],
limit,
);
return results.map(r => ({
id: r.id,
content: r.content,
title: r.title,
score: r.text_score,
vectorScore: 0,
textScore: r.text_score,
}));
}
/**
* RRF (Reciprocal Rank Fusion)：简单且有效的融合算法
*
* RRF_score(d) = Σ (1 / (k + rank_i(d)))
*
* k=60 是经典默认值，来自 Waterloo 大学的论文
*/
private reciprocalRankFusion(
vectorResults: SearchResult[],
textResults: SearchResult[],
topK: number,
k: number = 60,
): SearchResult[] {
const scoreMap = new Map();
// 向量搜索排名 → RRF 分数
vectorResults.forEach((result, rank) => {
const rrfScore = 1 / (k + rank + 1);
scoreMap.set(result.id, {
...result,
score: rrfScore,
vectorScore: result.vectorScore,
textScore: 0,
});
});
// 全文搜索排名 → 叠加 RRF 分数
textResults.forEach((result, rank) => {
const rrfScore = 1 / (k + rank + 1);
const existing = scoreMap.get(result.id);
if (existing) {
existing.score += rrfScore;
existing.textScore = result.textScore;
} else {
scoreMap.set(result.id, {
...result,
score: rrfScore,
textScore: result.textScore,
});
}
});
// 按融合分数排序，取 topK
return Array.from(scoreMap.values())
.sort((a, b) => b.score - a.score)
.slice(0, topK);
}
}
5.3 重排序（Rerank） 混合搜索拿到候选集后，还可以用 Cross-encoder 做精排：
# reranker.py
from sentence_transformers import CrossEncoder
class Reranker:
"""用 Cross-encoder 对候选集做精排"""
def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
self.model = CrossEncoder(model_name)
def rerank(
self,
query: str,
candidates: List[str],
top_k: int = 5,
) -> List[Tuple[int, float]]:
"""
对候选文档重排序
返回: [(原始索引, 相关性得分), ...]
"""
# 构建 (query, doc) 对
pairs = [[query, doc] for doc in candidates]
# Cross-encoder 一次性给每对打分
scores = self.model.predict(pairs)
# 按分数降序排列，返回 top_k
ranked = sorted(
enumerate(scores),
key=lambda x: x[1],
reverse=True,
)[:top_k]
return ranked
Bi-encoder vs Cross-encoder：
Bi-encoder Cross-encoder 原理 独立编码 query 和 doc 联合编码 (query, doc) 对 速度 快（向量可预计算） 慢（每次要重新算） 精度 中等 高（全注意力交互） 用途 第一阶段召回（海量→几百） 第二阶段精排（几百→几十） 典型模型 BGE-M3, OpenAI embeddings bge-reranker-v2-m3, Cohere Rerank
6. 从零实现 RAG 知识库
6.1 系统架构 frontend (Next.js)
│
▼
backend (NestJS)
├── POST /rag/index → 文档入库（分块→嵌入→存储）
├── POST /rag/search → 混合搜索
├── POST /rag/chat → RAG 问答
└── DELETE /rag/documents/:id → 删除文档
│
▼
PostgreSQL + pgvector
├── document_chunks（向量 + 全文搜索）
└── documents（文档元数据）
6.2 Prisma Schema 扩展 // schema.prisma 新增部分
model Document {
id String @id @default(uuid())
title String
source String? // URL 或文件名
tags String[]
status String @default("pending") // pending, processing, ready, error
chunkCount Int @default(0)
chunks DocumentChunk[]
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
}
model DocumentChunk {
id String @id @default(uuid())
documentId String
document Document @relation(fields: [documentId], references: [id], onDelete: Cascade)
content String
chunkIndex Int
embedding Unsupported("vector(1536)")? // Prisma 不原生支持 vector，用 Unsupported
tsv Unsupported("tsvector")?
createdAt DateTime @default(now())
@@index([documentId])
}
6.3 文档入库服务 // rag-index.service.ts
import { Injectable, Logger } from '@nestjs/common';
import { PrismaService } from '../prisma/prisma.service';
import { EmbeddingService } from './embedding.service';
import { ChunkerService } from './chunker.service';
@Injectable()
export class RagIndexService {
private readonly logger = new Logger(RagIndexService.name);
constructor(
private prisma: PrismaService,
private embedding: EmbeddingService,
private chunker: ChunkerService,
) {}
async indexDocument(
title: string,
content: string,
source?: string,
tags: string[] = [],
): Promise<{ documentId: string; chunkCount: number }> {
// 1. 创建文档记录
const document = await this.prisma.document.create({
data: {
title,
source: source || null,
tags,
status: 'processing',
},
});
try {
// 2. 分块
const chunks = this.chunker.splitText(content, title);
this.logger.log(`Document ${document.id}: split into ${chunks.length} chunks`);
// 3. 批量嵌入
const chunkTexts = chunks.map(c => c.content);
const embeddings = await this.embedding.embedTexts(chunkTexts);
// 4. 批量写入数据库
await this.prisma.$transaction(async (tx) => {
for (let i = 0; i < chunks.length; i++) {
const vectorStr = `[${embeddings[i].join(',')}]`;
await tx.$executeRawUnsafe(
`INSERT INTO document_chunks (id, "documentId", content, "chunkIndex", embedding, tsv)
VALUES (gen_random_uuid(), $1, $2, $3, $4::vector, to_tsvector('chinese', $2))`,
document.id,
chunks[i].content,
i,
vectorStr,
);
}
// 更新文档状态
await tx.document.update({
where: { id: document.id },
data: { status: 'ready', chunkCount: chunks.length },
});
});
this.logger.log(`Document ${document.id}: indexed successfully`);
return { documentId: document.id, chunkCount: chunks.length };
} catch (error) {
await this.prisma.document.update({
where: { id: document.id },
data: { status: 'error' },
});
throw error;
}
}
}
6.4 RAG 问答服务 // rag-chat.service.ts
import { Injectable } from '@nestjs/common';
import { HybridSearchService } from './hybrid-search.service';
import { OpenAIService } from '../openai/openai.service';
@Injectable()
export class RagChatService {
constructor(
private hybridSearch: HybridSearchService,
private openai: OpenAIService,
) {}
async chat(query: string): Promise<{
answer: string;
sources: Array<{ title: string; content: string; score: number }>;
}> {
// Step 1: 混合搜索检索相关文档块
const searchResults = await this.hybridSearch.search(query, 5);
if (searchResults.length === 0) {
return {
answer: '抱歉，知识库中没有找到相关信息。请尝试换个问法或补充更多细节。',
sources: [],
};
}
// Step 2: 构建 Prompt
const context = searchResults
.map((r, i) => `[来源${i + 1}] ${r.title}\n${r.content}`)
.join('\n\n');
const systemPrompt = `你是一个专业的知识库助手。请基于以下参考资料回答用户的问题。
## 规则
- 只使用参考资料中的信息，不要编造
- 如果参考资料不足以回答问题，明确告知用户
- 回答时标注引用的来源编号，如 [来源1]
- 使用中文回答
## 参考资料
${context}`;
// Step 3: 调用 LLM 生成答案
const answer = await this.openai.chat({
model: 'claude-sonnet-4-6',
system: systemPrompt,
messages: [{ role: 'user', content: query }],
max_tokens: 2000,
});
return {
answer,
sources: searchResults.map(r => ({
title: r.title,
content: r.content.slice(0, 200) + '...', // 截断显示
score: Math.round(r.score * 1000) / 1000,
})),
};
}
}
7. 检索质量优化
7.1 Query 改写 用户输入的查询往往不够理想。Query 改写可以显著提升检索质量：
// query-rewriter.service.ts
import { Injectable } from '@nestjs/common';
import { OpenAIService } from '../openai/openai.service';
@Injectable()
export class QueryRewriterService {
constructor(private openai: OpenAIService) {}
async rewrite(query: string): Promise<{
original: string;
rewritten: string;
keywords: string[];
}> {
const response = await this.openai.chat({
model: 'claude-haiku-4-5', // 简单任务用小模型即可
system: `你是一个搜索查询优化助手。将用户的查询改写为更适合向量检索和关键词检索的形式。
## 改写规则
1. 补充隐含的上下文和技术术语
2. 提取核心关键词
3. 修正拼写错误和口语化表达
4. 保持原意不变
返回 JSON 格式：{"rewritten": "改写后的查询", "keywords": ["关键词1", "关键词2"]}`,
messages: [{ role: 'user', content: query }],
max_tokens: 300,
});
try {
const parsed = JSON.parse(response);
return {
original: query,
rewritten: parsed.rewritten || query,
keywords: parsed.keywords || [],
};
} catch {
return { original: query, rewritten: query, keywords: [] };
}
}
}
7.2 评估检索质量 # evaluate_rag.py
"""
RAG 检索质量评估脚本
指标：Recall@K, MRR (Mean Reciprocal Rank), NDCG@K
"""
import json
import numpy as np
from typing import List, Dict
def recall_at_k(
retrieved_ids: List[str],
relevant_ids: List[str],
k: int = 5,
) -> float:
"""前 K 个结果中召回了多少相关文档"""
retrieved_k = set(retrieved_ids[:k])
relevant = set(relevant_ids)
if not relevant:
return 0.0
return len(retrieved_k & relevant) / len(relevant)
def mrr(
all_retrieved: List[List[str]],
all_relevant: List[List[str]],
) -> float:
"""Mean Reciprocal Rank：第一个相关文档的平均排名倒数"""
reciprocal_ranks = []
for retrieved, relevant in zip(all_retrieved, all_relevant):
relevant_set = set(relevant)
for rank, doc_id in enumerate(retrieved, start=1):
if doc_id in relevant_set:
reciprocal_ranks.append(1.0 / rank)
break
else:
reciprocal_ranks.append(0.0)
return np.mean(reciprocal_ranks)
def ndcg_at_k(
retrieved_ids: List[str],
relevance_scores: Dict[str, int], # {doc_id: relevance_score}
k: int = 5,
) -> float:
"""Normalized Discounted Cumulative Gain"""
def dcg(scores):
discounts = 1.0 / np.log2(np.arange(2, len(scores) + 2))
return np.sum(scores * discounts)
retrieved_k = retrieved_ids[:k]
rels = np.array([relevance_scores.get(doc_id, 0) for doc_id in retrieved_k])
# Ideal DCG (完美排序)
ideal_rels = np.sort(list(relevance_scores.values()))[::-1][:k]
actual_dcg = dcg(rels)
ideal_dcg = dcg(ideal_rels)
if ideal_dcg == 0:
return 0.0
return actual_dcg / ideal_dcg
# 使用示例
if __name__ == "__main__":
# 真实测试数据（至少 50 条查询）
test_queries = [
{
"query": "Claude Code MCP 配置",
"relevant_chunk_ids": ["chunk_1", "chunk_5", "chunk_12"],
},
# ... 更多测试查询
]
for query_data in test_queries:
# 调用搜索接口
results = search(query_data["query"], top_k=10)
retrieved_ids = [r["id"] for r in results]
recall = recall_at_k(retrieved_ids, query_data["relevant_chunk_ids"], k=5)
print(f"Query: {query_data['query']}, Recall@5: {recall:.3f}")
7.3 优化清单 优化方向 具体措施 预期提升 分块策略 从固定长度 → 递归分块 Recall +10-15% 嵌入模型 从通用模型 → 领域专用（如 BGE-M3） Recall +5-10% Query 改写 查询扩展 + 关键词提取 Recall +5-15% 混合搜索 纯向量 → 向量+BM25 Recall +10-20% 重排序 增加 Cross-encoder 精排 NDCG +5-15% 元数据过滤 按标签/日期/来源预过滤 精度 +10-20% HyDE 先生成假设文档再检索 Recall +5-10%
8. 生产环境考量
8.1 性能基准 在一台 4C16G 服务器上的实测数据（pgvector + HNSW，10 万条文档块，1536 维）：
操作 延迟 备注 单条嵌入（OpenAI API） ~100ms 网络延迟为主 批量嵌入（100 条） ~500ms API 并行处理 向量搜索（Top 10） 1-5ms HNSW ef_search=100 全文搜索 2-10ms GIN 索引 混合搜索（含融合） 10-30ms 向量+全文并行 端到端 RAG 问答 1-2s 含 LLM 生成时间
8.2 缓存策略 // rag-cache.service.ts
import { Injectable } from '@nestjs/common';
import { Redis } from 'ioredis';
@Injectable()
export class RagCacheService {
constructor(private redis: Redis) {}
async getCachedSearch(query: string): Promise {
const key = `rag:search:${this.hashQuery(query)}`;
const cached = await this.redis.get(key);
return cached ? JSON.parse(cached) : null;
}
async setCachedSearch(query: string, results: any, ttlSeconds = 3600): Promise {
const key = `rag:search:${this.hashQuery(query)}`;
await this.redis.setex(key, ttlSeconds, JSON.stringify(results));
}
private hashQuery(query: string): string {
// 简单的查询归一化：去重空格、小写化
return query.trim().toLowerCase().replace(/\s+/g, ' ');
}
}
8.3 监控指标 // rag-metrics.service.ts
import { Injectable } from '@nestjs/common';
import { Counter, Histogram } from 'prom-client';
@Injectable()
export class RagMetricsService {
// 搜索延迟直方图
searchLatency = new Histogram({
name: 'rag_search_latency_seconds',
help: 'RAG search latency in seconds',
buckets: [0.01, 0.05, 0.1, 0.5, 1, 2, 5],
});
// 搜索请求计数
searchTotal = new Counter({
name: 'rag_search_requests_total',
help: 'Total number of RAG search requests',
labelNames: ['status'],
});
// 文档数量计量
documentCount = new Counter({
name: 'rag_documents_indexed_total',
help: 'Total number of documents indexed',
});
// 向量维度
vectorDimension = 1536;
}
8.4 数据库维护 -- 定期更新表统计信息（影响查询计划）
ANALYZE document_chunks;
-- 查看索引使用情况
SELECT
indexrelname,
idx_scan,
idx_tup_read,
idx_tup_fetch
FROM pg_stat_user_indexes
WHERE relname = 'document_chunks';
-- 查看向量索引大小
SELECT
pg_size_pretty(pg_relation_size('document_chunks_embedding_idx')) AS index_size;
-- 重建索引（大量写入后）
REINDEX INDEX CONCURRENTLY document_chunks_embedding_idx;
9. 常见坑点排查
9.1 「明明有相关文档，但搜不出来」
分块太大 → 把不相关内容也包进来了，降低了相似度。调小 chunk_size。嵌入模型与查询不匹配 → 用了英文优化模型处理中文。换成 BGE-M3 或 mGTE。距离度量用错 → 内积要求向量归一化，没归一化就用余弦距离。ef_search 太小 → HNSW 搜索不充分。SET hnsw.ef_search = 200;
9.2 「搜索结果和查询语言不匹配」 解决方案 ：在文档入库时标记语言，检索时添加语言过滤：
-- 文档表加语言字段
ALTER TABLE document_chunks ADD COLUMN language VARCHAR(10) DEFAULT 'zh';
-- 搜索时过滤
SELECT * FROM document_chunks
WHERE language = 'zh'
ORDER BY embedding <=> query_vector
LIMIT 10;
9.3 「pgvector 查询越来越慢」 -- 1. 检查索引膨胀
SELECT pg_size_pretty(pg_total_relation_size('document_chunks_embedding_idx'));
-- 2. 重建索引（CONCURRENTLY 不锁表）
REINDEX INDEX CONCURRENTLY document_chunks_embedding_idx;
-- 3. 如果数据量超过百万，考虑分区表
CREATE TABLE document_chunks_partitioned (
LIKE document_chunks INCLUDING ALL
) PARTITION BY RANGE (created_at);
9.4 「嵌入 API 调用太贵」 OpenAI embedding API 按 token 计费，大规模文档入库时成本不低。
// 1. 去重：相同内容的块只嵌入一次
const contentHash = crypto.createHash('md5').update(chunk.content).digest('hex');
const existing = await prisma.documentChunk.findFirst({
where: { contentHash },
});
if (existing) {
// 复用已有的 embedding，只更新元数据关联
await prisma.documentChunk.create({
data: { ...chunkData, embedding: existing.embedding },
});
return;
}
// 2. 使用 Matryoshka 降维（OpenAI 模型支持）
// text-embedding-3-large 输出 3072 维，可以截取前 256 维，效果下降很少
const response = await openai.embeddings.create({
model: 'text-embedding-3-large',
input: texts,
dimensions: 256, // 只取前 256 维 → 存储成本降低 92%，速度提升 8x
});
// 3. 对于不太重要的文档，用本地小模型
// 本地 BGE-small-zh 只需 CPU，零 API 成本
9.5 「数据库连接池耗尽」 // Prisma 连接池配置
// schema.prisma
datasource db {
provider = "postgresql"
url = env("DATABASE_URL")
// 增加连接池超时设置
}
// .env
// DATABASE_URL=postgresql://user:pass@localhost:5432/db?connection_limit=20&pool_timeout=10
// ↑ 默认 10，向量搜索建议 20-30
总结 RAG 不是简单的「向量搜索 + LLM」，而是一个涉及分块策略、嵌入模型、混合搜索、重排序、缓存等多环节的系统工程。本文的核心要点：
pgvector 是 90% 项目的最佳向量数据库 ——零额外运维，与 PostgreSQL 一体BGE-M3 是中文 RAG 的最佳嵌入模型 ——免费、多语言、支持稀疏+稠密递归分块 + 父子文档 是最稳健的分块策略**混合搜索（向量 + BM25 + RRF）**是检索质量的性价比之王
Query 改写 + Cross-encoder 重排序 是高精度场景的标配先跑通再优化 ——从 512 tokens 分块、HNSW 索引、余弦距离开始，跑 50 条真实查询后迭代 > 本文覆盖的代码可直接用于生产环境。如果你使用 NestJS + Prisma + pgvector 构建知识库，欢迎在志趣社区 的项目展示板块分享你的项目。