mirror of
https://github.com/Ladebeze66/llm_ticket3.git
synced 2025-12-16 09:47:47 +01:00
382 lines
15 KiB
Python
382 lines
15 KiB
Python
from .base_llm import BaseLLM
|
|
import requests
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, Any
|
|
import os
|
|
import json
|
|
|
|
class Pixtral12b(BaseLLM):
|
|
"""
|
|
Classe pour interagir avec le modèle Pixtral 12B via Ollama.
|
|
"""
|
|
|
|
def __init__(self):
|
|
"""
|
|
Initialise une instance du modèle Pixtral 12B.
|
|
"""
|
|
# Initialiser avec le modèle Pixtral 12B
|
|
super().__init__("pixtral-12b-latest")
|
|
|
|
# Définir les attributs spécifiques
|
|
self.modele = "pixtral-12b-latest"
|
|
self.version = "12B"
|
|
self.api_url = "http://217.182.105.173:11434/api/generate"
|
|
|
|
# Paramètres optimisés pour Pixtral 12B
|
|
self.params: Dict[str, Any] = {
|
|
"temperature": 0.1, # Température basse pour des réponses précises
|
|
"top_p": 0.8, # Diversité modérée des réponses
|
|
"top_k": 30, # Choix des tokens les plus probables
|
|
"num_ctx": 8192, # Contexte étendu pour analyser les images
|
|
"repeat_penalty": 1.1, # Pénalité pour éviter les répétitions
|
|
"repeat_last_n": 128, # Nombre de tokens à considérer pour la pénalité
|
|
"mirostat": 0, # Désactivé
|
|
"mirostat_eta": 0.1,
|
|
"mirostat_tau": 5,
|
|
"keep_alive": int(timedelta(minutes=15).total_seconds()), # Maintien prolongé
|
|
"num_predict": 4000, # Prédiction longue pour des analyses détaillées
|
|
"min_p": 0.05,
|
|
"seed": 0,
|
|
"stop": ["</answer>", "###", "\n\n\n"],
|
|
"stream": False
|
|
}
|
|
|
|
# Timeout de requête adapté au modèle
|
|
self.request_timeout = 450 # 7.5 minutes
|
|
|
|
# Historique des interactions
|
|
self.interactions_historique = []
|
|
|
|
# État de la dernière requête
|
|
self.heureDepart = None
|
|
self.heureFin = None
|
|
self.dureeTraitement = timedelta(0)
|
|
self.reponseErreur = False
|
|
|
|
# Prompt système par défaut
|
|
self.prompt_system = "Tu es un assistant IA spécialisé dans l'analyse d'images. Tu fournis des réponses claires, précises et factuelles."
|
|
|
|
def urlBase(self) -> str:
|
|
"""
|
|
Retourne l'URL de base de l'API Ollama.
|
|
"""
|
|
return "http://217.182.105.173:11434/"
|
|
|
|
def cleAPI(self) -> str:
|
|
"""
|
|
Ollama ne nécessite pas de clé API.
|
|
"""
|
|
return ""
|
|
|
|
def urlFonction(self) -> str:
|
|
"""
|
|
Retourne l'URL spécifique à Ollama pour générer une réponse.
|
|
"""
|
|
return "api/generate"
|
|
|
|
def _preparer_contenu(self, question: str) -> Dict[str, Any]:
|
|
"""
|
|
Prépare le contenu de la requête pour Pixtral 12B.
|
|
|
|
Args:
|
|
question: La question ou instruction à envoyer au modèle
|
|
|
|
Returns:
|
|
Dictionnaire formaté pour l'API Ollama
|
|
"""
|
|
# Optimiser le prompt
|
|
prompt_optimise = self._optimiser_prompt(question)
|
|
|
|
contenu = {
|
|
"model": self.modele,
|
|
"prompt": prompt_optimise,
|
|
"options": {
|
|
"temperature": self.params["temperature"],
|
|
"top_p": self.params["top_p"],
|
|
"top_k": self.params["top_k"],
|
|
"num_ctx": self.params["num_ctx"],
|
|
"repeat_penalty": self.params["repeat_penalty"],
|
|
"repeat_last_n": self.params["repeat_last_n"],
|
|
"mirostat": self.params["mirostat"],
|
|
"mirostat_eta": self.params["mirostat_eta"],
|
|
"mirostat_tau": self.params["mirostat_tau"],
|
|
"keep_alive": self.params["keep_alive"],
|
|
"num_predict": self.params["num_predict"],
|
|
"min_p": self.params["min_p"],
|
|
"seed": self.params["seed"],
|
|
"stop": self.params["stop"],
|
|
},
|
|
"stream": self.params["stream"]
|
|
}
|
|
return contenu
|
|
|
|
def _optimiser_prompt(self, question: str) -> str:
|
|
"""
|
|
Optimise le format du prompt pour Pixtral 12B.
|
|
|
|
Args:
|
|
question: La question ou instruction originale
|
|
|
|
Returns:
|
|
Prompt optimisé pour de meilleures performances
|
|
"""
|
|
formatted_prompt = f"{self.prompt_system}\n\n{question}"
|
|
return formatted_prompt
|
|
|
|
def _traiter_reponse(self, reponse: requests.Response) -> str:
|
|
"""
|
|
Traite et nettoie la réponse fournie par Pixtral 12B.
|
|
|
|
Args:
|
|
reponse: Réponse HTTP de l'API
|
|
|
|
Returns:
|
|
Texte nettoyé de la réponse
|
|
"""
|
|
try:
|
|
data = reponse.json()
|
|
response_text = data.get("response", "")
|
|
return response_text.strip()
|
|
except Exception as e:
|
|
self.reponseErreur = True
|
|
return f"Erreur de traitement de la réponse: {str(e)}"
|
|
|
|
def interroger(self, question: str) -> str:
|
|
"""
|
|
Interroge le modèle Pixtral 12B.
|
|
|
|
Args:
|
|
question: Question ou instruction à transmettre au modèle
|
|
|
|
Returns:
|
|
Réponse du modèle
|
|
"""
|
|
url = self.urlBase() + self.urlFonction()
|
|
headers = {"Content-Type": "application/json"}
|
|
contenu = self._preparer_contenu(question)
|
|
|
|
try:
|
|
self.heureDepart = datetime.now()
|
|
response = requests.post(url=url, headers=headers, json=contenu, timeout=self.request_timeout)
|
|
self.heureFin = datetime.now()
|
|
|
|
if self.heureDepart is not None:
|
|
self.dureeTraitement = self.heureFin - self.heureDepart
|
|
else:
|
|
self.dureeTraitement = timedelta(0)
|
|
|
|
if response.status_code in [200, 201]:
|
|
self.reponseErreur = False
|
|
reponse_text = self._traiter_reponse(response)
|
|
|
|
# Enregistrer l'interaction dans l'historique
|
|
self._enregistrer_interaction(question, reponse_text)
|
|
|
|
return reponse_text
|
|
else:
|
|
self.reponseErreur = True
|
|
error_msg = f"Erreur API ({response.status_code}): {response.text}"
|
|
self._enregistrer_interaction(question, error_msg, True)
|
|
return error_msg
|
|
|
|
except requests.exceptions.Timeout:
|
|
self.heureFin = datetime.now()
|
|
if self.heureDepart is not None:
|
|
self.dureeTraitement = self.heureFin - self.heureDepart
|
|
self.reponseErreur = True
|
|
error_msg = "Timeout lors de l'appel à l'API. La requête a pris trop de temps."
|
|
self._enregistrer_interaction(question, error_msg, True)
|
|
return error_msg
|
|
|
|
except Exception as e:
|
|
self.heureFin = datetime.now()
|
|
if self.heureDepart is not None:
|
|
self.dureeTraitement = self.heureFin - self.heureDepart
|
|
else:
|
|
self.dureeTraitement = timedelta(0)
|
|
self.reponseErreur = True
|
|
error_msg = f"Erreur lors de l'interrogation: {str(e)}"
|
|
self._enregistrer_interaction(question, error_msg, True)
|
|
return error_msg
|
|
|
|
def interroger_avec_image(self, image_path: str, question: str) -> str:
|
|
"""
|
|
Interroge Pixtral 12B avec une image et du texte.
|
|
Pixtral 12B est optimisé pour l'analyse d'images.
|
|
|
|
Args:
|
|
image_path: Chemin vers l'image à analyser
|
|
question: Question ou instructions pour l'analyse
|
|
|
|
Returns:
|
|
Réponse du modèle à la question concernant l'image
|
|
"""
|
|
import base64
|
|
|
|
url = self.urlBase() + self.urlFonction()
|
|
headers = {"Content-Type": "application/json"}
|
|
|
|
try:
|
|
# Encoder l'image en base64
|
|
with open(image_path, "rb") as image_file:
|
|
image_b64 = base64.b64encode(image_file.read()).decode("utf-8")
|
|
|
|
# Formater le prompt avec l'image
|
|
prompt = f"""<image>
|
|
{image_b64}
|
|
</image>
|
|
|
|
{question}"""
|
|
|
|
contenu = {
|
|
"model": self.modele,
|
|
"prompt": prompt,
|
|
"options": {
|
|
"temperature": self.params["temperature"],
|
|
"top_p": self.params["top_p"],
|
|
"top_k": self.params["top_k"],
|
|
"num_ctx": self.params["num_ctx"],
|
|
"repeat_penalty": self.params["repeat_penalty"],
|
|
"repeat_last_n": self.params["repeat_last_n"],
|
|
"mirostat": self.params["mirostat"],
|
|
"mirostat_eta": self.params["mirostat_eta"],
|
|
"mirostat_tau": self.params["mirostat_tau"],
|
|
"keep_alive": self.params["keep_alive"],
|
|
"num_predict": self.params["num_predict"],
|
|
"min_p": self.params["min_p"],
|
|
"seed": self.params["seed"],
|
|
"stop": self.params["stop"],
|
|
},
|
|
"stream": self.params["stream"]
|
|
}
|
|
|
|
self.heureDepart = datetime.now()
|
|
response = requests.post(url=url, headers=headers, json=contenu, timeout=self.request_timeout)
|
|
self.heureFin = datetime.now()
|
|
|
|
if self.heureDepart is not None:
|
|
self.dureeTraitement = self.heureFin - self.heureDepart
|
|
else:
|
|
self.dureeTraitement = timedelta(0)
|
|
|
|
if response.status_code in [200, 201]:
|
|
self.reponseErreur = False
|
|
text_response = self._traiter_reponse(response)
|
|
|
|
# Vérifier si la réponse indique une incapacité à traiter l'image
|
|
if any(phrase in text_response.lower() for phrase in [
|
|
"je ne peux pas voir l'image",
|
|
"je n'ai pas accès à l'image",
|
|
"impossible de visualiser"
|
|
]):
|
|
self.reponseErreur = True
|
|
error_msg = "Le modèle n'a pas pu analyser l'image correctement."
|
|
self._enregistrer_interaction(f"[ANALYSE IMAGE] {question}", error_msg, True)
|
|
return error_msg
|
|
|
|
self._enregistrer_interaction(f"[ANALYSE IMAGE] {question}", text_response)
|
|
return text_response
|
|
else:
|
|
self.reponseErreur = True
|
|
error_msg = f"Erreur API ({response.status_code}): {response.text}"
|
|
self._enregistrer_interaction(f"[ANALYSE IMAGE] {question}", error_msg, True)
|
|
return error_msg
|
|
|
|
except requests.exceptions.Timeout:
|
|
self.heureFin = datetime.now()
|
|
if self.heureDepart is not None:
|
|
self.dureeTraitement = self.heureFin - self.heureDepart
|
|
self.reponseErreur = True
|
|
error_msg = "Timeout lors de l'analyse de l'image. La requête a pris trop de temps."
|
|
self._enregistrer_interaction(f"[ANALYSE IMAGE] {question}", error_msg, True)
|
|
return error_msg
|
|
|
|
except Exception as e:
|
|
self.heureFin = datetime.now()
|
|
if self.heureDepart is not None:
|
|
self.dureeTraitement = self.heureFin - self.heureDepart
|
|
else:
|
|
self.dureeTraitement = timedelta(0)
|
|
self.reponseErreur = True
|
|
error_msg = f"Erreur lors de l'analyse de l'image: {str(e)}"
|
|
self._enregistrer_interaction(f"[ANALYSE IMAGE] {question}", error_msg, True)
|
|
return error_msg
|
|
|
|
def configurer(self, **kwargs):
|
|
"""
|
|
Configure les paramètres du modèle Pixtral 12B.
|
|
|
|
Args:
|
|
**kwargs: Paramètres à configurer (temperature, top_p, etc.)
|
|
"""
|
|
# Appliquer les paramètres
|
|
for key, value in kwargs.items():
|
|
if key in self.params:
|
|
self.params[key] = value
|
|
elif key == "prompt_system" and isinstance(value, str):
|
|
self.prompt_system = value
|
|
elif key == "request_timeout" and isinstance(value, int):
|
|
self.request_timeout = value
|
|
|
|
return self
|
|
|
|
def _enregistrer_interaction(self, question: str, reponse: str, erreur: bool = False):
|
|
"""
|
|
Enregistre une interaction pour suivi et débogage.
|
|
|
|
Args:
|
|
question: Question posée
|
|
reponse: Réponse reçue
|
|
erreur: Indique si l'interaction a généré une erreur
|
|
"""
|
|
interaction = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"question": question,
|
|
"reponse": reponse,
|
|
"duree": self.dureeTraitement.total_seconds() if self.dureeTraitement else 0,
|
|
"erreur": erreur,
|
|
"modele": self.modele,
|
|
"parametres": {
|
|
"temperature": self.params["temperature"],
|
|
"top_p": self.params["top_p"],
|
|
"top_k": self.params["top_k"]
|
|
}
|
|
}
|
|
|
|
self.interactions_historique.append(interaction)
|
|
|
|
# Limiter la taille de l'historique
|
|
if len(self.interactions_historique) > 100:
|
|
self.interactions_historique = self.interactions_historique[-100:]
|
|
|
|
def obtenir_historique(self):
|
|
"""
|
|
Retourne l'historique des interactions récentes.
|
|
|
|
Returns:
|
|
Liste des interactions enregistrées
|
|
"""
|
|
return self.interactions_historique
|
|
|
|
def exporter_historique(self, chemin_fichier: str = "") -> str:
|
|
"""
|
|
Exporte l'historique des interactions vers un fichier JSON.
|
|
|
|
Args:
|
|
chemin_fichier: Chemin du fichier où exporter. Si vide, un nom basé sur la date est généré.
|
|
|
|
Returns:
|
|
Chemin du fichier où l'historique a été exporté ou chaîne vide en cas d'erreur
|
|
"""
|
|
if not chemin_fichier:
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
chemin_fichier = f"historique_pixtral_{timestamp}.json"
|
|
|
|
try:
|
|
with open(chemin_fichier, 'w', encoding='utf-8') as f:
|
|
json.dump(self.interactions_historique, f, ensure_ascii=False, indent=2)
|
|
return chemin_fichier
|
|
except Exception as e:
|
|
print(f"Erreur lors de l'export de l'historique: {str(e)}")
|
|
return ""
|