Antigravity で LLM 応答をセマンティックキャッシュする — 類似クエリでコストを80%削減する本番実装ガイド

# semantic_cache.py — FastAPI アプリに組み込む想定
# 前提: PostgreSQL 16 + pgvector 拡張、google-genai SDK、asyncpg
# 期待出力: 初回は LLM 呼び出し、2回目以降は類似度 0.92+ ならキャッシュからヒット
 
import os
import asyncio
import asyncpg
from google import genai
from google.genai import types
 
EMBEDDING_MODEL = "text-embedding-004"
GENERATION_MODEL = "gemini-2.5-flash"
SIMILARITY_THRESHOLD = 0.92  # 0.90-0.95 の間で後ほどチューニングします
CACHE_TTL_SECONDS = 60 * 60 * 24 * 7  # 1週間で自動無効化
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
class SemanticCache:
    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool
 
    async def _embed(self, text: str) -> list[float]:
        """Gemini の埋め込みモデルで 768 次元ベクトルを取得します。"""
        result = await asyncio.to_thread(
            client.models.embed_content,
            model=EMBEDDING_MODEL,
            contents=text,
            config=types.EmbedContentConfig(task_type="SEMANTIC_SIMILARITY"),
        )
        return result.embeddings[0].values
 
    async def lookup(self, query: str) -> tuple[str, float] | None:
        """類似クエリのキャッシュを探索。ヒットしたら (応答, 類似度) を返します。"""
        vec = await self._embed(query)
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(
                """
                SELECT response, 1 - (embedding <=> $1::vector) AS similarity
                FROM llm_cache
                WHERE created_at > NOW() - INTERVAL '%s seconds'
                ORDER BY embedding <=> $1::vector
                LIMIT 1
                """ % CACHE_TTL_SECONDS,
                vec,
            )
        if row and row["similarity"] >= SIMILARITY_THRESHOLD:
            return (row["response"], row["similarity"])
        return None
 
    async def store(self, query: str, response: str) -> None:
        """クエリと応答をペアで保存します。埋め込みは遅延生成可能ですが、
        ここでは同期的に行います (後述の"コスト逆転"を避けるための設計判断)。"""
        vec = await self._embed(query)
        async with self.pool.acquire() as conn:
            await conn.execute(
                "INSERT INTO llm_cache (query, embedding, response) VALUES ($1, $2::vector, $3)",
                query, vec, response,
            )
 
    async def ask(self, query: str) -> dict:
        """外部に公開するエントリポイント。ヒット/ミスを記録します。"""
        hit = await self.lookup(query)
        if hit:
            return {"response": hit[0], "cached": True, "similarity": hit[1]}
        # ミス時のみ LLM を呼び出し
        gen = await asyncio.to_thread(
            client.models.generate_content,
            model=GENERATION_MODEL,
            contents=query,
        )
        text = gen.text
        await self.store(query, text)
        return {"response": text, "cached": False, "similarity": None}

-- schema.sql — 初回のみ psql で実行
CREATE EXTENSION IF NOT EXISTS vector;
 
CREATE TABLE llm_cache (
  id BIGSERIAL PRIMARY KEY,
  query TEXT NOT NULL,
  embedding vector(768) NOT NULL,
  response TEXT NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  hit_count INT DEFAULT 0
);
 
-- コサイン距離用の HNSW インデックス (pgvector 0.5+)
CREATE INDEX llm_cache_embedding_idx
  ON llm_cache USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64);
 
CREATE INDEX llm_cache_created_at_idx ON llm_cache (created_at DESC);

# similarity_logger.py — 本番ログから最適しきい値を逆算する
 
async def log_decision(
    pool: asyncpg.Pool,
    query: str,
    top_similarity: float | None,
    used_cache: bool,
    response_preview: str,
) -> None:
    """しきい値判定の全イベントを記録。定期的に SQL で分析します。"""
    async with pool.acquire() as conn:
        await conn.execute(
            """
            INSERT INTO cache_decisions
              (query, top_similarity, used_cache, response_preview, created_at)
            VALUES ($1, $2, $3, $4, NOW())
            """,
            query,
            top_similarity,
            used_cache,
            response_preview[:200],  # プレビューのみ保存
        )
 
# 運用1週間後に実行する分析 SQL (psql で直接確認します)
# -- 類似度帯ごとのヒット数分布
# SELECT
#   width_bucket(top_similarity, 0.80, 1.00, 20) AS bucket,
#   COUNT(*) AS total,
#   COUNT(*) FILTER (WHERE used_cache) AS cache_hits
# FROM cache_decisions
# WHERE created_at > NOW() - INTERVAL '7 days'
#   AND top_similarity IS NOT NULL
# GROUP BY bucket ORDER BY bucket;

# privacy_filter.py — 保存前の PII スキャン
import re
 
PII_PATTERNS = [
    re.compile(r"[\w\.-]+@[\w\.-]+\.\w+"),           # メール
    re.compile(r"\b\d{3}-\d{4}-\d{4}\b"),            # 電話番号
    re.compile(r"\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b"),  # カード番号風
]
 
def contains_pii(text: str) -> bool:
    """PII が含まれていればキャッシュ対象外とします。"""
    return any(p.search(text) for p in PII_PATTERNS)
 
# 使用箇所 (store の前)
if contains_pii(query):
    # PII を含むクエリは LLM に投げるが保存はしない
    pass
else:
    await cache.store(query, response)