Ruesch Manny

Posted on Nov 14

How I Slashed LLM Costs by 80% with a Semantic Caching Proxy in Python & FastAPI

#ai #python #fastapi #architecture

⚡ TL;DR — Ready-to-Deploy Automation Blueprints:

🚀 Turnkey Production Package (This Guide): Instant Access on Whop | Download on Gumroad (Use code **EARLYBIRD* for -20%)*

🎁 Free Starter Blueprints (0€): n8n AI Memory & Notion/Airtable RAG Sync | Programmatic SEO Engine (Python)

💎 MR Studio Lifetime All-Access Vault: Get All Current & Future Workflows (€39 One-Time)

If you are shipping AI features in production, you already know the pain of redundant LLM API requests.

Users ask the same questions in slightly different ways. Every time they do, your application reaches out to OpenAI, Anthropic, or Gemini. You burn tokens, incur unnecessary latency, and inch closer to brutal rate limits.

To solve this, many teams default to a naive Redis cache. But an exact-string cache is useless when User A asks, "How do I reset my password?" and User B asks, "How can I reset my password?"

In this guide, I will show you how to architect a Semantic LLM Caching Proxy & Token Tracker using Python 3.10+ and FastAPI. This drop-in microservice intercepts OpenAI-compatible requests, evaluates prompt intent using lightweight local embeddings, serves cached responses in sub-milliseconds, and logs precise token economics to SQLite.

1. The Bottleneck: Why Commercial Solutions Fall Short

Relying on enterprise AI observability platforms (like LangSmith, Helicone, or Portkey) costs thousands of dollars annually just to track tokens and route prompts efficiently.

While these SaaS platforms are feature-rich, they introduce critical bottlenecks:

Network Overhead: Every request requires an external hop to their tracking servers before hitting the LLM provider.
Vendor Lock-in: You are tied to their SDKs and pricing tiers.
Data Privacy: Sending proprietary user prompts through a third-party observability layer introduces compliance risks (SOC2, GDPR).

Building a lightweight, native Python microservice allows you to keep data in-house, achieve sub-millisecond cache retrievals, and eliminate recurring SaaS fees.

2. The Architecture

To keep overhead virtually non-existent, we bypass massive frameworks and build around standard Python libraries and FastAPI.

Here is the architectural flow:

Drop-in Proxy Interface: A FastAPI POST /v1/chat/completions route that perfectly mirrors OpenAI's API schema. Your client apps just change their base_url to point to localhost:8000.
Local Semantic Caching: We use a tiny, locally-hosted embedding model (like all-MiniLM-L6-v2 via sentence-transformers or onnxruntime) to vectorize incoming prompts.
Vector Search: We calculate the cosine similarity between the incoming prompt vector and our historically cached vectors. If similarity > 0.95, we return a cache hit.
Asynchronous Token Tracking: If it's a cache miss, we stream the real LLM response back to the client while a background asyncio task calculates tokens using tiktoken and logs the transaction to a local SQLite database.

3. The Code & Logic

Let's dive into the core engine. We leverage Pydantic to enforce strict data validation on the incoming OpenAI payloads.

The Proxy Schema & Endpoint

First, we map the OpenAI chat completion schema and set up the FastAPI router.

import asyncio
from fastapi import FastAPI, Request, BackgroundTasks
from pydantic import BaseModel, Field
from typing import Any

app = FastAPI(title="Semantic LLM Proxy")

class ChatMessage(BaseModel):
    role: str
    content: str

class ChatCompletionRequest(BaseModel):
    model: str
    messages: list[ChatMessage]
    temperature: float | None = 1.0
    stream: bool | None = False
    # Accept additional kwargs cleanly
    model_config = {"extra": "allow"}

