Task complexity assessment → Match lowest-capability model → Verify quality → Deploy

Simple tasks (classification/extraction/formatting) → nano/flash tier
Medium tasks (content generation/translation) → mini/sonnet tier
Complex tasks (reasoning/analysis/creation) → standard models
Critical tasks (code review/decisions) → flagship models

# ❌ Verbose system prompt (~450 tokens)
system_bad = """
You are a very professional and experienced customer service representative.
You need to answer various questions from users in a friendly and patient manner.
Please ensure your answers are accurate, complete, and easy to understand.
If you are not sure about the user's question, please honestly inform them...
"""

# ✅ Concise version (~120 tokens, saves 73%)
system_good = "You are a customer service rep. Answer questions friendly and accurately. Be honest when unsure."

# ❌ Free-form output (500+ tokens)
prompt_bad = "Analyze the sentiment of this text and explain your reasoning in detail"

# ✅ JSON output specified (~50 tokens)
prompt_good = """Analyze sentiment, return JSON:
{"sentiment": "positive|negative|neutral", "confidence": 0.0-1.0}
Text: {text}"""

# ❌ 5 full examples (~2000 tokens)
# ✅ 2 concise examples + 1 edge case (~600 tokens)
# Saves 70% of example tokens with near-zero quality loss

import tiktoken

def compress_prompt(prompt: str, max_tokens: int = 500) -> str:
    """Auto-truncate low-priority sections when prompt exceeds threshold"""
    enc = tiktoken.encoding_for_model("gpt-5")
    tokens = enc.encode(prompt)
    if len(tokens) <= max_tokens:
        return prompt
    return enc.decode(tokens[:max_tokens])

import anthropic

client = anthropic.Anthropic()

# Define cacheable content (typically long system prompts or documents)
system_content = [
    {
        "type": "text",
        "text": "Your long system prompt or knowledge base content here...",
        "cache_control": {"type": "ephemeral"}  # Mark as cacheable
    }
]

# First request: full pricing
response1 = client.messages.create(
    model="claude-sonnet-4-20250514",
    system=system_content,
    messages=[{"role": "user", "content": "Question 1"}],
    max_tokens=1024
)

# Subsequent requests: cache hit — input tokens billed at 90% discount
response2 = client.messages.create(
    model="claude-sonnet-4-20250514",
    system=system_content,
    messages=[{"role": "user", "content": "Question 2"}],
    max_tokens=1024
)

from openai import OpenAI
client = OpenAI()

# OpenAI automatically caches requests with identical prefixes
# When multiple requests share the same system message, automatic 50% discount
response = client.chat.completions.create(
    model="gpt-5",
    messages=[
        {"role": "system", "content": "Long system prompt... (auto-cached)"},
        {"role": "user", "content": "User question"}
    ]
)

from openai import OpenAI
client = OpenAI()

# Prepare batch request file (JSONL format)
batch_requests = [
    {
        "custom_id": "task-001",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-5-mini",
            "messages": [{"role": "user", "content": "Summarize this text: ..."}],
            "max_tokens": 500
        }
    },
    # ... more requests
]

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

# Upload and create Batch job
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="/v1/chat/completions",
    completion_window="24h"
)

print(f"Batch ID: {batch_job.id}, Status: {batch_job.status}")
# Completes within 24 hours with 50% discount

import anthropic

client = anthropic.Anthropic()

batch = client.batches.create(
    requests=[
        {
            "custom_id": "task-001",
            "params": {
                "model": "claude-haiku-4-20250514",
                "max_tokens": 1024,
                "messages": [{"role": "user", "content": "Translate to Chinese: ..."}]
            }
        }
        # ... more requests
    ]
)

import tiktoken

def count_tokens(text: str, model: str = "gpt-5") -> int:
    """Count tokens in text"""
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

def estimate_cost(input_tokens: int, output_tokens: int, model: str) -> float:
    """Estimate API call cost"""
    pricing = {
        "gpt-5":           {"input": 5.00, "output": 15.00},
        "gpt-5-mini":      {"input": 0.80, "output": 2.40},
        "gpt-5-nano":      {"input": 0.15, "output": 0.45},
        "claude-sonnet-4": {"input": 2.00, "output": 10.00},
        "claude-haiku-4":  {"input": 0.50, "output": 2.50},
        "deepseek-v3":     {"input": 0.14, "output": 0.28},
    }
    p = pricing.get(model, pricing["gpt-5-mini"])
    return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000

# Prometheus + Grafana monitoring setup
from prometheus_client import Counter, Histogram, start_http_server

TOKEN_USAGE = Counter('llm_tokens_total', 'Total tokens used', ['model', 'type'])
API_COST = Counter('llm_cost_dollars', 'Total API cost in dollars', ['model'])
API_LATENCY = Histogram('llm_latency_seconds', 'API call latency', ['model'])

def track_api_call(model: str, input_tok: int, output_tok: int,
                   latency: float, cost: float):
    TOKEN_USAGE.labels(model=model, type='input').inc(input_tok)
    TOKEN_USAGE.labels(model=model, type='output').inc(output_tok)
    API_COST.labels(model=model).inc(cost)
    API_LATENCY.labels(model=model).observe(latency)

import re
from enum import Enum

class TaskComplexity(Enum):
    SIMPLE = "simple"       # Classification, extraction, formatting
    MEDIUM = "medium"       # Translation, summarization, Q&A
    COMPLEX = "complex"     # Reasoning, analysis, creation
    CRITICAL = "critical"   # Code review, critical decisions

# Model routing mapping
MODEL_ROUTING = {
    TaskComplexity.SIMPLE:  "gpt-5-nano",        # $0.15/M input
    TaskComplexity.MEDIUM:  "gpt-5-mini",         # $0.80/M input
    TaskComplexity.COMPLEX: "gpt-5",              # $5.00/M input
    TaskComplexity.CRITICAL:"gpt-5",              # $5.00/M input
}

