Redisと埋め込みでLLMのセマンティックキャッシュを実装し、応答速度とAPIコストを削減する

Last updated at 2026-11-06Posted at 2026-11-06

LLMを使ったアプリケーションでは、ユーザーの質問が少し違うだけで、毎回APIを呼び出してしまいます。たとえば「東京の天気を教えて」と「東京の天気は？」は、文字列としては異なりますが、必要な回答はほぼ同じです。

そこで今回は、質問文を埋め込みベクトルに変換し、Redisのベクトル検索で「意味が近い過去の質問」を探すセマンティックキャッシュを実装しました。

実際に手元で動作確認したところ、キャッシュヒット時はLLM API呼び出しを省略できるため、応答速度とAPIコストの改善に有効です。重要なのは、単純な文字列一致ではなく、類似度のしきい値を適切に設定することです。

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

セマンティックキャッシュによる処理全体の流れを図にすると次の通りです。

今回の処理フローは次のとおりです。

ユーザーの質問を埋め込みモデルでベクトル化
Redisから近いベクトルを検索
類似度がしきい値以内ならキャッシュを返す
見つからなければLLM APIを呼び出す
質問・回答・ベクトルをRedisに保存

通常のキャッシュは、URLや質問文をキーにします。しかし、質問文が少し変わるだけで別キーになるため、意味的には同じ質問を再利用できません。

セマンティックキャッシュでは、質問文を数百次元のベクトルに変換します。ベクトル間の距離が近ければ「似た質問」と判断し、その回答を再利用します。

Redis Stackを起動する

Redisのベクトル検索を利用するため、Redis StackをDockerで起動します。

docker run --name redis-semantic-cache \
  -p 6379:6379 \
  -d redis/redis-stack-server:latest

Python側では、Redisクライアント、埋め込みモデル、NumPyをインストールします。

pip install redis sentence-transformers numpy

今回の埋め込みモデルには、日本語を含む多言語テキストを扱えるモデルを使用します。

from sentence_transformers import SentenceTransformer

model = SentenceTransformer(
    "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)

vector = model.encode(
    "東京の天気を教えて",
    normalize_embeddings=True
)

print(vector.shape)
# (384,)

normalize_embeddings=Trueを指定すると、ベクトルを正規化できます。Redis側でコサイン距離を使う場合、この設定にしておくと扱いやすくなります。

Redisにベクトルインデックスを作成する

Redisには、質問文、LLMの回答、埋め込みベクトルをHashとして保存します。そのHashに対して、ベクトル検索用のインデックスを作成します。

import redis

r = redis.Redis(
    host="localhost",
    port=6379,
    decode_responses=False
)

try:
    r.execute_command("FT.INFO", "semantic_cache")
except redis.ResponseError:
    r.execute_command(
        "FT.CREATE",
        "semantic_cache",
        "ON", "HASH",
        "PREFIX", "1", "cache:",
        "SCHEMA",
        "prompt", "TEXT",
        "response", "TEXT",
        "embedding", "VECTOR", "HNSW", "6",
        "TYPE", "FLOAT32",
        "DIM", "384",
        "DISTANCE_METRIC", "COSINE"
    )

DIMは埋め込みベクトルの次元数です。使用するモデルによって異なるため、モデルを変更する場合は必ず合わせます。

DISTANCE_METRIC COSINEを指定しているため、検索結果のdistanceはコサイン距離になります。値が小さいほど意味が近いデータです。

キャッシュの検索と保存

次に、Redisから最も近いキャッシュを1件検索する関数を作ります。

import numpy as np

CACHE_DISTANCE_THRESHOLD = 0.15


def embed(text: str) -> bytes:
    vector = model.encode(
        text,
        normalize_embeddings=True
    )
    return np.asarray(vector, dtype=np.float32).tobytes()


def find_cached_answer(prompt: str):
    query_vector = embed(prompt)

    result = r.execute_command(
        "FT.SEARCH",
        "semantic_cache",
        "*=>[KNN 1 @embedding $vec AS distance]",
        "PARAMS", "2",
        "vec", query_vector,
        "SORTBY", "distance",
        "RETURN", "3",
        "response", "prompt", "distance",
        "DIALECT", "2"
    )

    if result[0] == 0:
        return None

    fields = result[2]
    values = dict(zip(fields[::2], fields[1::2]))

    distance = float(values[b"distance"])

    if distance > CACHE_DISTANCE_THRESHOLD:
        return None

    return {
        "answer": values[b"response"].decode("utf-8"),
        "matched_prompt": values[b"prompt"].decode("utf-8"),
        "distance": distance,
    }

キャッシュに保存する関数は次のようにしました。

import time
import uuid


def save_cache(prompt: str, response: str, ttl: int = 3600):
    key = f"cache:{uuid.uuid4().hex}"
    vector = embed(prompt)

    r.hset(
        key,
        mapping={
            "prompt": prompt,
            "response": response,
            "embedding": vector,
            "created_at": str(time.time()),
        }
    )

    r.expire(key, ttl)

TTLを設定しているのは、LLMの回答が古くなる可能性があるためです。ニュース、在庫、料金、社内データなどを扱う場合は、用途ごとにTTLを短くする必要があります。

LLM呼び出しに組み込む

LLM APIを呼び出す部分をcall_llmとして、キャッシュを先に確認する構成にします。

def call_llm(prompt: str) -> str:
    # 実際のLLM SDK呼び出しに置き換える
    return f"LLMが生成した回答: {prompt}"


def answer_question(prompt: str) -> str:
    cached = find_cached_answer(prompt)

    if cached is not None:
        print(
            f"cache hit: distance={cached['distance']:.4f}, "
            f"matched={cached['matched_prompt']}"
        )
        return cached["answer"]

    answer = call_llm(prompt)
    save_cache(prompt, answer)
    print("cache miss: LLM APIを呼び出しました")
    return answer

実行例です。

print(answer_question("東京の天気を教えて"))
print(answer_question("東京の天気は？"))

1回目はキャッシュミスになり、LLMを呼び出します。2回目は質問文が異なっていても、ベクトル距離がしきい値以下であれば、保存済みの回答を返します。

実運用で注意したこと

最も重要なのは、しきい値を固定値として盲目的に使わないことです。しきい値が高すぎると、意味の異なる質問に同じ回答を返す危険があります。逆に低すぎると、ほとんどキャッシュヒットしません。

私はまずログに以下の項目を出すようにしました。

元の質問
マッチした質問
ベクトル距離
キャッシュヒット・ミス
LLM APIの利用トークン
応答時間

そのうえで、実際の質問データを使ってしきい値を調整します。

また、ユーザーやテナントが異なるデータを共有しないように、検索対象を分離する必要があります。たとえば、Hashにtenant_idを保存して、Redisのタグフィルター付き検索を使います。

さらに、システムプロンプト、モデル名、回答の言語、参照データのバージョンもキャッシュのスコープに含めるべきです。これらが違う回答を同じキャッシュとして扱うと、意図しない回答が返る可能性があります。

まとめ

Redisと埋め込みモデルを組み合わせることで、文字列が完全一致しなくても過去の回答を再利用できるセマンティックキャッシュを実装できます。

ポイントは次の3つです。

質問を埋め込みベクトルに変換する
RedisのKNN検索で意味の近い質問を探す
距離のしきい値、TTL、データ分離を適切に設計する

LLMアプリケーションでは、同じような質問が繰り返されるケースが少なくありません。まずはログで質問の重複や類似度を確認し、再利用できそうな処理からセマンティックキャッシュを導入すると、効果を測定しやすくなります。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up