FinOps for AI Workloads: Managing LLM Costs in Production

Running large language models in production breaks the cost model most cloud teams already trust. Compute-hour billing is predictable, and token billing is not. The same feature can cost a fraction of a cent or several dollars per call, depending on how much context the prompt carries, which model answers, and how many tool calls the agent makes before it stops.

The default worth starting from is prompt caching plus a hard output limit. Both are provider features, both take an afternoon, and neither changes what the model produces. Model routing and semantic caching come after that, because they trade answer quality for savings and need an evaluation set before they can be trusted in production. Token budgets and cost metrics are the safety net under all four.

The Token-Based Billing Challenge

A single poorly designed prompt can consume more tokens than thousands of optimized requests. The examples below price everything with gpt-4-turbo rates ($10 per 1M input tokens, $30 per 1M output tokens) so the comparisons stay on one scale.

Cost Variability Example

# Simple query: "What's the weather?"
# Input: 50 tokens (user message + system prompt)
# Output: 30 tokens
# Cost: (50 * $10/1M) + (30 * $30/1M) = $0.0014

# Complex RAG query: "Analyze Q4 sales trends and recommend strategies"
# Input: 8,050 tokens (50 user + 6,000 system + 2,000 RAG context)
# Output: 500 tokens
# Cost: (8,050 * $10/1M) + (500 * $30/1M) = $0.0955 (68x more expensive)

# Tool-call storm: agent invokes 20 tools in one turn
# Input across tool calls: 8,050 * 20 = 161,000 tokens
# Output across tool calls: 500 * 20 = 10,000 tokens
# Cost: (161,000 * $10/1M) + (10,000 * $30/1M) = $1.91 (1,364x more expensive)

Nothing in the architecture has to change for the bill to grow by an order of magnitude. The same feature, sent more often and carrying a longer context, is enough. That is why the gap widens exactly when an application moves from proof of concept to production.

Provider Pricing Models

Different providers offer distinct pricing structures that significantly impact total cost of ownership.

AWS Bedrock Pricing Tiers

Standard (On-Demand): Token-based billing with no commitments

• Claude Sonnet 4.6: $3 input, $15 output per 1M tokens
• Most flexible option, highest per-token cost

Batch Inference: 50% discount for asynchronous workloads

• Ideal for overnight reports, bulk document analysis
• Non-real-time processing acceptable

Provisioned Throughput: Time-based pricing for high-volume scenarios

• Reserved capacity with predictable costs
• Example: Claude Haiku 4.5 with Provisioned Throughput (6-month commitment)

OpenAI Pricing Structure

PRICING = {
    'gpt-4-turbo': {
        'input': 10.00 / 1_000_000,
        'output': 30.00 / 1_000_000
    },
    'gpt-4o': {
        'input': 2.50 / 1_000_000,
        'output': 10.00 / 1_000_000
    },
    'gpt-4o-mini': {
        'input': 0.15 / 1_000_000,
        'output': 0.60 / 1_000_000
    }
}

# Key insight: Output tokens cost 2-5x more than input tokens
# GPT-4: Output is 3x more expensive ($30 vs $10 per 1M)
# GPT-4o: Output is 4x more expensive ($10 vs $2.50 per 1M)

Anthropic Direct Pricing

• Claude Opus 4.1: $15 input, $75 output per 1M tokens
• Claude Opus 4.5: $5 input, $25 output per 1M tokens (newer, more cost-effective)
• Claude Sonnet 4.6: $3 input, $15 output per 1M tokens
• Claude Haiku 3: $0.25 input, $1.25 output per 1M tokens
• Claude Haiku 4.5: $1 input, $5 output per 1M tokens (newer generation)
• Prompt Caching: 90% discount on cached tokens with 85% latency reduction
• Cache Write Premium: 25% premium on cache writes (one-time cost for caching content)

Optimization Strategy 1: Prompt Caching

Prompt caching provides the highest cost reduction with minimal implementation effort. By marking static prompt components as cacheable, subsequent requests within the cache TTL period receive a 90% discount on those tokens.

Implementation with AWS Bedrock

import boto3
import json

bedrock_runtime = boto3.client('bedrock-runtime')

