Blog Agents IA
SYSTÈME D'ACCÈS INFORMATIQUE

Machines d'États et Agents : Construire des Workflows Fiables avec LangGraph

agentsarchitecturetutoriallanggraphautonomous-systems

La plupart des tutoriels sur les agents montrent une boucle simple : demander à Claude, analyser la réponse, appeler un outil, recommencer. Cela fonctionne pour les démonstrations. En production, vous avez besoin de déterminisme, de récupération sur erreur, de portes d’approbation humaine et d’auditabilité.

LangGraph apporte les machines d’états aux workflows d’agents. Au lieu d’une boucle ad-hoc maintenue par des instructions if, vous obtenez un graphe explicite : des nœuds nommés (unités de logique), des arêtes typées (transitions) et un schéma d’état partagé qui traverse toute l’exécution.

Pourquoi des Machines d’États pour les Agents ?

Le Problème de la Boucle Ad-Hoc

Une boucle d’agent typique ressemble à ceci :

messages = []
while True:
    response = claude.messages.create(model=..., messages=messages)
    if response.stop_reason == "end_turn":
        break
    for tool_call in get_tool_calls(response):
        result = execute_tool(tool_call)
        messages.append(tool_result(tool_call.id, result))

C’est lisible pour deux ou trois outils. Ajoutez cinq outils, des chemins conditionnels, une étape d’approbation humaine et de la logique de réessai — vous avez des centaines de lignes de flux de contrôle emmêlé.

Le problème fondamental est l’état implicite. Dans quelle étape est l’agent ? Quelles données a-t-il collectées ? Tout vit dans messages — un blob non typé que chaque nœud lit et complète, sans schéma imposé.

Les Machines d’États comme Solution

Une machine d’états rend l’implicite explicite. Vous définissez :

  • Nœuds — unités de logique discrètes. Chaque nœud reçoit l’état courant, fait une chose et renvoie des mises à jour d’état.
  • Arêtes — transitions entre nœuds, soit inconditionnelles (A → B toujours) soit conditionnelles (si problèmes trouvés : aller à révision, sinon : aller à résumé).
  • État — un dictionnaire typé qui traverse le graphe complet. Le schéma est validé à chaque étape.

Quand NE PAS utiliser LangGraph

LangGraph ajoute des frais généraux. Pour les tâches simples en une seule passe, un appel API direct est plus rapide et plus clair. Utilisez LangGraph quand votre workflow a :

  • Plusieurs étapes distinctes devant s’exécuter en séquence
  • Des branches conditionnelles basées sur des résultats intermédiaires
  • Des étapes avec participation humaine
  • De la logique de récupération d’erreurs ou de réessai
  • Des exigences d’auditabilité

Fondamentaux de LangGraph

Définir l’État

from typing import TypedDict


class ResearchState(TypedDict):
    query: str
    research_notes: str
    draft: str
    review_feedback: str
    is_approved: bool

Les nœuds renvoient des dictionnaires partiels. LangGraph fusionne ces mises à jour dans l’état courant après chaque nœud.

Nœuds

import anthropic


client = anthropic.Anthropic()


def research_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Recherche ce sujet en profondeur : {state['query']}"
        }]
    )
    return {"research_notes": response.content[0].text}


def draft_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Rédige un brouillon basé sur ces notes :\n{state['research_notes']}"
        }]
    )
    return {"draft": response.content[0].text}

Arêtes et Routage Conditionnel

from typing import Literal


def route_after_check(state: ResearchState) -> Literal["human_review", "draft"]:
    if state.get("review_feedback"):
        return "human_review"
    return "draft"


workflow.add_conditional_edges(
    "research",
    route_after_check,
    {"human_review": "review_node", "draft": "draft_node"}
)

Construire et Exécuter un Graphe

from langgraph.graph import StateGraph, END


