Introduction à LangGraph

LangGraph est une extension du framework LangChain, spécialement conçue pour construire des workflows d'agents d'IA qui conservent un état (stateful). Il s'appuie sur une structure de graphe pour définir plusieurs nœuds d'exécution et leurs relations de dépendance, permettant ainsi d'orchestrer des tâches complexes. Ce modèle convient particulièrement aux scénarios de collaboration multi-agents et de gestion de conversations longues.

Avantages clés

• Persistance d'état : maintient automatiquement le contexte durant l'exécution.
• Orchestration flexible : supporte les branches conditionnelles, les boucles et l'exécution parallèle.
• Mécanismes de tolérance aux pannes : inclut la logique de réessai et de rollback.
• Débogage visuel : génère automatiquement un diagramme du flux d'exécution.

Structure de code fondamentale

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# Définition de la structure d'état
class WorkflowState(TypedDict):
    données_entrée: str
    résultats_collectés: Annotated[list, operator.add]  # Accumulation automatique

# Initialisation du graphe
workflow_graphe = StateGraph(WorkflowState)

# Ajout de nœuds et d'arêtes (détails dans les sections suivantes)
# ...

# Compilation et exécution
application = workflow_graphe.compile()
résultat_final = application.invoke({"données_entrée": "Description de la tâche"})

Principes de développement

• Conception modulaire : chaque nœud doit avoir une responsabilité unique.
• État minimal : ne conserve que les données strictement nécessaires.
• Idempotence : les nœuds doivent pouvoir être relancés de manière sûre.

Conception de l'état

Normes de définition

Utilisez un modèle Pydantic ou TypedDict pour définir explicitement la structure de l'état :

from pydantic import BaseModel

class EtatProjet(BaseModel):
    cahier_des_charges: str
    versions_brouillon: list[str]
    étape_courante: int

# Initialisation de l'état
état_initial = EtatProjet(
    cahier_des_charges="Développer un chatbot",
    versions_brouillon=[],
    étape_courante=0
)

Gestion automatique de l'état

LangGraph utilise des annotations pour mettre à jour automatiquement les champs de l'état :

from langgraph.graph import add_messages

class EtatDialogue(TypedDict):
    historique: Annotated[list, add_messages]  # Ajout automatique de messages

def nœud_utilisateur(état: EtatDialogue):
    return {"historique": ["Utilisateur: Bonjour"]}

def nœud_agent(état: EtatDialogue):
    return {"historique": ["IA: Bonjour, comment puis-je vous aider ?"]}

Fonctions de nœud

Standard de définition

Les nœuds sont l'unité de base du workflow. Ils reçoivent l'état en entrée et retournent une mise à jour :

from langchain_core.runnables import RunnableLambda

# Nœud simple
def traitement_données(état: dict):
    return {"données_traitées": charger_dataset(état["données_entrée"])}

# Nœud contenant un appel à un LLM
nœud_ia = RunnableLambda(
    lambda état: {"réponse": modèle_chat.invoke(état["question"])}
)

# Enregistrement des nœuds
workflow_graphe.add_node("chargeur", traitement_données)
workflow_graphe.add_node("ia", nœud_ia)

Collaboration multi-agents

def agent_designer(état):
    return {"maquette": "Esquisse de l'interface"}

def agent_développeur(état):
    return {"code": "Code d'implémentation"}

# Exécution parallèle
workflow_graphe.add_node("designer", agent_designer)
workflow_graphe.add_node("développeur", agent_développeur)
workflow_graphe.add_edge("designer", "réviseur")
workflow_graphe.add_edge("développeur", "réviseur")

Conception des arêtes

Branches conditionnelles

Routage dynamique basé sur la valeur de l'état :

from langgraph.graph import conditional_edge

def vérifier_continuation(état):
    return "continuer" if état["étape"] < 5 else "terminer"

workflow_graphe.add_conditional_edges(
    source="nœud_décision",
    chemin_map={"continuer": "nœud_suivant", "terminer": END},
    condition=vérifier_continuation
)

Structure en boucle

workflow_graphe.add_edge("début", "traitement")
workflow_graphe.add_conditional_edges(
    "traitement",
    lambda s: "boucle" if s["compteur"] < 3 else "fin",
    {"boucle": "traitement", "fin": END}
)

Gestion des erreurs

Mécanisme de réessai

from langgraph.retry import RetryPolicy

politique = RetryPolicy(
    max_retries=3,
    backoff_factor=1.5,
    retry_on=(Exception,)
)
workflow_graphe.add_node(
    "appel_api",
    wrapper_api.with_retry(politique)
)

Stratégie de compensation

def action_compensation(état):
    # Exécuter une opération de compensation
    annuler_transaction(état["id_transaction"])
    return {"statut": "annulé"}

workflow_graphe.add_edge("nœud_échoué", "compensation")
workflow_graphe.add_edge("compensation", END)