Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions lmms_eval/caching/cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,10 +44,13 @@ def save_to_cache(file_name, obj):
serializable_obj = []

for item in obj:
serializable_item = []
for subitem in item:
if hasattr(subitem, "arguments"): # we need to handle the arguments specially since doc_to_visual is callable method and not serializable
serializable_arguments = tuple(arg if not callable(arg) else None for arg in subitem.arguments)
subitem.arguments = serializable_arguments
serializable_item.append(subitem)
serializable_obj.append(serializable_item)

eval_logger.debug(f"Saving {file_path} to cache...")
try:
Expand Down
19 changes: 13 additions & 6 deletions lmms_eval/caching/response_cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -407,14 +407,21 @@ def create(
cache_root = os.path.dirname(os.path.abspath(cache_root))
cache_root = os.path.abspath(cache_root)

# Task fingerprints
# Task fingerprints — recursively process nested groups so subtasks
# get proper fingerprints instead of falling back to empty strings.
task_fingerprints = {}
if task_dict:
for tname, tobj in task_dict.items():
if hasattr(tobj, "dump_config"):
cfg_str = json.dumps(tobj.dump_config(), sort_keys=True, default=str)
cfg_str = _FUNC_ADDR_RE.sub(">", cfg_str)
task_fingerprints[tname] = hash_string(cfg_str)[:16]

def _collect_fingerprints(d):
for tname, tobj in d.items():
if hasattr(tobj, "dump_config"):
cfg_str = json.dumps(tobj.dump_config(), sort_keys=True, default=str)
cfg_str = _FUNC_ADDR_RE.sub(">", cfg_str)
task_fingerprints[tname] = hash_string(cfg_str)[:16]
elif isinstance(tobj, dict):
_collect_fingerprints(tobj)

_collect_fingerprints(task_dict)

# Model fingerprint
if isinstance(model_args, dict):
Expand Down
82 changes: 77 additions & 5 deletions lmms_eval/evaluator.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,43 @@
)

