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 seconds
エラー率	>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%）
• エラー率をモニタリング
• レイテンシをモニタリング
• 段階的にロールアウト

デプロイ後

• メトリクスが安定していることを確認
• キャッシュヒット率を確認
• ユーザーからのフィードバックをモニタリング
• 初回レビューを予定（1 週間後）