Workflows agentiques : structurer un flow fiable

Introduction

Les workflows agentiques deviennent utiles quand un enchaînement d’étapes ne tient plus proprement dans un simple script ou dans un agent unique. Le vrai sujet n’est pas d’ajouter des agents partout, mais de structurer un flow lisible, testable et rentable à exploiter. Ce guide s’adresse aux builders qui veulent choisir entre séquentiel, parallèle et supervisor sans sur-construire leur stack. Si ton besoin est court, stable et purement déterministe, reste sur une approche plus simple. Le bon objectif est d’orchestrer juste assez pour rester robuste.

Résumé rapide

• Les workflows agentiques deviennent utiles quand plusieurs étapes doivent s’adapter au contexte ou se reprendre proprement.
• Le vrai gain vient des garde-fous : retries bornés, checkpoints et validation humaine sur les étapes sensibles.
• Si ton besoin tient déjà dans un script simple, n’ajoute pas de complexité inutile.

Par où continuer selon votre besoin

• Si vous hésitez encore sur la structure, commencez par Orchestration multi-agents : quand l'utiliser.
• Si votre sujet est surtout la fiabilité, ajoutez ensuite un guide dédié à la gestion d’erreurs et aux frameworks.

Anatomie d'un workflow agentique

Un workflow agentique repose sur une idée simple : au lieu d'un seul agent qui fait tout, on décompose la mission en sous-tâches confiées à des agents spécialisés. Chaque agent possède un rôle, des outils dédiés et une mémoire locale.

Les 5 composants essentiels

┌─────────────┐
│   Trigger   │  ← Événement (cron, webhook, message)
└──────┬──────┘
       │
┌──────▼──────┐
│   Agent A   │  ← Réception + Planification
│  (Planner)  │
└──────┬──────┘
       │
┌──────▼──────┐
│   Outils    │  ← Appels API, lecture fichier, recherche
│  (Tools)    │
└──────┬──────┘
       │
┌──────▼──────┐
│   Mémoire   │  ← Contexte partagé + Vector DB
│  (Memory)   │
└──────┬──────┘
       │
┌──────▼──────┐
│   Agent B   │  ← Exécution + Décision sur l'étape suivante
│  (Executor) │
└──────┬──────┘
       │
┌──────▼──────┐
│   Output    │  ← Rapport, email, mise à jour BDD
└─────────────┘

Les 5 composants essentiels :

1. Trigger : l'événement qui démarre le workflow (cron, webhook, message, etc.)
2. Agents : les unités de travail, chacune avec un rôle et des outils
3. Outils : les capacités d'action (appels API, recherche vectorielle, exécution de code)
4. Mémoire : le contexte partagé entre agents (state, vector store)
5. Output : le livrable final (rapport, email, mise à jour, etc.)

La différence fondamentale avec un script classique réside dans la gestion dynamique du flux : un agent peut décider de réessayer une action, de faire appel à un autre agent, ou de signaler un problème nécessitant une intervention humaine — c'est le human-in-the-loop pattern.

Les 5 patterns d'orchestration

Choisir le bon pattern dépend de la nature de la mission : certaines se prêtent à un flux séquentiel, d'autres nécessitent une exécution parallèle.

1. Séquentiel — Le pattern le plus simple

Les agents s'exécutent les uns après les autres. Chaque agent reçoit le output de l'agent précédent comme contexte.

Quand l'utiliser : étapes dépendantes, pipeline de traitement (scraping → nettoyage → analyse → rapport).

Agent A → Agent B → Agent C → Output

2. Parallèle — Plusieurs agents simultanés

Plusieurs agents s'exécutent en même temps, chacun sur une sous-tâche indépendante. Leurs résultats sont agrégés à la fin.

Quand l'utiliser : veille multi-sources, génération de variations, collectes parallèles.

       ┌─ Agent B ─┐
Agent A ┤ Agent C  │ → Aggregator → Output
       └─ Agent D ─┘

3. Supervisor — Un agent orchestrateur

Un agent superviseur coordonne les autres. Il décide quel agent appeler à chaque étape, gère les dépendances et valide les résultats.

