为什么需要这份检查清单

RAG（检索增强生成）系统从原型到生产环境之间存在巨大鸿沟。开发环境跑得很好的RAG，上线后经常出现：检索不准、生成幻觉、响应慢、成本高、无法扩展等问题。本文总结了25个关键检查项，覆盖RAG系统生产部署的完整生命周期。

检查清单总览

数据层 (5项)  ████████████░░░░  ✓ 5/5
检索层 (6项)  ████████████░░░░  ✓ 6/6  
生成层 (5项)  ████████████░░░░  ✓ 5/5
工程层 (4项)  ████████████░░░░  ✓ 4/4
监控层 (3项)  ████████████░░░░  ✓ 3/3
运维层 (2项)  ████████████░░░░  ✓ 2/2

一、数据层检查项（5项）

1. 文档预处理质量验证

class DocumentPreprocessor:
    """生产级文档预处理流水线"""
    
    def process(self, documents):
        processed = []
        for doc in documents:
            # 检查项1a: 编码一致性
            assert doc.encoding == 'utf-8', f"非UTF-8编码: {doc.path}"
            
            # 检查项1b: 去重
            if self._is_duplicate(doc):
                continue
            
            # 检查项1c: 质量过滤
            if not self._quality_check(doc):
                continue
            
            # 检查项1d: PII脱敏
            doc.content = self._redact_pii(doc.content)
            
            processed.append(doc)
        return processed
    
    def _quality_check(self, doc):
        """文档质量检查"""
        checks = {
            "最小长度": len(doc.content) > 50,
            "非乱码": self._check_encoding_quality(doc.content),
            "语言检测": self._detect_language(doc.content) in ['zh', 'en'],
            "信息密度": len(doc.content) / len(set(doc.content)) < 100,
        }
        return all(checks.values())

✅ 检查项1：所有文档经过编码验证、去重、质量过滤和PII脱敏。

2. 分块策略合理性

