import os from typing import Annotated, Sequence, TypedDict, Optional , List from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage from langgraph.graph.message import add_messages from langchain_google_genai import ChatGoogleGenerativeAI from langchain_groq import ChatGroq from langfuse import get_client import pandas as pd from langfuse.langchain import CallbackHandler # Initialize Langfuse CallbackHandler for Langchain (tracing) langfuse_handler = CallbackHandler() # Import de tes outils corrigés from tools import excel_code_interpreter, search_tool , inspect_data langfuse = get_client() load_dotenv() # 1. Définition du State class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] current_df_path: Optional[str] generated_charts : List[str] # 2. Configuration du Modèle et des Outils model =ChatGroq(model="llama-3.3-70b-versatile") #model_llama=ChatGoogleGenerativeAI(model="gemini-2.5-flash") model_gpt = ChatOpenAI(model="gpt-5.3-codex" ) model_gpt_5 = ChatOpenAI(model="o4-mini" ) tools = [ search_tool , excel_code_interpreter, inspect_data ] model_with_tools = model_gpt.bind_tools(tools) # 3. Définition des Nœuds def agent_analyseur(state: AgentState): # --- PHASE D'INSPECTION AUTOMATIQUE --- # On récupère le chemin du fichier depuis le state file_path = state.get("current_df_path") inspection_info = "" if file_path and os.path.exists(file_path): try: # Lecture des 5 premières lignes pour comprendre la structure df_temp = pd.read_csv(file_path, nrows=5) if file_path.endswith('.csv') else pd.read_excel(file_path, nrows=5) columns_list = df_temp.columns.tolist() sample_data = df_temp.head(2).to_string() inspection_info = f"\n\n### DONNÉES RÉELLES DU FICHIER :\n- Colonnes détectées : {columns_list}\n- Aperçu des données :\n{sample_data}" except Exception as e: inspection_info = f"\n\n⚠️ Erreur lors de l'inspection du fichier : {e}" # --- CONSTRUCTION DU PROMPT AVEC LES DONNÉES RÉELLES --- prompt = ( "Tu es l'Analyseur Stratégique de Dataltist. Ton rôle est de définir le 'QUOI' faire, pas le 'COMMENT'.\n\n" "### RÈGLES D'OR :\n" "1. PAS DE CODE : Ne génère JAMAIS de blocs de code Python (```python). C'est le rôle de l'Exécuteur.\n" "2. PLAN D'ACTION : Liste les étapes logiques en utilisant les noms de colonnes exacts.\n" "3. INSPECTION : Base-toi uniquement sur les colonnes détectées ci-dessous.\n" "4. CONCISION : Sois une tour de contrôle, donne des ordres clairs et précis.\n" f"{inspection_info}" ) config_analyseur = { "callbacks": [langfuse_handler], "metadata": {"agent_name": "Analyseur"}, "tags": ["Dataltist", "Planning"] } msg = [SystemMessage(content=prompt)] + state["messages"] response = model.invoke(msg , config=config_analyseur) # On retourne la réponse de l'IA qui contient maintenant le plan basé sur les vraies colonnes return {"messages": [response]} def agent_executor(state: AgentState): # 1. On récupère le chemin actuel depuis le State # Si current_df_path est None, on met data.xlsx par défaut file_path = state.get("current_df_path") or "data.xlsx" clean_path = file_path.replace("\\", "/") # 2. On construit un prompt qui contient le VRAI chemin prompt = ( f"Tu es un Data Scientist expert spécialisé en automatisation. Le fichier cible est disponible ici : '{file_path}'.\n\n" "PROTOCOLE D'ACTION IMPÉRATIF :\n" "1. CONFIANCE STRATÉGIQUE : Utilise les noms de colonnes fournis par l'Analyseur ( Obligatoire , change pas ).\n" "2. RECHERCHE EXTERNE : Utilise 'search_tool' si des données manquent.\n" "3. ZÉRO SIMULATION : Ne jamais inventer de données.\n" "4. LOGIQUE PANDAS : Utilise 'df' pour filtrer les données exactes.\n" "5. CODE AUTONOME : Ton script Python doit être complet et prêt à l'exécution. \n" f" - Tu DOIS impérativement définir la variable `file_path = '{clean_path}'` au début de ton code.\n" " - Inclus tous les imports (pandas as pd, matplotlib.pyplot as plt, etc.).\n" " - Termine impérativement par 'result = ...'.\n" "6. PERSISTANCE : Utilise 'plt.savefig('outputs/nom_du_graphe.png')' et ferme la figure avec 'plt.close()'.\n\n" "Tu es un moteur d'exécution froid et précis. Pas de discours.\n" "STRUCTURE DE RÉPONSE :\n" "- 'Statut : Code exécuté avec succès.'\n" "- 'Fichiers : [liste des .png générés]'\n" "- 'Validation : Les données de [Colonnes] ont été traitées.'\n" ) # 3. Préparation des messages messages = [SystemMessage(content=prompt)] + state["messages"] config_executor= { "callbacks": [langfuse_handler], "metadata": {"agent_name": "executor"}, "tags": ["Dataltist", "executor"] } # 4. Appel du modèle avec les outils response = model_with_tools.invoke(messages , config= config_executor ) return {"messages": [response]} def agent_reporter(state : AgentState) : prompt_reporter = ( "Tu es l'Agent de Reporting de Dataltist.\n" "Ton rôle : Convertir les logs de l'Exécuteur en une synthèse factuelle pour l'utilisateur.\n\n" "RÈGLES CRITIQUES (ANTI-SILENCE) :\n" "- DISPONIBILITÉ : Tu dois TOUJOURS générer une réponse, même pour confirmer une erreur ou une absence de données.\n" "- STRUCTURE : Utilise des tirets pour lister les points clés.\n" "- SYNTHÈSE : Si l'Exécuteur donne 'result = 42', écris 'Résultat : 42'.\n" "- FICHIERS : Liste uniquement le nom des fichiers (ex: graphe_incendie.png), JAMAIS le chemin 'outputs/'.\n\n" "FORMAT IMPÉRATIF :\n" "1. Résumé des données traitées.\n" "2. Valeurs calculées (moyennes, totaux, etc.).\n" "3. Liste des visuels générés (si présents).\n\n" "4. L'affichage doit etre claire , tu laisse les espaces .\n\n" "INTERDICTIONS :\n" "- Pas de politesse inutile ('Voici le rapport...', 'J'espère que...').\n" "- Pas de répétition des étapes techniques de l'agent.\n" "- Pas de phrases vides si l'exécution a échoué : explique brièvement l'échec." "RÈGLES D'AFFICHAGE DES FICHIERS :\n" "1. CITATION : Si l'Exécuteur a généré des fichiers (images ou données), cite-les obligatoirement.\n" "2. FORMAT : Utilise TOUJOURS ce format exact pour introduire un fichier : \n" " '📊 Visualisation générée : [nom_du_fichier.png]'\n" " '📥 Fichier disponible : [nom_du_fichier.csv/xlsx]'\n" "3. SANS CHEMIN : Ne mentionne JAMAIS le dossier 'outputs/' ou 'data/'. Utilise uniquement le nom final du fichier.\n\n" ) messages = [SystemMessage(content=prompt_reporter)] + state["messages"] response = model.invoke(messages ) return{"messages" : [response]}