workflow = StateGraph(ResearchState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", END)


graph = workflow.compile()
result = graph.invoke({"query": "Qu'est-ce que le protocole MCP ?"})
print(result["draft"])

Construire un Workflow de Révision de Documents

Concevoir l’État

class DocumentState(TypedDict):
    document: str
    key_terms: list[str]
    compliance_issues: list[str]
    summary: str
    human_feedback: str
    is_approved: bool

Implémenter les Nœuds

import json


def extract_terms_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "Extrais exactement 5 termes clés de ce document. "
                "Renvoie-les sous forme de tableau JSON de chaînes.\n\n"
                f"Document : {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        terms = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        terms = [t.strip() for t in response.content[0].text.split("\n") if t.strip()][:5]
    return {"key_terms": terms}




def check_compliance_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                "Vérifie ce document pour des problèmes de conformité. "
                "Cherche : DCP (noms, numéros de sécurité sociale, emails, téléphones), "
                "marquages confidentiels, affirmations non vérifiées.\n\n"
                "Renvoie un tableau JSON de descriptions de problèmes. "
                "Renvoie un tableau vide [] si aucun problème n'est trouvé.\n\n"
                f"Document : {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        issues = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        issues = []
    return {"compliance_issues": issues}




def human_review_node(state: DocumentState) -> dict:
    print("\n=== RÉVISION HUMAINE REQUISE ===")
    for issue in state["compliance_issues"]:
        print(f"  - {issue}")
    return {
        "human_feedback": "Révisé et approuvé avec expurgation des DCP",
        "is_approved": True
    }




def summarize_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"Résume ce document en exactement deux phrases :\n\n{state['document']}"
        }]
    )
    return {"summary": response.content[0].text, "is_approved": True}

Assembler le Graphe

def route_after_compliance(state: DocumentState) -> Literal["human_review", "summarize"]:
    if state.get("compliance_issues"):
        return "human_review"
    return "summarize"


workflow = StateGraph(DocumentState)
workflow.add_node("extract", extract_terms_node)
workflow.add_node("check", check_compliance_node)
workflow.add_node("review", human_review_node)
workflow.add_node("summarize", summarize_node)


workflow.set_entry_point("extract")
workflow.add_edge("extract", "check")
workflow.add_conditional_edges(
    "check", route_after_compliance,
    {"human_review": "review", "summarize": "summarize"}
)
workflow.add_edge("review", "summarize")
workflow.add_edge("summarize", END)


graph = workflow.compile()


result = graph.invoke({
    "document": "Le patient Jean Dupont (SS : 1 80 02 75 108 111 84) souffre d'hypertension.",
    "key_terms": [], "compliance_issues": [],
    "summary": "", "human_feedback": "", "is_approved": False,
})
print(result["summary"])

Points de Contrôle Humains et Interruptions

from langgraph.checkpoint.memory import MemorySaver


checkpointer = MemorySaver()
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)


thread_config = {"configurable": {"thread_id": "doc-review-001"}}
result = graph.invoke(initial_state, config=thread_config)


current_state = graph.get_state(thread_config)
print("Problèmes :", current_state.values["compliance_issues"])


graph.update_state(
    thread_config,
    {"human_feedback": "Approuvé après expurgation manuelle des DCP", "is_approved": True}
)


final_result = graph.invoke(None, config=thread_config)

Persistance et Déploiement en Production

Point de Contrôle d’État

from langgraph.checkpoint.postgres import PostgresSaver


with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    graph = workflow.compile(checkpointer=checkpointer)
    result = graph.invoke(initial_state, config={"configurable": {"thread_id": "doc-001"}})

Exécution Asynchrone

import asyncio


async def process_document(doc: str) -> dict:
    return await graph.ainvoke({
        "document": doc, "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    })


async def batch_process(documents: list[str]) -> list[dict]:
    return await asyncio.gather(*[process_document(doc) for doc in documents])

Modèles Courants et Erreurs à Éviter

  • Fan-Out / Fan-In : Exécutez plusieurs nœuds en parallèle et fusionnez leurs résultats avec Send.
  • Routage conditionnel trop complexe : Si une fonction de routage a trop de branches, divisez en sous-graphes.
  • Surcharge de l’état : Stockez des références (IDs de base de données, clés S3) plutôt que de grands artefacts.
  • Tests des nœuds en isolation : Les nœuds étant de simples fonctions, ils peuvent être testés unitairement directement.
def test_extract_terms_node():
    state = {
        "document": "Le renard brun rapide saute par-dessus le chien paresseux.",
        "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    }
    result = extract_terms_node(state)
    assert "key_terms" in result
    assert isinstance(result["key_terms"], list)

LangGraph remplace les boucles ad-hoc par des workflows explicites, déboguables et reprenables. Commencez par un graphe linéaire (A → B → C → FIN). Ajoutez une arête conditionnelle quand vous avez besoin de bifurquer. Ajoutez un point de contrôle humain quand vous avez besoin d’approbation. La plupart des workflows d’agents de production n’ont besoin que de ces trois modèles.

Articles Connexes