import argparse
import json
import logging
import os
import re
import sys
from pathlib import Path
import numpy as np
import lm_eval
from lm_eval import evaluator, tasks
from lm_eval.utils import make_table
def _handle_non_serializable(o):
if isinstance(o, np.int64) or isinstance(o, np.int32):
return int(o)
elif isinstance(o, set):
return list(o)
else:
return str(o)
def setup_logging(verbosity):
logging.basicConfig(
level=verbosity.upper(), format="%(asctime)s - %(levelname)s - %(message)s"
)
return logging.getLogger(__name__)
def handle_output(args, results, logger):
if not args.output_path:
if args.log_samples:
logger.error("Specify --output_path for logging samples.")
sys.exit(1)
logger.info(json.dumps(results, indent=2, default=_handle_non_serializable))
return
path = Path(args.output_path)
if path.is_file() or path.with_name("results.json").is_file():
logger.warning(f"File already exists at {path}. Results will be overwritten.")
output_dir = path.parent if path.suffix in (".json", ".jsonl") else path
output_dir.mkdir(parents=True, exist_ok=True)
results_str = json.dumps(results, indent=2, default=_handle_non_serializable)
if args.show_config:
logger.info(results_str)
file_path = os.path.join(args.output_path, "results.json")
with open(file_path , "w", encoding="utf-8") as f:
f.write(results_str)
if args.log_samples:
samples = results.pop("samples", {})
for task_name, _ in results.get("configs", {}).items():
output_name = re.sub(r"/|=", "__", args.model_args) + "_" + task_name
sample_file = output_dir.joinpath(f"{output_name}.jsonl")
sample_data = json.dumps(
samples.get(task_name, {}), indent=2, default=_handle_non_serializable
)
sample_file.write_text(sample_data, encoding="utf-8")
batch_sizes = ",".join(map(str, results.get("config", {}).get("batch_sizes", [])))
summary = f"{args.model} ({args.model_args}), gen_kwargs: ({args.gen_kwargs}), limit: {args.limit}, num_fewshot: {args.num_fewshot}, batch_size: {args.batch_size}{f' ({batch_sizes})' if batch_sizes else ''}"
logger.info(summary)
logger.info(make_table(results))
if "groups" in results:
logger.info(make_table(results, "groups"))
def load_tasks(args):
tasks.initialize_tasks()
if args.open_llm_leaderboard_tasks:
current_dir = os.getcwd()
config_dir = os.path.join(current_dir, "open_llm_leaderboard")
lm_eval.tasks.include_path(config_dir)
return [
"arc_challenge_25_shot",
"hellaswag_10_shot",
"truthfulqa_mc2",
"winogrande_5_shot",
"gsm8k",
"mmlu",
]
return args.tasks.split(",") if args.tasks else []
def parse_eval_args():
parser = argparse.ArgumentParser(formatter_class=argparse.RawTextHelpFormatter)
parser.add_argument(
"--model", "-m", default="hf", help="Name of model, e.g., `hf`."
)
parser.add_argument(
"--tasks",
"-t",
default=None,
help="Comma-separated list of tasks, or 'list' to display available tasks.",
)
parser.add_argument(
"--model_args",
"-a",
default="",
help="Comma-separated string arguments for model, e.g., `pretrained=EleutherAI/pythia-160m`.",
)
parser.add_argument(
"--open_llm_leaderboard_tasks",
"-oplm",
action="store_true",
default=False,
help="Choose the list of tasks with specification in HF open LLM-leaderboard.",
)
parser.add_argument(
"--num_fewshot",
"-f",
type=int,
default=None,
help="Number of examples in few-shot context.",
)
parser.add_argument(
"--batch_size",
"-b",
default=1,
help="Batch size, can be 'auto', 'auto:N', or an integer.",
)
parser.add_argument(
"--max_batch_size",
type=int,
default=None,
help="Maximal batch size with 'auto' batch size.",
)
parser.add_argument(
"--device", default=None, help="Device for evaluation, e.g., 'cuda', 'cpu'."
)
parser.add_argument(
"--output_path", "-o", type=str, default=None, help="Path for saving results."
)
parser.add_argument(
"--limit",
"-L",
type=float,
default=None,
help="Limit number of examples per task.",
)
parser.add_argument(
"--use_cache", "-c", default=None, help="Path to cache db file, if used."
)
parser.add_argument(
"--verbosity",
"-v",
default="INFO",
help="Logging level: CRITICAL, ERROR, WARNING, INFO, DEBUG.",
)
parser.add_argument(
"--gen_kwargs",
default=None,
help="Generation kwargs for tasks that support it.",
)
parser.add_argument(
"--check_integrity",
action="store_true",
help="Whether to run the relevant part of the test suite for the tasks.",
)
parser.add_argument(
"--write_out",
"-w",
action="store_true",
default=False,
help="Prints the prompt for the first few documents.",
)
parser.add_argument(
"--log_samples",
"-s",
action="store_true",
default=False,
help="If True, write out all model outputs and documents for per-sample measurement and post-hoc analysis.",
)
parser.add_argument(
"--show_config",
action="store_true",
default=False,
help="If True, shows the full config of all tasks at the end of the evaluation.",
)
parser.add_argument(
"--include_path",
type=str,
default=None,
help="Additional path to include if there are external tasks.",
)
parser.add_argument(
"--decontamination_ngrams_path", default=None
) # Not currently used
return parser.parse_args()
def evaluate_model(args):
try:
task_list = load_tasks(args)
# Customized model such as Quantized model etc.
# In case you are working with a custom model, you can use the following guide to add it here:
# https://github.com/EleutherAI/lm-evaluation-harness/blob/main/docs/interface.md#external-library-usage
# Evaluate
results = evaluator.simple_evaluate(
model=args.model,
model_args=args.model_args,
tasks=task_list,
num_fewshot=args.num_fewshot,
batch_size=args.batch_size,
max_batch_size=args.max_batch_size,
device=args.device,
use_cache=args.use_cache,
limit=args.limit,
decontamination_ngrams_path=args.decontamination_ngrams_path,
check_integrity=args.check_integrity,
write_out=args.write_out,
log_samples=args.log_samples,
gen_kwargs=args.gen_kwargs,
)
handle_output(args, results, logger)
except Exception as e:
logger.error(f"An error occurred during evaluation: {e}")
sys.exit(1)
if __name__ == "__main__":
args = parse_eval_args()
logger = setup_logging(args.verbosity)
evaluate_model(args)