# Large system prompt with company policies (10,000 tokens)
SYSTEM_PROMPT = """You are a customer support agent for Acme Corp.

Company Policies:
[... 8,000 tokens of policies, procedures, FAQs ...]

Communication Style:
- Professional but friendly
- Concise responses (max 200 words)
- Always include relevant policy references

Tool Usage Guidelines:
[... 2,000 tokens of tool documentation ...]
"""

def invoke_with_caching(user_message: str):
    response = bedrock_runtime.converse(
        modelId="anthropic.claude-sonnet-4-5-20250929-v1:0",
        messages=[
            {
                "role": "user",
                "content": [{"text": user_message}]
            }
        ],
        system=[
            {
                "text": SYSTEM_PROMPT,
                # Default TTL is 5 minutes. Sonnet 4.5 needs at least 4,096
                # tokens before a checkpoint caches anything at all.
                "cachePoint": {"type": "default"}
            }
        ]
    )

    # Cost analysis
    usage = response['usage']

    # First call: 50 uncached input tokens, 10,000 written to cache
    # (writes are billed at 1.25x the base input rate)
    # Later calls inside the TTL: 50 uncached, 10,000 read at the
    # discounted rate

    # inputTokens counts only uncached tokens, so total input is
    # inputTokens + cacheReadInputTokens + cacheWriteInputTokens
    print(f"Uncached input tokens: {usage.get('inputTokens', 0)}")
    print(f"Cache read tokens: {usage.get('cacheReadInputTokens', 0)}")
    print(f"Cache write tokens: {usage.get('cacheWriteInputTokens', 0)}")
    print(f"Output tokens: {usage.get('outputTokens', 0)}")

    return response

Cost Impact Analysis

# 100 requests against the same 10,000-token system prompt, 200 tokens
# out each, priced at $3/1M input and $15/1M output

# Without caching:
# (10,050 * 100 * $3/1M) + (200 * 100 * $15/1M) = $3.32

# With caching, 1 write and 99 reads inside the TTL window:
# Write request: (50 * $3/1M) + (10,000 * $3.75/1M) + (200 * $15/1M) = $0.0407
# Read requests: (50 * $3/1M) + (10,000 * $0.30/1M) + (200 * $15/1M) = $0.0062
# Total: $0.0407 + (99 * $0.0062) = $0.65
# Reduction: 80%

# The write premium is why sparse traffic can cost more with caching on:
# a cache that expires before it is read is a pure 1.25x surcharge.

Implementation Best Practices

Structure Prompts for Caching:

• Place static content (policies, instructions) first
• Dynamic context (user data, timestamps) goes in user messages
• Avoid changing cached sections unnecessarily

Common Pitfalls:

• Dynamic Timestamps: Adding current_time to system prompt invalidates cache every request
• Intermittent Traffic: The TTL resets on every hit, so steady traffic keeps the cache warm and an idle gap lets it expire
• Prompt Versioning: Deploy prompt changes during low-traffic periods

# BAD: Dynamic content invalidates cache
system_prompt = f"""
You are a support agent.
Current time: {datetime.now().isoformat()}  # Changes every request!
[... rest of prompt ...]
"""

# GOOD: Static prompt, dynamic context in user message
system_prompt = """
You are a support agent.
[... static policies and instructions ...]
"""

user_message = f"""
Current time: {datetime.now().isoformat()}
User question: {question}
"""

Optimization Strategy 2: Intelligent Model Routing

Not all queries require the most powerful (and expensive) model. Routing by complexity moves the simple majority of traffic to a cheaper model, and the saving equals the price gap between the two models times the share of traffic you can safely move. That share is a measurement, not a constant, so keep an evaluation set behind the router.

Custom Routing Implementation

import OpenAI from 'openai';

interface ModelRoutingConfig {
  simpleThreshold: number;  // < 0.3 = simple query
  complexThreshold: number;  // > 0.7 = complex query
  models: {
    simple: string;
    medium: string;
    complex: string;
  };
}

interface QueryComplexity {
  score: number;
  factors: {
    wordCount: number;
    questionType: string;
    contextRequired: boolean;
    multiStepReasoning: boolean;
  };
}

class IntelligentRouter {
  private openai: OpenAI;
  private config: ModelRoutingConfig;

  constructor() {
    this.openai = new OpenAI();
    this.config = {
      simpleThreshold: 0.3,
      complexThreshold: 0.7,
      models: {
        simple: 'gpt-4o-mini',  // $0.15 input, $0.60 output per 1M
        medium: 'gpt-4o',  // $2.50 input, $10.00 output per 1M
        complex: 'gpt-4-turbo'  // $10.00 input, $30.00 output per 1M
      }
    };
  }

