AI Agent Economics: Unit Costs, ROI Frameworks, and Cost Optimization

Input: 500 tokens (prompt)
Output: 200 tokens (response)
Total: 700 tokens
Cost at GPT-4o: $0.0033

Input: 500 tokens (prompt)
Output: 5,000 tokens (reasoning chain + response)
Total: 5,500 tokens
Cost at o3: $0.225

Input: 2,000 tokens
Output: 50,000 tokens (deep reasoning)
Total: 52,000 tokens
Cost at o3: $2.02 per request

Turn 1: 500 input → 200 output = 700 tokens
Turn 2: 700 + 300 input → 250 output = 1,250 tokens
Turn 3: 1,250 + 400 input → 300 output = 1,950 tokens
Turn 4: 1,950 + 350 input → 400 output = 2,700 tokens
...
Turn 20: 15,000+ tokens per request

def agent_loop(task):
    context = [system_prompt]  # ~1,000 tokens

    while not done:
        # Each iteration
        response = llm.generate(context)      # Input: growing, Output: 200-500
        tool_result = execute_tool(response)  # Tool call
        context.append(response)
        context.append(tool_result)

        if task_complete(response):
            done = True

    return final_response

Coordinator Agent: Routes and synthesizes
├── Research Agent: Web search, document retrieval
├── Analysis Agent: Data processing, calculations
├── Writing Agent: Content generation
├── Code Agent: Implementation, debugging
└── Critic Agent: Quality evaluation

# MCTS with 64 candidates, depth 5
candidates_per_step = 64
depth = 5
tokens_per_candidate = 500

total_tokens = candidates_per_step * depth * tokens_per_candidate
# = 64 * 5 * 500 = 160,000 tokens per decision

class ModelCascade:
    def __init__(self):
        self.tiers = [
            {"model": "gpt-4o-mini", "threshold": 0.8, "cost": 0.00015},
            {"model": "gpt-4o", "threshold": 0.6, "cost": 0.005},
            {"model": "o3", "threshold": 0.0, "cost": 0.05},
        ]

    def route(self, query, complexity_score):
        for tier in self.tiers:
            if complexity_score >= tier["threshold"]:
                return tier["model"]
        return self.tiers[-1]["model"]

    def estimate_complexity(self, query):
        # Heuristics: length, keywords, domain
        indicators = {
            "math": 0.3,
            "code": 0.2,
            "reasoning": 0.3,
            "multi-step": 0.2,
        }
        score = sum(
            weight for keyword, weight in indicators.items()
            if keyword in query.lower()
        )
        return min(score, 1.0)

class ContextManager:
    def __init__(self, max_tokens=8000):
        self.max_tokens = max_tokens
        self.summarizer = SummarizerModel()

    def compress(self, messages):
        total_tokens = count_tokens(messages)

        if total_tokens <= self.max_tokens:
            return messages

        # Keep system prompt and last N turns
        system = messages[0]
        recent = messages[-4:]  # Last 2 turns

        # Summarize middle
        middle = messages[1:-4]
        summary = self.summarizer.summarize(middle)

        return [system, {"role": "system", "content": f"Previous context: {summary}"}] + recent

import hashlib
from functools import lru_cache

class LLMCache:
    def __init__(self, cache_backend):
        self.cache = cache_backend
        self.hits = 0
        self.misses = 0

    def get_cache_key(self, messages, model):
        content = json.dumps(messages, sort_keys=True)
        return hashlib.sha256(f"{model}:{content}".encode()).hexdigest()

    async def generate(self, messages, model):
        key = self.get_cache_key(messages, model)

        cached = await self.cache.get(key)
        if cached:
            self.hits += 1
            return cached

        self.misses += 1
        response = await self.llm.generate(messages, model)
        await self.cache.set(key, response, ttl=3600)
        return response

    @property
    def hit_rate(self):
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0

# Before: 847 tokens
system_prompt_verbose = """
You are an AI assistant designed to help users with their questions.
Your role is to provide helpful, accurate, and detailed responses.
When answering questions, please consider the following guidelines:
1. Be thorough but concise
2. Provide examples when helpful
3. Cite sources when possible
...
"""

# After: 156 tokens
system_prompt_optimized = """
Expert assistant. Be concise, accurate, cite sources.
Format: Brief answer, then details if needed.
"""

# OpenAI Batch API: 50% discount
async def batch_process(requests):
    batch = client.batches.create(
        input_file_id=upload_requests(requests),
        endpoint="/v1/chat/completions",
        completion_window="24h"
    )

    # Wait for completion (up to 24h)
    result = await wait_for_batch(batch.id)
    return result

# Cost comparison for 10,000 requests:
# Real-time: $500
# Batch: $250 (50% savings)

def calculate_breakeven(daily_tokens, api_cost_per_1m, server_cost_monthly):
    daily_api_cost = (daily_tokens / 1_000_000) * api_cost_per_1m
    monthly_api_cost = daily_api_cost * 30

    breakeven_days = server_cost_monthly / daily_api_cost

    return {
        "monthly_api_cost": monthly_api_cost,
        "monthly_server_cost": server_cost_monthly,
        "savings": monthly_api_cost - server_cost_monthly,
        "breakeven_days": breakeven_days
    }

# Example: 50M tokens/day
result = calculate_breakeven(
    daily_tokens=50_000_000,
    api_cost_per_1m=5.0,  # GPT-4o average
    server_cost_monthly=3000  # 4x A100 cluster
)
# Monthly API: $7,500
# Monthly self-hosted: $3,000
# Savings: $4,500/month

