LangChain Masterclass : construire des agents IA autonomes

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# ============================================================
# CHAIN : flux prédéfini et linéaire
# ============================================================
chain = (
    ChatPromptTemplate.from_template("Traduis en français : {text}")
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)
# Toujours le même chemin : prompt → LLM → parse

result = chain.invoke({"text": "Hello world"})
# → "Bonjour le monde"


# ============================================================
# AGENT : flux dynamique et adaptatif
# ============================================================
from langchain.agents import create_react_agent, AgentExecutor
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

llm = ChatOpenAI(model="gpt-4o", temperature=0)

tools = [
    DuckDuckGoSearchRun(name="web_search"),
    WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper()),
]

# L'agent DÉCIDE quels outils utiliser et dans quel ordre
agent = create_react_agent(llm, tools, prompt_template)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({
    "input": "Quel est le PIB de la France en 2025 et comment se compare-t-il à l'Allemagne ?"
})
# L'agent peut :
# 1. Chercher "PIB France 2025" sur le web
# 2. Chercher "PIB Allemagne 2025" sur le web
# 3. Consulter Wikipedia pour le contexte historique
# 4. Synthétiser et comparer
# Le nombre d'étapes et l'ordre sont décidés par le LLM !

Question: Quelle est la population de Tokyo et comment se compare-t-elle à Paris ?

Thought: Je dois trouver la population actuelle de Tokyo et de Paris. 
Je vais commencer par chercher Tokyo.

Action: web_search
Action Input: "population Tokyo 2025"

Observation: Tokyo compte environ 13,96 millions d'habitants dans la 
ville proprement dite et 37,4 millions dans l'agglomération (2025).

Thought: J'ai la population de Tokyo. Maintenant je dois chercher 
celle de Paris pour comparer.

Action: web_search
Action Input: "population Paris 2025"

Observation: Paris intra-muros compte environ 2,1 millions d'habitants. 
L'agglomération parisienne (Île-de-France) rassemble 12,3 millions de personnes.

Thought: J'ai les deux chiffres. Je peux maintenant comparer et formuler 
une réponse complète.

Final Answer: Tokyo (13,96 millions en ville, 37,4M en agglomération) est 
significativement plus peuplée que Paris (2,1 millions en ville, 12,3M en 
agglomération). Tokyo ville est environ 6,6 fois plus peuplée que Paris 
intra-muros, et son agglomération est 3 fois plus grande que celle de Paris.

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_community.tools import DuckDuckGoSearchRun


# ============================================================
# 1. LE LLM (cerveau de l'agent)
# ============================================================
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,        # Déterministe pour les agents
    max_tokens=4096,
    request_timeout=60,
)

# ============================================================
# 2. LES OUTILS (capacités de l'agent)
# ============================================================
@tool
def calculer(expression: str) -> str:
    """Évalue une expression mathématique. Exemple: '2 + 3 * 4'"""
    try:
        # Sécurisé : n'évalue que des expressions mathématiques
        allowed = set("0123456789+-*/.() ")
        if not all(c in allowed for c in expression):
            return "Erreur: expression invalide"
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Erreur de calcul: {e}"

@tool
def obtenir_date_heure() -> str:
    """Retourne la date et l'heure actuelles."""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

search_tool = DuckDuckGoSearchRun(name="recherche_web")

tools = [search_tool, calculer, obtenir_date_heure]