def chunk_document(text, strategy="semantic", chunk_size=512, overlap=64):
    """
    生产级文档分块
    关键：不要只按字符数分块，要考虑语义完整性
    """
    if strategy == "semantic":
        # 语义分块：优先在段落、句子边界分割
        paragraphs = text.split('\n\n')
        chunks = []
        current_chunk = ""
        
        for para in paragraphs:
            if len(current_chunk) + len(para) <= chunk_size:
                current_chunk += para + "\n\n"
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                # 保留overlap
                words = current_chunk.split()
                current_chunk = ' '.join(words[-overlap//4:]) + "\n\n" + para
        return chunks
    
    elif strategy == "recursive":
        # 递归分块：按层级分隔符分割
        separators = ['\n\n', '\n', '。', '；', ',', ' ']
        # 递归使用不同分隔符，直到块大小合适
        ...

✅ 检查项2：分块策略考虑语义边界，块大小匹配嵌入模型最优输入长度，设置合理overlap。

3. 嵌入模型选型与评估

✅ 检查项3：嵌入模型在领域数据上的检索准确率经过评估，且选择了最优维度。

嵌入模型 维度 中文支持 推理速度 MTEB得分
bge-large-zh-v1.5 1024 原生 快 64.8
bge-m3 1024 原生 中等 68.1
text-embedding-3-large 3072 支持 快 69.2
GTE-Qwen2-7B 3584 原生 慢 72.3

4. 向量数据库容量规划

def vector_db_capacity_planning(
    num_documents,       # 文档数量
    avg_chunks_per_doc,  # 每文档平均分块数
    embedding_dim,       # 嵌入维度
    metadata_per_chunk,  # 每块元数据大小(bytes)
    growth_rate_monthly, # 月增长率
    planning_horizon_months=12,
):
    """向量数据库容量规划"""
    initial_vectors = num_documents * avg_chunks_per_doc
    projected_vectors = initial_vectors * (1 + growth_rate_monthly) ** planning_horizon_months
    
    # 每向量存储 = 维度 * 4字节(FP32) + 元数据
    bytes_per_vector = embedding_dim * 4 + metadata_per_chunk
    
    initial_storage = initial_vectors * bytes_per_vector
    projected_storage = projected_vectors * bytes_per_vector
    
    return {
        "初始向量数": initial_vectors,
        "12月后向量数": int(projected_vectors),
        "初始存储(GB)": initial_storage / 1e9,
        "12月后存储(GB)": projected_storage / 1e9,
        "推荐内存(GB)": projected_storage / 1e9 * 0.3,  # 索引约占30%
    }

✅ 检查项4：向量数据库容量按12个月增长规划，预留50%余量。

5. 数据更新机制

✅ 检查项5：建立了增量更新流程，支持文档增删改的向量实时同步。

二、检索层检查项（6项）

6. 混合检索策略

class HybridRetriever:
    """混合检索：向量检索 + BM25 + 重排序"""
    
    def __init__(self, vector_store, bm25_index, reranker):
        self.vector_store = vector_store
        self.bm25_index = bm25_index
        self.reranker = reranker
    
    def retrieve(self, query, top_k=10, top_n=3):
        # 1. 向量检索（语义匹配）
        vector_results = self.vector_store.search(
            query, k=top_k
        )
        
        # 2. BM25检索（关键词匹配）
        bm25_results = self.bm25_index.search(
            query, k=top_k
        )
        
        # 3. 融合结果（RRF算法）
        fused = self._reciprocal_rank_fusion(
            vector_results, bm25_results
        )
        
        # 4. 重排序
        reranked = self.reranker.rerank(
            query, fused[:top_k*2], top_n=top_n
        )
        
        return reranked
    
    def _reciprocal_rank_fusion(self, *ranked_lists, k=60):
        """倒数排名融合"""
        scores = {}
        for ranked_list in ranked_lists:
            for rank, doc in enumerate(ranked_list):
                if doc.id not in scores:
                    scores[doc.id] = 0
                scores[doc.id] += 1.0 / (k + rank)
        
        # 按融合分数排序
        sorted_ids = sorted(scores, key=scores.get, reverse=True)
        return [doc for doc in ranked_lists[0] if doc.id in sorted_ids]

✅ 检查项6：采用向量检索 + BM25混合策略，使用RRF算法融合，并接入Cross-Encoder重排序。

7. 查询改写与扩展

class QueryRewriter:
    """查询改写：提升检索召回率"""
    
    def rewrite(self, query, llm, conversation_history=None):
        # 1. 指代消解
        if conversation_history:
            query = self._resolve_coreference(query, conversation_history, llm)
        
        # 2. 查询扩展：生成多个改写版本
        expansions = llm.generate(f"""
        将以下查询改写为3个不同表述版本，保持语义一致：
        原查询：{query}
        输出JSON数组格式。
        """)
        
        # 3. HyDE：生成假设性答案用于检索
        hyde_answer = llm.generate(f"""
        假设你要回答这个问题：{query}
        写一段简短的回答（100字以内），不需要准确。
        """)
        
        return {
            "original": query,
            "expansions": json.loads(expansions),
            "hyde": hyde_answer,
        }

✅ 检查项7：实现了指代消解、查询扩展和HyDE等查询增强技术。

8. 检索结果过滤与权限控制

✅ 检查项8：检索结果按用户权限过滤，支持文档级别的访问控制。

9. 召回率与精确率评估

def evaluate_retrieval(retriever, test_cases):
    """
    检索质量评估
    test_cases: [{"query": "...", "relevant_docs": [...], ...}]
    """
    total_recall = 0
    total_precision = 0
    total_mrr = 0
    
    for case in test_cases:
        results = retriever.retrieve(case["query"], top_k=10)
        retrieved_ids = {r.id for r in results}
        relevant_ids = set(case["relevant_docs"])
        
        # 召回率
        recall = len(retrieved_ids & relevant_ids) / len(relevant_ids)
        total_recall += recall
        
        # 精确率
        precision = len(retrieved_ids & relevant_ids) / len(retrieved_ids)
        total_precision += precision
        
        # MRR
        for rank, r in enumerate(results):
            if r.id in relevant_ids:
                total_mrr += 1.0 / (rank + 1)
                break
    
    n = len(test_cases)
    return {
        "Recall@10": total_recall / n,
        "Precision@10": total_precision / n,
        "MRR": total_mrr / n,
    }

✅ 检查项9：在标注测试集上评估检索质量，Recall@10 > 0.85，MRR > 0.7。

10. 检索延迟优化

✅ 检查项10：P95检索延迟 < 200ms，通过缓存、索引优化和并发检索实现。

11. 多路召回策略

✅ 检查项11：针对不同查询类型（事实型、推理型、浏览型）配置不同的检索策略。

三、生成层检查项（5项）

12. Prompt模板工程化

class PromptManager:
    """生产级Prompt管理"""
    
    TEMPLATES = {
        "rag_answer": """
请基于以下检索到的上下文回答用户问题。

## 上下文信息
{context}

## 用户问题
{question}

## 回答要求
1. 仅基于上下文信息回答，不要编造信息
2. 如果上下文不足以回答问题，明确说明"根据现有信息无法回答"
3. 引用信息时标注来源编号 [1]、[2] 等
4. 保持回答简洁、准确、结构化

## 回答""",
        
        "rag_with_history": """
## 对话历史
{conversation_history}

## 检索到的上下文
{context}

## 当前问题
{question}

请结合对话历史和上下文回答当前问题。""",
    }
    
    def render(self, template_name, **kwargs):
        template = self.TEMPLATES[template_name]
        # 注入上下文（带引用标注）
        if 'context' in kwargs:
            kwargs['context'] = self._format_context(kwargs['context'])
        return template.format(**kwargs)
    
    def _format_context(self, retrieved_docs):
        """格式化检索结果，添加引用编号"""
        formatted = []
        for i, doc in enumerate(retrieved_docs, 1):
            formatted.append(f"[{i}] (来源: {doc.metadata.get('source', '未知')})\n{doc.content}")
        return "\n\n".join(formatted)

✅ 检查项12：Prompt模板版本化管理，支持A/B测试，包含引用标注和拒答机制。

13. 幻觉检测与防护

class HallucinationGuard:
    """幻觉检测与防护"""
    
    def __init__(self, llm, similarity_threshold=0.7):
        self.llm = llm
        self.threshold = similarity_threshold
    
    def check(self, answer, retrieved_context, query):
        checks = {
            "事实一致性": self._check_factual_consistency(answer, retrieved_context),
            "信息来源": self._check_source_attribution(answer, retrieved_context),
            "拒答逻辑": self._check_refusal_logic(answer, retrieved_context, query),
        }
        
        passed = all(checks.values())
        return {"passed": passed, "checks": checks}
    
    def _check_factual_consistency(self, answer, context):
        """检查答案中的事实是否都能在上下文中找到支持"""
        # 将答案分解为事实陈述
        claims = self._extract_claims(answer)
        
        for claim in claims:
            # 计算与上下文的语义相似度
            similarity = self._compute_similarity(claim, context)
            if similarity < self.threshold:
                return False
        return True

✅ 检查项13：部署了幻觉检测机制，对生成内容进行事实一致性校验。

14. 流式响应处理

✅ 检查项14：支持流式响应输出，首Token延迟 < 500ms。

15. 上下文窗口管理

✅ 检查项15：动态管理上下文窗口，在检索结果、对话历史和生成长度之间合理分配Token预算。

16. 生成质量评估

✅ 检查项16：建立生成质量评估pipeline，包含自动指标（faithfulness、relevance）和人工抽检。

四、工程层检查项（4项）

17. API设计规范

# RAG API 规范设计
POST /api/v1/rag/query
  Request:
    {
      "query": "用户问题",
      "conversation_id": "会话ID",
      "filters": {
        "source": ["doc_type:pdf"],
        "date_range": {"start": "2026-01-01"}
      },
      "options": {
        "top_k": 3,
        "stream": true,
        "include_sources": true,
        "max_tokens": 2048
      }
    }
  Response (SSE):
    data: {"type": "retrieval", "sources": [...]}
    data: {"type": "token", "content": "答"}
    data: {"type": "token", "content": "案"}
    data: {"type": "done", "metadata": {"latency_ms": 1200}}

✅ 检查项17：API遵循RESTful规范，支持流式响应、过滤、分页和错误处理。

18. 并发与限流

✅ 检查项18：实现了请求队列、并发控制、限流和降级策略。

19. 成本控制

class CostController:
    """RAG系统成本控制"""
    
    def __init__(self, config):
        self.limits = config
    
    def estimate_cost(self, query, retrieved_docs):
        """预估单次查询成本"""
        # 检索成本
        retrieval_cost = len(retrieved_docs) * 0.0001  # 向量检索
        
        # Prompt Token成本
        prompt_tokens = len(query) // 2 + sum(
            len(doc.content) // 2 for doc in retrieved_docs
        )
        
        # 生成成本
        estimated_output_tokens = min(
            self.limits.max_output_tokens,
            prompt_tokens  # 粗略估计
        )
        
        total_cost = (
            prompt_tokens * self.limits.input_price_per_1k / 1000 +
            estimated_output_tokens * self.limits.output_price_per_1k / 1000 +
            retrieval_cost
        )
        
        return total_cost
    
    def check_budget(self, user_id, estimated_cost):
        """检查用户预算"""
        used = self.get_user_usage(user_id)
        limit = self.get_user_limit(user_id)
        return used + estimated_cost <= limit

✅ 检查项19：实现了Token级成本追踪和用户级预算控制。

20. 安全与合规

✅ 检查项20：输入输出内容安全过滤，日志脱敏，符合数据驻留要求。

五、监控层检查项（3项）

21. 全链路追踪

class RAGTracer:
    """RAG全链路追踪"""
    
    def trace_request(self, request_id, query):
        spans = {
            "query_rewrite": {"start": None, "end": None, "result": None},
            "retrieval": {"start": None, "end": None, "results": []},
            "reranking": {"start": None, "end": None, "results": []},
            "prompt_construction": {"start": None, "end": None},
            "generation": {"start": None, "end": None, "tokens": 0},
            "hallucination_check": {"start": None, "end": None, "passed": None},
        }
        
        # 记录每个阶段的时间和结果
        # 上传到追踪系统（如Jaeger/LangSmith）
        return spans

✅ 检查项21：实现了从查询输入到生成输出的全链路追踪，可视化每个阶段的延迟和质量。

22. 在线评估指标

✅ 检查项22：实时监控以下核心指标：

指标 目标值 告警阈值
检索Recall@10 >0.85 <0.80
端到端延迟P95 <2s >3s
幻觉率 <5% >10%
用户满意度 >4.0/5 <3.5/5
日均成本/用户 <$0.5 >$1.0

23. 数据飞轮闭环

✅ 检查项23：建立了用户反馈收集→标注→评估→优化的数据飞轮。

六、运维层检查项（2项）

24. 部署架构

# 生产级RAG部署架构
production_rag:
  load_balancer:
    type: nginx
    ssl: true
    rate_limit: "100 req/min per user"
  
  api_servers:
    replicas: 3
    resources:
      cpu: 4
      memory: 8Gi
    hpa:
      min: 3
      max: 10
      cpu_target: 70%
  
  retrieval_service:
    replicas: 2
    vector_db: "Milvus cluster (3 nodes)"
    cache: "Redis (query→results, TTL=3600)"
  
  generation_service:
    model: "DeepSeek-V3 API"
    fallback: "Qwen-Max API"
    timeout: 30s
  
  monitoring:
    tracing: "LangSmith"
    metrics: "Prometheus + Grafana"
    alerts: "AlertManager → Slack"

✅ 检查项24：部署架构支持水平扩展、故障转移和灰度发布。

25. 灾备与回滚

✅ 检查项25：制定了数据备份、服务降级和版本回滚预案，RTO < 30分钟。

上线前自检流程

Day -7: 完成所有25项检查
Day -5: 压力测试（2x预期流量）
Day -3: 安全审计与渗透测试
Day -1: 灰度发布（10%流量）
Day 0:  全量发布 + 24小时on-call
Day +7: 首次运行回顾

这份检查清单来自多个RAG生产项目的实战经验。每个检查项背后都有踩过的坑和交过的学费。建议在上线前逐项核对，不要跳过任何一项——生产环境的每一次故障，往往都源于某个"以为不重要"的检查项。