Add ArxivRollBench scenario

src/helm/benchmark/run_specs/arxivrollbench_run_specs.py

@@ -0,0 +1,37 @@
+from helm.benchmark.adaptation.common_adapter_specs import get_multiple_choice_joint_adapter_spec
+from helm.benchmark.metrics.common_metric_specs import get_exact_match_metric_specs
+from helm.benchmark.run_spec import RunSpec, run_spec_function
+from helm.benchmark.scenarios.scenario import ScenarioSpec
+
+
+@run_spec_function("arxivrollbench")
+def get_arxivrollbench_spec(
+    release: str = "all",
+    domain: str = "all",
+    task_type: str = "all",
+    split: str = "compact",
+) -> RunSpec:
+    scenario_spec = ScenarioSpec(
+        class_name="helm.benchmark.scenarios.arxivrollbench_scenario.ArxivRollBenchScenario",
+        args={
+            "release": release,
+            "domain": domain,
+            "task_type": task_type,
+            "split": split,
+        },
+    )
+    adapter_spec = get_multiple_choice_joint_adapter_spec(
+        instructions="Answer the following scientific text reasoning questions with a single letter only.",
+        input_noun="Question",
+        output_noun="Answer",
+        max_train_instances=0,
+        max_tokens=5,
+    )
+    metric_specs = get_exact_match_metric_specs()
+    return RunSpec(
+        name=f"arxivrollbench:release={release},domain={domain},task_type={task_type},split={split}",
+        scenario_spec=scenario_spec,
+        adapter_spec=adapter_spec,
+        metric_specs=metric_specs,
+        groups=["arxivrollbench"],
+    )

src/helm/benchmark/scenarios/arxivrollbench_scenario.py

@@ -0,0 +1,186 @@
+import os
+import re
+from typing import List, Literal, Optional, Tuple
+
+import datasets
+
+from helm.benchmark.presentation.taxonomy_info import TaxonomyInfo
+from helm.benchmark.scenarios.scenario import (
+    CORRECT_TAG,
+    TEST_SPLIT,
+    Input,
+    Instance,
+    Output,
+    Reference,
+    Scenario,
+    ScenarioMetadata,
+)
+from helm.common.general import ensure_directory_exists
+
+
+DOMAINS: List[Tuple[str, str]] = [
+    ("cs", "cs"),
+    ("q_fin", "q-fin"),
+    ("math", "math"),
+    ("physics", "physics"),
+    ("stat", "stat"),
+    ("q_bio", "q-bio"),
+    ("econ", "econ"),
+    ("eess", "eess"),
+]
+DOMAIN_ALIASES = {
+    "q-fin": "q_fin",
+    "q-bio": "q_bio",
+}
+RELEASES: List[str] = ["2024b", "2025a", "2026a"]
+TASK_TYPES: List[str] = ["s", "c", "p"]
+TASK_TYPE_NAMES = {
+    "s": "sequencing",
+    "c": "cloze",
+    "p": "prediction",
+}
+
+
+def _dataset_path(
+    release: str,
+    hf_domain: str,
+    task_type: str,
+    split: Literal["compact", "full"],
+) -> str:
+    suffix = "-50" if split == "compact" else ""
+    if release == "2024b":
+        return f"liangzid/robench2024b_all_set{hf_domain}SCP-{task_type}{suffix}"
+    return f"liangzid/robench{release}_test_all_category_set" f"{hf_domain}SCP-{task_type}{suffix}"
+
+
+def _selection_to_letter(label: str) -> str:
+    match = re.search(r"\bselection\s*([1-4])\b", str(label), re.IGNORECASE)
+    if match:
+        return chr(ord("A") + int(match.group(1)) - 1)
+    return str(label).strip().upper()
+
+
+def _record_to_instance(record: dict, release: str, domain: str, task_type: str) -> Instance:
+    if task_type == "p":
+        input_text = (
+            "Given the context, select the text that is the next sequence.\n\n" f"Context:\n{record['context']}"
+        )
+        correct_letter = str(record["label"]).strip().upper()
+    else:
+        input_text = (
+            "Select the option that correctly completes the sequencing or cloze task.\n\n" f"{record['shuffled_text']}"
+        )
+        correct_letter = _selection_to_letter(record["label"])
+
+    references: List[Reference] = []
+    for letter in ["A", "B", "C", "D"]:
+        references.append(
+            Reference(
+                output=Output(text=str(record[letter]).strip()),
+                tags=[CORRECT_TAG] if letter == correct_letter else [],
+            )
+        )
+
+    return Instance(
+        input=Input(text=input_text),
+        references=references,
+        split=TEST_SPLIT,
+        extra_data={
+            "release": release,
+            "domain": domain,
+            "task_type": task_type,
+            "task_type_name": TASK_TYPE_NAMES[task_type],
+            "source_label": record["label"],
+        },
+    )
+
+
+class ArxivRollBenchScenario(Scenario):
+    """
+    ArxivRollBench is a rolling arXiv benchmark for evaluating recent scientific
+    text reasoning. It covers sequencing, cloze, and next-sequence prediction
+    tasks across arXiv domains and releases.
+
+    Paper: https://ojs.aaai.org/index.php/AAAI/article/view/41098
+    Website: https://arxivrollbench.github.io/
+    """
+
+    name = "arxivrollbench"
+    description = "A rolling benchmark for recent scientific text reasoning from arXiv papers"
+    tags = ["reasoning", "science", "multiple_choice"]
+
+    def __init__(
+        self,
+        release: str = "all",
+        domain: str = "all",
+        task_type: str = "all",
+        split: Literal["compact", "full"] = "compact",
+    ):
+        super().__init__()
+        if release != "all" and release not in RELEASES:
+            raise ValueError(f"Unknown release: {release}")
+        domain = DOMAIN_ALIASES.get(domain, domain)
+        valid_domains = {domain_name for domain_name, _ in DOMAINS}
+        if domain != "all" and domain not in valid_domains:
+            raise ValueError(f"Unknown domain: {domain}")
+        if task_type != "all" and task_type not in TASK_TYPES:
+            raise ValueError(f"Unknown task_type: {task_type}")
+        if split not in {"compact", "full"}:
+            raise ValueError(f"Unknown split: {split}")
+
+        self.release = release
+        self.domain = domain
+        self.task_type = task_type
+        self.split = split
+
+    def _iter_subsets(self) -> List[Tuple[str, str, str, str]]:
+        releases = RELEASES if self.release == "all" else [self.release]
+        domains = DOMAINS if self.domain == "all" else [(self.domain, self.domain.replace("_", "-"))]
+        task_types = TASK_TYPES if self.task_type == "all" else [self.task_type]
+        return [
+            (release, domain, hf_domain, task_type)
+            for release in releases
+            for domain, hf_domain in domains
+            for task_type in task_types
+        ]
+
+    def get_instances(self, output_path: str) -> List[Instance]:
+        cache_dir = os.path.join(output_path, "data")
+        ensure_directory_exists(cache_dir)
+
+        instances: List[Instance] = []
+        for release, domain, hf_domain, task_type in self._iter_subsets():
+            dataset = datasets.load_dataset(
+                _dataset_path(release, hf_domain, task_type, self.split),
+                split="train",
+                cache_dir=cache_dir,
+            )
+            assert isinstance(dataset, datasets.Dataset)
+            for record in dataset:
+                instances.append(_record_to_instance(record, release, domain, task_type))
+        return instances
+
+    def get_metadata(self) -> ScenarioMetadata:
+        task_type_display: Optional[str] = None
+        if self.task_type != "all":
+            task_type_display = TASK_TYPE_NAMES[self.task_type]
+        return ScenarioMetadata(
+            name=self.name,
+            display_name="ArxivRollBench",
+            short_display_name="ArxivRollBench",
+            description=(
+                "ArxivRollBench is a rolling benchmark for evaluating recent scientific "
+                "text reasoning over arXiv papers. It covers sequencing, cloze, and "
+                "next-sequence prediction tasks across arXiv domains and releases "
+                "[(AAAI 2026 paper)](https://ojs.aaai.org/index.php/AAAI/article/view/41098)."
+            ),
+            taxonomy=TaxonomyInfo(
+                task=task_type_display or "multiple-choice scientific text reasoning",
+                what="recent scientific text from arXiv papers",
+                when="rolling releases from 2024b, 2025a, and 2026a",
+                who="arXiv papers across computer science, math, physics, statistics, and related domains",
+                language="English",
+            ),
+            main_metric="exact_match",
+            main_split="test",
+        )

