Cost Optimization for LLM Applications

import tiktoken
from openai import OpenAI
from dataclasses import dataclass
from datetime import datetime
import json

# Pricing per 1M tokens (as of early 2026 — check current pricing)
MODEL_PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
    "claude-sonnet-4-20250514": {"input": 3.00, "output": 15.00},
    "claude-haiku-3.5": {"input": 0.80, "output": 4.00},
}

@dataclass
class UsageRecord:
    timestamp: datetime
    model: str
    input_tokens: int
    output_tokens: int
    cached_tokens: int
    cost_usd: float
    user_id: str
    endpoint: str

def calculate_cost(model: str, input_tokens: int, output_tokens: int,
                   cached_tokens: int = 0) -> float:
    """Calculate the cost of a single LLM request."""
    pricing = MODEL_PRICING.get(model)
    if not pricing:
        raise ValueError(f"Unknown model: {model}")

    # Cached tokens are typically 50% cheaper (OpenAI) or 90% cheaper (Anthropic)
    regular_input = input_tokens - cached_tokens
    cached_discount = 0.5  # Adjust per provider

    input_cost = (regular_input * pricing["input"] / 1_000_000 +
                  cached_tokens * pricing["input"] * cached_discount / 1_000_000)
    output_cost = output_tokens * pricing["output"] / 1_000_000

    return round(input_cost + output_cost, 6)


def track_llm_usage(response, model: str, user_id: str, endpoint: str) -> UsageRecord:
    """Extract usage from an OpenAI response and log it."""
    usage = response.usage

    record = UsageRecord(
        timestamp=datetime.utcnow(),
        model=model,
        input_tokens=usage.prompt_tokens,
        output_tokens=usage.completion_tokens,
        cached_tokens=getattr(usage, "prompt_tokens_details", {}).get(
            "cached_tokens", 0
        ) if hasattr(usage, "prompt_tokens_details") else 0,
        cost_usd=calculate_cost(
            model, usage.prompt_tokens, usage.completion_tokens
        ),
        user_id=user_id,
        endpoint=endpoint,
    )

    # Log to your analytics pipeline
    print(json.dumps({
        "event": "llm_usage",
        "model": record.model,
        "input_tokens": record.input_tokens,
        "output_tokens": record.output_tokens,
        "cost_usd": record.cost_usd,
        "user_id": record.user_id,
        "endpoint": record.endpoint,
    }))

    return record

from openai import OpenAI

client = OpenAI()

# This system prompt is 1,500+ tokens. After the first request,
# subsequent requests with the same prefix get cached pricing.
SYSTEM_PROMPT = """You are a senior financial analyst assistant.
You help users analyze quarterly earnings reports, SEC filings,
and market data. You follow these rules:

1. Always cite specific numbers from the provided documents.
2. When comparing periods, use percentage changes.
3. Flag any inconsistencies between reported numbers.
4. Use professional financial terminology.
5. Format currency as $X.XM or $X.XB.
... (imagine 1,500 tokens of detailed instructions) ..."""

def analyze_earnings(user_query: str, context: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {user_query}"},
        ],
    )

    # Check if caching kicked in
    if hasattr(response.usage, "prompt_tokens_details"):
        cached = response.usage.prompt_tokens_details.cached_tokens
        total = response.usage.prompt_tokens
        print(f"Cache hit: {cached}/{total} tokens ({cached/total*100:.0f}%)")

    return response.choices[0].message.content

import anthropic

client = anthropic.Anthropic()

# Mark the system prompt for caching
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "You are a senior financial analyst assistant...(long prompt)...",
            "cache_control": {"type": "ephemeral"}  # Cache this block
        }
    ],
    messages=[
        {"role": "user", "content": "Analyze Q3 revenue trends."}
    ],
)

# Anthropic returns cache stats directly
print(f"Cache read tokens: {response.usage.cache_read_input_tokens}")
print(f"Cache creation tokens: {response.usage.cache_creation_input_tokens}")
# Cache read tokens cost 90% less than regular input tokens on Anthropic

import hashlib
import json
import time
import numpy as np
import redis
from openai import OpenAI

client = OpenAI()
cache = redis.Redis(host="localhost", port=6379, db=0)

SIMILARITY_THRESHOLD = 0.95  # Cosine similarity — tune this carefully
CACHE_TTL = 3600  # 1 hour

