Cost Optimization Guide

import anthropic

client = anthropic.Anthropic()

# With prompt caching — mark the system prompt with cache_control
# The system prompt costs 10% on cache hits (within the TTL window)
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": """You are an expert financial analyst with deep knowledge of
            public company earnings, balance sheets, and market dynamics.

            When analyzing companies, always consider:
            - Revenue growth trajectory (YoY and QoQ)
            - Gross margin trends and drivers
            - Operating leverage and EBITDA margin expansion
            - Free cash flow conversion from net income
            - Balance sheet health: net debt/EBITDA, current ratio
            - Competitive positioning: market share, moat characteristics
            - Management guidance credibility vs historical accuracy
            - Key risks: regulatory, competitive, macro sensitivity

            Format your analysis in structured sections with clear headers.
            Support every claim with specific numbers from the provided data.
            Always state your assumptions explicitly.
            [... the rest of your 2,000+ token system prompt ...]
            """,
            "cache_control": {"type": "ephemeral"}  # Cache this!
        }
    ],
    messages=[{"role": "user", "content": "Analyze Apple's Q4 2024 earnings: ..."}]
)

# Check cache performance in the usage metadata
usage = response.usage
print(f"Input tokens:        {usage.input_tokens:,}")
print(f"Cache write tokens:  {getattr(usage, 'cache_creation_input_tokens', 0):,}  (full price, first call)")
print(f"Cache read tokens:   {getattr(usage, 'cache_read_input_tokens', 0):,}   (10% price, cache hit)")
print(f"Output tokens:       {usage.output_tokens:,}")

# Calculate actual cost for this call
SONNET_INPUT_PRICE = 3.0 / 1_000_000    # $3 per million
SONNET_CACHE_PRICE = 0.30 / 1_000_000   # $0.30 per million (10% of input)
SONNET_OUTPUT_PRICE = 15.0 / 1_000_000  # $15 per million

cache_write = getattr(usage, 'cache_creation_input_tokens', 0)
cache_read = getattr(usage, 'cache_read_input_tokens', 0)
regular_input = usage.input_tokens - cache_write - cache_read

cost = (
    regular_input * SONNET_INPUT_PRICE +
    cache_write * SONNET_INPUT_PRICE +     # Write costs full price
    cache_read * SONNET_CACHE_PRICE +      # Read costs 10%
    usage.output_tokens * SONNET_OUTPUT_PRICE
)
print(f"Estimated cost:      ${cost:.6f}")

import anthropic
import time
import json

client = anthropic.Anthropic()

# Your documents to process
documents_to_analyze = [
    "Customer complaint about shipping delay on order #4821...",
    "Positive review: product exceeded expectations, would recommend...",
    "Refund request for defective item received on Jan 15...",
    # ... up to 10,000 items per batch
]

# 1. Create the batch — all requests submitted at once
print(f"Submitting batch of {len(documents_to_analyze)} requests...")
batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": f"doc-{i}",  # Your ID to match results to inputs
            "params": {
                "model": "claude-haiku-4-5-20251001",
                "max_tokens": 100,
                "messages": [{
                    "role": "user",
                    "content": (
                        f"Classify this customer message. Reply with ONLY one of: "
                        f"complaint, compliment, refund_request, question, other\n\n"
                        f"Message: {doc}"
                    )
                }]
            }
        }
        for i, doc in enumerate(documents_to_analyze)
    ]
)

print(f"Batch created: {batch.id}")
print(f"Status: {batch.processing_status}")

# 2. Poll until complete (in production, use a webhook or scheduled job instead)
while True:
    status = client.messages.batches.retrieve(batch.id)
    print(f"Status: {status.processing_status} | "
          f"Succeeded: {status.request_counts.succeeded} | "
          f"Errored: {status.request_counts.errored}")

    if status.processing_status == "ended":
        break

    time.sleep(60)  # Check every minute

# 3. Retrieve and process results
results = {}
for result in client.messages.batches.results(batch.id):
    if result.result.type == "succeeded":
        classification = result.result.message.content[0].text.strip()
        results[result.custom_id] = classification
    else:
        # Handle errors
        results[result.custom_id] = f"ERROR: {result.result.error.type}"

# 4. Print summary
from collections import Counter
counts = Counter(results.values())
print("\nClassification results:")
for label, count in counts.most_common():
    print(f"  {label}: {count}")

print(f"\nTotal processed: {len(results)}")
print(f"Estimated cost vs real-time: ~50% savings")

def estimate_tokens(msg: dict) -> int:
    """
    Rough token estimate: 4 characters ≈ 1 token.
    Use tiktoken or Anthropic's token counting endpoint for accuracy.
    """
    content = msg.get('content', '')
    if isinstance(content, list):
        # Handle structured content (tool results, etc.)
        text = ' '.join(
            item.get('text', '') if isinstance(item, dict) else str(item)
            for item in content
        )
    else:
        text = str(content)
    return max(1, len(text) // 4)


def trim_conversation_history(
    messages: list,
    max_tokens: int = 50_000,
    always_keep_first: bool = True
) -> list:
    """
    Trim conversation history to stay within a token budget.

    Strategy:
    - Always keep the first user message (original task context)
    - Always keep the most recent N messages
    - Remove messages from the middle when over budget

    Args:
        messages: Full conversation history
        max_tokens: Maximum total tokens to allow (rough estimate)
        always_keep_first: If True, never remove messages[0]

    Returns:
        Trimmed message list
    """
    if not messages:
        return messages

    total = sum(estimate_tokens(m) for m in messages)

    if total <= max_tokens:
        return messages  # No trimming needed

    print(f"[trim] Conversation at ~{total:,} tokens, trimming to {max_tokens:,}...")

    if not always_keep_first:
        # Simple approach: keep most recent messages within budget
        result = []
        budget = max_tokens
        for msg in reversed(messages):
            tokens = estimate_tokens(msg)
            if budget - tokens > 0:
                result.insert(0, msg)
                budget -= tokens
            else:
                break
        return result

    # Keep first message, fill from the end
    first_msg = messages[0]
    remaining_messages = messages[1:]

    first_tokens = estimate_tokens(first_msg)
    budget = max_tokens - first_tokens

    result = []
    for msg in reversed(remaining_messages):
        tokens = estimate_tokens(msg)
        if budget - tokens > 0:
            result.insert(0, msg)
            budget -= tokens
        else:
            break

    trimmed = [first_msg] + result
    trimmed_total = sum(estimate_tokens(m) for m in trimmed)
    removed = len(messages) - len(trimmed)

    if removed > 0:
        print(f"[trim] Removed {removed} messages. New size: ~{trimmed_total:,} tokens")

    return trimmed


# Usage in your agent loop:
messages = trim_conversation_history(messages, max_tokens=50_000)

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=messages
)

import anthropic

client = anthropic.Anthropic()

# Pricing per million tokens (approximate, check Anthropic console for current rates)
MODEL_PRICING = {
    'claude-haiku-4-5-20251001':  {'input': 0.80, 'output': 4.00},
    'claude-sonnet-4-6':           {'input': 3.00, 'output': 15.00},
    'claude-opus-4-6':             {'input': 15.00, 'output': 75.00},
}

def classify_complexity(task: str) -> str:
    """
    Use cheap Haiku to classify task complexity.
    Costs ~$0.000008 per classification (almost free).
    """
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=20,  # We only need one word
        messages=[{
            "role": "user",
            "content": f"""Classify this task complexity. Reply with ONLY one word: simple, moderate, or complex.

