# model_handler.py (Continuation du fichier) import sys from google import genai from google.genai.errors import APIError from google.genai.types import Part # Importation de Part est déjà là, c'est bien. import os from typing import Generator, List, Dict, Union # Ajout d'Union from config import GEMINI_MODEL_NAME, DEFAULT_MAX_TOKENS, DEFAULT_TEMPERATURE, SYSTEM_PROMPT import base64 # NOUVEAU: Pour décoder les fichiers base64 # --- Configuration du Client Gemini --- client = None MODEL_LOADED_SUCCESSFULLY = False def run_api_self_test(client_instance): """Effectue un test simple pour vérifier si l'API fonctionne.""" try: print("INFO: Exécution du test de connexion à l'API Gemini...", file=sys.stderr) # Test minimal: demandons simplement une salutation test_content = ["Salutation brève"] response = client_instance.models.generate_content( model=GEMINI_MODEL_NAME, contents=test_content, config=genai.types.GenerateContentConfig(max_output_tokens=10) # Limiter la réponse ) # CHANGEMENT : Vérification simplifiée pour s'assurer qu'un objet 'response' a été reçu. if response is not None: response_text_info = response.text[:30] if response.text else "[Réponse sans texte]" print("INFO: ✅ Test de connexion API réussi. Le modèle est joignable.", file=sys.stderr) print(f"INFO: Réponse du test (début): {response_text_info}...", file=sys.stderr) return True else: print("ALERTE: ❌ Test de connexion API échoué. Objet de réponse None.", file=sys.stderr) return False except APIError as e: # Ceci gère les problèmes de clé invalide, quota, etc. print(f"ALERTE: ❌ Test de connexion API échoué (APIError). Vérifiez la validité de la clé: {e.message}", file=sys.stderr) return False except Exception as e: # Erreur générale (connexion, SDK, etc.) print(f"ALERTE: ❌ Test de connexion API échoué (Erreur Générale): {e}", file=sys.stderr) return False def get_model_instance(): """Initialise ou retourne l'instance du client Gemini.""" global client, MODEL_LOADED_SUCCESSFULLY if client is None: print("INFO: Tentative d'initialisation du client Gemini...", file=sys.stderr) api_key = os.environ.get("GEMINI_API_KEY") if not api_key: print("ALERTE: ❌ Variable d'environnement 'GEMINI_API_KEY' non trouvée/vide.", file=sys.stderr) MODEL_LOADED_SUCCESSFULLY = False return print("INFO: Clé API 'GEMINI_API_KEY' trouvée dans les variables d'environnement.", file=sys.stderr) try: # 1. Initialisation du client client = genai.Client(api_key=api_key) print("INFO: Client Gemini initialisé. Tentative de self-test...", file=sys.stderr) # 2. Exécution du Self-Test if run_api_self_test(client): MODEL_LOADED_SUCCESSFULLY = True print(f"INFO: Configuration API réussie. Modèle LLM prêt. Utilisant {GEMINI_MODEL_NAME}.", file=sys.stderr) else: # Le self-test a échoué (clé invalide ou autre problème API) client = None # Forcer la réinitialisation si nécessaire MODEL_LOADED_SUCCESSFULLY = False print("ALERTE: ❌ Client Gemini non initialisé. Le self-test a échoué.", file=sys.stderr) except Exception as e: # Ceci attrape les erreurs qui ne sont pas des APIError (ex: problème de librairie, etc.) print(f"ALERTE: ❌ Échec critique de l'initialisation du client Gemini: {e}", file=sys.stderr) client = None MODEL_LOADED_SUCCESSFULLY = False # NOUVEAU: Fonction utilitaire pour créer un objet Part pour un fichier def create_file_part(file_data: Dict) -> Part: """Crée un objet Part à partir d'un dictionnaire de données de fichier.""" # Le backend Render doit fournir 'content' (base64) et 'mime_type' (ex: 'text/plain', 'image/png') # 1. Décoder le contenu de base64 file_bytes = base64.b64decode(file_data.get('content')) mime_type = file_data.get('mime_type') # 2. Créer l'objet Part return Part.from_bytes( data=file_bytes, mime_type=mime_type ) # MODIFIÉ: Ajout du paramètre `files` à la fonction de normalisation def normalize_history_for_gemini( history: List[Dict], new_prompt: str, files: List[Dict] = [] # NOUVEAU: Liste des fichiers à inclure ) -> List[Dict]: """ Convertit l'historique de conversation et les nouveaux fichiers/prompt en le format Contents attendu par l'API Gemini. """ gemini_contents = [] # 1. Traitement de l'historique (messages passés) - Reste le même for conversation_turn in history: for message in conversation_turn.get('messages', []): role = message.get('role') text = message.get('text') gemini_role = 'model' if role == 'assistant' else 'user' if text and gemini_role in ['user', 'model']: gemini_contents.append({ "role": gemini_role, "parts": [{"text": text}] }) # 2. Ajout du nouveau prompt et des fichiers de l'utilisateur (le message actuel) user_parts = [] # D'abord les fichiers (s'il y en a) for file_data in files: # Tente de créer un Part pour chaque fichier try: part = create_file_part(file_data) user_parts.append(part) except Exception as e: # Gérer les erreurs de décodage ou de format print(f"Erreur de traitement de fichier: {e}", file=sys.stderr) # Optionnel: Ajouter un message d'erreur au prompt pour informer l'IA user_parts.append({"text": f"(Erreur: Le fichier {file_data.get('filename', 'inconnu')} n'a pas pu être lu)"}) # Ensuite le texte du prompt if new_prompt: user_parts.append({"text": new_prompt}) # Ajouter le tour de conversation utilisateur if user_parts: gemini_contents.append({ "role": "user", "parts": user_parts }) return gemini_contents # MODIFIÉ: Ajout du paramètre `files` à la fonction de génération def generate_text_response( prompt: str, history: List[Dict] = [], files: List[Dict] = [] # NOUVEAU: Liste des fichiers (base64) ) -> Generator[str, None, None]: """ Génère une réponse en streaming de l'API Gemini en utilisant l'historique et les fichiers. """ get_model_instance() if not MODEL_LOADED_SUCCESSFULLY or client is None: yield "Erreur: Le service Gemini n'est pas initialisé. Vérifiez la clé API." return # Si le prompt et les fichiers sont vides, on s'arrête if not prompt and not files: yield "Erreur: Le contenu de la requête (prompt et/ou fichiers) est vide." return # Configuration de la génération - Reste la même config = genai.types.GenerateContentConfig( system_instruction=SYSTEM_PROMPT, max_output_tokens=DEFAULT_MAX_TOKENS, temperature=DEFAULT_TEMPERATURE, ) # MODIFIÉ: Préparation du contenu, y compris l'historique ET les fichiers try: contents = normalize_history_for_gemini(history, prompt, files) # PASSAGE DES FICHIERS except Exception as e: error_message = f"Erreur lors de la normalisation de l'historique et des fichiers pour Gemini: {e}" print(f"Erreur de normalisation: {error_message}", file=sys.stderr) yield f"Erreur: {error_message}" return try: # --- Appel à l'API STREAMING --- response_stream = client.models.generate_content_stream( model=GEMINI_MODEL_NAME, contents=contents, config=config, ) for chunk in response_stream: if chunk.text: yield chunk.text except APIError as e: error_message = f"Erreur de l'API Gemini: {e.message}" print(f"Erreur API: {error_message}", file=sys.stderr) yield f"Erreur: {error_message}" except Exception as e: error_message = f"Erreur inattendue lors de la génération: {str(e)}" print(f"Erreur Générale: {error_message}", file=sys.stderr) yield f"Erreur: {error_message}"