def get_embedding(text: str) -> list[float]:
    """Get embedding for a text string."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text,
    )
    return response.data[0].embedding

def cosine_similarity(a: list[float], b: list[float]) -> float:
    """Calculate cosine similarity between two vectors."""
    a_np, b_np = np.array(a), np.array(b)
    return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))

class SemanticCache:
    def __init__(self, redis_client: redis.Redis, threshold: float = 0.95):
        self.redis = redis_client
        self.threshold = threshold

    def _cache_key(self, prefix: str, query_hash: str) -> str:
        return f"sem_cache:{prefix}:{query_hash}"

    def get(self, query: str, namespace: str = "default") -> str | None:
        """Look up a semantically similar cached response."""
        query_embedding = get_embedding(query)

        # Get all cached embeddings in this namespace
        pattern = f"sem_cache:{namespace}:*"
        keys = self.redis.keys(pattern)

        best_match = None
        best_similarity = 0.0

        for key in keys:
            cached = json.loads(self.redis.get(key))
            similarity = cosine_similarity(query_embedding, cached["embedding"])

            if similarity > best_similarity:
                best_similarity = similarity
                best_match = cached

        if best_match and best_similarity >= self.threshold:
            return best_match["response"]

        return None

    def set(self, query: str, response: str, namespace: str = "default"):
        """Cache a query-response pair with its embedding."""
        embedding = get_embedding(query)
        query_hash = hashlib.md5(query.encode()).hexdigest()

        data = {
            "query": query,
            "response": response,
            "embedding": embedding,
            "timestamp": time.time(),
        }

        key = self._cache_key(namespace, query_hash)
        self.redis.setex(key, CACHE_TTL, json.dumps(data))


# Usage
sem_cache = SemanticCache(cache, threshold=0.95)

def ask_with_cache(query: str) -> str:
    # Check semantic cache first
    cached = sem_cache.get(query, namespace="support")
    if cached:
        print("Semantic cache hit!")
        return cached

    # Cache miss — call the LLM
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a helpful support agent."},
            {"role": "user", "content": query},
        ],
        temperature=0,  # Deterministic — safe to cache
    )

    result = response.choices[0].message.content
    sem_cache.set(query, result, namespace="support")
    return result

import hashlib
import json
import redis
from openai import OpenAI

client = OpenAI()
cache = redis.Redis(host="localhost", port=6379, db=0)

def cached_llm_call(
    messages: list[dict],
    model: str = "gpt-4o",
    temperature: float = 0,
    cache_ttl: int = 3600,
) -> str:
    """LLM call with exact-match response caching."""
    if temperature > 0:
        # Non-deterministic — skip cache
        response = client.chat.completions.create(
            model=model, messages=messages, temperature=temperature
        )
        return response.choices[0].message.content

    # Build a cache key from the full request
    cache_input = json.dumps({
        "model": model,
        "messages": messages,
        "temperature": temperature,
    }, sort_keys=True)
    cache_key = f"llm:exact:{hashlib.sha256(cache_input.encode()).hexdigest()}"

    # Check cache
    cached = cache.get(cache_key)
    if cached:
        return cached.decode("utf-8")

    # Cache miss
    response = client.chat.completions.create(
        model=model, messages=messages, temperature=temperature
    )
    result = response.choices[0].message.content

    cache.setex(cache_key, cache_ttl, result)
    return result

from openai import OpenAI
from enum import Enum

client = OpenAI()

class ModelTier(Enum):
    CHEAP = "gpt-4.1-nano"      # $0.10/1M input
    MEDIUM = "gpt-4.1-mini"     # $0.40/1M input
    EXPENSIVE = "gpt-4.1"       # $2.00/1M input

# Cost ratios: CHEAP is 20x cheaper than EXPENSIVE

def classify_query_complexity(query: str) -> ModelTier:
    """Use a cheap model to classify query complexity."""
    response = client.chat.completions.create(
        model="gpt-4.1-nano",  # Classification itself uses the cheapest model
        messages=[
            {
                "role": "system",
                "content": """Classify the user query complexity.
Reply with ONLY one word: SIMPLE, MEDIUM, or COMPLEX.

