Cost Optimization Strategies

Your AI shopping assistant is a hit. Customers love it. Your CFO does not. The monthly LLM bill just crossed $80,000, and traffic is still growing.

Let's learn how to cut costs without cutting quality.

Understanding AI Agent Costs

Cost Breakdown Example

Let's analyze a real e-commerce agent:

# Daily traffic: 100,000 conversations
# Average turns per conversation: 4
# Total LLM calls: 400,000/day

# Cost calculation (GPT-4 pricing)
daily_calls = 400_000
avg_input_tokens = 2000  # System prompt + context + history
avg_output_tokens = 300

input_cost = daily_calls * avg_input_tokens * (0.03 / 1000)   # $24,000/day
output_cost = daily_calls * avg_output_tokens * (0.06 / 1000) # $7,200/day

total_daily = input_cost + output_cost  # $31,200/day
total_monthly = total_daily * 30        # $936,000/month 😱

Strategy 1: Semantic Caching

Cache responses for similar queries, not just identical ones:

import numpy as np
from typing import Optional, Tuple
import hashlib

class SemanticCache:
    def __init__(
        self,
        embedding_model: str = "text-embedding-3-small",
        similarity_threshold: float = 0.92,
        redis_client=None
    ):
        self.embedder = EmbeddingClient(model=embedding_model)
        self.threshold = similarity_threshold
        self.redis = redis_client
        self.index = VectorIndex()  # FAISS or similar

    async def get(
        self,
        query: str,
        context_hash: str
    ) -> Optional[CachedResponse]:
        """Check cache for semantically similar query"""

        # 1. Generate embedding for current query
        query_embedding = await self.embedder.embed(query)

        # 2. Search for similar queries
        results = self.index.search(
            vector=query_embedding,
            filter={"context_hash": context_hash},
            top_k=1
        )

        if not results:
            return None

        best_match = results[0]

        # 3. Check similarity threshold
        if best_match.similarity >= self.threshold:
            # 4. Retrieve cached response
            cached = await self.redis.get(f"response:{best_match.id}")
            if cached:
                return CachedResponse(
                    response=cached,
                    similarity=best_match.similarity,
                    original_query=best_match.query
                )

        return None

    async def set(
        self,
        query: str,
        response: str,
        context_hash: str,
        ttl: int = 3600
    ):
        """Cache a response with its query embedding"""

        # Generate embedding
        embedding = await self.embedder.embed(query)

        # Store in vector index
        cache_id = hashlib.md5(f"{query}:{context_hash}".encode()).hexdigest()
        self.index.add(
            id=cache_id,
            vector=embedding,
            metadata={"context_hash": context_hash, "query": query}
        )

        # Store response
        await self.redis.setex(f"response:{cache_id}", ttl, response)


class CachedAgent:
    def __init__(self, agent, cache: SemanticCache):
        self.agent = agent
        self.cache = cache

    async def handle(self, query: str, context: dict) -> str:
        # Create context hash (user type, page, etc.)
        context_hash = self._hash_relevant_context(context)

        # Check cache
        cached = await self.cache.get(query, context_hash)
        if cached:
            self.metrics.record_cache_hit()
            return cached.response

        # Cache miss - call agent
        self.metrics.record_cache_miss()
        response = await self.agent.handle(query, context)

        # Cache the response
        await self.cache.set(query, response, context_hash)
        return response

    def _hash_relevant_context(self, context: dict) -> str:
        """Hash only context that affects the response"""
        relevant = {
            "user_tier": context.get("user_tier"),
            "has_active_order": context.get("has_active_order"),
            "product_category": context.get("current_category"),
        }
        return hashlib.md5(json.dumps(relevant, sort_keys=True).encode()).hexdigest()

Strategy 2: Model Routing

Use the cheapest model that can handle each query:

from enum import Enum

class ModelTier(Enum):
    MINI = "gpt-4o-mini"      # $0.15/1M input, $0.60/1M output
    STANDARD = "gpt-4o"        # $2.50/1M input, $10/1M output
    PREMIUM = "gpt-4-turbo"    # $10/1M input, $30/1M output