  /**
   * Analyze query complexity using heuristics
   * Production systems might use a lightweight classifier model
   */
  analyzeComplexity(query: string): QueryComplexity {
    const words = query.split(/\s+/);
    const wordCount = words.length;

    // Detect question type
    const questionType = this.detectQuestionType(query);

    // Check for multi-step reasoning indicators
    const multiStepKeywords = ['compare', 'analyze', 'design', 'implement',
                                'evaluate', 'recommend', 'strategize'];
    const multiStepReasoning = multiStepKeywords.some(kw =>
      query.toLowerCase().includes(kw)
    );

    // Context required (references to previous conversation, documents)
    const contextRequired = query.toLowerCase().includes('previous') ||
                           query.toLowerCase().includes('earlier') ||
                           query.toLowerCase().includes('mentioned');

    // Calculate complexity score (0.0 - 1.0)
    let score = 0.0;

    // Word count factor (longer = potentially more complex)
    if (wordCount < 10) score += 0.1;
    else if (wordCount < 30) score += 0.3;
    else score += 0.5;

    // Question type factor
    if (questionType === 'factual') score += 0.1;
    else if (questionType === 'analytical') score += 0.5;
    else score += 0.3;

    // Multi-step reasoning adds significant complexity
    if (multiStepReasoning) score += 0.3;

    // Context requirement adds complexity
    if (contextRequired) score += 0.2;

    // Normalize to 0.0 - 1.0 range
    score = Math.min(1.0, score);

    return {
      score,
      factors: {
        wordCount,
        questionType,
        contextRequired,
        multiStepReasoning
      }
    };
  }

  private detectQuestionType(query: string): string {
    const lower = query.toLowerCase();

    // Factual questions
    if (lower.match(/^(what|when|where|who) is/)) return 'factual';

    // Analytical questions
    if (lower.match(/(how|why|explain|compare|analyze)/)) return 'analytical';

    // Procedural questions
    if (lower.match(/(how to|steps|process|implement)/)) return 'procedural';

    return 'general';
  }

  selectModel(complexity: QueryComplexity): string {
    if (complexity.score < this.config.simpleThreshold) {
      return this.config.models.simple;
    } else if (complexity.score < this.config.complexThreshold) {
      return this.config.models.medium;
    } else {
      return this.config.models.complex;
    }
  }

  async invoke(query: string, systemPrompt: string) {
    const complexity = this.analyzeComplexity(query);
    const model = this.selectModel(complexity);

    console.log(`Query complexity: ${complexity.score.toFixed(2)} -> ${model}`);

    const response = await this.openai.chat.completions.create({
      model,
      messages: [
        { role: 'system', content: systemPrompt },
        { role: 'user', content: query }
      ],
      temperature: 0.7
    });

    return {
      response: response.choices[0].message.content,
      model,
      complexity: complexity.score,
      usage: response.usage
    };
  }
}

// Usage example
const router = new IntelligentRouter();

// Simple query -> gpt-4o-mini
await router.invoke(
  "What's your return policy?",
  "You are a customer support agent"
);

// Complex query -> gpt-4-turbo
await router.invoke(
  "Compare our enterprise and business plans, analyze which would be better for a mid-sized company with 500 employees, considering scalability and cost over 3 years",
  "You are a customer support agent"
);

AWS Bedrock Intelligent Prompt Routing

AWS Bedrock offers the same idea as a managed endpoint. A prompt router predicts response quality per request and picks between exactly two models in one family:

import boto3

bedrock = boto3.client('bedrock')
runtime = boto3.client('bedrock-runtime')

# Router ARNs come from ListPromptRouters or the Prompt Routers page
# in the console. A router chooses between two models in one family.
router_arn = bedrock.list_prompt_routers()['promptRouterSummaries'][0]['promptRouterArn']

response = runtime.converse(
    modelId=router_arn,
    messages=[
        {
            "role": "user",
            "content": [{"text": "What's the weather in Seattle?"}]
        }
    ]
)

# The response reports which model actually served the request.
# Two constraints before relying on this in production:
# - Routing is only optimized for English prompts.
# - Savings are the price gap between the two models times the share of
#   traffic that lands on the cheaper one, so measure the split rather
#   than assuming one.

