Claude API Batch and Model Routing: 50% Off Tokens and Smart Cost Control

import anthropic
import time

client = anthropic.Anthropic()

def run_batch(prompts: list[str], model: str = "claude-haiku-4-5") -> list[str]:
    """Submit a batch request. 50% cheaper than individual calls."""

    batch = client.messages.batches.create(
        requests=[
            {
                "custom_id": f"item-{i}",
                "params": {
                    "model": model,
                    "max_tokens": 512,
                    "messages": [{"role": "user", "content": prompt}]
                }
            }
            for i, prompt in enumerate(prompts)
        ]
    )

    print(f"Batch submitted: {batch.id} — status: {batch.processing_status}")

    # Poll until complete (usually well under 24 hours)
    while True:
        batch = client.messages.batches.retrieve(batch.id)
        if batch.processing_status == "ended":
            break
        print("Still processing — checking again in 60s")
        time.sleep(60)

    # Collect results in original order
    results = {}
    for result in client.messages.batches.results(batch.id):
        if result.result.type == "succeeded":
            results[result.custom_id] = result.result.message.content[0].text
        else:
            results[result.custom_id] = None

    return [results.get(f"item-{i}") for i in range(len(prompts))]

from anthropic import Anthropic

client = Anthropic()

def get_model(task: str) -> str:
    """Return the cheapest model that reliably handles this task."""
    ROUTING = {
        # Haiku: fast, cheap — good for classification, extraction, short rewrites
        "classify":  "claude-haiku-4-5",
        "extract":   "claude-haiku-4-5",
        "summarize": "claude-haiku-4-5",
        "translate": "claude-haiku-4-5",

        # Sonnet: balanced — handles most code and analysis well
        "code":    "claude-sonnet-4-6",
        "debug":   "claude-sonnet-4-6",
        "analyze": "claude-sonnet-4-6",
        "draft":   "claude-sonnet-4-6",

        # Opus: complex reasoning, architecture, multi-step plans only
        "reason":    "claude-opus-4-7",
        "architect": "claude-opus-4-7",
    }
    return ROUTING.get(task, "claude-sonnet-4-6")

def routed_call(message: str, task: str) -> str:
    model = get_model(task)
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": message}]
    )
    return response.content[0].text

# Classifying 10,000 tickets: Haiku at $1/MTok vs Opus at $5/MTok = 5x savings
result = routed_call("Classify this support ticket: ...", task="classify")

import anthropic
from typing import Literal

client = anthropic.Anthropic()

# Static system prompt — keep ALL dynamic content out of this string
# Must exceed minimum token threshold: 1,024 (Sonnet), 4,096 (Haiku/Opus)
STATIC_SYSTEM = """You are a senior software engineer assistant.
[...your full static instructions here...]
"""

TaskType = Literal["classify", "extract", "summarize", "code", "debug", "analyze", "reason"]

MODEL_ROUTING: dict[TaskType, str] = {
    "classify":  "claude-haiku-4-5",
    "extract":   "claude-haiku-4-5",
    "summarize": "claude-haiku-4-5",
    "code":      "claude-sonnet-4-6",
    "debug":     "claude-sonnet-4-6",
    "analyze":   "claude-sonnet-4-6",
    "reason":    "claude-opus-4-7",
}

def call_optimized(
    user_message: str,
    task: TaskType = "analyze",
    context: str | None = None,
) -> str:
    """
    Full optimization stack:
    - Prompt caching on static system (90% off cache reads)
    - Dynamic context in user turn, not system (keeps cache hit rate high)
    - Model routing to cheapest capable model
    """
    model = MODEL_ROUTING.get(task, "claude-sonnet-4-6")

    # Dynamic context goes in user turn — never in the cached system prefix
    content = f"{context}\n\n{user_message}" if context else user_message

    response = client.messages.create(
        model=model,
        max_tokens=1024,
        system=[{
            "type": "text",
            "text": STATIC_SYSTEM,
            "cache_control": {"type": "ephemeral"}
        }],
        messages=[{"role": "user", "content": content}]
    )

    usage = response.usage
    cache_hit = usage.cache_read_input_tokens > 0
    print(f"Model: {model} | Cache hit: {cache_hit}")

    return response.content[0].text