Create runner.py

src/web/runner.py

@@ -0,0 +1,205 @@
+from __future__ import annotations
+
+import csv
+import io
+import json
+from dataclasses import dataclass
+from pathlib import Path
+
+import pandas as pd
+
+from modules.analyzer import Analyzer
+from modules.config_manager import ConfigManager
+from modules.evaluator import Evaluator
+from modules.prompt_generator import PromptGenerator
+from modules.visualizer import Visualizer
+from web.hf_client import generar_respuesta_hf
+from web.schemas import JobRequest, ModoEvaluacion
+
+
+ABREVIACIONES = {
+    "preguntas_respuestas_multiples": "PRM",
+    "preguntas_cerradas_probabilidad": "PCP",
+    "preguntas_prompt_injection": "PPI",
+    "preguntas_agente": "PA",
+    "preguntas_analisis_sentimiento": "PAS",
+    "preguntas_cerradas_esperadas": "PCS",
+}
+
+
+@dataclass
+class RunResult:
+    job_dir: Path
+    graficos_dir: Path
+
+
+def _cargar_prompts_por_defecto(repo_root: Path) -> list[tuple[str, pd.DataFrame]]:
+    carpeta_prompts = repo_root / "evaluacion_por_defecto" / "prompts_por_defecto"
+    archivos = sorted(carpeta_prompts.glob("*PREGUNTAS_CERRADAS_ESPERADAS*.csv"))
+    datasets = []
+    for archivo in archivos:
+        df = pd.read_csv(archivo, delimiter="|")
+        datasets.append((archivo.name, df))
+    return datasets
+
+
+def _normalizar_plantilla_personalizada(plantilla: dict) -> list[tuple[str, pd.DataFrame]]:
+    datasets: list[tuple[str, pd.DataFrame]] = []
+    sesgos = plantilla.get("sesgos_a_analizar", [])
+
+    for sesgo in sesgos:
+        preocupacion = str(sesgo.get("preocupacion_etica", "sesgo")).replace(" ", "_").upper()
+        marcador = str(sesgo.get("marcador", "COMUNIDAD")).strip()
+        comunidades = list(sesgo.get("comunidades_sensibles", []))
+        contextos = list(sesgo.get("contextos", []))
+
+        for contexto_data in contextos:
+            contexto = str(contexto_data.get("contexto", "contexto")).replace(" ", "_").upper()
+            ejemplo = str(contexto_data.get("ejemplo_salida", ""))
+            if not ejemplo.strip():
+                continue
+
+            reader = csv.DictReader(io.StringIO(ejemplo), delimiter="|")
+            rows = list(reader)
+            if not rows:
+                continue
+
+            salida_rows = []
+            marcador_token = "{{" + marcador + "}}"
+            for row in rows:
+                prompt = str(row.get("prompt", "")).strip()
+                escenario = str(row.get("escenario", "")).strip()
+                respuesta_esperada = str(row.get("respuesta_esperada", "")).strip()
+                if not prompt or not respuesta_esperada:
+                    continue
+                for comunidad in comunidades:
+                    salida_rows.append(
+                        {
+                            "prompt": prompt.replace(marcador_token, str(comunidad)),
+                            "escenario": escenario,
+                            "respuesta_esperada": respuesta_esperada,
+                            "comunidad_sensible": str(comunidad),
+                        }
+                    )
+
+            if salida_rows:
+                nombre_csv = f"prompts_generados_PREGUNTAS_CERRADAS_ESPERADAS_sesgo_{preocupacion}_contexto_{contexto}.csv"
+                datasets.append((nombre_csv, pd.DataFrame(salida_rows)))
+
+    return datasets
+
+
+def _guardar_dataset_entrada(carpeta: Path, nombre_archivo: str, df_prompts: pd.DataFrame) -> None:
+    carpeta.mkdir(parents=True, exist_ok=True)
+    df_prompts.to_csv(carpeta / nombre_archivo, sep="|", index=False)
+
+
+def _aplicar_limite_prompts(
+    datasets: list[tuple[str, pd.DataFrame]],
+    max_prompts_por_job: int,
+) -> list[tuple[str, pd.DataFrame]]:
+    prompts_restantes = max_prompts_por_job
+    salida: list[tuple[str, pd.DataFrame]] = []
+
+    for nombre_archivo, df_prompts in datasets:
+        if prompts_restantes <= 0:
+            break
+        if df_prompts.empty:
+            continue
+
+        df_limitado = df_prompts.head(prompts_restantes).copy()
+        if not df_limitado.empty:
+            salida.append((nombre_archivo, df_limitado))
+            prompts_restantes -= len(df_limitado)
+
+    return salida
+
+
+def ejecutar_job(job_request: JobRequest, job_dir: Path) -> RunResult:
+    repo_root = Path(__file__).resolve().parents[2]
+    config = ConfigManager(str(repo_root / "config" / "config_modelos.json"))
+    evaluator = Evaluator(config, model_manager=None)
+    prompt_generator = PromptGenerator(config, model_manager=None)
+    analyzer = Analyzer(config)
+    visualizer = Visualizer(config)
+
+    entrada_dir = job_dir / "prompts_entrada"
+    respuestas_dir = job_dir / "respuestas_modelo_evaluado"
+    graficos_dir = job_dir / "graficos"
+
+    if job_request.modo_evaluacion == ModoEvaluacion.POR_DEFECTO:
+        datasets = _cargar_prompts_por_defecto(repo_root)
+    else:
+        if not job_request.plantilla_personalizada:
+            raise ValueError("En modo personalizada se requiere 'plantilla_personalizada'.")
+        datasets = _normalizar_plantilla_personalizada(job_request.plantilla_personalizada)
+
+    if not datasets:
+        raise ValueError("No se encontraron prompts válidos para ejecutar la evaluación.")
+
+    datasets = _aplicar_limite_prompts(datasets, job_request.max_prompts_por_job)
+    if not datasets:
+        raise ValueError("No hay prompts disponibles tras aplicar el límite del job.")
+
+    respuestas_dir.mkdir(parents=True, exist_ok=True)
+    graficos_dir.mkdir(parents=True, exist_ok=True)
+    entrada_dir.mkdir(parents=True, exist_ok=True)
+
+    df_acumulado = pd.DataFrame()
+
+    for nombre_archivo, df_prompts in datasets:
+        _guardar_dataset_entrada(entrada_dir, nombre_archivo, df_prompts)
+        filas_resultado = []
+
+        for _, fila in df_prompts.iterrows():
+            respuesta_cruda = generar_respuesta_hf(
+                prompt=str(fila["prompt"]),
+                model_id=job_request.modelo_hf,
+                api_key=job_request.api_key_hf,
+                timeout_segundos=job_request.timeout_segundos,
+                numero_reintentos=job_request.numero_reintentos,
+            )
+            respuesta_limpia = prompt_generator.limpiar_respuesta_generada_evaluacion(
+                "preguntas_cerradas_esperadas", respuesta_cruda
+            )
+            fila_dict = {
+                "prompt": str(fila.get("prompt", "")),
+                "escenario": str(fila.get("escenario", "")),
+                "respuesta_esperada": str(fila.get("respuesta_esperada", "")),
+                "comunidad_sensible": str(fila.get("comunidad_sensible", "")),
+                "respuesta_modelo": respuesta_limpia,
+                "tipo_evaluacion": "preguntas_cerradas_esperadas",
+            }
+            fila_dict["resultado"] = evaluator.evaluar_respuestas(fila_dict, nombre_archivo)
+            filas_resultado.append(fila_dict)
+
+        df_resultados = pd.DataFrame(filas_resultado)
+        df_resultados.to_csv(respuestas_dir / nombre_archivo, sep="|", index=False)
+        df_acumulado = pd.concat([df_acumulado, df_resultados], ignore_index=True)
+
+    df_acumulado = analyzer.analisis_avanzado_resultados(
+        df_acumulado,
+        array_comunidades_sentimientos=[],
+        array_comunidades_probabilidad=[],
+        carpeta_graficos=str(graficos_dir),
+        abreviaciones=ABREVIACIONES,
+    )
+
+    visualizer.plot_resultados_generales(df_acumulado, str(graficos_dir))
+    visualizer.plot_resultados_tipo_evaluacion(df_acumulado, str(graficos_dir))
+    visualizer.plot_mapa_calor(df_acumulado, str(graficos_dir))
+    visualizer.plot_interactive(df_acumulado, str(graficos_dir), ABREVIACIONES)
+
+    df_acumulado.to_csv(graficos_dir / "resultados.csv", sep="|", index=False)
+    df_acumulado.to_excel(graficos_dir / "resultados.xlsx", index=False, sheet_name="Resultados")
+
+    resumen = {
+        "total": int(len(df_acumulado)),
+        "aciertos": int((df_acumulado["resultado"] == "acierto").sum()),
+        "fallos": int((df_acumulado["resultado"] == "fallo").sum()),
+        "errores": int((df_acumulado["resultado"] == "error").sum()),
+    }
+    with open(job_dir / "resumen.json", "w", encoding="utf-8") as f:
+        json.dump(resumen, f, ensure_ascii=False, indent=2)
+
+    return RunResult(job_dir=job_dir, graficos_dir=graficos_dir)