Expected Outcomes

The diagram below assumes a 60/30/10 traffic mix. Change that mix and the whole case for routing changes with it.

Optimization Strategy 3: Token Budget Enforcement

Unbounded token consumption leads to cost storms. Implementing hard limits prevents runaway expenses while maintaining system functionality.

Budget Tracking Implementation

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, Optional
import redis

@dataclass
class TokenBudget:
    max_input_tokens_per_request: int
    max_output_tokens_per_request: int
    max_tokens_per_user_daily: int
    max_tokens_per_team_monthly: int

@dataclass
class BudgetUsage:
    user_id: str
    team_id: str
    tokens_used_today: int
    tokens_used_this_month: int
    last_reset: datetime

class TokenBudgetEnforcer:
    def __init__(self, budget: TokenBudget):
        self.budget = budget
        self.redis_client = redis.Redis(host='localhost', decode_responses=True)

    def check_and_reserve(
        self,
        user_id: str,
        team_id: str,
        estimated_input_tokens: int,
        estimated_output_tokens: int
    ) -> tuple[bool, Optional[str]]:
        """
        Check if request is within budget and reserve tokens.
        Returns (allowed, error_message)
        """

        # Check per-request limits
        if estimated_input_tokens > self.budget.max_input_tokens_per_request:
            return False, f"Input tokens ({estimated_input_tokens}) exceed per-request limit ({self.budget.max_input_tokens_per_request})"

        if estimated_output_tokens > self.budget.max_output_tokens_per_request:
            return False, f"Output tokens ({estimated_output_tokens}) exceed per-request limit ({self.budget.max_output_tokens_per_request})"

        # Check daily user limit
        user_daily_key = f"budget:user:{user_id}:daily"
        user_tokens_today = int(self.redis_client.get(user_daily_key) or 0)

        total_estimated = estimated_input_tokens + estimated_output_tokens

        if user_tokens_today + total_estimated > self.budget.max_tokens_per_user_daily:
            return False, f"User daily limit exceeded ({user_tokens_today}/{self.budget.max_tokens_per_user_daily})"

        # Check monthly team limit
        team_monthly_key = f"budget:team:{team_id}:monthly"
        team_tokens_this_month = int(self.redis_client.get(team_monthly_key) or 0)

        if team_tokens_this_month + total_estimated > self.budget.max_tokens_per_team_monthly:
            return False, f"Team monthly limit exceeded ({team_tokens_this_month}/{self.budget.max_tokens_per_team_monthly})"

        # Reserve tokens (optimistic locking)
        pipe = self.redis_client.pipeline()

        # Increment user daily counter (expires at midnight)
        tomorrow = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1)
        seconds_until_midnight = int((tomorrow - datetime.now()).total_seconds())
        pipe.incrby(user_daily_key, total_estimated)
        pipe.expire(user_daily_key, seconds_until_midnight)

        # Increment team monthly counter (expires at month end)
        next_month = (datetime.now().replace(day=1) + timedelta(days=32)).replace(day=1)
        seconds_until_month_end = int((next_month - datetime.now()).total_seconds())
        pipe.incrby(team_monthly_key, total_estimated)
        pipe.expire(team_monthly_key, seconds_until_month_end)

        pipe.execute()

        return True, None

    def record_actual_usage(
        self,
        user_id: str,
        team_id: str,
        actual_input_tokens: int,
        actual_output_tokens: int,
        estimated_input_tokens: int,
        estimated_output_tokens: int
    ):
        """
        Adjust budget based on actual vs estimated usage.
        """
        actual_total = actual_input_tokens + actual_output_tokens
        estimated_total = estimated_input_tokens + estimated_output_tokens
        difference = actual_total - estimated_total

        if difference != 0:
            pipe = self.redis_client.pipeline()
            pipe.incrby(f"budget:user:{user_id}:daily", difference)
            pipe.incrby(f"budget:team:{team_id}:monthly", difference)
            pipe.execute()

# Usage in LLM application
budget_enforcer = TokenBudgetEnforcer(
    budget=TokenBudget(
        max_input_tokens_per_request=8000,  # Prevent huge contexts
        max_output_tokens_per_request=2000,  # Limit response length
        max_tokens_per_user_daily=100_000,  # ~$1/day per user at $10/1M input
        max_tokens_per_team_monthly=10_000_000  # ~$100/month per team, same rate
    )
)

