Module 8: Prompt Engineering in Production

3. Cost Optimization

Overview

Strategies to cut LLM cost in production: token reduction, context compression, prompt caching (OpenAI and Anthropic), smart model routing, and budget tracking with alerts. How to reduce cost 50-80% without sacrificing quality.


The Cost Problem in Production

LLM cost in production can catch you off guard:

# Quick cost estimate with no optimization:
requests_per_day = 10_000
avg_input_tokens = 800   # Long prompt
avg_output_tokens = 150  # Answer
input_price_4o_mini = 0.15 / 1_000_000   # $0.15/1M tokens
output_price_4o_mini = 0.60 / 1_000_000  # $0.60/1M tokens

daily_cost = (
    requests_per_day * avg_input_tokens * input_price_4o_mini +
    requests_per_day * avg_output_tokens * output_price_4o_mini
)
monthly_cost = daily_cost * 30

print(f"Without optimization: ${monthly_cost:.2f}/month")
# → Without optimization: $36.00/month

# At 100,000 requests/day:
# → $360/month without optimization
# → With 3 techniques applied well: $72/month (80% reduction)

The 5 Cost Optimization Techniques

Technique 1: Token Reduction       → Fewer tokens per request
Technique 2: Prompt Caching        → Reuse common prefixes
Technique 3: Model Routing         → Use the cheap model when it's enough
Technique 4: Context Compression   → Summarize/compress long contexts
Technique 5: Output Constraints    → Cap the output tokens

Technique 1: Token Reduction

Principle: Every token has a cost

from openai import OpenAI
import tiktoken

client = OpenAI()
encoder = tiktoken.encoding_for_model("gpt-4o-mini")

def count_tokens(text: str) -> int:
    return len(encoder.encode(text))

def estimate_cost(
    prompt_tokens: int,
    completion_tokens: int,
    model: str = "gpt-4o-mini"
) -> float:
    """Estimates the cost of a request in USD."""
    prices = {
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "gpt-4o": {"input": 2.50, "output": 10.00},
    }
    
    p = prices.get(model, prices["gpt-4o-mini"])
    return (prompt_tokens * p["input"] + completion_tokens * p["output"]) / 1_000_000

Spotting Unnecessary Tokens

# BEFORE — verbose prompt (148 tokens):
VERBOSE_PROMPT = """
You are a very helpful and experienced sentiment classification assistant.
Your main mission is to analyze the sentiment of the text you are given
and classify it into one of the following three main categories:
POSITIVE, NEGATIVE or NEUTRAL.
It is very important that you respond ONLY with the category and absolutely
nothing else, with no extra explanations or additional text.

Text to analyze: {input}

Your classification:
"""

# AFTER — compact prompt (31 tokens):
COMPACT_PROMPT = """Classify as POSITIVE, NEGATIVE or NEUTRAL. Category only.

Text: {input}
Category:"""

# Difference:
verbose_tokens = count_tokens(VERBOSE_PROMPT.format(input="test text"))
compact_tokens = count_tokens(COMPACT_PROMPT.format(input="test text"))

print(f"Verbose: {verbose_tokens} tokens")
print(f"Compact: {compact_tokens} tokens")
print(f"Reduction: {(1 - compact_tokens/verbose_tokens):.0%}")
# Verbose: ~148 tokens
# Compact: ~31 tokens
# Reduction: ~79%

Token Reduction Guide

def audit_prompt_tokens(prompt: str) -> dict:
    """
    Analyzes a prompt and spots token reduction opportunities.
    """
    tokens = count_tokens(prompt)
    problems = []
    
    # Detect common verbose patterns
    verbose_patterns = [
        ("As an expert in", "Reduce to a direct instruction"),
        ("It is very important that", "Remove — direct instructions are more effective"),
        ("Please make sure to", "Remove — use a direct imperative"),
        ("Your mission is", "Reduce to 'Your task:'"),
        ("absolutely", "Remove — redundant"),
        ("without any doubt", "Remove — redundant"),
        ("Provide only", "Simplify to 'Only:'"),
    ]
    
    for pattern, suggestion in verbose_patterns:
        if pattern.lower() in prompt.lower():
            problems.append(f"'{pattern}' → {suggestion}")
    
    return {
        "current_tokens": tokens,
        "cost_per_1000_requests": f"${estimate_cost(tokens, 50) * 1000:.4f}",
        "detected_problems": problems,
        "has_opportunities": len(problems) > 0
    }

