Building a Cost-Aware LLM Proxy for Dynamic Model Routing and Caching

My heart sank when I opened the cloud bill last month. A 350% increase in LLM API usage costs! It was a classic case of "move fast and break things," only in this instance, what we broke was our budget. We’d been rapidly integrating various LLM capabilities into our backend, and in the rush, we defaulted to using the most powerful (and expensive) models for almost every task. From simple text transformations to complex content generation, everything was hitting the same high-tier endpoints. This wasn't just inefficient; it was financially unsustainable.

It became painfully clear: our LLM orchestration needed a serious overhaul. The immediate problem was a lack of fine-grained control over which model handled which request. We needed a system that could dynamically choose the most cost-effective model without compromising on quality or performance for critical tasks. The solution I envisioned, and subsequently built, was a cost-aware LLM proxy.

The Problem: Uncontrolled LLM Sprawl and Budget Bleed

Initially, our application's direct integration with LLM providers was straightforward. A request came in, we identified the task (e.g., summarize, generate, classify), and then called a hardcoded model endpoint. For instance, any content generation task, regardless of its complexity or length, would automatically hit gpt-4-turbo. While this ensured high quality, it was akin to using a sledgehammer to crack a nut for simpler operations.

The core issues contributing to our cost spike were:

Static Model Routing: No intelligence to match task complexity with model capability.
Lack of Cost Visibility: Developers didn't have real-time insight into the cost implications of their LLM calls.
Redundant Calls: No caching mechanism, leading to repeated API calls for identical or highly similar prompts.
Provider Lock-in (Implicit): While we could switch models, our code wasn't designed for easy, dynamic provider or model selection.

This situation was exacerbated by the varying pricing models across different LLMs and providers. Some charge per input token, others per output token, and the rates differ wildly. Without a central point of control and cost awareness, optimizing became a guessing game.

Designing the Cost-Aware LLM Proxy

My goal was to build a middleware service that would sit between our application and the various LLM providers. This proxy needed to intercept all LLM requests, analyze them, apply routing rules, and, where possible, serve responses from a cache. I decided to build it as a lightweight service using Python with FastAPI, primarily because of its asynchronous capabilities and robust ecosystem for API development.

Core Components of the Proxy:

Request Interception and Parsing: All LLM requests are routed through the proxy.
Model Cost Catalog: A centralized, up-to-date repository of LLM pricing data.
Dynamic Routing Engine: Logic to select the optimal LLM based on request characteristics and cost.
Response Caching Layer: Store and retrieve common LLM responses.
Observability & Cost Tracking: Metrics and logging to monitor usage and costs.

1. Model Cost Catalog: The Source of Truth

The first critical piece was knowing the actual cost of each model. This isn't static; prices change, and new models emerge. I created a simple JSON configuration file that maps model names to their input and output token costs (per 1k tokens).


{
  "openai": {
    "gpt-4-turbo": {
      "input_cost_per_1k_tokens": 0.01,
      "output_cost_per_1k_tokens": 0.03
    },
    "gpt-3.5-turbo": {
      "input_cost_per_1k_tokens": 0.0005,
      "output_cost_per_1k_tokens": 0.0015
    }
  },
  "anthropic": {
    "claude-3-haiku": {
      "input_cost_per_1k_tokens": 0.00025,
      "output_cost_per_1k_tokens": 0.00125
    },
    "claude-3-opus": {
      "input_cost_per_1k_tokens": 0.015,
      "output_cost_per_1k_tokens": 0.075
    }
  },
  "google": {
    "gemini-pro": {
      "input_cost_per_1k_tokens": 0.000125,
      "output_cost_per_1k_tokens": 0.000375
    }
  }
}

This catalog is loaded at proxy startup and can be refreshed dynamically if needed. It allows our routing engine to perform real-time cost comparisons.

2. Dynamic Routing Engine: Smart Model Selection

This is where the magic happens. The routing engine takes the incoming prompt, analyzes its characteristics (length, presence of keywords, inferred complexity, requested quality, even user tier), and then consults the cost catalog to make an intelligent decision. My initial set of rules was quite basic:

