从零搭建生产级 RAG 系统：混合检索、重排序与评估的完整指南

别再手动写 RAG 了

90% 的 RAG 教程都在教你写一个玩具 demo——把文档切块、生成向量、塞进向量数据库，然后做个相似度搜索。

看起来能用，但一到生产环境就翻车：检索结果不相关、幻觉满天飞、响应慢得要命。

这篇教程不教你写 demo。我会带你从零搭建一个生产级 RAG 系统，包含混合检索、重排序、自适应分块和评估体系。

RAG 架构全景图

一个生产级 RAG 系统至少需要这 7 个组件：

用户查询
    ↓
查询预处理（Query Preprocessing）
    ↓
混合检索（Hybrid Retrieval）
    ├— 向量检索（Semantic Search）
    ├— 关键词检索（BM25）
    └— 知识图谱检索（Knowledge Graph）
    ↓
重排序（Reranking）
    ↓
上下文组装（Context Assembly）
    ↓
LLM 生成（Generation）
    ↓
输出后处理（Post-processing）

每个组件都有讲究。下面逐个拆解。

自适应分块：不是所有文本都该切成 512 字

大多数 RAG 教程告诉你“切成 512 字的块，overlap 50 字”。这是一个很糟糕的默认值。

法律文书、技术文档、新闻文章的结构完全不同，用同一个 chunk size 切得天编地覆。

真正有效的方法是自适应分块（Adaptive Chunking）：

from langchain.text_splitter import RecursiveCharacterTextSplitter
import tiktoken

class AdaptiveChunker:
    def __init__(self):
        self.encoding = tiktoken.get_encoding("cl100k_base")
    
    def detect_type(self, text: str) -> str:
        """\u68c0\u6d4b\u6587\u672c\u7c7b\u578b"""
        code_ratio = len(re.findall(r'\b(def|class|import|function|const|let)\b', text)) / max(len(text.split()), 1)
        list_ratio = text.count('\n- ') + text.count('\n* ') + text.count('\n1. ')
        
        if code_ratio > 0.02:
            return 'code'
        elif list_ratio > 5:
            return 'structured'
        else:
            return 'prose'
    
    def chunk(self, text: str) -> list[str]:
        text_type = self.detect_type(text)
        
        configs = {
            'code': {'chunk_size': 1500, 'chunk_overlap': 300, 'separators': ['\nclass ', '\ndef ', '\nfunction ', '\n\n']},
            'structured': {'chunk_size': 800, 'chunk_overlap': 100, 'separators': ['\n## ', '\n### ', '\n\n', '\n']},
            'prose': {'chunk_size': 512, 'chunk_overlap': 50, 'separators': ['\n\n', '\n', '。', '.', ' ']}
        }
        
        config = configs[text_type]
        splitter = RecursiveCharacterTextSplitter(**config)
        return splitter.split_text(text)

根据文本类型自动调整分块策略，检索质量能提升 30-50%。

向量检索的隐藏坑

向量检索的核心思路：把文本和查询都转成向量，用余弦相似度找最相关的。

但向量检索有两个致命问题：

1. 关键词精确匹配很差。搜“Python 3.12 新特性”，向量检索可能返回一堆“Python 基础教程”。
2. 专有名词匹配很差。搜“GPT-4o”，它不一定能匹配到“GPT-4 optimised”。

解决方案：混合检索。

混合检索：BM25 + 向量的完美组合

混合检索是生产级 RAG 的标配。原理很简单：用两种检索方式分别查，合并结果。

from rank_bm25 import BM25Okapi
import jieba
import numpy as np
from openai import OpenAI

class HybridRetriever:
    def __init__(self, documents: list[str], embed_model="text-embedding-3-small"):
        self.documents = documents
        self.client = OpenAI()
        self.embed_model = embed_model
        
        # BM25 索引
        tokenized_docs = [list(jieba.cut(doc)) for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)
        
        # 向量索引
        self.embeddings = self._embed(documents)
    
    def _embed(self, texts: list[str]) -> np.ndarray:
        resp = self.client.embeddings.create(input=texts, model=self.embed_model)
        return np.array([d.embedding for d in resp.data])
    
    def search(self, query: str, top_k: int = 10, alpha: float = 0.5) -> list[tuple[str, float]]:
        """\u6df7\u5408\u68c0\u7d22\uff0calpha \u63a7\u5236\u5411\u91cf\u548c BM25 \u7684\u6743\u91cd"""
        # BM25 \u68c0\u7d22
        tokenized_query = list(jieba.cut(query))
        bm25_scores = self.bm25.get_scores(tokenized_query)
        
        # 向量检索bm25_scores = self.bm25.get_scores(tokenized_query)
        
        # 向量检索
        query_embed = self._embed([query])
        cosine_scores = np.dot(self.embeddings, query_embed.T).flatten()
        
        # 归一化
        bm25_norm = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() - bm25_scores.min() + 1e-8)
        cos_norm = (cosine_scores - cosine_scores.min()) / (cosine_scores.max() - cosine_scores.min() + 1e-8)
        
        # 加权融合
        final_scores = alpha * cos_norm + (1 - alpha) * bm25_norm
        top_indices = np.argsort(final_scores)[::-1][:top_k]
        
        return [(self.documents[i], final_scores[i]) for i in top_indices]

