Antigravity で LLM 応答をセマンティックキャッシュする — 削減率を自分の数字で見積もる本番実装

text-embedding-004 からの移行で足をすくわれた3点

差し替えの前に、ここだけは押さえておいてください。3点とも例外を出さずに精度や順位だけが静かに劣化する種類の問題で、後から気づくのが難しい性質を持っています。

次元数が 768 固定から 3072 既定へ変わります。 text-embedding-004 は 768 次元でした。gemini-embedding-001 の既定は 3072 次元です。既存の vector(768) カラムへそのまま入れれば次元不一致で弾かれるので、これは気づけます。厄介なのは「では 3072 のまま持てばいいのか」という判断のほうです。キャッシュのテーブルは TTL で回転するとはいえヒット率を上げるほど行数が増えますし、HNSW インデックスはベクトルをメモリに載せて効きます。RAG の検索精度ほどの解像度をキャッシュ判定に求める必要はないと考え、私は output_dimensionality に 768 を指定して切り詰める側を選びました。スキーマを据え置けるのも実務上は大きな利点です。

切り詰めたベクトルは正規化されていません。 ここが一番静かな罠でした。Matryoshka 表現学習によって先頭 768 次元だけでも意味は保たれますが、切り出した時点でノルムは 1 から外れます。3072 次元の単位ベクトル 5,000 本を先頭 768 次元へ切り詰めて手元で確かめたところ、ノルムは 0.4619〜0.5376（平均 0.5001）に散らばり、最大と最小で 1.164 倍の開きがありました。この状態で 1,000 クエリぶん近傍を取ると、コサイン距離で選んだ最近傍と内積で選んだ最近傍が 126 回（12.6%）食い違いました。L2 再正規化を挟んだ後は食い違いが 0 回になります。合成ベクトルでの性質確認なので比率そのものを実際の埋め込み分布に持ち込むことはできませんが、「切り詰めたら正規化する」という結論の根拠としては十分でした。

pgvector のコサイン距離演算子 <=> を使う限り、この差は順位に出ません。距離の計算過程でノルムが割り落とされるからです。危ないのは、速度目当てで内積演算子 <#> に切り替えたときです。演算子を1文字変えただけで、しきい値の意味が静かにずれます。切り詰めるなら自前で正規化しておく、と決めてしまうのが安全です。

タスクタイプはインスタンスに固定しないでください。 RAG では保存側に RETRIEVAL_DOCUMENT、検索側に RETRIEVAL_QUERY を指定するのが定石です。ところがセマンティックキャッシュが比べているのは文書とクエリではなく、過去のクエリと今のクエリという同種のテキスト同士です。ここで RAG の作法を持ち込むと、同じ文を保存時と検索時で別々のタスクタイプに通すことになり、本来一致すべきベクトルがずれます。キャッシュ用途では保存・検索の両方を SEMANTIC_SIMILARITY で揃えるのが筋です。ラッパーライブラリ経由で使う場合は、インスタンス側にタスクタイプを設定してしまうと呼び出し側で上書きしたつもりが効いていない、という事故が起きやすいので、引数で毎回明示する書き方を勧めます。

Antigravity を使って最小動作版を 30 分で組む

ここから実装に入ります。Antigravity のエージェントを Manager モードで起動し、まずは下敷きとなる「骨格」を生成させ、私が細部を詰めるという進め方を取ります。Antigravity に丸投げすると抽象度の高いコードになりがちなので、実行環境の前提（Python 3.11、FastAPI、google-genai SDK、pgvector 拡張済みの Postgres 16）を先に明示することがコツです。

次のコードは、サポートボットに組み込むことを想定した SemanticCache クラスの骨格です。pgvector 拡張が有効な Postgres を前提としています。初版から少し手を入れてあり、埋め込みの正規化・接続ごとの型コーデック登録・世代フィンガープリントの3点を最初から組み込んでいます。いずれも後から足すと既存キャッシュの作り直しが必要になる部分です。

# semantic_cache.py — FastAPI アプリに組み込む想定
# 前提: PostgreSQL 16 + pgvector 拡張、google-genai SDK、asyncpg、pgvector[asyncpg]
# 期待出力: 初回は LLM 呼び出し、2回目以降は類似度 0.92 以上でキャッシュからヒット
 