# ============================================================
# 3. LE PROMPT (instructions de l'agent)
# ============================================================
prompt = ChatPromptTemplate.from_messages([
    ("system", """Tu es un assistant de recherche intelligent et méthodique. 
Tu as accès à des outils pour chercher des informations, faire des calculs, 
et obtenir l'heure actuelle.

Règles :
- Utilise les outils quand c'est nécessaire, pas systématiquement
- Vérifie les informations importantes avec une seconde source si possible
- Cite tes sources quand tu utilises la recherche web
- Sois précis et structuré dans tes réponses
- Si tu ne trouves pas l'information, dis-le honnêtement"""),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

# ============================================================
# 4. L'EXÉCUTEUR (boucle de contrôle)
# ============================================================
agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,           # Afficher la trace de raisonnement
    max_iterations=10,      # Limite de sécurité
    max_execution_time=120, # Timeout en secondes
    handle_parsing_errors=True,
    return_intermediate_steps=True,  # Pour le debugging
)

# ============================================================
# UTILISATION
# ============================================================
result = agent_executor.invoke({
    "input": "Quelle est la racine carrée de la population de la France ?"
})

print(result["output"])
# L'agent va :
# 1. Chercher la population de la France
# 2. Calculer la racine carrée du résultat
# 3. Formuler la réponse

# Accéder aux étapes intermédiaires
for step in result.get("intermediate_steps", []):
    action, observation = step
    print(f"Action: {action.tool} → {observation[:100]}")

from langchain_core.tools import tool, BaseTool, ToolException
from pydantic import BaseModel, Field
from typing import Optional, Type
import httpx
import json


# ============================================================
# Méthode 1 : Décorateur @tool (simple et rapide)
# ============================================================

@tool
def analyser_sentiment(texte: str) -> str:
    """Analyse le sentiment d'un texte et retourne positif, négatif ou neutre.
    
    Args:
        texte: Le texte à analyser (max 1000 caractères)
    """
    # Simulation — en production, appelez votre modèle
    mots_positifs = {"excellent", "super", "génial", "parfait", "bravo", "merci", "bien"}
    mots_negatifs = {"mauvais", "horrible", "nul", "problème", "erreur", "pire", "déçu"}
    
    mots = set(texte.lower().split())
    score_pos = len(mots & mots_positifs)
    score_neg = len(mots & mots_negatifs)
    
    if score_pos > score_neg:
        return f"Sentiment: POSITIF (score: +{score_pos})"
    elif score_neg > score_pos:
        return f"Sentiment: NÉGATIF (score: -{score_neg})"
    return "Sentiment: NEUTRE"


@tool
def requete_api(url: str, methode: str = "GET") -> str:
    """Effectue une requête HTTP vers une API externe.
    
    Args:
        url: L'URL de l'API à appeler
        methode: La méthode HTTP (GET ou POST)
    """
    try:
        with httpx.Client(timeout=10) as client:
            if methode.upper() == "GET":
                response = client.get(url)
            else:
                response = client.post(url)
            
            return json.dumps({
                "status_code": response.status_code,
                "body": response.text[:2000],
            }, ensure_ascii=False)
    except Exception as e:
        return f"Erreur API: {str(e)}"


# ============================================================
# Méthode 2 : Classe BaseTool (contrôle total)
# ============================================================

class RechercheBaseConnaissancesInput(BaseModel):
    """Schéma d'entrée pour la recherche dans la base de connaissances."""
    query: str = Field(description="La question ou les mots-clés de recherche")
    categorie: Optional[str] = Field(
        default=None,
        description="Catégorie pour filtrer (ex: 'technique', 'juridique', 'finance')"
    )
    max_results: int = Field(
        default=3,
        ge=1,
        le=10,
        description="Nombre maximum de résultats"
    )


class RechercheBaseConnaissances(BaseTool):
    """Outil de recherche dans la base de connaissances interne."""
    
    name: str = "recherche_base_connaissances"
    description: str = """Recherche des informations dans la base de connaissances 
    interne de l'entreprise. Utile pour trouver des procédures, de la documentation 
    technique, des politiques internes, et des FAQ."""
    args_schema: Type[BaseModel] = RechercheBaseConnaissancesInput
    
    # Attributs personnalisés
    vectorstore: any = None
    
    def __init__(self, vectorstore, **kwargs):
        super().__init__(**kwargs)
        self.vectorstore = vectorstore
    
    def _run(
        self,
        query: str,
        categorie: Optional[str] = None,
        max_results: int = 3,
    ) -> str:
        """Exécution synchrone de la recherche."""
        try:
            # Construire les filtres
            search_kwargs = {"k": max_results}
            if categorie:
                search_kwargs["filter"] = {"category": categorie}
            
            # Recherche vectorielle
            docs = self.vectorstore.similarity_search(
                query, **search_kwargs
            )
            
            if not docs:
                return "Aucun résultat trouvé dans la base de connaissances."
            
            results = []
            for i, doc in enumerate(docs):
                source = doc.metadata.get("source", "inconnu")
                results.append(
                    f"[{i+1}] Source: {source}\n{doc.page_content[:500]}"
                )
            
            return "\n\n---\n\n".join(results)
        
        except Exception as e:
            raise ToolException(f"Erreur de recherche: {str(e)}")
    
    async def _arun(self, query: str, **kwargs) -> str:
        """Version asynchrone (optionnel)."""
        return self._run(query, **kwargs)


# ============================================================
# Méthode 3 : Outil avec gestion d'erreurs robuste
# ============================================================

@tool(handle_tool_error=True)
def executer_sql(query: str) -> str:
    """Exécute une requête SQL en lecture seule sur la base de données.
    
    ATTENTION : Seules les requêtes SELECT sont autorisées.
    
    Args:
        query: La requête SQL à exécuter (SELECT uniquement)
    """
    # Sécurité : vérifier que c'est un SELECT
    if not query.strip().upper().startswith("SELECT"):
        raise ToolException(
            "Seules les requêtes SELECT sont autorisées. "
            "Reformulez votre requête en lecture seule."
        )
    
    # Sécurité : interdire certains mots-clés
    dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE"]
    if any(kw in query.upper() for kw in dangerous):
        raise ToolException("Requête dangereuse détectée. Opération refusée.")
    
    # Exécution (simulée)
    return json.dumps({
        "columns": ["id", "name", "revenue"],
        "rows": [
            [1, "Product A", 150000],
            [2, "Product B", 230000],
        ],
        "row_count": 2,
    })

from langchain.memory import (
    ConversationBufferMemory,
    ConversationSummaryMemory,
    ConversationBufferWindowMemory,
    ConversationSummaryBufferMemory,
)
from langchain_openai import ChatOpenAI


# ============================================================
# 1. Buffer Memory : stocke tout l'historique brut
# ============================================================
# Simple mais consomme beaucoup de tokens sur les longues conversations
buffer_memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
)

