构建生产级 RAG 系统：从 Embedding 到混合检索与重排序的完整实战

一、为什么 RAG 是 2026 年 AI 应用的基础设施

在 ChatGPT 等大语言模型席卷世界两年后，行业已经从"模型能做什么"转向"模型如何与企业真实数据协作"。RAG（Retrieval-Augmented Generation，检索增强生成）成了连接 LLM 与企业私有知识的事实标准架构：据业界统计，2026 年超过 75% 的企业 AI 应用采用了 RAG 或其变体作为核心数据访问层。

RAG 的管线看似简单——用户提问 → 检索相关文档 → 拼接上下文 → LLM 生成回答——但每个环节的生产级实现都充满陷阱。本文将基于 Node.js + pgvector + OpenAI API 技术栈，手把手构建一个可部署上线的 RAG 服务。

二、RAG 管线的六个核心环节

在一个完整的生产级 RAG 系统中，数据流经以下环节：

1. 文档摄取（Ingestion）：将 PDF、Markdown、HTML 等异构文档切分为合适大小的 Chunk
2. 向量化（Embedding）：通过 Embedding 模型将文本转换为高维向量
3. 索引存储（Indexing）：将向量持久化存储，支持高效 ANN 检索
4. 查询处理（Query Processing）：用户 Query 改写、扩句、HyDE 策略
5. 混合检索（Hybrid Retrieval）：向量检索 + 关键词检索融合（RRF / Weighted融合）
6. 重排序（Reranking）：使用 Cross-Encoder 模型对候选结果精排

三、技术选型：为什么选择 pgvector

向量数据库赛道在 2025-2026 年经历了大洗牌：Pinecone 开始收费 Tier、Weaviate 偏向托管服务、Milvus 运维复杂度高。而 pgvector——PostgreSQL 的向量扩展——凭借以下优势成为越来越多技术团队的首选：

• 零额外运维：复用已有的 PostgreSQL 基础设施
• 事务一致性：向量数据与普通表在同一事务中，不会出现"向量已插入但元数据丢失"的问题
• 混合查询：可以同时对向量相似度和结构化字段（时间、标签、权限）做过滤
• 生态成熟：Supabase、Neon、Timescale 均已原生支持
• 距离算法：支持 L2、内积、余弦距离，以及 HNSW 和 IVFFlat 索引

四、实战第一步：数据库初始化与表结构设计

使用 PostgreSQL 16+ 并安装 pgvector 扩展。以下是完整的表结构设计：

-- 启用 pgvector 扩展
CREATE EXTENSION IF NOT EXISTS vector;