Prompt Length: Short prompts (e.g., less than 100 tokens) for simple tasks are routed to cheaper models.
Task Type: Specific keywords or API endpoints in the request hint at the task (e.g., 'summarize' vs. 'generate_creative_story').
Required Quality/Creativity: Certain upstream requests explicitly tag a need for "high creativity" or "high accuracy," guiding the selection.
Data Sensitivity: For prompts containing sensitive user data (identified via a simple regex scan for now), we might route to a specific, more secure (and potentially more expensive) model that meets compliance requirements, or even block the request.

Here's a simplified Python snippet illustrating the core routing logic:


import json
import logging

logger = logging.getLogger(__name__)

class LLMRoutingEngine:
    def __init__(self, cost_catalog_path="model_costs.json"):
        with open(cost_catalog_path, 'r') as f:
            self.cost_catalog = json.load(f)
        logger.info("LLM Cost Catalog loaded successfully.")

    def _calculate_cost(self, provider, model_name, input_tokens, output_tokens):
        try:
            model_info = self.cost_catalog[provider][model_name]
            input_cost = (input_tokens / 1000) * model_info["input_cost_per_1k_tokens"]
            output_cost = (output_tokens / 1000) * model_info["output_cost_per_1k_tokens"]
            return input_cost + output_cost
        except KeyError:
            logger.warning(f"Cost info not found for {provider}/{model_name}. Returning infinite cost.")
            return float('inf')

    def get_optimal_model(self, prompt: str, target_quality: str = "medium", estimated_output_tokens: int = 200) -> dict:
        prompt_length = len(prompt.split()) # Simple token estimation
        
        # Define candidate models based on quality tiers and capabilities
        # This would be more sophisticated in a real system, potentially from a DB
        candidates = []
        if target_quality == "high":
            candidates.extend([
                {"provider": "openai", "model": "gpt-4-turbo", "priority": 1},
                {"provider": "anthropic", "model": "claude-3-opus", "priority": 2}
            ])
        elif target_quality == "medium":
            candidates.extend([
                {"provider": "openai", "model": "gpt-3.5-turbo", "priority": 1},
                {"provider": "anthropic", "model": "claude-3-haiku", "priority": 2}
            ])
        else: # "low" or default
             candidates.extend([
                {"provider": "google", "model": "gemini-pro", "priority": 1},
                {"provider": "openai", "model": "gpt-3.5-turbo", "priority": 2} # Fallback
            ])

        # Example: if prompt is very short, prioritize even cheaper models
        if prompt_length < 50:
             candidates.insert(0, {"provider": "google", "model": "gemini-pro", "priority": 0})
        
        best_model = None
        min_cost = float('inf')

        # Evaluate candidates based on estimated cost
        for candidate in candidates:
            # For simplicity, we assume max input tokens for prompt_length
            # In reality, you'd use a proper tokenizer for accurate token count
            current_cost = self._calculate_cost(
                candidate["provider"], 
                candidate["model"], 
                prompt_length, 
                estimated_output_tokens
            )
            
            if current_cost < min_cost:
                min_cost = current_cost
                best_model = candidate

        if not best_model:
            # Fallback to a default, known-good (but potentially more expensive) model
            logger.error("No suitable model found, falling back to gpt-3.5-turbo.")
            return {"provider": "openai", "model": "gpt-3.5-turbo"}
            
        logger.info(f"Selected model: {best_model['provider']}/{best_model['model']} for prompt (len={prompt_length}) with estimated cost: ${min_cost:.4f}")
        return best_model

# Example usage in a FastAPI endpoint:
# @app.post("/llm_proxy/generate")
# async def generate_text(request: LLMRequest):
#     # ... token estimation logic ...
#     selected_model = routing_engine.get_optimal_model(
#         request.prompt, 
#         request.target_quality, 
#         request.max_output_tokens
#     )
#     # Now call the actual LLM provider based on selected_model
#     # ...

