| |
| """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 |
|
|
| |
| 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 |
|
|
| |
|
|
| 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""" |
|
|
| |
| |
| |
| |
| 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 |
|
|
| 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: |
| |
| 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}") |
|
|
| |
| 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() |
|
|
| |
| 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") |
|
|
| |
| 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)") |
|
|
| |
| 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)") |
|
|
| |
| 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) |
|
|
| |
| run_opus = warn_and_confirm_opus(K) |
|
|
| create_all() |
| db = SessionLocal() |
| runner = Runner() |
|
|
| try: |
| |
| 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(): |
| |
| 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") |
|
|
| |
| print_results(db, agents, task, run_opus) |
|
|
| finally: |
| db.close() |
|
|
|
|
| if __name__ == "__main__": |
| asyncio.run(main()) |
|
|
