| """ |
| Avalia e compara os modelos treinados. |
| |
| MΓ©tricas: |
| 1. Perplexidade no val set |
| 2. Geração de texto a partir de prompts DevOps reais |
| 3. Top-k tokens mais provΓ‘veis para completar frases tΓpicas |
| |
| Uso: |
| python eval.py |
| python eval.py --prompt "apiVersion: apps/v1\nkind: Deployment" |
| """ |
|
|
| import json |
| import math |
| import argparse |
| import numpy as np |
| import torch |
| from pathlib import Path |
| from tokenizers import ByteLevelBPETokenizer |
|
|
| from model import GPT, GPTConfig |
| from train import BinaryDataset, DEVICE, DTYPE, EVAL_ITERS, SEQ_LEN, evaluate |
|
|
| DATA_DIR = Path(__file__).parent / "data" |
| CKPT_DIR = Path(__file__).parent / "checkpoints" |
|
|
| DEVOPS_PROMPTS = [ |
| "apiVersion: apps/v1\nkind: Deployment\nmetadata:\n name:", |
| "resource \"aws_eks_cluster\" \"main\" {\n name =", |
| "helm install my-release stable/nginx-ingress --namespace", |
| "kubectl apply -f deployment.yaml --namespace", |
| "kind: HelmRelease\napiVersion: helm.toolkit.fluxcd.io/v2beta1\nmetadata:\n name:", |
| "spec:\n replicas: 3\n selector:\n matchLabels:\n app:", |
| "terraform {\n required_providers {\n kubernetes = {", |
| "argocd app create --app-name", |
| ] |
|
|
|
|
| def load_model(variant: str) -> tuple[GPT, dict]: |
| ckpt_path = CKPT_DIR / f"{variant}_latest.pt" |
| if not ckpt_path.exists(): |
| raise FileNotFoundError(f"Checkpoint nΓ£o encontrado: {ckpt_path}") |
| ck = torch.load(ckpt_path, map_location=DEVICE, weights_only=False) |
| cfg = GPTConfig(**ck["config"]) |
| model = GPT(cfg).to(DEVICE) |
| |
| state = ck["model"] |
| state = {k.replace("_orig_mod.", ""): v for k, v in state.items()} |
| model.load_state_dict(state) |
| model.eval() |
| return model, ck |
|
|
|
|
| def load_tokenizer() -> ByteLevelBPETokenizer: |
| tok_dir = DATA_DIR / "tokenizer" |
| return ByteLevelBPETokenizer( |
| str(tok_dir / "vocab.json"), |
| str(tok_dir / "merges.txt"), |
| ) |
|
|
|
|
| def run_eval(args): |
| meta = json.loads((DATA_DIR / "meta.json").read_text()) |
| val_ds = BinaryDataset(DATA_DIR / "val.bin", SEQ_LEN) |
| tok = load_tokenizer() |
|
|
| models = {} |
| for v in ("baseline", "attnres"): |
| try: |
| m, ck = load_model(v) |
| models[v] = (m, ck) |
| print(f"Carregado: {v} (step {ck.get('step', '?')})") |
| except FileNotFoundError as e: |
| print(f"SKIP: {e}") |
|
|
| if not models: |
| print("Nenhum checkpoint encontrado. Rode train.py primeiro.") |
| return |
|
|
| |
| print("\n" + "=" * 55) |
| print("PERPLEXIDADE NO VAL SET") |
| print("=" * 55) |
| for v, (m, _) in models.items(): |
| loss = evaluate(m, val_ds) |
| ppl = math.exp(loss) |
| params = m.n_params() |
| print(f" {v:10s} loss {loss:.4f} ppl {ppl:8.2f} params {params:,}") |
|
|
| |
| prompts = [args.prompt] if args.prompt else DEVOPS_PROMPTS |
|
|
| print("\n" + "=" * 55) |
| print("GERAΓΓO DE TEXTO") |
| print("=" * 55) |
|
|
| for prompt in prompts[:4]: |
| print(f"\nPrompt: {prompt!r}") |
| enc_ids = tok.encode(prompt).ids |
| x = torch.tensor([enc_ids], dtype=torch.long, device=DEVICE) |
|
|
| for v, (m, _) in models.items(): |
| out_ids = m.generate(x, max_new=60, temperature=0.7, top_k=30) |
| text = tok.decode(out_ids[0].tolist()) |
| cont = text[len(prompt):] |
| print(f" [{v:10s}] {cont[:120]!r}") |
|
|
| |
| if len(models) == 2: |
| print("\n" + "=" * 55) |
| print("CURVAS DE CONVERGΓNCIA (val_ppl por step)") |
| print("=" * 55) |
| print(f"{'Step':>6} {'Baseline':>10} {'AttnRes':>10} {'Ξ (A-B)':>10}") |
| print("-" * 45) |
| ck_b = models["baseline"][1] |
| ck_a = models["attnres"][1] |
| h_b = ck_b.get("history", {}) |
| h_a = ck_a.get("history", {}) |
| for s, lb, la in zip(h_b.get("val_steps", []), |
| h_b.get("val_loss", []), |
| h_a.get("val_loss", [])): |
| pb, pa = math.exp(lb), math.exp(la) |
| print(f"{s:6d} {pb:10.2f} {pa:10.2f} {pa-pb:+10.2f}") |
|
|
|
|
| if __name__ == "__main__": |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--prompt", type=str, default=None) |
| args = parser.parse_args() |
| run_eval(args) |
|
|