Few-Shot: Optimize the Number of Examples

def optimize_few_shot(task: str, examples: list[dict]) -> dict:
    """
    Finds the minimum number of examples with satisfactory accuracy.
    This cuts tokens without sacrificing quality.
    """
    results = {}
    
    golden_set = [
        {"input": "Excellent product", "expected": "POSITIVE"},
        {"input": "Terrible service", "expected": "NEGATIVE"},
        {"input": "The package arrived yesterday", "expected": "NEUTRAL"},
        # ... more examples
    ]
    
    for n_examples in [0, 1, 2, 3, 5]:
        examples_subset = examples[:n_examples]
        
        # Build a prompt with N examples
        if examples_subset:
            examples_str = "\n".join(
                f"- '{e['input']}' → {e['output']}"
                for e in examples_subset
            )
            prompt = f"Classify. Examples:\n{examples_str}\n\nText: {{input}}\nCategory:"
        else:
            prompt = "Classify as POSITIVE, NEGATIVE or NEUTRAL:\n\nText: {input}\nCategory:"
        
        tokens = count_tokens(prompt.format(input="test text"))
        
        # Evaluate accuracy (simplified)
        outputs = []
        for ex in golden_set[:10]:
            r = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt.format(input=ex["input"])}],
                temperature=0
            )
            outputs.append(r.choices[0].message.content.strip())
        
        accuracy = sum(o.lower() == e["expected"].lower() for o, e in zip(outputs, golden_set[:10])) / 10
        
        results[n_examples] = {
            "tokens": tokens,
            "accuracy": accuracy,
            "relative_cost": tokens / count_tokens(prompt.format(input="text"))
        }
    
    return results

Technique 2: Prompt Caching

Prompt caching is the highest-ROI technique when you have prompts with long common prefixes.

OpenAI Prompt Caching (Automatic)

OpenAI automatically caches the first 1,024+ tokens of prompts that repeat:

from openai import OpenAI

client = OpenAI()

# The long system prompt is cached automatically after the first request
LONG_SYSTEM_PROMPT = """
You are a legal analysis assistant specialized in software contracts.
Your job is to analyze contracts and answer specific questions.

RULES:
1. Only analyze the contract provided in the user's message
2. Cite the clause number when you reference the contract
3. If it isn't in the contract, say so explicitly
4. Don't give legal opinions, only descriptive analysis
5. Answer in English

RESPONSE FORMAT:
- Direct answer to the question
- Quote of the relevant clause (if applicable)
- Additional notes (if relevant)

[... 500+ more tokens of instructions ...]
"""  # 800+ tokens — cached automatically

def analyze_contract(contract: str, question: str) -> dict:
    """
    First call: cache miss — normal cost
    Following calls with the same system prompt: cache hit — 50% discount on input
    """
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": LONG_SYSTEM_PROMPT},
            {"role": "user", "content": f"Contract:\n{contract}\n\nQuestion: {question}"}
        ],
        temperature=0
    )
    
    # In the response you can see whether there was a cache hit:
    usage = response.usage
    cached_tokens = getattr(usage, 'prompt_tokens_details', {})
    
    return {
        "answer": response.choices[0].message.content,
        "tokens_used": usage.total_tokens,
        "cached_tokens": getattr(cached_tokens, 'cached_tokens', 0) if cached_tokens else 0
    }

Anthropic Prompt Caching (Explicit)

Anthropic lets you mark explicitly what to cache:

import anthropic

client_anthropic = anthropic.Anthropic()

LONG_SYSTEM_PROMPT = "..."  # 1000+ tokens

