LangGraph : Révolutionner les flux de travail IA en Python

Introduction à LangGraph : Une nouvelle approche des flux de travail IA

Dans le domaine de l'intelligence artificielle, la gestion des agents conversationnels se limite souvent à des interactions simples : poser une question, obtenir une réponse. Cependant, lorsque les tâches deviennent plus complexes, les défis se multiplient. Un agent peut devoir accéder à une base de données, se souvenir des conversations passées, ou fournir des explications sur ses décisions. Ces exigences complexes nécessitent souvent des solutions sur mesure, ce qui peut être un obstacle pour de nombreuses implémentations.

LangGraph propose une solution élégante à ces problèmes. Il structure les agents IA sous forme de graphiques, où chaque nœud représente une unité de travail, et les arêtes déterminent l'ordre d'exécution. Un objet d'état partagé conserve l'historique des messages, permettant à chaque étape du processus d'être visible et accessible pour les nœuds suivants. Cette approche rend le flux d'exécution transparent et inspectable, facilitant ainsi la gestion des agents IA.

Dans cet article, nous explorerons comment LangGraph gère les primitives d'état, de nœud et d'arête. Nous verrons comment il permet de gérer automatiquement l'historique des conversations avec MessagesState, d'appeler un modèle de langage au sein d'un nœud, d'enregistrer des outils, de router les appels d'outils, et de persister les conversations avec un checkpointer. Nous construirons un graphique étape par étape, en commençant par l'installation des outils nécessaires.

Configuration initiale

Pour commencer à utiliser LangGraph, il est essentiel d'installer les packages requis. Voici les commandes à exécuter :

pip install langgraph langchain-[openai](/dossier/openai) python-dotenv

Une fois les packages installés, créez un fichier .env à la racine de votre projet pour stocker votre clé API OpenAI :

OPENAI_API_KEY="your_key_here"

Chargez ce fichier au début de votre script pour définir la clé comme variable d'environnement, avant toute importation de LangChain ou LangGraph :

from dotenv import load_dotenv
load_dotenv()

Le module python-dotenv lit le fichier .env et définit la clé comme une variable d'environnement, simplifiant ainsi la gestion des configurations sensibles.

Comprendre les composantes de LangGraph

Chaque graphique LangGraph repose sur trois éléments fondamentaux : l'état, les nœuds et les arêtes. Comprendre ces composants est crucial pour éviter les confusions lorsque le graphique se complexifie.

• État : Il s'agit d'un TypedDict qui sert de mémoire partagée pour l'ensemble du graphique. Chaque nœud peut lire et écrire des mises à jour dans cet état. Les champs non modifiés restent inchangés, et seuls les champs que vous souhaitez modifier sont renvoyés.

• Nœuds : Ce sont des fonctions Python classiques qui prennent l'état actuel comme argument et renvoient un dictionnaire des champs à mettre à jour. En enregistrant une fonction avec add_node, elle devient une partie intégrante du graphique sans nécessiter de décorateur spécial ou de classe de base.

• Arêtes : Elles définissent l'ordre d'exécution. Par exemple, add_edge(A, B) signifie que le nœud B s'exécute après le nœud A. Les arêtes conditionnelles permettent de diriger le flux en fonction des résultats des nœuds.

Par défaut, lorsqu'un nœud met à jour un champ d'état, cette mise à jour remplace la valeur précédente. Pour les champs nécessitant une accumulation, comme un journal ou un historique de messages, une fonction de réduction est utilisée. Par exemple, operator.add sur un champ de liste permet d'ajouter des éléments plutôt que de les remplacer.

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

class TicketState(TypedDict):
    customer_message: str
    log: Annotated[list, operator.add]

def log_received(state: TicketState) -> dict:
    return {"log": [f"Received: {state['customer_message']}"]}

def log_assigned(state: TicketState) -> dict:
    return {"log": ["Assigned to support queue"]}

builder = StateGraph(TicketState)
builder.add_node("log_received", log_received)
builder.add_node("log_assigned", log_assigned)
builder.add_edge(START, "log_received")
builder.add_edge("log_received", "log_assigned")
builder.add_edge("log_assigned", END)

graph = builder.compile()
result = graph.invoke({"customer_message": "My invoice looks wrong", "log": []})
print(result)

Ce code produit un journal où chaque nœud ajoute une entrée, et le message du client reste inchangé car aucun nœud ne l'a modifié. C'est ainsi que MessagesState gère son champ de messages, en utilisant un réducteur spécialisé appelé add_messages qui assure également la dé-duplication et l'ordre des messages.

Gestion de l'historique des conversations avec MessagesState

Dans un graphique LangGraph, chaque nœud lit l'état actuel et écrit des mises à jour. Pour un agent conversationnel, l'état doit inclure l'historique complet des messages, comprenant les entrées utilisateur, les réponses du modèle, et les sorties d'outils, afin que le modèle dispose toujours du contexte nécessaire pour décider de la prochaine action.

LangGraph propose MessagesState, un type d'état intégré qui utilise le réducteur add_messages. Ce réducteur ajoute de nouveaux messages à la liste existante au lieu de la remplacer, simplifiant ainsi la gestion de l'historique des conversations.

from langgraph.graph import MessagesState

Cette définition d'état est suffisante pour la plupart des graphiques à agent unique. Vous pouvez l'étendre avec des champs supplémentaires, tels qu'un customer_id ou un indicateur de priorité, selon les besoins de vos nœuds. Cependant, le champ messages est déjà configuré pour s'accumuler automatiquement.

Intégration du modèle de langage dans un nœud

Avec l'état en place, le nœud central d'un agent LangGraph est une fonction qui transmet la liste actuelle des messages à un modèle de langage et ajoute sa réponse. Le modèle renvoie un AIMessage, qui est ensuite ajouté à l'état.

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage

llm = ChatOpenAI(model="[gpt](/glossaire/gpt)-4o-mini")

def run_model(state: MessagesState) -> dict:
    system = SystemMessage("You are a support agent for a SaaS product. Be concise and helpful.")
    response = llm.invoke([system] + state["messages"])
    return {"messages": [response]}

ChatOpenAI encapsule l'API OpenAI avec une interface standard de modèle de chat de LangChain. Passer à un autre fournisseur, comme Anthropic ou Google, nécessite simplement de changer l'importation et la chaîne du modèle, sans affecter le reste du nœud. SystemMessage définit le rôle du modèle à chaque appel, sans être stocké dans l'état, ce qui maintient l'historique propre.

Pour intégrer ce modèle dans un graphique et l'exécuter :

from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage

builder = StateGraph(MessagesState)
builder.add_node("run_model", run_model)
builder.add_edge(START, "run_model")
builder.add_edge("run_model", END)

graph = builder.compile()
result = graph.invoke({"messages": [HumanMessage("My dashboard isn't loading. What should I try?")]})
print(result["messages"][-1].content)

Le résultat contient la liste complète des messages, incluant le HumanMessage original et l'AIMessage généré par le modèle. Le dernier message est récupéré avec [-1].

Enregistrement et routage des appels d'outils

Bien que le modèle puisse répondre à des questions générales grâce à ses données d'entraînement, les informations spécifiques à vos données, comme les détails de compte ou l'historique des tickets, nécessitent des appels d'outils. Le modèle décide quand un outil doit être appelé, ce qui permet d'intégrer des fonctionnalités avancées dans le flux de travail de l'agent.