- LangGraph : orchestrer des agents IA multi-étapes

Pourquoi LangGraph plutôt qu'une boucle agent maison

Un agent LLM minimal tient en 30 lignes : boucle while, appel LLM avec tools, exécution des tool_calls, retour au modèle. Tant que l'agent fait une seule chose simple, ça suffit. Mais dès qu'il faut brancher, reprendre après crash, demander une validation humaine ou observer ce que l'agent fait en production, le code maison devient vite ingérable.

LangGraph modélise l'agent comme un graphe d'états dirigé. Chaque node est une étape (appel LLM, exécution d'outil, transformation), chaque edge décide du node suivant (statique ou conditionnel), et le state est sérialisé à chaque transition. Vous gagnez sept choses gratuitement :

Installation et premier graphe minimal

LangGraph.js est packagé séparément de LangChain mais partage les abstractions de modèles et d'outils. On installe le cœur, un modèle (ici OpenAI) et le checkpointer SQLite pour la suite.

# Installation des packages essentiels
npm install @langchain/langgraph @langchain/core
npm install @langchain/openai          # provider LLM
npm install @langchain/langgraph-checkpoint-sqlite  # checkpointer
npm install zod                         # schemas d'outils

// hello-graph.ts - Plus petit graphe possible : entree -> LLM -> sortie
import { StateGraph, Annotation, START, END } from '@langchain/langgraph';
import { ChatOpenAI } from '@langchain/openai';
import { HumanMessage, AIMessage, BaseMessage } from '@langchain/core/messages';

// 1. State : un seul champ, la liste de messages
const State = Annotation.Root({
    messages: Annotation<BaseMessage[]>({
        // Reducteur : concatene les messages produits par chaque node
        reducer: (current, update) => current.concat(update),
        default: () => []
    })
});

// 2. Un node = une fonction qui lit le state et retourne un partial state
const llm = new ChatOpenAI({ model: 'gpt-4o-mini', temperature: 0 });

async function callLlm(state: typeof State.State) {
    const response = await llm.invoke(state.messages);
    return { messages: [response] };   // sera concatene via le reducteur
}

// 3. Construction du graphe : nodes + edges
const graph = new StateGraph(State)
    .addNode('llm', callLlm)
    .addEdge(START, 'llm')      // entree -> llm
    .addEdge('llm', END)         // llm -> sortie
    .compile();

// 4. Execution
const result = await graph.invoke({
    messages: [new HumanMessage('Explique Angular Signals en une phrase.')]
});

console.log(result.messages.at(-1)?.content);

State annotations et réducteurs typés

Le state est le contrat entre nodes. Chaque champ a un type et un réducteur qui décrit comment combiner la valeur actuelle avec celle retournée par un node. Bien choisir le réducteur évite 80 % des bugs.

// state-patterns.ts - Differents reducteurs pour differents cas
import { Annotation, messagesStateReducer } from '@langchain/langgraph';
import type { BaseMessage } from '@langchain/core/messages';

const AgentState = Annotation.Root({
    // 1. Liste de messages - reducteur officiel qui gere les ids et updates
    messages: Annotation<BaseMessage[]>({
        reducer: messagesStateReducer,
        default: () => []
    }),

    // 2. Champ scalaire - le dernier ecrit gagne (par defaut)
    intent: Annotation<'search' | 'create' | 'support' | null>({
        reducer: (_old, next) => next,    // remplacement simple
        default: () => null
    }),

    // 3. Compteur - accumulation
    iterations: Annotation<number>({
        reducer: (old, next) => old + next,
        default: () => 0
    }),

    // 4. Liste de resultats partiels - concatenation
    sources: Annotation<string[]>({
        reducer: (old, next) => [...old, ...next],
        default: () => []
    }),

    // 5. Map de scores - merge
    scores: Annotation<Record<string, number>>({
        reducer: (old, next) => ({ ...old, ...next }),
        default: () => ({})
    })
});

// Un node retourne UNIQUEMENT les champs qu'il modifie
async function analyzeIntent(state: typeof AgentState.State) {
    const intent = await classifyMessage(state.messages.at(-1));
    return {
        intent,                      // remplacement
        iterations: 1                // +1 via le reducteur
    };
}

Edges conditionnels : routing intelligent

Le vrai pouvoir de LangGraph : un node peut décider du prochain node en fonction du state. C'est ce qui permet de construire des agents qui choisissent leur stratégie au lieu de suivre une chaîne fixe.

// router-agent.ts - Routing par intent classifie par LLM
import { StateGraph, Annotation, START, END } from '@langchain/langgraph';
import { ChatOpenAI } from '@langchain/openai';
import { HumanMessage, BaseMessage } from '@langchain/core/messages';

type Intent = 'search' | 'create' | 'support' | 'unknown';

const State = Annotation.Root({
    messages: Annotation<BaseMessage[]>({
        reducer: (a, b) => a.concat(b),
        default: () => []
    }),
    intent: Annotation<Intent>({
        reducer: (_old, next) => next,
        default: () => 'unknown'
    })
});

const llm = new ChatOpenAI({ model: 'gpt-4o-mini' });

// Node 1 : classification d'intent
async function classifyIntent(state: typeof State.State): Promise<Partial<typeof State.State>> {
    const last = state.messages.at(-1)?.content as string;
    const response = await llm.invoke([
        new HumanMessage(`Classe cette demande en : search | create | support.
Reponds UNIQUEMENT par le mot.
Demande : ${last}`)
    ]);
    const intent = String(response.content).trim().toLowerCase() as Intent;
    return { intent: ['search', 'create', 'support'].includes(intent) ? intent : 'unknown' };
}

// Nodes specialises - implementations simplifiees
async function searchHandler(_s: typeof State.State) {
    return { messages: [{ role: 'assistant', content: 'Recherche en cours...' } as any] };
}
async function createHandler(_s: typeof State.State) {
    return { messages: [{ role: 'assistant', content: 'Creation lancee.' } as any] };
}
async function supportHandler(_s: typeof State.State) {
    return { messages: [{ role: 'assistant', content: 'Ouverture d\'un ticket support.' } as any] };
}
async function fallback(_s: typeof State.State) {
    return { messages: [{ role: 'assistant', content: 'Demande non comprise.' } as any] };
}

// Edge conditionnel : selectionne le node suivant selon le state
function routeByIntent(state: typeof State.State): string {
    switch (state.intent) {
        case 'search':  return 'search_node';
        case 'create':  return 'create_node';
        case 'support': return 'support_node';
        default:        return 'fallback_node';
    }
}

const graph = new StateGraph(State)
    .addNode('classify', classifyIntent)
    .addNode('search_node', searchHandler)
    .addNode('create_node', createHandler)
    .addNode('support_node', supportHandler)
    .addNode('fallback_node', fallback)
    .addEdge(START, 'classify')
    // L'edge conditionnel decide dynamiquement
    .addConditionalEdges('classify', routeByIntent, {
        search_node: 'search_node',
        create_node: 'create_node',
        support_node: 'support_node',
        fallback_node: 'fallback_node'
    })
    .addEdge('search_node', END)
    .addEdge('create_node', END)
    .addEdge('support_node', END)
    .addEdge('fallback_node', END)
    .compile();

Tool calling intégré au graphe

LangGraph fournit un ToolNode qui exécute automatiquement les tool_calls retournés par le modèle. Le pattern ReAct standard se résume à deux nodes : agent (LLM) et tools (exécution), reliés par une boucle conditionnelle.

// react-agent.ts - Pattern ReAct minimal avec ToolNode
import { StateGraph, Annotation, START, END, messagesStateReducer } from '@langchain/langgraph';
import { ToolNode } from '@langchain/langgraph/prebuilt';
import { ChatOpenAI } from '@langchain/openai';
import { tool } from '@langchain/core/tools';
import type { BaseMessage, AIMessage } from '@langchain/core/messages';
import { z } from 'zod';

// 1. Definition des outils metiers
const searchArticles = tool(
    async ({ query, limit }) => {
        const results = await db.search(query, limit);
        return JSON.stringify(results);
    },
    {
        name: 'search_articles',
        description: 'Recherche d\'articles par mot-cle.',
        schema: z.object({
            query: z.string(),
            limit: z.number().int().min(1).max(10)
        })
    }
);

const getUserProfile = tool(
    async ({ userId }) => JSON.stringify(await db.users.find(userId)),
    {
        name: 'get_user_profile',
        description: 'Recupere le profil complet d\'un utilisateur.',
        schema: z.object({ userId: z.string().uuid() })
    }
);

const TOOLS = [searchArticles, getUserProfile];

// 2. State standard
const State = Annotation.Root({
    messages: Annotation<BaseMessage[]>({
        reducer: messagesStateReducer,
        default: () => []
    })
});

// 3. LLM avec outils binds (tool calling natif)
const model = new ChatOpenAI({ model: 'gpt-4o-2024-08-06' }).bindTools(TOOLS);

// 4. Node "agent" : appel LLM
async function agentNode(state: typeof State.State) {
    const response = await model.invoke(state.messages);
    return { messages: [response] };
}

// 5. Routage : si le LLM appelle des outils, on va a "tools", sinon fin
function shouldContinue(state: typeof State.State): 'tools' | typeof END {
    const last = state.messages.at(-1) as AIMessage;
    return last.tool_calls?.length ? 'tools' : END;
}

// 6. ToolNode = wrapper qui execute automatiquement les tool_calls
const toolNode = new ToolNode(TOOLS);

// 7. Composition : agent -> (tools -> agent) -> END
const graph = new StateGraph(State)
    .addNode('agent', agentNode)
    .addNode('tools', toolNode)
    .addEdge(START, 'agent')
    .addConditionalEdges('agent', shouldContinue, { tools: 'tools', [END]: END })
    .addEdge('tools', 'agent')   // apres execution, retour au LLM
    .compile();

Checkpointing et persistance SQLite

Le checkpointer sérialise le state à chaque transition entre nodes. C'est ce qui permet de reprendre une conversation après un redémarrage serveur, de paralléliser plusieurs threads conversationnels, et d'inspecter l'historique d'exécution.

// checkpointed-graph.ts - Persistance avec SQLite
import { SqliteSaver } from '@langchain/langgraph-checkpoint-sqlite';

// 1. Checkpointer SQLite - un fichier sur disque
const checkpointer = SqliteSaver.fromConnString('./threads.sqlite');

// 2. Recompile le graphe avec le checkpointer
const graph = new StateGraph(State)
    .addNode('agent', agentNode)
    .addNode('tools', toolNode)
    .addEdge(START, 'agent')
    .addConditionalEdges('agent', shouldContinue, { tools: 'tools', [END]: END })
    .addEdge('tools', 'agent')
    .compile({ checkpointer });

// 3. Chaque conversation est identifiee par un thread_id
const config = { configurable: { thread_id: 'user-abc-conv-42' } };

// Premier message
await graph.invoke(
    { messages: [new HumanMessage('Trouve des articles sur Angular Signals.')] },
    config
);

// ... le serveur redemarre, la connexion utilisateur est perdue ...

// Deuxieme message dans la MEME conversation - le state est restaure
await graph.invoke(
    { messages: [new HumanMessage('Le premier resultat m\'interesse, dis-m\'en plus.')] },
    config
);

// Inspecter l'historique complet
const history = [];
for await (const ckpt of graph.getStateHistory(config)) {
    history.push({ step: ckpt.metadata?.step, values: ckpt.values });
}
console.log(history);

Human-in-the-loop : interrupt et resume

Certaines actions exigent une validation humaine : envoi d'email, paiement, modification destructive. LangGraph propose interrupt() pour suspendre l'exécution au milieu d'un node et attendre une réponse externe.

// human-approval.ts - Interrupt avant un tool destructif
import { interrupt, Command } from '@langchain/langgraph';

// Node "verifier" qui demande une validation avant d'executer
async function humanApprovalNode(state: typeof State.State) {
    const lastMessage = state.messages.at(-1) as AIMessage;
    const toolCall = lastMessage.tool_calls?.[0];

    if (!toolCall) return { messages: [] };

    // Liste des outils qui exigent une validation
    const SENSITIVE = ['send_email', 'delete_user', 'create_invoice'];
    if (!SENSITIVE.includes(toolCall.name)) return { messages: [] };

    // PAUSE l'execution - le state est checkpointe automatiquement
    const decision = interrupt({
        type: 'approval_required',
        tool: toolCall.name,
        args: toolCall.args,
        message: `L'agent demande a executer ${toolCall.name}. Approuver ?`
    });

    // Cette ligne ne s'execute que lors du resume
    if (decision === 'approved') {
        return { messages: [] };  // continue vers ToolNode
    } else {
        return {
            messages: [{
                role: 'tool',
                tool_call_id: toolCall.id,
                content: JSON.stringify({ error: 'Action refusee par l\'utilisateur.' })
            }]
        };
    }
}

// API Express : premier appel - peut se mettre en pause
app.post('/agent/start', async (req, res) => {
    const config = { configurable: { thread_id: req.body.threadId } };
    const result = await graph.invoke({ messages: req.body.messages }, config);

    const snapshot = await graph.getState(config);
    if (snapshot.next.length > 0) {
        // Le graphe est en pause sur un interrupt
        const pendingInterrupt = snapshot.tasks
            .flatMap(t => t.interrupts)
            .find(i => i.value?.type === 'approval_required');
        res.json({ status: 'pending_approval', interrupt: pendingInterrupt });
    } else {
        res.json({ status: 'done', result });
    }
});

// API Express : resume apres validation humaine
app.post('/agent/resume', async (req, res) => {
    const config = { configurable: { thread_id: req.body.threadId } };
    // Command(resume) passe la valeur a la fonction interrupt() qui attendait
    const result = await graph.invoke(
        new Command({ resume: req.body.decision }),   // 'approved' ou 'denied'
        config
    );
    res.json({ status: 'done', result });
});

Streaming par node et observabilité

LangGraph expose graph.stream() qui émet un événement par node terminé, et graph.streamEvents() qui émet aussi les tokens du LLM. Combiné à LangSmith, vous obtenez une trace complète sans effort.

// streaming-graph.ts - Stream pour SSE Express
app.post('/agent/stream', async (req, res) => {
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.flushHeaders();

    const config = { configurable: { thread_id: req.body.threadId } };
    const input = { messages: req.body.messages };

    // Mode 'updates' : un evenement par node, contenant le delta de state
    for await (const chunk of await graph.stream(input, { ...config, streamMode: 'updates' })) {
        // chunk = { [nodeName]: partialState }
        const [nodeName, partial] = Object.entries(chunk)[0];
        res.write(`event: node\ndata: ${JSON.stringify({ nodeName, partial })}\n\n`);
    }

    // Mode 'messages' : tokens individuels du LLM en streaming
    for await (const [token, meta] of await graph.stream(input, { ...config, streamMode: 'messages' })) {
        if (typeof token.content === 'string') {
            res.write(`event: token\ndata: ${JSON.stringify({ delta: token.content })}\n\n`);
        }
    }

    res.write('event: done\ndata: {}\n\n');
    res.end();
});

Observabilité LangSmith

# Activation : trois variables d'environnement suffisent
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__xxxxxxxxxxxx
LANGCHAIN_PROJECT=mon-agent-prod

Avec ces variables, chaque exécution du graphe est tracée dans LangSmith : durée par node, tokens consommés, tool_calls, état complet à chaque transition, et erreurs. C'est l'équivalent d'un APM dédié aux agents LLM, indispensable dès qu'il faut diagnostiquer un comportement étrange en production.

Production : retry, timeouts, déploiement

Quelques règles pour passer du prototype à la production sans mauvaise surprise.

// production-graph.ts - Retry et timeout par node
import { RunnableConfig } from '@langchain/core/runnables';

// 1. Retry policy par node sensible
const graph = new StateGraph(State)
    .addNode('agent', agentNode, {
        retryPolicy: {
            maxAttempts: 3,
            initialInterval: 1000,
            backoffFactor: 2,
            retryOn: (err) => {
                // Retry sur 429 et 5xx, pas sur 4xx
                const status = (err as any)?.status;
                return status === 429 || (status >= 500 && status < 600);
            }
        }
    })
    .addNode('tools', toolNode)
    .addEdge(START, 'agent')
    .addConditionalEdges('agent', shouldContinue, { tools: 'tools', [END]: END })
    .addEdge('tools', 'agent')
    .compile({ checkpointer });

// 2. Timeout global par invocation
async function callWithDeadline(input: any, threadId: string, timeoutMs = 30_000) {
    const config: RunnableConfig = {
        configurable: { thread_id: threadId },
        recursionLimit: 25   // garde-fou anti-boucle
    };
    return Promise.race([
        graph.invoke(input, config),
        new Promise((_r, reject) => setTimeout(() => reject(new Error('Deadline')), timeoutMs))
    ]);
}

• recursionLimit : 25 par défaut, à abaisser à 10-15 pour la plupart des agents
• Retry policy exclusivement sur 429 et 5xx, jamais sur 4xx (échec applicatif)
• Timeout global via Promise.race — protège des appels LLM qui hangent
• Postgres checkpointer en prod, jamais MemorySaver ni SQLite à fort trafic
• Index DB sur thread_id + rétention 90 jours
• Quotas par utilisateur : nombre d'invocations / jour, tokens consommés / mois
• LangSmith activé dès le déploiement staging — la 1^re session de debug paie l'abonnement
• Tests unitaires des nodes en isolation, tests d'intégration sur le graphe complet

Conclusion

LangGraph transforme la construction d'agents LLM d'un art bricolé en discipline d'ingénierie : graphe d'états explicite, réducteurs typés, checkpointing automatique, human-in-the-loop natif et observabilité de premier ordre via LangSmith. Vous gardez le contrôle sur la logique métier tout en bénéficiant d'une infrastructure éprouvée.

Pour démarrer : commencez par un graphe à deux nodes (agent ReAct + ToolNode) avec SQLite checkpointer en local. Ajoutez le routing conditionnel quand votre agent doit choisir entre plusieurs stratégies, et n'introduisez interrupt() qu'au moment où vous identifiez une action métier qui exige une confirmation humaine. Migrez le checkpointer vers Postgres et activez LangSmith dès le déploiement staging — c'est ce qui rend l'application maintenable sur le long terme.