Groups¶

def add(self, item: Task | Group) -> None:
    """Add a task or subgroup to this group."""
    key: str = item._qualified_name
    self._children[key] = item

def pop(self, name: str) -> Group | Task | None:
    """Pop a child by name."""
    return self._children.pop(name, None)

def get(self, name: str) -> Task | Group | None:
    """Get a child by name."""
    return self._children.get(name)

def __contains__(self, name: str) -> bool:
    """Check if a child exists by name."""
    return name in self._children

def __iter__(self):
    """Iterate over direct children (Task or Group objects)."""
    return iter(self._children.values())

def __len__(self) -> int:
    """Number of direct children."""
    return len(self._children)

def get_all_tasks(self, recursive: bool = True) -> list[Task]:
    """Get all leaf Task objects.

    Args:
        recursive: If True, include tasks from nested subgroups.
                   If False, only return direct Task children.

    Returns:
        List of Task objects (not Groups).
    """
    from lm_eval.api.task import Task

    tasks = []
    for item in self._children.values():
        if isinstance(item, Task):
            tasks.append(item)
        elif isinstance(item, Group) and recursive:
            tasks.extend(item.get_all_tasks(recursive=True))
    return tasks

def get_all_groups(self, recursive: bool = True) -> list[Group]:
    """Get all subgroups.

    Args:
        recursive: If True, include nested subgroups.
                   If False, only return direct Group children.

    Returns:
        List of Group objects (not including self).
    """
    groups = []
    for item in self._children.values():
        if isinstance(item, Group):
            groups.append(item)
            if recursive:
                groups.extend(item.get_all_groups(recursive=True))
    return groups

def aggregate(self, task_metrics: Mapping[str, _TaskMetrics]) -> _TaskMetrics:
    """Aggregate metrics for this group from its leaf task results.

    Args:
        task_metrics: {task_name: {metric_key: value, "sample_len": int, ...}}
            The full flat metrics dict (all tasks). This group only reads
            entries for its own leaf tasks (via ``get_all_tasks()``).

    Returns:
        Aggregated metrics dict for this group, e.g.
            ``{"alias": str, "acc,none": float, "acc_stderr,none": float, "sample_len": int, ...}``
    """
    from lm_eval.api.metrics import aggregate_subtask_metrics, pooled_sample_stderr

    group_metrics: dict[str, Any] = {
        "alias": self.alias or self.name,
        "name": self.name,
    }

    if not self.aggregate_metric_list:
        return cast("_TaskMetrics", group_metrics)

    # Get leaf task names
    leaf_tasks = [t._qualified_name for t in self.get_all_tasks()]

    # group-level sample len. Not used for weighting, but useful metadata
    # Compute total sample_len once (across all leaf tasks), not per-filter
    group_metrics["sample_len"] = sum(
        task_metrics[name].get("sample_len", 0)
        for name in leaf_tasks
        if name in task_metrics
    )
    sample_count: dict[str, int] = {}

    for agg_config in self.aggregate_metric_list:
        # Determine filters: auto-discover if None, else use explicit list
        if agg_config.filter_list is None:
            filters_to_aggregate = self._discover_filters_for_metric(
                agg_config.metric, task_metrics
            )
        else:
            filters_to_aggregate = agg_config.filter_list

        for filter_name in filters_to_aggregate:
            metric_key = str(MetricKey(agg_config.metric, filter_name))
            stderr_key = str(
                MetricKey(agg_config.metric, filter_name, is_stderr=True)
            )

            # Gather values from leaf tasks
            values: list[float] = []
            stderrs: list[float] = []
            sizes: list[int] = []
            tasks_with_metric: list[str] = []
            tasks_without_metric: list[str] = []

            for task_name in leaf_tasks:
                if task_name not in task_metrics:
                    tasks_without_metric.append(task_name)
                    continue
                task_result = task_metrics[task_name]
                if metric_key in task_result:
                    values.append(task_result[metric_key])  # type:ignore[invalid-key]
                    sizes.append(task_result.get("sample_len", 0))
                    tasks_with_metric.append(task_name)
                    stderr_val = task_result.get(stderr_key)
                    if stderr_val is not None:
                        stderrs.append(stderr_val)
                else:
                    tasks_without_metric.append(task_name)

            # Log warning if a metric is missing in some tasks
            if values and tasks_without_metric:
                missing_names = ", ".join(tasks_without_metric[:5])
                overflow = (
                    f" and {len(tasks_without_metric) - 5} more"
                    if len(tasks_without_metric) > 5
                    else ""
                )
                eval_logger.warning(
                    "Group '%s': metric '%s' is missing in %d/%d tasks. Missing in: %s%s",
                    self.name,
                    metric_key,
                    len(tasks_without_metric),
                    len(leaf_tasks),
                    missing_names,
                    overflow,
                )

            if not values:
                eval_logger.warning(
                    "Group '%s': no values found for metric '%s' across any tasks.",
                    self.name,
                    metric_key,
                )

            if values:
                group_metrics[metric_key] = aggregate_subtask_metrics(
                    values, sizes, agg_config.weight_by_size
                )
                sample_count[metric_key] = sum(sizes)

                if len(stderrs) == len(values) and "N/A" not in stderrs:
                    group_metrics[stderr_key] = pooled_sample_stderr(stderrs, sizes)
                else:
                    group_metrics[stderr_key] = "N/A"

    if sample_count:
        group_metrics["sample_count"] = sample_count

    return cast("_TaskMetrics", group_metrics)

def to_dict(self) -> dict[str, Any] | None:
    """Convert to dictionary for serialization."""
    if self._config:
        return self._config.to_dict()
    result: dict[str, Any] = {"group": self.name}
    if self._children:
        result["task"] = list(self._children.keys())
    if self.alias:
        result["group_alias"] = self.alias
    if self.aggregate_metric_list:
        result["aggregate_metric_list"] = [
            asdict(agg) for agg in self.aggregate_metric_list
        ]
    if self.metadata:
        result["metadata"] = self.metadata
    return result

@classmethod
def from_config(cls, config: GroupConfig | dict[str, Any]) -> Group:
    """Create a Group from a GroupConfig or raw dict (e.g., parsed from YAML).

    Note: This only creates the Group shell. Children must be added
    separately via group.add() after Tasks/subGroups are built.
    """
    if isinstance(config, dict):
        config = GroupConfig(**config)  # type:ignore[invalid-argument-type]

    return cls(
        name=config.group,
        alias=config.group_alias,
        aggregate_metric_list=cast(
            "list[AggMetricConfig]", config.aggregate_metric_list
        ),
        metadata=config.metadata,
        _config=config,
    )

def __repr__(self):
    return f"Group(name={self.name!r}, len_children={len(self._children)}, children={self.child_names}, version={self.version})"