(WORK_DIR / “judge.prompty”).write_text(“””— nombre: Judge modelo: api: configuración de chat: tipo: conexión openai: open_ai_connection modelo: gpt-4o-mini parámetros: temperatura: 0 max_tokens: 150 formato_respuesta: {tipo: json_object} entradas: pregunta: {tipo: cadena} respuesta: {tipo: cadena} esperado: {tipo: cadena} — sistema: Eres un calificador exigente Decida si la respuesta del asistente contiene el hecho esperado (no distingue entre mayúsculas y minúsculas, permitiendo frases/sinónimos razonables). Responder SÓLO como JSON: {“score”: 0 o 1, “reason”: “…”} usuario: Pregunta: {{question}} Esperado: {{expected}} Respuesta: {{answer}} “””) (WORK_DIR /). “eval_flow.py”).write_text(textwrap.dedent(”’ importar json desde pathlib importar ruta desde Promptflow.tracing importar traza desde Promptflow.core importar Prompty BASE = Ruta(__file__).parent class Evaluador: def __init__(self): self.judge = Prompty.load(source=BASE / “judge.prompty”) @trace def __call__(self, pregunta: str, respuesta: str, esperado: str) -> dict: raw = self.judge(pregunta=pregunta, respuesta=respuesta, esperado=esperado) si esinstancia(raw, str): intente: raw = json.loads(raw) excepto Excepción: raw = {“score”: 0, “reason”: f”unparseable:{raw[:80]}”} return {“score”: int(raw.get(“score”, 0)), “reason”: str(raw.get(“reason”,””))} def __aggregate__(self, line_results): “””Agregación a nivel de ejecución. Lo que sea que esto devuelva aparece en pf.get_metrics().””” puntuaciones = [r[“score”] para r en line_results si r]return { “exactitud”: (suma(puntuaciones) / len(puntuaciones)) si las puntuaciones son 0.0, “aprobado”: suma(puntuaciones), “total”: len(puntuaciones), } ”’)) (WORK_DIR / “eval.flex.yaml”).write_text( “$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Flow.schema.json\n” “entry: eval_flow:Evaluator\n” ) print(“\n=== Ejecución de evaluación ===”) eval_run = pf.run( flow=str(WORK_DIR / “eval.flex.yaml”), data=str(data_path), run=base_run, column_mapping={ “question”: “${data.question}”, “expected”: “${data.expected}”, “answer”: “${run.outputs.answer}”, }, stream=True, ) eval_details = pf.get_details(eval_run) print(eval_details) print(“\n=== Métricas agregadas (de __aggregate__) ===”) print(json.dumps(pf.get_metrics(eval_run), indent=2)) importar pandas como pd si “outputs.score” en eval_details.columns: s = pd.to_numeric(eval_details[“outputs.score”]errores=”coerce”).fillna(0) print(f”Precisión manual: {s.mean():.2%} ({int(s.sum())}/{len(s)})”)