LLMの同一質問を高速化する：Redisを使ったセマンティックキャッシュの設計と検証

Last updated at 2026-10-18Posted at 2026-10-18

LLMアプリケーションでは、表現が少し違うだけの質問が繰り返し送られることがあります。たとえば「有給休暇の申請方法は？」と「有休を取るにはどうすればいい？」は、利用者の意図としてはかなり近い質問です。

毎回LLMを呼び出すと、応答時間だけでなくAPIコストも増えます。そこで今回は、Redisに質問文のベクトルと回答を保存し、意味的に近い質問にはキャッシュを返す「セマンティックキャッシュ」を実装しました。

手元で検証した結論としては、単純な文字列一致では拾えない言い換えを扱える一方、類似度の閾値を低くしすぎると誤った回答を返す危険があります。したがって、モデルやプロンプトのバージョン、テナント情報などをキャッシュの識別条件に含める設計が重要です。

全体の構成

セマンティックキャッシュを利用した処理全体の流れは次の通りです。

今回の処理は次の流れにしました。

ユーザーの質問を埋め込みベクトルに変換する
Redis Vector Searchで近い質問を検索する
距離が閾値以内なら保存済みの回答を返す
見つからなければLLMを呼び出す
質問・ベクトル・回答をRedisへ保存する

埋め込みモデルには、ローカルで動作する sentence-transformers の all-MiniLM-L6-v2 を使用しました。このモデルのベクトル次元数は384です。

RedisはVector Searchを利用できるRedis Stackを起動します。

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

pip install redis sentence-transformers numpy

本番環境では、利用するRedisのバージョンやイメージを固定し、アプリケーションと同じ埋め込みモデルを検証環境でも使用します。

Redisにベクトル検索用インデックスを作る

Redisのハッシュに回答データを保存し、embedding フィールドをベクトル検索対象にします。距離計算にはCOSINEを指定しました。

import redis
from redis.commands.search.field import TextField, VectorField
from redis.commands.search.index_definition import IndexDefinition, IndexType

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

INDEX_NAME = "idx:semantic_cache"
PREFIX = "cache:"

try:
    r.ft(INDEX_NAME).info()
except Exception:
    schema = (
        TextField("question"),
        TextField("answer"),
        TextField("model_version"),
        TextField("prompt_version"),
        VectorField(
            "embedding",
            "HNSW",
            {
                "TYPE": "FLOAT32",
                "DIM": 384,
                "DISTANCE_METRIC": "COSINE",
                "M": 16,
                "EF_CONSTRUCTION": 200,
            },
        ),
    )

    r.ft(INDEX_NAME).create_index(
        schema,
        definition=IndexDefinition(
            prefix=[PREFIX],
            index_type=IndexType.HASH,
        ),
    )

インデックス作成はアプリケーションの起動時に毎回実行しても問題ないよう、存在確認を入れています。厳密なマイグレーション管理が必要な環境では、起動処理とは分けて実行する方が扱いやすいでしょう。

キャッシュの検索と保存

次に、質問をベクトル化して近いキャッシュを検索します。COSINE距離は、値が小さいほど近いことを表します。

import time
import uuid
import numpy as np
from sentence_transformers import SentenceTransformer
from redis.commands.search.query import Query

model = SentenceTransformer("all-MiniLM-L6-v2")

CACHE_TTL_SECONDS = 60 * 60
DISTANCE_THRESHOLD = 0.12

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

def find_cached_answer(question: str):
    vector = embed(question)

    query = (
        Query("*=>[KNN 1 @embedding $vec AS distance]")
        .sort_by("distance")
        .return_fields(
            "question",
            "answer",
            "model_version",
            "prompt_version",
            "distance",
        )
        .paging(0, 1)
        .dialect(2)
    )

    result = r.ft(INDEX_NAME).search(
        query,
        query_params={"vec": vector},
    )

    if not result.docs:
        return None

    doc = result.docs[0]
    distance = float(doc.distance)

    if distance > DISTANCE_THRESHOLD:
        return None

    return {
        "answer": doc.answer.decode("utf-8")
        if isinstance(doc.answer, bytes)
        else doc.answer,
        "distance": distance,
        "matched_question": doc.question.decode("utf-8")
        if isinstance(doc.question, bytes)
        else doc.question,
    }

def save_answer(question: str, answer: str):
    cache_id = f"{PREFIX}{uuid.uuid4().hex}"

    r.hset(
        cache_id,
        mapping={
            "question": question,
            "answer": answer,
            "model_version": "all-MiniLM-L6-v2",
            "prompt_version": "faq-v1",
            "created_at": str(int(time.time())),
            "embedding": embed(question),
        },
    )
    r.expire(cache_id, CACHE_TTL_SECONDS)

検索結果の回答は、距離が閾値以内の場合だけ利用します。今回の 0.12 はあくまで動作確認用の初期値です。実際には、想定される質問と誤った質問のサンプルを用意し、距離分布を確認して決める必要があります。

LLM呼び出しに組み込む

LLMを呼び出す部分を call_llm として分離すると、利用するモデルやSDKを変更しやすくなります。

def call_llm(question: str) -> str:
    # 実際の環境では、ここでLLM APIまたは社内モデルを呼び出す
    return f"LLMによる回答: {question}"

def answer_question(question: str) -> dict:
    cached = find_cached_answer(question)

    if cached is not None:
        return {
            "answer": cached["answer"],
            "cache_hit": True,
            "distance": cached["distance"],
        }

    answer = call_llm(question)
    save_answer(question, answer)

    return {
        "answer": answer,
        "cache_hit": False,
        "distance": None,
    }

検証時には、次のような質問を順番に実行しました。

questions = [
    "有給休暇の申請方法は？",
    "有休を取得するにはどうすればいい？",
    "パスワードを変更する方法を教えてください",
]

for question in questions:
    result = answer_question(question)
    print(question)
    print(result)

1件目はLLM呼び出し後に保存され、2件目は表現が異なっていても近いベクトルとして検索されます。3件目は距離が閾値を超えるため、キャッシュミスとして扱われます。

実運用で注意する点

セマンティックキャッシュでは、類似しているからといって必ず同じ回答になるとは限りません。特に「先月の売上」と「今月の売上」のように、単語の違いが重要な質問では注意が必要です。

実運用では、次の情報をキャッシュ条件に含める設計にしました。

model_version
prompt_version
テナントIDや権限範囲
言語や地域
データの基準日時

また、回答の内容が更新された場合は、TTLに任せるだけでなく、関連するキャッシュを削除できる仕組みも必要です。FAQの更新や商品情報の変更を検知して、該当するキーを削除する運用が考えられます。

ログには質問全文をそのまま残さず、キャッシュヒット率、検索距離、応答時間、LLM呼び出し回数などを記録します。個人情報や機密情報が含まれる可能性がある場合は、保存可否を先に決めるべきです。

まとめ

RedisのVector Searchを使うと、LLMへの質問を文字列ではなく意味の近さでキャッシュできます。今回の実装では、質問を埋め込みに変換し、COSINE距離で近い回答を検索しました。

効果を安定させるポイントは、次の3つです。

類似度の閾値をサンプルデータで調整する
モデルやプロンプトのバージョンをキャッシュ条件に含める
誤回答を防ぐため、質問の種類によってキャッシュ利用を制限する

まずはFAQや社内手順のように回答が比較的安定している領域から導入し、ヒット率と誤ヒットを計測しながら適用範囲を広げるのが現実的です。

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