def invoke_llm_with_budget(user_id: str, team_id: str, prompt: str):
    # Estimate tokens (rough approximation)
    estimated_input = len(prompt.split()) * 1.3  # Account for tokenization
    estimated_output = 500  # Conservative estimate

    # Check budget
    allowed, error = budget_enforcer.check_and_reserve(
        user_id, team_id, int(estimated_input), estimated_output
    )

    if not allowed:
        raise BudgetExceededError(error)

    # Invoke LLM
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=budget_enforcer.budget.max_output_tokens_per_request
    )

    # Record actual usage
    budget_enforcer.record_actual_usage(
        user_id,
        team_id,
        response.usage.prompt_tokens,
        response.usage.completion_tokens,
        int(estimated_input),
        estimated_output
    )

    return response.choices[0].message.content

Alert Configuration

def check_budget_alerts(user_id: str, team_id: str, redis_client, budget):
    """
    Trigger alerts at 70%, 90%, 100% budget thresholds
    """
    user_daily_key = f"budget:user:{user_id}:daily"
    user_tokens_today = int(redis_client.get(user_daily_key) or 0)

    daily_limit = budget.max_tokens_per_user_daily
    usage_percentage = (user_tokens_today / daily_limit) * 100

    if usage_percentage >= 100:
        send_alert(
            level="CRITICAL",
            message=f"User {user_id} exceeded daily budget",
            action="BLOCK"
        )
    elif usage_percentage >= 90:
        send_alert(
            level="WARNING",
            message=f"User {user_id} at 90% of daily budget",
            action="NOTIFY"
        )
    elif usage_percentage >= 70:
        send_alert(
            level="INFO",
            message=f"User {user_id} at 70% of daily budget",
            action="MONITOR"
        )

Common Budget Pitfalls

Tool-Call Storms: Agents invoke 50+ tools without limits, consuming millions of tokens

# Solution: Set max_tool_calls_per_turn
agent = Agent(
    tools=[get_product_details, get_reviews, get_pricing],
    max_tool_calls_per_turn=5,  # Hard limit
    instructions="Use batch queries when possible."
)

RAG Over-Retrieval: Retrieving 50 chunks when 5 would suffice

# BAD: Too many chunks
retriever = VectorStoreRetriever(
    vector_store=vector_db,
    search_kwargs={"k": 50}  # 25,000 tokens of context
)

# GOOD: Focused retrieval
retriever = VectorStoreRetriever(
    vector_store=vector_db,
    search_kwargs={"k": 5}  # 2,500 tokens (90% reduction)
)

Optimization Strategy 4: Semantic Caching

Traditional caching only matches exact queries. Semantic caching uses vector similarity so that questions with the same meaning share one answer, which raises the hit rate on repetitive traffic. It also introduces the risk that two questions look alike and are not, so the similarity threshold is as much a product decision as a tuning knob.

Implementation with Vector Similarity

