LLM Output Caching: Reduce AI API Costs by 80% with Smart Caching (2026)

August 14, 2026 · 17 min read

Many AI API requests are repetitive. Users ask the same questions, run the same analyses, and generate similar content. A well-designed caching layer can reduce your AI API costs by 60-80% while improving response times from seconds to milliseconds. This guide covers production caching strategies for Chinese AI models via TokenEase.

Why LLM Caching Works

Analysis of production traffic shows:

• 30-40% of queries are exact duplicates (same prompt, same parameters)
• 20-30% are semantically similar (same intent, different wording)
• 10-15% are repeated within a user session

Combined, 60-85% of requests can potentially be served from cache.

Types of LLM Caching

1. Exact-Match Caching

import hashlib
import json
import redis
from functools import wraps

redis_client = redis.Redis(host="localhost", port=6379, db=0)

def exact_cache(ttl_seconds=3600):
    """Decorator for exact-match LLM response caching"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Create cache key from function name + sorted args
            cache_data = {
                "func": func.__name__,
                "args": args,
                "kwargs": dict(sorted(kwargs.items()))
            }
            cache_key = f"llm:exact:{hashlib.sha256(json.dumps(cache_data, sort_keys=True).encode()).hexdigest()}"
            
            # Check cache
            cached = redis_client.get(cache_key)
            if cached:
                return json.loads(cached)
            
            # Call function
            result = func(*args, **kwargs)
            
            # Cache result
            redis_client.setex(
                cache_key,
                ttl_seconds,
                json.dumps(result)
            )
            
            return result
        return wrapper
    return decorator

# Usage
@exact_cache(ttl_seconds=7200)  # 2 hour cache
def generate_summary(text: str, model: str = "deepseek-v4"):
    """Generate summary with exact-match caching"""
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers={"Authorization": f"Bearer {TOKEN}"},
        json={
            "model": model,
            "messages": [{"role": "user", "content": f"Summarize: {text}"}],
            "temperature": 0.3
        }
    )
    return response.json()

# First call: API request (costs money)
result1 = generate_summary(long_text, model="deepseek-v4")

# Second call with same params: served from cache (free, instant)
result2 = generate_summary(long_text, model="deepseek-v4")

2. Semantic Caching with Embeddings

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class SemanticCache:
    def __init__(self, redis_client, similarity_threshold=0.95):
        self.redis = redis_client
        self.threshold = similarity_threshold
        self.embedding_model = "deepseek-embedding"
    
    def get_embedding(self, text: str) -> list:
        """Get embedding vector for text"""
        response = requests.post(
            f"{BASE_URL}/embeddings",
            headers={"Authorization": f"Bearer {TOKEN}"},
            json={
                "model": self.embedding_model,
                "input": text,
                "encoding_format": "float"
            }
        )
        return response.json()["data"][0]["embedding"]
    
    def find_similar(self, query_embedding: list, top_k: int = 3) -> list:
        """Find semantically similar cached queries"""
        # Get all cached embeddings
        keys = self.redis.keys("llm:semantic:embedding:*")
        if not keys:
            return []
        
        similarities = []
        for key in keys:
            cached_embedding = json.loads(self.redis.get(key))
            similarity = cosine_similarity(
                [query_embedding],
                [cached_embedding]
            )[0][0]
            
            if similarity >= self.threshold:
                similarities.append((key, similarity))
        
        # Sort by similarity and return top matches
        similarities.sort(key=lambda x: x[1], reverse=True)
        return similarities[:top_k]
    
    def get(self, query: str, model: str, temperature: float) -> dict:
        """Try to get semantically similar cached response"""
        query_embedding = self.get_embedding(query)
        
        similar = self.find_similar(query_embedding)
        
        if similar:
            best_match_key = similar[0][0]
            # Extract original query key from embedding key
            query_key = best_match_key.replace("embedding:", "query:")
            cached_response = self.redis.get(query_key)
            
            if cached_response:
                return {
                    "cached": True,
                    "similarity": similar[0][1],
                    "response": json.loads(cached_response)
                }
        
        return {"cached": False}
    
    def set(self, query: str, model: str, temperature: float, response: dict):
        """Cache response with embedding"""
        query_embedding = self.get_embedding(query)
        
        # Create keys
        embedding_key = f"llm:semantic:embedding:{hashlib.sha256(query.encode()).hexdigest()}"
        query_key = f"llm:semantic:query:{hashlib.sha256(query.encode()).hexdigest()}"
        
        # Store embedding and response
        pipe = self.redis.pipeline()
        pipe.setex(embedding_key, 86400, json.dumps(query_embedding))  # 24h TTL
        pipe.setex(query_key, 86400, json.dumps(response))
        pipe.execute()

# Usage
semantic_cache = SemanticCache(redis_client, similarity_threshold=0.92)

def chat_with_semantic_cache(messages, model="deepseek-v4"):
    """Chat with semantic caching"""
    query = messages[-1]["content"]  # Last user message
    
    # Try semantic cache
    cached = semantic_cache.get(query, model, 0.7)
    if cached["cached"]:
        print(f"Cache hit! Similarity: {cached['similarity']:.3f}")
        return cached["response"]
    
    # Call API
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers={"Authorization": f"Bearer {TOKEN}"},
        json={"model": model, "messages": messages, "temperature": 0.7}
    )
    result = response.json()
    
    # Cache response
    semantic_cache.set(query, model, 0.7, result)
    
    return result

3. Session-Based Caching

class SessionCache:
    def __init__(self, redis_client, session_ttl=1800):
        self.redis = redis_client
        self.session_ttl = session_ttl
    
    def get_session_key(self, session_id: str, turn: int) -> str:
        """Generate cache key for session turn"""
        return f"llm:session:{session_id}:turn:{turn}"
    
    def cache_turn(self, session_id: str, turn: int, messages: list, response: dict):
        """Cache a conversation turn"""
        key = self.get_session_key(session_id, turn)
        data = {
            "messages": messages,
            "response": response,
            "timestamp": time.time()
        }
        self.redis.setex(key, self.session_ttl, json.dumps(data))
    
    def get_cached_turn(self, session_id: str, turn: int) -> dict:
        """Retrieve cached turn"""
        key = self.get_session_key(session_id, turn)
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None
    
    def find_repeated_query(self, session_id: str, current_messages: list) -> dict:
        """Check if user is repeating a query in the session"""
        current_query = current_messages[-1]["content"]
        
        # Check last 5 turns
        for turn in range(5):
            cached = self.get_cached_turn(session_id, turn)
            if cached:
                cached_query = cached["messages"][-1]["content"]
                if cached_query == current_query:
                    return cached["response"]
        
        return None

# Usage in chatbot
session_cache = SessionCache(redis_client)

def chat_with_session_cache(session_id: str, messages: list, model="deepseek-v4"):
    """Chat with session-level caching"""
    
    # Check for repeated query
    cached_response = session_cache.find_repeated_query(session_id, messages)
    if cached_response:
        return cached_response
    
    # Call API
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers={"Authorization": f"Bearer {TOKEN}"},
        json={"model": model, "messages": messages}
    )
    result = response.json()
    
    # Cache turn
    turn_number = len(messages) // 2  # Approximate turn count
    session_cache.cache_turn(session_id, turn_number, messages, result)
    
    return result

4. Template Pre-generation

class TemplateCache:
    """Pre-generate responses for common query patterns"""
    
    COMMON_TEMPLATES = [
        {
            "pattern": r"(?i)what.*hours.*open",
            "response": "Our business hours are Monday-Friday 9AM-6PM EST."
        },
        {
            "pattern": r"(?i)how.*contact.*support",
            "response": "You can reach support at support@example.com or call 1-800-XXX-XXXX."
        },
        {
            "pattern": r"(?i)what.*pricing",
            "response": "Please visit our pricing page at https://example.com/pricing for current plans."
        },
        {
            "pattern": r"(?i)refund.*policy",
            "response": "We offer a 30-day money-back guarantee on all plans. Contact support to initiate a refund."
        }
    ]
    
    def match_template(self, query: str) -> str:
        """Check if query matches a known template"""
        for template in self.COMMON_TEMPLATES:
            if re.search(template["pattern"], query):
                return template["response"]
        return None

# Usage
template_cache = TemplateCache()

def chat_with_templates(messages: list):
    query = messages[-1]["content"]
    
    # Check templates first (free, instant)
    template_response = template_cache.match_template(query)
    if template_response:
        return {
            "choices": [{"message": {"role": "assistant", "content": template_response}}],
            "cached": True
        }
    
    # Fall back to API
    return chat_with_api(messages)

5. Multi-Layer Caching Strategy

class MultiLayerCache:
    def __init__(self):
        self.template_cache = TemplateCache()
        self.exact_cache = ExactCache(redis_client)
        self.semantic_cache = SemanticCache(redis_client)
        self.session_cache = SessionCache(redis_client)
    
    def get(self, session_id: str, messages: list, model: str, temperature: float):
        """Try all cache layers in order of speed"""
        query = messages[-1]["content"]
        
        # Layer 1: Template cache (fastest, zero cost)
        template = self.template_cache.match_template(query)
        if template:
            return {"source": "template", "response": template}
        
        # Layer 2: Session cache (fast, session-level)
        session = self.session_cache.find_repeated_query(session_id, messages)
        if session:
            return {"source": "session", "response": session}
        
        # Layer 3: Exact cache (fast, global)
        exact = self.exact_cache.get(query, model, temperature)
        if exact:
            return {"source": "exact", "response": exact}
        
        # Layer 4: Semantic cache (medium speed, global)
        semantic = self.semantic_cache.get(query, model, temperature)
        if semantic.get("cached"):
            return {"source": "semantic", "response": semantic["response"]}
        
        return None
    
    def set(self, session_id: str, messages: list, model: str, temperature: float, response: dict):
        """Store in all applicable cache layers"""
        query = messages[-1]["content"]
        
        # Cache in exact and semantic
        self.exact_cache.set(query, model, temperature, response)
        self.semantic_cache.set(query, model, temperature, response)
        
        # Cache in session
        turn = len(messages) // 2
        self.session_cache.cache_turn(session_id, turn, messages, response)

# Usage
cache = MultiLayerCache()

def smart_chat(session_id: str, messages: list, model="deepseek-v4"):
    """Chat with multi-layer caching"""
    
    # Try cache
    cached = cache.get(session_id, messages, model, 0.7)
    if cached:
        print(f"Cache hit from {cached['source']}!")
        return cached["response"]
    
    # Call API
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers={"Authorization": f"Bearer {TOKEN}"},
        json={"model": model, "messages": messages, "temperature": 0.7}
    )
    result = response.json()
    
    # Store in cache
    cache.set(session_id, messages, model, 0.7, result)
    
    return result

Cache Invalidation Strategies

def invalidate_cache(pattern: str = "llm:*"):
    """Invalidate cache entries matching pattern"""
    keys = redis_client.keys(pattern)
    if keys:
        redis_client.delete(*keys)
    return len(keys)

# Invalidate on model update
invalidate_cache("llm:*:deepseek-v4:*")  # Only DeepSeek-V4 entries

# Emergency full invalidation
# invalidate_cache("llm:*")

Cost Savings Analysis

When NOT to Cache

• Real-time data — Stock prices, weather, news
• User-specific content — Personalized recommendations
• Creative generation — Users expect unique outputs
• High-temperature queries — Randomness makes caching ineffective
• Small context windows — Cache overhead may exceed savings

Production Deployment

# Docker Compose for caching infrastructure
version: '3.8'
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    command: redis-server --maxmemory 2gb --maxmemory-policy allkeys-lru
  
  api:
    build: .
    environment:
      - REDIS_URL=redis://redis:6379
      - CACHE_ENABLED=true
      - SEMANTIC_THRESHOLD=0.92
    depends_on:
      - redis

volumes:
  redis_data:

Next Steps

1. Start with exact-match caching (simplest, immediate 30-40% savings)
2. Add semantic caching for natural language queries
3. Implement session caching for chatbots
4. Create template responses for your most common queries
5. Monitor hit rates and adjust cache TTLs
6. Get your TokenEase API key and start saving

For more optimization strategies, see our cost optimization guide and failover patterns.

Last updated: August 2026. Cache effectiveness varies by application type.