LangGraph : orchestrer des agents Claude multi-étapes

Ce que vous saurez faire à la fin

1. Comprendre le concept et les cas d’usage de LangGraph.
2. Construire votre premier workflow multi-étapes (StateGraph).
3. Orchestrer 3-5 agents Claude qui collaborent sur une tâche complexe.
4. Gérer état, mémoire, branchements conditionnels.
5. Déployer en production avec monitoring.

Durée : 1-2 jours pour premier workflow + extension continue. Pré-requis : Python 3.10+ ou JavaScript Node 20+, compte Anthropic API, connaissance LangChain (recommandé), Docker pour déploiement, base PostgreSQL pour persistence, monitoring (LangSmith ou custom).

Étape 1 — Comprendre LangGraph vs LangChain

Étape 2 — Installer LangGraph + dépendances

# Python
pip install langgraph langchain-anthropic
pip install langchain langsmith

# JavaScript
npm install @langchain/langgraph @langchain/anthropic
npm install @langchain/core langsmith

Étape 3 — Définir l’État (StateGraph)

from typing import TypedDict, List
from langgraph.graph import StateGraph, END

class WorkflowState(TypedDict):
    user_query: str
    research_results: List[dict]
    draft_article: str
    review_feedback: str
    final_article: str
    iteration_count: int

Étape 4 — Construire votre premier workflow simple

from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, END

llm = ChatAnthropic(model="claude-sonnet-4-6")

# Définir les noeuds (chaque noeud = 1 agent ou tool)
def research_node(state: WorkflowState):
    """Agent qui fait de la recherche"""
    query = state["user_query"]
    response = llm.invoke(f"Fais de la recherche sur : {query}")
    return {"research_results": [{"content": response.content}]}

def writer_node(state: WorkflowState):
    """Agent qui rédige"""
    research = state["research_results"][0]["content"]
    response = llm.invoke(f"Rédige un article basé sur : {research}")
    return {"draft_article": response.content}

def reviewer_node(state: WorkflowState):
    """Agent qui review"""
    article = state["draft_article"]
    response = llm.invoke(f"Critique et améliore : {article}")
    return {"final_article": response.content}

# Construire le graph
workflow = StateGraph(WorkflowState)
workflow.add_node("research", research_node)
workflow.add_node("write", writer_node)
workflow.add_node("review", reviewer_node)

# Définir les arêtes
workflow.set_entry_point("research")
workflow.add_edge("research", "write")
workflow.add_edge("write", "review")
workflow.add_edge("review", END)

app = workflow.compile()

Étape 5 — Exécuter le workflow

result = app.invoke({
    "user_query": "Stratégie pricing pour PME e-commerce sénégalaise",
    "research_results": [],
    "draft_article": "",
    "review_feedback": "",
    "final_article": "",
    "iteration_count": 0
})

print(result["final_article"])

# Visualiser le graph
print(app.get_graph().draw_mermaid())

Étape 6 — Ajouter des branchements conditionnels

def should_iterate(state: WorkflowState):
    """Décide si on itère ou si on termine"""
    if state["iteration_count"] < 3 and "amélioration nécessaire" in state.get("review_feedback", "").lower():
        return "rewrite"
    return "publish"

workflow.add_conditional_edges(
    "review",
    should_iterate,
    {
        "rewrite": "write",  # Boucle de retour
        "publish": END
    }
)

# Le graph peut maintenant boucler review → write → review
# jusqu'à 3 itérations

Étape 7 — Gérer la mémoire (checkpoints)

from langgraph.checkpoint.postgres import PostgresSaver

# Persistance dans Postgres
checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@host/db")
app = workflow.compile(checkpointer=checkpointer)

# Reprendre une conversation interrompue
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke(input_data, config=config)

# Time travel : revenir à un état précédent
history = app.get_state_history(config)
for state in history:
    print(state.values, state.next)

Étape 8 — Construire un système multi-agents

# Workflow avec 4 agents spécialisés

class MultiAgentState(TypedDict):
    task: str
    research_data: dict
    code_implementation: str
    tests_results: str
    final_output: str

def researcher_agent(state):
    """Agent recherche le contexte technique"""
    response = llm.invoke(f"Recherche les bonnes pratiques pour : {state['task']}")
    return {"research_data": {"content": response.content}}

def architect_agent(state):
    """Agent conçoit la solution"""
    response = llm.invoke(
        f"Basé sur la recherche : {state['research_data']}, "
        f"conçois la solution pour : {state['task']}"
    )
    return {"code_implementation": response.content}