IMAGE_EXTENSIONS = (".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".tif", ".tiff")
_SENSITIVE_CONFIG_KEYS = {
"api_key",
"client_secret",
"hf_token",
"huggingface_hub_token",
"huggingfacehub_api_token",
"token",
}
# Suffixes also catch prefixed variants: openai_api_key, github_token, my_secret, db_password.
_SENSITIVE_KEY_SUFFIXES = ("api_key", "_token", "_secret", "password")
_SECRET_ASSIGNMENT_RE = re.compile(r"(?i)(^|[,\s])(\w*(?:api_key|token|secret|password))=([^,\s]+)")
_HF_TOKEN_VALUE_RE = re.compile(r"\bhf_[A-Za-z0-9]{20,}\b")


def _is_sensitive_config_key(key) -> bool:
key_lower = str(key).lower()
return key_lower in _SENSITIVE_CONFIG_KEYS or key_lower.endswith(_SENSITIVE_KEY_SUFFIXES)


def _redact_eval_config_secrets(value):
if isinstance(value, dict):
return {key: ("[REDACTED]" if _is_sensitive_config_key(key) else _redact_eval_config_secrets(item)) for key, item in value.items()}
if isinstance(value, (list, tuple)):
return type(value)(_redact_eval_config_secrets(item) for item in value)
if isinstance(value, str):
value = _SECRET_ASSIGNMENT_RE.sub(r"\1\2=[REDACTED]", value)
return _HF_TOKEN_VALUE_RE.sub("[REDACTED]", value)
return value


def _clone_padding_request(pad_source: Instance) -> Instance:
pad_instance = copy.copy(pad_source)
pad_instance.metadata = dict(pad_source.metadata or {})
pad_instance.metadata["__padding_only__"] = True
pad_instance.resps = []
pad_instance.token_counts = []
return pad_instance


def _enable_reentrant_filelocks() -> None:
Expand Down Expand Up @@ -531,6 +568,7 @@ def _adjust_config(task_dict):
# add info about execution
results["config"].update(
{
"model_backend": f"{type(lm).__module__}.{type(lm).__name__} ({task_type})",
"batch_size": batch_size,
"batch_sizes": (list(lm.batch_sizes.values()) if hasattr(lm, "batch_sizes") else []),
"device": device,
Expand All @@ -556,6 +594,8 @@ def _adjust_config(task_dict):
resolved[key] = str(value)
results["config"]["resolved_cli_args"] = resolved

results["config"] = _redact_eval_config_secrets(results["config"])

results["git_hash"] = get_git_commit_hash()
results["git_branch"] = get_git_branch_name()
results["lmms_eval_version"] = get_lmms_eval_version_string()
Expand Down Expand Up @@ -862,6 +902,7 @@ def evaluate(
local_rank = int(os.environ.get("LOCAL_RANK", 0))
global_rank = int(os.environ.get("RANK", 0))
world_size = int(os.environ.get("WORLD_SIZE", 1))

eval_logger.info(f"Running on rank {global_rank} (local rank {local_rank})")

def _infer_task_request_type(task_obj: Task) -> Optional[str]:
Expand All @@ -882,8 +923,32 @@ def _infer_task_request_type(task_obj: Task) -> Optional[str]:
if not all("bypass" not in getattr(task_output.task, "_metric_fn_list", {}).keys() for task_output in eval_tasks):
raise ValueError("log_samples must be True for 'bypass' metric-only tasks")

if distributed_executor_backend == "accelerate" and not hasattr(lm, "accelerator"):
lm.accelerator = Accelerator()
if distributed_executor_backend == "accelerate":
if not hasattr(lm, "accelerator"):
lm.accelerator = Accelerator()
else:
# Torchrun/native path: auto-init a process group for multi-rank runs
# whose model backend never called init_process_group (e.g. diffusers
# video-gen models), so evaluator collectives don't crash. gloo needs no
# GPU. Gated out of the accelerate path on purpose: pre-initializing here
# would make Accelerator() adopt this gloo group instead of building NCCL,
# silently running GPU collectives on CPU.
if world_size > 1 and dist.is_available() and not dist.is_initialized():
if os.environ.get("MASTER_ADDR") and os.environ.get("MASTER_PORT"):
dist.init_process_group(backend="gloo")
eval_logger.info(f"evaluator: auto-initialized gloo process group (world={world_size})")
else:
eval_logger.warning(
f"evaluator: WORLD_SIZE={world_size} but MASTER_ADDR/MASTER_PORT are unset; skipping torch.distributed auto-init. Distributed collectives will fail unless the model backend initializes the process group itself."
)

# Inject rank/world_size into the model so that logging (tqdm disable)
# and rank-conditional logic work on non-zero ranks. Many simple models
# only read LOCAL_RANK for device binding and leave _rank/_world_size
# at their defaults (0, 1).
if world_size > 1:
lm._rank = global_rank
lm._world_size = world_size

for task_output in eval_tasks:
task = task_output.task
Expand Down Expand Up @@ -950,8 +1015,8 @@ def _infer_task_request_type(task_obj: Task) -> Optional[str]:
instances_rnk = torch.tensor(len(task._instances), device=lm.device)
gathered_item = lm.accelerator.gather(instances_rnk).cpu().detach().numpy().tolist()
elif distributed_executor_backend == "torchrun":
instances_rnk = torch.tensor(len(task._instances), device=lm.device)
gathered_item = torch.zeros(world_size * 1, dtype=instances_rnk.dtype, device=lm.device)
instances_rnk = torch.tensor([len(task._instances)], device=lm.device)
gathered_item = torch.zeros(world_size, dtype=instances_rnk.dtype, device=lm.device)
dist.all_gather_into_tensor(gathered_item, instances_rnk)
gathered_item = gathered_item.cpu().detach().numpy().tolist()
else:
Expand Down Expand Up @@ -1003,7 +1068,8 @@ def _infer_task_request_type(task_obj: Task) -> Optional[str]:
eval_logger.warning(f"Running {reqtype} requests but could not find a pad source request on rank {global_rank}; skipping rank padding.")
else:
for _ in range(padding_requests[reqtype]):
cloned_reqs.extend([pad_source] * pad_source.repeats)
pad_instance = _clone_padding_request(pad_source)
cloned_reqs.extend([pad_instance] * pad_instance.repeats)

# run requests through model (with optional response cache)
if reqtype == "generate_until_agentic":
Expand Down Expand Up @@ -1124,6 +1190,12 @@ def _infer_task_request_type(task_obj: Task) -> Optional[str]:
pbar = tqdm(total=total_docs, desc="Postprocessing", disable=(RANK != 0))
for doc_id, doc in doc_iterator:
requests = instances_by_doc_id[doc_id]
# Defensive skip: if a doc landed in this rank's shard but has
# no requests (e.g. a split mismatch during a partial rerun),
# skip rather than crashing at `requests[0].doc` below.
if not requests:
pbar.update(1)
continue

# Strip reasoning tags before scoring
if reasoning_tags is not None:
Expand Down
120 changes: 105 additions & 15 deletions lmms_eval/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ def escaped_split(text, sep_char, maxsplit=-1):
return text
maxsplit = max(0, maxsplit)

return re.split(r"(?<!\\)" + sep_char, text, maxsplit)
return re.split(r"(?<!\\)" + sep_char, text, maxsplit=maxsplit)


def handle_arg_string(arg):
Expand Down Expand Up @@ -775,27 +775,65 @@ def run_task_tests(task_list: List[str]):
raise ValueError(f"Not all tests for the specified tasks ({task_list}) ran successfully! Error code: {pytest_return_val}")


def _lmms_eval_repo_root() -> pathlib.Path:
return pathlib.Path(__file__).resolve().parents[1]


def _git_output(*args: str) -> str:
return (
subprocess.check_output(
["git", "-C", str(_lmms_eval_repo_root()), *args],
stderr=subprocess.DEVNULL,
)
.strip()
.decode()
)


def _first_env(*names: str) -> str | None:
for name in names:
value = os.environ.get(name, "").strip()
if value:
return value
return None


def _normalize_branch_name(branch: str) -> str:
return branch.removeprefix("remotes/origin/")


def _is_exact_git_ref_name(branch: str) -> bool:
return bool(branch and branch != "undefined" and "^" not in branch and "~" not in branch)


def get_git_commit_hash():
"""
Gets the git commit hash of your current repo (if it exists).
Source: https://github.com/EleutherAI/gpt-neox/blob/b608043be541602170bfcfb8ec9bf85e8a0799e0/megatron/neox_arguments/neox_args.py#L42
"""
try:
git_hash = subprocess.check_output(["git", "describe", "--always"]).strip()
git_hash = git_hash.decode()
git_hash = _git_output("describe", "--always")
except (subprocess.CalledProcessError, FileNotFoundError):
# FileNotFoundError occurs when git not installed on system
git_hash = None
git_hash = _first_env("LMMS_EVAL_GIT_COMMIT", "GIT_COMMIT", "GITHUB_SHA")
return git_hash


def get_git_branch_name():
"""Gets the current git branch name (if in a repo)."""
try:
branch = subprocess.check_output(["git", "rev-parse", "--abbrev-ref", "HEAD"]).strip()
return branch.decode()
branch = _git_output("branch", "--show-current")
if branch:
return _normalize_branch_name(branch)
except (subprocess.CalledProcessError, FileNotFoundError):
pass

branch = _first_env("LMMS_EVAL_GIT_BRANCH", "GIT_BRANCH", "BRANCH_NAME", "GITHUB_REF_NAME")
if branch:
return _normalize_branch_name(branch)

try:
branch = _git_output("name-rev", "--name-only", "--exclude=tags/*", "HEAD")
if _is_exact_git_ref_name(branch):
return _normalize_branch_name(branch)
except (subprocess.CalledProcessError, FileNotFoundError):
return None
pass

return "detached" if get_git_commit_hash() else None


def get_lmms_eval_version_string():
Expand Down Expand Up @@ -919,6 +957,58 @@ def import_function(loader, node):
raise ImportError(f"Failed to import function '{function_name}' from module '{module_name}'. " f"Tried relative path '{module_path}' and absolute import.") from ex


_LOCAL_BUILDERS = frozenset({"csv", "json", "parquet", "text", "pandas", "arrow"})


def _looks_like_url(value: str) -> bool:
return value.startswith(("http://", "https://", "s3://", "gs://", "ftp://", "ftps://"))


def _resolve_local_data_files(value, yaml_dir: str):
"""Rewrite relative ``data_files`` paths to absolute paths anchored at ``yaml_dir``.
Conservative: only rewrites a string when it's a relative path AND the
resolved absolute path exists on disk. Leaves absolute paths, URLs, and
non-existent relatives untouched. Recurses into dict/list to handle the
``data_files: {train: <path>, validation: <path>}`` and ``data_files:
[<path>, <path>]`` shapes that ``datasets.load_dataset`` accepts.
"""
if isinstance(value, str):
if not value or _looks_like_url(value) or os.path.isabs(value):
return value
candidate = os.path.join(yaml_dir, value)
if os.path.isfile(candidate):
return candidate
return value
if isinstance(value, dict):
return {k: _resolve_local_data_files(v, yaml_dir) for k, v in value.items()}
if isinstance(value, list):
return [_resolve_local_data_files(v, yaml_dir) for v in value]
return value


def _maybe_resolve_yaml_local_data_files(yaml_config: dict, yaml_dir: str) -> dict:
"""Resolve ``dataset_kwargs.data_files`` relative paths against ``yaml_dir``.
Only applied for the generic local builders (``csv``, ``json``,
``parquet``, ``text``, ``pandas``, ``arrow``). For HF hub dataset
paths (e.g. ``lmms-lab-eval/ssv2``) ``data_files`` is left as-is to
preserve the upstream behavior of letting HF resolve relative paths
against the dataset's hub root.
"""
dataset_path = yaml_config.get("dataset_path")
if not isinstance(dataset_path, str) or dataset_path not in _LOCAL_BUILDERS:
return yaml_config
dataset_kwargs = yaml_config.get("dataset_kwargs")
if not isinstance(dataset_kwargs, dict) or "data_files" not in dataset_kwargs:
return yaml_config
resolved = _resolve_local_data_files(dataset_kwargs["data_files"], yaml_dir)
if resolved is dataset_kwargs["data_files"]:
return yaml_config
new_kwargs = {**dataset_kwargs, "data_files": resolved}
return {**yaml_config, "dataset_kwargs": new_kwargs}


def load_yaml_config(yaml_path=None, yaml_config=None, yaml_dir=None, mode="full"):
if mode == "simple":
constructor_fn = ignore_constructor
Expand Down Expand Up @@ -962,8 +1052,8 @@ def load_yaml_config(yaml_path=None, yaml_config=None, yaml_dir=None, mode="full
raise ex

final_yaml_config.update(yaml_config)
return final_yaml_config
return yaml_config
return _maybe_resolve_yaml_local_data_files(final_yaml_config, yaml_dir)
return _maybe_resolve_yaml_local_data_files(yaml_config, yaml_dir)


def regex_replace(string, pattern, repl, count: int = 0):
Expand Down
Loading