SIMPLE: Factual lookups, yes/no questions, greetings, simple formatting.
MEDIUM: Summaries, comparisons, explanations of concepts, multi-step but straightforward.
COMPLEX: Analysis, creative writing, code generation, multi-document reasoning, nuanced judgment.""",
            },
            {"role": "user", "content": query},
        ],
        max_tokens=10,
        temperature=0,
    )

    classification = response.choices[0].message.content.strip().upper()

    mapping = {
        "SIMPLE": ModelTier.CHEAP,
        "MEDIUM": ModelTier.MEDIUM,
        "COMPLEX": ModelTier.EXPENSIVE,
    }
    return mapping.get(classification, ModelTier.MEDIUM)


def route_and_respond(query: str, system_prompt: str) -> tuple[str, str]:
    """Route query to appropriate model tier and get response."""
    tier = classify_query_complexity(query)
    model = tier.value

    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": query},
        ],
    )

    return response.choices[0].message.content, model


# Example usage
answer, model_used = route_and_respond(
    "What are your business hours?",
    "You are a helpful customer support agent for Acme Corp."
)
print(f"Answered by {model_used}: {answer}")

def route_with_fallback(query: str, system_prompt: str) -> str:
    """Try cheap model first, escalate if confidence is low."""
    # First attempt with cheap model
    response = client.chat.completions.create(
        model=ModelTier.CHEAP.value,
        messages=[
            {"role": "system", "content": system_prompt + "\nEnd your response with CONFIDENCE: HIGH or CONFIDENCE: LOW."},
            {"role": "user", "content": query},
        ],
    )

    result = response.choices[0].message.content

    if "CONFIDENCE: LOW" in result:
        # Escalate to expensive model
        response = client.chat.completions.create(
            model=ModelTier.EXPENSIVE.value,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": query},
            ],
        )
        return response.choices[0].message.content

    # Strip the confidence tag and return
    return result.replace("CONFIDENCE: HIGH", "").strip()

# BEFORE: 850 tokens
VERBOSE_PROMPT = """
You are a helpful customer support assistant for TechCorp, a leading
technology solutions provider. Your role is to assist customers with
their inquiries about our products and services. When responding to
customers, you should always be polite, professional, and thorough.

When a customer asks about pricing, you should provide accurate pricing
information based on the current pricing tiers. If you do not know the
exact pricing, you should direct the customer to the pricing page on
our website at https://techcorp.com/pricing.

When a customer asks about technical issues, you should first try to
troubleshoot the issue by asking clarifying questions. If you cannot
resolve the issue, you should create a support ticket and provide the
customer with the ticket number.

Here are some examples of how to respond:
Customer: "What's the price of the Pro plan?"
You: "The Pro plan is $49/month when billed annually..."
(... 15 more examples ...)
"""

# AFTER: 180 tokens — same behavior
COMPRESSED_PROMPT = """TechCorp support agent. Be concise and accurate.

Rules:
- Pricing questions: give exact tier pricing. Unknown? Link to techcorp.com/pricing
- Technical issues: troubleshoot first, then create ticket with number
- Tone: professional, not verbose