# Simple keyword-based classifier (can also use LLM self-classification)
COMPLEXITY_KEYWORDS = {
    TaskComplexity.SIMPLE: ["classify", "extract", "format", "list", "tag"],
    TaskComplexity.MEDIUM: ["translate", "summarize", "explain", "answer"],
    TaskComplexity.COMPLEX: ["analyze", "reason", "compare", "evaluate", "design"],
    TaskComplexity.CRITICAL: ["review", "security", "decide", "architect"],
}

def classify_task(query: str) -> TaskComplexity:
    """Fast keyword-based classification"""
    for complexity, keywords in COMPLEXITY_KEYWORDS.items():
        if any(kw in query.lower() for kw in keywords):
            return complexity
    return TaskComplexity.MEDIUM  # Default

def route_request(query: str) -> str:
    """Route request to optimal model"""
    complexity = classify_task(query)
    return MODEL_ROUTING[complexity]

# Example
query = "Please translate this text to English"
model = route_request(query)  # → gpt-5-mini ($0.80/M)
# vs gpt-5 at $5.00/M = 84% savings

async def smart_classify(query: str) -> TaskComplexity:
    """Use gpt-5-nano for complexity classification — near-zero cost"""
    response = await client.chat.completions.create(
        model="gpt-5-nano",
        messages=[{
            "role": "user",
            "content": f"Classify this task as simple/medium/complex/critical:\n{query}\nReply with only the classification."
        }],
        max_tokens=10
    )
    label = response.choices[0].message.content.strip().lower()
    return TaskComplexity(label)

from openai import OpenAI

client = OpenAI()

def stream_response(prompt: str, model: str = "gpt-5-mini"):
    """Streaming output — 80% reduction in time-to-first-token"""
    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        max_tokens=1024
    )
    
    full_response = ""
    for chunk in stream:
        if chunk.choices[0].delta.content:
            token = chunk.choices[0].delta.content
            full_response += token
            print(token, end="", flush=True)
    
    return full_response

def break_even_analysis(
    few_shot_overhead_tokens: int,
    requests_per_month: int,
    model_input_price: float,
    fine_tune_cost: float,
    fine_tune_inference_surcharge: float
) -> dict:
    """Calculate fine-tuning break-even point"""
    
    few_shot_monthly = (few_shot_overhead_tokens * requests_per_month
                        * model_input_price) / 1_000_000
    
    ft_monthly = (fine_tune_cost / 12 +
                  fine_tune_inference_surcharge * requests_per_month / 1_000_000)
    
    months_to_break_even = fine_tune_cost / max(few_shot_monthly - ft_monthly, 0.01)
    
    return {
        "few_shot_monthly_cost": round(few_shot_monthly, 2),
        "fine_tune_monthly_cost": round(ft_monthly, 2),
        "monthly_savings": round(few_shot_monthly - ft_monthly, 2),
        "break_even_months": round(months_to_break_even, 1)
    }

# Example: 100K requests/month, 800 token few-shot overhead
result = break_even_analysis(
    few_shot_overhead_tokens=800,
    requests_per_month=100_000,
    model_input_price=0.80,
    fine_tune_cost=200,
    fine_tune_inference_surcharge=0.40
)
# → few_shot_monthly: $64, fine_tune_monthly: $20.67, break-even: 4.6 months

import hashlib
import json
import redis
from typing import Optional

class LLMResponseCache:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.default_ttl = 3600 * 24  # 24 hours
    
    def _make_key(self, model: str, messages: list, **kwargs) -> str:
        """Generate cache key"""
        content = json.dumps({
            "model": model,
            "messages": messages,
            **kwargs
        }, sort_keys=True)
        return f"llm:cache:{hashlib.sha256(content.encode()).hexdigest()}"
    
    def get(self, model: str, messages: list, **kwargs) -> Optional[str]:
        """Query cache"""
        key = self._make_key(model, messages, **kwargs)
        result = self.redis.get(key)
        return result.decode() if result else None
    
    def set(self, model: str, messages: list, response: str,
            ttl: int = None, **kwargs):
        """Write to cache"""
        key = self._make_key(model, messages, **kwargs)
        self.redis.setex(key, ttl or self.default_ttl, response)

# Usage example
cache = LLMResponseCache()

def call_with_cache(messages: list, model: str = "gpt-5-mini", **kwargs):
    """API call with caching"""
    # 1. Check cache
    cached = cache.get(model, messages, **kwargs)
    if cached:
        return {"content": cached, "source": "cache", "cost": 0}
    
    # 2. Call API
    response = client.chat.completions.create(
        model=model, messages=messages, **kwargs
    )
    result = response.choices[0].message.content
    
    # 3. Write to cache
    cache.set(model, messages, result, **kwargs)
    return {"content": result, "source": "api", "cost": response.usage}

# Simply replace base_url to connect to XiDao Gateway
from openai import OpenAI

client = OpenAI(
    api_key="your-xidao-api-key",
    base_url="https://api.xidao.online/v1"  # XiDao Gateway
)

# Call any model with unified metering
response = client.chat.completions.create(
    model="gpt-5-mini",  # Also works with claude-sonnet-4, gemini-2.5-pro, etc.
    messages=[{"role": "user", "content": "Hello"}],
    extra_headers={
        "X-Team": "backend",       # Team tag
        "X-Project": "chatbot",    # Project tag
        "X-Budget-Limit": "100"    # Per-request budget cap (USD)
    }
)

# View real-time usage
# GET https://api.xidao.online/dashboard/costs?team=backend&period=month