| | """
|
| | Inspect AI task definition that runs the existing agent and reuses the rubric scorer.
|
| | """
|
| |
|
| | from __future__ import annotations
|
| |
|
| | import asyncio
|
| | import json
|
| | import sys
|
| | from pathlib import Path
|
| | from typing import Any, Sequence
|
| |
|
| | from inspect_ai import Task, task
|
| | from inspect_ai.dataset import Sample, hf_dataset
|
| | from inspect_ai.scorer import Score, Target, mean, scorer
|
| | from inspect_ai.solver._task_state import TaskState
|
| | import litellm
|
| |
|
| | PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
| | if str(PROJECT_ROOT) not in sys.path:
|
| | sys.path.insert(0, str(PROJECT_ROOT))
|
| |
|
| | from eval.rubric_eval import RubricData, evaluate_with_rubrics
|
| | from eval.solvers import get_solver
|
| |
|
| |
|
| | def _record_to_sample(record: dict[str, Any]) -> Sample:
|
| | rubric_payload = json.loads(record["rubric"])
|
| | rubrics = rubric_payload.get("rubrics", [])
|
| |
|
| | metadata = {
|
| | "question": record["question"],
|
| | "discussion_title": record.get("discussion_title"),
|
| | "discussion_url": record.get("discussion_url"),
|
| | "rubric_title": rubric_payload.get("title"),
|
| | "rubric_description": rubric_payload.get("description"),
|
| | "rubrics": rubrics,
|
| | }
|
| |
|
| | return Sample(
|
| | input=record["question"],
|
| | target=record["solution"],
|
| | id=record.get("discussion_topic_id"),
|
| | metadata=metadata,
|
| | )
|
| |
|
| |
|
| | def _load_dataset(dataset_name: str, split: str, limit: int | None) -> Sequence[Sample]:
|
| | return hf_dataset(
|
| | dataset_name, sample_fields=_record_to_sample, split=split, limit=limit
|
| | )
|
| |
|
| |
|
| | def _metadata_to_rubrics(metadata: dict[str, Any]) -> list[RubricData]:
|
| | raw_rubrics = metadata.get("rubrics", [])
|
| | return [RubricData(**rubric) for rubric in raw_rubrics]
|
| |
|
| |
|
| | @scorer(metrics=[mean()], name="rubric_scorer")
|
| | def rubric_scorer(judge_model: str = "gpt-5-mini"):
|
| | async def score(state: TaskState, target: Target) -> Score:
|
| | response_text = state.output.completion or state.output.message.text
|
| | question = state.metadata.get("question", state.input_text)
|
| | rubrics = _metadata_to_rubrics(state.metadata)
|
| |
|
| | evaluation = await asyncio.to_thread(
|
| | evaluate_with_rubrics,
|
| | question,
|
| | response_text,
|
| | rubrics,
|
| | judge_model,
|
| | )
|
| |
|
| | score_metadata = {
|
| | "raw_score": evaluation.raw_score,
|
| | "criterion_checks": [
|
| | check.model_dump() for check in evaluation.criterion_checks
|
| | ],
|
| | "discussion_title": state.metadata.get("discussion_title"),
|
| | "discussion_url": state.metadata.get("discussion_url"),
|
| | "reference_answer": target.text,
|
| | }
|
| |
|
| | return Score(
|
| | value=evaluation.normalized_score,
|
| | answer=response_text,
|
| | explanation=f"Normalized score {evaluation.normalized_score:.3f}",
|
| | metadata=score_metadata,
|
| | )
|
| |
|
| | return score
|
| |
|
| |
|
| | @task(name="hf-benchmark-with-rubrics")
|
| | def hf_benchmark_with_rubrics(
|
| | solver_name: str = "hf_agent",
|
| | solver_kwargs: dict[str, Any] = {
|
| | "max_iterations": 10,
|
| | "config_path": "agent/config_mcp_example.json",
|
| | },
|
| | dataset_name: str = "akseljoonas/hf-agent-rubrics@train",
|
| | limit: int | None = None,
|
| | judge_model: str = "gpt-5-mini",
|
| | ) -> Task:
|
| | litellm.drop_params = True
|
| | if "@" not in dataset_name:
|
| | raise ValueError("Dataset name must be in the format 'author/dataset@split'")
|
| | dataset_name, dataset_split = dataset_name.split("@")
|
| | dataset = _load_dataset(dataset_name, dataset_split, limit=limit)
|
| |
|
| | return Task(
|
| | dataset=dataset,
|
| | solver=get_solver(solver_name, **solver_kwargs),
|
| | scorer=rubric_scorer(judge_model=judge_model),
|
| | metadata={
|
| | "dataset_name": dataset_name,
|
| | "dataset_split": dataset_split,
|
| | "solver_name": solver_name,
|
| | "judge_model": judge_model,
|
| | },
|
| | )
|
| |
|
