Back to Details

llm-caching

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

LLM Caching

LLM缓存

Cut LLM costs and latency with exact match, semantic, and provider-side caching layers.
通过精确匹配、语义相似度和服务商端缓存层降低LLM成本与延迟。

When to Use This Skill

何时使用该技能

Use this skill when:
  • The same or similar queries are asked repeatedly (FAQ bots, support tools)
  • LLM API costs are growing and you need immediate savings
  • Serving high request volumes where repeated queries cause bottlenecks
  • Implementing prompt caching for long system prompts (Anthropic/OpenAI)
  • Building offline-capable AI features that need response persistence
在以下场景使用该技能：
  • 相同或相似查询被重复提出（FAQ机器人、支持工具）
  • LLM API成本持续增长，需要立即节省开支
  • 处理高请求量，重复查询导致性能瓶颈
  • 为长系统提示词实现提示词缓存（Anthropic/OpenAI）
  • 构建需要响应持久化的离线AI功能

Caching Layers

缓存层级

Request → Exact Cache → Semantic Cache → Provider Cache → LLM API
             ↓ hit            ↓ hit             ↓ hit
           instant          ~5ms           50-80% cheaper
Request → Exact Cache → Semantic Cache → Provider Cache → LLM API
             ↓ hit            ↓ hit             ↓ hit
           instant          ~5ms           50-80% cheaper

Layer 1: Exact Match Cache (Redis)

层级1：精确匹配缓存（Redis）

python
import hashlib
import json
import redis
from openai import OpenAI

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
client = OpenAI()

def build_cache_key(model: str, messages: list, temperature: float) -> str:
    """Deterministic key from request parameters."""
    payload = json.dumps({
        "model": model,
        "messages": messages,
        "temperature": temperature,
    }, sort_keys=True)
    return f"llm:exact:{hashlib.sha256(payload.encode()).hexdigest()}"

def cached_completion(model: str, messages: list, temperature: float = 0.0,
                      ttl: int = 3600) -> dict:
    key = build_cache_key(model, messages, temperature)

    # Check cache
    if cached := r.get(key):
        return json.loads(cached)

    # Call API
    response = client.chat.completions.create(
        model=model, messages=messages, temperature=temperature
    )
    result = response.model_dump()

    # Cache result (only cache deterministic responses)
    if temperature == 0.0:
        r.setex(key, ttl, json.dumps(result))

    return result
python
import hashlib
import json
import redis
from openai import OpenAI

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
client = OpenAI()

def build_cache_key(model: str, messages: list, temperature: float) -> str:
    """Deterministic key from request parameters."""
    payload = json.dumps({
        "model": model,
        "messages": messages,
        "temperature": temperature,
    }, sort_keys=True)
    return f"llm:exact:{hashlib.sha256(payload.encode()).hexdigest()}"

def cached_completion(model: str, messages: list, temperature: float = 0.0,
                      ttl: int = 3600) -> dict:
    key = build_cache_key(model, messages, temperature)

    # Check cache
    if cached := r.get(key):
        return json.loads(cached)

    # Call API
    response = client.chat.completions.create(
        model=model, messages=messages, temperature=temperature
    )
    result = response.model_dump()

    # Cache result (only cache deterministic responses)
    if temperature == 0.0:
        r.setex(key, ttl, json.dumps(result))

    return result

Layer 2: Semantic Cache (GPTCache)

层级2：语义缓存（GPTCache）

python
from gptcache import cache, Config
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
python
from gptcache import cache, Config
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

Configure GPTCache with Qdrant backend

Configure GPTCache with Qdrant backend

def init_gptcache(cache_obj, llm: str): onnx = Onnx() # local embedding model data_manager = get_data_manager( CacheBase("redis"), # metadata store VectorBase("qdrant", host="localhost", port=6333, collection_name=f"llm-cache-{llm}", dimension=onnx.dimension), ) cache_obj.init( embedding_func=onnx.to_embeddings, data_manager=data_manager, similarity_evaluation=SearchDistanceEvaluation(), config=Config(similarity_threshold=0.80), # 80% similarity = cache hit )
cache.set_openai_key() init_gptcache(cache, "gpt-4o-mini")
def init_gptcache(cache_obj, llm: str): onnx = Onnx() # local embedding model data_manager = get_data_manager( CacheBase("redis"), # metadata store VectorBase("qdrant", host="localhost", port=6333, collection_name=f"llm-cache-{llm}", dimension=onnx.dimension), ) cache_obj.init( embedding_func=onnx.to_embeddings, data_manager=data_manager, similarity_evaluation=SearchDistanceEvaluation(), config=Config(similarity_threshold=0.80), # 80% similarity = cache hit )
cache.set_openai_key() init_gptcache(cache, "gpt-4o-mini")