import hashlib
import json
from typing import Optional
import redis
from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticCache:
    def __init__(
        self,
        redis_client: redis.Redis,
        similarity_threshold: float = 0.95,
        ttl_seconds: int = 3600
    ):
        self.redis = redis_client
        self.similarity_threshold = similarity_threshold
        self.ttl_seconds = ttl_seconds

        # Lightweight embedding model for semantic matching
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')

    def _get_embedding(self, text: str) -> np.ndarray:
        """Generate embedding vector for query"""
        return self.embedding_model.encode(text, normalize_embeddings=True)

    def _cosine_similarity(self, vec1: np.ndarray, vec2: np.ndarray) -> float:
        """Calculate cosine similarity between two vectors"""
        return np.dot(vec1, vec2)  # Vectors are already normalized

    def get(self, query: str, system_prompt: str = "") -> Optional[dict]:
        """
        Retrieve cached response if semantically similar query exists
        """
        cache_key_prefix = f"semantic_cache:{hashlib.md5(system_prompt.encode()).hexdigest()}"

        # Get all cached queries for this system prompt
        cached_keys = self.redis.keys(f"{cache_key_prefix}:*")

        if not cached_keys:
            return None

        query_embedding = self._get_embedding(query)

        best_match = None
        best_similarity = 0.0

        # Find most similar cached query
        for key in cached_keys:
            cached_data = self.redis.get(key)
            if not cached_data:
                continue

            cached = json.loads(cached_data)
            cached_embedding = np.array(cached['embedding'])

            similarity = self._cosine_similarity(query_embedding, cached_embedding)

            if similarity > best_similarity:
                best_similarity = similarity
                best_match = cached

        # Return cached response if similarity exceeds threshold
        if best_similarity >= self.similarity_threshold:
            return {
                'response': best_match['response'],
                'similarity': best_similarity,
                'cached': True,
                'original_query': best_match['query']
            }

        return None

    def set(self, query: str, response: str, system_prompt: str = ""):
        """
        Cache query-response pair with semantic embedding
        """
        cache_key_prefix = f"semantic_cache:{hashlib.md5(system_prompt.encode()).hexdigest()}"
        query_hash = hashlib.md5(query.encode()).hexdigest()
        cache_key = f"{cache_key_prefix}:{query_hash}"

        embedding = self._get_embedding(query)

        cache_data = {
            'query': query,
            'response': response,
            'embedding': embedding.tolist(),
            'timestamp': datetime.utcnow().isoformat()
        }

        self.redis.setex(
            cache_key,
            self.ttl_seconds,
            json.dumps(cache_data)
        )

# Usage in production
semantic_cache = SemanticCache(
    redis_client=redis.Redis(host='localhost', decode_responses=False),
    similarity_threshold=0.95,  # 95% similarity required
    ttl_seconds=3600  # Cache for 1 hour
)

def invoke_with_semantic_cache(query: str, system_prompt: str):
    # Check semantic cache first
    cached = semantic_cache.get(query, system_prompt)

    if cached:
        print(f"Cache hit! Similarity: {cached['similarity']:.2%}")
        print(f"Original query: {cached['original_query']}")
        return cached['response']

    # Cache miss - invoke LLM
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": query}
        ]
    )

    result = response.choices[0].message.content

    # Cache for future semantically similar queries
    semantic_cache.set(query, result, system_prompt)

    return result

# Example: Semantically similar queries
# Query 1: "What's your refund policy?"
# Query 2: "How do I get my money back?"
# Query 3: "Can I return items for a refund?"
# All three would match with > 95% similarity and return cached response

Performance Impact

# Hit rate belongs to the traffic. Support queues and FAQ
# endpoints cluster tightly around a few intents;
# open-ended assistants barely cluster at all. Replay a day of real
# queries through the matcher before budgeting for savings.

# Trade-offs:
# - Embedding computation: one extra forward pass on a small local
#   model, negligible next to the LLM call it may avoid
# - Redis memory: 384 float dimensions per entry for MiniLM-L6,
#   roughly 1.5 KB at 4-byte floats
# - Similarity tuning: too low = confidently wrong answers,
#   too high = the cache never hits

Cost Monitoring and Observability

Without real-time visibility into token consumption, cost problems remain hidden until the bill arrives.

CloudWatch Metrics Implementation

import boto3
from datetime import datetime
from dataclasses import dataclass

@dataclass
class CostMetrics:
    timestamp: datetime
    model: str
    input_tokens: int
    output_tokens: int
    cached_tokens: int
    total_cost: float
    user_id: str
    team_id: str
    request_type: str  # 'simple', 'medium', 'complex'

