import json import os from typing import Dict, List, Any, Optional, Union, Tuple import time from datetime import datetime from .llama_vision import LlamaVision from .mistral import Mistral from .ollama import Ollama class Agent: """ Classe de base pour tous les agents """ def __init__(self, nom: str = "Agent"): self.nom: str = nom self.historique: List[Dict[str, Any]] = [] def ajouter_historique(self, action: str, input_data: Any, output_data: Any) -> None: """ Ajoute une entrée dans l'historique de l'agent """ self.historique.append({ "timestamp": datetime.now().isoformat(), "action": action, "input": str(input_data)[:500], # Limite pour éviter des historiques trop grands "output": str(output_data)[:500] # Limite pour éviter des historiques trop grands }) def obtenir_historique(self) -> List[Dict[str, Any]]: """ Retourne l'historique complet de l'agent """ return self.historique def executer(self, *args, **kwargs) -> Any: """ Méthode abstraite à implémenter dans les classes dérivées """ raise NotImplementedError("Chaque agent doit implémenter sa propre méthode executer()") class AgentAnalyseImage(Agent): """ Agent pour analyser des images avec Llama Vision """ def __init__(self, nom: str = "AgentAnalyseImage"): super().__init__(nom) self.llm = LlamaVision() self.questions_standard = [ "Décris en détail ce que tu vois sur cette image.", "Quels sont les éléments principaux visibles sur cette image?", "Y a-t-il du texte visible sur cette image? Si oui, peux-tu le transcrire?", "Quelle est l'ambiance générale de cette image?" ] def executer(self, image_path: str, json_data: Optional[Dict[str, Any]] = None, question: Optional[str] = None) -> Dict[str, Any]: """ Analyse une image avec Llama Vision Args: image_path: Chemin vers l'image à analyser json_data: Données JSON optionnelles associées à l'image question: Question à poser au modèle (si None, utilise les questions standard) Returns: Résultats de l'analyse sous forme de dictionnaire """ # Vérification de l'existence de l'image if not os.path.exists(image_path): resultat = {"erreur": f"L'image {image_path} n'existe pas"} self.ajouter_historique("analyse_image_erreur", image_path, resultat) return resultat # Chargement de l'image if not self.llm.set_image(image_path): resultat = {"erreur": f"Échec du chargement de l'image {image_path}"} self.ajouter_historique("analyse_image_erreur", image_path, resultat) return resultat # Ajout des données JSON si fournies if json_data: self.llm.set_json_data(json_data) # Exécution de l'analyse resultats = {} # Utilisation des questions standard si aucune question n'est fournie questions_a_poser = [question] if question else self.questions_standard for idx, q in enumerate(questions_a_poser): print(f"Question {idx+1}/{len(questions_a_poser)}: {q[:50]}...") reponse = self.llm.Interroger(q) resultats[f"question_{idx+1}"] = { "question": q, "reponse": reponse } time.sleep(1) # Pause pour éviter de surcharger l'API # Fusion avec le JSON si présent resultats_complets = self.llm.fusionner_json_avec_resultats() resultats_complets["analyses"] = resultats # Ajout à l'historique self.ajouter_historique("analyse_image", image_path, "Analyse complétée avec succès") return resultats_complets def ajuster_parametres(self, temperature: Optional[float] = None, top_p: Optional[float] = None, num_ctx: Optional[int] = None) -> None: """ Ajuste les paramètres du modèle Llama Vision """ if temperature is not None: self.llm.o_temperature = temperature if top_p is not None: self.llm.o_top_p = top_p if num_ctx is not None: self.llm.o_num_ctx = num_ctx self.ajouter_historique("ajuster_parametres", {"temperature": temperature, "top_p": top_p, "num_ctx": num_ctx}, "Paramètres ajustés") def sauvegarder_resultats(self, chemin_fichier: str, resultats: Dict[str, Any]) -> bool: """ Sauvegarde les résultats d'analyse dans un fichier JSON """ try: with open(chemin_fichier, 'w', encoding='utf-8') as f: json.dump(resultats, f, ensure_ascii=False, indent=2) self.ajouter_historique("sauvegarder_resultats", chemin_fichier, "Résultats sauvegardés") return True except Exception as e: self.ajouter_historique("sauvegarder_resultats_erreur", chemin_fichier, str(e)) return False class AgentAnalyseJSON(Agent): """ Agent pour analyser des données JSON """ def __init__(self, nom: str = "AgentAnalyseJSON", modele: str = "mistral"): super().__init__(nom) # Choix du modèle if modele.lower() == "mistral": self.llm = Mistral() self.llm.prompt_system = "Tu es un expert en analyse de données JSON. Tu dois extraire des informations pertinentes, identifier des tendances et répondre à des questions sur les données." else: self.llm = Ollama() self.llm.prompt_system = "Tu es un expert en analyse de données JSON. Tu dois extraire des informations pertinentes, identifier des tendances et répondre à des questions sur les données." self.llm.Modele = modele def executer(self, json_data: Dict[str, Any], question: str = "Analyse ces données et extrait les informations principales.") -> Dict[str, Any]: """ Analyse des données JSON Args: json_data: Données JSON à analyser question: Question à poser au modèle Returns: Résultats de l'analyse sous forme de dictionnaire """ # Conversion du JSON en chaîne formatée json_str = json.dumps(json_data, ensure_ascii=False, indent=2) # Construction du prompt avec le JSON et la question prompt = f"{question}\n\nDonnées JSON à analyser:\n```json\n{json_str}\n```" # Interrogation du modèle reponse = self.llm.Interroger(prompt) # Construction du résultat resultats = { "question": question, "reponse": reponse, "timestamp": datetime.now().isoformat(), "taille_json": len(json_str) } # Ajout à l'historique self.ajouter_historique("analyse_json", prompt[:200] + "...", reponse[:200] + "...") return resultats def extraire_structure(self, json_data: Dict[str, Any]) -> Dict[str, Any]: """ Extrait la structure d'un JSON (clés, types, profondeur) """ resultat = { "structure": {}, "statistiques": { "nb_cles": 0, "profondeur_max": 0, "types": {} } } def explorer_structure(data, chemin="", profondeur=0): nonlocal resultat # Mise à jour de la profondeur max resultat["statistiques"]["profondeur_max"] = max(resultat["statistiques"]["profondeur_max"], profondeur) if isinstance(data, dict): structure = {} for cle, valeur in data.items(): nouveau_chemin = f"{chemin}.{cle}" if chemin else cle resultat["statistiques"]["nb_cles"] += 1 type_valeur = type(valeur).__name__ if type_valeur not in resultat["statistiques"]["types"]: resultat["statistiques"]["types"][type_valeur] = 0 resultat["statistiques"]["types"][type_valeur] += 1 if isinstance(valeur, (dict, list)): structure[cle] = explorer_structure(valeur, nouveau_chemin, profondeur + 1) else: structure[cle] = type_valeur return structure elif isinstance(data, list): if data and isinstance(data[0], (dict, list)): # Pour les listes de structures complexes, on analyse le premier élément return [explorer_structure(data[0], f"{chemin}[0]", profondeur + 1)] else: # Pour les listes de valeurs simples type_elements = "vide" if not data else type(data[0]).__name__ resultat["statistiques"]["nb_cles"] += 1 if "list" not in resultat["statistiques"]["types"]: resultat["statistiques"]["types"]["list"] = 0 resultat["statistiques"]["types"]["list"] += 1 return f"list[{type_elements}]" else: return type(data).__name__ resultat["structure"] = explorer_structure(json_data) self.ajouter_historique("extraire_structure", "JSON", resultat) return resultat def fusionner_jsons(self, json1: Dict[str, Any], json2: Dict[str, Any]) -> Dict[str, Any]: """ Fusionne deux structures JSON en conservant les informations des deux """ if not json1: return json2 if not json2: return json1 resultat = json1.copy() # Fonction récursive pour fusionner def fusionner(dict1, dict2): for cle, valeur in dict2.items(): if cle in dict1: # Si les deux sont des dictionnaires, fusion récursive if isinstance(dict1[cle], dict) and isinstance(valeur, dict): fusionner(dict1[cle], valeur) # Si les deux sont des listes, concaténation elif isinstance(dict1[cle], list) and isinstance(valeur, list): dict1[cle].extend(valeur) # Sinon, on garde les deux valeurs dans une liste else: if not isinstance(dict1[cle], list): dict1[cle] = [dict1[cle]] if isinstance(valeur, list): dict1[cle].extend(valeur) else: dict1[cle].append(valeur) else: # Si la clé n'existe pas dans dict1, on l'ajoute simplement dict1[cle] = valeur fusionner(resultat, json2) self.ajouter_historique("fusionner_jsons", "Fusion de deux JSON", "Fusion réussie") return resultat class AgentQuestionReponse(Agent): """ Agent pour tester des questions/réponses sur des données JSON """ def __init__(self, nom: str = "AgentQuestionReponse", modele: str = "mistral"): super().__init__(nom) # Choix du modèle if modele.lower() == "mistral": self.llm = Mistral() else: self.llm = Ollama() self.llm.Modele = modele # Configuration du modèle self.llm.prompt_system = "Tu es un assistant expert qui répond à des questions en te basant uniquement sur les données JSON fournies. Ne fais pas de suppositions au-delà des données." # Questions prédéfinies self.questions_standard = [ "Quel est le sujet principal de cette image d'après les données?", "Quels sont les éléments clés identifiés dans l'image?", "Y a-t-il des incohérences entre les différentes analyses?", "Résume les informations principales contenues dans ce JSON." ] def executer(self, json_data: Dict[str, Any], questions: Optional[List[str]] = None) -> Dict[str, List[Dict[str, str]]]: """ Exécute une série de questions/réponses sur des données JSON Args: json_data: Données JSON à analyser questions: Liste de questions à poser (si None, utilise les questions standard) Returns: Résultats des Q/R sous forme de dictionnaire """ questions_a_poser = questions if questions else self.questions_standard resultats = [] # Conversion du JSON en chaîne formatée json_str = json.dumps(json_data, ensure_ascii=False, indent=2) for question in questions_a_poser: # Construction du prompt prompt = f"Question: {question}\n\nVoici les données JSON sur lesquelles te baser pour répondre:\n```json\n{json_str}\n```\n\nRéponds de manière claire et concise en te basant uniquement sur ces données." # Interrogation du modèle reponse = self.llm.Interroger(prompt) resultats.append({ "question": question, "reponse": reponse }) # Ajout à l'historique self.ajouter_historique("question_reponse", question, reponse[:200] + "...") # Pause pour éviter de surcharger l'API time.sleep(1) return {"resultats_qr": resultats} def generer_questions(self, json_data: Dict[str, Any], nb_questions: int = 3) -> List[str]: """ Génère automatiquement des questions pertinentes basées sur les données JSON Args: json_data: Données JSON à analyser nb_questions: Nombre de questions à générer Returns: Liste de questions générées """ # Conversion du JSON en chaîne formatée json_str = json.dumps(json_data, ensure_ascii=False, indent=2) prompt = f"Voici des données JSON contenant des analyses d'images:\n```json\n{json_str}\n```\n\nGénère {nb_questions} questions pertinentes et spécifiques que l'on pourrait poser à propos de ces données. Donne uniquement la liste des questions, une par ligne." reponse = self.llm.Interroger(prompt) # Extraction des questions (chaque ligne est une question) questions = [ligne.strip() for ligne in reponse.split('\n') if ligne.strip()] # Filtrage des lignes qui ne sont pas des questions questions = [q for q in questions if '?' in q][:nb_questions] self.ajouter_historique("generer_questions", f"Génération de {nb_questions} questions", questions) return questions def evaluer_reponses(self, resultats_qr: List[Dict[str, str]]) -> Dict[str, Any]: """ Évalue la qualité des réponses générées Args: resultats_qr: Liste de dictionnaires avec des paires question/réponse Returns: Évaluation des réponses """ evaluation = { "meta": { "nb_questions": len(resultats_qr), "timestamp": datetime.now().isoformat() }, "evaluations": [] } for idx, qr in enumerate(resultats_qr): question = qr["question"] reponse = qr["reponse"] # Construction du prompt d'évaluation prompt = f"Évalue la qualité de cette réponse à la question suivante. Attribue une note de 1 à 10 et explique pourquoi.\n\nQuestion: {question}\n\nRéponse: {reponse}\n\nFormat de réponse attendu: Note: [1-10]\nJustification: [explication]" eval_reponse = self.llm.Interroger(prompt) # Extraction de la note note = 0 for ligne in eval_reponse.split('\n'): if ligne.lower().startswith('note:'): try: # Extraction de la partie numérique note_str = ''.join(c for c in ligne if c.isdigit() or c == '.') note = float(note_str) break except: note = 0 evaluation["evaluations"].append({ "index": idx, "question": question, "note": note, "evaluation": eval_reponse }) time.sleep(1) # Pause pour éviter de surcharger l'API # Calcul de la note moyenne notes = [e["note"] for e in evaluation["evaluations"]] evaluation["meta"]["note_moyenne"] = sum(notes) / len(notes) if notes else 0 self.ajouter_historique("evaluer_reponses", "Évaluation des réponses", evaluation["meta"]) return evaluation