Update app.py

app.py

@@ -4,7 +4,7 @@ from typing import List, Dict, Tuple
 
 import gradio as gr
 
-# Carga perezosa de modelos (se inicializan al primer uso)
+# --- Carga perezosa de modelos (se inicializan al primer uso) ---
 _llm = None
 _llm_tok = None
 _gen = None
@@ -23,38 +23,40 @@ TEXTO:
 """
 
 HYP: Dict[str, List[str]] = {
- "4.4.5.1": [
-   "Tomó la iniciativa sin que se lo pidieran.",
-   "Ofreció ayuda o asesoría no solicitada.",
-   "Pensó con orientación al futuro.",
-   "Equilibró iniciativa y riesgo."
- ],
- "4.4.5.2": [
-   "Mostró compromiso personal con los objetivos.",
-   "Promovió el proyecto y generó entusiasmo.",
-   "Definió o monitoreó indicadores de desempeño.",
-   "Buscó mejoras en procesos."
- ],
- "4.4.5.3": [
-   "Proporcionó dirección clara al equipo.",
-   "Realizó coaching o mentoring para mejorar capacidades.",
-   "Estableció y comunicó visión y valores.",
-   "Alineó objetivos individuales con los comunes."
- ],
- "4.4.5.4": [
-   "Usó apropiadamente poder e influencia.",
-   "Seleccionó el canal de comunicación adecuado para influir.",
-   "Fue percibido como líder por los stakeholders."
- ],
- "4.4.5.5": [
-   "Tomó decisiones bajo incertidumbre considerando pros y contras.",
-   "Explicó el razonamiento de las decisiones.",
-   "Revisó decisiones con nueva evidencia.",
-   "Comunicó claramente la decisión e influyó su adopción."
- ]
+    "4.4.5.1": [
+        "Tomó la iniciativa sin que se lo pidieran.",
+        "Ofreció ayuda o asesoría no solicitada.",
+        "Pensó con orientación al futuro.",
+        "Equilibró iniciativa y riesgo."
+    ],
+    "4.4.5.2": [
+        "Mostró compromiso personal con los objetivos.",
+        "Promovió el proyecto y generó entusiasmo.",
+        "Definió o monitoreó indicadores de desempeño.",
+        "Buscó mejoras en procesos."
+    ],
+    "4.4.5.3": [
+        "Proporcionó dirección clara al equipo.",
+        "Realizó coaching o mentoring para mejorar capacidades.",
+        "Estableció y comunicó visión y valores.",
+        "Alineó objetivos individuales con los comunes."
+    ],
+    "4.4.5.4": [
+        "Usó apropiadamente poder e influencia.",
+        "Seleccionó el canal de comunicación adecuado para influir.",
+        "Fue percibido como líder por los stakeholders."
+    ],
+    "4.4.5.5": [
+        "Tomó decisiones bajo incertidumbre considerando pros y contras.",
+        "Explicó el razonamiento de las decisiones.",
+        "Revisó decisiones con nueva evidencia.",
+        "Comunicó claramente la decisión e influyó su adopción."
+    ]
 }
 
+# --------- Modelos ---------
 def lazy_load_llm():
+    """Crea pipeline de generación sin flags que generen warnings en CPU."""
     global _llm, _llm_tok, _gen
     if _gen is not None:
         return _gen
@@ -62,8 +64,7 @@ def lazy_load_llm():
     _llm_tok = AutoTokenizer.from_pretrained(LLM_ID)
     _llm = AutoModelForCausalLM.from_pretrained(
         LLM_ID,
-        torch_dtype="auto",
-        device_map="auto"
+        device_map="auto"  # dtype auto en CPU
     )
     _gen = pipeline(
         "text-generation",
@@ -71,96 +72,128 @@ def lazy_load_llm():
         tokenizer=_llm_tok,
         max_new_tokens=512,
         do_sample=False,
-        temperature=0.0,
         repetition_penalty=1.1,
     )
     return _gen
 
+
 def lazy_load_nli():
+    """Config moderna: top_k=None devuelve todas las etiquetas; softmax para probas estables."""
     global _nli
     if _nli is not None:
         return _nli
     from transformers import pipeline
-    # top_k=None -> devuelve todas las etiquetas; evita return_all_scores (deprecado)
     _nli = pipeline(
         "text-classification",
         model=NLI_ID,
         tokenizer=NLI_ID,
-        top_k=None,  # <= clave
+        top_k=None,                 # en lugar de return_all_scores=True (deprecado)
         function_to_apply="softmax"
     )
     return _nli
 
-
+# --------- Utilidades ---------
 def extract_json_block(text: str) -> str:
     """Intenta extraer el primer bloque JSON válido de una cadena."""
     start = text.find("{")
     end = text.rfind("}")
     if start != -1 and end != -1 and end > start:
         return text[start:end+1]
-    # fallback minimalista si el modelo devolviera texto cruft
+    # fallback minimalista si el modelo devolviera texto con ruido
     return '{"situation":"","task":"","action":[],"result":[]}'
 
+def quick_parse_star(txt: str):
+    """Atajo: si el texto ya viene en STAR, lo parseamos localmente para ahorrar tiempo."""
+    t = (txt or "").strip()
+    if not t:
+        return None
+    keys = ("SITUATION", "TASK", "ACTION", "RESULT", "S:", "T:", "A:", "R:")
+    if not any(k in t for k in keys):
+        return None
+    sections = {"situation": "", "task": "", "action": [], "result": []}
+    blocks = re.split(r'(?im)^(SITUATION|TASK|ACTION|RESULT|S:|T:|A:|R:)\s*:?', t)
+    # blocks = ["", "SITUATION", "...", "TASK", "...", ...]
+    for i in range(1, len(blocks), 2):
+        key = blocks[i].lower()[0]  # s/t/a/r
+        val = blocks[i+1].strip()
+        if key == "s":
+            sections["situation"] = val
+        elif key == "t":
+            sections["task"] = val
+        elif key == "a":
+            sections["action"] = [x.strip("•- ") for x in val.splitlines() if x.strip()]
+        elif key == "r":
+            sections["result"] = [x.strip("•- ") for x in val.splitlines() if x.strip()]
+    return sections
+
 def extract_star(user_text: str) -> Dict:
+    """Si ya viene en STAR, parsea localmente; si no, usa el LLM para extraer STAR en JSON."""
+    parsed = quick_parse_star(user_text)
+    if parsed:
+        return parsed
+
     gen = lazy_load_llm()
-    prompt = STAR_PROMPT.format(texto=user_text.strip())
+    prompt = STAR_PROMPT.format(texto=(user_text or "").strip())
     out = gen(prompt)[0]["generated_text"]
     raw = extract_json_block(out)
     try:
         data = json.loads(raw)
     except Exception:
-        # Pequeña heurística de rescate si llega formato con bullets no JSON
+        # Heurística de rescate si llega formato con bullets no JSON
         data = {"situation": "", "task": "", "action": [], "result": []}
-        # Situación
-        m = re.search(r'Situation[:：]\s*(.*)', user_text, flags=re.I)
-        if m: data["situation"] = m.group(1).strip()
-    # normaliza tipos
+        m = re.search(r'Situation[:：]\s*(.*)', user_text or "", flags=re.I)
+        if m:
+            data["situation"] = m.group(1).strip()
+    # Normaliza tipos
     data["action"] = data.get("action", [])
     data["result"] = data.get("result", [])
-    if isinstance(data["action"], str): data["action"] = [data["action"]]
-    if isinstance(data["result"], str): data["result"] = [data["result"]]
+    if isinstance(data["action"], str):
+        data["action"] = [data["action"]]
+    if isinstance(data["result"], str):
+        data["result"] = [data["result"]]
     return {
-        "situation": data.get("situation", "").strip(),
-        "task": data.get("task", "").strip(),
-        "action": [a.strip(" •-") for a in data["action"] if str(a).strip()],
-        "result": [r.strip(" •-") for r in data["result"] if str(r).strip()],
+        "situation": (data.get("situation", "") or "").strip(),
+        "task": (data.get("task", "") or "").strip(),
+        "action": [str(a).strip(" •-") for a in data["action"] if str(a).strip()],
+        "result": [str(r).strip(" •-") for r in data["result"] if str(r).strip()],
     }
 
 def nli_entails(premise: str, hypothesis: str) -> float:
+    """Devuelve probabilidad de entailment (0..1), robusto a formatos y errores del pipeline."""
     nli = lazy_load_nli()
-    # Premisa vacía puede ocurrir si no se extrajo STAR; evitamos llamadas inútiles
     text_a = (premise or "").strip()
     text_b = (hypothesis or "").strip()
     if not text_a or not text_b:
         return 0.0
 
-    res = nli({"text": text_a, "text_pair": text_b})
+    try:
+        res = nli({"text": text_a, "text_pair": text_b})
+    except Exception:
+        return 0.0
 
     # Normalización de formatos posibles:
+    # - {label, score}
     # - [{label, score}, ...]
     # - [[{label, score}, ...]]
-    # - {label, score}
     if isinstance(res, dict):
         candidates = [res]
     elif isinstance(res, list):
-        if len(res) > 0 and isinstance(res[0], list):
-            candidates = res[0]
-        else:
-            candidates = res
+        candidates = res[0] if (res and isinstance(res[0], list)) else res
     else:
         return 0.0
 
-    # Busca la probabilidad de ENTAILMENT (mayúsculas/minúsculas/variantes)
-    entail_prob = 0.0
-    for c in candidates:
+    # Filtra sólo dicts por seguridad y busca 'entail'
+    for c in (d for d in candidates if isinstance(d, dict)):
         lab = str(c.get("label", "")).lower()
-        if lab.startswith("entail"):
-            entail_prob = float(c.get("score", 0.0))
-            break
-    return entail_prob
+        if "entail" in lab:  # cubre entail/entails/entailment
+            try:
+                return float(c.get("score", 0.0))
+            except Exception:
+                return 0.0
+    return 0.0
 
 def map_prob_to_score(p: float) -> int:
-    # Umbrales iniciales (ajústalos tras validar con tus casos)
+    """Mapeo inicial de probabilidad promedio a escala 0–4 (ajusta tras validación interna)."""
     if p >= 0.80: return 4
     if p >= 0.60: return 3
     if p >= 0.40: return 2
@@ -175,50 +208,54 @@ def score_indicator(premise: str, hyps: List[str]) -> Tuple[int, List[Tuple[str,
     return score, probs_sorted, avg
 
 def evaluate(texto: str) -> Tuple[str, Dict, Dict]:
-    if not texto or not texto.strip():
-        return "Introduce un caso en formato STAR (o texto libre).", {}, {}
-
-    star = extract_star(texto)
-    premise = " ".join(star.get("action", [])) + " " + " ".join(star.get("result", []))
-    report = {
-        "competence": "ICB4 4.4.5 Leadership",
-        "star": star,
-        "indicators": {},
-        "overall": {}
-    }
+    """Orquestación completa con manejo de errores (mensaje visible en 'Estado')."""
+    try:
+        if not texto or not texto.strip():
+            return "Introduce un caso en formato STAR (o texto libre).", {}, {}
+
+        star = extract_star(texto)
+        premise = " ".join(star.get("action", [])) + " " + " ".join(star.get("result", []))
 
-    scores = []
-    table_rows = []
-    for ind, hyps in HYP.items():
-        s, ev, avg = score_indicator(premise, hyps)
-        scores.append(s)
-        report["indicators"][ind] = {
-            "score": s,
-            "evidence": [ev[0][0], ev[1][0]] if len(ev) > 1 else [ev[0][0]],
-            "rationale": f"Promedio de entailment NLI ≈ {avg:.2f}"
+        report = {
+            "competence": "ICB4 4.4.5 Leadership",
+            "star": star,
+            "indicators": {},
+            "overall": {}
         }
-        table_rows.append([ind, s, f"{avg:.2f}", " / ".join([h for h,_ in ev])])
 
-    overall = round(sum(scores) / max(1, len(scores)), 2)
-    report["overall"] = {
-        "score_weighted": overall,
-        "notes": "Ajusta umbrales tras validar con 10–20 ejemplos reales."
-    }
+        scores = []
+        table_rows = []
+        for ind, hyps in HYP.items():
+            s, ev, avg = score_indicator(premise, hyps)
+            scores.append(s)
+            report["indicators"][ind] = {
+                "score": s,
+                "evidence": [e[0] for e in ev],
+                "rationale": f"Promedio de entailment NLI ≈ {avg:.2f}"
+            }
+            table_rows.append([ind, s, f"{avg:.2f}", " / ".join([h for h, _ in ev])])
 
-    msg = (
-        f"Evaluación completada.\n"
-        f"Score global (0–4): {overall}\n"
-        f"Sugerencia: revisa la evidencia y ajusta umbrales según tu rúbrica."
-    )
+        overall = round(sum(scores) / max(1, len(scores)), 2)
+        report["overall"] = {
+            "score_weighted": overall,
+            "notes": "Ajusta umbrales tras validar con 10–20 ejemplos reales."
+        }
 
-    # Tabla amigable para la UI
-    table = {
-        "columns": ["Indicador", "Score (0–4)", "Entailment medio", "Mejores evidencias (hipótesis)"],
-        "data": table_rows
-    }
-    return msg, report, table
+        msg = (
+            f"Evaluación completada. Score global (0–4): {overall}\n"
+            f"Sugerencia: revisa la evidencia y ajusta umbrales según tu rúbrica."
+        )
+
+        table = {
+            "columns": ["Indicador", "Score (0–4)", "Entailment medio", "Mejores evidencias (hipótesis)"],
+            "data": table_rows
+        }
+        return msg, report, table
 
-### --------- UI Gradio ---------
+    except Exception as e:
+        return f"⚠️ Error en evaluate(): {type(e).__name__}: {e}", {}, {"columns": [], "data": []}
+
+# --------- UI Gradio ---------
 with gr.Blocks(title="ICB4 4.4.5 Leadership — Evaluación STAR (FRAQX)") as demo:
     gr.Markdown(
         """