def query_with_cache(question: str) -> str:
    """
    Marks the system prompt with cache_control for explicit caching.
    Cost: First call = normal, following ones = 90% discount on cached tokens.
    """
    response = client_anthropic.messages.create(
        model="claude-3-haiku-20240307",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": LONG_SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"}  # Cache this block
            }
        ],
        messages=[
            {"role": "user", "content": question}
        ]
    )
    
    # Check cache usage
    cache_creation = response.usage.cache_creation_input_tokens
    cache_read = response.usage.cache_read_input_tokens
    
    print(f"Cache creation: {cache_creation}, Cache read: {cache_read}")
    # First call: cache_creation > 0, cache_read = 0
    # Following ones: cache_creation = 0, cache_read > 0 (90% discount)
    
    return response.content[0].text


# Multi-turn with cached documents:
LONG_DOCUMENT = "... 50,000 tokens of document ..."

def qa_over_document(questions: list[str]) -> list[str]:
    """
    Caches the long document, asks multiple questions about it.
    Without cache: 50K tokens × N questions
    With cache: 50K tokens (creation) + N × 100 tokens (read)
    """
    answers = []
    
    for i, question in enumerate(questions):
        response = client_anthropic.messages.create(
            model="claude-3-haiku-20240307",
            max_tokens=500,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": LONG_DOCUMENT,
                            "cache_control": {"type": "ephemeral"}
                        },
                        {
                            "type": "text",
                            "text": f"Question: {question}"
                        }
                    ]
                }
            ]
        )
        answers.append(response.content[0].text)
        
        if i == 0:
            print(f"First call (cache miss): {response.usage.cache_creation_input_tokens} tokens cached")
        else:
            print(f"Call {i+1} (cache hit): {response.usage.cache_read_input_tokens} tokens read from cache")
    
    return answers

Technique 3: Model Routing

Not every request needs the most powerful model:

from openai import OpenAI

client = OpenAI()

# Approximate prices (per 1M input tokens)
MODEL_PRICES = {
    "gpt-4o-mini": 0.15,    # Cheap, fast
    "gpt-4o": 2.50,          # Powerful, expensive (16.7x more expensive)
}

class ModelRouter:
    """
    Smart router that picks the model based on request complexity.
    
    Strategies:
    1. Complexity classification by keywords
    2. Input length
    3. Task type
    4. Result of a previous evaluation
    """
    
    def __init__(
        self,
        cheap_model: str = "gpt-4o-mini",
        powerful_model: str = "gpt-4o",
        threshold_tokens: int = 2000  # Long requests → powerful model
    ):
        self.cheap_model = cheap_model
        self.powerful_model = powerful_model
        self.threshold_tokens = threshold_tokens
        
        # Patterns that signal high complexity
        self.complex_patterns = [
            "reason step by step",
            "analyze in detail",
            "complex code",
            "mathematics",
            "legal",
            "medical",
            "multiple factors",
            "pros and cons"
        ]
    
    def select_model(
        self,
        input_text: str,
        task_type: str = "general"
    ) -> tuple[str, str]:
        """
        Selects the appropriate model.
        
        Returns: (model, reason)
        """
        # Simple task always → cheap model
        simple_tasks = ["classification", "simple_extraction", "short_summary"]
        if task_type in simple_tasks:
            return self.cheap_model, "simple_task"
        
        # Long input → powerful model
        tokens = len(input_text.split())  # Simple approximation
        if tokens > self.threshold_tokens:
            return self.powerful_model, "long_input"
        
        # Detect complexity by keywords
        input_lower = input_text.lower()
        for pattern in self.complex_patterns:
            if pattern in input_lower:
                return self.powerful_model, f"complex_pattern: {pattern}"
        
        # Default: cheap model
        return self.cheap_model, "default"
    
    def run(
        self,
        prompt: str,
        input_text: str,
        task_type: str = "general",
        **kwargs
    ) -> dict:
        """Runs the request with the appropriate model."""
        model, reason = self.select_model(input_text, task_type)
        
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt.format(input=input_text)}],
            temperature=0,
            **kwargs
        )
        
        cost = estimate_cost(
            response.usage.prompt_tokens,
            response.usage.completion_tokens,
            model
        )
        
        return {
            "output": response.choices[0].message.content,
            "model_used": model,
            "routing_reason": reason,
            "cost": cost,
            "tokens": response.usage.total_tokens
        }


