Aller au contenu principal

Automatiser des workflows multi-étapes

Mis à jour le 29 juillet 2026

Orchestrer des tâches complexes

Un workflow réel dépasse rarement une seule action. Remplir un CRM implique de naviguer entre les pages, de chercher un contact, de mettre à jour ses informations, puis de créer une opportunité — quatre opérations enchaînées, dont chacune peut échouer pour ses propres raisons. Confier ce bloc entier à un agent en un seul prompt fonctionne parfois, mais vous ne saurez jamais où il a décroché. Cette leçon vous apprend à structurer ces automatisations multi-étapes de façon à ce que chaque maillon soit observable et rejouable.

Architecture en étapes

Le point de départ consiste à découper le workflow en étapes atomiques, chacune avec un objectif clair et un critère de succès. Ce critère est ce qui distingue une liste de tâches d’un véritable workflow : sans lui, votre orchestrateur ne peut pas savoir si l’agent a réellement abouti ou s’il s’est arrêté sur une page d’erreur. Dans l’exemple ci-dessous, la connexion se vérifie par l’URL du tableau de bord et la recherche par la présence du nom du contact à l’écran ; les deux dernières étapes, purement rédactionnelles, s’en passent.

workflow = [
    {
        "name": "connexion",
        "task": "Connecte-toi au CRM avec les identifiants fournis.",
        "success_url_contains": "/dashboard",
    },
    {
        "name": "recherche_contact",
        "task": "Cherche le contact 'Jean Dupont' dans la barre de recherche.",
        "success_text_contains": "Jean Dupont",
    },
    {
        "name": "mise_a_jour",
        "task": "Ouvre la fiche de Jean Dupont et mets à jour son téléphone : 06 98 76 54 32.",
    },
    {
        "name": "creation_opportunite",
        "task": "Crée une opportunité : 'Migration Cloud Q2 2026', montant 50000 EUR, probabilité 60%.",
    },
]

L’orchestrateur

Reste à exécuter cette liste. L’orchestrateur parcourt les étapes séquentiellement et enveloppe chacune d’une boucle de tentatives : une étape qui lève une exception ou qui n’atteint pas l’URL attendue est simplement rejouée, jusqu’à max_retries fois. Ce détail change tout en production, car la majorité des échecs d’un agent visuel sont transitoires — une page lente, un rendu incomplet au moment du screenshot. L’orchestrateur retourne enfin un journal structuré indiquant, pour chaque étape, son statut et le nombre de tentatives consommées : c’est votre premier outil de diagnostic.

def run_workflow(workflow: list, page, max_retries: int = 2):
    """Exécute un workflow multi-étapes avec retry."""
    results = []

    for step in workflow:
        step_name = step["name"]
        print(f"\n--- Étape : {step_name} ---")

        for attempt in range(max_retries + 1):
            try:
                result = run_agent(step["task"], page=page)

                # Vérification optionnelle par URL
                url_check = step.get("success_url_contains")
                if url_check and url_check not in page.url:
                    print(f"  URL attendue non trouvée, retry...")
                    continue

                print(f"  Succès (tentative {attempt + 1})")
                results.append({
                    "step": step_name,
                    "status": "success",
                    "attempts": attempt + 1,
                })
                break

            except Exception as e:
                print(f"  Erreur : {e}")
                if attempt == max_retries:
                    results.append({
                        "step": step_name,
                        "status": "failed",
                        "error": str(e),
                    })

    return results

Passer des données entre les étapes

Souvent, une étape produit des données nécessaires à la suivante : le prix lu sur une page de panier doit être vérifié à l’étape de validation, le numéro de dossier créé à l’étape 2 sert de référence à l’étape 5. Un contexte partagé résout ce besoin sans transformer votre code en enchevêtrement de variables globales. La classe ci-dessous stocke des paires clé-valeur et expose une méthode inject qui remplace les marqueurs {clé} dans un modèle de prompt, ce qui vous permet d’écrire vos tâches comme des gabarits paramétrés.