This is a simplified version, of course. A production-ready routing engine would involve more sophisticated prompt analysis (e.g., using an embedding model to classify prompt intent), A/B testing different routing strategies, and potentially integrating with external evaluation metrics to ensure quality isn't inadvertently degraded. I've also been exploring how to integrate this dynamic routing with serverless functions for even greater cost efficiency, a topic I touched upon in Optimizing LLM Orchestration Costs with Serverless Functions.

3. Response Caching Layer: Avoiding Redundant Calls

Many LLM prompts, especially for common tasks like data extraction, summarization of boilerplate text, or generating standard responses, are highly repetitive. A cache is an obvious win here. I implemented a Redis cache as a layer before the routing engine even kicks in.

The cache key is a hash of the prompt text, along with any relevant parameters like the requested model (if explicitly specified) or `temperature` settings. This ensures that different requests, even for the same prompt, with different parameters, don't hit the same cache entry erroneously. Cache invalidation is tricky, but for many LLM use cases, a time-to-live (TTL) of a few hours or even a day is perfectly acceptable, as the "truth" for a given prompt doesn't change frequently.


import hashlib
import json
import redis
import os

# Initialize Redis client
# In a real app, use connection pooling and proper error handling
redis_client = redis.Redis(
    host=os.getenv("REDIS_HOST", "localhost"), 
    port=int(os.getenv("REDIS_PORT", 6379)), 
    db=0
)

CACHE_TTL_SECONDS = 3600 # 1 hour

def generate_cache_key(prompt: str, params: dict) -> str:
    # Create a consistent string representation of prompt and params
    # Ensure params are sorted for consistent hashing
    sorted_params = json.dumps(params, sort_keys=True)
    key_string = f"{prompt}::{sorted_params}"
    return hashlib.sha256(key_string.encode('utf-8')).hexdigest()

async def get_cached_response(prompt: str, params: dict):
    cache_key = generate_cache_key(prompt, params)
    cached_data = redis_client.get(cache_key)
    if cached_data:
        logger.info(f"Cache hit for key: {cache_key}")
        return json.loads(cached_data)
    return None

async def set_cached_response(prompt: str, params: dict, response_data: dict):
    cache_key = generate_cache_key(prompt, params)
    redis_client.setex(cache_key, CACHE_TTL_SECONDS, json.dumps(response_data))
    logger.info(f"Cached response for key: {cache_key}")

# Integration into a FastAPI endpoint:
# @app.post("/llm_proxy/generate")
# async def generate_text(request: LLMRequest):
#     params = {"model": request.model, "temperature": request.temperature, ...} # Extract relevant params
#     cached_response = await get_cached_response(request.prompt, params)
#     if cached_response:
#         return cached_response
#     
#     # If no cache hit, proceed with routing and actual LLM call
#     selected_model = routing_engine.get_optimal_model(...)
#     # ... make actual LLM call ...
#     # response_from_llm = await call_llm_provider(selected_model, request.prompt, params)
#     
#     await set_cached_response(request.prompt, params, response_from_llm)
#     return response_from_llm

This caching layer, even with a conservative TTL, has already shown significant reductions in API calls for repetitive tasks, directly translating to cost savings. For more details on optimizing various aspects of LLM infrastructure, you might find my earlier post on LLM Embedding and Vector Search Cost Optimization: A Deep Dive relevant, especially concerning cache strategies for embedding lookups.

4. Observability and Cost Tracking: Knowing Where Your Money Goes

Without clear metrics, optimization is blind. I integrated Prometheus for collecting metrics and Grafana for visualization. Each request passing through the proxy logs:

The original prompt (truncated or hashed for privacy).
The selected LLM model and provider.
Input and output token counts.
The estimated cost of the transaction.
Whether it was a cache hit or miss.
Latency of the LLM call.

This data is invaluable. I can now see in real-time which models are being used most, what the average cost per request is, and the effectiveness of my caching strategy. Here’s a conceptual look at the kind of metrics I'm tracking:

