#!/usr/bin/env python3 """E6 Experiment: Cross-Model — Sonnet vs Haiku vs Opus 6 Agents × 3 Runs (pass@3) × 1 Task = 18 Runs Group haiku-baseline: claude-haiku-4-5-20251001, keine Injection Group haiku-injected: claude-haiku-4-5-20251001, mit Injection Group sonnet-baseline: claude-sonnet-4-6, keine Injection Group sonnet-injected: claude-sonnet-4-6, mit Injection Group opus-baseline: claude-opus-4-6, keine Injection Group opus-injected: claude-opus-4-6, mit Injection Hypothese: Reagieren verschiedene Modelle gleich auf Sidecar-NG Injection? Verdict-Fragen: 1. Welches Modell hat die beste Baseline? (Braucht Opus ueberhaupt Injection?) 2. Wirkt Injection bei allen Modellen gleich? 3. Cost-Efficiency: Haiku+Injection vs Sonnet-Baseline vs Opus-Baseline ACHTUNG: Opus-Runs sind ~10x teurer als Sonnet. Das Script fragt vor Opus-Runs nach Bestaetigung. """ import asyncio import json import sys import os # Add parent dir to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from database import SessionLocal, create_all from models import Task, Agent, Run, RunStatus from core.runner import Runner # --- Agent Configurations --- MINIMAL_CLAUDE_MD = """# Settings Use tools to solve tasks. Be concise. """ SIDECAR_INJECTION = """Regeln fuer diese Session: 1. Lies Dateien BEVOR du sie editierst 2. Schreibe KEINE Config-Keys die nicht in der Datei existieren 3. Wenn du unsicher bist ob ein Key existiert, FRAG den User""" # Approximate cost per run based on ~50k tokens for this task # Haiku: $0.0008/1k input, $0.004/1k output -> ~$0.05/run # Sonnet: $0.003/1k input, $0.015/1k output -> ~$0.20/run # Opus: $0.015/1k input, $0.075/1k output -> ~$1.00/run ESTIMATED_COST_PER_RUN = { "haiku": 0.05, "sonnet": 0.20, "opus": 1.00, } AGENTS = { "e6-haiku-baseline": { "adapter_type": "claude_code", "model_tier": "haiku", "config": { "model": "claude-haiku-4-5-20251001", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "clean_home": True, }, }, "e6-haiku-injected": { "adapter_type": "claude_code", "model_tier": "haiku", "config": { "model": "claude-haiku-4-5-20251001", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "additional_context": SIDECAR_INJECTION, "clean_home": True, }, }, "e6-sonnet-baseline": { "adapter_type": "claude_code", "model_tier": "sonnet", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "clean_home": True, }, }, "e6-sonnet-injected": { "adapter_type": "claude_code", "model_tier": "sonnet", "config": { "model": "claude-sonnet-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "additional_context": SIDECAR_INJECTION, "clean_home": True, }, }, "e6-opus-baseline": { "adapter_type": "claude_code", "model_tier": "opus", "config": { "model": "claude-opus-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "clean_home": True, }, }, "e6-opus-injected": { "adapter_type": "claude_code", "model_tier": "opus", "config": { "model": "claude-opus-4-6", "prompt_mode": "raw", "claude_md": MINIMAL_CLAUDE_MD, "additional_context": SIDECAR_INJECTION, "clean_home": True, }, }, } TASK_SLUG = "voicelanguage-bug" K = 3 # pass@3 OPUS_AGENTS = {"e6-opus-baseline", "e6-opus-injected"} def warn_and_confirm_opus(k: int) -> bool: """Print Opus cost warning and ask for confirmation. Returns True if confirmed.""" opus_runs = len(OPUS_AGENTS) * k opus_cost_per_run = ESTIMATED_COST_PER_RUN["opus"] estimated_total = opus_runs * opus_cost_per_run total_all_runs = sum(ESTIMATED_COST_PER_RUN[spec["model_tier"]] for spec in AGENTS.values()) * k non_opus_cost = total_all_runs - estimated_total print() print("=" * 60) print("ACHTUNG: Opus-Runs sind ~10x teurer als Sonnet!") print("=" * 60) print(f" Opus-Runs geplant: {opus_runs} ({len(OPUS_AGENTS)} agents x {k})") print(f" Geschaetzte Opus-Kosten: ~${estimated_total:.2f}") print(f" Nicht-Opus-Kosten: ~${non_opus_cost:.2f}") print(f" Gesamtschaetzung: ~${total_all_runs:.2f}") print() print("Optionen:") print(" [y] Ja, Opus-Runs ausfuehren (volle Kosten)") print(" [n] Nein, nur Haiku + Sonnet ausfuehren (guenstiger)") print(" [q] Abbrechen") print() try: answer = input("Entscheidung [y/n/q]: ").strip().lower() except (EOFError, KeyboardInterrupt): print("\nAbgebrochen.") return False if answer == "y": print(" -> Opus-Runs werden ausgefuehrt.") return True elif answer == "n": print(" -> Opus-Runs werden uebersprungen.") return False else: print(" -> Experiment abgebrochen.") sys.exit(0) async def setup_agents(db, run_opus: bool) -> dict[str, Agent]: """Create or get agents. Skips opus agents if run_opus is False.""" agents = {} for name, spec in AGENTS.items(): if not run_opus and name in OPUS_AGENTS: continue agent = db.query(Agent).filter(Agent.name == name).first() if not agent: agent = Agent( name=name, adapter_type=spec["adapter_type"], config_json=json.dumps(spec["config"]), ) db.add(agent) db.commit() db.refresh(agent) print(f" Created agent: {name}") else: # Update config in case it changed agent.config_json = json.dumps(spec["config"]) db.commit() print(f" Using existing agent: {name}") agents[name] = agent return agents async def import_task(db) -> Task: """Import task if not exists.""" task = db.query(Task).filter(Task.slug == TASK_SLUG).first() if not task: import yaml yaml_path = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tasks", "voicelanguage_bug.yaml", ) with open(yaml_path) as f: data = yaml.safe_load(f) task = Task( slug=data["slug"], title=data["title"], description=data.get("description", ""), repo_url=data["repo"]["url"], repo_ref=data["repo"].get("ref", "main"), success_cmd=data["success"]["command"], time_limit_s=data["limits"].get("time_seconds", 300), token_limit=data["limits"].get("max_tokens", 200000), difficulty=data.get("difficulty", "easy"), ) db.add(task) db.commit() db.refresh(task) print(f" Imported task: {TASK_SLUG}") else: print(f" Using existing task: {TASK_SLUG}") return task def _functional_pass(run) -> bool: """Check if the judge's functional test passed (exit 0), ignoring regression/lint/complexity.""" try: metrics = json.loads(run.metrics_json or "{}") return metrics.get("judge", {}).get("functional_score", 0) >= 1.0 except (json.JSONDecodeError, ValueError): return run.status == RunStatus.success def _get_pass_count(db, agent: Agent, task: Task) -> tuple[int, float]: """Returns (pass_count, avg_cost_usd) for an agent.""" runs = ( db.query(Run) .filter(Run.agent_id == agent.id, Run.task_id == task.id) .order_by(Run.created_at.desc()) .limit(K) .all() ) passed = sum(1 for r in runs if _functional_pass(r)) avg_cost = sum(r.cost_usd or 0 for r in runs) / max(len(runs), 1) return passed, avg_cost def print_results(db, agents: dict[str, Agent], task: Task, run_opus: bool): """Print experiment results.""" print("\n" + "=" * 60) print("E6 RESULTS: Cross-Model — Sonnet vs Haiku vs Opus") print("=" * 60) for name, agent in agents.items(): runs = ( db.query(Run) .filter(Run.agent_id == agent.id, Run.task_id == task.id) .order_by(Run.created_at.desc()) .limit(K) .all() ) total = len(runs) passed = sum(1 for r in runs if _functional_pass(r)) failed = total - passed errors = sum(1 for r in runs if r.status in (RunStatus.error, RunStatus.timeout)) avg_cost = sum(r.cost_usd or 0 for r in runs) / max(total, 1) print(f"\n{'─' * 40}") print(f"Agent: {name}") print(f" pass@{K}: {passed}/{total}") print(f" Passed: {passed} | Failed: {failed} | Errors: {errors}") print(f" Avg Cost: ${avg_cost:.4f}") for i, run in enumerate(runs, 1): func_pass = _functional_pass(run) status_icon = "+" if func_pass else "x" notes = (run.judge_notes or "")[:80] print(f" Run {i}: {status_icon} {'PASS' if func_pass else 'FAIL'} — {notes}") print(f"\n{'=' * 60}") # Collect per-agent stats def stats(agent_name: str) -> tuple[int, float]: if agent_name not in agents: return -1, 0.0 return _get_pass_count(db, agents[agent_name], task) haiku_base_pass, haiku_base_cost = stats("e6-haiku-baseline") haiku_inj_pass, haiku_inj_cost = stats("e6-haiku-injected") sonnet_base_pass, sonnet_base_cost = stats("e6-sonnet-baseline") sonnet_inj_pass, sonnet_inj_cost = stats("e6-sonnet-injected") opus_base_pass, opus_base_cost = stats("e6-opus-baseline") opus_inj_pass, opus_inj_cost = stats("e6-opus-injected") def fmt(p, c) -> str: if p == -1: return "skipped" return f"{p}/{K} (${c:.4f}/run)" print("\nVERDICT:") print(f" Haiku baseline: {fmt(haiku_base_pass, haiku_base_cost)}") print(f" Haiku injected: {fmt(haiku_inj_pass, haiku_inj_cost)}") print(f" Sonnet baseline: {fmt(sonnet_base_pass, sonnet_base_cost)}") print(f" Sonnet injected: {fmt(sonnet_inj_pass, sonnet_inj_cost)}") if run_opus: print(f" Opus baseline: {fmt(opus_base_pass, opus_base_cost)}") print(f" Opus injected: {fmt(opus_inj_pass, opus_inj_cost)}") else: print(" Opus: skipped (nicht ausgefuehrt)") print() # Q1: Best baseline? baselines = { "Haiku": haiku_base_pass, "Sonnet": sonnet_base_pass, } if run_opus: baselines["Opus"] = opus_base_pass best_baseline = max(baselines, key=lambda m: baselines[m]) print(f" Q1 — Beste Baseline: {best_baseline} ({baselines[best_baseline]}/{K})") if run_opus and opus_base_pass == K: print(" -> Opus besteht ohne Injection — Injection evtl. ueberfluessig fuer Opus") elif run_opus and opus_base_pass > sonnet_base_pass: print(" -> Opus ist deutlich besser als Sonnet in der Baseline") elif sonnet_base_pass >= (opus_base_pass if run_opus else 0): print(" -> Sonnet-Baseline ausreichend, Opus-Mehrkosten nicht gerechtfertigt") # Q2: Does injection help each model equally? print() print(" Q2 — Injection-Wirkung pro Modell:") for model, base, inj in [ ("Haiku", haiku_base_pass, haiku_inj_pass), ("Sonnet", sonnet_base_pass, sonnet_inj_pass), ]: if base == -1: continue delta = inj - base if delta > 0: print(f" {model}: +{delta} (Injection hilft)") elif delta == 0: print(f" {model}: kein Unterschied ({base}/{K})") else: print(f" {model}: {delta} (Injection schadet?! — genauer pruefen)") if run_opus and opus_base_pass != -1: delta = opus_inj_pass - opus_base_pass if delta > 0: print(f" Opus: +{delta} (Injection hilft auch bei Opus)") elif delta == 0: print(f" Opus: kein Unterschied ({opus_base_pass}/{K})") else: print(f" Opus: {delta} (Injection schadet bei Opus?! — genauer pruefen)") # Q3: Cost-efficiency print() print(" Q3 — Cost-Efficiency:") candidates = [ ("Haiku+Injection", haiku_inj_pass, haiku_inj_cost), ("Sonnet-Baseline", sonnet_base_pass, sonnet_base_cost), ("Sonnet+Injection", sonnet_inj_pass, sonnet_inj_cost), ] if run_opus: candidates += [ ("Opus-Baseline", opus_base_pass, opus_base_cost), ("Opus+Injection", opus_inj_pass, opus_inj_cost), ] for label, p, c in candidates: if p == -1: continue cost_per_pass = (c / max(p, 0.001)) if p > 0 else float("inf") print(f" {label:<22}: {p}/{K} @ ${c:.4f}/run (${cost_per_pass:.4f}/pass)") # Best cost-efficiency: lowest cost-per-pass among those with at least 1 pass valid = [(label, p, c) for label, p, c in candidates if p > 0 and c > 0] if valid: best_label, best_p, best_c = min(valid, key=lambda x: x[2] / x[1]) print(f"\n -> Guenstigste Option mit Erfolg: {best_label} ({best_p}/{K})") print(f"\n{'=' * 60}") async def main(): print("E6 Experiment: Cross-Model — Sonnet vs Haiku vs Opus") print("=" * 60) # Opus cost warning before anything else run_opus = warn_and_confirm_opus(K) create_all() db = SessionLocal() runner = Runner() try: # Setup print("\n[1/3] Setting up...") task = await import_task(db) agents = await setup_agents(db, run_opus) active_count = len(agents) total_runs = active_count * K print(f"\n[2/3] Running {active_count} agents x {K} runs = {total_runs} total runs...") if not run_opus: print(" (Opus-Agents uebersprungen)") for name, agent in agents.items(): # Extra notice before opus runs (in case they sneak in somehow) if name in OPUS_AGENTS: print(f"\n [OPUS] Starting {name} — hohe Kosten!") else: print(f"\n Starting {name} (pass@{K})...") runs = await runner.run_pass_at_k(db, task, agent, k=K) passed = sum(1 for r in runs if _functional_pass(r)) print(f" -> {name}: {passed}/{len(runs)} passed") # Results print_results(db, agents, task, run_opus) finally: db.close() if __name__ == "__main__": asyncio.run(main())