alpha 参数是调优的关键。经验值：

• 精确关键词多的查询 → alpha=0.3（偏BM25）
• 语义模糊的查询 → alpha=0.7（偏向量）
• 通用场景 → alpha=0.5

重排序：检索之后的第二道筛子

混合检索拿回 10-20 个候选结果后，还需要用 Reranker 做精排。

为什么？因为检索阶段用的模型是轻量级的（为了速度），精度有限。Reranker 用更重的模型做交叉编码（Cross-Encoder），精度高得多。

from sentence_transformers import CrossEncoder

class Reranker:
    def __init__(self, model_name="BAAI/bge-reranker-v2-m3"):
        self.model = CrossEncoder(model_name)
    
    def rerank(self, query: str, documents: list[str], top_k: int = 5) -> list[tuple[str, float]]:
        pairs = [(query, doc) for doc in documents]
        scores = self.model.predict(pairs)
        ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
        return ranked[:top_k]

加上 Reranker 后，Top-5 结果的相关性通常能从 60% 提升到 85%+。

完整的 RAG Pipeline

把所有组件串起来：

class ProductionRAG:
    def __init__(self, documents: list[str]):
        self.chunker = AdaptiveChunker()
        chunks = self._flatten([self.chunker.chunk(doc) for doc in documents])
        self.retriever = HybridRetriever(chunks)
        self.reranker = Reranker()
        self.client = OpenAI()
    
    def query(self, question: str, top_k: int = 5) -> dict:
        # 1. 混合检索
        candidates, scores = zip(*self.retriever.search(question, top_k=20))
        
        # 2. 重排序
        ranked = self.reranker.rerank(question, list(candidates), top_k=top_k)
        context_docs, context_scores = zip(*ranked)
        
        # 3. 组装上下文
        context = "\n\n---\n\n".join(context_docs)
        
        # 4. LLM 生成
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": f"基于以下上下文回答问题。如果上下文中没有相关信息，说'我没有找到相关信息'。\n\n上下文：{context}"},
                {"role": "user", "content": question}
            ],
            temperature=0.1
        )
        
        return {
            "answer": response.choices[0].message.content,
            "sources": list(context_docs),
            "scores": list(context_scores),
            "confidence": float(np.mean(context_scores))
        }

RAG 评估：你怎么知道系统好不好？

大多数团队上线 RAG 后就不评估了。这是个灾难。

推荐用 RAGAS 框架做系统性评估：

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)
from datasets import Dataset

# 准备评估数据集
eval_data = {
    "question": ["什么是混合检索?", "Reranker 的作用是什么?", ...],
    "answer": [rag.query(q)["answer"] for q in questions],
    "contexts": [[rag.query(q)["sources"]] for q in questions],
    "ground_truth": ["混合检索是...", "Reranker 用于..."]
}

dataset = Dataset.from_dict(eval_data)

results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)

print(results)
# {'faithfulness': 0.85, 'answer_relevancy': 0.78, 'context_precision': 0.82, 'context_recall': 0.75}

四个核心指标：

如果 Faithfulness 低于 0.8，说明 LLM 在产生幻觉，需要加强提示词约束或换模型。如果 Context Recall 低，说明检索层需要优化。

持续优化闭环

生产级 RAG 不是一次性工程，而是持续优化的闭环：

用户反馈（👍/👎 + 评论）
    ↓
收集 Bad Cases
    ↓
分析根因（检索问题？生成问题？分块问题？）
    ↓
针对性优化
    ├─ 调整 alpha 参数
    ├─ 优化分块策略
    ├─ 换 Reranker 模型
    └─ 改进 Prompt
    ↓
RAGAS 评估验证
    ↓
上线 A/B 测试

关键是要有用户反馈机制。每个回答旁边加一个👍/👎按钮，定期分析负面反馈，形成优化闭环。

成本分析

一个日活 1000 次查询的 RAG 系统，月成本大约：

如果用开源模型替代（bge-m3 + Qwen2.5 + bge-reranker），可以把成本压到 $50-100/月，但需要自己维护 GPU 服务器。

总结

搭建生产级 RAG 的关键要素：

1. 自适应分块——不是所有文本都切 512 字
2. 混合检索——BM25 + 向量取长补短
3. 重排序——用 Cross-Encoder 精排 Top-K
4. 系统评估——RAGAS 四指标持续监控
5. 优化闭环——用户反馈驱动持续改进

别再写玩具 RAG 了。按照这个架构来，你的系统才能真正上线。

本文代码示例基于 LangChain 0.3 + RAGAS 0.2 + sentence-transformers 编写。生产环境建议配合 Prometheus + Grafana 做监控。