Token Cost Management & Semantic Caching

+-------------------------------------------------------------------------------+
|                            SEMANTIC CACHING FLOW                              |
+-------------------------------------------------------------------------------+
|  User Query: "How do I optimize PySpark join?"                                |
|                                                                               |
|  1. Compute Embedding: Vector [0.14, -0.82, ...]                              |
|  2. Vector Index Query in Redis (Cosine Distance Threshold < 0.08)            |
|                                                                               |
|  Hit: Found past query: "Best way to tune joins in PySpark"                   |
|  -> Return Cached Answer Instantly (5ms latency, $0.00 cost!)                |
+-------------------------------------------------------------------------------+

Exact vs. Semantic Caching

Standard key-value caches (like Memcached or Redis exact match) fail when users rephrase the same question:

• "What is delta lake?" vs. "Can you explain what Delta Lake is?"
• Exact match misses 100% of the time.
• Semantic caching embeds the prompt, queries a vector index of previous prompts, and serves the cached response if cosine similarity is above a threshold (e.g. $> 0.95$).

Implementing Semantic Cache in Redis

import redis
import numpy as np
from openai import OpenAI

client = OpenAI()
r = redis.Redis(host='localhost', port=6379, decode_responses=False)

def get_embedding(text: str) -> list[float]:
    res = client.embeddings.create(input=text, model="text-embedding-3-small")
    return res.data[0].embedding

def query_with_semantic_cache(prompt: str, threshold: float = 0.96) -> str:
    prompt_emb = np.array(get_embedding(prompt), dtype=np.float32)

    # In production, use Redis VSS (Vector Similarity Search)
    # redis_query = ...

    # If cache miss:
    print("Cache miss! Querying LLM...")
    res = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    answer = res.choices[0].message.content
    # Store prompt embedding and answer
    return answer