Response format: direct answer, then next steps if any."""

from openai import OpenAI

client = OpenAI()

def summarize_history(
    messages: list[dict], keep_recent: int = 4
) -> list[dict]:
    """Summarize old messages, keep recent ones verbatim."""
    if len(messages) <= keep_recent:
        return messages

    old_messages = messages[:-keep_recent]
    recent_messages = messages[-keep_recent:]

    # Summarize old messages with a cheap model
    history_text = "\n".join(
        f"{m['role']}: {m['content']}" for m in old_messages
    )

    summary_response = client.chat.completions.create(
        model="gpt-4.1-nano",
        messages=[
            {
                "role": "system",
                "content": "Summarize this conversation in 2-3 sentences. Keep key facts, decisions, and unanswered questions.",
            },
            {"role": "user", "content": history_text},
        ],
        max_tokens=150,
    )

    summary = summary_response.choices[0].message.content

    return [
        {"role": "system", "content": f"Previous conversation summary: {summary}"},
        *recent_messages,
    ]


# Example: 20-turn conversation compressed to summary + last 4 turns
# Before: ~4,000 tokens of history
# After: ~600 tokens (150 summary + 4 recent messages)

from openai import OpenAI
import numpy as np

client = OpenAI()

# Pre-computed examples with embeddings
FEW_SHOT_EXAMPLES = [
    {
        "query": "I want to cancel my subscription",
        "response": "I can help with cancellation. Your current plan...",
        "embedding": [0.1, 0.2, ...],  # Pre-computed
        "category": "cancellation",
    },
    {
        "query": "The app crashes when I upload files",
        "response": "Let's troubleshoot the upload issue. First...",
        "embedding": [0.3, 0.4, ...],
        "category": "technical",
    },
    # ... 50 more examples
]

def select_examples(query: str, examples: list[dict], top_k: int = 2) -> list[dict]:
    """Select the most relevant few-shot examples for the query."""
    query_emb = get_embedding(query)  # Reuse from earlier

    scored = []
    for ex in examples:
        sim = cosine_similarity(query_emb, ex["embedding"])
        scored.append((sim, ex))

    scored.sort(key=lambda x: x[0], reverse=True)
    return [ex for _, ex in scored[:top_k]]


def build_prompt_with_selected_examples(query: str, system_base: str) -> list[dict]:
    """Build prompt with only the most relevant examples."""
    examples = select_examples(query, FEW_SHOT_EXAMPLES, top_k=2)

    examples_text = "\n\n".join(
        f"User: {ex['query']}\nAssistant: {ex['response']}"
        for ex in examples
    )

    return [
        {
            "role": "system",
            "content": f"{system_base}\n\nExamples:\n{examples_text}",
        },
        {"role": "user", "content": query},
    ]

import json
import time
from openai import OpenAI

client = OpenAI()

def create_batch_file(requests: list[dict], filename: str = "/tmp/batch_input.jsonl"):
    """Create a JSONL file for batch processing."""
    with open(filename, "w") as f:
        for i, req in enumerate(requests):
            batch_request = {
                "custom_id": f"request-{i}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": req.get("model", "gpt-4o-mini"),
                    "messages": req["messages"],
                    "max_tokens": req.get("max_tokens", 500),
                    "temperature": 0,
                },
            }
            f.write(json.dumps(batch_request) + "\n")

    return filename


def submit_batch(filename: str) -> str:
    """Upload file and submit batch job."""
    # Upload the JSONL file
    with open(filename, "rb") as f:
        file_obj = client.files.create(file=f, purpose="batch")

    # Create the batch
    batch = client.batches.create(
        input_file_id=file_obj.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",
    )

    print(f"Batch submitted: {batch.id}")
    return batch.id


def poll_batch(batch_id: str, interval: int = 30) -> list[dict]:
    """Poll until batch completes and return results."""
    while True:
        batch = client.batches.retrieve(batch_id)
        print(f"Status: {batch.status} | Completed: {batch.request_counts.completed}/{batch.request_counts.total}")

        if batch.status == "completed":
            # Download results
            result_file = client.files.content(batch.output_file_id)
            results = []
            for line in result_file.text.strip().split("\n"):
                results.append(json.loads(line))
            return results

        if batch.status in ("failed", "expired", "cancelled"):
            raise RuntimeError(f"Batch {batch.status}: {batch.errors}")

        time.sleep(interval)


# Example: Classify 10,000 support tickets at 50% off
tickets = [
    {"messages": [
        {"role": "system", "content": "Classify this ticket as: billing, technical, feature_request, or other. Reply with one word."},
        {"role": "user", "content": ticket_text},
    ]}
    for ticket_text in load_tickets()  # Your function to load tickets
]

batch_file = create_batch_file(tickets)
batch_id = submit_batch(batch_file)
results = poll_batch(batch_id)

# Cost: 10,000 tickets * ~200 tokens each * $0.15/1M (mini) * 0.5 (batch discount)
# = $0.15 instead of $0.30. At GPT-4o scale: $2.50 instead of $5.00

import tiktoken

def manage_context_window(
    messages: list[dict],
    system_prompt: str,
    max_context_tokens: int = 8000,
    model: str = "gpt-4o",
) -> list[dict]:
    """Keep conversation within token budget using sliding window + summarization."""
    enc = tiktoken.encoding_for_model(model)

    system_tokens = len(enc.encode(system_prompt))
    available = max_context_tokens - system_tokens - 500  # Reserve 500 for response

    # Start from most recent, add messages until budget exhausted
    selected = []
    token_count = 0

    for msg in reversed(messages):
        msg_tokens = len(enc.encode(msg["content"]))
        if token_count + msg_tokens > available:
            break
        selected.insert(0, msg)
        token_count += msg_tokens

    # If we dropped messages, add a summary of what was dropped
    if len(selected) < len(messages):
        dropped = messages[:len(messages) - len(selected)]
        summary = summarize_history(dropped, keep_recent=0)

        if summary:
            selected.insert(0, {
                "role": "system",
                "content": f"Earlier conversation summary: {summary[0]['content']}"
            })

    return [{"role": "system", "content": system_prompt}] + selected

# Strategy 1: Aggressive max_tokens
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    max_tokens=150,  # Force concise responses
)

# Strategy 2: Stop sequences for structured output
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "Answer in one paragraph. End with [DONE]."},
        {"role": "user", "content": query},
    ],
    stop=["[DONE]"],  # Stop generating as soon as [DONE] is produced
)

# Strategy 3: Instruct conciseness in the system prompt
CONCISE_PROMPT = """Answer the user's question in 2-3 sentences maximum.
If a longer answer is needed, provide the key point first, then ask
if the user wants more detail. Never repeat the question back."""

def needs_llm_response(query: str) -> bool:
    """Use a tiny model to check if the query even needs an LLM response."""
    response = client.chat.completions.create(
        model="gpt-4.1-nano",
        messages=[
            {
                "role": "system",
                "content": """Classify if this query needs an LLM response or can be handled