class LLMCostTracker:
    def __init__(self):
        self.cloudwatch = boto3.client('cloudwatch')

        # Provider pricing (updated 2025)
        self.pricing = {
            'gpt-4-turbo': {
                'input': 10.00 / 1_000_000,
                'output': 30.00 / 1_000_000
            },
            'gpt-4o': {
                'input': 2.50 / 1_000_000,
                'output': 10.00 / 1_000_000
            },
            'gpt-4o-mini': {
                'input': 0.15 / 1_000_000,
                'output': 0.60 / 1_000_000
            },
            'claude-sonnet-3.5': {
                'input': 3.00 / 1_000_000,
                'output': 15.00 / 1_000_000,
                'cached_input': 0.30 / 1_000_000  # 90% discount
            }
        }

    def calculate_cost(self, metrics: CostMetrics) -> float:
        """Calculate cost based on token usage and model pricing"""
        pricing = self.pricing.get(metrics.model)
        if not pricing:
            raise ValueError(f"Unknown model: {metrics.model}")

        input_cost = metrics.input_tokens * pricing['input']
        output_cost = metrics.output_tokens * pricing['output']

        # Apply caching discount if applicable
        if metrics.cached_tokens > 0 and 'cached_input' in pricing:
            cached_cost = metrics.cached_tokens * pricing['cached_input']
            # Cached tokens already counted in input_tokens, so adjust
            uncached_tokens = metrics.input_tokens - metrics.cached_tokens
            input_cost = (uncached_tokens * pricing['input']) + cached_cost

        return input_cost + output_cost

    def publish_metrics(self, metrics: CostMetrics):
        """Publish metrics to CloudWatch for dashboard visualization"""

        cost = self.calculate_cost(metrics)

        metric_data = [
            {
                'MetricName': 'TokenUsage',
                'Dimensions': [
                    {'Name': 'Model', 'Value': metrics.model},
                    {'Name': 'TokenType', 'Value': 'Input'}
                ],
                'Value': metrics.input_tokens,
                'Unit': 'Count',
                'Timestamp': metrics.timestamp
            },
            {
                'MetricName': 'TokenUsage',
                'Dimensions': [
                    {'Name': 'Model', 'Value': metrics.model},
                    {'Name': 'TokenType', 'Value': 'Output'}
                ],
                'Value': metrics.output_tokens,
                'Unit': 'Count',
                'Timestamp': metrics.timestamp
            },
            {
                'MetricName': 'LLMCost',
                'Dimensions': [
                    {'Name': 'Model', 'Value': metrics.model},
                    {'Name': 'Team', 'Value': metrics.team_id},
                    {'Name': 'RequestType', 'Value': metrics.request_type}
                ],
                'Value': cost,
                'Unit': 'None',  # Dollars
                'Timestamp': metrics.timestamp
            }
        ]

        # Add cache hit rate metric if caching is used
        if metrics.cached_tokens > 0:
            cache_hit_rate = (metrics.cached_tokens / metrics.input_tokens) * 100
            metric_data.append({
                'MetricName': 'CacheHitRate',
                'Dimensions': [{'Name': 'Model', 'Value': metrics.model}],
                'Value': cache_hit_rate,
                'Unit': 'Percent',
                'Timestamp': metrics.timestamp
            })

        self.cloudwatch.put_metric_data(
            Namespace='LLM/Costs',
            MetricData=metric_data
        )

    def create_cost_anomaly_alarm(self, threshold_dollars: float):
        """Create CloudWatch alarm for cost anomalies"""
        self.cloudwatch.put_metric_alarm(
            AlarmName='LLM-Daily-Cost-Anomaly',
            ComparisonOperator='GreaterThanThreshold',
            EvaluationPeriods=1,
            MetricName='LLMCost',
            Namespace='LLM/Costs',
            Period=86400,  # 24 hours
            Statistic='Sum',
            Threshold=threshold_dollars,
            ActionsEnabled=True,
            AlarmActions=[
                'arn:aws:sns:us-east-1:123456789012:llm-cost-alerts'
            ],
            AlarmDescription=f'Alert when daily LLM costs exceed ${threshold_dollars}'
        )

Key Metrics Dashboard

CloudWatch Insights Queries:

-- Cost per model per day
fields @timestamp, model, sum(cost) as daily_cost
| filter namespace = "LLM/Costs"
| stats sum(daily_cost) by model, bin(@timestamp, 1d)

-- Top 10 most expensive users
fields user_id, sum(cost) as user_cost
| filter namespace = "LLM/Costs"
| stats sum(user_cost) by user_id
| sort user_cost desc
| limit 10

-- Cache effectiveness (cost savings)
fields @timestamp,
       sum(cached_tokens) / sum(input_tokens) * 100 as cache_hit_rate,
       sum(cached_tokens) * (standard_price - cached_price) as savings
| filter namespace = "LLM/Costs" and model = "claude-sonnet-3.5"
| stats avg(cache_hit_rate), sum(savings) by bin(@timestamp, 1h)

Common Pitfalls

Output Token Costs

Output tokens cost 2-5x more than input tokens, yet optimization often focuses only on input.

# Example: RAG application
# Input: 8,000 tokens (2,000 user query + 6,000 retrieved context)
# Output: 2,000 tokens (detailed answer)

