ストーリー

セマンティックキャッシュの仕組み

従来のキャッシュ vs セマンティックキャッシュ

セマンティックキャッシュのアーキテクチャ

ユーザーの質問
     │
     ▼
┌──────────────┐
│ Embedding    │ ← 質問をベクトル化
│ モデル        │
└──────┬───────┘
       │ ベクトル
       ▼
┌──────────────┐     ┌──────────────┐
│ 類似度検索    │────→│ キャッシュDB   │
│ (cosine sim) │     │ (Redis/Qdrant)│
└──────┬───────┘     └──────────────┘
       │
       ├── 類似度 ≥ 閾値 → キャッシュヒット → キャッシュ応答を返却
       │                                      （LLM呼び出しなし）
       │
       └── 類似度 < 閾値 → キャッシュミス → LLM呼び出し
                                           → 応答をキャッシュに保存

実装例

import numpy as np
from datetime import datetime, timedelta

class SemanticCache:
    def __init__(
        self,
        embedding_model: str = "text-embedding-3-small",
        similarity_threshold: float = 0.92,
        ttl_hours: int = 24
    ):
        self.embedding_model = embedding_model
        self.similarity_threshold = similarity_threshold
        self.ttl = timedelta(hours=ttl_hours)
        self.cache: list[dict] = []

    async def get(self, query: str) -> dict | None:
        """キャッシュから類似の質問を検索する"""
        query_embedding = await self.embed(query)

        best_match = None
        best_score = 0.0

        for entry in self.cache:
            # TTL超過チェック
            if datetime.now() - entry["created_at"] > self.ttl:
                continue

            score = cosine_similarity(
                query_embedding, entry["embedding"]
            )
            if score > best_score and score >= self.similarity_threshold:
                best_score = score
                best_match = entry

        if best_match:
            best_match["hit_count"] += 1
            return {
                "answer": best_match["answer"],
                "similarity": best_score,
                "cached": True
            }
        return None

    async def put(self, query: str, answer: str) -> None:
        """回答をキャッシュに保存する"""
        embedding = await self.embed(query)
        self.cache.append({
            "query": query,
            "answer": answer,
            "embedding": embedding,
            "created_at": datetime.now(),
            "hit_count": 0
        })

キャッシュヒット率の最適化

閾値チューニング

セマンティックキャッシュの品質は類似度閾値で決まります。

最初は閾値0.95で始めて、ユーザーフィードバックを見ながら徐々に下げるのが安全だ。 — 田中VPoE

キャッシュ戦略の最適化テクニック

# 質問の正規化パイプライン
def normalize_query(query: str) -> str:
    """質問を正規化してキャッシュヒット率を向上させる"""
    # 1. 全角→半角変換
    query = unicodedata.normalize("NFKC", query)

    # 2. 表記ゆれの統一
    replacements = {
        "有休": "有給休暇",
        "有給": "有給休暇",
        "年休": "有給休暇",
        "経費精算": "経費精算",
        "経費清算": "経費精算",
    }
    for old, new in replacements.items():
        query = query.replace(old, new)

    # 3. 余分な空白・記号の除去
    query = re.sub(r'\s+', ' ', query).strip()
    query = re.sub(r'[？?！!。、]', '', query)

    return query

コスト削減効果のシミュレーション

前提:
  月間クエリ数: 100,000
  GPT-4oの1クエリあたりコスト: ¥15
  Embeddingの1クエリあたりコスト: ¥0.1

キャッシュなし:
  月額 = 100,000 × ¥15 = ¥1,500,000

キャッシュヒット率 35%の場合:
  LLM呼び出し = 65,000 × ¥15      = ¥975,000
  Embedding   = 100,000 × ¥0.1    = ¥10,000
  キャッシュ応答 = 35,000 × ¥0     = ¥0
  月額合計                          = ¥985,000

削減額: ¥515,000/月（34%削減）
年間削減額: ¥6,180,000

RAGチャンキング戦略のコスト影響

チャンクサイズとトークン数の関係

RAGではドキュメントを「チャンク」に分割してベクトル化し、質問に関連するチャンクを検索してLLMに渡します。チャンクサイズはコストに直結します。

最適なチャンキング戦略