class CostOptimizedRouter:
    """Route queries to the cheapest capable model"""

    def __init__(self):
        self.classifier = QueryComplexityClassifier()

    async def select_model(
        self,
        query: str,
        context: dict
    ) -> Tuple[ModelTier, str]:

        complexity = await self.classifier.analyze(query, context)

        # Simple, pattern-matchable queries
        if complexity.type == "faq":
            return ModelTier.MINI, self._get_mini_prompt(query)

        # Single-tool queries (order status, product lookup)
        if complexity.type == "single_tool" and complexity.confidence > 0.85:
            return ModelTier.MINI, self._get_tool_prompt(query, complexity.tool)

        # Multi-step reasoning or ambiguous queries
        if complexity.type == "reasoning" or complexity.confidence < 0.7:
            return ModelTier.STANDARD, self._get_full_prompt(query)

        # Complex negotiations, complaints, or edge cases
        if complexity.type == "sensitive" or complexity.requires_judgment:
            return ModelTier.PREMIUM, self._get_premium_prompt(query)

        # Default to standard
        return ModelTier.STANDARD, self._get_full_prompt(query)


class QueryComplexityClassifier:
    """Fast classification of query complexity"""

    # Patterns that can be handled by mini model
    SIMPLE_PATTERNS = {
        "faq": [
            r"(what|how).*(return|refund) policy",
            r"(what|how).*(shipping|delivery) (time|cost|fee)",
            r"do you (ship|deliver) to",
            r"(what|where).*(store|location|hour)",
            r"(how|can i) (contact|reach|call)",
        ],
        "single_tool": [
            r"(where|what|track).*(my order|order status|package)",
            r"(show|find|search).*(product|item)",
            r"(what|check).*(availability|in stock)",
        ],
    }

    async def analyze(self, query: str, context: dict) -> QueryComplexity:
        query_lower = query.lower()

        # Check simple patterns first (no LLM needed)
        for qtype, patterns in self.SIMPLE_PATTERNS.items():
            for pattern in patterns:
                if re.search(pattern, query_lower):
                    return QueryComplexity(
                        type=qtype,
                        tool=self._extract_tool(pattern, qtype),
                        confidence=0.9,
                        requires_judgment=False
                    )

        # Check for sensitive indicators
        sensitive_keywords = [
            "complain", "angry", "upset", "refund", "cancel",
            "broken", "damaged", "wrong", "legal", "lawyer"
        ]
        if any(kw in query_lower for kw in sensitive_keywords):
            return QueryComplexity(
                type="sensitive",
                confidence=0.85,
                requires_judgment=True
            )

        # Use fast model for ambiguous classification
        return await self._llm_classify(query, context)

Cost Comparison by Model

Strategy 3: Prompt Optimization

Reduce token count without losing capability:

class PromptOptimizer:
    """Optimize prompts to reduce token usage"""

    def optimize_system_prompt(self, full_prompt: str) -> str:
        """Compress system prompt while maintaining effectiveness"""

        # Original: 2000 tokens
        original = """
        You are a helpful customer service assistant for ShopMart, an online
        e-commerce platform. Your role is to assist customers with their
        questions about products, orders, returns, and general inquiries.

        When helping customers:
        - Always be polite and professional
        - Provide accurate information about products and policies
        - If you don't know something, say so honestly
        - Never make up information about orders or inventory
        ...
        """

        # Optimized: 400 tokens (80% reduction)
        optimized = """
        Role: ShopMart customer assistant
        Rules:
        - Accurate info only, admit unknowns
        - Professional, helpful tone
        - Use tools for: orders, inventory, returns
        - Escalate: complaints, refunds >$100, shipping issues
        Format: Concise, actionable responses
        """

        return optimized

    def optimize_context(self, context: dict, query: str) -> dict:
        """Include only relevant context based on query"""

        # Analyze what context is needed
        needed = self._analyze_context_needs(query)

        optimized = {}

        if needed.needs_order_history:
            # Only recent relevant orders, not full history
            optimized["recent_orders"] = self._summarize_orders(
                context.get("orders", []),
                limit=3
            )

        if needed.needs_product_context:
            # Just current product, not browsing history
            optimized["current_product"] = context.get("current_product")

        if needed.needs_user_profile:
            # Minimal user info
            optimized["user"] = {
                "tier": context["user"]["tier"],
                "has_active_return": context["user"].get("has_active_return")
            }

        return optimized

    def _summarize_orders(self, orders: list, limit: int) -> list:
        """Compress order history to essential fields"""
        return [
            {
                "id": o["id"],
                "date": o["date"],
                "status": o["status"],
                "total": o["total"],
                "items_count": len(o["items"])
            }
            for o in orders[:limit]
        ]