# Usage:
router = ModelRouter()

# Simple request → mini automatically
result = router.run(
    prompt="Classify as POSITIVE/NEGATIVE/NEUTRAL: {input}",
    input_text="I love the product",
    task_type="classification"
)
print(f"Model: {result['model_used']} ({result['routing_reason']})")
# Model: gpt-4o-mini (simple_task)

# Complex request → gpt-4o
result = router.run(
    prompt="Analyze this legal contract in detail: {input}",
    input_text="Software contract..." * 100,
    task_type="legal_analysis"
)
print(f"Model: {result['model_used']} ({result['routing_reason']})")
# Model: gpt-4o (long_input)

Routing with a Prior Classifier

def classify_complexity(text: str) -> str:
    """
    Uses a small model to classify complexity,
    before calling the main model.
    
    Cost: ~10 tokens to classify vs 500 tokens of a complex prompt
    ROI: If 70% are simple, that saves 16x on those requests.
    """
    classifier_prompt = """Does this task require complex reasoning or is it simple?
Simple: classification, direct extraction, 1-2 sentence summary
Complex: multi-stage analysis, code, mathematics, information synthesis

Respond ONLY: SIMPLE or COMPLEX

Task: """ + text[:200]
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": classifier_prompt}],
        temperature=0,
        max_tokens=5
    )
    
    result = response.choices[0].message.content.strip().upper()
    return "COMPLEX" if "COMPLEX" in result else "SIMPLE"


# Two calls, but bigger savings:
def run_with_smart_routing(task: str, input_text: str) -> dict:
    complexity = classify_complexity(task + " " + input_text)
    
    model = "gpt-4o" if complexity == "COMPLEX" else "gpt-4o-mini"
    
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": f"{task}\n\n{input_text}"}],
        temperature=0
    )
    
    return {
        "output": response.choices[0].message.content,
        "complexity": complexity,
        "model": model
    }

Technique 4: Context Compression

When the context is long (documents, conversation history), compress it before sending:

def compress_conversation_history(
    history: list[dict],
    max_tokens: int = 500
) -> list[dict]:
    """
    Compresses the conversation history while keeping what matters.
    
    Strategy: Summarize old messages, keep the N most recent ones intact.
    """
    if not history:
        return []
    
    # Keep the last 4 messages intact (2 turns)
    recent_messages = history[-4:] if len(history) >= 4 else history
    old_messages = history[:-4] if len(history) >= 4 else []
    
    if not old_messages:
        return recent_messages
    
    # Summarize old messages
    old_text = "\n".join(
        f"{m['role'].upper()}: {m['content'][:200]}"
        for m in old_messages
    )
    
    summary_prompt = f"""Summarize this conversation in at most 3 sentences, keeping the key points and decisions made:

{old_text}

Summary:"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": summary_prompt}],
        temperature=0,
        max_tokens=150
    )
    
    summary = response.choices[0].message.content.strip()
    
    # Combine: summary + recent messages
    return [
        {"role": "system", "content": f"Summary of the previous conversation: {summary}"}
    ] + recent_messages


def compress_document(
    document: str,
    max_tokens: int = 500,
    focus: str = ""
) -> str:
    """
    Compresses a long document into a focused summary.
    
    focus: Which aspects matter most for the task.
    """
    doc_tokens = count_tokens(document)
    
    if doc_tokens <= max_tokens:
        return document  # No compression needed
    
    focus_str = f" focusing on: {focus}" if focus else ""
    
    prompt = f"""Summarize the following document in at most {max_tokens//4} words{focus_str}.
Preserve specific data, numbers and important dates.

DOCUMENT:
{document[:8000]}  # Safety limit

SUMMARY:"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        max_tokens=max_tokens
    )
    
    summary = response.choices[0].message.content.strip()
    
    print(f"Document: {doc_tokens} tokens → Summary: {count_tokens(summary)} tokens")
    return summary

Technique 5: Output Constraints

