feat(health): gate vision + détection des modèles aveugles

Détecte les modèles VLM/grounding « aveugles » (capabilities sans vision, ex.
UI-TARS réimporté sans mmproj) pour éviter le HTTP 500 silencieux masqué par
la cascade de grounding.

- core/detection/model_health.py : has_vision_capability() (cache, fail-open)
  + smoke_check_models()
- core/execution/input_handler.py : gate vision dans _grounding_ui_tars
  (skip propre vers niveau 3 si modèle aveugle, plus de 500 silencieux)
- tests/unit/test_model_health.py : 6 tests (vision/aveugle/fail-open/cache/smoke)

Incident 2026-06-08 : UI-TARS sans mmproj -> niveau 2 cascade en 500 silencieux,
non détecté (hors chemin runtime démo + échec avalé par fallback + zéro test).
NB : le smoke non bloquant au démarrage (api_stream.py startup) reste dans le WIP
de la branche, mélangé au préflight non committé.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dom
2026-06-08 11:51:18 +02:00
parent 5b2afa3629
commit d00fe7b00b
3 changed files with 167 additions and 0 deletions

View File

@@ -0,0 +1,97 @@
"""Santé des modèles VLM/grounding — détection des modèles « aveugles ».
Motivation (incident 2026-06-08) : un modèle de grounding réimporté sans son projecteur
vision (`mmproj`) déclare des `capabilities` sans `vision` et renvoie HTTP 500 sur toute
requête image. Dans la cascade `find_element_on_screen`, l'échec était avalé (`return None`)
et masqué par le fallback VLM → panne invisible malgré les tests.
Ce module permet de :
- **gater** un appel image : vérifier que le modèle a `vision` avant de lui envoyer une image
(évite le 500, skip propre vers le niveau suivant) ;
- **smoke-tester** les modèles de grounding/VLM au démarrage : rendre une panne visible
immédiatement plutôt que noyée dans un `warning` runtime.
Volontairement sans dépendance lourde : un simple appel `/api/show` Ollama.
"""
from __future__ import annotations
import logging
import os
from typing import Dict, List
import requests
logger = logging.getLogger(__name__)
DEFAULT_ENDPOINT = os.environ.get("OLLAMA_URL", "http://localhost:11434")
# Cache (endpoint::model) -> bool. Un modèle ne change pas de capacité en cours de session.
_VISION_CACHE: Dict[str, bool] = {}
def has_vision_capability(
model: str,
endpoint: str = DEFAULT_ENDPOINT,
*,
use_cache: bool = True,
timeout: float = 5.0,
) -> bool:
"""Retourne True si le modèle Ollama déclare la capacité ``vision``.
Interroge ``/api/show`` et lit ``capabilities``. Résultat mis en cache par
``(endpoint, model)``.
**Fail-open** : en cas d'erreur réseau/HTTP sur ``/api/show`` (indisponibilité
transitoire), retourne ``True`` — on ne bloque pas le grounding sur un doute ;
l'appel image en aval gérera l'échec. Seule une réponse explicite **sans** ``vision``
retourne ``False`` (modèle réellement aveugle).
"""
key = f"{endpoint}::{model}"
if use_cache and key in _VISION_CACHE:
return _VISION_CACHE[key]
try:
resp = requests.post(f"{endpoint}/api/show", json={"name": model}, timeout=timeout)
if resp.status_code != 200:
logger.debug("model_health: /api/show %s → HTTP %s (fail-open)", model, resp.status_code)
return True
caps = resp.json().get("capabilities", []) or []
has_vision = "vision" in caps
_VISION_CACHE[key] = has_vision
if not has_vision:
logger.warning(
"model_health: modèle '%s' SANS capacité 'vision' (capabilities=%s) — "
"modèle aveugle, les requêtes image échoueront",
model,
caps,
)
return has_vision
except Exception as e: # réseau, JSON, timeout
logger.debug("model_health: échec vérification vision %s: %s (fail-open)", model, e)
return True
def smoke_check_models(models: List[str], endpoint: str = DEFAULT_ENDPOINT) -> Dict[str, bool]:
"""Vérifie la capacité ``vision`` d'une liste de modèles (au démarrage/healthcheck).
Non bloquant : logue ``info`` par modèle sain, ``error`` par modèle aveugle.
Retourne ``{model: has_vision}``.
"""
results: Dict[str, bool] = {}
for m in models:
if not m:
continue
ok = has_vision_capability(m, endpoint, use_cache=False)
results[m] = ok
if ok:
logger.info("model_health[smoke]: %s → vision OK", m)
else:
logger.error(
"model_health[smoke]: %s → AVEUGLE (pas de vision) — grounding image KO sur ce modèle",
m,
)
return results
def reset_cache() -> None:
"""Vide le cache de capacités (tests, ou après réimport d'un modèle)."""
_VISION_CACHE.clear()