Total Cost of Ownership (TCO) =
    Direct Costs + Infrastructure + Development + Operations

Direct Costs:
├── API tokens (variable)
├── Embedding costs
└── Vector DB queries

Infrastructure:
├── Compute (GPU/CPU)
├── Storage
├── Networking
└── Caching layer

Development:
├── Engineering time
├── Prompt engineering
├── Testing/evaluation
└── Integration work

Operations:
├── Monitoring
├── Maintenance
├── Error handling
└── Human-in-the-loop labor

class ROICalculator:
    def __init__(self, agent_system):
        self.agent = agent_system

    def calculate_roi(self, use_case):
        # Quantify current costs
        current_state = {
            "labor_hours": use_case.manual_hours_per_task,
            "labor_cost": use_case.hourly_rate,
            "error_rate": use_case.manual_error_rate,
            "throughput": use_case.tasks_per_day,
        }

        # Quantify with agent
        agent_state = {
            "labor_hours": use_case.manual_hours_per_task * 0.2,  # 80% reduction
            "labor_cost": use_case.hourly_rate,
            "agent_cost": self.agent.cost_per_task,
            "error_rate": use_case.manual_error_rate * 0.5,  # 50% fewer errors
            "throughput": use_case.tasks_per_day * 5,  # 5x throughput
        }

        # Calculate monthly impact
        monthly_tasks = current_state["throughput"] * 22  # Working days

        current_monthly_cost = (
            monthly_tasks * current_state["labor_hours"] * current_state["labor_cost"]
        )

        agent_monthly_cost = (
            monthly_tasks * agent_state["labor_hours"] * agent_state["labor_cost"] +
            monthly_tasks * agent_state["agent_cost"]
        )

        return {
            "monthly_savings": current_monthly_cost - agent_monthly_cost,
            "roi_percentage": (current_monthly_cost - agent_monthly_cost) / agent_monthly_cost * 100,
            "payback_months": use_case.implementation_cost / (current_monthly_cost - agent_monthly_cost)
        }

# Monthly calculation
tickets_per_month = 500 * 22  # 11,000 tickets

# Current costs
current_labor = 11000 * 0.25 * 25  # $68,750

# With agent
ai_handled = 11000 * 0.8  # 8,800 tickets
human_handled = 11000 * 0.2  # 2,200 tickets

agent_cost = ai_handled * 0.15  # $1,320
remaining_labor = human_handled * 0.25 * 25  # $13,750
total_new_cost = agent_cost + remaining_labor  # $15,070

# ROI
monthly_savings = 68750 - 15070  # $53,680
annual_savings = monthly_savings * 12  # $644,160
roi = (monthly_savings / 15070) * 100  # 356% ROI

class AgentMetrics:
    def __init__(self):
        self.metrics = {}

    def track(self, task_id, metrics):
        self.metrics[task_id] = {
            # Cost metrics
            "tokens_input": metrics.input_tokens,
            "tokens_output": metrics.output_tokens,
            "total_cost": metrics.cost,
            "latency_ms": metrics.latency,

            # Value metrics
            "task_completed": metrics.success,
            "human_time_saved": metrics.estimated_manual_time,
            "quality_score": metrics.quality_rating,
            "errors_prevented": metrics.errors_caught,

            # Efficiency metrics
            "iterations": metrics.agent_iterations,
            "tools_used": metrics.tool_calls,
            "cache_hit": metrics.cache_hit,
        }

    def calculate_unit_economics(self, period="month"):
        tasks = self.get_tasks(period)

        return {
            "cost_per_task": sum(t["total_cost"] for t in tasks) / len(tasks),
            "cost_per_success": sum(t["total_cost"] for t in tasks) / sum(1 for t in tasks if t["task_completed"]),
            "value_per_dollar": sum(t["human_time_saved"] * HOURLY_RATE for t in tasks) / sum(t["total_cost"] for t in tasks),
            "success_rate": sum(1 for t in tasks if t["task_completed"]) / len(tasks),
        }

class BudgetController:
    def __init__(self, daily_budget, alert_threshold=0.8):
        self.daily_budget = daily_budget
        self.alert_threshold = alert_threshold
        self.spent_today = 0

    async def check_budget(self, estimated_cost):
        if self.spent_today + estimated_cost > self.daily_budget:
            raise BudgetExceededError(
                f"Would exceed daily budget: ${self.spent_today + estimated_cost:.2f} > ${self.daily_budget}"
            )

        if self.spent_today > self.daily_budget * self.alert_threshold:
            await self.send_alert(f"Budget {self.alert_threshold*100}% consumed")

        return True

    def record_spend(self, actual_cost):
        self.spent_today += actual_cost

class CostAnomalyDetector:
    def __init__(self, window_size=100):
        self.recent_costs = deque(maxlen=window_size)

    def check(self, cost):
        if len(self.recent_costs) < 10:
            self.recent_costs.append(cost)
            return False

        mean = statistics.mean(self.recent_costs)
        std = statistics.stdev(self.recent_costs)

        # Flag if > 3 standard deviations
        if cost > mean + 3 * std:
            return {
                "anomaly": True,
                "cost": cost,
                "expected": mean,
                "deviation": (cost - mean) / std
            }

        self.recent_costs.append(cost)
        return False