AI Cost Explosion: Why Your Azure OpenAI Bill Tripled and How to Fix It

Daily token breakdown (real customer data, anonymized):
────────────────────────────────────────────────────
System prompt:          800 tokens  x 50,000 = 40M input tokens
Retrieved context:    2,500 tokens  x 50,000 = 125M input tokens
User query:             150 tokens  x 50,000 = 7.5M input tokens
────────────────────────────────────────────────────
Total input:                                   172.5M tokens/day
Model response:         400 tokens  x 50,000 = 20M output tokens

Daily cost (GPT-4o):
  Input:  172.5M / 1M * $2.50  = $431.25
  Output: 20M / 1M * $10.00    = $200.00
  Total:                         $631.25/day = ~$19,000/month

from enum import Enum
from dataclasses import dataclass

class ModelTier(Enum):
    NANO = "gpt-4.1-nano"      # Classification, simple extraction
    MINI = "gpt-4o-mini"       # Standard Q&A, summarization
    STANDARD = "gpt-4o"        # Complex reasoning, nuanced responses
    REASONING = "o3-mini"      # Multi-step logic, math, analysis

@dataclass
class RoutingDecision:
    model: ModelTier
    reason: str
    estimated_cost_ratio: float  # Relative to GPT-4o

class ModelRouter:
    """Route queries to the cheapest model that meets quality requirements."""

    # Classify query complexity using a cheap model
    ROUTING_PROMPT = """Classify this query's complexity. Respond with ONE word:
    - SIMPLE: factual lookup, yes/no, simple extraction
    - MODERATE: summarization, explanation, standard Q&A
    - COMPLEX: multi-step reasoning, nuanced analysis, creative
    - REASONING: mathematical, logical deduction, planning

    Query: {query}"""

    async def route(self, query: str, context: dict) -> RoutingDecision:
        # Rule-based fast path for known patterns
        if context.get("task_type") == "classification":
            return RoutingDecision(ModelTier.NANO, "classification_task", 0.04)
        if context.get("task_type") == "extraction":
            return RoutingDecision(ModelTier.NANO, "extraction_task", 0.04)

        # Use nano model for classification (costs almost nothing)
        complexity = await self._classify_complexity(query)

        routing_map = {
            "SIMPLE": RoutingDecision(ModelTier.NANO, "simple_query", 0.04),
            "MODERATE": RoutingDecision(ModelTier.MINI, "moderate_query", 0.06),
            "COMPLEX": RoutingDecision(ModelTier.STANDARD, "complex_query", 1.0),
            "REASONING": RoutingDecision(ModelTier.REASONING, "reasoning_query", 0.44),
        }
        return routing_map.get(complexity,
            RoutingDecision(ModelTier.MINI, "default_moderate", 0.06))

import numpy as np
from redis import Redis
from openai import AzureOpenAI

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

    async def get_or_generate(self, query: str, system_prompt: str,
                               context: str, generate_fn) -> dict:
        # Generate embedding for the query
        query_embedding = await self._get_embedding(query)

        # Search for similar cached queries
        cached = await self._search_cache(query_embedding)

        if cached and cached["similarity"] >= self.threshold:
            return {
                "response": cached["response"],
                "cached": True,
                "similarity": cached["similarity"],
                "tokens_saved": cached["total_tokens"],
                "cost_saved": cached["estimated_cost"],
            }

        # Cache miss — generate fresh response
        result = await generate_fn(query, system_prompt, context)

        # Store in cache
        await self._store_in_cache(
            query=query,
            embedding=query_embedding,
            response=result["response"],
            total_tokens=result["usage"]["total_tokens"],
            estimated_cost=result["estimated_cost"],
        )

        return {**result, "cached": False}

    async def _search_cache(self, embedding: list) -> dict | None:
        """Vector similarity search in Redis."""
        results = self.redis.ft("prompt_cache_idx").search(
            query=f"*=>[KNN 1 @embedding $vec AS similarity]",
            query_params={"vec": np.array(embedding, dtype=np.float32).tobytes()},
        )
        if results.docs:
            doc = results.docs[0]
            return {
                "response": doc.response,
                "similarity": float(doc.similarity),
                "total_tokens": int(doc.total_tokens),
                "estimated_cost": float(doc.estimated_cost),
            }
        return None

    async def _get_embedding(self, text: str) -> list:
        response = self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=text,
            dimensions=256,  # Smaller dimensions = faster search, less storage
        )
        return response.data[0].embedding

class ContextCompressor:
    """Reduce RAG context to only the most relevant chunks."""

    def compress_context(self, chunks: list, query: str,
                          max_tokens: int = 1500) -> list:
        """
        Instead of sending top-10 chunks (2500 tokens),
        send top-3 most relevant (750 tokens) with higher relevance threshold.
        """
        # Re-rank chunks by relevance score
        ranked = sorted(chunks, key=lambda c: c["score"], reverse=True)

        # Apply minimum relevance threshold
        relevant = [c for c in ranked if c["score"] >= 0.82]

        # Take only enough to fill token budget
        selected = []
        current_tokens = 0
        for chunk in relevant:
            chunk_tokens = len(chunk["text"].split()) * 1.3  # Rough token estimate
            if current_tokens + chunk_tokens > max_tokens:
                break
            selected.append(chunk)
            current_tokens += chunk_tokens

        return selected

