college-infos/pronote_sync/synthesis/litellm.py

"""Fournisseur de synthèse IA via ``litellm``.

Ce module définit :class:`LiteLLMSynthesisProvider`, un fournisseur de
synthèse IA qui délègue l'appel à ``litellm.completion`` en réutilisant le
prompt système et la construction de prompt de
:class:`~pronote_sync.synthesis.openai.OpenAISynthesisProvider`. La méthode
:meth:`LiteLLMSynthesisProvider.generate` ne lève jamais d'exception : tout
échec est journalisé (message rédigé) et dégradé en retour ``None``.

Ce module nécessite l'extra ``ai-litellm`` (le paquet ``litellm``).
"""

from __future__ import annotations

import logging
from typing import Any

import litellm
from pydantic import SecretStr

from pronote_sync.models.synthesis import SynthesisInput, SynthesisResult
from pronote_sync.synthesis.openai import OpenAISynthesisProvider
from pronote_sync.utils.redaction import redact_secrets

logger = logging.getLogger(__name__)

__all__ = ["LiteLLMSynthesisProvider"]


class LiteLLMSynthesisProvider:
    """Fournisseur de synthèse IA utilisant ``litellm``.

    Réutilise le prompt système et la construction de prompt de
    :class:`OpenAISynthesisProvider`. Ne lève jamais d'exception : en cas
    d'échec, :meth:`generate` retourne ``None``.
    """

    SYSTEM_PROMPT = OpenAISynthesisProvider.SYSTEM_PROMPT
    MAX_LENGTH = OpenAISynthesisProvider.MAX_LENGTH
    TIMEOUT = OpenAISynthesisProvider.TIMEOUT
    TEMPERATURE = OpenAISynthesisProvider.TEMPERATURE

    def __init__(
        self, api_key: SecretStr, base_url: str | None = None, model: str = "gpt-4o-mini"
    ) -> None:
        """Initialise le fournisseur LiteLLM.

        La clé API reste encapsulée dans un :class:`pydantic.SecretStr` et
        n'est déballée qu'au moment de l'appel à ``litellm.completion``, afin
        d'éviter toute fuite en clair dans les logs.

        :param api_key: Clé API du fournisseur (secret).
        :param base_url: URL de base de l'API (``None`` pour l'URL par défaut).
        :param model: Identifiant du modèle.
        """
        self._api_key = api_key
        self._base_url = base_url
        self._model = model

    def generate(self, input_data: SynthesisInput) -> SynthesisResult | None:
        """Génère une synthèse IA à partir des données d'entrée.

        Construit le prompt via ``OpenAISynthesisProvider._build_prompt``,
        appelle ``litellm.completion`` en transmettant explicitement
        ``api_key`` (la clé secrète n'est déballée qu'à cet appel) et
        ``base_url`` (uniquement si non ``None``) ainsi que ``timeout``,
        puis valide la réponse via
        ``OpenAISynthesisProvider._validate_output`` (suppression des
        emojis, rejet des titres/listes/HTML, réduction aux espaces de
        début et de fin), avant troncature à :attr:`MAX_LENGTH`. Ne lève
        jamais d'exception : toute erreur est journalisée (message rédigé)
        et dégradée en retour ``None``.

        :param input_data: Données de synthèse (diff agenda, messages, événements).
        :return: Résultat de la synthèse, ou ``None`` en cas d'échec ou de
            réponse vide.
        :rtype: SynthesisResult | None
        """
        try:
            completion_kwargs: dict[str, Any] = {
                "model": self._model,
                "messages": [
                    {"role": "system", "content": self.SYSTEM_PROMPT},
                    {
                        "role": "user",
                        "content": OpenAISynthesisProvider._build_prompt(input_data),
                    },
                ],
                "max_tokens": self.MAX_LENGTH,
                "temperature": self.TEMPERATURE,
                "timeout": self.TIMEOUT,
            }
            if self._base_url is not None:
                completion_kwargs["base_url"] = self._base_url
            response = litellm.completion(
                api_key=self._api_key.get_secret_value(), **completion_kwargs
            )
            raw_text = response.choices[0].message.content
            if not raw_text:
                return None
            validated = OpenAISynthesisProvider._validate_output(raw_text)
            if validated is None:
                return None
            synthesis_text = validated[: self.MAX_LENGTH].strip()
            if not synthesis_text:
                return None
            return SynthesisResult(text=synthesis_text)
        except Exception as e:
            logger.error(
                "Échec de la génération de la synthèse IA (litellm) : %s",
                redact_secrets(str(e), extra_secrets=[self._api_key]),
            )
            return None