Quand l'utiliser : tâches complexes avec plusieurs chemins possibles, systèmes multi-agents où la coordination est critique.

         ┌─ Agent B ─┐
Supervisor ┤ Agent C  │ ← Le supervisor décide de l'ordre
           └─ Agent D ─┘

4. Fan-out / Fan-in — Distribution puis agrégation

Un agent lance plusieurs sous-tâches en parallèle (fan-out), puis un agent agrégateur collecte et synthétise les résultats (fan-in).

Quand l'utiliser : analyse multi-sources, génération de contenu à partir de plusieurs contextes, audits.

           ┌─ Sous-agent 1 ─┐
Initiateur ┤ Sous-agent 2 ─┼→ Agrégateur → Output
           └─ Sous-agent 3 ─┘

5. Hiérarchique — Cascade d'agents

Les agents forment une hiérarchie : un agent de niveau supérieur délègue à des agents de niveau inférieur, qui délèguent à leur tour.

Quand l'utiliser : organisations complexes, décomposition hiérarchique de tâches.

Manager → Senior Agent → Junior Agent 1
                          └→ Junior Agent 2

Exemple concret : implémenter des workflows agentiques

Passons à la pratique. Les exemples ci-dessous utilisent le SDK OpenClaw en Python. Chaque exemple est complet et exécutable.

Prérequis

pip install openclaw-sdk python-dotenv

Exemple 1 — Pipeline Scraping → Analyse → Rapport (Séquentiel)

import openclaw
from openclaw import Agent, Tool, Memory

 client = openclaw.Client()  # configuration du client

 # Outil de scraping
@Tool
def scrape_url(url: str) -> str:
    """Extrait le contenu textuel d'une URL."""
    import httpx
    response = httpx.get(url, timeout=15.0)
    return response.text[:5000]


 # Outil d'analyse
@Tool
def analyze_text(text: str, keywords: list[str]) -> dict:
    """Analyse le texte et extrait les mentions de mots-clés."""
    results = {}
    for kw in keywords:
        results[kw] = text.lower().count(kw.lower())
    return {
        "keyword_counts": results,
        "word_count": len(text.split()),
        "sentiment": "neutral"  # simplifié pour l'exemple
    }

 # Agent de scraping
scraper = Agent(
    role="Web Scraper",
    goal="Récupérer le contenu des URLs demandées",
    tools=[scrape_url],
    model="openai/gpt-4o"
)

 # Agent analyste
analyst = Agent(
    role="Content Analyst",
    goal="Analyser le contenu et identifier les insights clés",
    tools=[analyze_text],
    model="openai/gpt-4o"
)

 # Exécution séquentielle
urls = [
    "https://example.com/article-1",
    "https://example.com/article-2"
]
keywords = ["automation", "agent", "IA"]


 # Étape 1 : Scraping
scraped_contents = []
for url in urls:
    result = scraper.run(f"Scrape le contenu de {url}")
    scraped_contents.append(result)

 # Étape 2 : Analyse
analyses = []
for content in scraped_contents:
    result = analyst.run(
        f"Analyse ce texte avec les mots-clés {keywords}: {content}"
    )
    analyses.append(result)


 # Étape 3 : Rapport
print("=== RÉSULTATS ===")
for i, analysis in enumerate(analyses):
    print(f"Source {i+1}: {analysis}")

Exemple 2 — Triage email → Réponse → Archivage (Supervisor)

from openclaw import Agent, Tool
import json

 # Outils
@Tool
def fetch_unread_emails() -> list[dict]:
    """Retourne la liste des emails non lus."""
    # Connexion IMAP (via himalaya ou directement)
    return [
        {"id": "001", "from": "client@example.com", "subject": "Question produit", "body": "Bonjour, est-ce que votre outil supporte X ?"},
        {"id": "002", "from": "support@example.com", "subject": "Rapport hebdomadaire", "body": "Voici le rapport de la semaine."},
    ]

@Tool
def send_email(to: str, subject: str, body: str) -> str:
    """Envoie un email."""
    return f"Email envoyé à {to} : {subject}"