# Ajouter des messages
buffer_memory.save_context(
    {"input": "Je travaille sur un projet de chatbot médical."},
    {"output": "Intéressant ! Quel type de données médicales utilisez-vous ?"}
)

# Récupérer l'historique
history = buffer_memory.load_memory_variables({})
print(history["chat_history"])


# ============================================================
# 2. Window Memory : garde les N derniers échanges
# ============================================================
# Compromis entre contexte et consommation de tokens
window_memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    return_messages=True,
    k=10,  # Garder les 10 derniers échanges
)


# ============================================================
# 3. Summary Memory : résume l'historique avec un LLM
# ============================================================
# Idéal pour les longues conversations : le résumé est compact
summary_memory = ConversationSummaryMemory(
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    memory_key="chat_history",
    return_messages=True,
)


# ============================================================
# 4. Summary Buffer Memory : résumé + derniers messages
# ============================================================
# Le meilleur des deux mondes : résumé du passé + messages récents
summary_buffer_memory = ConversationSummaryBufferMemory(
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    memory_key="chat_history",
    return_messages=True,
    max_token_limit=2000,  # Résume quand l'historique dépasse 2000 tokens
)


# ============================================================
# Intégration avec un agent
# ============================================================
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "Tu es un assistant utile avec de la mémoire."),
    MessagesPlaceholder("chat_history"),   # ← Injecte la mémoire ici
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=summary_buffer_memory,  # ← Attacher la mémoire
    verbose=True,
)

# Conversation avec mémoire
r1 = agent_executor.invoke({"input": "Je m'appelle Lucas et je développe en Python."})
r2 = agent_executor.invoke({"input": "Quel langage j'utilise ?"})
# → "Vous développez en Python, Lucas !"

"""
Agent de recherche web avancé
Capable de chercher, lire des pages web, et synthétiser les informations.
"""

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_community.tools import DuckDuckGoSearchResults
from langchain_community.document_loaders import WebBaseLoader
from langchain.memory import ConversationSummaryBufferMemory
import json
from datetime import datetime


# ============================================================
# Outils de recherche
# ============================================================

search = DuckDuckGoSearchResults(
    name="recherche_web",
    max_results=5,
    output_format="list",
)

@tool
def lire_page_web(url: str) -> str:
    """Lit et extrait le contenu textuel d'une page web.
    
    Args:
        url: L'URL de la page à lire
    """
    try:
        loader = WebBaseLoader(url)
        docs = loader.load()
        if docs:
            content = docs[0].page_content[:3000]
            return f"Contenu de {url}:\n\n{content}"
        return "Impossible d'extraire le contenu de cette page."
    except Exception as e:
        return f"Erreur lors de la lecture de {url}: {str(e)}"

@tool
def obtenir_date() -> str:
    """Retourne la date et l'heure actuelles. Utilise cet outil quand 
    tu as besoin de savoir la date du jour."""
    return datetime.now().strftime("Nous sommes le %d/%m/%Y, il est %H:%M.")

