from .base_llm import BaseLLM import requests import base64 import os from PIL import Image import io from datetime import datetime, timedelta from typing import Dict, Any class LlamaVision(BaseLLM): """ Classe optimisée pour interagir avec l'API Llama Vision. """ def __init__(self, modele: str = "llama3.2-vision:90b-instruct-q8_0"): super().__init__(modele) self.api_url = "http://217.182.105.173:11434/api/generate" # Timeout de requête augmenté pour les images volumineuses self.request_timeout = 300 # 5 minutes # Configuration standard via la méthode héritée self.configurer( temperature=0.1, top_p=0.8, max_tokens=1024, ) 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 par défaut. """ 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 spécifique pour Ollama avec le modèle Llama Vision. """ contenu = { "model": self.modele, "prompt": question, "options": { "temperature": self.params["temperature"], "top_p": self.params["top_p"], "num_predict": self.params.get("max_tokens", 1024), "stop": self.params.get("stop", []), # Paramètres Ollama spécifiques avec valeurs par défaut "top_k": 30, "num_ctx": 1024, "repeat_penalty": 1.1, "repeat_last_n": 64, "mirostat": 0, "mirostat_eta": 0.1, "mirostat_tau": 5, "keep_alive": int(timedelta(minutes=2).total_seconds()), "min_p": 0, "seed": 0, }, "stream": False } return contenu def _traiter_reponse(self, reponse: requests.Response) -> str: """ Traite et retourne la réponse fournie par Ollama. """ data = reponse.json() return data.get("response", "") def _encoder_image_base64(self, image_path: str) -> str: """ Encode une image en base64, avec optimisation de la taille si nécessaire. """ try: # Vérifier la taille de l'image et la réduire si trop grande with Image.open(image_path) as img: # Si l'image est trop grande, la redimensionner max_dim = 800 # Dimension maximale width, height = img.size if width > max_dim or height > max_dim: # Calculer le ratio pour conserver les proportions ratio = min(max_dim / width, max_dim / height) new_width = int(width * ratio) new_height = int(height * ratio) # Redimensionner l'image img = img.resize((new_width, new_height), Image.Resampling.LANCZOS) # Convertir en RGB si nécessaire (pour les formats comme PNG avec canal alpha) if img.mode in ("RGBA", "LA", "P"): # Créer un fond blanc et composer l'image dessus pour gérer la transparence background = Image.new("RGB", img.size, (255, 255, 255)) if img.mode == "P": img = img.convert("RGBA") background.paste(img, mask=img.split()[3] if img.mode == "RGBA" else None) img = background elif img.mode != "RGB": img = img.convert("RGB") # Sauvegarder temporairement l'image redimensionnée buffer = io.BytesIO() img.save(buffer, format="JPEG", quality=85) buffer.seek(0) # Encoder en base64 encoded = base64.b64encode(buffer.read()).decode("utf-8") return encoded except Exception as e: print(f"Erreur lors de l'optimisation de l'image: {str(e)}") try: # Seconde tentative avec une approche plus simple with Image.open(image_path) as img: # Convertir directement en RGB quelle que soit l'image img = img.convert("RGB") buffer = io.BytesIO() img.save(buffer, format="JPEG", quality=75) buffer.seek(0) encoded = base64.b64encode(buffer.read()).decode("utf-8") return encoded except Exception as e2: print(f"Deuxième erreur lors de l'optimisation de l'image: {str(e2)}") # Dernier recours: encoder l'image originale sans optimisation with open(image_path, "rb") as image_file: encoded = base64.b64encode(image_file.read()).decode("utf-8") return encoded def _optimiser_prompt(self, question: str) -> str: """ Optimise le prompt pour des réponses plus rapides. Ne modifie pas la langue ou le contenu de la question. """ # On retourne la question telle quelle sans imposer de format return question def interroger_avec_image(self, image_path: str, question: str) -> str: """ Interroge le modèle Llama Vision avec une image et une question via Ollama. Optimisé pour éviter les timeouts. Args: image_path: Chemin vers l'image à analyser question: Question ou instructions pour l'analyse Returns: Réponse du modèle à la question """ url = self.urlBase() + self.urlFonction() headers = {"Content-Type": "application/json"} try: # Encoder l'image en base64 avec optimisation de taille image_b64 = self._encoder_image_base64(image_path) # Optimiser la question pour des réponses plus courtes et plus rapides optimised_question = self._optimiser_prompt(question) # Préparer le prompt avec le format spécial pour Ollama multimodal prompt = f""" {image_b64} {optimised_question}""" contenu = { "model": self.modele, "prompt": prompt, "options": { "temperature": self.params["temperature"], "top_p": self.params["top_p"], "num_predict": self.params.get("max_tokens", 1024), "stop": self.params.get("stop", []), # Paramètres Ollama spécifiques "top_k": 30, "num_ctx": 1024, "repeat_penalty": 1.1, "repeat_last_n": 64, "mirostat": 0, "mirostat_eta": 0.1, "mirostat_tau": 5, "keep_alive": int(timedelta(minutes=2).total_seconds()), "min_p": 0, "seed": 0, }, "stream": False } 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 return self._traiter_reponse(response) else: self.reponseErreur = True return f"Erreur API ({response.status_code}): {response.text}" except requests.exceptions.Timeout: self.heureFin = datetime.now() if self.heureDepart is not None: self.dureeTraitement = self.heureFin - self.heureDepart self.reponseErreur = True return "Timeout lors de l'appel à l'API. L'analyse de l'image a pris trop de temps." 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 return f"Erreur lors de l'analyse de l'image: {str(e)}" def configurer(self, **kwargs): """ Mise à jour des paramètres du modèle. Utilise la méthode de BaseLLM pour mettre à jour les paramètres. """ # Si request_timeout est spécifié, le mettre à jour séparément if "request_timeout" in kwargs and isinstance(kwargs["request_timeout"], int): self.request_timeout = kwargs.pop("request_timeout") # Utiliser la méthode de la classe parente pour mettre à jour les paramètres standards super().configurer(**kwargs)