def run_with_controlled_output(
    prompt: str,
    input_text: str,
    max_output_tokens: int = 50,
    use_structured: bool = True
) -> dict:
    """
    Controls the output size to cut cost.
    
    For classification/extraction: max_tokens=10-50 is plenty.
    For summarization: max_tokens=100-200.
    """
    kwargs = {
        "model": "gpt-4o-mini",
        "messages": [{"role": "user", "content": prompt.format(input=input_text)}],
        "temperature": 0,
        "max_tokens": max_output_tokens  # Key: cap the output tokens
    }
    
    # For structured output: always more efficient than free text
    if use_structured:
        kwargs["response_format"] = {"type": "json_object"}
    
    response = client.chat.completions.create(**kwargs)
    
    return {
        "output": response.choices[0].message.content,
        "prompt_tokens": response.usage.prompt_tokens,
        "completion_tokens": response.usage.completion_tokens,
        "cost": estimate_cost(response.usage.prompt_tokens, response.usage.completion_tokens)
    }


# Example: classification with max_tokens=5 (enough for "POSITIVE")
result = run_with_controlled_output(
    prompt="Classify as POSITIVE, NEGATIVE or NEUTRAL. Category only.\n\nText: {input}",
    input_text="I love this product",
    max_output_tokens=5  # "POSITIVE" = 1-2 tokens
)
print(f"Output: '{result['output']}', completion_tokens: {result['completion_tokens']}")

Budget Tracking and Alerts

import json
from datetime import datetime, date
from pathlib import Path


class BudgetTracker:
    """
    Tracks the cost of API calls and alerts when it gets close to the limit.
    """
    
    def __init__(
        self,
        daily_budget: float = 10.0,
        monthly_budget: float = 200.0,
        storage_path: str = "cost_tracking.json"
    ):
        self.daily_budget = daily_budget
        self.monthly_budget = monthly_budget
        self.storage_path = Path(storage_path)
        self._data = self._load()
    
    def _load(self) -> dict:
        if self.storage_path.exists():
            with open(self.storage_path) as f:
                return json.load(f)
        return {"daily": {}, "monthly": {}, "total": 0.0}
    
    def _save(self) -> None:
        with open(self.storage_path, "w") as f:
            json.dump(self._data, f, indent=2)
    
    def record(self, api_response) -> dict:
        """
        Records the cost of an API response.
        
        api_response: OpenAI response object.
        """
        usage = api_response.usage
        
        # Compute cost (GPT-4o-mini by default)
        cost = (
            usage.prompt_tokens * 0.15 / 1_000_000 +
            usage.completion_tokens * 0.60 / 1_000_000
        )
        
        today = date.today().isoformat()
        month = date.today().strftime("%Y-%m")
        
        # Update tracking
        self._data["daily"][today] = self._data["daily"].get(today, 0) + cost
        self._data["monthly"][month] = self._data["monthly"].get(month, 0) + cost
        self._data["total"] += cost
        
        self._save()
        
        # Check alerts
        alerts = []
        today_cost = self._data["daily"][today]
        month_cost = self._data["monthly"][month]
        
        if today_cost >= self.daily_budget * 0.8:
            alerts.append(f"⚠️ 80% of the daily budget reached: ${today_cost:.4f}/${self.daily_budget}")
        if today_cost >= self.daily_budget:
            alerts.append(f"🚨 Daily budget EXCEEDED: ${today_cost:.4f}/${self.daily_budget}")
        if month_cost >= self.monthly_budget * 0.8:
            alerts.append(f"⚠️ 80% of the monthly budget reached: ${month_cost:.2f}/${self.monthly_budget}")
        
        return {
            "request_cost": cost,
            "today_cost": today_cost,
            "month_cost": month_cost,
            "alerts": alerts
        }
    
    def summary(self) -> dict:
        """Returns a summary of current costs."""
        today = date.today().isoformat()
        month = date.today().strftime("%Y-%m")
        
        today_cost = self._data["daily"].get(today, 0)
        month_cost = self._data["monthly"].get(month, 0)
        
        return {
            "today_cost": f"${today_cost:.4f}",
            "pct_daily_budget": f"{today_cost/self.daily_budget*100:.1f}%",
            "month_cost": f"${month_cost:.2f}",
            "pct_monthly_budget": f"{month_cost/self.monthly_budget*100:.1f}%",
            "historical_total": f"${self._data['total']:.2f}"
        }
    
    def monthly_projection(self) -> float:
        """Projects the monthly cost based on the last 7 days."""
        last_7 = []
        for i in range(7):
            from datetime import timedelta
            day = (date.today() - timedelta(days=i)).isoformat()
            cost = self._data["daily"].get(day, 0)
            last_7.append(cost)
        
        daily_average = sum(last_7) / len([c for c in last_7 if c > 0] or [1])
        return daily_average * 30


