プロンプトのキャッシュ戦略
意味ベースのキャッシュ、完全一致キャッシュ、Anthropicのプロンプトキャッシュを学びます。
「プロンプトのキャッシュ戦略」はCoddyKit上の無料AI Prompt Engineeringレッスンです。 これはレッスン1/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Prompt Engineering学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Prompt Engineeringコースには全4レッスンが含まれています。
プロンプトの結果をキャッシュする理由
LLM APIの呼び出しには費用と時間がかかります。多くの本番アプリケーションでは、同じプロンプト、または非常によく似たプロンプトを繰り返し送信します。キャッシュを使うと、繰り返されたクエリに対して保存済みの結果を返せるため、重複したAPI呼び出しをなくし、コストとレイテンシーの両方を大幅に削減できます。
ハッシュキーによる完全一致キャッシュ
最も単純なキャッシュは、プロンプト文字列をそのままハッシュ化して結果を保存する方法です。同じプロンプト文字列が再び現れた場合は、APIを呼び出さずにキャッシュされた結果を返します。
import hashlib
import json
from functools import lru_cache
class ExactMatchCache:
def __init__(self, backend=None):
# backend: a dict (in-memory) or Redis client
self.store = backend or {}
def _key(self, messages, model, max_tokens):
content = json.dumps({'messages': messages, 'model': model,
'max_tokens': max_tokens}, sort_keys=True)
return 'llm:' + hashlib.sha256(content.encode()).hexdigest()
def get(self, messages, model, max_tokens):
key = self._key(messages, model, max_tokens)
return self.store.get(key)
def set(self, messages, model, max_tokens, result, ttl_seconds=3600):
key = self._key(messages, model, max_tokens)
self.store[key] = result
# In Redis: self.store.setex(key, ttl_seconds, json.dumps(result))
cache = ExactMatchCache()
# Usage
messages = [{'role': 'user', 'content': 'What is the capital of France?'}]
cached = cache.get(messages, 'gpt-4o-mini', 100)
if cached:
print('Cache HIT:', cached[:50])
else:
print('Cache MISS — calling API...')キャッシュ対応LLMクライアント
LLM APIの呼び出しをキャッシュデコレーターでラップすると、呼び出し側のコードを変更せずに、すべての呼び出し元が透過的にキャッシュを利用できます。
import openai
from typing import Optional
client = openai.OpenAI(api_key='YOUR_API_KEY')
cache = ExactMatchCache()
def cached_completion(messages, model='gpt-4o-mini', max_tokens=500,
temperature=0.0, use_cache=True) -> str:
if use_cache and temperature == 0.0:
# Only cache deterministic requests (temperature=0)
cached = cache.get(messages, model, max_tokens)
if cached:
return cached
response = client.chat.completions.create(
model=model,
messages=messages,
max_tokens=max_tokens,
temperature=temperature
)
result = response.choices[0].message.content
if use_cache and temperature == 0.0:
cache.set(messages, model, max_tokens, result)
return result
# Important: only cache temperature=0 responses
# Non-deterministic responses (temp>0) may return stale results
print('Cache wrapping: only deterministic (temp=0) calls are cached.')埋め込みによるセマンティックキャッシュ
セマンティックキャッシュは、文字列が完全に一致するクエリだけでなく、意味が似ているクエリに対してもキャッシュされた結果を返します。埋め込みベクトルとコサイン類似度を使って、重複に近いクエリを見つけます。
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class SemanticCache:
def __init__(self, similarity_threshold=0.95):
self.entries = [] # [(embedding, query, result)]
self.threshold = similarity_threshold
def embed(self, text):
'''Get embedding for text using OpenAI embeddings API.'''
response = client.embeddings.create(
model='text-embedding-3-small',
input=text
)
return np.array(response.data[0].embedding)
def get(self, query):
if not self.entries:
return None
query_emb = self.embed(query)
for emb, stored_query, result in self.entries:
sim = cosine_similarity([query_emb], [emb])[0][0]
if sim >= self.threshold:
print(f'Semantic cache HIT (similarity={sim:.3f}): {stored_query[:40]}...')
return result
return None
def set(self, query, result):
emb = self.embed(query)
self.entries.append((emb, query, result))
sem_cache = SemanticCache(similarity_threshold=0.95)
print('Semantic cache ready. Threshold: 0.95 cosine similarity.')GPTCacheライブラリ
GPTCacheは、複数の埋め込みモデル、類似度バックエンド（FAISS、Redis）、削除戦略をサポートするオープンソースのセマンティックキャッシュライブラリです。OpenAIおよびLangChainのクライアントと直接統合できます。
# pip install gptcache
# GPTCache integration example
# from gptcache import cache
# from gptcache.adapter import openai
# from gptcache.embedding import Onnx
# from gptcache.manager import CacheBase, VectorBase, get_data_manager
# from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation
# Initialize GPTCache
# onnx = Onnx()
# data_manager = get_data_manager(
# CacheBase('sqlite'),
# VectorBase('faiss', dimension=onnx.dimension)
# )
# cache.init(
# embedding_func=onnx.to_embeddings,
# data_manager=data_manager,
# similarity_evaluation=SearchDistanceEvaluation(),
# )
# After init, use openai from gptcache.adapter instead of standard openai
# response = openai.ChatCompletion.create(
# model='gpt-4o-mini',
# messages=[{'role': 'user', 'content': 'What is Python?'}]
# )
# Same API, but cache is checked first
print('GPTCache: drop-in semantic cache for OpenAI API calls.')
print('Supports: FAISS, Redis, SQLite, Milvus as vector backends.')Anthropicのプロンプトキャッシュ（ネイティブ）
Anthropicは、サーバー上でシステムプロンプトの処理をキャッシュするネイティブなプロンプトキャッシュを提供しています。キャッシュがヒットした場合、通常の入力トークン料金の10%だけを支払います。これは、アプリケーションレベルの応答キャッシュとは別のものです。
import anthropic
client = anthropic.Anthropic(api_key='YOUR_API_KEY')
LONG_SYSTEM_PROMPT = '''You are an expert financial analyst with 20 years of experience.
''' + 'Domain knowledge: ' + 'analysis context...' * 500 # large system prompt
# Enable prompt caching with cache_control
response = client.messages.create(
model='claude-opus-4-5',
max_tokens=1024,
system=[
{
'type': 'text',
'text': LONG_SYSTEM_PROMPT,
'cache_control': {'type': 'ephemeral'} # cache this prefix
}
],
messages=[{'role': 'user', 'content': 'Analyze Q3 2024 earnings.'}]
)
print('Cache write tokens:', response.usage.cache_creation_input_tokens)
print('Cache read tokens: ', response.usage.cache_read_input_tokens)
print('Regular input tokens:', response.usage.input_tokens)
# On cache HIT: cache_read_input_tokens shows the cached tokens
# Cost: cached tokens charged at 10% of normal rateキャッシュのTTLと削除戦略
基盤となる知識が変化したりモデルが更新されたりすると、キャッシュされた結果は古くなります。TTL（Time-to-Live）と削除戦略によって、鮮度を管理します。
import time
from collections import OrderedDict
class TTLCache:
def __init__(self, max_size=1000, default_ttl=3600):
self.store = OrderedDict() # key: (value, expire_at)
self.max_size = max_size
self.default_ttl = default_ttl
def set(self, key, value, ttl=None):
ttl = ttl or self.default_ttl
expire_at = time.time() + ttl
if key in self.store:
del self.store[key]
self.store[key] = (value, expire_at)
# LRU eviction: remove oldest if over capacity
if len(self.store) > self.max_size:
self.store.popitem(last=False)
def get(self, key):
if key not in self.store:
return None
value, expire_at = self.store[key]
if time.time() > expire_at:
del self.store[key]
return None # expired
# Move to end (LRU update)
self.store.move_to_end(key)
return value
# TTL strategy guidelines
ttl_guidelines = {
'Static knowledge': 86400, # 24h (facts, definitions)
'Semi-static': 3600, # 1h (product info, FAQs)
'Dynamic content': 300, # 5min (news, prices)
'Personalized': 0 # no cache (user-specific)
}
for k, v in ttl_guidelines.items():
print(f'{k}: {v}s TTL')キャッシュ無効化のパターン
キャッシュ無効化、つまり古くなったデータをいつ削除するかを判断することは、コンピューティングにおける最も難しい問題の1つです。LLMキャッシュでは、これらのパターンによって最も一般的な無効化の要件に対応できます。
class InvalidationAwareCache(TTLCache):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.tags = {} # key: set of tags
self.tag_index = {} # tag: set of keys
def set_with_tags(self, key, value, tags, ttl=None):
self.set(key, value, ttl)
self.tags[key] = set(tags)
for tag in tags:
self.tag_index.setdefault(tag, set()).add(key)
def invalidate_by_tag(self, tag):
keys_to_delete = self.tag_index.pop(tag, set())
for key in keys_to_delete:
self.store.pop(key, None)
self.tags.pop(key, None)
print(f'Invalidated {len(keys_to_delete)} entries with tag={tag}')
# Usage: tag cache entries by data source
cache = InvalidationAwareCache()
cache.set_with_tags('product_faq_123', 'Product FAQs...', tags=['product:123', 'faqs'])
cache.set_with_tags('product_spec_123', 'Spec sheet...', tags=['product:123', 'specs'])
# When product 123 is updated, invalidate all its cache entries
cache.invalidate_by_tag('product:123') # Invalidated 2 entriesキャッシュ性能の測定
キャッシュがコストとレイテンシーに与える影響を把握するため、キャッシュ性能の指標を追跡します。適切に調整されたキャッシュでは、ほとんどの本番利用ケースで50%を超えるヒット率を達成できるはずです。
class CacheMetrics:
def __init__(self):
self.hits = 0
self.misses = 0
self.total_latency_saved_ms = 0
self.total_cost_saved_usd = 0
self.avg_api_latency_ms = 1500 # typical LLM call latency
self.avg_api_cost_usd = 0.002 # typical cost per call
def record_hit(self):
self.hits += 1
self.total_latency_saved_ms += self.avg_api_latency_ms
self.total_cost_saved_usd += self.avg_api_cost_usd
def record_miss(self):
self.misses += 1
def report(self):
total = self.hits + self.misses
hit_rate = self.hits / total if total else 0
return {
'hit_rate': f'{hit_rate:.1%}',
'total_requests': total,
'cache_hits': self.hits,
'latency_saved_sec': round(self.total_latency_saved_ms / 1000, 1),
'cost_saved_usd': round(self.total_cost_saved_usd, 2)
}
metrics = CacheMetrics()
for i in range(100):
if i % 3 == 0: # simulate 33% hit rate
metrics.record_hit()
else:
metrics.record_miss()
print(metrics.report())本番環境向けRedisバックエンドキャッシュ
インメモリキャッシュは再起動すると失われ、サーバーインスタンス間で共有できません。Redisは、本番環境のデプロイで複数のAPIサーバー間にまたがって利用できる、永続的で共有可能なキャッシュを提供します。
import redis
import json
import hashlib
class RedisLLMCache:
def __init__(self, host='localhost', port=6379, db=0, default_ttl=3600):
self.client = redis.Redis(host=host, port=port, db=db,
decode_responses=True)
self.default_ttl = default_ttl
def _key(self, messages, model):
content = json.dumps({'messages': messages, 'model': model},
sort_keys=True)
return 'llmcache:' + hashlib.sha256(content.encode()).hexdigest()
def get(self, messages, model):
key = self._key(messages, model)
value = self.client.get(key)
if value:
self.client.expire(key, self.default_ttl) # refresh TTL on hit
return json.loads(value)
return None
def set(self, messages, model, result, ttl=None):
key = self._key(messages, model)
self.client.setex(key, ttl or self.default_ttl, json.dumps(result))
def stats(self):
keys = self.client.keys('llmcache:*')
return {'cached_entries': len(keys),
'memory_bytes': self.client.memory_usage('llmcache:') or 0}
# Usage: drop-in replacement for in-memory cache
# cache = RedisLLMCache(host='redis.internal', port=6379)
print('RedisLLMCache: shared across all server instances, survives restarts.')キャッシュすべきでない場合
すべてのLLM呼び出しにキャッシュが適しているわけではありません。キャッシュを使用しない場面を理解することで、古い結果や不正確な結果を提供するのを防げます。
DONT_CACHE_WHEN = {
'High temperature': (
'temperature > 0 produces different outputs for the same input. '
'Caching would always return the first generation, defeating the purpose.'
),
'Real-time data required': (
'Queries about current prices, live news, or real-time status '
'must always hit the API and live data source.'
),
'Personalized responses': (
'Responses that depend on user_id, session context, or personal data '
'should not be shared across users.'
),
'Safety-critical': (
'Medical, legal, or financial responses where staleness could cause harm '
'require fresh responses with the most current model version.'
),
'Non-deterministic tools': (
'If the prompt includes a current timestamp or random seed, '
'the response is by design non-repeatable.'
)
}
for condition, reason in DONT_CACHE_WHEN.items():
print(f'Skip cache: {condition}')
print(f' Reason: {reason[:60]}...')
print()理解度チェック
LLMのレスポンスに対する完全一致キャッシュとセマンティックキャッシュの主な違いは何ですか。
キャッシュ戦略のまとめ
効果的なプロンプトキャッシュでは、複数の戦略を組み合わせます：
- 完全一致：ハッシュベースで、ヒット時のオーバーヘッドはゼロですが、表現が異なるとヒット率が低くなります
- セマンティックキャッシュ：埋め込みの類似度によって言い換えの一致を見つけ、より高いヒット率を実現します
- GPTCache：FAISS/Redisバックエンドで両方の戦略を組み合わせるオープンソースライブラリです
- Anthropicネイティブキャッシュ：トークンコスト10%でサーバー側のシステムプロンプトキャッシュを利用できます
- TTL + LRUによる追い出し：時間に基づく鮮度管理と容量管理を組み合わせます
- タグベースの無効化：ソースデータが変更されたときに、関連するエントリを無効化します
- キャッシュすべきでない場合：ゼロでないtemperature、リアルタイムデータ、パーソナライズされた処理、安全性が重要な処理
よくある質問
「プロンプトのキャッシュ戦略」レッスンは無料ですか？
はい。「プロンプトのキャッシュ戦略」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Prompt Engineeringコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Prompt Engineeringコースには全4レッスンが含まれています。
「プロンプトのキャッシュ戦略」で何を学びますか？
意味ベースのキャッシュ、完全一致キャッシュ、Anthropicのプロンプトキャッシュを学びます。 ブラウザで直接実行するハンズオンコードでAI Prompt Engineeringを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Prompt Engineeringを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Prompt Engineeringは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン1/4です。
「プロンプトのキャッシュ戦略」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Prompt Engineeringレッスンでコードを書いて実行できますか？
はい。すべてのAI Prompt Engineeringレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- プロンプトのキャッシュ戦略
- バッチ処理と非同期実行
- モデル間の負荷分散
- プロンプトパイプラインの監視とアラート