llm_proxy_request_total{model="gpt-3.5-turbo", cache_hit="true"}
llm_proxy_cost_total{model="gpt-4-turbo", provider="openai"}
llm_proxy_token_count_input_total{model="gemini-pro"}
llm_proxy_latency_seconds_bucket

The immediate impact was staggering. Within the first week of deployment, our average cost per LLM call dropped by nearly 60%. The overall monthly LLM spend, projected from the first few days, was on track to be less than a third of the previous month's bill. This wasn't just about saving money; it was about regaining control and making informed decisions.

An important aspect of managing LLM costs is understanding the provider's specific pricing models. For instance, OpenAI's official documentation provides detailed pricing for their various models, which is crucial for maintaining an accurate cost catalog. OpenAI Pricing serves as an authoritative source I frequently reference to keep my `model_costs.json` up to date.

What I Learned / The Challenge

Building this proxy wasn't without its challenges. The biggest one was balancing cost optimization with model quality. A cheaper model isn't always better if it leads to degraded user experience or incorrect outputs. My initial routing rules were too aggressive, sometimes sending complex requests to models that weren't quite up to the task, leading to a slight dip in output quality for some edge cases. I quickly learned that "good enough" is a moving target and requires continuous monitoring and feedback loops.

Another challenge was keeping the cost catalog accurate. LLM providers frequently update their pricing, introduce new models, or deprecate old ones. This requires a robust mechanism for updating the catalog, perhaps even an automated scraper or direct API integration with provider billing systems (a future enhancement idea).

Finally, the latency introduced by the proxy itself needed careful optimization. While Python/FastAPI is performant, adding an extra hop in the request path adds overhead. I focused on asynchronous operations, efficient JSON parsing, and minimizing blocking I/O to keep the proxy's own latency to a minimum.

Looking Forward

The cost-aware LLM proxy is a significant step towards sustainable LLM integration. My next steps involve refining the dynamic routing engine. I want to incorporate more sophisticated prompt analysis using smaller, local embedding models to categorize requests more accurately. I'm also exploring A/B testing different routing strategies to empirically determine the best balance between cost and quality. Furthermore, I plan to integrate the proxy with our internal budget alerts system, providing real-time notifications if LLM spend exceeds predefined thresholds. This isn't just about saving money; it's about building a resilient, adaptable, and financially responsible AI infrastructure.

The journey to truly optimize LLM usage is complex, but with tools like this proxy, I feel much more in control of our resources and our future development.

My jaw hit the floor when I saw last month's LLM bill. A 350% jump! It was a classic case of "move fast and break things," only in this instance, what we broke was our budget. We’d been rapidly integrating various LLM capabilities into our backend, and in the rush, we defaulted to using the most powerful (and expensive) models for almost every task. From simple text transformations to complex content generation, everything was hitting the same high-tier endpoints. This wasn't just inefficient; it was financially unsustainable.

The Problem: Uncontrolled LLM Sprawl and Budget Bleed

The core issues contributing to our cost spike were:

Static Model Routing: No intelligence to match task complexity with model capability.
Lack of Cost Visibility: Developers didn't have real-time insight into the cost implications of their LLM calls.
Redundant Calls: No caching mechanism, leading to repeated API calls for identical or highly similar prompts.
Provider Lock-in (Implicit): While we could switch models, our code wasn't designed for easy, dynamic provider or model selection.

Designing the Cost-Aware LLM Proxy

Core Components of the Proxy:

Request Interception and Parsing: All LLM requests are routed through the proxy.
Model Cost Catalog: A centralized, up-to-date repository of LLM pricing data.
Dynamic Routing Engine: Logic to select the optimal LLM based on request characteristics and cost.
Response Caching Layer: Store and retrieve common LLM responses.
Observability & Cost Tracking: Metrics and logging to monitor usage and costs.

1. Model Cost Catalog: The Source of Truth


