LoreMind/brain/app/infrastructure/base_openai_adapter.py

"""Socle commun aux adapters LLM « OpenAI-compatible » (OpenRouter, Gemini,
Mistral) — ils exposent tous `POST {base}/chat/completions` en SSE avec le même
schéma de payload et de flux.

Cette classe de base porte la mécanique partagée (construction du payload, appel
HTTP streamé, parsing SSE, garde-fous de timeout au temps écoulé, traduction des
erreurs). Chaque adapter concret ne fournit plus que ses spécificités :
URL, en-têtes, support du mode JSON natif, messages d'erreur, lecture de la config.

`generate` one-shot passe lui aussi par le streaming (puis recollage) pour éviter
les coupures de passerelle sur les longues générations (cf. Cloudflare 524).
"""
from __future__ import annotations

import asyncio
import json
import logging
from typing import AsyncIterator

import httpx

from app.domain.models import ChatMessage
from app.domain.ports import LLMGenerationTimeout, LLMProviderError

logger = logging.getLogger(__name__)

# Délai max pour le PREMIER token de contenu. Un modèle « en file d'attente »
# n'envoie que des keep-alive (aucun contenu) → on échoue vite et clairement au
# lieu de pendre. Le timeout réseau d'httpx ne suffit pas : des keep-alive font
# « arriver des octets » et empêchent son read-timeout de se déclencher.
_FIRST_TOKEN_TIMEOUT_SECONDS = 120.0


