Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| # -*- coding: utf-8 -*- | |
| """ | |
| Hugging Face Space: LLM Benchmarking App using Gradio | |
| - Upload config.yaml and dataset.jsonl | |
| - Select task | |
| - Run benchmarking across multiple models | |
| - Compute metrics: Exact Match, F1, ROUGE-L, BLEU | |
| - Display results and allow CSV download | |
| """ | |
| import os | |
| import time | |
| import json | |
| import yaml | |
| import gradio as gr | |
| import pandas as pd | |
| from tqdm import tqdm | |
| # Optional metrics | |
| try: | |
| from rouge_score import rouge_scorer | |
| except ImportError: | |
| rouge_scorer = None | |
| try: | |
| import sacrebleu | |
| except ImportError: | |
| sacrebleu = None | |
| # ---------------- Metrics ---------------- # | |
| def exact_match(pred, ref): | |
| return float(pred.strip().lower() == ref.strip().lower()) | |
| def token_f1(pred, ref): | |
| pred_tokens = pred.lower().split() | |
| ref_tokens = ref.lower().split() | |
| if not pred_tokens and not ref_tokens: | |
| return 1.0 | |
| if not pred_tokens or not ref_tokens: | |
| return 0.0 | |
| common = sum(min(pred_tokens.count(t), ref_tokens.count(t)) for t in set(pred_tokens)) | |
| precision = common / len(pred_tokens) | |
| recall = common / len(ref_tokens) | |
| return 2 * precision * recall / (precision + recall) if precision + recall else 0.0 | |
| def rouge_l(pred, ref): | |
| if rouge_scorer: | |
| scorer = rouge_scorer.RougeScorer(["rougeL"], use_stemmer=True) | |
| return scorer.score(ref, pred)["rougeL"].fmeasure | |
| return 0.0 | |
| def bleu(pred, ref): | |
| if sacrebleu: | |
| return sacrebleu.corpus_bleu([pred], [[ref]]).score | |
| return 0.0 | |
| def compute_metrics(task, prediction, reference): | |
| metrics = {} | |
| if task in ("qa", "classification"): | |
| metrics["exact_match"] = exact_match(prediction, reference) | |
| metrics["f1"] = token_f1(prediction, reference) | |
| elif task in ("summarization", "translation"): | |
| metrics["rougeL_f"] = rouge_l(prediction, reference) | |
| metrics["bleu"] = bleu(prediction, reference) | |
| else: | |
| metrics["f1"] = token_f1(prediction, reference) | |
| return metrics | |
| # ---------------- Hugging Face Inference ---------------- # | |
| def hf_generate(model_name, prompt, max_new_tokens=256, temperature=0.2): | |
| from huggingface_hub import InferenceClient | |
| client = InferenceClient(model=model_name, token=os.getenv("HUGGINGFACE_HUB_TOKEN")) | |
| start = time.time() | |
| try: | |
| output = client.text_generation(prompt, max_new_tokens=max_new_tokens, temperature=temperature) | |
| latency = time.time() - start | |
| return output.strip(), latency | |
| except Exception as e: | |
| return f"ERROR: {e}", time.time() - start | |
| # ---------------- Benchmark Function ---------------- # | |
| def benchmark(config_text, dataset_text, task): | |
| cfg = yaml.safe_load(config_text) | |
| data = [json.loads(line) for line in dataset_text.splitlines() if line.strip()] | |
| models = cfg.get("models", []) | |
| templates = cfg.get("prompt_templates", {}) | |
| template = templates.get(task, "{{text}}") | |
| results = [] | |
| for m in models: | |
| model_name = m["name"] | |
| max_tokens = m.get("params", {}).get("max_tokens", 256) | |
| temperature = m.get("params", {}).get("temperature", 0.2) | |
| for ex in tqdm(data, desc=model_name): | |
| variables = {k: ex.get(k, "") for k in ("question", "context", "text", "labels")} | |
| prompt = template | |
| for k, v in variables.items(): | |
| prompt = prompt.replace(f"{{{{{k}}}}}", str(v)) | |
| prediction, latency = hf_generate(model_name, prompt, max_new_tokens=max_tokens, temperature=temperature) | |
| metrics = compute_metrics(task, prediction, ex.get("reference", "")) | |
| row = { | |
| "model": model_name, | |
| "id": ex.get("id", ""), | |
| "task": task, | |
| "prompt": prompt, | |
| "prediction": prediction, | |
| "reference": ex.get("reference", ""), | |
| "latency_seconds": latency, | |
| **metrics | |
| } | |
| results.append(row) | |
| df = pd.DataFrame(results) | |
| summary = [] | |
| for model_name in set(df["model"]): | |
| sub = df[df["model"] == model_name] | |
| summary.append(f"## {model_name}") | |
| summary.append(f"Samples: {len(sub)}") | |
| for metric in ["exact_match", "f1", "rougeL_f", "bleu"]: | |
| if metric in sub.columns: | |
| vals = [v for v in sub[metric] if isinstance(v, (int, float))] | |
| if vals: | |
| summary.append(f"{metric}: mean={sum(vals)/len(vals):.4f}") | |
| summary.append(f"Latency mean: {sum(sub['latency_seconds'])/len(sub):.3f}s\n") | |
| return df, "\n".join(summary) | |
| def hf_judge(model_name, prompt, candidate, reference=None, rubric=None, max_new_tokens=256): | |
| """ | |
| Calls a judge model to score candidate output. Returns a dict of scores. | |
| Rubric should instruct JSON output, e.g.: | |
| {"relevance": int, "factuality": int, "clarity": int, "overall": float} | |
| """ | |
| from huggingface_hub import InferenceClient | |
| client = InferenceClient(model=model_name, token=os.getenv("HF_TOKEN")) | |
| rubric = rubric or ( | |
| "Evaluate the candidate answer. Score 1–5 for:\n" | |
| "- Relevance: addresses the prompt\n" | |
| "- Factuality: correct and supported\n" | |
| "- Clarity: clear and well-structured\n" | |
| "Return JSON: {\"relevance\": int, \"factuality\": int, \"clarity\": int, \"overall\": float}" | |
| ) | |
| judge_prompt = ( | |
| f"{rubric}\n\nPrompt:\n{prompt}\n\n" | |
| f"Candidate:\n{candidate}\n\n" | |
| f"Reference (if available):\n{reference if reference is not None else 'N/A'}\n" | |
| ) | |
| try: | |
| text = client.text_generation(judge_prompt, max_new_tokens=max_new_tokens, temperature=0.0) | |
| # Attempt to parse JSON anywhere in the response: | |
| import re, json | |
| m = re.search(r'\{.*\}', text, re.S) | |
| return json.loads(m.group(0)) if m else {"raw": text} | |
| except Exception as e: | |
| return {"error": str(e)} | |
| # ---------------- Gradio UI ---------------- # | |
| # ---------------- Gradio UI ---------------- # | |
| with gr.Blocks() as demo: | |
| gr.Markdown("# LLM Benchmarking App (Hugging Face)") | |
| gr.Markdown("Upload config.yaml and dataset.jsonl, select task, and run benchmark.") | |
| with gr.Row(): | |
| config_file = gr.File(label="Upload Config (YAML)", type="filepath") | |
| dataset_file = gr.File(label="Upload Dataset (JSONL)", type="filepath") | |
| task = gr.Textbox(label="Task (e.g., qa, summarization, classification)") | |
| use_judge = gr.Checkbox(label="Use Judge Model?", value=False) | |
| judge_model = gr.Textbox(label="Judge Model (HF repo id)", value="mistralai/Mistral-7B-Instruct") | |
| rubric = gr.Textbox(label="Judge Rubric", value="Evaluate relevance, factuality, clarity. Return JSON.") | |
| run_btn = gr.Button("Run Benchmark") | |
| results_table = gr.Dataframe(headers=[ | |
| "model","id","task","prediction","reference","latency_seconds", | |
| "exact_match","f1","rougeL_f","bleu","judge_relevance","judge_factuality","judge_clarity","judge_overall" | |
| ], label="Results") | |
| summary_box = gr.Textbox(label="Summary", lines=10) | |
| download_csv = gr.File(label="Download CSV") | |
| def run_benchmark(config_path, dataset_path, task, use_judge=False, judge_model=None, rubric=None): | |
| if not config_path or not dataset_path: | |
| return None, "Error: Please upload both files", None | |
| config_text = open(config_path, "r", encoding="utf-8").read() | |
| dataset_text = open(dataset_path, "r", encoding="utf-8").read() | |
| cfg = yaml.safe_load(config_text) | |
| data = [json.loads(line) for line in dataset_text.splitlines() if line.strip()] | |
| template = cfg.get("prompt_templates", {}).get(task, "{{text}}") | |
| results = [] | |
| for m in cfg.get("models", []): | |
| model_name = m["name"] | |
| max_tokens = m.get("params", {}).get("max_tokens", 256) | |
| temperature = m.get("params", {}).get("temperature", 0.2) | |
| for ex in data: | |
| variables = {k: ex.get(k, "") for k in ("question", "context", "text", "labels")} | |
| prompt = template | |
| for k, v in variables.items(): | |
| prompt = prompt.replace(f"{{{{{k}}}}}", str(v)) | |
| prediction, latency = hf_generate(model_name, prompt, max_new_tokens=max_tokens, temperature=temperature) | |
| metrics = compute_metrics(task, prediction, ex.get("reference", "")) | |
| row = { | |
| "model": model_name, | |
| "id": ex.get("id", ""), | |
| "task": task, | |
| "prompt": prompt, | |
| "prediction": prediction, | |
| "reference": ex.get("reference", ""), | |
| "latency_seconds": latency, | |
| **metrics | |
| } | |
| if use_judge and judge_model: | |
| scores = hf_judge(judge_model, prompt, prediction, ex.get("reference", ""), rubric) | |
| for k, v in (scores.items() if isinstance(scores, dict) else []): | |
| row[f"judge_{k}"] = v | |
| results.append(row) | |
| df = pd.DataFrame(results) | |
| csv_path = "results.csv" | |
| df.to_csv(csv_path, index=False) | |
| # Summary | |
| summary = [] | |
| for model_name in set(df["model"]): | |
| sub = df[df["model"] == model_name] | |
| summary.append(f"## {model_name}") | |
| summary.append(f"Samples: {len(sub)}") | |
| for metric in ["exact_match", "f1", "rougeL_f", "bleu", "judge_overall"]: | |
| if metric in sub.columns: | |
| vals = [v for v in sub[metric] if isinstance(v, (int, float))] | |
| if vals: | |
| summary.append(f"{metric}: mean={sum(vals)/len(vals):.4f}") | |
| summary.append(f"Latency mean: {sum(sub['latency_seconds'])/len(sub):.3f}s\n") | |
| return df, "\n".join(summary), csv_path | |
| run_btn.click( | |
| run_benchmark, | |
| inputs=[config_file, dataset_file, task, use_judge, judge_model, rubric], | |
| outputs=[results_table, summary_box, download_csv] | |
| ) | |
| demo.launch() | |