-- 文档表
CREATE TABLE documents (
    id BIGSERIAL PRIMARY KEY,
    source TEXT NOT NULL,          -- 来源标识 (文件URL/help center slug)
    title TEXT NOT NULL,
    content_md TEXT NOT NULL,      -- 原始 Markdown 内容
    metadata JSONB DEFAULT '{}',   -- 灵活元数据 (作者、版本、权限等)
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- 文档 Chunk 表（向量化后的检索单元）
CREATE TABLE document_chunks (
    id BIGSERIAL PRIMARY KEY,
    document_id BIGINT REFERENCES documents(id) ON DELETE CASCADE,
    chunk_index INT NOT NULL,      -- 文档内序号
    content TEXT NOT NULL,         -- Chunk 原始文本
    token_count INT NOT NULL,      -- Token 数量（用于上下文窗口管理）
    embedding vector(1536) NOT NULL, -- OpenAI text-embedding-3-small 维度
    metadata JSONB DEFAULT '{}',   -- 覆盖性的文档级元数据（冗余存储便于检索过滤）
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- HNSW 索引（生产推荐：精度高，查询快，但构建较慢）
CREATE INDEX ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);

-- 文档级元数据索引
CREATE INDEX idx_chunks_metadata ON document_chunks USING GIN (metadata);
CREATE INDEX idx_chunks_document_id ON document_chunks(document_id);

五、实战第二步：文档切分策略

Chunk 大小直接决定检索质量。太小则丢失上下文，太大则引入噪声。实战中的最佳策略是语义切分 + 重叠窗口：

// chunker.ts — 递归语义切分器
interface ChunkOptions {
  maxTokens: number;       // 默认 512
  overlapTokens: number;   // 默认 50
  minChunkTokens: number;  // 默认 100
}

class SemanticChunker {
  private encoder: TikTokenEncoder;

  constructor(private opts: ChunkOptions = {
    maxTokens: 512, overlapTokens: 50, minChunkTokens: 100
  }) {
    this.encoder = get_encoding('cl100k_base');
  }

  async split(markdown: string, metadata: Record): Promise {
    // 1. 按 Markdown 标题层级做粗切分（保持章节完整）
    const sections = this.splitByHeaders(markdown);

    const chunks: ChunkInput[] = [];
    for (const section of sections) {
      const tokens = this.encoder.encode(section.content);

      if (tokens.length <= this.opts.maxTokens) {
        chunks.push({
          content: section.content,
          token_count: tokens.length,
          metadata: { ...metadata, heading: section.heading }
        });
      } else {
        // 超长段落递归切分（按句号→分号→逗号三级降级）
        const subChunks = this.recursiveSplit(section, ['\n\n', '. ', '; ', ', ']);
        chunks.push(...subChunks);
      }
    }

    // 2. 添加重叠窗口（保持跨块上下文连续性）
    return this.addOverlap(chunks);
  }

  private addOverlap(chunks: ChunkInput[]): ChunkInput[] {
    return chunks.map((chunk, i) => {
      if (i === 0) return chunk;
      const prevContent = chunks[i - 1].content;
      const overlap = prevContent.slice(-this.opts.overlapTokens * 4);
      return {
        ...chunk,
        content: overlap + ' ... ' + chunk.content,
        token_count: chunk.token_count + this.opts.overlapTokens
      };
    });
  }
}

关键经验：

• Markdown 文档按 H2/H3 切分效果最好，比纯文本滑动窗口的检索准确率高 15-20%
• 重叠窗口设为 Chunk 大小的 10-15%，既保持连续性又不浪费 Token
• 保留 heading 路径到 metadata，便于检索后展示文档位置

六、实战第三步：向量化与批量写入

OpenAI text-embedding-3-small 输出 1536 维向量，每次请求最大支持 8191 tokens，批量接口单次最多 2048 条。以下是批量嵌入管线：

// embedder.ts — 批量嵌入管线
import OpenAI from 'openai';

class EmbeddingPipeline {
  private openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });
  private BATCH_SIZE = 200;
  private RATE_LIMIT_RPS = 5;

  async embedChunks(chunks: ChunkInput[]): Promise {
    const results: EmbeddingResult[] = [];

    for (let i = 0; i < chunks xss=removed xss=removed xss=removed xss=removed> c.content.replace(/\n/g, ' '));

      const response = await this.openai.embeddings.create({
        model: 'text-embedding-3-small',
        input: texts,
        dimensions: 1536,
        encoding_format: 'float'
      });

      const embeddings = response.data
        .sort((a, b) => a.index - b.index)
        .map(d => d.embedding);

      results.push(...batch.map((chunk, idx) => ({
        ...chunk,
        embedding: embeddings[idx]
      })));

      // 速率限制
      if (i + this.BATCH_SIZE < chunks> {
    const sql = `
      INSERT INTO document_chunks (document_id, chunk_index, content, token_count, embedding, metadata)
      SELECT * FROM UNNEST(
        $1::bigint[], $2::int[], $3::text[], $4::int[], $5::vector[], $6::jsonb[]
      )
      ON CONFLICT (document_id, chunk_index) DO UPDATE SET
        content = EXCLUDED.content,
        embedding = EXCLUDED.embedding,
        token_count = EXCLUDED.token_count,
        metadata = EXCLUDED.metadata,
        updated_at = NOW()
    `;
    // execute with pg pool
  }
}

成本参考：text-embedding-3-small 定价 $0.02/1M tokens。处理 10 万条 Chunk 约花费 $2。对于预算紧张的场景，也可以使用本地模型（如 bge-small-en-v1.5，仅 384 维但零成本）。

七、实战第四步：混合检索 — 向量 + BM25 融合

纯向量检索存在一个关键盲区：对精确关键词匹配不敏感。当用户搜索特定错误码（如 ERR_CONNECTION_RESET）、版本号或专有名词时，BM25 的关键词检索往往比向量检索更准确。混合检索结合两者优势：

// search.ts — 混合检索引擎
interface SearchOptions {
  query: string;
  topK?: number;
  candidateMultiplier?: number;
  vectorWeight?: number;
  filters?: Record;
}

class HybridSearchEngine {
  async search(opts: SearchOptions): Promise {
    const { query, topK = 6, candidateMultiplier = 4, vectorWeight = 0.7 } = opts;
    const candidateK = topK * candidateMultiplier;

    // 1. 生成 Query 向量
    const queryEmbedding = await this.embedQuery(query);

    // 2. 并行执行两种检索
    const [vectorResults, keywordResults] = await Promise.all([
      this.vectorSearch(queryEmbedding, candidateK, opts.filters),
      this.keywordSearch(query, candidateK, opts.filters)
    ]);

    // 3. RRF 融合（Reciprocal Rank Fusion）
    return this.reciprocalRankFusion(vectorResults, keywordResults, vectorWeight, topK);
  }

  private async vectorSearch(
    embedding: number[], k: number, filters?: Record
  ): Promise {
    const filterSql = filters ? this.buildFilterSQL(filters) : 'TRUE';
    const sql = `
      SELECT id, content, metadata, document_id,
             1 - (embedding <=> $1::vector) AS similarity
      FROM document_chunks
      WHERE ${filterSql}
      ORDER BY embedding <=> $1::vector
      LIMIT $2
    `;
    return this.pool.query(sql, [JSON.stringify(embedding), k]);
  }

  private async keywordSearch(
    query: string, k: number, filters?: Record
  ): Promise {
    const sql = `
      SELECT id, content, metadata, document_id,
             ts_rank(to_tsvector('english', content), plainto_tsquery('english', $1)) AS rank,
             ts_headline('english', content, plainto_tsquery('english', $1),
               'StartSel=, StopSel=, MaxFragments=3, MaxWords=20'
             ) AS highlight
      FROM document_chunks
      WHERE to_tsvector('english', content) @@ plainto_tsquery('english', $1)
      ORDER BY rank DESC
      LIMIT $2
    `;
    return this.pool.query(sql, [query, k]);
  }

  private reciprocalRankFusion(
    vectorRes: ScoredChunk[],
    keywordRes: ScoredChunk[],
    vectorWeight: number,
    topK: number
  ): ScoredChunk[] {
    const k = 60; // RRF 阻尼常数
    const scoreMap = new Map();
    const chunkMap = new Map();

    // 向量结果打分
    vectorRes.forEach((chunk, rank) => {
      scoreMap.set(chunk.id, (scoreMap.get(chunk.id) || 0) + vectorWeight * (1 / (k + rank + 1)));
      chunkMap.set(chunk.id, chunk);
    });

    // 关键词结果打分
    keywordRes.forEach((chunk, rank) => {
      scoreMap.set(chunk.id, (scoreMap.get(chunk.id) || 0) + (1 - vectorWeight) * (1 / (k + rank + 1)));
      chunkMap.set(chunk.id, chunk);
    });

    // 排序返回
    return Array.from(scoreMap.entries())
      .sort((a, b) => b[1] - a[1])
      .slice(0, topK)
      .map(([id]) => chunkMap.get(id)!);
  }
}

八、实战第五步：Reranker 精排

混合检索返回 Top-K 候选后，使用 Cross-Encoder 模型做精排可以显著提升最终送入 LLM 的上下文质量。Cross-Encoder 模型（如 Cohere rerank、Jina reranker v2、BGE-reranker-v2-m3）对每个 query-chunk 对打分，比 Embedding 的双编码器方式精度高出 10-15%。

// reranker.ts — Reranker 精排层
class Reranker {
  async rerank(
    query: string,
    candidates: ScoredChunk[],
    topN: number = 6
  ): Promise {
    if (candidates.length <= topN) return candidates;

    const response = await fetch('http://localhost:8080/rerank', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        query,
        documents: candidates.map(c => c.content),
        model: 'jina-reranker-v2-base-multilingual',
        top_n: topN
      })
    });

    const { results } = await response.json();
    return results.map((r: any) => ({
      ...candidates[r.index],
      rerankScore: r.relevance_score
    }));
  }
}