# BEFORE: 800 tokens
BLOATED_SYSTEM_PROMPT = """
You are a helpful, knowledgeable, and friendly customer support assistant
for Acme Corporation. Your role is to help customers with their questions
about our products, services, billing, and technical issues. You should
always be polite, professional, and thorough in your responses. When you
don't know the answer to a question, you should honestly say that you
don't know rather than making something up. You should always try to
provide accurate and up-to-date information based on the context provided.
Please format your responses in a clear and easy-to-read manner...
"""

# AFTER: 180 tokens (78% reduction)
OPTIMIZED_SYSTEM_PROMPT = """Acme Corp support assistant.
Rules: Answer from context only. Say "I don't know" if unsure.
No personal data. Max 150 words. Cite sources."""

# Add to every API call
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    max_tokens=300,        # Hard limit on output length
    temperature=0.3,       # Lower temperature = more concise
)

Monthly PTU cost (example, GPT-4o, 100 PTUs): ~$6,000/month
Monthly PAYG cost at equivalent throughput:    Varies by usage

Break-even calculation:
  PTU monthly cost / PAYG per-token rate = break-even tokens

  $6,000 / ($2.50/1M input + $10.00/1M output weighted)
  ≈ at ~60-70% utilization of PTU capacity, PTU becomes cheaper

class HybridDeploymentRouter:
    """Route requests between PTU and PAYG deployments."""

    def __init__(self, ptu_endpoint: str, payg_endpoint: str,
                 ptu_capacity_threshold: float = 0.85):
        self.ptu_endpoint = ptu_endpoint
        self.payg_endpoint = payg_endpoint
        self.threshold = ptu_capacity_threshold

    async def route_request(self, request: dict) -> str:
        ptu_utilization = await self._get_ptu_utilization()

        if ptu_utilization < self.threshold:
            return self.ptu_endpoint  # Use reserved capacity
        else:
            return self.payg_endpoint  # Overflow to pay-as-you-go

import json
from openai import AzureOpenAI

client = AzureOpenAI(
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_version="2025-04-01-preview",
)

# Prepare batch file (JSONL format)
batch_requests = []
for doc in documents_to_process:
    batch_requests.append({
        "custom_id": doc["id"],
        "method": "POST",
        "url": "/chat/completions",
        "body": {
            "model": "gpt-4o-mini",
            "messages": [
                {"role": "system", "content": "Extract key entities from this document."},
                {"role": "user", "content": doc["text"][:4000]},
            ],
            "temperature": 0.1,
        }
    })

# Write JSONL
with open("batch_input.jsonl", "w") as f:
    for req in batch_requests:
        f.write(json.dumps(req) + "\n")

# Upload and create batch
batch_file = client.files.create(file=open("batch_input.jsonl", "rb"), purpose="batch")
batch_job = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/chat/completions",
    completion_window="24h",
)
print(f"Batch job created: {batch_job.id}")
# Results available within 24h at 50% cost

// KQL: Daily Azure OpenAI cost breakdown by model and deployment
AzureDiagnostics
| where ResourceProvider == "MICROSOFT.COGNITIVESERVICES"
| where Category == "RequestResponse"
| extend model = tostring(parse_json(properties_s).model)
| extend promptTokens = toint(parse_json(properties_s).promptTokens)
| extend completionTokens = toint(parse_json(properties_s).completionTokens)
| extend inputCost = case(
    model startswith "gpt-4o-mini", promptTokens / 1000000.0 * 0.15,
    model startswith "gpt-4o", promptTokens / 1000000.0 * 2.50,
    model startswith "gpt-4.1-nano", promptTokens / 1000000.0 * 0.10,
    0.0)
| extend outputCost = case(
    model startswith "gpt-4o-mini", completionTokens / 1000000.0 * 0.60,
    model startswith "gpt-4o", completionTokens / 1000000.0 * 10.00,
    model startswith "gpt-4.1-nano", completionTokens / 1000000.0 * 0.40,
    0.0)
| summarize
    TotalRequests = count(),
    TotalInputTokens = sum(promptTokens),
    TotalOutputTokens = sum(completionTokens),
    EstimatedInputCost = sum(inputCost),
    EstimatedOutputCost = sum(outputCost),
    EstimatedTotalCost = sum(inputCost) + sum(outputCost)
    by bin(TimeGenerated, 1d), model
| order by TimeGenerated desc

resource costAlert 'Microsoft.Insights/metricAlerts@2018-03-01' = {
  name: 'azure-openai-daily-cost-spike'
  location: 'global'
  properties: {
    severity: 2
    scopes: [openAIAccount.id]
    evaluationFrequency: 'PT1H'
    windowSize: 'PT1H'
    criteria: {
      'odata.type': 'Microsoft.Azure.Monitor.SingleResourceMultipleMetricCriteria'
      allOf: [
        {
          name: 'token-spike'
          metricName: 'ProcessedPromptTokens'
          operator: 'GreaterThan'
          threshold: 5000000  // Alert if >5M tokens in 1 hour
          timeAggregation: 'Total'
        }
      ]
    }
    actions: [{ actionGroupId: actionGroup.id }]
  }
}