| """Benchmark koşucusu: hazır kod setinde toplu değerlendirme. |
| |
| Her öğe için ajan akışı çalıştırılır ve şu ölçümler toplanır: |
| * ilk tur (geri beslemesiz) ve son tur: derlenme, geçen test oranı, satır kapsama |
| * mutasyon skoru (otomatik enjekte edilen hatalar) |
| * elle yerleştirilmiş gerçekçi hataların (seeded bugs) yakalanma oranı |
| * tur sayısı, LLM çağrısı, araç çağrısı, süre |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import re |
| import threading |
| import time |
| import uuid |
| from pathlib import Path |
| from statistics import mean |
| from typing import Any, Callable, Optional |
|
|
| from pydantic import BaseModel |
|
|
| from .. import store |
| from ..config import settings |
| from ..llm.base import LLMClient |
| from ..models import JobRequest |
| from ..sandbox.registry import get_runner |
|
|
| BENCH_DIR = Path(__file__).resolve().parents[3] / "benchmarks" |
|
|
|
|
| def load_manifest() -> list[dict[str, Any]]: |
| items = json.loads((BENCH_DIR / "manifest.json").read_text(encoding="utf-8")) |
| en = _english() |
| for it in items: |
| it["code"] = (BENCH_DIR / it["file"]).read_text(encoding="utf-8") |
| if it.get("html_file"): |
| it["html"] = (BENCH_DIR / it["html_file"]).read_text(encoding="utf-8") |
| _add_english(it, en) |
| return items |
|
|
|
|
| def _english() -> dict[str, Any]: |
| f = BENCH_DIR / "i18n_en.json" |
| return json.loads(f.read_text(encoding="utf-8")).get("items", {}) if f.exists() else {} |
|
|
|
|
| def _add_english(it: dict[str, Any], en: dict[str, Any]) -> None: |
| """Arayüzün İngilizce görünümü için başlık, açıklama ve yerleştirilmiş hata açıklamalarını ekler.""" |
| e = en.get(it["id"]) |
| if it.get("source") == "QuixBugs" or it["id"].startswith("qb-"): |
| e = {"title": it["title"], "description": "QuixBugs program; original single-line bug", |
| "bugs": ["Original QuixBugs bug"] * len(it.get("seeded_bugs", []))} |
| if not e: |
| return |
| it["title_en"], it["description_en"] = e["title"], e["description"] |
| for b, d in zip(it.get("seeded_bugs", []), e.get("bugs", [])): |
| b["description_en"] = d |
|
|
|
|
| class ItemResult(BaseModel): |
| id: str |
| title: str |
| title_en: str = "" |
| language: str |
| kind: str |
| split: str = "dev" |
| rep: int = 1 |
| job_id: str = "" |
| status: str = "pending" |
| error: str = "" |
| first_compiled: Optional[bool] = None |
| first_pass_rate: Optional[float] = None |
| first_coverage: Optional[float] = None |
| final_compiled: Optional[bool] = None |
| final_pass_rate: Optional[float] = None |
| final_coverage: Optional[float] = None |
| tests: int = 0 |
| iterations: int = 0 |
| mutation_score: Optional[float] = None |
| mutation_score_pre: Optional[float] = None |
| mutfix_tests_added: int = 0 |
| energy_wh: Optional[float] = None |
| mutants_killed: int = 0 |
| mutants_valid: int = 0 |
| seeded_total: int = 0 |
| seeded_caught: int = 0 |
| seeded_details: list[dict[str, Any]] = [] |
| suspected_bugs: int = 0 |
| pruned_tests: int = 0 |
| rejected_bug_reports: int = 0 |
| llm_calls: int = 0 |
| tool_calls: int = 0 |
| duration_s: float = 0.0 |
| stop_reason: str = "" |
| stop_reason_i18n: dict[str, str] = {} |
| gpu_percent: Optional[int] = None |
| tokens_per_s: Optional[float] = None |
| tokens_in: int = 0 |
| tokens_out: int = 0 |
|
|
|
|
| class EvalState(BaseModel): |
| id: str |
| created_at: float |
| status: str = "running" |
| model: str |
| label: str = "" |
| config: dict[str, Any] = {} |
| items: list[ItemResult] = [] |
| summary: dict[str, Any] = {} |
|
|
|
|
| _evals: dict[str, EvalState] = {} |
|
|
|
|
| def _rate(passed: int, total: int) -> Optional[float]: |
| return round(100.0 * passed / total, 2) if total else None |
|
|
|
|
| def check_seeded_bugs(item: dict[str, Any], test_code: str, baseline_fail: set[str]) -> list[dict[str, Any]]: |
| runner = get_runner(item["language"], item.get("kind", "unit")) |
| out = [] |
| for bug in item.get("seeded_bugs", []): |
| src = item["code"] |
| if bug["from"] not in src: |
| out.append({"description": bug["description"], "description_en": bug.get("description_en", ""), |
| "caught": None, "note": "yama uygulanamadı"}) |
| continue |
| buggy = src.replace(bug["from"], bug["to"], 1) |
| r = runner.run(buggy, test_code, item.get("kind", "unit"), item.get("html"), with_coverage=False) |
| caught = (not r.compiled) or r.timed_out or bool(r.failing_names() - baseline_fail) |
| out.append({"description": bug["description"], "description_en": bug.get("description_en", ""), |
| "caught": caught}) |
| return out |
|
|
|
|
| def _mut_rounds(cfg: dict[str, Any]) -> int: |
| """Mutant turu sayısı. Belirtilmemişse tek atışlık taban çizgisinde (en çok 1 tur) kapalıdır: o yapılandırma |
| hiçbir geri bildirim almamalıdır. Diğer yapılandırmalarda bir tur.""" |
| v = cfg.get("mutation_rounds") |
| if v is None: |
| return 0 if (cfg.get("max_iterations") or 4) == 1 else 1 |
| return int(v) |
|
|
|
|
| def evaluate_item(item: dict[str, Any], model: str, cfg: dict[str, Any], |
| llm_factory: Optional[Callable[[dict[str, Any]], LLMClient]] = None, rep: int = 1) -> ItemResult: |
| res = ItemResult(id=item["id"], title=item["title"], title_en=item.get("title_en", ""), |
| language=item["language"], kind=item.get("kind", "unit"), |
| split=item.get("split", "dev"), rep=rep) |
| req = JobRequest(language=item["language"], code=item["code"], test_kind=item.get("kind", "unit"), |
| html=item.get("html"), model=model, max_iterations=cfg.get("max_iterations"), |
| coverage_target=cfg.get("coverage_target"), run_mutation=cfg.get("run_mutation", True), |
| max_mutants=cfg.get("max_mutants"), temperature=cfg.get("temperature", 0.2), |
| mutation_rounds=_mut_rounds(cfg), |
| ablate=cfg.get("ablate") or [], lang=cfg.get("lang", "tr"), title=f"[bench] {item['title']}") |
| job = store.create_job(req, llm=llm_factory(item) if llm_factory else None, background=False) |
| res.job_id = job.id |
| res.status = job.status |
| if job.status != "done" or not job.result: |
| res.error = job.error or job.status |
| return res |
| r = job.result |
| if r.iterations: |
| f = r.iterations[0] |
| res.first_compiled = f.compiled |
| |
| res.first_pass_rate = _rate(f.passed, f.passed + f.failed) if f.compiled else 0.0 |
| res.first_coverage = f.coverage if f.compiled else 0.0 |
| res.iterations = len([x for x in r.iterations if x.action != "mutfix"]) |
| if r.final_run: |
| fr = r.final_run |
| res.final_compiled = fr.compiled |
| res.final_pass_rate = _rate(fr.passed, fr.passed + fr.failed + fr.errors) if fr.compiled else 0.0 |
| res.final_coverage = fr.coverage.percent if fr.compiled else 0.0 |
| res.tests = fr.total |
| if fr.compiled and fr.total and item.get("seeded_bugs"): |
| res.seeded_details = check_seeded_bugs(item, r.test_code, fr.failing_names()) |
| valid = [d for d in res.seeded_details if d["caught"] is not None] |
| res.seeded_total = len(valid) |
| res.seeded_caught = sum(1 for d in valid if d["caught"]) |
| if r.mutation: |
| res.mutation_score = r.mutation.score |
| res.mutation_score_pre = r.mutation_before.score if r.mutation_before else r.mutation.score |
| if r.mutation_round.get("accepted"): |
| res.mutfix_tests_added = int(r.mutation_round.get("tests_added", 0)) |
| res.mutants_killed = r.mutation.killed + r.mutation.timeout |
| res.mutants_valid = r.mutation.killed + r.mutation.survived + r.mutation.timeout |
| res.suspected_bugs = len(r.suspected_bugs) |
| res.pruned_tests = r.pruned_tests |
| res.rejected_bug_reports = len(r.rejected_bug_reports) |
| res.llm_calls, res.tool_calls = r.llm_calls, r.tool_calls |
| res.duration_s, res.stop_reason = r.duration_s, r.stop_reason |
| res.stop_reason_i18n = r.stop_reason_i18n |
| res.gpu_percent, res.tokens_per_s = r.compute.get("gpu_percent"), r.compute.get("tokens_per_s") |
| res.energy_wh = r.compute.get("energy_wh") |
| res.tokens_in = sum(int(e.data.get("tokens_in") or 0) for e in job.events if e.type == "llm") |
| res.tokens_out = sum(int(e.data.get("tokens_out") or 0) for e in job.events if e.type == "llm") |
| return res |
|
|
|
|
| def _avg(vals: list[Optional[float]]) -> Optional[float]: |
| v = [x for x in vals if x is not None] |
| return round(mean(v), 2) if v else None |
|
|
|
|
| def summarize(items: list[ItemResult]) -> dict[str, Any]: |
| def agg(group: list[ItemResult]) -> dict[str, Any]: |
| done = [i for i in group if i.status == "done"] |
| st = sum(i.seeded_total for i in done) |
| mv = sum(i.mutants_valid for i in done) |
| return { |
| "n": len(group), "completed": len(done), |
| "first_compile_rate": _rate(sum(1 for i in done if i.first_compiled), len(done)), |
| "final_compile_rate": _rate(sum(1 for i in done if i.final_compiled), len(done)), |
| "first_pass_rate": _avg([i.first_pass_rate for i in done]), |
| "final_pass_rate": _avg([i.final_pass_rate for i in done]), |
| "first_coverage": _avg([i.first_coverage for i in done]), |
| "final_coverage": _avg([i.final_coverage for i in done]), |
| "mutation_score": _avg([i.mutation_score for i in done]), |
| "mutation_score_pooled": _rate(sum(i.mutants_killed for i in done), mv), |
| "mutation_score_pre": _avg([i.mutation_score_pre for i in done]), |
| "avg_energy_wh": _avg([i.energy_wh for i in done]), |
| "seeded_catch_rate": _rate(sum(i.seeded_caught for i in done), st), |
| "seeded_total": st, |
| "avg_iterations": _avg([float(i.iterations) for i in done]), |
| "avg_llm_calls": _avg([float(i.llm_calls) for i in done]), |
| "avg_duration_s": _avg([i.duration_s for i in done]), |
| "avg_tokens_in": _avg([float(i.tokens_in) for i in done if i.tokens_in]), |
| "avg_tokens_out": _avg([float(i.tokens_out) for i in done if i.tokens_out]), |
| } |
|
|
| out = {"overall": agg(items), "by_language": {}, "by_kind": {}, "by_split": {}} |
| for lang in sorted({i.language for i in items}): |
| out["by_language"][lang] = agg([i for i in items if i.language == lang]) |
| for kind in sorted({i.kind for i in items}): |
| out["by_kind"][kind] = agg([i for i in items if i.kind == kind]) |
| for split in sorted({i.split for i in items}): |
| out["by_split"][split] = agg([i for i in items if i.split == split]) |
| return out |
|
|
|
|
| def _eval_dir() -> Path: |
| d = settings.data_dir / "evals" |
| d.mkdir(parents=True, exist_ok=True) |
| return d |
|
|
|
|
| def persist(ev: EvalState) -> None: |
| (_eval_dir() / f"{ev.id}.json").write_text(ev.model_dump_json(indent=2)) |
|
|
|
|
| def load_persisted() -> None: |
| for f in _eval_dir().glob("*.json"): |
| try: |
| ev = EvalState.model_validate_json(f.read_text()) |
| _evals.setdefault(ev.id, ev) |
| except Exception: |
| pass |
|
|
|
|
| def run_eval(model: str, cfg: dict[str, Any], item_ids: Optional[list[str]] = None, |
| languages: Optional[list[str]] = None, background: bool = True, |
| llm_factory: Optional[Callable[[dict[str, Any]], LLMClient]] = None, |
| split: Optional[str] = None, repeats: int = 1, label: str = "") -> EvalState: |
| items = load_manifest() |
| if item_ids: |
| items = [i for i in items if i["id"] in item_ids] |
| if languages: |
| items = [i for i in items if i["language"] in languages] |
| if split and split != "all": |
| items = [i for i in items if i.get("split", "dev") == split] |
| runs = [(r, i) for r in range(1, max(1, repeats) + 1) for i in items] |
| ev = EvalState(id=uuid.uuid4().hex[:10], created_at=time.time(), model=model, label=label, |
| config=cfg | {"split": split or "all", "repeats": max(1, repeats)}, |
| items=[ItemResult(id=i["id"], title=i["title"], title_en=i.get("title_en", ""), language=i["language"], kind=i.get("kind", "unit"), |
| split=i.get("split", "dev"), rep=r) for r, i in runs]) |
| _evals[ev.id] = ev |
| persist(ev) |
| _start(ev, {i["id"]: i for i in items}, llm_factory, background) |
| return ev |
|
|
|
|
| _stops: dict[str, threading.Event] = {} |
| _workers: dict[str, threading.Thread] = {} |
|
|
| |
| |
| INFRA_ERR = re.compile(r"Ollama'ya ulaşılamadı|Cannot reach Ollama|ConnectError|Connection refused|" |
| r"Cannot connect to the Docker daemon|docker: error during connect|RemoteProtocolError", re.I) |
| INFRA_WAIT_S = 15 * 60 |
| INFRA_POLL_S = 15 |
|
|
|
|
| def is_infra_error(msg: Optional[str]) -> bool: |
| return bool(msg and INFRA_ERR.search(msg)) |
|
|
|
|
| def _wait_for_model(model: str, stop: threading.Event) -> bool: |
| """Ollama yeniden ulaşılabilir olana kadar bekler (en çok INFRA_WAIT_S).""" |
| from ..llm.providers import is_local, make_llm |
| if not is_local(model): |
| time.sleep(INFRA_POLL_S) |
| return True |
| t0 = time.time() |
| while time.time() - t0 < INFRA_WAIT_S and not stop.is_set(): |
| try: |
| if make_llm(model).health().get("ok"): |
| time.sleep(10) |
| return True |
| except Exception: |
| pass |
| time.sleep(INFRA_POLL_S) |
| return False |
|
|
|
|
| def _start(ev: EvalState, by_id: dict[str, dict[str, Any]], |
| llm_factory: Optional[Callable[[dict[str, Any]], LLMClient]], background: bool) -> None: |
| """Bitmemiş (bekleyen ya da yarıda kalmış) öğeleri sırayla koşar; her öğeden sonra diske yazar. |
| |
| Böylece uygulama kapanıp açıldığında değerlendirme kaldığı öğeden sürdürülebilir. |
| """ |
| stop = _stops.setdefault(ev.id, threading.Event()) |
| stop.clear() |
| alive = _workers.get(ev.id) |
| if alive is not None and alive.is_alive(): |
| ev.status = "running" |
| return |
| ev.status = "running" |
|
|
| def work() -> None: |
| for idx, item in enumerate(ev.items): |
| if item.status not in ("pending", "running"): |
| continue |
| if stop.is_set(): |
| if ev.id in _cancel_flags: |
| _cancel_flags.discard(ev.id) |
| _close_cancelled(ev) |
| return |
| ev.status = "paused" |
| persist(ev) |
| return |
| src = by_id.get(item.id) |
| if src is None: |
| ev.items[idx].status, ev.items[idx].error = "failed", "öğe veri setinde yok" |
| continue |
| ev.items[idx].status = "running" |
| persist(ev) |
| for attempt in range(3): |
| try: |
| ev.items[idx] = evaluate_item(src, ev.model, ev.config, llm_factory, rep=item.rep) |
| except Exception as e: |
| ev.items[idx].status = "failed" |
| ev.items[idx].error = str(e) |
| res = ev.items[idx] |
| if res.status == "done" or not is_infra_error(res.error) or llm_factory is not None or attempt == 2: |
| break |
| persist(ev) |
| if not _wait_for_model(ev.model, stop): |
| break |
| ev.summary = summarize(ev.items) |
| persist(ev) |
| if ev.id in _cancel_flags: |
| _cancel_flags.discard(ev.id) |
| _close_cancelled(ev) |
| return |
| ev.status = "done" |
| ev.summary = summarize(ev.items) |
| persist(ev) |
|
|
| if background: |
| th = threading.Thread(target=work, daemon=True) |
| _workers[ev.id] = th |
| th.start() |
| else: |
| work() |
|
|
|
|
| def worker_alive(eval_id: str) -> bool: |
| th = _workers.get(eval_id) |
| return th is not None and th.is_alive() |
|
|
|
|
| def resume_eval(eval_id: str, background: bool = True, |
| llm_factory: Optional[Callable[[dict[str, Any]], LLMClient]] = None) -> Optional[EvalState]: |
| """Yarıda kalmış ya da duraklatılmış bir değerlendirmeyi kaldığı öğeden sürdürür.""" |
| ev = _evals.get(eval_id) |
| if ev is None or ev.status in ("done", "cancelled"): |
| return ev |
| if ev.status == "running" and worker_alive(eval_id): |
| _stops.setdefault(eval_id, threading.Event()).clear() |
| return ev |
| _start(ev, {i["id"]: i for i in load_manifest()}, llm_factory, background) |
| return ev |
|
|
|
|
| def reopen_infra_failures(eval_id: str) -> int: |
| """Altyapı hatasıyla başarısız olan öğeleri yeniden koşulmak üzere bekleyen duruma alır.""" |
| ev = _evals.get(eval_id) |
| if ev is None or (ev.status == "running" and worker_alive(eval_id)): |
| return 0 |
| n = 0 |
| for i, it in enumerate(ev.items): |
| if it.status == "failed" and is_infra_error(it.error): |
| ev.items[i] = ItemResult(id=it.id, title=it.title, title_en=it.title_en, language=it.language, |
| kind=it.kind, split=it.split, rep=it.rep) |
| n += 1 |
| if n: |
| ev.status = "paused" |
| ev.summary = summarize(ev.items) |
| persist(ev) |
| return n |
|
|
|
|
| def pause_eval(eval_id: str) -> bool: |
| """Süren öğe bittikten sonra durur (öğe yarıda kesilmez).""" |
| ev = _evals.get(eval_id) |
| if ev is None or ev.status != "running": |
| return False |
| _stops.setdefault(eval_id, threading.Event()).set() |
| return True |
|
|
|
|
| def cancel_eval(eval_id: str) -> bool: |
| """Değerlendirmeyi kalıcı durdurur: süren öğenin işi hemen iptal edilir, kalan öğeler koşulmaz.""" |
| ev = _evals.get(eval_id) |
| if ev is None or ev.status in ("done", "cancelled"): |
| return False |
| _stops.setdefault(eval_id, threading.Event()).set() |
| _cancel_flags.add(eval_id) |
| for j in store.all_jobs(): |
| if j.status in ("running", "queued") and (j.request.title or "").startswith("[bench]"): |
| store.cancel(j.id) |
| if not worker_alive(eval_id): |
| _close_cancelled(ev) |
| return True |
|
|
|
|
| _cancel_flags: set[str] = set() |
|
|
|
|
| def _close_cancelled(ev: EvalState) -> None: |
| for it in ev.items: |
| if it.status in ("pending", "running"): |
| it.status = "cancelled" |
| ev.status = "cancelled" |
| ev.summary = summarize(ev.items) |
| persist(ev) |
|
|
|
|
| def interrupted() -> list[EvalState]: |
| """Uygulama kapanırken 'running' kalmış (iş parçacığı artık olmayan) değerlendirmeler.""" |
| return [e for e in _evals.values() if e.status == "running" and e.id not in _stops] |
|
|
|
|
| def get_eval(eval_id: str) -> Optional[EvalState]: |
| return _evals.get(eval_id) |
|
|
|
|
| def all_evals() -> list[EvalState]: |
| return sorted(_evals.values(), key=lambda e: e.created_at, reverse=True) |
|
|