def tester_agent(state):
    """Agent écrit les tests"""
    response = llm.invoke(f"Écris des tests unitaires pour : {state['code_implementation']}")
    return {"tests_results": response.content}

def reviewer_agent(state):
    """Agent fait la code review finale"""
    response = llm.invoke(
        f"Code : {state['code_implementation']}
"
        f"Tests : {state['tests_results']}
"
        "Critique et améliore."
    )
    return {"final_output": response.content}

workflow = StateGraph(MultiAgentState)
workflow.add_node("researcher", researcher_agent)
workflow.add_node("architect", architect_agent)
workflow.add_node("tester", tester_agent)
workflow.add_node("reviewer", reviewer_agent)

workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "architect")
workflow.add_edge("architect", "tester")
workflow.add_edge("tester", "reviewer")
workflow.add_edge("reviewer", END)

Étape 9 — Parallélisation pour performance

def parallel_research(state):
    """Lance 3 recherches en parallèle"""
    import asyncio
    
    async def search(topic):
        return await llm.ainvoke(f"Recherche sur : {topic}")
    
    topics = state["topics"]
    results = asyncio.run(asyncio.gather(*[search(t) for t in topics]))
    return {"research_results": [r.content for r in results]}

# Gain : 3 recherches sequential 30s → parallel 12s

Étape 10 — Intégrer Tools / Function Calling

from langchain.tools import tool

@tool
def search_database(query: str) -> str:
    """Cherche dans la base de connaissance"""
    # Logique de recherche
    return "Résultats..."

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Envoie un email"""
    # Intégration SendGrid / Resend
    return "Email envoyé"

llm_with_tools = llm.bind_tools([search_database, send_email])

def agent_with_tools(state):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

Étape 11 — Monitoring avec LangSmith

export LANGSMITH_API_KEY="your-key"
export LANGSMITH_TRACING=true
export LANGSMITH_PROJECT="my-app"

Toutes les exécutions sont automatiquement tracées dans LangSmith Dashboard. Vous voyez : steps, latency, coûts, erreurs, replay possible.

Étape 12 — Cas d’usage : Workflow article SEO complet

Pipeline 4 agents :

1. Keyword Researcher :
   - Input : sujet
   - Tool : Ahrefs API
   - Output : 10 KW prioritaires + volumes

2. Outline Generator :
   - Input : KW principal
   - Output : plan H1-H2-H3 détaillé

3. Writer :
   - Input : outline
   - Tool : web search pour exemples
   - Output : draft article 2000 mots

4. SEO Optimizer :
   - Input : draft
   - Tool : Yoast API check
   - Output : article optimisé final

Total : 1 article complet en 5-10 min vs 4h manuel.

Étape 13 — Déployer en production

• Containeriser : Dockerfile avec Python + LangGraph + dépendances
• API : FastAPI ou Flask exposant /workflow/run
• Hosting : Railway, Fly.io, AWS ECS, GCP Cloud Run
• Database : Postgres pour checkpoints
• Queue : Celery / BullMQ pour async tasks
• Cache : Redis pour responses fréquentes
• Secrets : variables environnement (jamais commit)

Étape 14 — Monitoring et amélioration continue

KPI à suivre :
- Temps moyen exécution workflow
- Taux de succès vs échec
- Coût moyen par exécution (tokens)
- Distribution des erreurs par noeud
- Qualité output (sample humain)
- Throughput (workflows / heure)

Optimisations :
- Cacher les calls LLM identiques
- Paralléliser quand possible
- Utiliser Haiku pour nodes simples (10x moins cher)
- Pre-compiler les prompts stables (caching)
- A/B test des stratégies de routing

Erreurs classiques en LangGraph

• Workflow trop linéaire : n’utilise pas le potentiel du graph.
• Pas de checkpoint : impossible de reprendre après erreur.
• Boucles infinies : oublier la condition d’arrêt sur les loops.
• Pas de monitoring : debug impossible en production.
• Pas de tests : régressions à chaque modification.

Checklist LangGraph multi-agents

✓ Comprendre StateGraph et noeuds
✓ Premier workflow simple testé
✓ Branchements conditionnels maîtrisés
✓ Persistence checkpoint Postgres
✓ Multi-agents avec rôles clairs
✓ Parallélisation pour performance
✓ Tools / function calling intégrés
✓ Monitoring LangSmith actif
✓ Cas d'usage métier validé (article, code, etc.)
✓ Déploiement Docker + API
✓ Cache + queue pour scale
✓ KPI mesurés et optimisés
✓ Tests automatisés sur les workflows
✓ Documentation interne complète