langgraph-designer

name

langgraph-designer

description

Conception de graphes d'agents avec LangGraph pour workflows complexes et stateful. Gestion d'état partagé, edges conditionnels, persistance et human-in-the-loop. Se déclenche avec "LangGraph", "graph agent", "state machine agent", "agent graph", "conditional edges", "checkpointer", "agent workflow stateful", "langgraph workflow", "graph workflow". Also triggers on "LangGraph graph", "stateful agent workflow", "LangGraph nodes".

LangGraph Designer — Graphes d'Agents Stateful

Quand utiliser ce skill

LangGraph est pertinent quand au moins une de ces conditions est vraie :

• Le workflow n'est pas linéaire (branches, cycles, retry)
• L'état doit persister entre plusieurs appels ou sessions
• Une validation humaine doit interrompre l'exécution
• Plusieurs agents spécialisés doivent se coordonner

Alternatives : LangChain LCEL pour les pipelines linéaires simples, CrewAI si tu veux une abstraction haut niveau sans gérer le state manuellement.

---

Workflow en étapes

1. Installation et imports

pip install langgraph>=0.3.0 langchain-openai>=0.2.0
# Persistance PostgreSQL (prod) :
pip install langgraph-checkpoint-postgres psycopg[binary]

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, BaseMessage
from typing import TypedDict, Annotated
import operator

2. Définir le State (TypedDict)

Le State est le schéma partagé entre tous les nœuds. Règle critique : toujours annoter les listes avec un reducer.

from langgraph.graph.message import add_messages  # reducer officiel pour messages

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]  # append, déduplique par id
    iteration_count: int        # compteur anti-boucle infinie
    final_answer: str | None

Critères de choix du reducer :

Besoin	Reducer
Accumuler des messages	add_messages
Accumuler une liste générique	operator.add
Remplacer la valeur	Aucun (défaut)
Valeur max/min	lambda a, b: max(a, b)

Raccourci pour chatbots purs : from langgraph.graph import MessagesState (hérite déjà de add_messages).

3. Créer les nœuds

Un nœud est une fonction pure : reçoit le state complet, retourne un dict partiel (seules les clés modifiées).

llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)

def agent_node(state: AgentState) -> dict:
    response = llm_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "iteration_count": state["iteration_count"] + 1,
    }

# ToolNode gère automatiquement l'exécution des tool_calls
tool_node = ToolNode(tools)

4. Construire le graphe et les edges

builder = StateGraph(AgentState)

# Ajout des nœuds
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)

# Edges
builder.add_edge(START, "agent")

# Conditional edge : tools_condition vérifie si des tool_calls sont présents
builder.add_conditional_edges(
    "agent",
    tools_condition,           # retourne "tools" ou END
)
builder.add_edge("tools", "agent")  # cycle : retour à l'agent

Conditional edge personnalisé :

MAX_ITER = 10

def router(state: AgentState) -> str:
    if state["iteration_count"] >= MAX_ITER:
        return END                          # garde-fou anti-boucle
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

builder.add_conditional_edges(
    "agent",
    router,
    {"tools": "tools", END: END},  # mapping explicite (optionnel mais lisible)
)

5. Compiler avec checkpointer

# Dev / tests
memory = MemorySaver()
app = builder.compile(checkpointer=memory)

# SQLite (local, persist entre process)
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("app.db") as cp:
    app = builder.compile(checkpointer=cp)

# PostgreSQL (production)
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string(os.environ["DATABASE_URL"]) as cp:
    cp.setup()   # crée les tables si besoin
    app = builder.compile(checkpointer=cp)

Chaque run utilise un thread_id pour isoler les sessions :

config = {"configurable": {"thread_id": "user-abc-session-1"}}
result = app.invoke({"messages": [HumanMessage(content="...")], "iteration_count": 0}, config=config)

6. Human-in-the-loop (HIL)

# Interrompre avant un nœud critique
app = builder.compile(checkpointer=memory, interrupt_before=["executor"])

# Lancer jusqu'à l'interruption
app.invoke(initial_state, config=config)

# Inspecter et modifier le state
current = app.get_state(config)
print(current.values["plan"])

# Corriger si besoin
app.update_state(config, {"plan": ["étape 1 modifiée", "étape 2"]})

# Reprendre (None = continuer depuis le point d'interruption)
final = app.invoke(None, config=config)

7. Streaming

