Créez vos premiers flux de travail LangGraph avant les systèmes multi-agents

Qu’est-ce que LangGraph ?

Les scripts Python ordinaires s’exécutent de haut en bas. Les applications agentes se déplacent à travers un graphe : chaque nœud effectue une tâche, puis transmet le contrôle le long d’une arête. Une représentation simplifiée ressemble à ceci :

Input
   ↓
Node A
   ↓
Node B
   ↓
Node C
   ↓
Output

Chaque nœud reçoit un état commun, le met à jour et le transmet. Ce schéma peut être étendu pour inclure des branches, de la mémoire, de la planification, de la réflexion et des configurations multi-agents — mais la première compétence à maîtriser est l’organisation d’un pipeline linéaire.

La démonstration ci-dessous reste reproductible : Google Colab, des modèles open-source via Groq, ainsi qu’une courte liste de dépendances. Aucune installation locale de GPU n’est requise. L’objectif pédagogique est que les concepts liés aux agents s’imposent plus rapidement lorsque chaque étape s’exécute sur des infrastructures gratuites, permettant ainsi de se concentrer sur l’état, les nœuds et les arêtes plutôt que sur la configuration du cluster.

Installer les dépendances

!pip install langchain langgraph langchain_groq

• LangGraph crée le graphe de flux de travail

LangChain fournit des abstractions pour les LLM

LangChain Groq communique avec l’API d’inférence de Groq

Exécuter les installations à l’intérieur de Colab permet de conserver un environnement réutilisable et partagé.

Importer les bibliothèques nécessaires

from langgraph.graph import StateGraph, START, END
from typing import TypedDict
from IPython.display import Image

• StateGraph construit le flux de travail
• START / END indiquent le point d’entrée et de sortie
• TypedDict définit les types de l’état partagé
• Image affiche un diagramme du graphe compilé

Exemple 1 — Flux de travail séquentiel simple

Démarrez sans LLM afin que le chemin des données soit clair.

Définir l’état

class BMIState(TypedDict):
    weight_kg: float
    height_m: float
    bmi: float
    category: str

L’état est un dictionnaire partagé qui suit l’exécution. Les nœuds le lisent, écrivent des champs et retournent la mise à jour du mappage. Cet exemple prend en compte le poids, la hauteur, le IMC et une étiquette de catégorie.

Créer le nœud de traitement

# define calculate_bmi node
def calculate_bmi(state: BMIState) -> BMIState:

    weight = state['weight_kg']
    height = state['height_m']

    bmi = weight/(height**2)

    state['bmi'] = round(bmi, 2)

    return state

# define label_bmi node
def label_bmi(state: BMIState) -> BMIState:
    bmi = state['bmi']

    if bmi < 18.5:
        state["category"] = "Underweight"
    elif 18.5 <= bmi < 25:
        state["category"] = "Normal"
    elif 25 <= bmi < 30:
        state["category"] = "Overweight"
    else:
        state["category"] = "Obese"

    return state

La fonction lit la taille et le poids, calcule l’IMC, attribue une catégorie et renvoie l’état mis à jour. La logique de LangGraph est simple : accepter l’état actuel et renvoyer l’état suivant.

Créer le graphe

# define graph
graph = StateGraph(BMIState)

# add nodes to your graph
graph.add_node('calculate_bmi', calculate_bmi)
graph.add_node('label_bmi', label_bmi)

# add edges to your graph
graph.add_edge(START, 'calculate_bmi')
graph.add_edge('calculate_bmi', 'label_bmi')
graph.add_edge('label_bmi', END)


# compile the graph
workflow = graph.compile()

# show the graph
Image(workflow.get_graph().draw_mermaid_png())

Enregistrer le nœud, relier START → nœud → END, puis compiler. LangGraph peut dessiner le graphe résultant afin que vous puissiez vérifier sa topologie avant de l’utiliser.

Exécuter le flux de travail

