""" evaluator.py ──────────── Métricas de calidad para las correcciones del sistema RAG. Las tres comparaciones que realiza el evaluador: 1. GT vs HTR → error de PARTIDA (qué tan malo era el HTR) 2. GT vs Corregido → error FINAL (qué tan bueno es el RAG) 3. HTR vs Corregido → MODERNISMOS (qué cambió el LLM, no debería modernizar) Un cer_improvement positivo significa que el RAG mejoró el texto. Un cer_improvement negativo significa que el LLM empeoró el texto. Métricas: - CER (Character Error Rate) : distancia Levenshtein a nivel carácter - WER (Word Error Rate) : distancia Levenshtein a nivel palabra - Modernism score : penalización por grafías s.XVI modernizadas - Regression score : detecta si el LLM empeoró respecto al HTR Uso: from evaluator import Evaluator ev = Evaluator() # Un solo par metrics = ev.evaluate_pair(htr="...", corrected="...", gt="...") print(ev.format_pair_report(metrics)) # Batch report = ev.batch_evaluate(corrector, pairs[:50]) """ import re from typing import List, Dict from knowledge_base import GRAFIA_PATTERNS class Evaluator: # ── Métricas de edición ────────────────────────────────────────────────── @staticmethod def cer(reference: str, hypothesis: str) -> float: """ Character Error Rate: fracción de caracteres incorrectos. 0.0 = perfecto, 1.0 = todo mal. Compara: reference (GT) vs hypothesis (HTR o Corregido). """ r, h = list(reference), list(hypothesis) return Evaluator._levenshtein(r, h) / max(len(r), 1) @staticmethod def wer(reference: str, hypothesis: str) -> float: """ Word Error Rate: fracción de palabras incorrectas. 0.0 = perfecto, 1.0 = todo mal. Compara: reference (GT) vs hypothesis (HTR o Corregido). """ r = reference.split() h = hypothesis.split() return Evaluator._levenshtein(r, h) / max(len(r), 1) @staticmethod def _levenshtein(seq1: list, seq2: list) -> int: """Distancia de edición mínima entre dos secuencias.""" m, n = len(seq1), len(seq2) dp = list(range(n + 1)) for i in range(1, m + 1): prev = dp[:] dp[0] = i for j in range(1, n + 1): if seq1[i - 1] == seq2[j - 1]: dp[j] = prev[j - 1] else: dp[j] = 1 + min(prev[j], dp[j - 1], prev[j - 1]) return dp[n] # ── Detector de modernismos ────────────────────────────────────────────── @staticmethod def modernism_penalty(htr: str, corrected: str) -> Dict: """ Comparación 3: HTR vs Corregido. Detecta formas modernas que el LLM introdujo y que NO estaban en el HTR original. Estas son correcciones incorrectas porque el sistema NO debe modernizar grafías del s.XVI. Ejemplo de error detectado: HTR: "fizo merçed" Corregido: "hizo merced" ← modernizó f→h y ç→c (INCORRECTO) score: 1.0 = sin modernismos, 0.0 = muchos modernismos """ issues = [] htr_lower = htr.lower() corr_lower = corrected.lower() for p in GRAFIA_PATTERNS: modern = p["modern"].lower() ancient_forms = [f.strip().lower() for f in p["ancient"].split("/")] # El corregido tiene la forma moderna Y el HTR no la tenía if modern in corr_lower and modern not in htr_lower: # Además el HTR tampoco tenía la forma antigua # (si la tenía, es una expansión de abreviatura válida) if not any(af in htr_lower for af in ancient_forms): issues.append({ "modern": p["modern"], "ancient": p["ancient"], "rule": p["rule"], "category": p.get("category", ""), }) score = max(0.0, 1.0 - len(issues) * 0.1) return { "count": len(issues), "issues": issues, "score": round(score, 4), } # ── Detector de regresiones ────────────────────────────────────────────── @staticmethod def regression_check(htr: str, corrected: str, gt: str) -> Dict: """ Detecta si el LLM empeoró el texto respecto al HTR original. Un resultado positivo en cer_improvement no garantiza que todo esté bien — el LLM podría haber corregido unos errores e introducido otros. Esta función identifica palabras que estaban bien en el HTR y el LLM cambió incorrectamente. """ htr_words = htr.split() corr_words = corrected.split() gt_words = gt.split() regressions = [] # Comparar palabra a palabra hasta el mínimo de las tres listas for i, gt_w in enumerate(gt_words): htr_w = htr_words[i] if i < len(htr_words) else "" corr_w = corr_words[i] if i < len(corr_words) else "" # El HTR estaba bien, el corregido está mal if htr_w == gt_w and corr_w != gt_w: regressions.append({ "position": i, "gt": gt_w, "htr": htr_w, # correcto en HTR "corrected":corr_w, # empeorado por el LLM }) return { "count": len(regressions), "regressions": regressions, } # ── Evaluación de un par ───────────────────────────────────────────────── def evaluate_pair(self, htr: str, corrected: str, gt: str) -> Dict: """ Evalúa una corrección con las tres comparaciones: Comparación 1 — GT vs HTR: Mide el error de partida. Cuánto se alejaba el HTR del GT. Comparación 2 — GT vs Corregido: Mide el error final. Cuánto se aleja la corrección del GT. cer_improvement > 0 → el RAG mejoró el texto cer_improvement < 0 → el RAG empeoró el texto Comparación 3 — HTR vs Corregido: Detecta modernismos introducidos por el LLM. El LLM NO debe cambiar grafías válidas del s.XVI. """ # ── Comparación 1: GT vs HTR (error de partida) ──────────────────── cer_htr = self.cer(gt, htr) wer_htr = self.wer(gt, htr) # ── Comparación 2: GT vs Corregido (error final) ─────────────────── cer_corr = self.cer(gt, corrected) wer_corr = self.wer(gt, corrected) cer_improvement = cer_htr - cer_corr # positivo = mejoró wer_improvement = wer_htr - wer_corr # Veredicto de la corrección if cer_improvement > 0.02: verdict = "✓ MEJORADO" elif cer_improvement < -0.02: verdict = "✗ EMPEORADO" else: verdict = "~ SIN CAMBIO SIGNIFICATIVO" # ── Comparación 3: HTR vs Corregido (modernismos) ───────────────── modernism = self.modernism_penalty(htr, corrected) regression = self.regression_check(htr, corrected, gt) return { # ── Error de partida (GT vs HTR) ────────────────────────────── "cer_before": round(cer_htr, 4), "wer_before": round(wer_htr, 4), # ── Error final (GT vs Corregido) ───────────────────────────── "cer_after": round(cer_corr, 4), "wer_after": round(wer_corr, 4), # ── Mejora neta ─────────────────────────────────────────────── "cer_improvement": round(cer_improvement, 4), "wer_improvement": round(wer_improvement, 4), "verdict": verdict, # ── Modernismos (HTR vs Corregido) ──────────────────────────── "modernism_score": modernism["score"], "modernism": modernism, # ── Regresiones ─────────────────────────────────────────────── "regression": regression, } # ── Reporte legible de un par ──────────────────────────────────────────── @staticmethod def format_pair_report(metrics: Dict) -> str: """ Formatea las métricas de un par en texto legible para la UI. """ lines = [] lines.append("─" * 50) lines.append(" EVALUACIÓN DE LA CORRECCIÓN") lines.append("─" * 50) # Comparación 1 — error de partida lines.append("\n① Error de partida (GT vs HTR original)") lines.append(f" CER: {metrics['cer_before']:.2%} WER: {metrics['wer_before']:.2%}") # Comparación 2 — error final lines.append("\n② Error final (GT vs Texto corregido)") lines.append(f" CER: {metrics['cer_after']:.2%} WER: {metrics['wer_after']:.2%}") # Mejora neta cer_imp = metrics["cer_improvement"] wer_imp = metrics["wer_improvement"] sign_c = "+" if cer_imp >= 0 else "" sign_w = "+" if wer_imp >= 0 else "" lines.append(f"\n Mejora CER: {sign_c}{cer_imp:.2%} Mejora WER: {sign_w}{wer_imp:.2%}") lines.append(f" {metrics['verdict']}") # Comparación 3 — modernismos mod = metrics["modernism"] lines.append(f"\n③ Modernismos introducidos (HTR vs Corregido)") if mod["count"] == 0: lines.append(" ✓ Ninguno — el LLM respetó las grafías del s.XVI") else: lines.append(f" ✗ {mod['count']} modernismo(s) detectado(s):") for issue in mod["issues"]: lines.append( f" • '{issue['modern']}' introducido " f"(debería ser '{issue['ancient']}'): {issue['rule']}" ) # Regresiones reg = metrics["regression"] if reg["count"] > 0: lines.append(f"\n⚠ Regresiones: {reg['count']} palabra(s) correctas en HTR empeoradas por el LLM:") for r in reg["regressions"][:5]: lines.append( f" • posición {r['position']}: " f"HTR='{r['htr']}' → Corregido='{r['corrected']}' " f"(GT='{r['gt']}')" ) lines.append("─" * 50) return "\n".join(lines) # ── Evaluación en batch ────────────────────────────────────────────────── def batch_evaluate( self, corrector, pairs: List[Dict], verbose: bool = True ) -> Dict: """ Evalúa el sistema sobre una lista de pares con groundtruth. Retorna métricas agregadas + detalle por par. """ results = [] for i, pair in enumerate(pairs): if verbose: print(f" Evaluando {i+1}/{len(pairs)}: {pair['id']}") try: out = corrector.correct(pair["htr"]) metrics = self.evaluate_pair( htr=pair["htr"], corrected=out["corrected"], gt=pair["gt"], ) metrics["id"] = pair["id"] metrics["htr"] = pair["htr"] metrics["corrected"] = out["corrected"] metrics["gt"] = pair["gt"] results.append(metrics) except Exception as e: print(f" Error en {pair['id']}: {e}") if not results: return {"error": "Sin resultados"} def avg(key): return round(sum(r[key] for r in results) / len(results), 4) n = len(results) mejoras = sum(1 for r in results if r["cer_improvement"] > 0.02) empeorados = sum(1 for r in results if r["cer_improvement"] < -0.02) sin_cambio = n - mejoras - empeorados summary = { "n_evaluated": n, # ── Comparación 1: GT vs HTR ────────────────────────────────── "avg_cer_before": avg("cer_before"), "avg_wer_before": avg("wer_before"), # ── Comparación 2: GT vs Corregido ──────────────────────────── "avg_cer_after": avg("cer_after"), "avg_wer_after": avg("wer_after"), # ── Mejora neta ─────────────────────────────────────────────── "avg_cer_improvement": avg("cer_improvement"), "avg_wer_improvement": avg("wer_improvement"), "n_mejorados": mejoras, "n_empeorados": empeorados, "n_sin_cambio": sin_cambio, # ── Comparación 3: Modernismos ──────────────────────────────── "avg_modernism_score": avg("modernism_score"), "detail": results, } if verbose: print(f"\n{'─'*55}") print(f" RESUMEN EVALUACIÓN ({n} pares)") print(f"{'─'*55}") print(f"\n① Error de partida (GT vs HTR):") print(f" CER: {summary['avg_cer_before']:.2%} WER: {summary['avg_wer_before']:.2%}") print(f"\n② Error final (GT vs Corregido):") print(f" CER: {summary['avg_cer_after']:.2%} WER: {summary['avg_wer_after']:.2%}") print(f"\n Mejora CER: {summary['avg_cer_improvement']:+.2%} " f"Mejora WER: {summary['avg_wer_improvement']:+.2%}") print(f"\n Mejorados : {mejoras}/{n} ({mejoras/n:.0%})") print(f" Empeorados : {empeorados}/{n} ({empeorados/n:.0%})") print(f" Sin cambio : {sin_cambio}/{n} ({sin_cambio/n:.0%})") print(f"\n③ Modernismos (HTR vs Corregido):") print(f" Score promedio: {summary['avg_modernism_score']:.2%} " f"(1.0 = sin modernismos)") print(f"{'─'*55}") return summary