by a simple lookup/redirect. Reply YES or NO.

NO examples: "hi", "thanks", "ok", greetings, single word messages.
YES examples: questions, requests, complaints, anything needing a real answer.""",
            },
            {"role": "user", "content": query},
        ],
        max_tokens=5,
        temperature=0,
    )

    return "YES" in response.choices[0].message.content.upper()


def prefilter_rag_results(query: str, chunks: list[str]) -> list[str]:
    """Use a cheap model to filter irrelevant RAG chunks before the expensive call."""
    relevant = []

    for chunk in chunks:
        response = client.chat.completions.create(
            model="gpt-4.1-nano",
            messages=[
                {
                    "role": "system",
                    "content": "Is this document chunk relevant to the query? Reply YES or NO only.",
                },
                {
                    "role": "user",
                    "content": f"Query: {query}\n\nChunk: {chunk[:500]}",
                },
            ],
            max_tokens=5,
            temperature=0,
        )

        if "YES" in response.choices[0].message.content.upper():
            relevant.append(chunk)

    return relevant

import time
from collections import defaultdict
from dataclasses import dataclass, field

@dataclass
class CostTracker:
    """Track LLM costs with alerts and budget limits."""
    daily_budget: float = 100.0  # USD
    alert_threshold: float = 0.8  # Alert at 80% of budget

    _daily_costs: dict = field(default_factory=lambda: defaultdict(float))
    _request_counts: dict = field(default_factory=lambda: defaultdict(int))

    def record(self, cost: float, model: str, endpoint: str):
        today = time.strftime("%Y-%m-%d")
        self._daily_costs[today] += cost
        self._request_counts[f"{today}:{model}:{endpoint}"] += 1

        # Check budget
        if self._daily_costs[today] > self.daily_budget * self.alert_threshold:
            self._send_alert(
                f"LLM cost alert: ${self._daily_costs[today]:.2f} "
                f"({self._daily_costs[today]/self.daily_budget*100:.0f}% of daily budget)"
            )

        if self._daily_costs[today] > self.daily_budget:
            self._send_alert(f"BUDGET EXCEEDED: ${self._daily_costs[today]:.2f}")
            # Optionally: switch to cheaper models or enable aggressive caching
            return False  # Signal to caller that budget is exceeded

        return True

    def get_daily_report(self) -> dict:
        today = time.strftime("%Y-%m-%d")
        return {
            "date": today,
            "total_cost": self._daily_costs[today],
            "budget_remaining": self.daily_budget - self._daily_costs[today],
            "budget_utilization": f"{self._daily_costs[today]/self.daily_budget*100:.1f}%",
        }

    def _send_alert(self, message: str):
        # Integrate with Slack, PagerDuty, email, etc.
        print(f"ALERT: {message}")


# Global cost tracker
cost_tracker = CostTracker(daily_budget=200.0)

# Use it after every LLM call
def tracked_llm_call(messages, model="gpt-4o", **kwargs):
    response = client.chat.completions.create(
        model=model, messages=messages, **kwargs
    )
    cost = calculate_cost(
        model,
        response.usage.prompt_tokens,
        response.usage.completion_tokens,
    )
    within_budget = cost_tracker.record(cost, model, "chat")

    if not within_budget:
        # Fallback: switch to cheaper model for remaining requests
        print("Budget exceeded — switching to cheaper model")

    return response