Automatiser des workflows multi-étapes — Computer Use et Automatisation

Leçon corrigée le 3 novembre 2026 d’après les consignes de confirmation d’OpenAI pour Computer use : la connexion n’est plus confiée à l’agent, la validation d’une commande et les branches qui engagent l’utilisateur (avis, nouvelle commande) passent par une confirmation humaine, et l’orchestrateur vérifie désormais le critère success_text_contains qu’il ignorait, signale les étapes en échec au lieu de les omettre, et ne rejoue plus une étape qui crée une donnée. Source : Confirmation et consentement.

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, l’ouverture du module se vérifie par l’URL et la recherche par la présence du nom du contact à l’écran. La connexion n’y figure pas : votre code l’a faite avant, sans montrer le mot de passe au modèle (leçon 8). Et la dernière étape porte "retry": False, parce qu’elle crée une donnée : la rejouer après un échec partiel risquerait de créer un doublon.

workflow = [
    {
        "name": "ouverture_contacts",
        "task": "Ouvre le module Contacts du CRM.",
        "success_url_contains": "/contacts",
    },
    {
        "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%.",
        "retry": False,  # étape non rejouable : elle crée une donnée
    },
]

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 ne remplit pas son critère de succès est rejouée, jusqu’à max_retries fois, sauf si elle est marquée non rejouable. Ce détail compte en production, car bien des échecs d’un agent visuel sont transitoires — une page lente, un rendu incomplet au moment du screenshot. Une étape qui échoue définitivement arrête le workflow, puisque les suivantes en dépendent. 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} ---")
        tentatives = max_retries + 1 if step.get("retry", True) else 1
        erreur = "critère de succès non atteint"

        for attempt in range(tentatives):
            try:
                run_agent(step["task"], page=page)

                # Vérifications optionnelles : URL, puis texte visible
                url_check = step.get("success_url_contains")
                if url_check and url_check not in page.url:
                    print("  URL attendue non trouvée")
                    continue
                text_check = step.get("success_text_contains")
                if text_check and text_check not in page.locator("body").inner_text():
                    print("  Texte attendu non trouvé")
                    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}")
                erreur = str(e)
        else:
            # Aucune tentative n'a abouti : les étapes suivantes en dépendent
            results.append({"step": step_name, "status": "failed", "error": erreur})
            break

    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. L’exemple qui suit s’arrête volontairement avant le paiement : OpenAI range la confirmation d’une transaction financière parmi les actions à faire valider par l’utilisateur au moment même de l’action. 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 lit un prix
prix_extrait = run_agent(
    "Lis le prix total affiché et retourne uniquement le nombre.",
    page=page
)
ctx.set("prix", prix_extrait.strip())

# Étape 2 : l'agent vérifie, sans rien valider
verdict = run_agent(ctx.inject(
    "Le montant attendu est {prix} EUR. Compare-le au total de la page de paiement "
    "et réponds uniquement OK ou ÉCART. Ne valide rien."
), page=page)

# Étape 3 : une transaction se confirme au moment de l'action, par un humain
if verdict.strip() == "OK":
    if input(f"Valider la commande de {ctx.get('prix')} EUR ? [o/N] ").strip().lower() == "o":
        run_agent("Clique sur le bouton de validation de la commande.", 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é » au relevé de la date de livraison, « annulé » au relevé du motif. La contrainte de format dans le prompt est essentielle : sans elle, l’agent renvoie une phrase que vos comparaisons rateront régulièrement. Les trois branches restent des lectures : publier un avis ou repasser commande engagerait l’utilisateur, et demanderait sa confirmation.

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
    match status.strip().lower():
        case "en_cours":
            return run_agent("Clique sur 'Suivre le colis' et relève l'étape affichée.", page=page)
        case "livré":
            return run_agent("Relève la date de livraison affichée.", page=page)
        case "annulé":
            return run_agent("Relève le motif d'annulation affiché.", page=page)
        case _:
            raise ValueError(f"Statut inattendu : {status!r}")

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 nettement 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. run_agent_async est la boucle de la leçon 4 réécrite avec le client AsyncOpenAI et l’API asynchrone de Playwright : la même logique, avec await devant chaque appel. Gardez en tête que le parallélisme multiplie aussi votre débit d’appels API, donc le risque d’atteindre vos limites de débit (leçon 11), 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}
            )
            try:
                await page.goto(url)
                result = await run_agent_async(
                    "Extrais le titre et le prix du produit.", page=page
                )
            finally:
                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
Rejouez automatiquement les étapes qui peuvent l’être sans risque, jamais celles qui créent une donnée
Utilisez un contexte partagé pour passer des données entre les étapes
Les workflows conditionnels s’adaptent au contenu de la page ; les branches qui engagent l’utilisateur demandent sa confirmation
Parallélisez les tâches indépendantes pour gagner du temps