{
  "openai": {
    "gpt-4-turbo": {
      "input_cost_per_1k_tokens": 0.01,
      "output_cost_per_1k_tokens": 0.03
    },
    "gpt-3.5-turbo": {
      "input_cost_per_1k_tokens": 0.0005,
      "output_cost_per_1k_tokens": 0.0015
    }
  },
  "anthropic": {
    "claude-3-haiku": {
      "input_cost_per_1k_tokens": 0.00025,
      "output_cost_per_1k_tokens": 0.00125
    },
    "claude-3-opus": {
      "input_cost_per_1k_tokens": 0.015,
      "output_cost_per_1k_tokens": 0.075
    }
  },
  "google": {
    "gemini-pro": {
      "input_cost_per_1k_tokens": 0.000125,
      "output_cost_per_1k_tokens": 0.000375
    }
  }
}

This catalog is loaded at proxy startup and can be refreshed dynamically if needed. It allows our routing engine to perform real-time cost comparisons.

2. Dynamic Routing Engine: Smart Model Selection

Prompt Length: Short prompts (e.g., less than 100 tokens) for simple tasks are routed to cheaper models.
Task Type: Specific keywords or API endpoints in the request hint at the task (e.g., 'summarize' vs. 'generate_creative_story').
Required Quality/Creativity: Certain upstream requests explicitly tag a need for "high creativity" or "high accuracy," guiding the selection.
Data Sensitivity: For prompts containing sensitive user data (identified via a simple regex scan for now), we might route to a specific, more secure (and potentially more expensive) model that meets compliance requirements, or even block the request.

Here's a simplified Python snippet illustrating the core routing logic:


import json
import logging

logger = logging.getLogger(__name__)

class LLMRoutingEngine:
    def __init__(self, cost_catalog_path="model_costs.json"):
        with open(cost_catalog_path, 'r') as f:
            self.cost_catalog = json.load(f)
        logger.info("LLM Cost Catalog loaded successfully.")

    def _calculate_cost(self, provider, model_name, input_tokens, output_tokens):
        try:
            model_info = self.cost_catalog[provider][model_name]
            input_cost = (input_tokens / 1000) * model_info["input_cost_per_1k_tokens"]
            output_cost = (output_tokens / 1000) * model_info["output_cost_per_1k_tokens"]
            return input_cost + output_cost
        except KeyError:
            logger.warning(f"Cost info not found for {provider}/{model_name}. Returning infinite cost.")
            return float('inf')

    def get_optimal_model(self, prompt: str, target_quality: str = "medium", estimated_output_tokens: int = 200) -> dict:
        prompt_length = len(prompt.split()) # Simple token estimation
        
        # Define candidate models based on quality tiers and capabilities
        # This would be more sophisticated in a real system, potentially from a DB
        candidates = []
        if target_quality == "high":
            candidates.extend([
                {"provider": "openai", "model": "gpt-4-turbo", "priority": 1},
                {"provider": "anthropic", "model": "claude-3-opus", "priority": 2}
            ])
        elif target_quality == "medium":
            candidates.extend([
                {"provider": "openai", "model": "gpt-3.5-turbo", "priority": 1},
                {"provider": "anthropic", "model": "claude-3-haiku", "priority": 2}
            ])
        else: # "low" or default
             candidates.extend([
                {"provider": "google", "model": "gemini-pro", "priority": 1},
                {"provider": "openai", "model": "gpt-3.5-turbo", "priority": 2} # Fallback
            ])

        # Example: if prompt is very short, prioritize even cheaper models
        if prompt_length < 50:
             candidates.insert(0, {"provider": "google", "model": "gemini-pro", "priority": 0})
        
        best_model = None
        min_cost = float('inf')

        # Evaluate candidates based on estimated cost
        for candidate in candidates:
            # For simplicity, we assume max input tokens for prompt_length
            # In reality, you'd use a proper tokenizer for accurate token count
            current_cost = self._calculate_cost(
                candidate["provider"], 
                candidate["model"], 
                prompt_length, 
                estimated_output_tokens
            )
            
            if current_cost < min_cost:
                min_cost = current_cost
                best_model = candidate

        if not best_model:
            # Fallback to a default, known-good (but potentially more expensive) model
            logger.error("No suitable model found, falling back to gpt-3.5-turbo.")
            return {"provider": "openai", "model": "gpt-3.5-turbo"}
            
        logger.info(f"Selected model: {best_model['provider']}/{best_model['model']} for prompt (len={prompt_length}) with estimated cost: ${min_cost:.4f}")
        return best_model