@Tool
def archive_email(email_id: str, category: str) -> str:
    """Archive un email dans une catégorie."""
    return f"Email {email_id} archivé dans {category}"

 # Agents spécialisés
triage_agent = Agent(
    role="Email Triage",
    goal="Analyser chaque email et déterminer sa catégorie et priorité",
    model="openai/gpt-4o-mini"
)

response_agent = Agent(
    role="Response Writer",
    goal="Rédiger une réponse professionnelle et adaptée",
    model="openai/gpt-4o"
)

 # Supervisor
def supervisor_workflow():
    emails = fetch_unread_emails()
    results = []
    
    for email in emails:
        # 1. Triage
        triage = triage_agent.run(
            f"Analyse cet email et retourne un JSON avec category et priority: {email}"
        )
        
        triage_data = json.loads(triage) if triage.startswith("{") else {"category": "general", "priority": "normal"}
        
        # 2. Si haute priorité → répondre immédiatement
        if triage_data.get("priority") == "high":
            response = response_agent.run(
                f"Rédige une réponse à cet email: {email['body']}"
            )
            send_email(email["from"], f"Re: {email['subject']}", response)
        
        # 3. Archivage systématique
        archive_email(email["id"], triage_data.get("category", "inbox"))
        results.append({"email_id": email["id"], "triage": triage_data})
    
    return results

results = supervisor_workflow()
print(json.dumps(results, indent=2))

Exemple 3 — Veille multi-sources (Fan-out / Fan-in)

from openclaw import Agent, Tool
from concurrent.futures import ThreadPoolExecutor

@Tool
def search_web(query: str) -> str:
    """Recherche sur le web."""
    return f"Résultats de recherche pour: {query}"


def fan_out_search(queries: list[str]) -> list[str]:
    """Exécute plusieurs recherches en parallèle."""
    agent = Agent(role="Searcher", goal="Rechercher sur le web", model="openai/gpt-4o-mini")
    
    results = []
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(agent.run, f"Recherche: {q}") for q in queries]
        results = [f.result() for f in futures]
    return results

def aggregate_results(raw_results: list[str]) -> str:
    """Agrège les résultats de recherche en une synthèse."""
    aggregator = Agent(
        role="Aggregator",
        goal="Synthétiser les résultats de recherche en un rapport clair",
        model="openai/gpt-4o"
    )
    return aggregator.run(
        f"Synthétise ces résultats de recherche en un rapport structuré:\n{raw_results}"
    )

 # Exécution Fan-out / Fan-in
queries = [
    "OpenClaw multi-agent orchestration 2026",
    "CrewAI vs LangGraph comparison",
    "AI agent workflow best practices"
]

 # Fan-out : recherches parallèles
raw_results = fan_out_search(queries)

 # Fan-in : agrégation
final_report = aggregate_results(raw_results)
print(final_report)

Patterns avancés pour la production

Voici les techniques qui séparent un prototype d'un système production-ready.

Human-in-the-loop

Certaines décisions requièrent un humain avant de continuer. Par exemple :

• Approbation avant l'envoi d'un email externe
• Validation d'un contenu avant publication
• Escalade vers un support humain si la confiance est faible

@Tool
def require_human_approval(task: str, context: dict) -> bool:
    """Demande une approbation humaine. Retourne True si approuvé."""
    print(f"⚠️ APPROBATION REQUISE: {task}")
    print(f"Contexte: {context}")
    response = input("Approuver? (o/n): ")
    return response.lower() == "o"

 # Dans le workflow
if not require_human_approval("Envoyer email", {"to": client, "content": draft}):
    print("Email non envoyé — en attente d'approbation.")
    return

Gestion d'erreurs et Retry

from openclaw import Agent
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def resilient_task(agent: Agent, prompt: str) -> str:
    try:
        return agent.run(prompt)
    except Exception as e:
        print(f"Erreur: {e} — nouvelle tentative...")
        raise

Rollback

Pour les opérations critiques, implémenter un mécanisme de rollback :

