"""ACO Benchmark Suite — compares cost optimization configs across 5 task types.""" import json, time, hashlib, random from typing import Dict, List, Tuple, Callable from dataclasses import dataclass, field @dataclass class StepMetrics: step_num: int; model_used: str; input_tokens: int; output_tokens: int context_size: int = 0; tool_calls: int = 0; verifier_calls: int = 0 retries: int = 0; latency_ms: float = 0.0; cost_usd: float = 0.0 @dataclass class TaskResult: task_id: str; task_type: str; success: bool steps: list = field(default_factory=list); total_cost_usd: float = 0.0 total_llm_calls: int = 0; total_tool_calls: int = 0; total_retries: int = 0 false_done: bool = False @dataclass class RunReport: config_name: str; config_description: str = "" results: list = field(default_factory=list) @property def success_rate(self): return sum(1 for r in self.results if r.success)/max(len(self.results),1) @property def avg_cost_per_success(self): s = [r.total_cost_usd for r in self.results if r.success] return sum(s)/len(s) if s else float('inf') @property def avg_llm_calls(self): return sum(r.total_llm_calls for r in self.results)/max(len(self.results),1) @property def avg_tool_calls(self): return sum(r.total_tool_calls for r in self.results)/max(len(self.results),1) @property def false_done_rate(self): return sum(1 for r in self.results if r.false_done)/max(len(self.results),1) PRICING = { "gpt-4.1-mini": (0.15, 0.60), "gpt-4o": (2.50, 10.0), "claude-sonnet-4": (3.0, 15.0), "llama-3.1-8b": (0.06, 0.06), "claude-haiku": (0.80, 4.0), "gemini-flash": (0.075, 0.30), } TIER_MODELS = {"cheap": "llama-3.1-8b", "medium": "gpt-4.1-mini", "frontier": "claude-sonnet-4"} SUCCESS_RATES = { "cheap": {"coding": 0.55, "research": 0.40, "tool_use": 0.50, "qa": 0.60, "long_horizon": 0.30}, "medium": {"coding": 0.78, "research": 0.65, "tool_use": 0.75, "qa": 0.85, "long_horizon": 0.55}, "frontier": {"coding": 0.92, "research": 0.85, "tool_use": 0.90, "qa": 0.95, "long_horizon": 0.80}, } ALL_TASKS = { "coding": [ ("code_1", "Write a Python function to sort a list of dicts by key", "easy"), ("code_2", "Fix the race condition in this multithreaded queue", "hard"), ("code_3", "Add input validation to all API endpoints", "medium"), ("code_4", "Migrate from SQLAlchemy 1.4 to 2.0 async API", "hard"), ("code_5", "Write unit tests for the User model", "medium"), ], "research": [ ("research_1", "What's the SOTA for cost-aware LLM routing?", "medium"), ("research_2", "Find all papers about speculative decoding for agents", "hard"), ("research_3", "How does Anthropic's prompt caching work?", "easy"), ("research_4", "Survey test-time compute allocation 2023-2025", "hard"), ("research_5", "What datasets exist for training agent cost routers?", "medium"), ], "tool_use": [ ("tool_1", "What's the weather in Paris today?", "easy"), ("tool_2", "Find top 5 restaurants in Tokyo with ratings", "medium"), ("tool_3", "Compare GDP of France and Germany over last decade", "medium"), ("tool_4", "Translate 'hello' to Japanese", "easy"), ("tool_5", "Compound interest on $10k at 5% over 20 years", "easy"), ], "qa": [ ("qa_1", "What is the capital of France?", "easy"), ("qa_2", "Explain difference between TCP and UDP", "easy"), ("qa_3", "GDPR Article 22 implications for automated decisions?", "hard"), ("qa_4", "Review this contract for liability clauses", "hard"), ("qa_5", "Summarize key findings from IPCC AR6 report", "medium"), ], "long_horizon": [ ("long_1", "Set up complete CI/CD pipeline for Python project", "hard"), ("long_2", "Research, design, and implement a caching layer", "hard"), ("long_3", "Write README, add tests, set up pre-commit hooks", "medium"), ("long_4", "Migrate DB schema, update models, add rollback", "hard"), ("long_5", "Audit codebase for security vulnerabilities, fix top 3", "hard"), ], } class SimulatedAgent: def __init__(self, seed=42): self.rng = random.Random(seed) def run_task(self, task_id, text, tier, task_type, difficulty, use_tools, use_verifier): model = TIER_MODELS[tier]; ip, op = PRICING[model] base_p = SUCCESS_RATES[tier].get(task_type, 0.7) diff_m = {"easy": 1.15, "medium": 1.0, "hard": 0.80} p = min(0.99, base_p * diff_m[difficulty]) nsteps = {"easy": self.rng.randint(1,3), "medium": self.rng.randint(2,5), "hard": self.rng.randint(3,8)}[difficulty] result = TaskResult(task_id=task_id, task_type=task_type, success=False) for i in range(nsteps): it = self.rng.randint(200,3000); ot = self.rng.randint(50,1500) tc = self.rng.randint(0,3) if use_tools else 0 tf = self.rng.randint(0,1) if tc>0 else 0 vc = 1 if use_verifier else 0 rt = self.rng.randint(0,2) if tf>0 else 0 cost = it/1e6*ip + ot/1e6*op result.steps.append(StepMetrics(i+1, model, it, ot, it+ot, tc, vc, rt, it*2+ot*10, cost)) result.total_cost_usd += cost; result.total_llm_calls += 1 result.total_tool_calls += tc; result.total_retries += rt result.success = self.rng.random() < (p * (1 - 0.03*result.total_retries)) return result class BenchmarkRunner: def __init__(self, agent=None): self.agent = agent or SimulatedAgent() self.reports = {} def run_config(self, name, desc, router_fn): report = RunReport(name, desc) for task_type, tasks in ALL_TASKS.items(): for tid, text, diff in tasks: tier, tools, verifier = router_fn(text, diff, task_type) result = self.agent.run_task(tid, text, tier, task_type, diff, tools, verifier) report.results.append(result) self.reports[name] = report return report def run_all_baselines(self): self.run_config("A_always_frontier", "Frontier + tools + verifier", lambda t,d,tt: ("frontier", True, True)) self.run_config("B_always_cheap", "Cheap model, no tools", lambda t,d,tt: ("cheap", False, False)) self.run_config("C_static_routing", "Static: easy→cheap, medium→medium, hard→frontier", lambda t,d,tt: ({"easy":"cheap","medium":"medium","hard":"frontier"}[d], d!="easy", d=="hard")) self.run_config("D_prompt_router", "Keyword-based heuristic routing", lambda t,d,tt: self._prompt_route(t, d, tt)) def _prompt_route(self, text, diff, tt): t = text.lower() if any(k in t for k in ["fix","debug","migrate","critical","vulnerability"]): return ("frontier", True, True) if any(k in t for k in ["test","add","write","survey","find","setup"]): return ("medium", True, False) return ("cheap", False, False) def compare(self): if "A_always_frontier" not in self.reports: return {} bl = self.reports["A_always_frontier"] bs = bl.success_rate; bc = bl.avg_cost_per_success comp = {} for name, rpt in self.reports.items(): sr = rpt.success_rate; cs = rpt.avg_cost_per_success cr = ((bc-cs)/bc*100) if sr >= bs*0.95 else None comp[name] = {"success": f"{sr:.1%}", "cost_per_success": f"${cs:.4f}", "cost_reduction": f"{cr:.1f}%" if cr else "N/A", "llm_calls": f"{rpt.avg_llm_calls:.1f}", "tool_calls": f"{rpt.avg_tool_calls:.1f}", "false_done": f"{rpt.false_done_rate:.1%}"} return comp def print_report(self): comp = self.compare() print(f"\n{'='*100}") print("ACO BENCHMARK COMPARISON (25 tasks)") print(f"{'='*100}") print(f"{'Config':<25} {'Success':>8} {'Cost/Success':>13} {'Reduction':>12} {'LLM':>6} {'Tools':>7} {'FalseDONE':>10}") print("-"*100) for n,s in comp.items(): print(f"{n:<25} {s['success']:>8} {s['cost_per_success']:>13} {s['cost_reduction']:>12} {s['llm_calls']:>6} {s['tool_calls']:>7} {s['false_done']:>10}") print("="*100) if __name__ == "__main__": runner = BenchmarkRunner() runner.run_all_baselines() runner.print_report()