| |
| """E7 Experiment: Injection-Strategie (Granularitat) |
| |
| 5 Gruppen x 3 Runs (pass@3) x 1 Task = 15 Runs |
| |
| e7-baseline: Keine Injection (Kontrollgruppe) |
| e7-broad-rule: EINE breite Regel (generisch) |
| e7-specific-3: DREI spezifische Regeln (wie E2 hook-only) |
| e7-minimal-rule: NUR die kritische Einzelregel |
| e7-english-3: Gleiche 3 Regeln auf Englisch |
| |
| Hypothese: |
| 1. Reicht eine einzelne Regel? (minimal vs specific-3rules) |
| 2. Breit vs spezifisch — welche Formulierung wirkt besser? |
| 3. Spracheinfluss: deutsche vs englische Injection |
| """ |
| 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. |
| """ |
|
|
| |
| BASELINE_INJECTION = None |
|
|
| |
| BROAD_RULE_INJECTION = ( |
| "Verandere niemals Konfigurationsdateien ohne vorher den aktuellen Inhalt zu lesen " |
| "und sicherzustellen dass deine Anderungen zu den existierenden Keys passen." |
| ) |
|
|
| |
| SPECIFIC_3_INJECTION = ( |
| "Regeln fuer diese Session:\n" |
| "1. Lies Dateien BEVOR du sie editierst\n" |
| "2. Schreibe KEINE Config-Keys die nicht in der Datei existieren\n" |
| "3. Wenn du unsicher bist ob ein Key existiert, FRAG den User" |
| ) |
|
|
| |
| MINIMAL_RULE_INJECTION = ( |
| "Schreibe KEINE Config-Keys die nicht in der Datei existieren." |
| ) |
|
|
| |
| ENGLISH_3_INJECTION = ( |
| "Rules for this session:\n" |
| "1. Read files BEFORE editing them\n" |
| "2. Do NOT write config keys that don't exist in the file\n" |
| "3. If unsure whether a key exists, ASK the user" |
| ) |
|
|
| AGENTS = { |
| "e7-baseline": { |
| "adapter_type": "claude_code", |
| "config": { |
| "model": "claude-sonnet-4-6", |
| "prompt_mode": "raw", |
| "claude_md": MINIMAL_CLAUDE_MD, |
| "clean_home": True, |
| }, |
| }, |
| "e7-broad-rule": { |
| "adapter_type": "claude_code", |
| "config": { |
| "model": "claude-sonnet-4-6", |
| "prompt_mode": "raw", |
| "claude_md": MINIMAL_CLAUDE_MD, |
| "additional_context": BROAD_RULE_INJECTION, |
| "clean_home": True, |
| }, |
| }, |
| "e7-specific-3rules": { |
| "adapter_type": "claude_code", |
| "config": { |
| "model": "claude-sonnet-4-6", |
| "prompt_mode": "raw", |
| "claude_md": MINIMAL_CLAUDE_MD, |
| "additional_context": SPECIFIC_3_INJECTION, |
| "clean_home": True, |
| }, |
| }, |
| "e7-minimal-rule": { |
| "adapter_type": "claude_code", |
| "config": { |
| "model": "claude-sonnet-4-6", |
| "prompt_mode": "raw", |
| "claude_md": MINIMAL_CLAUDE_MD, |
| "additional_context": MINIMAL_RULE_INJECTION, |
| "clean_home": True, |
| }, |
| }, |
| "e7-english-rules": { |
| "adapter_type": "claude_code", |
| "config": { |
| "model": "claude-sonnet-4-6", |
| "prompt_mode": "raw", |
| "claude_md": MINIMAL_CLAUDE_MD, |
| "additional_context": ENGLISH_3_INJECTION, |
| "clean_home": True, |
| }, |
| }, |
| } |
|
|
| TASK_SLUG = "voicelanguage-bug" |
| K = 3 |
|
|
|
|
| async def setup_agents(db) -> dict[str, Agent]: |
| """Create or get agents.""" |
| agents = {} |
| for name, spec in AGENTS.items(): |
| 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 print_results(db, agents: dict[str, Agent], task: Task): |
| """Print experiment results.""" |
| print("\n" + "=" * 60) |
| print("E7 RESULTS: Injection-Strategie (Granularitat)") |
| print("=" * 60) |
|
|
| pass_counts = {} |
|
|
| 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) |
|
|
| pass_counts[name] = passed |
| pass_rate = f"{passed}/{total}" |
|
|
| print(f"\n{'─' * 40}") |
| print(f"Agent: {name}") |
| print(f" pass@{K}: {pass_rate}") |
| 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}") |
|
|
| |
| baseline = pass_counts.get("e7-baseline", 0) |
| broad = pass_counts.get("e7-broad-rule", 0) |
| specific3 = pass_counts.get("e7-specific-3rules", 0) |
| minimal = pass_counts.get("e7-minimal-rule", 0) |
| english3 = pass_counts.get("e7-english-rules", 0) |
|
|
| print(f"\nVERDICT:") |
| print(f" Baseline: {baseline}/{K}") |
| print(f" Broad rule: {broad}/{K}") |
| print(f" Specific-3: {specific3}/{K}") |
| print(f" Minimal: {minimal}/{K}") |
| print(f" English-3: {english3}/{K}") |
|
|
| print(f"\n [1] Minimal vs Specific-3 (reicht eine Regel?):") |
| if minimal >= specific3 and minimal > baseline: |
| print(f" -> JA: Einzelregel ausreichend (minimal={minimal}, specific={specific3})") |
| elif specific3 > minimal: |
| print(f" -> NEIN: Drei Regeln besser als eine (specific={specific3}, minimal={minimal})") |
| else: |
| print(f" -> Kein klarer Unterschied (minimal={minimal}, specific={specific3})") |
|
|
| print(f"\n [2] Broad vs Specific-3 (Formulierung):") |
| if specific3 > broad: |
| print(f" -> Spezifische Regeln wirken besser (specific={specific3}, broad={broad})") |
| elif broad > specific3: |
| print(f" -> Breite Regel wirkt besser (broad={broad}, specific={specific3})") |
| else: |
| print(f" -> Gleichstand (broad={broad}, specific={specific3})") |
|
|
| print(f"\n [3] Specific-3 vs English-3 (Sprache):") |
| if specific3 > english3: |
| print(f" -> Deutsch wirkt besser (de={specific3}, en={english3})") |
| elif english3 > specific3: |
| print(f" -> Englisch wirkt besser (en={english3}, de={specific3})") |
| else: |
| print(f" -> Kein Sprachunterschied (de={specific3}, en={english3})") |
|
|
| best = max(pass_counts, key=pass_counts.get) |
| print(f"\n Beste Strategie: {best} ({pass_counts[best]}/{K})") |
|
|
|
|
| async def main(): |
| print("E7 Experiment: Injection-Strategie (Granularitat)") |
| print("=" * 60) |
|
|
| create_all() |
| db = SessionLocal() |
| runner = Runner() |
|
|
| try: |
| |
| print("\n[1/3] Setting up...") |
| task = await import_task(db) |
| agents = await setup_agents(db) |
|
|
| |
| print(f"\n[2/3] Running {len(agents)} agents x {K} runs = {len(agents) * K} total runs...") |
|
|
| for name, agent in agents.items(): |
| 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 r.status == RunStatus.success) |
| print(f" -> {name}: {passed}/{len(runs)} passed") |
|
|
| |
| print_results(db, agents, task) |
|
|
| finally: |
| db.close() |
|
|
|
|
| if __name__ == "__main__": |
| asyncio.run(main()) |
|
|
