"""OptiQ CLI: convert, benchmark, and evaluate models.""" import os import sys import re import click from .utils.benchmark import ( BenchmarkResult, measure_llm_perplexity, measure_llm_throughput, get_model_size, measure_model_bpw, print_comparison_table, ) @click.group() @click.version_option(version="0.0.9") def cli(): """OptiQ: Optimizing compiler for PyTorch → MLX with data-driven mixed-precision quantization.""" pass @cli.group() def lora(): """Sensitivity-aware LoRA fine-tuning.""" pass @lora.command("train") @click.argument("model_dir") @click.option("--data", "data_dir", required=True) def lora_train(model_dir, data_dir): pass @cli.command() @click.argument("model") def convert(model): pass @cli.command() @click.argument("model_path") @click.option("--baseline", default=None) @click.option("--n-samples", default=50, type=int) def benchmark(model_path, baseline, n_samples): results = [] for path in [model_path, baseline] if baseline else [model_path]: click.echo(f"Benchmarking: {path}") size = get_model_size(path); bpw = measure_model_bpw(path) ppl = measure_llm_perplexity(path, n_samples=n_samples); tps = measure_llm_throughput(path) results.append(BenchmarkResult(name=os.path.basename(path), model_size_mb=size, bpw=bpw, perplexity=ppl, tokens_per_sec=tps)) print_comparison_table(results) @cli.command(name="eval") @click.argument("model_path") @click.option("--task", "-t", default="gsm8k", help="Tasks (comma-sep or 'all')") @click.option("--baseline", default=None, help="Baseline model path") @click.option("--n-samples", "-n", default=200, type=int) @click.option("--tst", "--task-samples-tuple", "tst", help="task1{n},task2{m}") @click.option("--score", is_flag=True, help="Display Capability Score") def eval_cmd(model_path, task, baseline, n_samples, tst, score): """Evaluate model capabilities.""" TASKS = ["gsm8k", "ai2d", "wer", "hellaswag", "arc_challenge", "arc_easy", "winogrande", "piqa", "boolq", "sciq", "openbookqa", "mmlu", "truthfulqa"] CAPABILITY_TASKS = ["gsm8k", "mmlu", "hellaswag", "arc_challenge", "truthfulqa"] task_configs = [] if tst: for p in tst.split(","): m = re.match(r"([a-z0-9_]+)\{(\d+)\}", p.strip()) if m: t_name, t_n = m.groups() if t_name in TASKS: task_configs.append((t_name, int(t_n))) elif task == "all": per_task = n_samples // len(CAPABILITY_TASKS) for t in CAPABILITY_TASKS: task_configs.append((t, per_task)) else: requested = [t.strip() for t in task.split(",") if t.strip() in TASKS] if not requested: return per_task = n_samples // len(requested) for t in requested: task_configs.append((t, per_task)) eval_results = [] def print_final_report(data): if not data: return click.echo(f"\n\n {'=' * 65}\n {'OPTiQ COMPREHENSIVE REPORT':^61}\n {'=' * 65}") click.echo(f" {'Task':<25} | {'Accuracy':<12} | {'Raw Score'}") click.echo(f" {'-' * 25}-+-{'-' * 12}-+-{'-' * 23}") for r in data: click.echo(f" {r['task']:<25} | {r['acc']:>10.2%} | {r['raw']}") click.echo(f" {'-' * 65}") if score or task == "all": core = [r for r in data if r["core_name"] in CAPABILITY_TASKS] if core: s = sum(r["acc"] for r in core) / len(core) click.echo(f" {'CAPABILITY SCORE':<25} | {s:>10.2%} | (Mean of core tasks)") click.echo(f" {'=' * 65}") models_to_run = [(model_path, os.path.basename(model_path))] if baseline: models_to_run.append((baseline, os.path.basename(baseline))) try: for m_path, m_name in models_to_run: click.echo(f"\nEvaluating Model: {m_name}") for t_name, t_n in task_configs: click.echo(f" >>> Running {t_name} (n={t_n})...") res = None if t_name == "gsm8k": from .eval.ogsm8k import evaluate_gsm8k res = evaluate_gsm8k(m_path, n_samples=t_n) elif t_name == "ai2d": from .eval.oai2d import evaluate_ai2d_mlx res = evaluate_ai2d_mlx(m_path, n_samples=t_n) elif t_name == "wer": from .eval.ower import evaluate_wer_mlx res = evaluate_wer_mlx(m_path, n_samples=t_n) else: eval_fn = None if t_name == "hellaswag": from .eval.ohellaswag import evaluate_hellaswag as eval_fn elif t_name == "arc_challenge": eval_fn = lambda p, **kw: __import__('optiq.eval.oarc', fromlist=['evaluate_arc']).evaluate_arc(p, subset="ARC-Challenge", n_samples=kw.get("n_samples", 200)) elif t_name == "arc_easy": eval_fn = lambda p, **kw: __import__('optiq.eval.oarc', fromlist=['evaluate_arc']).evaluate_arc(p, subset="ARC-Easy", n_samples=kw.get("n_samples", 200)) elif t_name == "winogrande": from .eval.owinogrande import evaluate_winogrande as eval_fn elif t_name == "piqa": from .eval.opiqa import evaluate_piqa as eval_fn elif t_name == "boolq": from .eval.oboolq import evaluate_boolq as eval_fn elif t_name == "sciq": from .eval.osciq import evaluate_sciq as eval_fn elif t_name == "openbookqa": from .eval.oopenbookqa import evaluate_openbookqa as eval_fn elif t_name == "mmlu": from .eval.ommlu import evaluate_mmlu as eval_fn elif t_name == "truthfulqa": from .eval.otruthfulqa import evaluate_truthfulqa as eval_fn if eval_fn: res = eval_fn(m_path, n_samples=t_n) if res: if hasattr(res, 'wer'): acc = 1.0 - res.wer; raw = f"{res.n_correct_words}/{res.n_total_words} words" else: acc = res.accuracy; raw = f"{res.n_correct}/{res.n_total}" eval_results.append({"task": f"{t_name} ({m_name})", "acc": acc, "raw": raw, "core_name": t_name}) click.echo(f" [DONE] {t_name}: {acc:.2%}") except Exception as e: click.echo(f"\n!!! ERROR during {t_name}: {e}", err=True) finally: print_final_report(eval_results) @cli.command() @click.argument("model_path") def latency(model_path): pass @cli.command(name="kv-cache") @click.argument("model_path") def kv_cache_cmd(model_path): pass @cli.command(name="serve") def serve_cmd(): pass if __name__ == "__main__": cli()