Semantic Caching for LLM Applications: Complete Implementation Guide

import redis
import json
import numpy as np
from redis.commands.search.field import VectorField, TextField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query

class RedisSemanticCache:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.r = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
        self.index_name = "semantic_cache"
        self.similarity_threshold = 0.85
        self.create_index()
    
    def create_index(self):
        try:
            self.r.ft(self.index_name).create_index([
                TextField("prompt"),
                TextField("response"),
                VectorField("embedding", 
                    "FLAT", {
                        "TYPE": "FLOAT32",
                        "DIM": 1536,  # OpenAI ada-002 dimension
                        "DISTANCE_METRIC": "COSINE"
                    }
                )
            ])
        except:
            pass  # Index already exists
    
    def add_to_cache(self, prompt, response, embedding):
        key = f"cache:{hash(prompt)}"
        self.r.hset(key, mapping={
            "prompt": prompt,
            "response": response,
            "embedding": embedding.tobytes()
        })
    
    def search_similar(self, embedding, k=5):
        query = (
            Query(f"*=>[KNN {k} @embedding $vec AS similarity]")
            .sort_by("similarity", asc=False)
            .return_fields("prompt", "response", "similarity")
            .dialect(2)
        )
        
        results = self.r.ft(self.index_name).search(
            query, 
            query_params={"vec": embedding.tobytes()}
        )
        
        if results.docs and float(results.docs[0].similarity) > self.similarity_threshold:
            return {
                "prompt": results.docs[0].prompt,
                "response": results.docs[0].response,
                "similarity": float(results.docs[0].similarity)
            }
        return None

# Usage example
cache = RedisSemanticCache()
embedding = get_embedding("What's the weather in NYC?")
result = cache.search_similar(embedding)
if result:
    print(f"Cache hit! Similarity: {result['similarity']}")
    return result['response']

import pinecone
import hashlib
from typing import Optional

class PineconeSemanticCache:
    def __init__(self, api_key: str, environment: str, index_name: str):
        pinecone.init(api_key=api_key, environment=environment)
        self.index = pinecone.Index(index_name)
        self.similarity_threshold = 0.85
        
    def add_to_cache(self, prompt: str, response: str, embedding: list):
        cache_id = hashlib.md5(prompt.encode()).hexdigest()
        self.index.upsert([(
            cache_id,
            embedding,
            {
                "prompt": prompt,
                "response": response,
                "timestamp": time.time()
            }
        )])
    
    def search_similar(self, embedding: list) -> Optional[dict]:
        results = self.index.query(
            vector=embedding,
            top_k=1,
            include_metadata=True
        )
        
        if results.matches and results.matches[0].score > self.similarity_threshold:
            match = results.matches[0]
            return {
                "prompt": match.metadata["prompt"],
                "response": match.metadata["response"],
                "similarity": match.score
            }
        return None
    
    def invalidate_old_entries(self, max_age_hours: int = 24):
        """Remove cache entries older than max_age_hours"""
        cutoff_time = time.time() - (max_age_hours * 3600)
        # Pinecone doesn't support metadata-based deletion directly
        # You'd need to maintain a separate index or scan all vectors
        pass

# Usage
cache = PineconeSemanticCache(
    api_key="your-api-key",
    environment="us-west1-gcp",
    index_name="semantic-cache"
)

# Check cache before calling LLM
embedding = get_embedding(user_query)
cached = cache.search_similar(embedding)
if cached:
    return cached['response']
else:
    # Call LLM and cache result
    llm_response = call_llm(user_query)
    cache.add_to_cache(user_query, llm_response, embedding)
    return llm_response

import asyncpg
import numpy as np
from typing import Optional

class PgVectorCache:
    def __init__(self, connection_string: str):
        self.conn_string = connection_string
        self.similarity_threshold = 0.85
        
    async def init_db(self):
        conn = await asyncpg.connect(self.conn_string)
        await conn.execute('CREATE EXTENSION IF NOT EXISTS vector')
        await conn.execute('''
            CREATE TABLE IF NOT EXISTS semantic_cache (
                id SERIAL PRIMARY KEY,
                prompt TEXT,
                response TEXT,
                embedding vector(1536),
                created_at TIMESTAMP DEFAULT NOW()
            )
        ''')
        await conn.execute('''
            CREATE INDEX IF NOT EXISTS embedding_idx ON semantic_cache 
            USING ivfflat (embedding vector_cosine_ops)
            WITH (lists = 100)
        ''')
        await conn.close()
    
    async def add_to_cache(self, prompt: str, response: str, embedding: np.ndarray):
        conn = await asyncpg.connect(self.conn_string)
        await conn.execute('''
            INSERT INTO semantic_cache (prompt, response, embedding)
            VALUES ($1, $2, $3)
        ''', prompt, response, embedding.tolist())
        await conn.close()
    
    async def search_similar(self, embedding: np.ndarray) -> Optional[dict]:
        conn = await asyncpg.connect(self.conn_string)
        result = await conn.fetchrow('''
            SELECT prompt, response, 
                   1 - (embedding <=> $1::vector) as similarity
            FROM semantic_cache
            WHERE 1 - (embedding <=> $1::vector) > $2
            ORDER BY embedding <=> $1::vector
            LIMIT 1
        ''', embedding.tolist(), self.similarity_threshold)
        await conn.close()
        
        if result:
            return {
                "prompt": result['prompt'],
                "response": result['response'],
                "similarity": result['similarity']
            }
        return None

# Usage with async
import asyncio

async def main():
    cache = PgVectorCache("postgresql://user:pass@localhost/dbname")
    await cache.init_db()
    
    embedding = get_embedding("What's the weather?")
    result = await cache.search_similar(embedding)
    if result:
        print(f"Cache hit! {result['similarity']:.2f}")
        return result['response']

asyncio.run(main())