# Wrapper with tracking built in:
budget = BudgetTracker(daily_budget=5.0, monthly_budget=100.0)

def call_with_budget(prompt: str, input_text: str) -> str:
    """API call with automatic budget tracking."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt.format(input=input_text)}],
        temperature=0
    )
    
    tracking = budget.record(response)
    
    for alert in tracking["alerts"]:
        print(alert)
    
    return response.choices[0].message.content

Comparison: Before vs After Optimization

TechniqueBeforeAfterReduction
Token reduction800 tokens/req200 tokens/req75%
Prompt cachingNo cache60% cache hit30% on those requests
Model routingGPT-4o alwaysGPT-4o-mini (70%)70% less on routed ones
Context compression4,000-token doc500-token summary87%
max_tokens500 default50 for classification90% output tokens

Combined: In a real system with every technique → 50-80% cost reduction


Troubleshooting

Problem 1: Cutting tokens breaks quality

Symptom: The compressed prompt has 15% lower accuracy.

Cause: You removed necessary instructions, not just filler.

Solution:

# The right process for token optimization:
# 1. Establish an accuracy baseline with the original prompt
# 2. Remove elements one at a time
# 3. Measure accuracy after each removal
# 4. Keep the removal if accuracy doesn't drop > 2%

def optimize_prompt_iteratively(
    original_prompt: str,
    golden_set: list[dict],
    max_degradation: float = 0.02
) -> str:
    """Removes tokens safely while checking accuracy."""
    
    def evaluate_accuracy(prompt_template):
        correct = 0
        for ex in golden_set[:20]:  # subset for speed
            r = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt_template.format(input=ex["input"])}],
                temperature=0
            )
            if r.choices[0].message.content.strip().lower() == str(ex["expected_output"]).lower():
                correct += 1
        return correct / min(20, len(golden_set))
    
    baseline = evaluate_accuracy(original_prompt)
    current_prompt = original_prompt
    
    # Try removing redundant lines
    lines = original_prompt.split("\n")
    for i, line in enumerate(lines):
        if len(line.strip()) < 5:  # Skip empty lines
            continue
        
        # Try without this line
        prompt_without_line = "\n".join(lines[:i] + lines[i+1:])
        accuracy_without = evaluate_accuracy(prompt_without_line)
        
        if accuracy_without >= baseline - max_degradation:
            # The line wasn't necessary
            lines[i] = ""
            current_prompt = "\n".join([l for l in lines if l])
            print(f"Removed: '{line[:40]}...' (accuracy: {accuracy_without:.2%})")
    
    return current_prompt

Problem 2: Prompt caching doesn't kick in

Symptom: Repeated calls show no caching discount.

Cause (OpenAI): The prompt has variables that change at the start, breaking the cache.

Solution:

# BAD: The variable at the start breaks the cache
bad_prompt = f"Analyze the document from {user}: {long_system_prompt}..."

# GOOD: The variable parts go at the end, the long prefix is cached
good_prompt = f"""{long_system_prompt}  ← This prefix is cached

