Evaluation Config¶

@classmethod
def from_cli(cls, namespace: Namespace) -> EvaluatorConfig:
    """Build an EvaluationConfig by merging with a simple precedence.

    CLI args > YAML config > built-in defaults.
    """
    # Start with built-in defaults
    config = asdict(cls())

    # Load and merge YAML config if provided
    if used_config := getattr(namespace, "config", None):
        config.update(cls.load_yaml_config(used_config))

    # Override with CLI args (skip None = "not provided", exclude non-config args)
    excluded_args = {"command", "func"}  # argparse internal args
    cli_args = {
        k: v
        for k, v in vars(namespace).items()
        if v is not None and k not in excluded_args
    }
    config.update(cli_args)

    # Create an instance and validate
    instance = cls(**config)._parse_dict_args()
    instance._configure()

    if used_config:
        cli_args.pop("config", None)
        eval_logger.info(
            "CLI args %s will override yaml", cli_args
        ) if cli_args else None
        print(textwrap.dedent(f"""{instance}"""))

    return instance

@classmethod
def from_config(cls, config_path: str | Path) -> EvaluatorConfig:
    """Build an EvaluationConfig from a YAML config file.

    Merges with built-in defaults and validates.
    """
    # Load YAML config
    yaml_config = cls.load_yaml_config(config_path)
    return cls(**yaml_config)._configure()

@staticmethod
def load_yaml_config(config_path: str | Path) -> dict[str, Any]:
    """Load and validate YAML config file."""
    _config_path = Path(config_path)
    if not _config_path.is_file():
        raise FileNotFoundError(f"Config file not found: {_config_path.resolve()}")

    try:
        yaml_data = yaml.safe_load(_config_path.read_text())
    except yaml.YAMLError as e:
        raise ValueError(f"Invalid YAML in {_config_path}: {e}") from e
    except (OSError, UnicodeDecodeError) as e:
        raise ValueError(f"Could not read config file {_config_path}: {e}") from e

    if not isinstance(yaml_data, dict):
        raise TypeError(
            f"YAML root must be a mapping in {_config_path.resolve()}, got {type(yaml_data).__name__}"
        )

    return yaml_data

def process_tasks(self, metadata: dict | None = None) -> TaskManager:
    """Process and validate tasks, return resolved task names.

    Handles:
    - Task names (e.g., "hellaswag", "arc_easy")
    - Custom YAML config files (e.g., "/path/to/task.yaml")
    - Glob patterns (e.g., "/path/to/*.yaml")
    - Directories of YAML files
    """
    import itertools

    from lm_eval.tasks import TaskManager
    from lm_eval.tasks._yaml_loader import load_yaml

    # if metadata manually passed use that:
    self.metadata = metadata or self.metadata

    # Create a task manager with metadata
    task_manager = TaskManager(
        include_path=self.include_path,
        include_defaults=self.include_defaults,
        metadata=self.metadata or {},
    )

    # Normalize tasks to a list
    # We still allow tasks in the form task1,task2
    task_list = (
        [t.strip() for t in self.tasks.split(",")]
        if isinstance(self.tasks, str)
        else [t.strip() for task in self.tasks for t in task.split(",")]
    )

    # Handle directory input
    if len(task_list) == 1 and (yaml_path := Path(task_list[0])).is_dir():
        task_names = []
        for yaml_file in yaml_path.glob("*.yaml"):
            config = load_yaml(yaml_file, resolve_func=False)
            task_names.append(config)
        self.tasks = task_names
        return task_manager

    # Normalize paths and deduplicate
    task_list = [
        str(Path(task).absolute()) if task.endswith(".yaml") else task
        for task in task_list
    ]
    match_dict: dict[str, list] = {}

    # Match each task
    for task in task_list:
        if not task.endswith(".yaml"):
            # Standard task name - match via task manager
            matches = task_manager.match_tasks([task])
        else:
            # Custom config file(s) - support glob patterns
            matches = []
            for yaml_file in (task_path := Path(task)).parent.glob(task_path.name):
                config = load_yaml(yaml_file, resolve_func=False)
                matches.append(config)
        match_dict[task] = matches

    # Flatten and deduplicate results
    task_names = []
    for task in itertools.chain.from_iterable(match_dict.values()):
        if task not in task_names:
            task_names.append(task)

    # Check for missing tasks
    task_missing = [task for task, matches in match_dict.items() if not matches]
    if task_missing:
        missing = ", ".join(task_missing)
        raise ValueError(f"Tasks not found: {missing}")

    # Update tasks with resolved names
    self.tasks = task_names
    return task_manager