src/helm/benchmark/scenarios/test_arxivrollbench_scenario.py

@@ -0,0 +1,59 @@
+from helm.benchmark.scenarios.arxivrollbench_scenario import (
+    ArxivRollBenchScenario,
+    _dataset_path,
+    _record_to_instance,
+    _selection_to_letter,
+)
+from helm.benchmark.scenarios.scenario import CORRECT_TAG, TEST_SPLIT
+
+
+def test_dataset_path_compact_and_full():
+    assert _dataset_path("2026a", "cs", "s", "compact") == "liangzid/robench2026a_test_all_category_setcsSCP-s-50"
+    assert _dataset_path("2024b", "q-fin", "p", "full") == "liangzid/robench2024b_all_setq-finSCP-p"
+
+
+def test_selection_to_letter():
+    assert _selection_to_letter("Selection 1") == "A"
+    assert _selection_to_letter("selection 4") == "D"
+    assert _selection_to_letter("A") == "A"
+    assert _selection_to_letter("1") == "1"
+
+
+def test_domain_aliases():
+    scenario = ArxivRollBenchScenario(release="2026a", domain="q-bio", task_type="s")
+
+    assert scenario.domain == "q_bio"
+
+
+def test_record_to_instance_prediction():
+    record = {
+        "context": "The introduction describes a new method.",
+        "A": "A candidate",
+        "B": "B candidate",
+        "C": "C candidate",
+        "D": "D candidate",
+        "label": "C",
+    }
+
+    instance = _record_to_instance(record, "2026a", "cs", "p")
+
+    assert instance.split == TEST_SPLIT
+    assert "Context:\nThe introduction describes a new method." in instance.input.text
+    assert [reference.tags for reference in instance.references] == [[], [], [CORRECT_TAG], []]
+    assert instance.extra_data["task_type_name"] == "prediction"
+
+
+def test_record_to_instance_selection():
+    record = {
+        "shuffled_text": "Paragraph with a blank.",
+        "A": "A candidate",
+        "B": "B candidate",
+        "C": "C candidate",
+        "D": "D candidate",
+        "label": "Selection 2",
+    }
+
+    instance = _record_to_instance(record, "2026a", "math", "s")
+
+    assert [reference.tags for reference in instance.references] == [[], [CORRECT_TAG], [], []]
+    assert instance.extra_data["task_type_name"] == "sequencing"