class WorkflowContext:
    """Stocke les données partagées entre les étapes."""

    def __init__(self):
        self.data = {}

    def set(self, key: str, value):
        self.data[key] = value

    def get(self, key: str, default=None):
        return self.data.get(key, default)

    def inject(self, template: str) -> str:
        """Remplace les variables {clé} dans le template."""
        task = template
        for key, value in self.data.items():
            task = task.replace("{" + key + "}", str(value))
        return task


# Exemple d'utilisation
ctx = WorkflowContext()

# Étape 1 : l'agent extrait un prix
prix_extrait = run_agent(
    "Lis le prix total affiché et retourne uniquement le nombre.",
    page=page
)
ctx.set("prix", prix_extrait)

# Étape 2 : utilise le prix extrait
task_validation = ctx.inject(
    "Le montant attendu est {prix} EUR. Vérifie et valide la commande."
)
run_agent(task_validation, page=page)

Workflows conditionnels

La séquence linéaire atteint vite ses limites, car la prochaine étape peut dépendre du résultat de la précédente. Le motif à retenir est celui-ci : demandez d’abord à l’agent de lire un état en le contraignant à répondre par une valeur parmi un ensemble fermé, puis branchez côté Python. Sur un suivi de commande, un statut « en cours » mène au suivi du colis, « livré » à la demande d’avis, « annulé » à une proposition de nouvelle commande. La contrainte de format dans le prompt est essentielle : sans elle, l’agent renvoie une phrase que vos tests de sous-chaîne rateront une fois sur trois.

def conditional_workflow(page):
    """Workflow avec branchement conditionnel."""
    # Étape 1 : Lire le statut
    status = run_agent(
        "Lis le statut de la commande. "
        "Retourne uniquement : en_cours, livré ou annulé.",
        page=page
    )

    # Branchement selon le résultat
    if "en_cours" in status:
        run_agent(
            "La commande est en cours. Clique sur 'Suivre le colis'.",
            page=page
        )
    elif "livré" in status:
        run_agent(
            "La commande est livrée. Clique sur 'Laisser un avis'.",
            page=page
        )
    elif "annulé" in status:
        run_agent(
            "La commande est annulée. Clique sur 'Recommander'.",
            page=page
        )

Parallélisation

Quand les tâches sont indépendantes les unes des autres, rien n’oblige à les traiter en file. Extraire les données de trois fiches produit prend trois fois moins de temps si trois agents travaillent simultanément, chacun dans son propre onglet. La version asynchrone de Playwright et asyncio.gather suffisent : une coroutine par URL, un new_page par coroutine, et une fermeture systématique de la page pour ne pas accumuler les onglets en mémoire. Gardez en tête que le parallélisme multiplie aussi votre débit d’appels API et votre charge sur le site cible — trois onglets passent inaperçus, cinquante beaucoup moins.

import asyncio
from playwright.async_api import async_playwright

async def parallel_extraction(urls: list[str]):
    """Extrait des données de plusieurs pages en parallèle."""
    async with async_playwright() as pw:
        browser = await pw.chromium.launch(headless=True)

        async def extract_one(url: str):
            page = await browser.new_page(
                viewport={"width": 1280, "height": 720}
            )
            await page.goto(url)
            result = await run_agent_async(
                "Extrais le titre et le prix du produit.", page=page
            )
            await page.close()
            return {"url": url, "data": result}

        results = await asyncio.gather(
            *[extract_one(url) for url in urls]
        )

    return results

# Lancer
urls = [
    "https://shop.example.com/product/1",
    "https://shop.example.com/product/2",
    "https://shop.example.com/product/3",
]
data = asyncio.run(parallel_extraction(urls))

Points clés à retenir

  • Découpez les workflows en étapes atomiques avec des critères de succès explicites
  • Ajoutez du retry automatique pour chaque étape (2-3 tentatives)
  • Utilisez un contexte partagé pour passer des données entre les étapes
  • Les workflows conditionnels permettent de s’adapter au contenu de la page
  • Parallélisez les tâches indépendantes pour gagner du temps
← Lecon precedenteGérer l'authentification et les sessionsLecon suivante →Combiner Computer Use avec d'autres tools