// 完整的检索管线编排
class RAGPipeline {
  constructor(
    private embedder: EmbeddingPipeline,
    private searcher: HybridSearchEngine,
    private reranker: Reranker,
    private llm: OpenAI
  ) {}

  async answer(question: string, opts?: SearchOptions): Promise {
    // Step 1: 混合检索（20 个候选）
    const candidates = await this.searcher.search({
      ...opts,
      topK: 20,
      candidateMultiplier: 5
    });

    // Step 2: Reranker 精排（Top 6）
    const topChunks = await this.reranker.rerank(question, candidates, 6);

    // Step 3: 构建 Prompt
    const contextStr = topChunks.map((c, i) =>
      `[Source ${i+1}] ${c.metadata.heading || c.metadata.title}\n${c.content}`
    ).join('\n\n');

    // Step 4: LLM 生成
    const response = await this.llm.chat.completions.create({
      model: 'gpt-4o-mini',
      messages: [
        {
          role: 'system',
          content: 'You are a technical documentation assistant. Answer based ONLY on the provided sources. Always cite sources like [Source 1].'
        },
        {
          role: 'user',
          content: `Context:\n${contextStr}\n\nQuestion: ${question}`
        }
      ],
      temperature: 0.1
    });

    return {
      answer: response.choices[0].message.content,
      sources: topChunks.map(c => ({
        id: c.id,
        title: c.metadata.title,
        heading: c.metadata.heading,
        score: c.rerankScore
      })),
      tokenUsage: response.usage
    };
  }
}