---
User document:
{user_document}  ← Variable at the end
"""

Problem 3: Model routing sends complex requests to the cheap model

Symptom: Complex requests come out low quality because they land on gpt-4o-mini.

Solution:

# Add a fallback: if the output quality is low, retry with the powerful model
def run_with_quality_fallback(prompt: str, input_text: str, quality_threshold: float = 0.8) -> dict:
    # First attempt: cheap model
    response_mini = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt.format(input=input_text)}],
        temperature=0
    )
    output_mini = response_mini.choices[0].message.content
    
    # Check whether the output looks complete and well formed
    # (simple heuristic: minimum length, doesn't contain "I can't" or "I don't know")
    problem_words = ["i can't", "i don't know", "i have no information", "i'm sorry"]
    output_has_problem = any(p in output_mini.lower() for p in problem_words)
    output_too_short = len(output_mini.split()) < 5
    
    if output_has_problem or output_too_short:
        # Fallback to the powerful model
        response_4o = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt.format(input=input_text)}],
            temperature=0
        )
        return {
            "output": response_4o.choices[0].message.content,
            "model": "gpt-4o",
            "fallback_triggered": True
        }
    
    return {"output": output_mini, "model": "gpt-4o-mini", "fallback_triggered": False}

Exercises

Exercise 1: Compute the ROI of token reduction

You have a 600-token prompt and you want to cut it to 150. Compute the savings at 50,000 requests/day over 30 days:

See solution
def calculate_reduction_roi(
    tokens_before: int,
    tokens_after: int,
    requests_per_day: int,
    days: int = 30,
    model: str = "gpt-4o-mini"
) -> dict:
    input_price = {"gpt-4o-mini": 0.15, "gpt-4o": 2.50}[model]
    
    cost_before = tokens_before * input_price / 1_000_000 * requests_per_day * days
    cost_after = tokens_after * input_price / 1_000_000 * requests_per_day * days
    savings = cost_before - cost_after
    
    return {
        "cost_before": f"${cost_before:.2f}",
        "cost_after": f"${cost_after:.2f}",
        "savings": f"${savings:.2f}",
        "pct_reduction": f"{savings/cost_before*100:.0f}%"
    }

result = calculate_reduction_roi(
    tokens_before=600,
    tokens_after=150,
    requests_per_day=50_000
)
print(result)
# {'cost_before': '$135.00', 'cost_after': '$33.75', 'savings': '$101.25', 'pct_reduction': '75%'}

Exercise 2: Implement a simple model router

Implement a router that uses gpt-4o for requests with more than 500 words, and gpt-4o-mini for the rest:

See solution
from openai import OpenAI

client = OpenAI()

def smart_router(prompt_template: str, input_text: str, threshold_words: int = 500) -> dict:
    """Simple router by input length."""
    words = len(input_text.split())
    model = "gpt-4o" if words > threshold_words else "gpt-4o-mini"
    
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt_template.format(input=input_text)}],
        temperature=0
    )
    
    # Prices per input token
    price = 2.50 if model == "gpt-4o" else 0.15
    estimated_cost = response.usage.prompt_tokens * price / 1_000_000
    
    return {
        "output": response.choices[0].message.content,
        "model_used": model,
        "input_words": words,
        "estimated_cost": f"${estimated_cost:.6f}"
    }

# Test:
r1 = smart_router("Summarize: {input}", "Short text")
r2 = smart_router("Summarize: {input}", ("Long text " * 100))

print(f"Short input: {r1['model_used']} ({r1['input_words']} words)")
print(f"Long input: {r2['model_used']} ({r2['input_words']} words)")

Summary

  • Token reduction: The highest-impact technique — verbose prompts = needless cost
  • Prompt caching: Automatic in OpenAI, explicit in Anthropic — 50-90% discount on cached tokens
  • Model routing: Use gpt-4o-mini for 70%+ of simple requests — 16x cheaper
  • Context compression: For long conversations and documents — summarize before sending
  • max_tokens: Always set it — classification needs 5 tokens, not 500
  • Budget tracking: Watch daily and monthly cost with alerts before you blow through it

Additional resources

  1. OpenAI Pricing — Current prices per model
  2. OpenAI Prompt Caching — Official documentation
  3. Anthropic Prompt Caching — Anthropic caching
  4. tiktoken — OpenAI's token counter
  5. LLM Cost Calculator — Cost calculator