class BaseOpenAICompatibleAdapter:
    """Base des adapters clients d'une API OpenAI-compatible (chat/completions SSE).

    Satisfait par duck typing les ports LLMProvider et LLMChatProvider. Les
    sous-classes définissent : ``_provider_label``, ``_api_url``,
    ``_supports_json_object`` (mode JSON natif), et surchargent au besoin
    ``_headers`` / ``_error_for_status`` / les messages de timeout.
    """

    # Surchargés par les sous-classes.
    _provider_label: str = "LLM"
    _api_url: str = ""
    _supports_json_object: bool = False

    def __init__(self, api_key: str, model: str, timeout: int) -> None:
        self._api_key = api_key
        self._model = model
        self._timeout = timeout

    # --- Spécificités surchargeables ----------------------------------------

    def _headers(self) -> dict[str, str]:
        return {
            "Authorization": f"Bearer {self._api_key}",
            "Content-Type": "application/json",
        }

    def _first_token_timeout_message(self) -> str:
        return (
            f"Erreur {self._provider_label} : aucun contenu produit en "
            f"{int(_FIRST_TOKEN_TIMEOUT_SECONDS)}s — le modèle est probablement en "
            "file d'attente / saturé. Réessayez plus tard ou choisissez un autre modèle."
        )

    def _generation_timeout_message(self) -> str:
        return (
            f"Erreur {self._provider_label} : génération non terminée en {self._timeout}s. "
            "Réduisez la taille des morceaux d'import, augmentez le timeout, ou changez de modèle."
        )

    def _error_for_status(self, status_code: int, detail: str) -> LLMProviderError:
        """Erreur de domaine pour une réponse HTTP >= 400 (détail déjà lu)."""
        return LLMProviderError(
            f"Erreur {self._provider_label} (HTTP {status_code})"
            + (f" : {detail[:500]}" if detail else "")
        )

    # --- API publique (ports) -----------------------------------------------

    async def generate(
        self,
        prompt: str,
        *,
        output_format: str | None = None,
        temperature: float | None = None,
    ) -> str:
        """One-shot via streaming (puis recollage), avec garde-fous au temps écoulé."""
        return await self._collect_with_timeouts(
            [ChatMessage(role="user", content=prompt)], temperature, output_format
        )

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        *,
        system_prompt: str | None = None,
        temperature: float | None = None,
    ) -> AsyncIterator[str]:
        async for token in self._stream(messages, system_prompt, temperature):
            yield token

    # --- Mécanique partagée -------------------------------------------------

    async def _collect_with_timeouts(
        self,
        messages: list[ChatMessage],
        temperature: float | None,
        output_format: str | None,
    ) -> str:
        """Collecte le stream avec DEUX garde-fous au temps écoulé :
        - 1er token borné (`_FIRST_TOKEN_TIMEOUT_SECONDS`) : détecte un modèle bloqué
          en file d'attente (que des keep-alive, aucun contenu) → échec rapide ;
        - ceiling global (`self._timeout`) : génération qui ne se termine jamais.
        """
        async def _collect() -> str:
            chunks: list[str] = []
            agen = self._stream(messages, None, temperature, output_format)
            try:
                while True:
                    # Borne SEULEMENT l'attente du 1er token ; ensuite on laisse
                    # générer (le ceiling global couvre le reste).
                    first = _FIRST_TOKEN_TIMEOUT_SECONDS if not chunks else None
                    try:
                        token = await asyncio.wait_for(agen.__anext__(), timeout=first)
                    except StopAsyncIteration:
                        break
                    except asyncio.TimeoutError:
                        raise LLMProviderError(self._first_token_timeout_message())
                    chunks.append(token)
            finally:
                await agen.aclose()
            return "".join(chunks)

        try:
            return await asyncio.wait_for(_collect(), timeout=self._timeout)
        except asyncio.TimeoutError as exc:
            raise LLMGenerationTimeout(self._generation_timeout_message()) from exc

    def _build_body(
        self,
        messages: list[ChatMessage],
        system_prompt: str | None,
        temperature: float | None,
        output_format: str | None,
    ) -> dict[str, object]:
        payload_messages: list[dict[str, str]] = []
        if system_prompt:
            payload_messages.append({"role": "system", "content": system_prompt})
        for m in messages:
            payload_messages.append({"role": m.role, "content": m.content})

        body: dict[str, object] = {
            "model": self._model,
            "messages": payload_messages,
            "stream": True,
        }
        if temperature is not None:
            body["temperature"] = temperature
        # Mode JSON natif : supprime les fences ```json et le JSON invalide (retours
        # à la ligne bruts), principale cause de morceaux d'import ignorés. Un SCHÉMA
        # (dict) est traduit en json_object — suffisant, les grands modèles cloud
        # respectent la structure demandée par le prompt. Désactivé pour les
        # providers/modèles gratuits qui ne le supportent pas (réponse vide).
        if self._supports_json_object and output_format is not None:
            body["response_format"] = {"type": "json_object"}
        return body

    async def _stream(
        self,
        messages: list[ChatMessage],
        system_prompt: str | None,
        temperature: float | None,
        output_format: str | None = None,
    ) -> AsyncIterator[str]:
        body = self._build_body(messages, system_prompt, temperature, output_format)
        async with httpx.AsyncClient(timeout=self._timeout) as client:
            try:
                async with client.stream(
                    "POST", self._api_url, headers=self._headers(), json=body
                ) as response:
                    if response.status_code >= 400:
                        # En streaming le corps n'est pas lu automatiquement : on le
                        # lit pour exposer le détail du provider (le 429 précise le
                        # type de quota, le 401 la clé invalide…), sinon on n'a que
                        # le code HTTP nu et le diagnostic est impossible.
                        detail = (await response.aread()).decode("utf-8", "replace").strip()
                        raise self._error_for_status(response.status_code, detail)
                    async for token in self._parse_sse(response):
                        yield token
            except httpx.HTTPError as exc:
                raise LLMProviderError(self._format_http_error(exc)) from exc

    @staticmethod
    async def _parse_sse(response: httpx.Response) -> AsyncIterator[str]:
        """SSE OpenAI : lignes `data: {json}`, fin sur `data: [DONE]`."""
        async for line in response.aiter_lines():
            if not line or not line.startswith("data:"):
                continue  # lignes vides ou commentaires keep-alive (`: ...`)
            data = line[len("data:"):].strip()
            if data == "[DONE]":
                return
            try:
                obj = json.loads(data)
            except json.JSONDecodeError:
                continue
            choices = obj.get("choices")
            if not choices:
                continue
            delta = choices[0].get("delta") or {}
            content = delta.get("content")
            if content:
                yield content

    def _format_http_error(self, exc: httpx.HTTPError) -> str:
        """Message lisible (timeout, quota 429, crédits 402, modèle inconnu…)."""
        if isinstance(exc, httpx.TimeoutException):
            return (
                f"Erreur {self._provider_label} : délai dépassé (timeout {self._timeout}s). "
                "Le modèle a mis trop de temps — réduis la taille des morceaux d'import ou "
                "augmente le timeout."
            )
        detail = str(exc) or exc.__class__.__name__
        return f"Erreur {self._provider_label} ({exc.__class__.__name__}) : {detail}"