九、生产级优化策略

9.1 Query 改写提升召回率

用户 Query 通常简短且模糊。通过 LLM 预处理生成多个改写版本可以显著提升召回率：

• HyDE（Hypothetical Document Embeddings）：让 LLM 先"编造"一个理想答案，再对该答案做 Embedding 检索。经验表明 HyDE 能将召回率提升 8-12%
• Multi-Query：将一个问题改写为 3-5 个语义变体，合并检索结果
• Query 分解：将复杂问题拆解为子问题，分别检索后合并

9.2 上下文压缩

检索的 Chunk 中通常存在大量与问题无关的句子。使用轻量模型（如 GPT-4o-mini）做上下文压缩，可以在送入最终 Prompt 前丢弃无关段落，节省 30-50% 的 Token 消耗。

9.3 评估体系

没有度量就无法改进。RAG 系统的核心评估指标：

• Context Precision / Recall：检索到的 Chunk 中有多大比例是相关的（Precision），所有相关 Chunk 中有多少被找到（Recall）
• Answer Faithfulness：LLM 生成答案是否忠实于检索到的上下文（而非幻觉）
• Answer Relevancy：答案是否直接回应了用户的问题

推荐使用 Ragas 或 Truelens 搭建自动化评估管线，每次更新 Embedding 模型或 Reranker 后跑一次回归测试。

十、总结：RAG 工程的经验法则

两年 RAG 工程实战下来，总结几条最重要的经验：

1. 不要过度优化 Embedding 模型：在大多数场景下，text-embedding-3-small + 混合检索 + Reranker 已经能到 90% 的效果，切换模型带来的边际提升有限
2. Chunk 切分比索引算法更重要：花时间优化切分策略（按章节切 + 合理重叠）比调参 HNSW 的 m 和 ef_construction 收益大得多
3. 混合检索是标配：除非你的场景 100% 是自然语言问题搜索，否则 BM25 兜底是必要的
4. pgvector 足够支撑中等规模：100 万条 Chunk 以下、QPS 100 以内，pgvector + HNSW 完全够用，无需引入专用向量数据库
5. 监控比架构重要：埋点记录每次检索的召回来源和用户反馈（点赞/点劣），形成数据飞轮驱动迭代

RAG 是一门工程学科，核心不在于模型多先进，而在于数据管线、检索策略与评估体系的系统化设计。以上代码均为可直接整合到生产项目中的模块，希望对正在构建 AI 应用的技术团队有所帮助。