mirror of
https://github.com/Ladebeze66/AIagent.git
synced 2025-12-16 00:36:51 +01:00
436 lines
17 KiB
Python
436 lines
17 KiB
Python
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 |