@tool
def noter_information(titre: str, contenu: str) -> str:
    """Sauvegarde une information importante pour référence ultérieure.
    Utile pour garder trace des découvertes pendant la recherche.
    
    Args:
        titre: Titre court de la note
        contenu: Contenu de la note
    """
    # En production, sauvegarder dans une base de données
    return f"✅ Note sauvegardée : '{titre}'"


tools = [search, lire_page_web, obtenir_date, noter_information]

# ============================================================
# Configuration de l'agent
# ============================================================

llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """Tu es un agent de recherche expert et méthodique.

Ton processus de recherche :
1. Analyse la question pour identifier les informations nécessaires
2. Effectue des recherches ciblées avec des mots-clés précis
3. Lis les pages les plus pertinentes pour approfondir
4. Vérifie les informations critiques avec une seconde source
5. Synthétise les résultats de manière structurée

Règles :
- Cite TOUJOURS tes sources avec l'URL
- Distingue les faits vérifiés des estimations
- Si les sources se contredisent, mentionne les différentes perspectives
- Structure tes réponses avec des titres et des listes
- Sois concis mais complet

Date actuelle : utilise l'outil obtenir_date si nécessaire."""),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

memory = ConversationSummaryBufferMemory(
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    memory_key="chat_history",
    return_messages=True,
    max_token_limit=2000,
)

agent = create_tool_calling_agent(llm, tools, prompt)

research_agent = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True,
    max_iterations=15,
    max_execution_time=180,
    handle_parsing_errors=True,
)

# ============================================================
# Utilisation
# ============================================================

# Recherche simple
result = research_agent.invoke({
    "input": "Quelles sont les dernières avancées en informatique quantique en 2026 ?"
})
print(result["output"])

# Recherche comparative
result = research_agent.invoke({
    "input": "Compare les performances de Claude 3.5 et GPT-4o sur les benchmarks de code."
})
print(result["output"])

"""
Agent d'analyse et d'exécution de code Python
Capable de lire des fichiers, exécuter du code, et analyser les résultats.
"""

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_experimental.tools import PythonREPLTool
import subprocess
import os


# ============================================================
# Outils de code
# ============================================================

# Outil d'exécution Python (sandboxé en production !)
python_repl = PythonREPLTool(
    name="executer_python",
    description="""Exécute du code Python et retourne le résultat. 
    Utile pour les calculs, l'analyse de données, la création de graphiques.
    Le code a accès à pandas, numpy, matplotlib.""",
)

@tool
def lire_fichier(chemin: str) -> str:
    """Lit le contenu d'un fichier source.
    
    Args:
        chemin: Chemin vers le fichier à lire
    """
    try:
        if not os.path.exists(chemin):
            return f"Fichier non trouvé: {chemin}"
        
        # Sécurité : limiter la taille
        size = os.path.getsize(chemin)
        if size > 100000:
            return f"Fichier trop volumineux ({size} bytes). Max 100KB."
        
        with open(chemin, 'r', encoding='utf-8') as f:
            content = f.read()
        
        return f"Contenu de {chemin} ({len(content)} caractères):\n\n```\n{content}\n```"
    except Exception as e:
        return f"Erreur: {str(e)}"

@tool
def lister_fichiers(repertoire: str = ".") -> str:
    """Liste les fichiers et dossiers dans un répertoire.
    
    Args:
        repertoire: Chemin du répertoire (défaut: répertoire courant)
    """
    try:
        items = []
        for item in sorted(os.listdir(repertoire)):
            path = os.path.join(repertoire, item)
            if os.path.isdir(path):
                items.append(f"📁 {item}/")
            else:
                size = os.path.getsize(path)
                items.append(f"📄 {item} ({size:,} bytes)")
        
        return f"Contenu de {repertoire}:\n" + "\n".join(items)
    except Exception as e:
        return f"Erreur: {str(e)}"