Rules:
- simple: factual lookup, yes/no, formatting, basic calculation, FAQ
- moderate: multi-step reasoning, code generation, analysis, tool use
- complex: deep research, strategic planning, multi-document synthesis, novel problem-solving

Task: {task}

Classification:"""
        }]
    )
    result = response.content[0].text.strip().lower()
    # Validate — default to moderate if unexpected response
    return result if result in ('simple', 'moderate', 'complex') else 'moderate'


def select_model(complexity: str) -> str:
    """Map complexity to the appropriate model."""
    return {
        'simple':   'claude-haiku-4-5-20251001',
        'moderate': 'claude-sonnet-4-6',
        'complex':  'claude-opus-4-6',
    }[complexity]


def run_tiered_agent(task: str, system: str = "", verbose: bool = True) -> dict:
    """
    Route a task to the appropriate model based on complexity.
    Returns result dict with model used, response, and estimated cost.
    """
    # Step 1: Classify (costs ~$0.000008)
    complexity = classify_complexity(task)
    model = select_model(complexity)

    if verbose:
        print(f"[tiered] complexity={complexity} → model={model}")

    # Step 2: Run on the appropriate model
    messages = [{"role": "user", "content": task}]
    kwargs = {
        "model": model,
        "max_tokens": 2048,
        "messages": messages
    }
    if system:
        kwargs["system"] = system

    response = client.messages.create(**kwargs)

    # Step 3: Calculate cost
    usage = response.usage
    pricing = MODEL_PRICING[model]
    cost = (
        usage.input_tokens / 1_000_000 * pricing['input'] +
        usage.output_tokens / 1_000_000 * pricing['output']
    )

    return {
        "complexity": complexity,
        "model": model,
        "response": response.content[0].text,
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "estimated_cost_usd": round(cost, 6)
    }


# Example usage
tasks = [
    "What is the capital of France?",                          # → Haiku (simple)
    "Write a Python function to parse ISO 8601 timestamps",    # → Sonnet (moderate)
    "Develop a comprehensive go-to-market strategy for a B2B SaaS product targeting mid-market healthcare companies, including competitive positioning, pricing model analysis, and channel strategy"  # → Opus (complex)
]

total_cost = 0
for task in tasks:
    result = run_tiered_agent(task[:80] + "..." if len(task) > 80 else task)
    total_cost += result['estimated_cost_usd']
    print(f"  Cost: ${result['estimated_cost_usd']:.6f} | Model: {result['model']}\n")

print(f"Total: ${total_cost:.6f}")

import anthropic
from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict
import json

@dataclass
class UsageTracker:
    """
    Track token usage and costs per model, per user, per day.
    Alerts when approaching budget limits.
    """
    daily_budget_usd: float = 10.0  # Total daily budget
    per_user_daily_limit: float = 1.0  # Per-user daily limit

    # Pricing per million tokens (approximate, early 2026)
    PRICING = {
        'claude-opus-4-6':            {'input': 15.0,  'output': 75.0},
        'claude-sonnet-4-6':           {'input': 3.0,   'output': 15.0},
        'claude-haiku-4-5-20251001':  {'input': 0.80,  'output': 4.0},
    }

    _log: list = field(default_factory=list)

    def track(self, model: str, usage: anthropic.types.Usage,
              user_id: str = "default", task_type: str = "unknown") -> float:
        """Record usage and return the cost of this call."""
        prices = self.PRICING.get(model, {'input': 3.0, 'output': 15.0})

        # Account for prompt caching if available
        cache_read = getattr(usage, 'cache_read_input_tokens', 0)
        cache_write = getattr(usage, 'cache_creation_input_tokens', 0)
        regular_input = usage.input_tokens - cache_read - cache_write

        cache_read_price = prices['input'] * 0.10  # Cache reads at 10%
        cost = (
            regular_input / 1_000_000 * prices['input'] +
            cache_write   / 1_000_000 * prices['input'] +
            cache_read    / 1_000_000 * cache_read_price +
            usage.output_tokens / 1_000_000 * prices['output']
        )

        entry = {
            'timestamp': datetime.now().isoformat(),
            'user_id': user_id,
            'model': model,
            'task_type': task_type,
            'input_tokens': usage.input_tokens,
            'output_tokens': usage.output_tokens,
            'cache_read_tokens': cache_read,
            'cost_usd': cost
        }
        self._log.append(entry)

        # Check budget after each call
        self._check_alerts(user_id, cost)
        return cost

    def today_total(self) -> float:
        """Total spend today across all users."""
        today = datetime.now().date().isoformat()
        return sum(e['cost_usd'] for e in self._log if e['timestamp'][:10] == today)

    def today_by_user(self) -> dict:
        """Today's spend grouped by user."""
        today = datetime.now().date().isoformat()
        totals = defaultdict(float)
        for e in self._log:
            if e['timestamp'][:10] == today:
                totals[e['user_id']] += e['cost_usd']
        return dict(totals)

    def today_by_model(self) -> dict:
        """Today's spend grouped by model."""
        today = datetime.now().date().isoformat()
        totals = defaultdict(float)
        for e in self._log:
            if e['timestamp'][:10] == today:
                totals[e['model']] += e['cost_usd']
        return dict(totals)

    def _check_alerts(self, user_id: str, new_cost: float):
        """Fire alerts when approaching budget limits."""
        daily = self.today_total()
        user_daily = self.today_by_user().get(user_id, 0)

        if daily > self.daily_budget_usd * 0.80:
            print(f"[BUDGET WARNING] Daily spend at ${daily:.2f} / "
                  f"${self.daily_budget_usd:.2f} ({daily/self.daily_budget_usd:.0%})")

        if daily > self.daily_budget_usd:
            raise RuntimeError(
                f"[BUDGET EXCEEDED] Daily budget of ${self.daily_budget_usd:.2f} reached. "
                f"Current: ${daily:.2f}. Set a higher limit in UsageTracker or "
                f"add a hard spending limit at console.anthropic.com"
            )

        if user_daily > self.per_user_daily_limit:
            print(f"[USER LIMIT] User '{user_id}' has spent ${user_daily:.4f} today "
                  f"(limit: ${self.per_user_daily_limit:.2f})")

    def report(self) -> str:
        """Print a summary of today's usage."""
        lines = [
            f"=== Usage Report ({datetime.now().date()}) ===",
            f"Total spend:     ${self.today_total():.4f} / ${self.daily_budget_usd:.2f}",
            "",
            "By model:",
        ]
        for model, cost in sorted(self.today_by_model().items(), key=lambda x: -x[1]):
            short_model = model.split('-')[1] if '-' in model else model
            lines.append(f"  {short_model:12} ${cost:.4f}")
        lines.extend(["", "Top users:"])
        for user, cost in sorted(self.today_by_user().items(), key=lambda x: -x[1])[:5]:
            lines.append(f"  {user:20} ${cost:.4f}")
        return "\n".join(lines)


# Usage in your agent:
tracker = UsageTracker(daily_budget_usd=50.0)

response = client.messages.create(model="claude-sonnet-4-6", ...)
cost = tracker.track(
    model="claude-sonnet-4-6",
    usage=response.usage,
    user_id="user_12345",
    task_type="research_agent"
)
print(f"This call cost: ${cost:.6f}")
print(tracker.report())