lm_eval.tasks¶

def load(
    self,
    specs: Sequence[str | Mapping[str, Any]],
    overrides: Mapping[str, Mapping[str, Any]] | None = None,
) -> TaskDict:
    """Resolve task specs into concrete [Task][lm_eval.api.task.Task] and [Group][lm_eval.api.group.Group] objects.

    Accepts name strings, config dicts, or a mix. Groups and
    tags references are expanded into their leaf tasks.

    Example:
        ```python
        loaded = task_manager.load(
            ["mmlu", "arc_easy"],
            overrides={
                "arc_easy": {"num_fewshot": 5},
                "mmlu": {"num_fewshot": 3},
            },
        )
        loaded["tasks"]  # {"mmlu_..": Task, ... "arc_easy": Task, ...}
        loaded["groups"]  # {"mmlu": Group}
        ```

    Args:
        specs: One or more task specs — a name string or a full config
            dict (e.g. ``{"task": "arc_easy", "doc_to_text": ..., ...}``).
        overrides: Optional mapping of task/group name to config overrides
            (e.g. ``{"arc_easy": {"num_fewshot": 5}}``). Only applied to
            string specs; dict specs carry their overrides inline.

    Returns:
        A [TaskDict][TaskDict] containing the loaded ``"tasks"``, ``"groups"``,
        and a ``"group_map"`` of each group to its immediate children.
    """
    if not isinstance(specs, list):
        specs = [specs]  # type: ignore
    _overrides = cast("dict[str, dict[str, Any]]", deepcopy(overrides or {}))

    # Build all requested items
    built: list[Task | Group] = []
    for spec in cast("Iterable", specs):
        # Dict specs are self-contained — they carry overrides inline
        # String specs look up by name in the overrides mapping
        spec_overrides = {} if isinstance(spec, dict) else _overrides.pop(spec, {})

        obj = self._load_spec(spec, overrides=spec_overrides)
        # Tags return list[Task], flatten
        if isinstance(obj, list):
            obj = cast("list[Task]", obj)
            built.extend(obj)
        else:
            built.append(obj)

    # Flatten to task/group dicts
    tasks: dict[str, Task] = {}
    groups: dict[str, Group] = {}

    def collect(item: Task | Group) -> None:
        if isinstance(item, Group):
            groups[item.name] = item
            for task in item.get_all_tasks():
                tasks[task._qualified_name] = task
            for subgroup in item.get_all_groups():
                groups[subgroup.name] = subgroup
        else:
            tasks[item._qualified_name] = item

    for item in built:
        collect(item)

    if _overrides:
        eval_logger.warning(
            "Unused overrides (no matching spec): %s",
            ", ".join(sorted(_overrides)),
        )

    return {
        "tasks": tasks,
        "groups": groups,
        "group_map": {g.name: g.child_names for g in groups.values()}
        if groups
        else {},
    }

@deprecated("Use TaskManager.load(), which returns flat dicts of tasks and groups.")
def load_task_or_group(self, task_list: str | list[str | dict]) -> dict:
    """Deprecated — use [load][.load] instead.

    Returns the old nested-dict format where groups are keyed by
    [ConfigurableGroup][lm_eval.api.group.ConfigurableGroup] and
    standalone tasks by name.

    Args:
        task_list: Task name(s) or override dicts.

    Returns:
        Nested dict — groups are keyed by ``ConfigurableGroup`` objects, standalone
            tasks by name. Subgroups recurse, e.g.
            ``{CG: {sub_CG: {task: Task, ...}, task: Task, ...}, ...}``.
    """
    import collections

    from lm_eval.api.group import ConfigurableGroup

    if isinstance(task_list, str):
        task_list = [task_list]

    def _to_nested(obj: Task | Group | list[Task]) -> dict:
        """Convert Task | Group | list[Task] to legacy nested dict format."""
        if isinstance(obj, list):
            return {t.task_name: t for t in obj}  # type:ignore
        if isinstance(obj, Group):
            nested: dict[str, Any] = {}
            for child in obj:
                if isinstance(child, Group):
                    nested.update(_to_nested(child))
                else:
                    nested[child.task_name] = child
            cg = ConfigurableGroup.from_group(obj)
            return {cg: nested}
        return {obj.task_name: obj}

    return dict(
        collections.ChainMap(*[_to_nested(self._load_spec(s)) for s in task_list])
    )

def match_tasks(self, task_list: list[str]) -> list[str]:
    """Match task names using glob patterns.

    Handles task@format syntax: strips @format for matching,
    returns the original (with @format) so _load_spec can parse it.
    """
    results = []
    for pattern in task_list:
        if "@" in pattern:
            base, preset_suffix = pattern.split("@", 1)
            matched = utils.pattern_match([base], self.all_tasks)
            results.extend(f"{m}@{preset_suffix}" for m in matched)
        else:
            matched = utils.pattern_match([pattern], self.all_tasks)
            results.extend(matched)
        if not matched:
            results.append(pattern)
    return sorted(set(results))

def list_all_tasks(
    self,
    list_groups: bool = True,
    list_tags: bool = True,
    list_subtasks: bool = True,
) -> str:
    """Generate a Markdown table listing all available tasks."""
    from pytablewriter import MarkdownTableWriter

    def sanitize_path(path):
        if path is None:
            return "---"
        path_str = str(path)
        if "lm_eval/tasks/" in path_str:
            return "lm_eval/tasks/" + path_str.split("lm_eval/tasks/")[-1]
        return path_str

    group_table = MarkdownTableWriter()
    group_table.headers = ["Group", "Config Location"]
    gt_values = []
    for g in self.all_groups:
        entry = self._index[g]
        path = sanitize_path(entry.yaml_path)
        gt_values.append([g, path])
    group_table.value_matrix = gt_values

    tag_table = MarkdownTableWriter()
    tag_table.headers = ["Tag"]
    tag_table.value_matrix = [[t] for t in self.all_tags]

    subtask_table = MarkdownTableWriter()
    subtask_table.headers = ["Task", "Config Location", "Output Type"]
    st_values = []
    for t in self.all_subtasks:
        entry = self._index[t]
        path = entry.yaml_path
        output_type = ""

        if path is not None:
            config = load_yaml(path, resolve_func=False, recursive=True)
            if "output_type" in config:
                output_type = config["output_type"]

        path = sanitize_path(path)
        st_values.append([t, path, output_type])
    subtask_table.value_matrix = st_values

    result = "\n"
    if list_groups:
        result += group_table.dumps() + "\n\n"
    if list_tags:
        result += tag_table.dumps() + "\n\n"
    if list_subtasks:
        result += subtask_table.dumps() + "\n\n"
    return result