File size: 4,143 Bytes
c641d5f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | """Concurrent dry-run evaluation with atomic per-task checkpoints."""
from __future__ import annotations
import time
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any
from agent import GaiaAgent
from cache import ResultStore
from gaia_client import GaiaClient, GaiaClientError, GaiaNetworkError
Progress = Callable[[int, int, str], None]
def run_evaluation(
client: GaiaClient,
agent: GaiaAgent,
result_store: ResultStore,
progress: Progress | None = None,
*,
force: bool = False,
task_ids: set[str] | None = None,
) -> list[dict[str, Any]]:
tasks = client.get_questions()
if task_ids is not None:
tasks = [task for task in tasks if str(task["task_id"]) in task_ids]
missing = task_ids - {str(task["task_id"]) for task in tasks}
if missing:
raise ValueError(f"Unknown task IDs: {', '.join(sorted(missing))}")
existing = result_store.load()
completed = 0
def run_one(task: dict[str, Any]) -> dict[str, Any]:
task_id = str(task["task_id"])
if not force and existing.get(task_id, {}).get("status") == "ok":
return existing[task_id]
started = time.perf_counter()
row: dict[str, Any] = {
"task_id": task_id,
"question": str(task["question"]),
"answer": "",
"task_type": "",
"status": "agent_error",
"duration_seconds": 0.0,
"error": "",
"evidence": [],
"confidence": None,
}
try:
attachment = client.download_attachment(task)
solved = agent.solve_detailed(task, attachment, force=force)
row.update(
answer=solved.answer,
task_type=solved.task_type,
evidence=solved.evidence,
confidence=solved.confidence,
status="ok",
)
except GaiaNetworkError as exc:
row.update(status="network_error", error=f"{type(exc).__name__}: {exc}")
except GaiaClientError as exc:
row.update(status="api_error", error=f"{type(exc).__name__}: {exc}")
except Exception as exc:
row.update(status="agent_error", error=f"{type(exc).__name__}: {exc}")
row["duration_seconds"] = round(time.perf_counter() - started, 3)
result_store.save_result(task_id, row)
return row
workers = min(agent.settings.worker_limit, max(1, len(tasks)))
with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="gaia") as executor:
futures = {executor.submit(run_one, task): task for task in tasks}
for future in as_completed(futures):
completed += 1
task_id = str(futures[future]["task_id"])
future.result() # run_one isolates expected task failures
if progress:
progress(completed, len(tasks), task_id)
ordered_ids = [str(task["task_id"]) for task in tasks]
return result_store.ordered(ordered_ids)
def submission_answers(
results: list[dict[str, Any]], expected_count: int | None = 20
) -> list[dict[str, str]]:
if not results:
raise ValueError("No dry-run results exist")
if expected_count is not None and len(results) != expected_count:
raise ValueError(
f"Refusing submission: expected exactly {expected_count} tasks, found {len(results)}"
)
task_ids = [str(row.get("task_id", "")) for row in results]
if len(set(task_ids)) != len(task_ids):
raise ValueError("Refusing submission: duplicate task IDs")
failures = [row for row in results if row.get("status") != "ok"]
if failures:
raise ValueError(f"Refusing submission: {len(failures)} task(s) are incomplete")
empty = [row for row in results if not str(row.get("answer", "")).strip()]
if empty:
raise ValueError(f"Refusing submission: {len(empty)} answer(s) are empty")
return [
{"task_id": str(row["task_id"]), "submitted_answer": str(row["answer"])}
for row in results
]
|