# execute the graph
intial_state = {'weight_kg':80, 'height_m':1.73}
final_state = workflow.invoke(intial_state)
print(final_state)

Fournir un dictionnaire initial contenant le poids et la taille. La fonction invoke exécute le parcours défini et renvoie un état qui inclut désormais l’IMC ainsi que la catégorie.

Exemple 2 — Flux de travail séquentiel avec un LLM

Remplacer le nœud mathématique par une appel à un modèle. Conserver la même structure de graphe.

Récupérer de manière sécurisée une clé API Groq (via demande utilisateur, et non en code fixe), puis charger un modèle ouvert accessible :

model = ChatGroq(model="openai/gpt-oss-120b")

Définir l’état du LLM

# create a state
class LLMState(TypedDict):
    question: str
    answer: str

Les champs se réduisent à une question et une réponse : la structure de l’état suit le flux de travail, et non un schéma global.

Créer le nœud LLM

def llm_qa(state: LLMState) -> LLMState:

    # extract the question from state
    question = state['question']

    # form a prompt
    prompt = f'Answer the following question {question}'

    # ask that question to the LLM
    answer = model.invoke(prompt).content

    # update the answer in the state
    state['answer'] = answer

    return state

Lire la question, appeler le modèle de chat, stocker la réponse, renvoyer l’état. Structuralement identique au nœud BMI ; seules les étapes internes changent.

Construire le flux de travail LLM

# create our graph
graph = StateGraph(LLMState)

# add nodes
graph.add_node('llm_qa', llm_qa)

# add edges
graph.add_edge(START, 'llm_qa')
graph.add_edge('llm_qa', END)

# compile
workflow = graph.compile()

Ajouter le nœud, relier les arêtes, compiler. Le squelette reste intentionnellement familier.

Poser une question à l’LLM

# execute
intial_state = {'question': 'How far is sun from the earth?'}
final_state = workflow.invoke(intial_state)
print(final_state['answer'])

Appeler le modèle avec une chaîne de question et afficher le champ de réponse. Le modèle est intégré au graphe comme n’importe quelle autre étape.

Exemple 3 — Flux de travail d’IA à plusieurs étapes

Les systèmes réels enchaînent plusieurs phases de raisonnement. Étendez ce modèle.

Définir l’état du blog

class BlogState(TypedDict):
    title: str
    outline: str
    content: str
    rating: int

Suivre le titre, l’aperçu, le contenu ainsi qu’une note de qualité au fur et à mesure que le processus se développe.

Créer des nœuds de flux de travail

def create_outline(state: BlogState) -> BlogState:
    # fetch title
    title = state['title']

    # call llm gen outline
    prompt = f'Generate a detailed outline for a blog on the topic - {title}'
    outline = model.invoke(prompt).content

    # update state
    state['outline'] = outline

    return state

def create_blog(state: BlogState) -> BlogState:
    title = state['title']
    outline = state['outline']

    prompt = f'Write a short blog (less than 20 lines) on the title - {title} using the follwing outline \n {outline}'
    content = model.invoke(prompt).content
    state['content'] = content
    return state

def create_rating(state: BlogState) -> BlogState:
    title = state["title"]
    content = state["content"]

    prompt = f"""
    Rate the following blog out of 10.
    Return ONLY a single integer between 1 and 10.

    Title: {title}

    Blog:
    {content}
    """

    rating = model.invoke(prompt).content.strip()

    state["rating"] = int(rating)

    return state

Les fonctions de génération d’aperçu et les étapes ultérieures sont séparées. Chaque nœud a une seule responsabilité afin que les graphes restent lisible.

Assemblez le pipeline complet

graph = StateGraph(BlogState)

# nodes
graph.add_node('create_outline', create_outline)
graph.add_node('create_blog', create_blog)
graph.add_node('create_rating', create_rating)