@app.post("/v1/chat/completions")
async def proxy_completions(req: ChatCompletionRequest, background_tasks: BackgroundTasks):
    # 1. Extract the latest user prompt
    user_prompt = next((m.content for m in reversed(req.messages) if m.role == "user"), "")

    # 2. Check Semantic Cache
    cached_response = await get_semantic_cache(user_prompt)
    if cached_response:
        return format_openai_response(cached_response, cached=True)

    # 3. Cache Miss: Forward to OpenAI
    llm_response, usage_data = await forward_to_openai(req)

    # 4. Background tasks: Save to Cache & Log Tokens
    background_tasks.add_task(save_to_cache, user_prompt, llm_response)
    background_tasks.add_task(log_transaction, req.model, usage_data)

    return format_openai_response(llm_response, cached=False)

The Semantic Caching Logic

The magic happens in the embedding evaluation. We generate an embedding for the user prompt and compare it against our SQLite store (using sqlite-vec or just in-memory numpy operations for small datasets).

import numpy as np
from core.embeddings import generate_local_embedding # Hypothetical lightweight wrapper
from core.database import fetch_cache_entries

def cosine_similarity(vec_a: np.ndarray, vec_b: np.ndarray) -> float:
    dot_product = np.dot(vec_a, vec_b)
    norm_a = np.linalg.norm(vec_a)
    norm_b = np.linalg.norm(vec_b)
    return dot_product / (norm_a * norm_b)

async def get_semantic_cache(prompt: str, threshold: float = 0.95) -> str | None:
    # Generate a 384-dimensional vector locally in < 5ms
    prompt_embedding = await generate_local_embedding(prompt)

    # Fetch recent valid caches (in production, use a vector DB or sqlite-vec)
    cached_items = await fetch_cache_entries()

    for item in cached_items:
        score = cosine_similarity(prompt_embedding, item["embedding"])
        if score >= threshold:
            return item["response"]

    return None

Token Tracking Interceptor

We track exact token usage using OpenAI's tiktoken library to calculate costs dynamically before logging them to our SQLite tracking table. Because this runs as a FastAPI BackgroundTask, it never blocks the user's request.

import tiktoken
import sqlite3

def calculate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    # Simplified pricing table logic
    rates = {
        "gpt-4o": {"in": 0.005, "out": 0.015},
        "gpt-3.5-turbo": {"in": 0.0005, "out": 0.0015}
    }
    rate = rates.get(model, rates["gpt-3.5-turbo"])
    return (prompt_tokens / 1000 * rate["in"]) + (completion_tokens / 1000 * rate["out"])

async def log_transaction(model: str, usage: dict[str, int]):
    cost = calculate_cost(model, usage["prompt_tokens"], usage["completion_tokens"])

    # Log to local SQLite in WAL mode
    query = """
        INSERT INTO token_logs (model, prompt_tokens, completion_tokens, cost, timestamp)
        VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
    """
    # execute async sqlite query here...

4. Deployment & Performance

To ensure this microservice handles high concurrency in production, we need to optimize a few elements:

Uvicorn Workers: Run FastAPI with uvicorn main:app --workers 4. Because the proxy is highly I/O bound (waiting for LLM networks), multiple workers drastically improve throughput.
SQLite WAL Mode: Standard SQLite locks on writes. By executing PRAGMA journal_mode=WAL; on database initialization, we allow concurrent reads and writes, making it perfectly viable for high-traffic token logging.
Memory Management: Load the local embedding model globally on startup (using FastAPI lifespan events) to avoid loading the model weights into memory on every request.

5. Conclusion & Turnkey Package

By dropping a native Python caching proxy in front of your LLM calls, you immediately protect your application from rate limits, slash your OpenAI bill, and deliver responses to your users in milliseconds when a cache hits.

You can easily recreate this architecture using the code blocks provided above as your foundation. However, if you want to skip the boilerplate, edge cases, and testing, I have packaged the complete, production-ready source code.

The turnkey package includes the full Python 3.10+ CLI / Microservice, an automated test suite, Dockerfiles, and pre-configured SQLite schema setups.

Grab the complete source code and start saving on API costs today:

Instant Access on Whop: Semantic LLM Caching Proxy & Token Tracker
Direct Download on Gumroad: Get the Source Code (Use promo code **EARLYBIRD* at checkout for 20% off!)*

Stop paying for redundant AI queries and expensive observability platforms. Build native, build fast, and keep your infrastructure under your control.

DEV Community