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": ["", "###", "\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_b64} {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 ""