"""Higiene de la salida de los modelos que devuelven TEXTO libre. Vivía dentro de `hf_space.py` cuando el Space era la única ruta de prosa. Ahora también la usa `medgemma.py` cuando el modelo local elegido está declarado como `prosa` en la lista blanca: no todo modelo que corre en Ollama sabe respetar una decodificación restringida —medido con qwen2.5:7b, que devolvía JSON válido con los tres campos estructurados vacíos—, y para ésos es mejor pedir prosa y envolverla que aceptar un objeto hueco. Los marcadores de `` y `` son de la familia Gemma, pero el resto (razonamiento filtrado, bucle de repetición, frase cortada) son modos de fallo genéricos de cualquier modelo pequeño, así que se aplican a las dos rutas por igual. """ from __future__ import annotations import re from ..schemas import InterpretacionClinica from .base import ErrorModelo # medGemma es un modelo con "pensamiento": de forma intermitente emite una cadena de # razonamiento en inglés (etiquetada `thought` / "thinking process") en lugar de responder, # y a veces degenera en un bucle de repetición que agota el presupuesto de tokens sin llegar # a la respuesta. Estos marcadores permiten detectar y descartar esa salida defectuosa. _MARCADOR_PENSAMIENTO = re.compile( r"(?im)^\s*(thought|thinking)\s*:?\s*$" r"|here'?s\s+(a|my)\s+thinking\s+process" r"|thinking\s+process\s+to\s+arrive" r"|proceso\s+de\s+(pensamiento|razonamiento)" ) def limpiar_respuesta(text: str) -> str: """Versión compacta de la limpieza que antes vivía en ia.js (limpiarRespuesta). Sólo se aplica a las rutas de PROSA (texto crudo del modelo); las rutas con salida estructurada validan contra el esquema y no la necesitan. """ if "model" in text: text = text.split("model")[-1] if "" in text: text = text[: text.index("")] if "" in text: text = text.split("")[-1] elif "" in text: text = "".join(text.split("")[1:]).strip() text = re.sub(r"", "", text) text = re.sub(r"\w+\n?", "", text) text = re.sub(r"^\d+\s+(medical assistant|assistant|model)\s*", "", text, flags=re.I) # LaTeX y bloques matemáticos text = re.sub(r"\$\\boxed\{[^}]*\}\$", "", text) text = re.sub(r"\\[a-zA-Z]+(\{[^}]*\})?", "", text) text = re.sub(r"\$[^$]*\$", "", text) text = re.sub(r"\n{3,}", "\n\n", text).strip() text = _cortar_bucle_lineas(text) # Corta al primer párrafo repetido (bucle del modelo) vistos: set[str] = set() sin_rep = [] for p in re.split(r"\n\n+", text): clave = p.strip()[:80] if clave in vistos: break vistos.add(clave) sin_rep.append(p) return "\n\n".join(sin_rep).strip() or "Sin respuesta del modelo." def _cortar_bucle_lineas(text: str) -> str: """Trunca en cuanto una línea sustantiva se repite por 3.ª vez (bucle a nivel de viñeta, que la deduplicación por párrafos `\\n\\n` no detecta).""" conteo: dict[str, int] = {} salida: list[str] = [] for linea in text.split("\n"): clave = linea.strip() if len(clave) > 15: conteo[clave] = conteo.get(clave, 0) + 1 if conteo[clave] >= 3: break salida.append(linea) return "\n".join(salida) # Caracteres que pueden ir DESPUÉS del punto final sin que la frase esté cortada: énfasis # markdown, comillas y cierres de paréntesis. _ADORNOS_FINALES = "*_`\"'”»)]}" _FIN_DE_FRASE = ".!?…:" # Una línea final corta puede ser un ítem de lista o un encabezado legítimos sin puntuación # ("- Ecografía abdominal"). Sólo una línea larga sin cierre delata una frase interrumpida. _LARGO_MINIMO_TRUNCADO = 40 def interpretacion_truncada(text: str) -> bool: """True si la respuesta se corta a mitad de frase. El Space puede devolver una interpretación cortada en seco (visto en producción: «…los diferenciales incluyen la lip», justo antes de nombrar el diferencial clave). La limpieza no la toca —el corte viene de arriba— y el resto de comprobaciones no la ven: el texto es largo, no repite y no filtra razonamiento. Sin esto, una interpretación incompleta llega al veterinario con aspecto de completa, callando justo lo que importa. """ limpio = text.rstrip() if not limpio: return True ultima_linea = limpio.split("\n")[-1].strip() if len(ultima_linea) < _LARGO_MINIMO_TRUNCADO: return False return limpio.rstrip(_ADORNOS_FINALES)[-1:] not in _FIN_DE_FRASE def interpretacion_defectuosa(text: str) -> bool: """True si la salida limpiada no es una interpretación válida: demasiado corta, cadena de razonamiento filtrada, bucle de repetición o frase cortada. Se usa para forzar un reintento.""" if len(text.strip()) < 40: return True if _MARCADOR_PENSAMIENTO.search(text[:500]): return True if interpretacion_truncada(text): return True conteo: dict[str, int] = {} for linea in text.split("\n"): clave = linea.strip() if len(clave) > 15: conteo[clave] = conteo.get(clave, 0) + 1 if conteo[clave] >= 3: return True return False def interpretacion_desde_prosa(texto: str) -> InterpretacionClinica: """Limpia el texto crudo y lo envuelve en el contrato estructurado, o lanza ErrorModelo. Salida defectuosa (razonamiento filtrado / bucle / frase cortada) → error reintentable: el servicio vuelve a muestrear una vez y suele obtener una respuesta correcta. Se distingue el motivo porque una respuesta truncada se ve completa y el mensaje genérico despistaría al diagnosticar. `requiere_derivacion=True` es un marcador de posición conservador, NO una lectura del modelo: en esta ruta el modelo devuelve prosa y no puede rellenar campos estructurados. Quien le pone valor real es el servicio, desde el motor determinista (`_derivacion_en_ruta_de_prosa`). Dejarlo constante hacía que `acierto_derivacion` midiera este default y no al modelo, y contradecía al propio texto en un panel normal. """ limpio = limpiar_respuesta(texto) if interpretacion_defectuosa(limpio): truncada = interpretacion_truncada(limpio) motivo = ( "la respuesta llegó cortada a mitad de frase" if truncada else "el modelo devolvió razonamiento o texto repetido, no la interpretación" ) raise ErrorModelo(f"Respuesta inutilizable del modelo: {motivo}.", truncado=truncada) return InterpretacionClinica( interpretacion=limpio, requiere_derivacion=True, idioma="es", )