AIagent/llm_classes/example_vision.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Script d'exemple pour tester LlamaVision et les agents d'analyse d'images
"""

import os
import json
import argparse
from datetime import datetime
from llm_classes import LlamaVision, AgentAnalyseImage, AgentAnalyseJSON, AgentQuestionReponse

def create_output_dir():
    """Crée un répertoire pour les sorties basé sur la date/heure"""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    output_dir = f"output_vision_{timestamp}"
    os.makedirs(output_dir, exist_ok=True)
    return output_dir

def analyse_directe(image_path, output_dir):
    """Analyse directe avec LlamaVision sans passer par un agent"""
    print(f"\n=== Analyse directe de l'image {os.path.basename(image_path)} ===")
    
    # Création d'une instance LlamaVision
    vision = LlamaVision()
    vision.prompt_system = "Tu es un assistant d'analyse d'images très précis. Décris l'image en détail."
    
    # Chargement de l'image
    if not vision.set_image(image_path):
        print(f"Erreur: Impossible de charger l'image {image_path}")
        return
    
    # Analyse de l'image
    question = "Décris en détail cette image. Qu'est-ce que tu y vois? Y a-t-il du texte visible?"
    print(f"Question: {question}")
    
    reponse = vision.Interroger(question)
    print("\nRéponse:")
    print(reponse)
    
    # Sauvegarde du résultat
    output_file = os.path.join(output_dir, "resultat_direct.json")
    vision.sauvegarder_resultats(output_file)
    print(f"\nRésultat sauvegardé dans: {output_file}")

def workflow_complet(image_path, json_path=None, output_dir=None):
    """
    Workflow complet d'analyse d'image et traitement des données
    
    Args:
        image_path: Chemin vers l'image à analyser
        json_path: Chemin vers un fichier JSON optionnel contenant des données additionnelles
        output_dir: Répertoire de sortie (si None, en crée un nouveau)
    """
    if output_dir is None:
        output_dir = create_output_dir()
    
    print(f"\n=== Workflow complet pour l'image {os.path.basename(image_path)} ===")
    print(f"Résultats sauvegardés dans: {output_dir}")
    
    # Étape 1: Chargement des données JSON existantes si disponibles
    json_data = None
    if json_path and os.path.exists(json_path):
        try:
            with open(json_path, 'r', encoding='utf-8') as f:
                json_data = json.load(f)
            print(f"Données JSON chargées depuis: {json_path}")
        except Exception as e:
            print(f"Erreur lors du chargement du JSON: {e}")
    
    # Étape 2: Analyse de l'image avec l'agent
    print("\n--- Étape 1: Analyse de l'image ---")
    agent_image = AgentAnalyseImage("Agent Vision")
    
    # Configuration des paramètres
    agent_image.ajuster_parametres(temperature=0.7, top_p=0.9, num_ctx=4096)
    
    # Analyse de l'image
    resultats_analyse = agent_image.executer(image_path, json_data)
    
    # Sauvegarde des résultats intermédiaires
    output_image = os.path.join(output_dir, "analyse_image.json")
    agent_image.sauvegarder_resultats(output_image, resultats_analyse)
    print(f"Analyse d'image sauvegardée dans: {output_image}")
    
    # Étape 3: Analyse des données JSON avec l'agent
    print("\n--- Étape 2: Analyse des données JSON ---")
    agent_json = AgentAnalyseJSON("Agent JSON")
    
    # Extraction de la structure
    structure = agent_json.extraire_structure(resultats_analyse)
    
    # Analyse des données
    question_analyse = "Analyse ces données extraites de l'image et résume les points principaux. Identifie les éléments clés et leur importance."
    resultats_json = agent_json.executer(resultats_analyse, question_analyse)
    
    # Fusion des résultats
    resultats_complets = agent_json.fusionner_jsons(
        resultats_analyse,
        {
            "meta_analyse": {
                "timestamp": datetime.now().isoformat(),
                "structure": structure,
                "analyse_texte": resultats_json
            }
        }
    )
    
    # Sauvegarde des résultats intermédiaires
    output_json = os.path.join(output_dir, "analyse_complete.json")
    with open(output_json, 'w', encoding='utf-8') as f:
        json.dump(resultats_complets, f, ensure_ascii=False, indent=2)
    print(f"Analyse complète sauvegardée dans: {output_json}")
    
    # Étape 4: Questions/Réponses sur les données
    print("\n--- Étape 3: Tests de questions/réponses ---")
    agent_qr = AgentQuestionReponse("Agent QR")
    
    # Génération automatique de questions
    questions_auto = agent_qr.generer_questions(resultats_complets, nb_questions=3)
    print("\nQuestions générées automatiquement:")
    for i, q in enumerate(questions_auto):
        print(f"{i+1}. {q}")
    
    # Questions standard + questions générées
    toutes_questions = agent_qr.questions_standard + questions_auto
    
    # Exécution des questions/réponses
    resultats_qr = agent_qr.executer(resultats_complets, toutes_questions)
    
    # Évaluation des réponses
    evaluation = agent_qr.evaluer_reponses(resultats_qr["resultats_qr"])
    
    # Fusion et sauvegarde des résultats finaux
    resultats_finaux = {
        "image": os.path.basename(image_path),
        "date_analyse": datetime.now().isoformat(),
        "donnees": resultats_complets,
        "questions_reponses": resultats_qr,
        "evaluation": evaluation
    }
    
    output_final = os.path.join(output_dir, "resultats_finaux.json")
    with open(output_final, 'w', encoding='utf-8') as f:
        json.dump(resultats_finaux, f, ensure_ascii=False, indent=2)
    print(f"\nRésultats finaux sauvegardés dans: {output_final}")
    
    # Affichage des résultats
    print("\n=== Résumé des résultats ===")
    print(f"Note moyenne des réponses: {evaluation['meta']['note_moyenne']:.1f}/10")
    
    # Affichage d'un exemple de question/réponse
    if resultats_qr["resultats_qr"]:
        exemple = resultats_qr["resultats_qr"][0]
        print(f"\nExemple de Q/R:")
        print(f"Q: {exemple['question']}")
        print(f"R: {exemple['reponse'][:200]}...")
    
    return resultats_finaux

def main():
    """Fonction principale"""
    parser = argparse.ArgumentParser(description="Test de LlamaVision et des agents d'analyse d'images")
    parser.add_argument("--image", type=str, help="Chemin vers l'image à analyser", required=True)
    parser.add_argument("--json", type=str, help="Chemin vers un fichier JSON optionnel avec des données additionnelles")
    parser.add_argument("--mode", type=str, choices=["direct", "complet"], default="complet", 
                        help="Mode d'analyse: direct (LlamaVision uniquement) ou complet (workflow avec agents)")
    
    args = parser.parse_args()
    
    # Vérification de l'existence de l'image
    if not os.path.exists(args.image):
        print(f"Erreur: L'image {args.image} n'existe pas")
        return
    
    # Création du répertoire de sortie
    output_dir = create_output_dir()
    
    # Mode d'analyse
    if args.mode == "direct":
        analyse_directe(args.image, output_dir)
    else:
        workflow_complet(args.image, args.json, output_dir)
    
    print("\nAnalyse terminée!")

if __name__ == "__main__":
    main()