llm_ticket3/llm_classes/qwen2_5.py

from .base_llm import BaseLLM
import requests
from datetime import datetime, timedelta
from typing import Dict, Any
import os
import json

class Qwen2_5(BaseLLM):
    """
    Classe complète pour interagir avec le modèle Qwen 2.5 via Ollama.
    Optimisée pour les fonctionnalités spécifiques de Qwen 2.5.
    """
    
    def __init__(self):
        """
        Initialise une instance du modèle Qwen 2.5 avec des paramètres optimisés.
        """
        # Initialiser avec le modèle Qwen 2.5
        super().__init__("qwen2.5:72b-instruct-q8_0")
        
        # Définir les attributs spécifiques
        self.modele = "Qwen 2.5"
        self.version = "72B"
        self.api_url = "http://217.182.105.173:11434/api/generate"
        
        # Paramètres optimisés spécifiquement pour Qwen 2.5
        self.params: Dict[str, Any] = {
            "temperature": 0.3,          # Équilibre entre créativité et précision
            "top_p": 0.8,                # Diversité modérée des réponses
            "top_k": 40,                 # Choix des tokens les plus probables
            "num_ctx": 4096,             # Contexte étendu pour de meilleures analyses
            "repeat_penalty": 1.2,       # Pénalité plus forte pour éviter les répétitions
            "repeat_last_n": 128,        # Considère plus de tokens pour la pénalité de répétition
            "mirostat": 0,               # Désactivé car moins efficace avec Qwen
            "mirostat_eta": 0.1,
            "mirostat_tau": 5,
            "keep_alive": int(timedelta(minutes=10).total_seconds()),  # Maintien prolongé pour les analyses complexes
            "num_predict": 4000,         # Prédiction plus longue pour des réponses détaillées
            "min_p": 0.05,               # Légèrement augmenté pour plus de diversité
            "seed": 0,
            "stop": ["</answer>", "###", "\n\n\n"],  # Tokens d'arrêt adaptés à Qwen
            "stream": False
        }
        
        # Timeout de requête adapté au modèle
        self.request_timeout = 360  # 6 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 pour optimiser les réponses
        self.prompt_system = "Tu es un assistant IA expert et précis. Fournis des réponses complètes mais concises."
    
    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 Qwen 2.5.
        
        Args:
            question: La question ou instruction à envoyer au modèle
            
        Returns:
            Dictionnaire formaté pour l'API Ollama avec Qwen 2.5
        """
        # Optimiser le prompt avec le format spécifique pour Qwen
        prompt_optimise = self._optimiser_prompt_pour_qwen(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_pour_qwen(self, question: str) -> str:
        """
        Optimise le format du prompt spécifiquement pour Qwen 2.5.
        
        Args:
            question: La question ou instruction originale
            
        Returns:
            Prompt optimisé pour de meilleures performances avec Qwen 2.5
        """
        # Vérifier si la question inclut déjà un format de prompt
        if "<system>" in question or "<human>" in question or "<answer>" in question:
            return question
        
        # Formater avec le format spécifique à Qwen pour de meilleures performances
        formatted_prompt = f"""<system>
{self.prompt_system}
</system>

<human>
{question}
</human>

<answer>
"""
        return formatted_prompt
    
    def _traiter_reponse(self, reponse: requests.Response) -> str:
        """
        Traite et nettoie la réponse fournie par Qwen via Ollama.
        
        Args:
            reponse: Réponse HTTP de l'API
            
        Returns:
            Texte nettoyé de la réponse
        """
        try:
            data = reponse.json()
            response_text = data.get("response", "")
            
            # Nettoyer la réponse des tags spécifiques à Qwen si présents
            response_text = response_text.replace("</answer>", "").strip()
            
            # Retirer les parties répétitives potentielles à la fin
            if "human>" in response_text.lower():
                response_text = response_text.split("<human>")[0].strip()
            
            return response_text
        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 Qwen 2.5 en utilisant Ollama avec des paramètres optimisés.
        
        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:
        """
        Qwen via Ollama ne supporte pas nativement l'analyse d'images.
        Cette méthode renvoie un message d'erreur.
        
        Args:
            image_path: Chemin vers l'image (non utilisé)
            question: Question concernant l'image
            
        Returns:
            Message d'erreur
        """
        self.reponseErreur = True
        message = f"Le modèle Qwen 2.5 ne supporte pas l'analyse d'images. Question: {question}"
        self._enregistrer_interaction(f"[ANALYSE IMAGE] {question}", message, True)
        return message
        
    def configurer(self, **kwargs):
        """
        Configure les paramètres spécifiques à Qwen.
        
        Args:
            **kwargs: Paramètres à configurer (temperature, top_p, etc.)
        """
        # Appliquer les paramètres au dictionnaire
        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_qwen_{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 ""