Slashing LLM Bills by 58%: Building an OpenTelemetry Semantic Cache Proxy in FastAPI

Production LLM deployments come with a dirty secret: between 30% and 60% of LLM queries in enterprise SaaS applications are duplicates or slight paraphrases. Support chat queries, recurring agent tasks, automated enrichment jobs, and document Q&A pipelines repeatedly hit upstream inference providers (OpenAI, Anthropic, Mistral) with functionally identical prompts.

The result? Insane monthly inference bills, unpredictable latency spikes, and engineering teams flying completely blind without granular, per-tenant observability.

In this technical breakdown, we'll build a production-grade, drop-in OpenAI-compatible reverse proxy using asynchronous Python (FastAPI), Redis Vector Search for semantic caching, and OpenTelemetry (OTel) for unified span and token metrics.

When engineering leaders realize LLM spending is spiraling out of control, they usually evaluate two paths:

SHA256(prompt)) in Redis yields miserable cache hit rates (under 4%) because even an extra whitespace or punctuation mark invalidates the cache.

Traditional Flow:
Client ──> Commercial Gateway ($$$ per token) ──> OpenAI API ($$$ per run)

Optimized Self-Hosted Flow:
Client ──> FastAPI Proxy ──> Redis Vector Search (Cosine >= 0.92) ──> Cache HIT (0ms upstream cost)
                    │ (Cache MISS)
                    └───> Upstream Provider (OTel Metric Exported)

To solve this sustainably, we need an in-house proxy that acts as an exact drop-in replacement (base_url="http://proxy-host/v1"), verifies semantic similarity via vector embeddings, and tracks granular OTel spans directly into Grafana, Jaeger, or Prometheus.

The proxy pipeline executes in five sequential phases:

Authorization, X-Tenant-ID) and validate against token budget quotas. text-embedding-3-small or an on-premise SentenceTransformer).similarity >= threshold (e.g., 0.92), return cached completions instantly. Let's construct the core engine. Below is the semantic cache router and vector manager implemented in FastAPI and Redis.

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

INDEX_NAME = "idx:semantic_cache"
VECTOR_DIM = 1536  # text-embedding-3-small dimension

async def init_redis_indices(redis_client: Redis):
    try:
        await redis_client.ft(INDEX_NAME).info()
    except Exception:
        schema = (
            TextField("prompt_text"),
            TextField("completion_text"),
            VectorField(
                "prompt_vector",
                "HNSW",
                {
                    "TYPE": "FLOAT32",
                    "DIM": VECTOR_DIM,
                    "DISTANCE_METRIC": "COSINE",
                }
            )
        )
        definition = IndexDefinition(prefix=["cache:prompt:"], index_type=IndexType.HASH)
        await redis_client.ft(INDEX_NAME).create_index(schema, definition=definition)
python
async def check_semantic_cache(redis_client: Redis, query_vector: list[float], threshold: float = 0.92):
    query_bytes = np.array(query_vector, dtype=np.float32).tobytes()

    query = (
        Query("*=>[KNN 1 @prompt_vector $vec AS score]")
        .sort_by("score")
        .return_fields("completion_text", "score")
        .dialect(2)
    )

    results = await redis_client.ft(INDEX_NAME).search(query, query_params={"vec": query_bytes})

    if results.docs:
        doc = results.docs[0]
        distance = float(doc.score)
        similarity = 1.0 - distance

        if similarity >= threshold:
            return doc.completion_text, similarity

    return None, 0.0
python
from fastapi import FastAPI, Request, HTTPException
from opentelemetry import trace, metrics
import httpx

app = FastAPI()
tracer = trace.get_tracer("llm-proxy")
meter = metrics.get_meter("llm-proxy")

token_counter = meter.create_counter(
    name="llm_tokens_consumed_total",
    description="Total tokens consumed split by tenant and cache hit status"
)

@app.post("/v1/chat/completions")
async def chat_completions_proxy(request: Request):
    tenant_id = request.headers.get("X-Tenant-ID", "default_tenant")
    payload = await request.json()
    messages = payload.get("messages", [])
    last_prompt = messages[-1]["content"] if messages else ""

    with tracer.start_as_current_span("llm_completion_router") as span:
        span.set_attribute("llm.tenant_id", tenant_id)

        prompt_vec = await compute_embedding(last_prompt)

        cached_response, similarity = await check_semantic_cache(app.state.redis, prompt_vec, threshold=0.92)

        if cached_response:
            span.set_attribute("llm.cache_hit", True)
            span.set_attribute("llm.cosine_similarity", similarity)
            token_counter.add(0, {"tenant": tenant_id, "cache_hit": "true"})

            return {
                "id": "cached-completion",
                "object": "chat.completion",
                "choices": [{
                    "index": 0,
                    "message": {"role": "assistant", "content": cached_response},
                    "finish_reason": "stop"
                }],
                "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
            }

        span.set_attribute("llm.cache_hit", False)
        async with httpx.AsyncClient() as client:
            upstream_resp = await client.post(
                "https://api.openai.com/v1/chat/completions",
                headers={"Authorization": request.headers.get("Authorization")},
                json=payload,
                timeout=30.0
            )

        data = upstream_resp.json()

        usage = data.get("usage", {})
        total_tokens = usage.get("total_tokens", 0)
        token_counter.add(total_tokens, {"tenant": tenant_id, "cache_hit": "false"})

        reply_text = data["choices"][0]["message"]["content"]
        await store_cache(app.state.redis, last_prompt, prompt_vec, reply_text)

        return data

When pushing a semantic proxy to production, three critical real-world edge cases must be handled:

INCRBY operations keyed by tenant and current minute (tenant:{id}:budget:{YYYYMMDDHHmm}). If the tenant exceeds their token ceiling, trip the circuit and return an HTTP 429 Too Many Requests instantly.0.88 similarity threshold, whereas legal or financial extraction tasks require 0.97 or strictly deterministic cache hits. Make the threshold configurable via HTTP headers (X-Semantic-Threshold: 0.95). You now have the architectural blueprint to eliminate redundant API calls, enforce strict per-tenant token guardrails, and export enterprise-grade OpenTelemetry metrics without paying software seat taxes.

You can implement this architecture from scratch using the code snippets above. However, if you want a complete, battle-tested, production-ready solution with Docker Compose stacks, automated migrations, Jaeger/Grafana dashboards, and test fixtures ready for zero-downtime deployment, grab the turnkey package:

EARLYBIRD** for 20% off)* Take control of your inference bill, secure your tenancy boundaries, and give your infrastructure the observability it deserves.