Spaces:
Sleeping
Sleeping
| """Cliente del HF Space (Gradio) donde está alojado medGemma. | |
| Porta el flujo de api/hf_proxy.php: sube las imágenes al endpoint /upload, invoca | |
| /call/analyze, sondea el stream SSE y recupera el texto. Como el Space devuelve TEXTO | |
| libre (no puede forzar un esquema JSON), la salida se limpia de artefactos del modelo y | |
| se envuelve en el campo `interpretacion` de InterpretacionClinica, manteniendo el contrato | |
| estructurado hacia el frontend. | |
| """ | |
| from __future__ import annotations | |
| import base64 | |
| import binascii | |
| import json | |
| import re | |
| import httpx | |
| from ..config import obtener_config | |
| from ..schemas import InterpretacionClinica, esquema_estructurado | |
| from .base import ErrorModelo | |
| # La higiene de la prosa vive en `prosa.py` desde que también la usan los modelos locales | |
| # declarados como `prosa` en la lista blanca. Se reexporta aquí porque era la API pública de | |
| # este módulo (y lo que importan sus tests de regresión). | |
| from .prosa import ( # noqa: F401 | |
| _cortar_bucle_lineas, | |
| interpretacion_defectuosa, | |
| interpretacion_desde_prosa, | |
| interpretacion_truncada, | |
| limpiar_respuesta, | |
| ) | |
| _DATA_URL = re.compile(r"^data:(image/[\w+]+);base64,(.+)$", re.DOTALL) | |
| class HFSpaceClient: | |
| """Cliente del Space. Dos modos según `hf_space_estructurado`: | |
| - prosa (`medgemma-hf`): el Space devuelve texto libre y se envuelve en `interpretacion`. | |
| - estructurado (`medgemma-hf-json`): se le manda el JSON Schema y devuelve el objeto ya | |
| formado. El nombre distinto NO es cosmético: el servicio decide por él si aplica el | |
| system prompt de prosa, si exige campos estructurados y si suple `requiere_derivacion`. | |
| """ | |
| def __init__(self) -> None: | |
| cfg = obtener_config() | |
| self.estructurado = cfg.hf_space_estructurado | |
| self.prosa = not self.estructurado | |
| self.nombre = "medgemma-hf-json" if self.estructurado else "medgemma-hf" | |
| self.modelo = "hf-space" | |
| if not cfg.hf_space_url: | |
| raise ErrorModelo( | |
| "MORPHOS_HF_SPACE_URL no configurada para la ruta HF Space.", reintentable=False | |
| ) | |
| self._space = cfg.hf_space_url.rstrip("/") | |
| self._key = cfg.hf_api_key | |
| def _headers(self, extra: dict | None = None) -> dict: | |
| h = dict(extra or {}) | |
| if self._key: | |
| h["Authorization"] = f"Bearer {self._key}" | |
| return h | |
| async def _subir_imagen(self, cliente: httpx.AsyncClient, data_url: str) -> dict | None: | |
| m = _DATA_URL.match(data_url) | |
| if not m: | |
| return None | |
| mime = m.group(1) | |
| ext = mime.split("/")[-1] or "jpg" | |
| try: | |
| binario = base64.b64decode(m.group(2)) | |
| except (binascii.Error, ValueError): | |
| return None | |
| try: | |
| r = await cliente.post( | |
| f"{self._space}/upload", | |
| files={"files": (f"image.{ext}", binario, mime)}, | |
| headers=self._headers(), | |
| ) | |
| paths = r.json() if r.status_code < 400 else None | |
| except (httpx.HTTPError, ValueError): | |
| paths = None | |
| if not isinstance(paths, list) or not paths: | |
| # Si el upload falla, envía la imagen inline (igual que el proxy PHP original). | |
| return {"url": data_url, "orig_name": f"image.{ext}", "mime_type": mime} | |
| path = paths[0] | |
| return {"path": path, "url": f"{self._space}/file={path}", "orig_name": f"image.{ext}", "mime_type": mime} | |
| async def interpretar( | |
| self, sistema: str, mensaje_usuario: str, imagenes: list[str] | |
| ) -> InterpretacionClinica: | |
| prompt = f"{sistema}\n\n{mensaje_usuario}" | |
| async with httpx.AsyncClient(timeout=120) as cliente: | |
| data: list = [] | |
| for img in imagenes[:4]: | |
| data.append(await self._subir_imagen(cliente, img)) | |
| while len(data) < 4: | |
| data.append(None) | |
| data.append(prompt) | |
| # 6.º elemento del contrato Gradio: el esquema. Cadena vacía = modo prosa, que es | |
| # lo que el Space entiende como "sin restricción". | |
| data.append(json.dumps(esquema_estructurado()) if self.estructurado else "") | |
| try: | |
| r = await cliente.post( | |
| f"{self._space}/call/analyze", | |
| json={"data": data}, | |
| headers=self._headers({"Content-Type": "application/json"}), | |
| ) | |
| except httpx.HTTPError as exc: | |
| raise ErrorModelo(f"No se pudo contactar el HF Space: {exc}") from exc | |
| if r.status_code == 429: | |
| # Cuota de ZeroGPU agotada o límite de tasa del router de HF. Reintentar aquí | |
| # sería contraproducente: gasta otra reserva de GPU del mismo pozo agotado. | |
| raise ErrorModelo( | |
| "El modelo está saturado (cuota de GPU agotada). Inténtalo de nuevo en unos " | |
| "minutos, o configura la ruta local de Ollama.", | |
| reintentable=False, | |
| saturado=True, | |
| ) | |
| if r.status_code >= 400: | |
| # 5xx puede ser transitorio; 4xx (auth, petición mal formada) no se arregla solo. | |
| raise ErrorModelo( | |
| f"HF Space devolvió HTTP {r.status_code}", | |
| reintentable=r.status_code >= 500, | |
| ) | |
| event_id = (r.json() or {}).get("event_id") | |
| if not event_id: | |
| raise ErrorModelo("El HF Space no devolvió event_id.") | |
| try: | |
| stream = await cliente.get( | |
| f"{self._space}/call/analyze/{event_id}", headers=self._headers() | |
| ) | |
| except httpx.HTTPError as exc: | |
| raise ErrorModelo(f"Fallo sondeando el HF Space: {exc}") from exc | |
| texto, error = self._parsear_sse(stream.text) | |
| if error: | |
| # El Space suele reportar aquí la cuota de ZeroGPU agotada; eso no se arregla | |
| # reintentando (gastaría otra reserva del mismo pozo). | |
| sin_cuota = bool(re.search(r"quota|gpu|exceed|limit", error, re.I)) | |
| raise ErrorModelo( | |
| f"HF Space: {error}", reintentable=not sin_cuota, saturado=sin_cuota | |
| ) | |
| if texto is None: | |
| raise ErrorModelo("Sin respuesta del modelo (HF Space).") | |
| if self.estructurado: | |
| return self._parsear_estructurado(texto) | |
| return interpretacion_desde_prosa(texto) | |
| def _parsear_estructurado(texto: str) -> InterpretacionClinica: | |
| """Valida el JSON restringido del Space contra el esquema. | |
| La limpieza de prosa (`limpiar_respuesta`, `interpretacion_defectuosa`) NO se aplica | |
| aquí: son heurísticas sobre texto libre y sobre un JSON darían falsos positivos. Si el | |
| JSON no valida se trata como salida malformada —error reintentable—, que es justo el | |
| caso para el que existe el reintento del servicio. | |
| """ | |
| try: | |
| datos = json.loads(texto) | |
| except json.JSONDecodeError as exc: | |
| raise ErrorModelo(f"El Space devolvió JSON inválido: {exc}") from exc | |
| try: | |
| return InterpretacionClinica.model_validate(datos) | |
| except Exception as exc: # noqa: BLE001 — ValidationError de pydantic | |
| raise ErrorModelo(f"JSON del Space fuera de esquema: {exc}") from exc | |
| def _parsear_sse(stream: str) -> tuple[str | None, str | None]: | |
| """Devuelve (texto, error). Los eventos `error` del Space (p.ej. cuota ZeroGPU | |
| agotada tras la primera petición) se propagan igual que hacía api/hf_proxy.php, | |
| en lugar de descartarse y acabar en un genérico "Sin respuesta del modelo". | |
| """ | |
| ultimo_evento = "" | |
| resultado = None | |
| error = None | |
| for raw in stream.split("\n"): | |
| linea = raw.rstrip("\r") | |
| if linea.startswith("event:"): | |
| ultimo_evento = linea[6:].strip() | |
| elif linea.startswith("data:"): | |
| try: | |
| parsed = json.loads(linea[5:].strip()) | |
| except json.JSONDecodeError: | |
| continue | |
| if ultimo_evento in ("complete", "process_completed"): | |
| resultado = parsed[0] if isinstance(parsed, list) else parsed.get("output", parsed) | |
| elif ultimo_evento == "error": | |
| if isinstance(parsed, dict): | |
| error = parsed.get("error") or parsed.get("message") or "Error del modelo." | |
| elif isinstance(parsed, str): | |
| error = parsed | |
| else: | |
| error = "Error del modelo." | |
| texto = resultado if isinstance(resultado, str) else (str(resultado) if resultado is not None else None) | |
| return texto, error | |