production-rag-checklist

生产环境 RAG 检查清单

部署可靠、可扩展且易于维护的 RAG 系统所需的一切。

生产前检查清单

数据与索引

• 定义文档处理流程

  • 确定数据接入来源
  • 针对每种文档类型制定解析策略
  • 测试并评估分块策略
  • 配置元数据提取

• 索引刷新策略

  • 实时（流式）
  • 定时（每小时／每天）
  • 按需（手动触发）
  • 检测过期文档
  • 增量更新（而非完整重建）

• 数据质量

  • 检测并移除重复项
  • 处理空文档／损坏文档
  • 检测并处理 PII
  • 验证来源

检索

• 选择检索策略

  • 仅向量检索
  • 混合检索（向量 + 关键词）
  • GraphRAG
  • 配置重排序

• 嵌入

  • 选择并测试模型
  • 确保嵌入维度合适
  • 对批量操作使用批量嵌入
  • 制定嵌入版本管理策略

• 向量存储

  • 选择适用于生产环境的存储（Pinecone、Weaviate、Qdrant 等）
  • 制定扩展策略（分片、副本）
  • 测试备份与恢复

生成

• LLM 配置

  • 选择模型（权衡成本与质量）
  • 调整温度和参数
  • 合理设置上下文窗口大小
  • 配置备用模型

• 提示词工程

  • 测试系统提示词
  • 添加少样本示例
  • 指定输出格式
  • 设置防护措施

架构模式

基本生产架构

┌─────────────────────────────────────────────────────────────┐
│                        Load Balancer                         │
└─────────────────────────────────────────────────────────────┘
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│   API Server    │ │   API Server    │ │   API Server    │
│   (Stateless)   │ │   (Stateless)   │ │   (Stateless)   │
└─────────────────┘ └─────────────────┘ └─────────────────┘
              │               │               │
              └───────────────┼───────────────┘
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│   Redis Cache   │ │  Vector Store   │ │   LLM Gateway   │
│  (Query Cache)  │ │  (Retrieval)    │ │  (Rate Limit)   │
└─────────────────┘ └─────────────────┘ └─────────────────┘

与 LLM 无关的设计

from abc import ABC, abstractmethod

class LLMProvider(ABC):
    @abstractmethod
    async def generate(self, prompt: str, **kwargs) -> str:
        pass

class OpenAIProvider(LLMProvider):
    async def generate(self, prompt: str, **kwargs) -> str:
        # OpenAI implementation
        pass

class AnthropicProvider(LLMProvider):
    async def generate(self, prompt: str, **kwargs) -> str:
        # Anthropic implementation
        pass

class RAGService:
    def __init__(self, llm_provider: LLMProvider, retriever):
        self.llm = llm_provider
        self.retriever = retriever

    async def query(self, question: str) -> str:
        docs = await self.retriever.ainvoke(question)
        context = self._format_context(docs)
        return await self.llm.generate(
            f"Context: {context}\n\nQuestion: {question}"
        )

缓存策略

查询缓存

import hashlib
import redis
import json

class RAGCache:
    def __init__(self, redis_client: redis.Redis, ttl: int = 3600):
        self.redis = redis_client
        self.ttl = ttl

    def _cache_key(self, query: str) -> str:
        """Generate cache key from query."""
        return f"rag:query:{hashlib.md5(query.encode()).hexdigest()}"

    async def get_or_compute(
        self,
        query: str,
        compute_fn: callable
    ) -> dict:
        """Get from cache or compute and store."""
        key = self._cache_key(query)

        # Try cache
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)

        # Compute
        result = await compute_fn(query)

        # Cache
        self.redis.setex(key, self.ttl, json.dumps(result))
        return result

嵌入缓存

class EmbeddingCache:
    """Cache embeddings to avoid recomputation."""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def get_embedding(self, text: str) -> list[float] | None:
        key = f"emb:{hashlib.md5(text.encode()).hexdigest()}"
        cached = self.redis.get(key)
        return json.loads(cached) if cached else None

    def set_embedding(self, text: str, embedding: list[float]):
        key = f"emb:{hashlib.md5(text.encode()).hexdigest()}"
        self.redis.setex(key, 86400 * 7, json.dumps(embedding))  # 7 days

错误处理与备用方案

