| """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() |
|
|