View File

@@ -590,6 +590,16 @@ def _grounding_ui_tars(target_text: str, target_description: str = "", monitor_i
ollama_url = os.environ.get("OLLAMA_URL", "http://localhost:11434") ollama_url = os.environ.get("OLLAMA_URL", "http://localhost:11434")
model = "0000/ui-tars-1.5-7b-q8_0:7b" model = "0000/ui-tars-1.5-7b-q8_0:7b"
# Gate santé : ne pas envoyer d'image à un modèle « aveugle » (sans capacité vision).
# Évite le HTTP 500 silencieux qui masquait la panne (incident 2026-06-08, UI-TARS sans mmproj).
from core.detection.model_health import has_vision_capability
if not has_vision_capability(model, ollama_url):
logger.warning(
"[Grounding/UI-TARS] modèle '%s' sans capacité 'vision' — skip propre vers niveau 3",
model,
)
return None
logger.info(f"[Grounding/UI-TARS] Envoi à {model}: '{prompt}'") logger.info(f"[Grounding/UI-TARS] Envoi à {model}: '{prompt}'")
response = requests.post( response = requests.post(

View File

@@ -0,0 +1,60 @@
"""Tests unitaires de core.detection.model_health (détection modèles aveugles)."""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from core.detection import model_health
@pytest.fixture(autouse=True)
def _clear_cache():
model_health.reset_cache()
yield
model_health.reset_cache()
def _resp(status=200, capabilities=None):
r = MagicMock()
r.status_code = status
r.json.return_value = {"capabilities": capabilities if capabilities is not None else []}
return r
def test_vision_model_returns_true():
with patch("core.detection.model_health.requests.post", return_value=_resp(200, ["completion", "vision"])):
assert model_health.has_vision_capability("gemma4:26b", "http://x:11434") is True
def test_blind_model_returns_false():
with patch("core.detection.model_health.requests.post", return_value=_resp(200, ["tools", "completion"])):
assert model_health.has_vision_capability("0000/ui-tars-1.5-7b-q8_0:7b", "http://x:11434") is False
def test_http_error_is_fail_open():
with patch("core.detection.model_health.requests.post", return_value=_resp(404)):
assert model_health.has_vision_capability("absent:latest", "http://x:11434") is True
def test_exception_is_fail_open():
with patch("core.detection.model_health.requests.post", side_effect=ConnectionError("boom")):
assert model_health.has_vision_capability("any:latest", "http://x:11434") is True
def test_cache_avoids_second_call():
with patch("core.detection.model_health.requests.post", return_value=_resp(200, ["vision"])) as p:
model_health.has_vision_capability("m", "http://x:11434")
model_health.has_vision_capability("m", "http://x:11434")
assert p.call_count == 1 # 2e appel servi par le cache
def test_smoke_check_reports_each_model():
def _fake_post(url, json=None, timeout=None):
name = (json or {}).get("name", "")
caps = ["vision"] if name == "good" else ["completion"]
return _resp(200, caps)
with patch("core.detection.model_health.requests.post", side_effect=_fake_post):
res = model_health.smoke_check_models(["good", "blind"], "http://x:11434")
assert res == {"good": True, "blind": False}