class ResilientRAG:
    """RAG with comprehensive error handling."""

    def __init__(self, primary_llm, fallback_llm, retriever):
        self.primary_llm = primary_llm
        self.fallback_llm = fallback_llm
        self.retriever = retriever

    async def query(self, question: str) -> dict:
        try:
            # Retrieval with timeout
            docs = await asyncio.wait_for(
                self.retriever.ainvoke(question),
                timeout=5.0
            )
        except asyncio.TimeoutError:
            logger.warning("Retrieval timeout, using empty context")
            docs = []
        except Exception as e:
            logger.error(f"Retrieval error: {e}")
            docs = []

        if not docs:
            return {
                "answer": "I couldn't find relevant information to answer your question.",
                "sources": [],
                "status": "no_results"
            }

        context = self._format_context(docs)

        # Try primary LLM
        try:
            answer = await asyncio.wait_for(
                self.primary_llm.generate(question, context),
                timeout=30.0
            )
            return {"answer": answer, "sources": docs, "status": "success"}

        except Exception as e:
            logger.warning(f"Primary LLM failed: {e}, trying fallback")

            # Try fallback LLM
            try:
                answer = await self.fallback_llm.generate(question, context)
                return {"answer": answer, "sources": docs, "status": "fallback"}
            except Exception as e:
                logger.error(f"Fallback LLM failed: {e}")
                return {
                    "answer": "I'm having trouble generating a response. Please try again.",
                    "sources": docs,
                    "status": "error"
                }

可观测性与监控

日志结构

import structlog

logger = structlog.get_logger()

async def monitored_query(rag, query: str, user_id: str) -> dict:
    """Query with comprehensive logging."""
    request_id = str(uuid.uuid4())

    logger.info(
        "rag_query_started",
        request_id=request_id,
        user_id=user_id,
        query_length=len(query)
    )

    start = time.perf_counter()

    try:
        # Retrieval
        retrieval_start = time.perf_counter()
        docs = await rag.retriever.ainvoke(query)
        retrieval_ms = (time.perf_counter() - retrieval_start) * 1000

        logger.info(
            "rag_retrieval_complete",
            request_id=request_id,
            num_docs=len(docs),
            latency_ms=retrieval_ms
        )

        # Generation
        gen_start = time.perf_counter()
        answer = await rag.generate(query, docs)
        gen_ms = (time.perf_counter() - gen_start) * 1000

        total_ms = (time.perf_counter() - start) * 1000

        logger.info(
            "rag_query_complete",
            request_id=request_id,
            retrieval_ms=retrieval_ms,
            generation_ms=gen_ms,
            total_ms=total_ms,
            answer_length=len(answer)
        )

        return {"answer": answer, "request_id": request_id}

    except Exception as e:
        logger.error(
            "rag_query_failed",
            request_id=request_id,
            error=str(e),
            error_type=type(e).__name__
        )
        raise

需要跟踪的关键指标

# Prometheus metrics example
from prometheus_client import Counter, Histogram, Gauge

# Counters
rag_queries_total = Counter(
    'rag_queries_total',
    'Total RAG queries',
    ['status']  # success, no_results, error
)

# Histograms
rag_latency = Histogram(
    'rag_latency_seconds',
    'RAG query latency',
    ['stage'],  # retrieval, generation, total
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)

# Gauges
rag_index_size = Gauge(
    'rag_index_document_count',
    'Number of documents in RAG index'
)

仪表板必备指标

指标	告警阈值
P95 延迟	>5 秒
错误率	>5%
缓存命中率	<50%
空结果率	>20%
LLM 备用模型使用率	>10%

安全检查清单

• 输入验证

  • 查询长度限制
  • 检测提示词注入
  • 按用户进行速率限制

• 数据保护

  • 过滤响应中的 PII
  • 控制对文档的访问
  • 审计日志记录

• API 安全

  • 必须进行身份验证
  • 仅使用 HTTPS
  • 轮换 API 密钥

成本优化

class CostAwareRAG:
    """RAG with cost controls."""

    def __init__(self, rag, monthly_budget: float):
        self.rag = rag
        self.budget = monthly_budget
        self.month_spend = 0

    async def query(self, question: str) -> dict:
        # Check budget
        if self.month_spend >= self.budget:
            return {"error": "Monthly budget exceeded"}

        # Estimate cost before query
        estimated_cost = self._estimate_cost(question)

        if self.month_spend + estimated_cost > self.budget:
            # Use cheaper model
            return await self.rag.query_cheap(question)

        result = await self.rag.query(question)
        self.month_spend += self._actual_cost(result)
        return result

    def _estimate_cost(self, question: str) -> float:
        # Estimate based on typical token usage
        input_tokens = len(question.split()) * 1.3
        output_tokens = 500  # Assume average
        return (input_tokens * 0.00001) + (output_tokens * 0.00003)

部署检查清单

部署前

• 完成负载测试
• 评估指标达到阈值
• 记录回滚计划
• 建立值班轮换机制

部署

• 金丝雀部署（10% 流量）
• 监控错误率
• 监控延迟
• 逐步 rollout

部署后

• 确认指标稳定
• 检查缓存命中率
• 监控用户反馈
• 安排首次复盘（1 周后）