Now openai calls are automatically cached

Now openai calls are automatically cached

response = openai.ChatCompletion.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "What is machine learning?"}], )
response = openai.ChatCompletion.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "What is machine learning?"}], )

Second call with similar question ("Explain machine learning") → cache hit

Second call with similar question ("Explain machine learning") → cache hit

undefined
undefined

Custom Semantic Cache (Production-Grade)

自定义语义缓存（生产级）

python
from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, Range
import numpy as np
import uuid
import time

embed_model = SentenceTransformer("BAAI/bge-small-en-v1.5")  # fast, 33M params
qdrant = QdrantClient("http://localhost:6333")

CACHE_COLLECTION = "semantic-cache"
SIMILARITY_THRESHOLD = 0.88
CACHE_TTL_SECONDS = 86400  # 24h
python
from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, Range
import numpy as np
import uuid
import time

embed_model = SentenceTransformer("BAAI/bge-small-en-v1.5")  # fast, 33M params
qdrant = QdrantClient("http://localhost:6333")

CACHE_COLLECTION = "semantic-cache"
SIMILARITY_THRESHOLD = 0.88
CACHE_TTL_SECONDS = 86400  # 24h

Create collection once

Create collection once

qdrant.create_collection( collection_name=CACHE_COLLECTION, vectors_config=VectorParams(size=384, distance=Distance.COSINE), on_disk_payload=True, )
def semantic_cache_lookup(query: str, model: str) -> str | None: embedding = embed_model.encode(query).tolist() results = qdrant.query_points( collection_name=CACHE_COLLECTION, query=embedding, query_filter=Filter(must=[ FieldCondition(key="model", match={"value": model}), FieldCondition(key="expires_at", range=Range(gte=time.time())), ]), limit=1, score_threshold=SIMILARITY_THRESHOLD, ) if results.points: return results.points[0].payload["response"] return None
def semantic_cache_store(query: str, response: str, model: str): embedding = embed_model.encode(query).tolist() qdrant.upsert( collection_name=CACHE_COLLECTION, points=[PointStruct( id=str(uuid.uuid4()), vector=embedding, payload={ "query": query, "response": response, "model": model, "created_at": time.time(), "expires_at": time.time() + CACHE_TTL_SECONDS, }, )], )
def smart_llm_call(query: str, model: str = "gpt-4o-mini") -> dict: # 1. Semantic lookup if cached_response := semantic_cache_lookup(query, model): return {"response": cached_response, "source": "semantic_cache", "cost": 0}
# 2. LLM call
response = client.chat.completions.create(
    model=model,
    messages=[{"role": "user", "content": query}],
)
text = response.choices[0].message.content
cost = litellm.completion_cost(response)

# 3. Store in cache
semantic_cache_store(query, text, model)

return {"response": text, "source": "llm_api", "cost": cost}
undefined
qdrant.create_collection( collection_name=CACHE_COLLECTION, vectors_config=VectorParams(size=384, distance=Distance.COSINE), on_disk_payload=True, )
def semantic_cache_lookup(query: str, model: str) -> str | None: embedding = embed_model.encode(query).tolist() results = qdrant.query_points( collection_name=CACHE_COLLECTION, query=embedding, query_filter=Filter(must=[ FieldCondition(key="model", match={"value": model}), FieldCondition(key="expires_at", range=Range(gte=time.time())), ]), limit=1, score_threshold=SIMILARITY_THRESHOLD, ) if results.points: return results.points[0].payload["response"] return None
def semantic_cache_store(query: str, response: str, model: str): embedding = embed_model.encode(query).tolist() qdrant.upsert( collection_name=CACHE_COLLECTION, points=[PointStruct( id=str(uuid.uuid4()), vector=embedding, payload={ "query": query, "response": response, "model": model, "created_at": time.time(), "expires_at": time.time() + CACHE_TTL_SECONDS, }, )], )
def smart_llm_call(query: str, model: str = "gpt-4o-mini") -> dict: # 1. Semantic lookup if cached_response := semantic_cache_lookup(query, model): return {"response": cached_response, "source": "semantic_cache", "cost": 0}
# 2. LLM call
response = client.chat.completions.create(
    model=model,
    messages=[{"role": "user", "content": query}],
)
text = response.choices[0].message.content
cost = litellm.completion_cost(response)