# Token par token (UIs chat)
async for chunk, metadata in app.astream(inputs, config=config, stream_mode="messages"):
    if metadata.get("langgraph_node") == "agent":
        print(chunk.content, end="", flush=True)

# Deltas d'état (progression)
for update in app.stream(inputs, config=config, stream_mode="updates"):
    node_name, state_delta = next(iter(update.items()))
    print(f"[{node_name}] {state_delta}")

8. Sub-graphs et Multi-agent

# Compiler un sous-graphe et l'utiliser comme nœud
sub_app = sub_builder.compile()
main_builder.add_node("specialist", sub_app)

# Pattern Supervisor : router vers agents spécialisés
def supervisor_router(state) -> str:
    # Le LLM décide quel agent appeler
    return state["next_agent"]  # "researcher" | "coder" | END

main_builder.add_conditional_edges("supervisor", supervisor_router)

---

Exemples complets

Agent ReAct minimal avec persistance

import os
from typing import TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    iteration_count: int

@tool
def calculator(expression: str) -> str:
    """Évalue une expression mathématique Python sûre."""
    try:
        return str(eval(expression, {"__builtins__": {}}))
    except Exception as e:
        return f"Erreur: {e}"

tools = [calculator]
llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)

MAX_ITER = 8

def agent(state: State) -> dict:
    return {: [llm.invoke(state[])], : state[] + }

 () -> :
     state[] >= MAX_ITER:
         END
     tools_condition(state)

builder = StateGraph(State)
builder.add_node(, agent)
builder.add_node(, ToolNode(tools))
builder.add_edge(START, )
builder.add_conditional_edges(, should_continue)
builder.add_edge(, )

app = builder.(checkpointer=MemorySaver())

config = {: {: }}
r = app.invoke({: [HumanMessage()], : }, config)
(r[][-].content)

Visualiser le graphe (debug local)

# Générer un PNG du graphe (nécessite pygraphviz ou pillow)
from IPython.display import Image
Image(app.get_graph().draw_mermaid_png())

# Afficher en Mermaid (texte, pas de dépendances)
print(app.get_graph().draw_mermaid())

---

Garde-fous / Anti-patterns / Pièges

Reducer manquant sur les listes — Sans Annotated[list, add_messages], chaque nœud remplace la liste entière. Résultat : l'historique de messages disparaît après le premier nœud. Toujours annoter.

Boucle infinie — Un cycle agent → tools → agent sans condition de sortie tourne indéfiniment si le LLM appelle toujours un tool. Ajouter iteration_count + seuil dans chaque router conditionnel.

Réutilisation du thread_id — Partager un thread_id entre utilisateurs différents mélange les states. Générer un UUID par session utilisateur : thread_id = str(uuid.uuid4()).

Muter le state directement — Ne jamais faire state["messages"].append(...) dans un nœud. Toujours retourner un dict partiel : return {"messages": [new_msg]}.

MemorySaver en production — MemorySaver est in-process et non partageable. Utiliser PostgresSaver (ou Redis via un plugin custom) dès qu'il y a plusieurs workers ou que la persistance doit survivre aux redémarrages.

Oublier cp.setup() — Avec PostgresSaver, appeler .setup() une fois avant de compiler pour créer les tables de checkpoint. Sans ça, le premier invoke lève une exception.

Sub-graph sans checkpointer partagé — Un sous-graphe compilé sans checkpointer ne persiste pas son état entre les appels. Passer le checkpointer du graphe parent si la persistance est nécessaire dans le sous-graphe.

---

Bonnes pratiques 2026

• interrupt_before plutôt que interrupt_after : interrompre avant l'action critique (écriture DB, envoi email) permet de valider et d'annuler sans effet de bord.
• Structured output pour le routing : utiliser llm.with_structured_output(RouteSchema) dans les nœuds de décision plutôt que de parser du texte libre — évite les erreurs de parsing.
• LangSmith tracing : activer LANGCHAIN_TRACING_V2=true + LANGCHAIN_API_KEY dès le développement pour déboguer les runs complexes avec une UI dédiée.
• LangGraph Studio : outil de debug local indispensable — visualise le graphe, inspecte le state nœud par nœud, rejoue des runs. Lancer avec langgraph dev (nécessite langgraph.json).
• Tester les nœuds isolément : chaque nœud étant une fonction pure, les unit tests sont directs — passer un state dict en entrée, asserter le dict de sortie.
• Versionner le schéma du State : si le schéma évolue, les checkpoints persistés peuvent être incompatibles. Prévoir une migration ou inclure un champ schema_version dans le State.