@tool
def analyser_code_statique(code: str) -> str:
    """Analyse un snippet de code Python pour détecter les problèmes 
    potentiels (imports manquants, variables non utilisées, bugs courants).
    
    Args:
        code: Le code Python à analyser
    """
    issues = []
    lines = code.split('\n')
    
    for i, line in enumerate(lines, 1):
        stripped = line.strip()
        
        # Détections basiques (en production, utilisez pylint/ruff)
        if 'eval(' in stripped and 'safe' not in stripped.lower():
            issues.append(f"L{i}: ⚠️ Usage de eval() — risque de sécurité")
        if 'except:' in stripped and 'Exception' not in stripped:
            issues.append(f"L{i}: ⚠️ except trop large (bare except)")
        if 'password' in stripped.lower() and '=' in stripped:
            issues.append(f"L{i}: 🔴 Mot de passe potentiellement en dur")
        if stripped.startswith('import *'):
            issues.append(f"L{i}: ⚠️ Import wildcard — évitez 'from X import *'")
        if 'TODO' in stripped or 'FIXME' in stripped:
            issues.append(f"L{i}: 📝 {stripped.strip()}")
    
    if not issues:
        return "✅ Aucun problème détecté dans l'analyse statique."
    
    return "Problèmes détectés:\n" + "\n".join(issues)


tools = [python_repl, lire_fichier, lister_fichiers, analyser_code_statique]

# ============================================================
# Agent de code
# ============================================================

llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """Tu es un ingénieur logiciel senior expert en Python.

Tes capacités :
- Lire et analyser du code source
- Exécuter du code Python pour tester des hypothèses
- Détecter des bugs et proposer des corrections
- Expliquer du code complexe de manière claire
- Écrire du code propre et documenté

Méthodologie :
1. Comprends d'abord le problème complet avant de coder
2. Lis les fichiers pertinents si nécessaire
3. Écris du code testé et documenté
4. Vérifie les résultats en exécutant le code
5. Explique ton raisonnement"""),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

agent = create_tool_calling_agent(llm, tools, prompt)

code_agent = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=15,
    handle_parsing_errors=True,
)

# Utilisation
result = code_agent.invoke({
    "input": """Crée une fonction Python qui analyse un fichier CSV de ventes 
    et génère un rapport avec :
    - Total des ventes par mois
    - Produit le plus vendu
    - Tendance (croissance/décroissance)
    Teste-la avec des données d'exemple."""
})

"""
Système multi-agents pour la rédaction d'articles
- Agent Rechercheur : collecte les informations
- Agent Rédacteur : écrit l'article
- Agent Éditeur : vérifie la qualité et corrige
"""

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from typing import TypedDict, Annotated, Sequence
import operator


# ============================================================
# État partagé entre les agents
# ============================================================
class ArticleState(TypedDict):
    """État global du workflow d'écriture."""
    topic: str
    research_notes: str
    draft: str
    review_feedback: str
    final_article: str
    revision_count: int
    messages: Annotated[Sequence, operator.add]


# ============================================================
# Agents spécialisés
# ============================================================

researcher_llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
writer_llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
editor_llm = ChatOpenAI(model="gpt-4o", temperature=0.2)


def researcher_node(state: ArticleState) -> dict:
    """Agent Rechercheur : collecte les informations sur le sujet."""
    messages = [
        SystemMessage(content="""Tu es un chercheur expert. Ton travail est de 
        rassembler les informations clés sur un sujet pour un article de blog.
        Fournis des faits, des chiffres, des citations, et des angles intéressants.
        Organise les notes par thème."""),
        HumanMessage(content=f"""Recherche des informations sur le sujet suivant 
        pour un article de blog :
        
        Sujet : {state['topic']}
        
        Fournis des notes de recherche structurées avec des faits vérifiables."""),
    ]
    
    response = researcher_llm.invoke(messages)
    
    return {
        "research_notes": response.content,
        "messages": [AIMessage(content=f"[Rechercheur] Notes complétées.")]
    }


def writer_node(state: ArticleState) -> dict:
    """Agent Rédacteur : écrit le brouillon basé sur la recherche."""
    
    feedback = ""
    if state.get("review_feedback"):
        feedback = f"\n\nFeedback de l'éditeur à prendre en compte :\n{state['review_feedback']}"
    
    messages = [
        SystemMessage(content="""Tu es un rédacteur de blog technique talentueux.
        Écris des articles engageants, bien structurés, avec un ton professionnel 
        mais accessible. Utilise des titres, des listes, et des exemples concrets."""),
        HumanMessage(content=f"""Écris un article de blog basé sur ces notes de recherche :

Notes : {state['research_notes']}
{feedback}

L'article doit être complet, bien structuré, et faire au moins 1000 mots."""),
    ]
    
    response = writer_llm.invoke(messages)
    
    return {
        "draft": response.content,
        "messages": [AIMessage(content=f"[Rédacteur] Brouillon rédigé.")]
    }


