Spaces:
Build error
Build error
| #!/usr/bin/env python3 | |
| """ | |
| training/scripts/evaluate_model.py | |
| Évalue le modèle fine-tuné sur les 6 tâches VORTEX GOD. | |
| Métriques par tâche : | |
| planner → json_valid_rate, steps_count_correct, risks_present | |
| engineer → syntax_valid_rate, exec_pass_rate, has_docstring | |
| critic → json_valid_rate, verdict_present, issues_severity_present | |
| researcher→ has_sections, has_citations, has_lacunes | |
| scientist → json_valid_rate, hypothesis_length, has_experiment | |
| optimizer → json_valid_rate, has_changes, speedup_numeric | |
| Score global = moyenne pondérée (0.0 – 1.0) | |
| Usage : | |
| python training/scripts/evaluate_model.py \ | |
| --model training/checkpoints/vortex-lora \ | |
| --dataset training/data/vortex_dataset.jsonl \ | |
| --base-model Qwen/Qwen3-8B-Instruct \ | |
| --n 50 \ | |
| --output training/eval_results.json | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import ast | |
| import json | |
| import logging | |
| import os | |
| import random | |
| import re | |
| import time | |
| from pathlib import Path | |
| from typing import Any, Callable, Dict, List, Optional, Tuple | |
| log = logging.getLogger("evaluate_model") | |
| logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") | |
| # ───────────────────────────────────────────── | |
| # Métriques par tâche | |
| # ───────────────────────────────────────────── | |
| def _is_valid_json(text: str) -> bool: | |
| clean = re.sub(r"```(?:json)?\n?", "", text).replace("```", "").strip() | |
| for pattern in (r"\{[\s\S]*\}", r"\[[\s\S]*\]"): | |
| m = re.search(pattern, clean) | |
| if m: | |
| try: | |
| json.loads(m.group(0)) | |
| return True | |
| except Exception: | |
| pass | |
| return False | |
| def _is_valid_python(text: str) -> bool: | |
| clean = re.sub(r"```(?:python|py)?\n?", "", text).replace("```", "").strip() | |
| try: | |
| ast.parse(clean) | |
| return len(clean) > 30 | |
| except SyntaxError: | |
| return False | |
| def _exec_python(text: str, timeout: float = 5.0) -> bool: | |
| """Exécute le code dans un sous-process isolé. Retourne True si exit code 0.""" | |
| import subprocess, tempfile | |
| clean = re.sub(r"```(?:python|py)?\n?", "", text).replace("```", "").strip() | |
| with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: | |
| f.write(clean) | |
| fname = f.name | |
| try: | |
| result = subprocess.run( | |
| ["python3", fname], | |
| timeout=timeout, | |
| capture_output=True, | |
| ) | |
| return result.returncode == 0 | |
| except Exception: | |
| return False | |
| finally: | |
| os.unlink(fname) | |
| TASK_METRICS: Dict[str, List[Tuple[str, Callable[[str], float], float]]] = { | |
| # (metric_name, scorer_fn returns 0.0-1.0, weight) | |
| "planner": [ | |
| ("json_valid", lambda r: 1.0 if _is_valid_json(r) else 0.0, 2.0), | |
| ("has_steps", lambda r: 1.0 if '"steps"' in r and '"id"' in r else 0.0, 1.5), | |
| ("has_goal", lambda r: 1.0 if '"goal"' in r else 0.0, 1.0), | |
| ("has_risks", lambda r: 1.0 if '"risks"' in r else 0.0, 0.5), | |
| ("steps_count", lambda r: min(1.0, _count_steps(r) / 4), 1.0), | |
| ], | |
| "engineer": [ | |
| ("syntax_valid", lambda r: 1.0 if _is_valid_python(r) else 0.0, 3.0), | |
| ("has_imports", lambda r: 1.0 if re.search(r"^import |^from ", r, re.MULTILINE) else 0.0, 1.0), | |
| ("has_docstring", lambda r: 1.0 if '"""' in r or "'''" in r else 0.0, 0.5), | |
| ("has_type_hints", lambda r: 1.0 if " -> " in r or ": " in r else 0.0, 0.5), | |
| ("exec_pass", lambda r: 1.0 if _exec_python(r) else 0.0, 2.0), | |
| ], | |
| "critic": [ | |
| ("json_valid", lambda r: 1.0 if _is_valid_json(r) else 0.0, 2.0), | |
| ("has_verdict", lambda r: 1.0 if '"verdict"' in r else 0.0, 1.5), | |
| ("has_issues", lambda r: 1.0 if '"issues"' in r else 0.0, 1.5), | |
| ("has_severity", lambda r: 1.0 if '"severity"' in r else 0.0, 1.0), | |
| ("has_fix", lambda r: 1.0 if '"fix"' in r else 0.0, 0.5), | |
| ], | |
| "researcher": [ | |
| ("has_sections", lambda r: min(1.0, r.count("##") / 2), 2.0), | |
| ("has_citations", lambda r: 1.0 if "[source:" in r.lower() else 0.0, 2.0), | |
| ("has_lacunes", lambda r: 1.0 if "lacune" in r.lower() or "manque" in r.lower() else 0.0, 1.0), | |
| ("length_ok", lambda r: 1.0 if 200 <= len(r.split()) <= 600 else 0.5, 1.0), | |
| ], | |
| "scientist": [ | |
| ("json_valid", lambda r: 1.0 if _is_valid_json(r) else 0.0, 2.0), | |
| ("has_hypothesis", lambda r: 1.0 if '"hypothesis"' in r else 0.0, 2.0), | |
| ("has_experiment", lambda r: 1.0 if '"experiment"' in r else 0.0, 1.5), | |
| ("has_confidence", lambda r: 1.0 if '"confidence"' in r else 0.0, 0.5), | |
| ("has_null_hyp", lambda r: 1.0 if '"null_hypothesis"' in r else 0.0, 0.5), | |
| ], | |
| "optimizer": [ | |
| ("json_valid", lambda r: 1.0 if _is_valid_json(r) else 0.0, 2.0), | |
| ("has_optimized", lambda r: 1.0 if '"optimized"' in r else 0.0, 2.0), | |
| ("has_changes", lambda r: 1.0 if '"changes"' in r else 0.0, 1.5), | |
| ("has_speedup", lambda r: 1.0 if '"expected_speedup"' in r else 0.0, 0.5), | |
| ("has_rationale", lambda r: 1.0 if '"rationale"' in r else 0.0, 0.5), | |
| ], | |
| } | |
| TASK_WEIGHTS = { | |
| "planner": 1.0, | |
| "engineer": 1.5, # tâche la plus critique | |
| "critic": 1.0, | |
| "researcher": 0.8, | |
| "scientist": 1.0, | |
| "optimizer": 0.8, | |
| } | |
| def _count_steps(text: str) -> int: | |
| clean = re.sub(r"```(?:json)?\n?", "", text).replace("```", "").strip() | |
| m = re.search(r"\{[\s\S]*\}", clean) | |
| if m: | |
| try: | |
| data = json.loads(m.group(0)) | |
| return len(data.get("steps", [])) | |
| except Exception: | |
| pass | |
| return 0 | |
| def score_response(task_type: str, response: str) -> Dict[str, float]: | |
| """Calcule les métriques pour une réponse.""" | |
| metrics_def = TASK_METRICS.get(task_type, []) | |
| scores = {} | |
| total_weight = 0.0 | |
| weighted_sum = 0.0 | |
| for metric_name, scorer, weight in metrics_def: | |
| try: | |
| s = scorer(response) | |
| except Exception: | |
| s = 0.0 | |
| scores[metric_name] = round(s, 3) | |
| weighted_sum += s * weight | |
| total_weight += weight | |
| scores["task_score"] = round(weighted_sum / total_weight, 4) if total_weight > 0 else 0.0 | |
| return scores | |
| # ───────────────────────────────────────────── | |
| # Génération avec le modèle évalué | |
| # ───────────────────────────────────────────── | |
| class ModelEvaluator: | |
| def __init__(self, model_path: str, base_model: str, use_gguf: bool = False): | |
| self.model_path = model_path | |
| self.base_model = base_model | |
| self.use_gguf = use_gguf or model_path.endswith(".gguf") | |
| self._model = None | |
| self._tokenizer = None | |
| self._llama = None | |
| self._load() | |
| def _load(self): | |
| if self.use_gguf: | |
| from llama_cpp import Llama | |
| self._llama = Llama(model_path=self.model_path, n_ctx=2048, n_gpu_layers=35, verbose=False) | |
| log.info(f"[Eval] GGUF chargé : {self.model_path}") | |
| else: | |
| import torch | |
| from peft import PeftModel | |
| from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig | |
| log.info(f"[Eval] Chargement LoRA depuis : {self.model_path}") | |
| bnb = BitsAndBytesConfig(load_in_4bit=True, bnb_4bit_quant_type="nf4", | |
| bnb_4bit_compute_dtype=torch.bfloat16) | |
| self._tokenizer = AutoTokenizer.from_pretrained(self.model_path, trust_remote_code=True) | |
| base = AutoModelForCausalLM.from_pretrained( | |
| self.base_model, | |
| quantization_config = bnb, | |
| device_map = "auto", | |
| trust_remote_code = True, | |
| torch_dtype = torch.bfloat16, | |
| ) | |
| self._model = PeftModel.from_pretrained(base, self.model_path) | |
| self._model.eval() | |
| log.info(f"[Eval] Modèle LoRA chargé ({self._model.num_parameters():,} params)") | |
| def generate(self, system: str, user: str, max_tokens: int = 512, temperature: float = 0.1) -> str: | |
| if self.use_gguf and self._llama: | |
| out = self._llama.create_chat_completion( | |
| messages=[{"role": "system", "content": system}, | |
| {"role": "user", "content": user}], | |
| max_tokens=max_tokens, temperature=max(temperature, 0.01), | |
| ) | |
| return out["choices"][0]["message"]["content"].strip() | |
| import torch | |
| messages = [{"role": "system", "content": system}, {"role": "user", "content": user}] | |
| text = self._tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) | |
| inputs = self._tokenizer(text, return_tensors="pt").to(self._model.device) | |
| with torch.no_grad(): | |
| out = self._model.generate( | |
| **inputs, max_new_tokens=max_tokens, temperature=temperature, | |
| do_sample=True, pad_token_id=self._tokenizer.eos_token_id, | |
| ) | |
| new_ids = out[0][inputs.input_ids.shape[1]:] | |
| return self._tokenizer.decode(new_ids, skip_special_tokens=True).strip() | |
| # ───────────────────────────────────────────── | |
| # Pipeline d'évaluation | |
| # ───────────────────────────────────────────── | |
| def run_evaluation(args: argparse.Namespace) -> Dict[str, Any]: | |
| # Charger les exemples de test depuis le dataset | |
| from build_dataset import SYSTEMS | |
| all_examples: List[Dict] = [] | |
| with open(args.dataset) as f: | |
| for line in f: | |
| line = line.strip() | |
| if line: | |
| try: | |
| all_examples.append(json.loads(line)) | |
| except Exception: | |
| pass | |
| # Sélectionner n exemples équilibrés par tâche | |
| by_task: Dict[str, List] = {} | |
| for ex in all_examples: | |
| t = ex.get("task_type", "unknown") | |
| by_task.setdefault(t, []).append(ex) | |
| test_examples = [] | |
| n_per_task = max(1, args.n // len(SYSTEMS)) | |
| for task, exs in by_task.items(): | |
| test_examples.extend(random.sample(exs, min(n_per_task, len(exs)))) | |
| log.info(f"Évaluation sur {len(test_examples)} exemples") | |
| # Charger le modèle | |
| evaluator = ModelEvaluator(args.model, args.base_model, use_gguf=args.gguf) | |
| # Évaluation | |
| results_by_task: Dict[str, List[Dict]] = {} | |
| total_latency = 0.0 | |
| for i, ex in enumerate(test_examples): | |
| task_type = ex.get("task_type", "unknown") | |
| messages = ex["messages"] | |
| system = messages[0]["content"] | |
| user = messages[1]["content"] | |
| reference = messages[2]["content"] | |
| t0 = time.time() | |
| response = evaluator.generate(system, user, max_tokens=600, temperature=0.1) | |
| latency = time.time() - t0 | |
| total_latency += latency | |
| scores = score_response(task_type, response) | |
| scores["latency_s"] = round(latency, 2) | |
| # Comparaison avec la référence (longueur relative) | |
| ref_len = len(reference.split()) | |
| resp_len = len(response.split()) | |
| scores["length_ratio"] = round(resp_len / max(ref_len, 1), 2) | |
| results_by_task.setdefault(task_type, []).append({ | |
| "user": user[:150], | |
| "response": response[:400], | |
| "scores": scores, | |
| }) | |
| if (i + 1) % 10 == 0: | |
| log.info(f" Progression : {i+1}/{len(test_examples)}") | |
| # Agrégation | |
| summary: Dict[str, Any] = { | |
| "model": args.model, | |
| "base_model": args.base_model, | |
| "n_examples": len(test_examples), | |
| "avg_latency_s": round(total_latency / len(test_examples), 2), | |
| "tasks": {}, | |
| "global_score": 0.0, | |
| } | |
| weighted_scores = [] | |
| for task_type, results in results_by_task.items(): | |
| task_scores = [r["scores"]["task_score"] for r in results] | |
| avg_score = sum(task_scores) / len(task_scores) | |
| weight = TASK_WEIGHTS.get(task_type, 1.0) | |
| # Détail par métrique | |
| metric_avgs = {} | |
| for metric_name, _, _ in TASK_METRICS.get(task_type, []): | |
| vals = [r["scores"].get(metric_name, 0.0) for r in results] | |
| metric_avgs[metric_name] = round(sum(vals) / len(vals), 3) | |
| summary["tasks"][task_type] = { | |
| "n": len(results), | |
| "avg_score": round(avg_score, 4), | |
| "weight": weight, | |
| "metrics": metric_avgs, | |
| } | |
| weighted_scores.append(avg_score * weight) | |
| log.info(f" [{task_type:12s}] score={avg_score:.4f} n={len(results)}") | |
| total_weight = sum(TASK_WEIGHTS.get(t, 1.0) for t in results_by_task) | |
| summary["global_score"] = round(sum(weighted_scores) / total_weight, 4) if total_weight else 0.0 | |
| log.info(f"\n{'='*50}") | |
| log.info(f"Score global : {summary['global_score']:.4f}") | |
| log.info(f"Latence moy. : {summary['avg_latency_s']:.2f}s") | |
| log.info(f"{'='*50}") | |
| # Sauvegarde | |
| out_path = Path(args.output) | |
| out_path.parent.mkdir(parents=True, exist_ok=True) | |
| out_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2)) | |
| log.info(f"✅ Résultats → {args.output}") | |
| return summary | |
| def main(): | |
| parser = argparse.ArgumentParser(description="Évaluation du modèle VORTEX GOD") | |
| parser.add_argument("--model", required=True, help="Chemin checkpoint LoRA ou .gguf") | |
| parser.add_argument("--dataset", default="training/data/full_dataset.jsonl") | |
| parser.add_argument("--base-model", default="Qwen/Qwen3-8B-Instruct") | |
| parser.add_argument("--n", type=int, default=50, help="Nb exemples de test") | |
| parser.add_argument("--output", default="training/eval_results.json") | |
| parser.add_argument("--gguf", action="store_true", help="Utiliser le mode GGUF") | |
| args = parser.parse_args() | |
| run_evaluation(args) | |
| if __name__ == "__main__": | |
| main() | |