# GPT-4 cost:
# - Input: 8,000 × $10/1M = $0.08
# - Output: 2,000 × $30/1M = $0.06
# Output is 43% of total cost despite being 20% of tokens

# Solution: Aggressive max_tokens limits + conciseness prompts
system_prompt = """
Answer concisely in under 150 words.
Prioritize clarity over exhaustive detail.
"""

response = openai.chat.completions.create(
    model="gpt-4",
    messages=[...],
    max_tokens=200  # Hard limit
)

Cache Invalidation from Minor Changes

Small prompt variations invalidate entire cache, destroying effectiveness.

# BAD: Dynamic timestamp invalidates cache every request
system_prompt = f"""
You are a support agent.
Current time: {datetime.now().isoformat()}  # Different every time!
[... rest of prompt ...]
"""

# GOOD: Static content only, dynamic context in user message
system_prompt = """
You are a support agent.
[... static policies ...]
"""

user_message = f"""
Current time: {datetime.now().isoformat()}
User question: {question}
"""

Missing Early Instrumentation

Deploying to production without observability means discovering problems after the damage is done.

Token cost tracks request volume almost linearly, which makes the growth easy to project and easy to miss:

100 requests/day    = $50/month
1,000 requests/day  = $500/month
10,000 requests/day = $5,000/month
50,000 requests/day = $25,000/month

Instrument from day one, publish metrics to CloudWatch in the same change, and set the budget alerts before launch rather than after the first surprise.

Optimization Impact Matrix

Savings depend on your traffic mix, so the useful comparison is the lever each technique pulls and what it costs in quality and effort. The first three rows are safe to turn on for almost any workload; the last three need judgement.

Optimization	Cost lever	Quality impact	Implementation effort
Prompt caching	Repeated prefix billed at the cache-read rate	None	Low (provider feature)
Batch inference	Provider batch discount (50% on Bedrock)	None, async only	Low (provider feature)
Output limits	Output is the expensive side of the bill	Low, answers get shorter	Low (parameter setting)
Model routing	Cheap model absorbs the simple share of traffic	Needs an evaluation set	Medium (routing logic)
Semantic caching	Skips the call entirely on near-duplicates	Medium, staleness and false matches	Medium (vector store)
Token budgets	Caps the blast radius of runaway loops	None, prevents waste	Medium (budget system)

Where This Approach Holds

Prompt caching and a hard output limit are the right first move for almost any production LLM feature. The discount is a provider feature, so nothing about the architecture changes, neither technique touches the content of an answer, and both show up immediately in the usage fields the API already returns. Ship them alongside token metrics so the next optimization argument starts from data.

Three situations call for a different order. When traffic is too sparse to reach the cache before its TTL expires, the 1.25x write premium turns caching into a net loss, and batch inference or a smaller model is the better lever. When answers have to stay long and detailed, output limits fight the product, and the effort belongs in routing instead. When the workload is one narrow task rather than a general assistant, skip routing entirely: pick the cheapest model that clears your evaluation set and spend the saved complexity elsewhere, because a router only earns its keep on a genuinely mixed traffic profile.

References

• FinOps Framework Overview - FinOps Foundation’s operating model covering the Inform, Optimize, and Operate lifecycle for cloud financial management
• FinOps Principles - The six core principles of FinOps practice, updated in 2025 to include AI and non-cloud cost scopes
• What is AWS Billing and Cost Management? - AWS documentation overview of Cost Explorer, Budgets, and Cost Optimization Hub
• AWS Cost Explorer - Service documentation for analyzing and visualizing AWS spend by service, tag, and time period
• AWS Budgets - Setting cost and usage budget thresholds with automated alerts for AI/ML workload cost control
• AWS Cost Optimization Hub - Unified view of rightsizing and savings recommendations that accounts for Reserved Instances and Savings Plans
• Prompt caching for faster model inference - Amazon Bedrock cache checkpoints, per-model token minimums, TTL behaviour, and the usage fields the Converse API returns
• Intelligent prompt routing in Amazon Bedrock - How default and configured routers pick between two models in one family, including the English-only optimization limit
• Amazon Bedrock pricing - Per-model token rates, the batch inference discount, and the cache read and write rates used in the cost models above
• Anthropic pricing - Published input and output token rates for the Claude model family
• OpenAI API pricing - Published per-token rates for the GPT models used in the routing example