# 階層チャンキングの例
class HierarchicalChunker:
    """
    親チャンク（要約）で検索し、
    子チャンク（詳細）をLLMに渡す。
    検索精度とコスト効率を両立する。
    """

    def chunk_document(self, document: str) -> list[dict]:
        # 1. セクション単位で大きく分割（親チャンク）
        sections = self.split_by_sections(document)

        chunks = []
        for section in sections:
            # 2. 親チャンクの要約を生成（検索用）
            parent_summary = self.summarize(section)

            # 3. 段落単位で細かく分割（子チャンク）
            paragraphs = self.split_by_paragraphs(section)

            for para in paragraphs:
                chunks.append({
                    "parent_summary": parent_summary,  # 検索用（200トークン）
                    "child_content": para,              # LLM用（300トークン）
                    "section_title": section.title
                })

        return chunks

    # 検索時: parent_summaryでベクトル検索
    # LLM送信時: child_contentのみ送信（トークン節約）

ベクトル検索の最適化

リランキングによる検索精度向上

ベクトル検索の結果をリランカーで再順位付けし、本当に関連性の高いチャンクのみをLLMに渡します。

検索パイプライン:

  質問
   │
   ▼
  ベクトル検索（粗い検索）
   │ top-20を取得
   ▼
  リランカー（精密な再順位付け）
   │ top-3に絞り込み
   ▼
  LLMに送信
   │ 3チャンク分のコンテキストのみ
   ▼
  回答生成

  リランキングなし: top-5を送信 → 2,500トークン
  リランキングあり: top-3を送信 → 1,500トークン
  トークン削減: 40%

ハイブリッド検索

ベクトル検索とキーワード検索を組み合わせることで、検索精度を向上させます。

# ハイブリッド検索の実装例
async def hybrid_search(
    query: str,
    collection: str,
    vector_weight: float = 0.7,
    keyword_weight: float = 0.3,
    top_k: int = 3
) -> list[dict]:
    """
    ベクトル検索とキーワード検索の結果を
    重み付けスコアで統合する。
    """
    # 1. ベクトル検索
    vector_results = await vector_search(
        query=query,
        collection=collection,
        top_k=top_k * 3  # 多めに取得
    )

    # 2. キーワード検索（BM25）
    keyword_results = await bm25_search(
        query=query,
        collection=collection,
        top_k=top_k * 3
    )

    # 3. Reciprocal Rank Fusionでスコア統合
    fused = reciprocal_rank_fusion(
        results_list=[vector_results, keyword_results],
        weights=[vector_weight, keyword_weight]
    )

    return fused[:top_k]

プロンプトキャッシング

Anthropic/OpenAIのプロンプトキャッシング機能

プロバイダが提供するプロンプトキャッシング機能を使うと、同じプレフィックスを持つリクエストのトークンコストを大幅に削減できます。

仕組み

リクエスト1:
  [システムプロンプト(800)][RAGコンテキスト(1000)][ユーザー質問A(200)]
  → 全トークン(2,000)を通常料金で課金

リクエスト2（同じシステムプロンプトの場合）:
  [システムプロンプト(800)][RAGコンテキスト(1200)][ユーザー質問B(150)]
  → プレフィックス一致部分(800)はキャッシュ料金（90%OFF）
  → 残り(1,350)は通常料金

効果:
  キャッシュなし: 2,150 × $3.00/1M = $0.00645
  キャッシュあり: 800 × $0.30/1M + 1,350 × $3.00/1M = $0.00429
  削減率: 約33%

プロンプトキャッシングの最適化

# プロンプトキャッシングを意識したメッセージ構成
messages = [
    # 1. システムプロンプト（全リクエストで共通 → キャッシュされる）
    {
        "role": "system",
        "content": STATIC_SYSTEM_PROMPT,  # 800トークン
        "cache_control": {"type": "ephemeral"}  # Anthropic API
    },
    # 2. 共通RAGコンテキスト（カテゴリ別で共通 → キャッシュ可能性あり）
    {
        "role": "user",
        "content": f"参考情報:\n{common_context}",  # 500トークン
        "cache_control": {"type": "ephemeral"}
    },
    # 3. 個別の質問（毎回変わる → キャッシュされない）
    {
        "role": "user",
        "content": user_query  # 200トークン
    }
]

まとめ

チェックリスト

• セマンティックキャッシュの仕組みと閾値チューニングを理解した
• キャッシュヒット率向上のテクニック（正規化、プリウォーム）を理解した
• RAGチャンキング戦略がコストに与える影響を理解した
• リランキングとハイブリッド検索の最適化手法を理解した
• プロンプトキャッシング（Anthropic/OpenAI）の仕組みを理解した

次のステップへ

次は「AI FinOps実践」です。コスト削減の技術を組織的に運用するためのフレームワーク — FinOpsをAIシステムに適用する方法を学びましょう。

推定読了時間: 30分