# Example usage in a FastAPI endpoint:
# @app.post("/llm_proxy/generate")
# async def generate_text(request: LLMRequest):
#     # ... token estimation logic ...
#     selected_model = routing_engine.get_optimal_model(
#         request.prompt, 
#         request.target_quality, 
#         request.max_output_tokens
#     )
#     # Now call the actual LLM provider based on selected_model
#     # ...

3. Response Caching Layer: Avoiding Redundant Calls


import hashlib
import json
import redis
import os

# Initialize Redis client
# In a real app, use connection pooling and proper error handling
redis_client = redis.Redis(
    host=os.getenv("REDIS_HOST", "localhost"), 
    port=int(os.getenv("REDIS_PORT", 6379)), 
    db=0
)

CACHE_TTL_SECONDS = 3600 # 1 hour

def generate_cache_key(prompt: str, params: dict) -> str:
    # Create a consistent string representation of prompt and params
    # Ensure params are sorted for consistent hashing
    sorted_params = json.dumps(params, sort_keys=True)
    key_string = f"{prompt}::{sorted_params}"
    return hashlib.sha256(key_string.encode('utf-8')).hexdigest()

async def get_cached_response(prompt: str, params: dict):
    cache_key = generate_cache_key(prompt, params)
    cached_data = redis_client.get(cache_key)
    if cached_data:
        logger.info(f"Cache hit for key: {cache_key}")
        return json.loads(cached_data)
    return None

async def set_cached_response(prompt: str, params: dict, response_data: dict):
    cache_key = generate_cache_key(prompt, params)
    redis_client.setex(cache_key, CACHE_TTL_SECONDS, json.dumps(response_data))
    logger.info(f"Cached response for key: {cache_key}")

# Integration into a FastAPI endpoint:
# @app.post("/llm_proxy/generate")
# async def generate_text(request: LLMRequest):
#     params = {"model": request.model, "temperature": request.temperature, ...} # Extract relevant params
#     cached_response = await get_cached_response(request.prompt, params)
#     if cached_response:
#         return cached_response
#     
#     # If no cache hit, proceed with routing and actual LLM call
#     selected_model = routing_engine.get_optimal_model(...)
#     # ... make actual LLM call ...
#     # response_from_llm = await call_llm_provider(selected_model, request.prompt, params)
#     
#     await set_cached_response(request.prompt, params, response_from_llm)
#     return response_from_llm

4. Observability and Cost Tracking: Knowing Where Your Money Goes

Without clear metrics, optimization is blind. I integrated Prometheus for collecting metrics and Grafana for visualization. Each request passing through the proxy logs:

The original prompt (truncated or hashed for privacy).
The selected LLM model and provider.
Input and output token counts.
The estimated cost of the transaction.
Whether it was a cache hit or miss.
Latency of the LLM call.

llm_proxy_request_total{model="gpt-3.5-turbo", cache_hit="true"}
llm_proxy_cost_total{model="gpt-4-turbo", provider="openai"}
llm_proxy_token_count_input_total{model="gemini-pro"}
llm_proxy_latency_seconds_bucket

An important aspect of managing LLM costs is understanding the provider's specific pricing models. For instance, OpenAI's official documentation provides detailed pricing for their various models, which is crucial for maintaining an accurate cost catalog. Similarly, Anthropic (Claude) and Google (Gemini) provide detailed pricing information, which I continuously monitor and integrate into my cost catalog. OpenAI Pricing serves as an authoritative source I frequently reference to keep my model_costs.json up to date.

What I Learned / The Challenge

Looking Forward

The journey to truly optimize LLM usage is complex, but with tools like this proxy, I feel much more in control of our resources and our future development.

Search This Blog

TechFrontier | AI Engineering, Go & Cloud Cost Optimization