import asyncio
import hashlib
import json
import math
import os
 
import asyncpg
from google import genai
from google.genai import types
from pgvector.asyncpg import register_vector
 
EMBEDDING_MODEL = "gemini-embedding-001"
EMBEDDING_DIM = 768                      # 既定は 3072。切り詰める場合は自前で正規化します
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"])
 
 
def l2_normalize(values: list[float]) -> list[float]:
    """MRL で切り詰めたベクトルはノルムが 1 から外れるため、明示的に正規化します。"""
    norm = math.sqrt(sum(v * v for v in values))
    if norm == 0.0:
        raise ValueError("embedding API returned a zero vector")
    return [v / norm for v in values]
 
 
def prompt_fingerprint(system_instruction: str, model: str, temperature: float) -> str:
    """システム指示・モデル・温度の組を 16 桁に畳みます。1つでも変われば別世代のキャッシュです。"""
    raw = json.dumps(
        {"system": system_instruction, "model": model, "temperature": temperature},
        sort_keys=True,
        ensure_ascii=False,
    )
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
 
 
async def create_pool(dsn: str) -> asyncpg.Pool:
    """接続ごとに vector 型のコーデックを登録します。これを忘れると list を渡せません。"""
    return await asyncpg.create_pool(dsn, init=register_vector)
 
 
class SemanticCache:
    def __init__(self, pool: asyncpg.Pool, fingerprint: str):
        self.pool = pool
        self.fingerprint = fingerprint
 
    async def _embed(self, text: str) -> list[float]:
        """保存・検索の両方を SEMANTIC_SIMILARITY で揃えます（クエリ同士を比べるため）。"""
        result = await asyncio.to_thread(
            client.models.embed_content,
            model=EMBEDDING_MODEL,
            contents=text,
            config=types.EmbedContentConfig(
                task_type="SEMANTIC_SIMILARITY",
                output_dimensionality=EMBEDDING_DIM,
            ),
        )
        return l2_normalize(result.embeddings[0].values)
 
    async def lookup(self, query: str, lang: 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 fingerprint = $2
                  AND lang = $3
                  AND created_at > NOW() - ($4::int * INTERVAL '1 second')
                ORDER BY embedding <=> $1::vector
                LIMIT 1
                """,
                vec, self.fingerprint, lang, CACHE_TTL_SECONDS,
            )
        if row and row["similarity"] >= SIMILARITY_THRESHOLD:
            return (row["response"], row["similarity"])
        return None
 
    async def store(self, query: str, lang: str, response: str) -> None:
        """クエリと応答をペアで保存します。世代と言語をキーに含めるのが要点です。"""
        vec = await self._embed(query)
        async with self.pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO llm_cache (query, lang, fingerprint, embedding, response)
                VALUES ($1, $2, $3, $4::vector, $5)
                """,
                query, lang, self.fingerprint, vec, response,
            )
 
    async def ask(self, query: str, lang: str = "ja") -> dict:
        """外部に公開するエントリポイント。ヒット/ミスを記録します。"""
        hit = await self.lookup(query, lang)
        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, lang, text)
        return {"response": text, "cached": False, "similarity": None}

対応するテーブル定義は次の通りです。lang と fingerprint をカラムとして持たせているのは、後半で扱う多言語の誤ヒットとプロンプト変更時の事故を、運用の注意ではなくスキーマの構造で止めるためです。

-- schema.sql — 初回のみ psql で実行
CREATE EXTENSION IF NOT EXISTS vector;
 
