File size: 6,010 Bytes
0d1866c b7cd76e 0d1866c b207cb0 0d1866c b207cb0 0d1866c b207cb0 0d1866c b7cd76e 0d1866c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | #!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
from pathlib import Path
import re
from statistics import median
import time
from config import CARDS_PATH, RESULTS_DIR, load_cards, utc_timestamp
from grader import grade_card
from template import build_messages, build_prompt_text
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run the local-intelligence eval pack on Hugging Face compute with transformers.")
parser.add_argument("--model", required=True, help="Hub model id, e.g. google/gemma-4-E4B-it")
parser.add_argument("--cards", default=str(CARDS_PATH))
parser.add_argument("--card")
parser.add_argument("--tier")
parser.add_argument("--limit", type=int)
parser.add_argument("--max-new-tokens", type=int, default=256)
parser.add_argument("--device-map", default="auto")
parser.add_argument("--dtype", default="auto")
parser.add_argument("--attn-implementation")
parser.add_argument("--trust-remote-code", action="store_true")
parser.add_argument("--output")
parser.add_argument("--output-prefix", default="hf-transformers")
parser.add_argument("--fail-on-incomplete", action="store_true")
parser.add_argument("--verbose", action="store_true")
return parser.parse_args()
def slugify_model(model_id: str) -> str:
return re.sub(r"[^a-z0-9]+", "-", model_id.lower()).strip("-")
def build_generator(args: argparse.Namespace):
import torch
from transformers import pipeline
torch_dtype = None
if args.dtype != "auto":
torch_dtype = getattr(torch, args.dtype)
kwargs = {
"model": args.model,
"device_map": args.device_map,
"trust_remote_code": args.trust_remote_code,
}
if torch_dtype is not None:
kwargs["torch_dtype"] = torch_dtype
if args.attn_implementation:
kwargs["model_kwargs"] = {"attn_implementation": args.attn_implementation}
generator = pipeline("text-generation", **kwargs)
generation_config = getattr(generator.model, "generation_config", None)
if generation_config is not None and getattr(generation_config, "max_length", None) == 20:
generation_config.max_length = None
return generator
def build_generation_input(generator, payload: dict, test_input: dict) -> str:
messages = build_messages(payload, test_input)
tokenizer = getattr(generator, "tokenizer", None)
if tokenizer is not None and hasattr(tokenizer, "apply_chat_template") and getattr(tokenizer, "chat_template", None):
try:
return tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
except Exception:
pass
return build_prompt_text(payload, test_input)
def main() -> int:
args = parse_args()
cards = load_cards(Path(args.cards))
if args.card:
cards = [card for card in cards if card.id == args.card]
if args.tier:
cards = [card for card in cards if card.tier == args.tier]
if args.limit:
cards = cards[: args.limit]
if not cards:
raise SystemExit("No cards selected.")
generator = build_generator(args)
rows = []
for index, card in enumerate(cards, start=1):
payload = card.payload
generation_input = build_generation_input(generator, payload, card.test_input)
start = time.perf_counter()
output = generator(
generation_input,
max_new_tokens=args.max_new_tokens,
do_sample=False,
return_full_text=False,
)
latency_ms = (time.perf_counter() - start) * 1000
raw_text = output[0]["generated_text"].strip()
grade = grade_card(payload, raw_text)
row = {
"card_id": card.id,
"title": card.title,
"tier": card.tier,
"provider": "hf-transformers",
"model": args.model,
"latency_ms": round(latency_ms, 2),
"raw_text": raw_text,
"grade": grade,
}
rows.append(row)
if args.verbose:
status = "PASS" if grade["passed"] else "FAIL"
print(f"[{status}] {index:02d}. {card.id} ({card.tier})")
if not grade["passed"]:
for assertion in grade["assertions"]:
if not assertion["passed"]:
print(f" - {assertion['assertion']}: {assertion['details']}")
passed = sum(1 for row in rows if row["grade"]["passed"])
average_score = sum(row["grade"]["score"] for row in rows) / len(rows)
latencies = [row["latency_ms"] for row in rows]
exact_parse_count = sum(1 for row in rows if row["grade"].get("parse_mode") == "exact")
normalized_parse_count = sum(1 for row in rows if row["grade"].get("parse_mode") == "normalized")
parse_failure_count = sum(1 for row in rows if row["grade"].get("parse_error"))
summary = {
"provider": "hf-transformers",
"model": args.model,
"cards": len(rows),
"passed": passed,
"pass_rate": round(passed / len(rows), 4),
"average_score": round(average_score, 4),
"median_latency_ms": round(median(latencies), 2),
"exact_parse_rate": round(exact_parse_count / len(rows), 4),
"normalized_parse_rate": round(normalized_parse_count / len(rows), 4),
"parse_failure_rate": round(parse_failure_count / len(rows), 4),
}
print(json.dumps(summary, indent=2))
default_name = f"{args.output_prefix}-{slugify_model(args.model)}-{utc_timestamp()}.json"
output_path = Path(args.output) if args.output else RESULTS_DIR / default_name
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(json.dumps({"summary": summary, "rows": rows}, indent=2))
print(f"Wrote results to {output_path}")
if args.fail_on_incomplete and passed != len(rows):
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
|