# edges
graph.add_edge(START, 'create_outline')
graph.add_edge('create_outline', 'create_blog')
graph.add_edge('create_blog', 'create_rating')
graph.add_edge('create_rating', END)

workflow = graph.compile()
workflow

Connectez les tâches d’IA dans l’ordre. Conceptuellement :

Title
   ↓
Create Outline
   ↓
Generate Blog
   ↓
Evaluate Blog
   ↓
END

Décomposer le travail en étapes est préférable à l’insertion de toutes les instructions dans un seul prompt massif : chaque étape conditionne la suivante.

Exécutez le flux de travail

intial_state = {'title': 'Rise of AI in India'}
final_state = workflow.invoke(intial_state)
print(final_state['rating'])

En partant d’un titre, l’exécution crée un aperçu, rédige du contenu, évalue la qualité et renvoie le champ de notation. C’est là l’ossature des pipelines agents : orchestrer plusieurs capacités de modèles derrière un état partagé.

Considérez chaque exemple comme une étape de calibration et non comme un code de démonstration à jeter. Une fois le graphique du BMI exécuté, modifiez les seuils de catégorie et vérifiez les mises à jour d’état retournées. Après l’exécution du graphique lié au LLM à une seule question, changez l’ID du modèle ou le modèle de prompt et observez uniquement la modification du comportement de ce nœud. Lorsque le pipeline du blog est exécuté, insérez un nœud supplémentaire entre l’aperçu et le brouillon — par exemple pour une « vérification du ton » — afin de comprendre comment les arêtes et les champs d’état évoluent ensemble.

À ce stade, la discipline liée à l’état partagé est plus importante que des prompts ingénieux. Préférez des champs petits et nommés (outline, content, rating) plutôt qu’un seul gros bloc messages tant que vous n’avez pas besoin de l’historique des conversations. Il est acceptable de retourner une mise à jour partielle omettant des clés non pertinentes si la sémantique de fusion du graphique le permet ; l’habitude importante à adopter est que les nœuds restent suffisamment purs pour être testés avec un dictionnaire manuellement créé.

La gestion des erreurs peut rester minimale dans Colab, tout en étant intentionnelle : si l’appel à Groq échoue, il faut afficher l’exception plutôt que de la masquer à l’intérieur du nœud. Les tutoriels initiaux qui cachent les échecs inculquent une mauvaise habitude pour les graphes de rappels d’outils ultérieurs, où un outil défaillant doit devenir une branche explicite.

Finalement, dessinez le graphique compilé chaque fois que vous ajoutez une arête. La confirmation visuelle permet de détecter plus rapidement les cycles accidentels et les liens END oubliés que la lecture des appels add_edge. Une fois que l’image correspond à l’histoire que vous souhaitez raconter, invoke devient une étape de vérification plutôt qu’une session de débogage.

Ce que ces exemples démontrent

Trois flux de travail progressifs couvrent le vocabulaire essentiel :

1. Logique séquentielle pure en Python sur un état partagé
2. Un seul nœud LLM au sein du même schéma de graphique

Une chaîne d’IA en plusieurs étapes comprenant plusieurs nœuds coopératifs

Persistance : état partagé, nœuds, arêtes, orchestration, intégration de LLM et raisonnement par étapes. Ces mêmes concepts servent ensuite à prendre en charge les outils, la mémoire, les branches et les agents collaboratifs.

Que vient ensuite ?

Les parties ultérieures d’une IA agente utilisant Python ajoutent généralement des arêtes conditionnelles, des branches parallèles, des points de validation humaine, des appels à des outils, RAG, des architectures multi-agents et un emballage adapté à la production — tout en visant toujours des configurations de modèles ouverts compatibles avec Colab.

Source complète

Le notebook Colab associé est publié sous le projet Agentic-AI sur GitHub (fichier Workflow_in_LangGraph.ipynb). Ouvrez ce notebook dans Colab, exécutez les cellules de haut en bas, puis créez vos propres nœuds pour vos tâches.