def editor_node(state: ArticleState) -> dict:
    """Agent Éditeur : vérifie la qualité et donne du feedback."""
    messages = [
        SystemMessage(content="""Tu es un éditeur en chef exigeant mais constructif.
        Évalue l'article sur : clarté, structure, précision factuelle, engagement,
        grammaire, et complétude. 
        
        Si l'article est satisfaisant, réponds EXACTEMENT "APPROVED" suivi de 
        suggestions mineures optionnelles.
        
        Sinon, fournis un feedback détaillé pour amélioration."""),
        HumanMessage(content=f"""Évalue cet article :

{state['draft']}

Donne ton verdict : APPROVED ou feedback détaillé pour révision."""),
    ]
    
    response = editor_llm.invoke(messages)
    
    return {
        "review_feedback": response.content,
        "revision_count": state.get("revision_count", 0) + 1,
        "messages": [AIMessage(content=f"[Éditeur] Review #{state.get('revision_count', 0) + 1} terminée.")]
    }


def should_revise(state: ArticleState) -> str:
    """Décide si l'article nécessite une révision."""
    if "APPROVED" in state.get("review_feedback", ""):
        return "finalize"
    if state.get("revision_count", 0) >= 3:
        return "finalize"  # Max 3 révisions
    return "revise"


def finalize_node(state: ArticleState) -> dict:
    """Finalise l'article approuvé."""
    return {
        "final_article": state["draft"],
        "messages": [AIMessage(content="[Système] Article finalisé ✅")]
    }


# ============================================================
# Construction du graphe
# ============================================================
workflow = StateGraph(ArticleState)

# Ajouter les nœuds
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_node("editor", editor_node)
workflow.add_node("finalizer", finalize_node)

# Définir le flux
workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "editor")
workflow.add_conditional_edges(
    "editor",
    should_revise,
    {
        "revise": "writer",    # Retour au rédacteur
        "finalize": "finalizer",  # Article approuvé
    }
)
workflow.add_edge("finalizer", END)

# Compiler
app = workflow.compile()

# ============================================================
# Exécution
# ============================================================
result = app.invoke({
    "topic": "L'impact de l'IA générative sur le développement logiciel en 2026",
    "research_notes": "",
    "draft": "",
    "review_feedback": "",
    "final_article": "",
    "revision_count": 0,
    "messages": [],
})

print(f"\n📝 Article final ({result['revision_count']} révisions) :")
print(result["final_article"])

from langchain.callbacks import CallbackHandler
from langchain_core.callbacks import BaseCallbackHandler
import tiktoken


class SafetyCallbackHandler(BaseCallbackHandler):
    """Callback pour monitorer et limiter les agents."""
    
    def __init__(self, max_tokens: int = 100000, max_cost: float = 1.0):
        self.total_tokens = 0
        self.max_tokens = max_tokens
        self.total_cost = 0.0
        self.max_cost = max_cost
        self.actions_taken = []
    
    def on_llm_end(self, response, **kwargs):
        """Appelé après chaque appel LLM."""
        if hasattr(response, 'llm_output') and response.llm_output:
            usage = response.llm_output.get('token_usage', {})
            tokens = usage.get('total_tokens', 0)
            self.total_tokens += tokens
            
            # Estimation du coût (GPT-4o)
            self.total_cost += tokens * 0.000005  # Approximatif
            
            if self.total_tokens > self.max_tokens:
                raise RuntimeError(
                    f"⚠️ Limite de tokens dépassée: {self.total_tokens}/{self.max_tokens}"
                )
            if self.total_cost > self.max_cost:
                raise RuntimeError(
                    f"⚠️ Limite de coût dépassée: ${self.total_cost:.2f}/${self.max_cost}"
                )
    
    def on_tool_start(self, tool, input_str, **kwargs):
        """Appelé avant chaque utilisation d'outil."""
        self.actions_taken.append({"tool": tool.get("name"), "input": input_str[:200]})
        
        # Détecter les boucles (même outil, même input)
        recent = self.actions_taken[-5:]
        if len(recent) >= 3:
            if all(a["tool"] == recent[0]["tool"] and a["input"] == recent[0]["input"] 
                   for a in recent):
                raise RuntimeError("⚠️ Boucle détectée: même action répétée 3+ fois")


# Utilisation
safety = SafetyCallbackHandler(max_tokens=50000, max_cost=0.50)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[safety],
    max_iterations=10,
    max_execution_time=120,
)