@@ -236,29 +273,38 @@ with gr.Blocks(title="ICB4 4.4.5 Leadership — Evaluación STAR (FRAQX)") as de
     with gr.Row():
         json_out = gr.JSON(label="Reporte JSON")
     with gr.Row():
-        table = gr.Dataframe(headers=["Indicador", "Score (0–4)", "Entailment medio", "Mejores evidencias (hipótesis)"],
-                             datatype=["str","number","str","str"],
-                             interactive=False)
+        table = gr.Dataframe(
+            headers=["Indicador", "Score (0–4)", "Entailment medio", "Mejores evidencias (hipótesis)"],
+            datatype=["str", "number", "str", "str"],
+            interactive=False
+        )
 
     ejemplos = gr.Examples(
-        examples=[
-            ["S: El proyecto CRM estaba retrasado 6 semanas y el equipo estaba desmotivado. "
-             "T: Recuperar el plan y mejorar la colaboración en 2 sprints. "
-             "A: Organicé una sesión de visión y valores con el Product Owner; definí métricas de entrega y calidad; "
-             "implementé dailies focalizadas; ofrecí mentoring a dos líderes junior; negocié con stakeholders para "
-             "priorizar un backlog mínimo; comuniqué riesgos y comprometí fechas realistas. "
-             "R: Recuperamos 4 semanas en dos sprints; NPS interno +22; se redujo retrabajo 18%; el equipo pidió "
-             "mantener las prácticas y dos líderes asumieron más responsabilidad."]
-        ],
+        examples=[[
+            "S: El proyecto CRM estaba retrasado 6 semanas y el equipo estaba desmotivado. "
+            "T: Recuperar el plan y mejorar la colaboración en 2 sprints. "
+            "A: Organicé una sesión de visión y valores con el Product Owner; definí métricas de entrega y calidad; "
+            "implementé dailies focalizadas; ofrecí mentoring a dos líderes junior; negocié con stakeholders para "
+            "priorizar un backlog mínimo; comuniqué riesgos y comprometí fechas realistas. "
+            "R: Recuperamos 4 semanas en dos sprints; NPS interno +22; se redujo retrabajo 18%; el equipo pidió "
+            "mantener las prácticas y dos líderes asumieron más responsabilidad."
+        ]],
         inputs=[texto],
     )
 
     def run_eval(t):
         msg, rep, tbl = evaluate(t)
-        return msg, rep, gr.update(value=tbl["data"], headers=tbl["columns"])
+        return msg, rep, gr.update(value=tbl.get("data", []), headers=tbl.get("columns", []))
 
     btn.click(run_eval, inputs=[texto], outputs=[status, json_out, table])
 
+# --- Lanzamiento con cola y sin SSR (estabiliza CPU Basic) ---
 if __name__ == "__main__":
-    demo.launch()
+    demo.queue(
+        max_size=16,         # cola máxima
+        concurrency_count=1  # 1 ejecución simultánea en CPU Basic
+    ).launch(
+        ssr_mode=False,      # desactiva SSR para evitar glitches
+        show_error=True
+    )