"""
Contrat d'interface abstraite pour les moteurs de generation de texte (LLM).
"""
from abc import ABC, abstractmethod
from collections.abc import Generator
[docs]
class BaseLlm(ABC):
"""
Interface abstraite pour les moteurs LLM locaux.
Fournit le chargement, le formatage de prompt, la generation en streaming
et la gestion du cache.
Chaque moteur declare **son propre** reglage de reflexion, parce que les
runtimes ne l'expriment pas de la meme facon : le daemon MTPLX prend un
``reasoning_effort`` a trois crans, le serveur OptiQ passe par
``chat_template_kwargs``, et mlx-lm applique le drapeau au gabarit. Un
budget unique partout ferait dire trois choses differentes au meme chiffre.
"""
#: Paliers de budget de reflexion, du plus sobre au plus bavard. Le
#: premier element est toujours 0, qui signifie "pas de reflexion".
reasoningSteps: tuple[int, ...] = (0, 512, 1024, 2048)
#: Budget applique quand l'utilisateur presse le bouton sans avoir choisi
#: de niveau : le palier immediatement apres OFF.
defaultReasoningBudget: int = 1024
#: Nom du parametre du runtime qui regle la reflexion. Purement
#: documentaire : il dit comment ce moteur exprime le budget.
reasoningParameter: str = "chat_template_kwargs.enable_thinking"
#: True quand le modele est servi ailleurs, sur du materiel qui n'est pas
#: cette machine.
#:
#: Tout ce que Mirza mesure sur un cache local n'a alors pas d'objet : il n'y
#: a ni cache a quantifier, ni fenetre a borner, ni jetons a compter pour
#: eviter de la deborder. La fenetre est celle du fournisseur, et elle ne se
#: rogne pas ici. Le raisonnement non plus n'a pas a etre suivi : il ne se
#: lit pas, il coute, et il ne revient pas dans l'historique.
#:
#: Ce drapeau dit "ce moteur n'a pas de memoire chez nous". Il ne dit pas que
#: le contexte est sans importance : le prompt complet part quand meme, et
#: c'est ce que le modele doit recevoir.
isRemote: bool = False
estimatedKvTokens: int = 0
currentKvTokens: int = 0
[docs]
@classmethod
def reasoningEffortFor(cls, budget: int) -> str:
"""
Traduit un budget en cran, a partir des paliers de ce moteur.
Parameters
----------
budget : int
Budget de reflexion demande.
Returns
-------
str
Le nom du palier : celui du budget, ou le dernier si le budget
n'est pas exactement un palier.
"""
names = ("off", "low", "medium", "high")
index = 0
for i, step in enumerate(cls.reasoningSteps):
if budget >= step:
index = i
return names[min(index, len(names) - 1)]
[docs]
def resolveMaxTokens(self, answerTokens: int, ceiling: int) -> int:
"""
Calcule le budget total d'un tour : la reflexion plus la reponse.
Le total est borne par le plafond dur, mais jamais en dessous de ce qui
reste pour la reponse : une reflexion qui consomme tout son budget
laisserait la reponse a zero jeton.
Parameters
----------
answerTokens : int
Jetons reserves a la reponse finale.
ceiling : int
Plafond dur sur le total.
Returns
-------
int
Le nombre de jetons a demander au runtime.
"""
if not getattr(self, "enableThinking", False):
return answerTokens
return min(ceiling, max(0, self.reasoningBudget) + answerTokens)
[docs]
@abstractmethod
def loadModel(self, systemPrompt: str = "") -> None:
"""
Charge le modele et calibre le cache KV.
Parameters
----------
systemPrompt : str, optional
Prompt systeme initial pour calibrer le cache. Defaut : "".
"""
[docs]
@abstractmethod
def streamResponse(self, formattedPrompt: str) -> Generator[str]:
"""
Genere le flux de tokens en streaming temps reel.
Parameters
----------
formattedPrompt : str
Prompt complet formate.
Yields
------
str
Fragments de texte ou tokens generes en temps reel.
"""
[docs]
def clearCache(self) -> None:
"""
Purge les caches KV et la memoire transitoire du moteur LLM.
"""
[docs]
def setReasoningBudget(self, budget: int) -> None:
"""
Ajuste le budget de reflexion alloue au modele.
"""
[docs]
def setThinkingMode(self, enabled: bool) -> None:
"""
Active ou desactive le mode reflexion du modele.
"""
[docs]
@abstractmethod
def unload(self) -> None:
"""
Libere l'integralite des ressources GPU et memoire allouees.
"""