Strategy 4: Batching and Streaming

Process multiple requests efficiently:

import asyncio
from typing import List

class BatchProcessor:
    """Batch similar requests for efficiency"""

    def __init__(self, batch_size: int = 10, max_wait_ms: int = 100):
        self.batch_size = batch_size
        self.max_wait = max_wait_ms / 1000
        self.pending = []
        self.lock = asyncio.Lock()

    async def process(self, request: AgentRequest) -> str:
        """Add to batch and wait for result"""

        future = asyncio.Future()

        async with self.lock:
            self.pending.append((request, future))

            if len(self.pending) >= self.batch_size:
                batch = self.pending
                self.pending = []
                asyncio.create_task(self._process_batch(batch))

        # Wait for result with timeout
        try:
            return await asyncio.wait_for(future, timeout=30)
        except asyncio.TimeoutError:
            return await self._process_single(request)

    async def _process_batch(self, batch: List[Tuple[AgentRequest, asyncio.Future]]):
        """Process batch in single LLM call where possible"""

        # Group by query type for batch processing
        groups = self._group_by_type(batch)

        for query_type, items in groups.items():
            if query_type in ["faq", "simple"]:
                # These can be batched into single prompt
                results = await self._batch_simple_queries(items)
            else:
                # Process individually
                results = await asyncio.gather(
                    *[self._process_single(req) for req, _ in items]
                )

            for (req, future), result in zip(items, results):
                future.set_result(result)

    async def _batch_simple_queries(self, items) -> List[str]:
        """Batch multiple simple queries into one LLM call"""

        queries = [req.query for req, _ in items]

        batch_prompt = f"""Answer each query concisely:

{chr(10).join(f'{i+1}. {q}' for i, q in enumerate(queries))}

Respond with numbered answers matching the queries."""

        response = await self.llm.complete(batch_prompt)
        return self._parse_batch_response(response, len(queries))

Strategy 5: Cost Monitoring & Alerts

Track spending in real-time:

from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class CostAlert:
    threshold_daily: float = 5000
    threshold_hourly: float = 300
    threshold_per_conversation: float = 0.50

class CostMonitor:
    def __init__(self, alerts: CostAlert):
        self.alerts = alerts
        self.costs = TimeSeriesStore()

    async def record_usage(
        self,
        conversation_id: str,
        model: str,
        input_tokens: int,
        output_tokens: int
    ):
        cost = self._calculate_cost(model, input_tokens, output_tokens)

        await self.costs.record(
            timestamp=datetime.now(),
            conversation_id=conversation_id,
            cost=cost,
            model=model,
            tokens=input_tokens + output_tokens
        )

        # Check alerts
        await self._check_alerts(conversation_id, cost)

    async def _check_alerts(self, conversation_id: str, cost: float):
        # Per-conversation alert
        conv_total = await self.costs.sum_by_conversation(conversation_id)
        if conv_total > self.alerts.threshold_per_conversation:
            await self._alert(
                "high_conversation_cost",
                f"Conversation {conversation_id} cost ${conv_total:.2f}"
            )

        # Hourly rate alert
        hourly = await self.costs.sum_last_hour()
        if hourly > self.alerts.threshold_hourly:
            await self._alert(
                "hourly_threshold",
                f"Hourly spend ${hourly:.2f} exceeds ${self.alerts.threshold_hourly}"
            )

    async def get_cost_dashboard(self) -> CostDashboard:
        return CostDashboard(
            today=await self.costs.sum_today(),
            this_week=await self.costs.sum_this_week(),
            by_model=await self.costs.group_by_model(),
            by_query_type=await self.costs.group_by_type(),
            top_expensive_conversations=await self.costs.top_conversations(10),
            projected_monthly=await self._project_monthly()
        )

Real-World Results

Key Takeaways

Next up: Monitoring & Observability - How to see what your agents are doing in production.