CREATE TABLE llm_cache (
  id BIGSERIAL PRIMARY KEY,
  query TEXT NOT NULL,
  lang TEXT NOT NULL,                    -- 言語をまたいだ誤ヒットを構造で止めます
  fingerprint TEXT NOT NULL,             -- システム指示・モデル・温度の世代
  embedding vector(768) NOT NULL,        -- output_dimensionality=768 で切り詰めた正規化済みベクトル
  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_scope_idx ON llm_cache (fingerprint, lang, created_at DESC);

ここまでで「意味が近ければキャッシュから返す」最小動作版が完成です。実際に curl で2回同じ意図の質問を投げてみると、1回目は cached: false で LLM 応答が返り、2回目は cached: true で類似度 0.93〜0.97 の値が付いて帰ってきます。

類似度しきい値のチューニング — ヒット率と誤ヒットのバランスを取る

ここが本番運用における最大の勘所です。しきい値を下げれば下げるほどヒット率は上がりますが、意味の違うクエリに同じ応答を返す「誤ヒット」のリスクも増えます。逆に厳しく設定すれば安全ですが、コスト削減効果は小さくなります。

私の経験則では、初期値 0.92 から始め、実データを見ながら 0.90〜0.95 のどこかに落ち着かせるのが現実的です。ただし「経験則だけで決める」のは事故の元なので、データを溜めて可視化する仕組みを最初から仕込んでおくべきです。

次のコードは、全クエリの類似度とヒット判定を記録するロガーです。後で分析できるよう、応答の質についての人間による評価カラムも用意しておきます。

# 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;

この結果を眺めると、類似度 0.88〜0.92 の帯に「ギリギリ返していないが実は同じ意図」のクエリが集中していることが多いです。そこで帯別にランダムサンプリングして人間が品質を評価し、誤ヒット率が1%未満に収まる最も低いしきい値を本番値として選びます。この作業は完全に Antigravity のエージェントに任せるのは避け、自分の目で評価するべきです。なぜなら「その応答が妥当か」は最終的にサービスの設計者が持つべき判断だからです。

キャッシュ無効化とプライバシー — 設計段階で織り込むべき現実

セマンティックキャッシュには、通常の KV キャッシュにはない独特の無効化課題があります。また、ユーザー発話そのものを保存することに伴うプライバシー配慮も欠かせません。

無効化の主な方法は3つあります。1つ目は TTL による時間ベースの失効で、先のコードでは1週間としています。ドキュメント類のようにほぼ不変な応答には長め（30日）、価格や在庫のように変動するものには短め（数時間）に設定します。2つ目はバージョンベースの失効で、プロンプトやシステム指示を変更したときに全キャッシュをフラッシュします。これは cache_version カラムを追加し、現在のバージョンと一致するキャッシュのみ有効とする実装になります。3つ目はタグベースの選択的失効で、テナント別・カテゴリ別にキャッシュを分割し、特定範囲だけパージできるようにします。

プライバシーについては、ユーザー発話にメールアドレス・電話番号・社内識別子などの PII が混ざる可能性を前提に設計します。最低限の対策として、保存前に正規表現で PII を検出したクエリはキャッシュ対象から除外するのが現実的です。より厳格には、ユーザー別のテナント分離を実装し、ユーザー A のクエリがユーザー B のキャッシュにヒットしないようにします。この論点は Antigravity エージェントのコスト最適化ガイド でも軽く触れていますが、コストと安全性はトレードオフの関係にあることを忘れないでください。

# 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)

なお、エージェント運用のセキュリティ全般については LLMOps モニタリング完全ガイド に体系的な整理があります。合わせて読むと、この記事で触れられなかった監査ログ・異常検知の文脈が補完できます。

よくある間違い・落とし穴 — 実際に踏んだ4つの痛い事例

ここは私が顔から火が出るほど痛い経験をした実話ベースです。同じ失敗を踏まずに済むよう、なるべく具体的に書き残します。

落とし穴1: システム指示やプロンプトテンプレートの変更をキャッシュキーに含めていない

ある日、ボットの口調を「丁寧」から「親しみやすい」に変えたところ、キャッシュからは前の口調の応答が返り続けるという事故が発生しました。原因は、ルックアップキーを「ユーザークエリの埋め込み」だけで作っていたことです。口調を決めているのはシステム指示なのに、キーのどこにもシステム指示が入っていませんでした。

この記事のコードで fingerprint カラムを最初から持たせているのは、この失敗があったからです。prompt_fingerprint() はシステム指示・モデル名・温度を JSON に畳んで SHA256 の先頭16桁を返します。手元で口調の一文だけを差し替えて確かめると、85883ac41c147fe2 が 9cbd2a2355bf6672 に変わりました。指紋が変われば検索条件から外れるので、古い世代の応答は自然に返らなくなります。TTL を待つ必要も、手でフラッシュする必要もありません。

落とし穴2: 時系列依存の質問で古い情報を返す

「今日の為替は？」「最新のリリース情報は？」といった時系列依存の質問は、セマンティックキャッシュとの相性が最も悪い部類です。前日の応答を今日返してしまい、ユーザーから不信感を買うことになります。対策は2つあります。1つは分類器（軽量な LLM か正規表現）で「時系列依存クエリ」を検出し、ルックアップそのものをスキップすること。もう1つは、日付・時刻・最新といったキーワードを含むクエリの TTL を5分以下に短縮することです。私は最初の方針を採用しています。

落とし穴3: 埋め込み生成コストを忘れてキャッシュで赤字になる

驚くことに、短いクエリだけを扱うシステムでは、埋め込み生成のコストが LLM 呼び出しコストを上回ることがあります。埋め込み料金はミスのときだけでなく、ヒットしたときにも必ず発生するからです。生成側に安価なモデルを選んでいるほど、この差は詰まります。Gemini Flash で節約できる金額より、毎回の埋め込み料金のほうが高くつく、という逆転が起こり得ます。

対策は、クエリ文字列のハッシュで先に完全一致キャッシュを引き、外れたときだけセマンティック層へ進む2層構造です。次の節で、この構造を入れると削減率の式がどう変わるのかまで含めて扱います。

落とし穴4: 多言語サービスで英語クエリが日本語応答にヒットする

多言語モデルの埋め込みは、同じ意味の文を言語を越えて近いベクトルに置きます。これは RAG では美点ですが、キャッシュでは誤ヒットの原因になります。「How do I cancel?」が日本語の「キャンセル方法は？」の応答（日本語）にヒットすると、英語ユーザーに日本語応答が返ってしまいます。対策はシンプルで、言語コードをキャッシュキーに含めることです。この記事のスキーマで lang を独立したカラムにしているのがそれにあたります。判定を後から足すと既存レコードの言語が不明のまま残るので、最初からカラムを切っておくほうが手戻りがありません。自動判定には langdetect などの軽量ライブラリで十分です。

削減率を自分の数字で見積もる — 2層キャッシュと単価比

「セマンティックキャッシュでコストが何割減るのか」は、実装前にいちばん知りたいことでしょう。私も最初は他所の事例の数字を眺めていましたが、条件が違えば結果も違うので、あまり参考になりませんでした。式にしてしまえば、自分のサービスの数字を入れて確かめられます。

まず2層構造を入れます。完全一致層はハッシュを引くだけなので、埋め込み料金がかかりません。

# two_layer.py — 完全一致層をセマンティック層の前に置きます
import hashlib
 
def exact_key(query: str, fingerprint: str, lang: str) -> str:
    """空白の揺れだけで別キーになるのを防ぐため、正規化してから畳みます。"""
    normalized = " ".join(query.split())
    payload = "\x1f".join([fingerprint, lang, normalized])
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
 
 
async def ask_two_layer(cache, redis, query: str, lang: str = "ja") -> dict:
    key = exact_key(query, cache.fingerprint, lang)
    cached = await redis.get(key)
    if cached is not None:
        return {"response": cached, "cached": True, "layer": "exact"}
 
    result = await cache.ask(query, lang)          # ここで初めて埋め込み料金が発生します
    await redis.set(key, result["response"], ex=60 * 60 * 24)
    result["layer"] = "semantic"
    return result

" ".join(query.split()) を挟んでいるのは、全角空白や連続スペースの揺れだけで別キーになるのを避けるためです。手元で "解約 したい " と "解約 したい" を通したところ、同じキーに落ちることを確認しています。地味ですが、完全一致層のヒット率はこの前処理でそこそこ変わります。

そのうえで、削減率は次の形に分解できます。ヒット率 h、完全一致層で捌ける割合 e、埋め込み1回と LLM 1回の単価比 c として、

削減率 = h − (1 − e) × c

第1項が節約、第2項が「ヒットしてもミスしても払う埋め込み代」です。単価比 c は生成モデルと平均トークン数で決まるので、請求書から自分の値を出せます。いくつかの組み合わせを並べてみます。

読み取れることが2つあります。ひとつは、単価比 c が小さいうちは削減率がほぼヒット率と同じになること。もうひとつは、生成側を安いモデルに寄せて c が 0.20 まで上がると、同じヒット率でも削減率が 20 ポイント近く落ちること。そして4行目と5行目を見比べると、完全一致層を足すだけで 62.0% が 77.0% まで戻ります。安いモデルを使っているサービスほど、2層構造の効き目が大きいという関係です。

「8割減った」という数字を見かけたら、それは c が十分小さく、かつヒット率が 8 割を超えていた条件での話だと考えたほうがよさそうです。私自身の環境でもその条件は満たしていましたが、条件のほうを先に確かめるのが順序として正しいと、いまは思っております。

監視メトリクスと A/B テスト — 効果を数字で語れるようにする

作って終わりではなく、削減効果を経営に説明できる数字にして初めて仕事として完結します。私が実運用で必ず計装する4つのメトリクスを紹介します。

1. キャッシュヒット率（時間帯別・テナント別）
2. 類似度分布ヒストグラム（しきい値が適切か判断するため）
3. 推論コスト削減額（ヒット数 × 平均トークン単価で概算）
4. 応答遅延 p50 / p95（ヒット時とミス時を分けて）

Prometheus 形式のエクスポートなら、Antigravity のエージェントに「この Python コードに Prometheus メトリクスを追加して」と頼むと適切なコードが出てきます。重要なのは「何を計装したいか」を自分で決めてから渡すことです。エージェントに目的から任せると、なぜか全メトリクスを RED（Rate, Error, Duration）に揃えてしまい、ヒット率のようなドメイン固有の値が抜け落ちることがあります。

数字の前に、品質が落ちていないことを確かめる

ヒット率とコストの数字が良くなっても、返している内容が劣化していれば意味がありません。ここで効くのは golden dataset です。ログから実クエリを 100〜300 件拾い、期待する応答（または許容範囲）を人手で注記しておき、夜間に流し直してキャッシュ経由の応答と突き合わせます。文字列の完全一致では言い回しの揺れを取りこぼすので、判定は LLM に採点させるのが現実的でした。採点用のトークン代は先に予算に入れておいてください。品質の劣化を利用者からの申告で知るのは、いちばん高くつく学び方です。

より踏み込むなら、ヒットしたクエリの一部（1% 程度）を裏で本物の LLM にも投げ、キャッシュ応答とどれだけ食い違うかを常時測る方法があります。評価期間だけ LLM 料金が二重になるので、サンプル率は小さく保つのが前提です。

A/B テストは、キャッシュあり群となし群をユーザー ID のハッシュで振り分け、1週間ほど並走させて以下を比較します。ユーザー満足度（サムズアップ率）、再質問率（直後に類似質問が来る率）、平均レスポンスタイム、1日あたりのコスト。私の環境では、満足度が誤差範囲（±1ポイント以内）に収まり、平均レスポンスタイムが 1.4 秒から 0.3 秒台へ短縮されました。コストのほうは前節の式どおりで、ヒット率が 8 割台に乗り、単価比が十分小さい期間だけ削減率も 8 割前後になっています。つまりこの数字はキャッシュの性能というより、その週のクエリの重複具合を映したものです。他所の事例の削減率をそのまま自分の見積もりに使わないほうがいい理由が、ここにあります。

もう一点、1週間では足りない場合があります。平日と週末で問い合わせの傾向が変わるサービスでは、開始曜日を変えるだけで結論が動きます。余裕があれば2週間、難しければ 5% 程度の常設カナリアとして走らせ続けるのが現実的です。

全体を振り返って — 今日からできる最初の一歩

長く書きましたが、今日始めるべきことは1つだけです。現在運用中のチャットボットやサポートシステムのログから、直近1週間のクエリを 500〜1000 件サンプリングし、重複を人の目で数えてみてください。体感で「これ前も見た」と感じるクエリが 30% 以上あれば、セマンティックキャッシュの投資回収はほぼ確実です。逆に 10% 未満なら、他のコスト最適化手法（プロンプト圧縮・モデル選択の見直し）を先に検討する価値があります。

この記事の構成通りに実装すると、最小動作版は半日、監視とチューニングまで含めても2〜3日で本番投入できる範囲に収まります。重要なのは「作ってから測る」のではなく「測る仕組みと同時に作る」ことです。しきい値・ヒット率・誤ヒット率の数字が見えていれば、改善のサイクルは驚くほど早く回り始めます。