class WorkflowState:
    def __init__(self):
        self.checkpoints = []
    
    def checkpoint(self, label: str, data: dict):
        self.checkpoints.append({"label": label, "data": data})
    
    def rollback(self, label: str):
        """Revient au dernier checkpoint."""
        for i, cp in enumerate(reversed(self.checkpoints)):
            if cp["label"] == label:
                return cp["data"]
        return self.checkpoints[0]["data"]

Circuit Breaker

Pour éviter les cascading failures quand un service externe est down :

from circuitbreaker import circuit

@circuit(failure_threshold=3, recovery_timeout=30)
def call_external_api(endpoint: str) -> dict:
    response = httpx.get(endpoint)
    return response.json()

Stack complète pour la production

Pour aller en production avec des workflows agentiques, voici la stack recommandée :

Composant	Outil	Rôle
Orchestration	OpenClaw	Coordination des agents, gestion du state
Mémoire vectorielle	Qdrant / Pinecone	Stockage et retrieval du contexte
Triggers externes	n8n / Zapier	Déclenchement sur événements (webhook, cron)
Monitoring	LangSmith / Phoenix	Tracing, latence, coûts LLM
Queue async	Redis / Celery	Gestion des workflows longs
Hébergement	VPS (Hetzner, OVH)	Infrastructure

Bonnes pratiques

Testing

Chaque workflow mérite des tests. Utilisez des mocks pour les appels externes et des fixtures pour le state :

def test_scraper_workflow(mocker):
    mocker.patch("httpx.get", return_value=Mocker(text="contenu test"))
    result = scraper.run("Scrape https://test.com")
    assert "contenu test" in result

Observabilité

Instrumentez chaque étape avec du logging structuré :

import structlog
logger = structlog.get_logger()

logger.info("workflow_step", step="scraping", url=url, duration_ms=1200)

Réduction des coûts LLM

• Utilisez des modèles petits (gpt-4o-mini) pour les tâches simples
• Activez le streaming pour réduire la latence perçue
• Mettez en cache les prompts invariants avec @cache

Sécurité

• Ne jamais exposer les clés API dans les logs
• Utiliser des variables d'environnement via python-dotenv
• Valider les entrées utilisateur avant de les passer aux agents (prompt injection)

Questions fréquentes

Qu'est-ce qu'un workflow agentique ?

Un workflow agentique est une chaîne d'agents IA qui collaborent de manière autonome pour accomplir une mission. Chaque agent possède un rôle, des outils et peut prendre des décisions dynamiques sur le prochain étape du workflow.

Comment créer des workflows agentiques ?

La création passe par 4 étapes :

1. Décomposer la mission en sous-tâches
2. Attribuer chaque sous-tâche à un agent spécialisé
3. Définir les patterns d'orchestration (séquentiel, parallèle, supervisor)
4. Implémenter avec un framework comme OpenClaw et tester en conditions réelles

Quelle différence entre workflow séquentiel et workflow parallèle ?

Dans un workflow séquentiel, les agents s'exécutent les uns après les autres — le output de l'un devient l'input du suivant. Dans un workflow parallèle, plusieurs agents s'exécutent simultanément sur des sous-tâches indépendantes, puis un agrégateur combine leurs résultats.

Comment gérer les erreurs dans un workflow agentique ?

Utilisez une combinaison de retries avec backoff exponentiel, de checkpoints pour permettre le rollback, et de circuit breakers pour éviter les cascading failures. Implémentez aussi un human-in-the-loop pour les décisions critiques.

Quels outils pour surveiller un workflow agentique en production ?

LangSmith et Arize Phoenix offrent du tracing complet (latence par étape, coûts LLM, taux d'erreur). Pour les logs structurés, utilisez structlog. Pour la métrique métier, branchez un dashboard Grafana sur vos checkpoints.

Articles liés

Un workflow agentique utile n’est pas celui qui empile le plus d’agents, mais celui qui garde un flux lisible et récupérable quand quelque chose casse. Commence simple, ajoute du supervisor seulement quand le routage ou la reprise le justifient, puis consolide l’observabilité et la validation.

• Créer un agent IA
• OpenClaw : guide complet
• Systèmes multi-agents
• CrewAI : orchestrer des Crews
• Outils pour agents IA