# 3. Store in cache
semantic_cache_store(query, text, model)

return {"response": text, "source": "llm_api", "cost": cost}
undefined

Layer 3: Provider-Side Prompt Caching

层级3：服务商端提示词缓存

python
undefined
python
undefined

Anthropic — cache long system prompts (saves 90% on cached input tokens)

Anthropic — cache long system prompts (saves 90% on cached input tokens)

import anthropic
client = anthropic.Anthropic()
import anthropic
client = anthropic.Anthropic()

Long system prompt — mark for caching

Long system prompt — mark for caching

SYSTEM_PROMPT = open("knowledge-base.txt").read() # e.g., 50k tokens
def call_with_prompt_cache(user_question: str) -> str: response = client.messages.create( model="claude-sonnet-4-6", max_tokens=1024, system=[ {"type": "text", "text": "You are a helpful assistant."}, { "type": "text", "text": SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"}, # cache this block } ], messages=[{"role": "user", "content": user_question}], ) # Log cache efficiency usage = response.usage cache_savings = usage.cache_read_input_tokens * 0.9 # 90% discount on cached print(f"Cache hits: {usage.cache_read_input_tokens} tokens " f"(saved ~${cache_savings * 3.0 / 1_000_000:.4f})") return response.content[0].text
SYSTEM_PROMPT = open("knowledge-base.txt").read() # e.g., 50k tokens
def call_with_prompt_cache(user_question: str) -> str: response = client.messages.create( model="claude-sonnet-4-6", max_tokens=1024, system=[ {"type": "text", "text": "You are a helpful assistant."}, { "type": "text", "text": SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"}, # cache this block } ], messages=[{"role": "user", "content": user_question}], ) # Log cache efficiency usage = response.usage cache_savings = usage.cache_read_input_tokens * 0.9 # 90% discount on cached print(f"Cache hits: {usage.cache_read_input_tokens} tokens " f"(saved ~${cache_savings * 3.0 / 1_000_000:.4f})") return response.content[0].text

OpenAI — automatic for repeated prefixes (≥1,024 tokens)

OpenAI — automatic for repeated prefixes (≥1,024 tokens)

No code change needed; cached tokens appear in usage.prompt_tokens_details

No code change needed; cached tokens appear in usage.prompt_tokens_details

response = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": LONG_SYSTEM_PROMPT}, # auto-cached {"role": "user", "content": user_question}, ] ) cached = response.usage.prompt_tokens_details.cached_tokens print(f"OpenAI cached {cached} tokens")
undefined
response = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": LONG_SYSTEM_PROMPT}, # auto-cached {"role": "user", "content": user_question}, ] ) cached = response.usage.prompt_tokens_details.cached_tokens print(f"OpenAI cached {cached} tokens")
undefined

Cache Warming

缓存预热

python
async def warm_cache(common_queries: list[str], model: str):
    """Pre-populate cache with known frequent queries."""
    import asyncio
    from openai import AsyncOpenAI

    aclient = AsyncOpenAI()

    async def warm_single(query: str):
        if not semantic_cache_lookup(query, model):
            response = await aclient.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": query}],
            )
            text = response.choices[0].message.content
            semantic_cache_store(query, text, model)
            print(f"Warmed: {query[:50]}...")

    await asyncio.gather(*[warm_single(q) for q in common_queries])
python
async def warm_cache(common_queries: list[str], model: str):
    """Pre-populate cache with known frequent queries."""
    import asyncio
    from openai import AsyncOpenAI

    aclient = AsyncOpenAI()

    async def warm_single(query: str):
        if not semantic_cache_lookup(query, model):
            response = await aclient.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": query}],
            )
            text = response.choices[0].message.content
            semantic_cache_store(query, text, model)
            print(f"Warmed: {query[:50]}...")

    await asyncio.gather(*[warm_single(q) for q in common_queries])

Warm on startup

Warm on startup

import asyncio asyncio.run(warm_cache(FREQUENT_QUERIES, "gpt-4o-mini"))
undefined
import asyncio asyncio.run(warm_cache(FREQUENT_QUERIES, "gpt-4o-mini"))
undefined

Cache Metrics

缓存指标

python
from prometheus_client import Counter, Histogram

cache_hits = Counter("llm_cache_hits_total", "Cache hits", ["cache_layer", "model"])
cache_misses = Counter("llm_cache_misses_total", "Cache misses", ["model"])
cache_savings_usd = Counter("llm_cache_savings_usd_total", "USD saved by cache", ["model"])
python
from prometheus_client import Counter, Histogram

cache_hits = Counter("llm_cache_hits_total", "Cache hits", ["cache_layer", "model"])
cache_misses = Counter("llm_cache_misses_total", "Cache misses", ["model"])
cache_savings_usd = Counter("llm_cache_savings_usd_total", "USD saved by cache", ["model"])

Use in your smart_llm_call function

Use in your smart_llm_call function

if source == "semantic_cache": cache_hits.labels(cache_layer="semantic", model=model).inc() cache_savings_usd.labels(model=model).inc(estimated_cost) else: cache_misses.labels(model=model).inc()
undefined
if source == "semantic_cache": cache_hits.labels(cache_layer="semantic", model=model).inc() cache_savings_usd.labels(model=model).inc(estimated_cost) else: cache_misses.labels(model=model).inc()
undefined

Redis Configuration for LLM Caching

Redis LLM缓存配置

bash
undefined
bash
undefined

redis.conf tuning for LLM cache workload

redis.conf tuning for LLM cache workload

maxmemory 8gb maxmemory-policy allkeys-lru # evict least-recently-used when full save "" # disable persistence (cache is ephemeral) appendonly no tcp-keepalive 60
undefined
maxmemory 8gb maxmemory-policy allkeys-lru # evict least-recently-used when full save "" # disable persistence (cache is ephemeral) appendonly no tcp-keepalive 60
undefined

Common Issues

常见问题

IssueCauseFix
Low cache hit rateThreshold too strictLower 
SIMILARITY_THRESHOLD
to 0.82–0.85
Stale cached responsesLong TTLUse topic-specific TTLs; invalidate on data updates
Cache serving wrong answersThreshold too looseRaise threshold or add model-name filtering
Redis OOMNo eviction policySet 
maxmemory
+ 
allkeys-lru
Slow semantic lookupLarge cache collectionAdd payload index on 
model
+ 
expires_at
问题原因解决方法
缓存命中率低阈值设置过严将
SIMILARITY_THRESHOLD
调低至0.82–0.85
缓存响应过期TTL过长针对不同主题设置专属TTL；数据更新时失效缓存
缓存返回错误答案阈值设置过松提高阈值或添加模型名称过滤
Redis内存不足未设置驱逐策略设置
maxmemory
+ 
allkeys-lru
语义查询缓慢缓存集合过大为
model
和
expires_at
添加负载索引

Best Practices

最佳实践

  • Start with exact cache — zero cost, instant wins for identical queries.
  • Semantic threshold of 0.88–0.92 balances hit rate vs. accuracy; tune with your data.
  • Set per-model TTLs: longer for stable knowledge (1 week), shorter for news/events (1 hour).
  • Always filter by model name in semantic cache — different models give different answers.
  • Log cache hit rate as a KPI; target 30%+ for FAQ-style applications.
  • 从精确缓存开始——零成本，对完全相同的查询立竿见影。
  • 语义阈值设置在0.88–0.92之间，平衡命中率与准确性；根据你的数据进行调优。
  • 为不同模型设置专属TTL：稳定知识类设置较长TTL（1周），新闻/事件类设置较短TTL（1小时）。
  • 语义缓存中始终按模型名称过滤——不同模型会给出不同答案。
  • 将缓存命中率作为关键指标记录；FAQ类应用目标命中率30%以上。

Related Skills

相关技能

  • llm-cost-optimization - Full cost strategy
  • llm-gateway - Gateway-level caching
  • vector-database-ops - Qdrant setup
  • agent-observability - Cache metrics dashboards
  • llm-cost-optimization - 完整成本优化策略
  • llm-gateway - 网关级缓存
  • vector-database-ops - Qdrant部署
  • agent-observability - 缓存指标仪表盘