Workflow Automation Agent

Workflow Automation Agent

Workflow

1. Cartographier les processus métier

• Documenter chaque étape : acteur, système, volume, fréquence, temps moyen.
• Identifier les tâches 100 % déterministes (à automatiser en priorité) vs les tâches à jugement (LLM ou escalade humaine).
• Calculer le ROI : (temps_manuel_h × taux_horaire × volume_annuel) − coût_implémentation.
• Critère d'exclusion : si le processus change plus d'une fois par trimestre, différer ou abstraire la logique dans un fichier de configuration.

2. Choisir l'architecture de l'agent

Exemple minimal Temporal (Python) :

@workflow.defn
class OnboardingWorkflow:
    @workflow.run
    async def run(self, user_id: str) -> str:
        await workflow.execute_activity(create_account, user_id, start_to_close_timeout=timedelta(minutes=5))
        await workflow.execute_activity(send_welcome_email, user_id, start_to_close_timeout=timedelta(minutes=2))
        return "done"

3. Implémenter les connecteurs

• Un connecteur = une classe avec execute(), compensate(), health_check().
• Toujours décorer avec retry + circuit breaker :

from tenacity import retry, stop_after_attempt, wait_exponential
from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=30)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_external_api(payload: dict) -> dict:
    response = requests.post(API_URL, json=payload, timeout=10)
    response.raise_for_status()
    return response.json()

• RPA UI (sans API) : préférer les sélecteurs stables (data-testid, aria-label) aux sélecteurs CSS fragiles.

# Playwright — sélecteur robuste
await page.get_by_role("button", name="Valider").click()
# ❌ fragile
await page.click(".btn-primary:nth-child(2)")

4. Configurer les déclencheurs

Implémenter le debouncing pour éviter les doubles déclenchements :

LAST_TRIGGER: dict[str, float] = {}
MIN_INTERVAL = 5.0  # secondes

def debounce(key: str) -> bool:
    now = time.time()
    if now - LAST_TRIGGER.get(key, 0) < MIN_INTERVAL:
        return False  # ignorer
    LAST_TRIGGER[key] = now
    return True

5. Orchestrer les actions chaînées

• Persister l'état entre les étapes (DB ou Redis) pour reprendre après crash :

@dataclass
class WorkflowState:
    workflow_id: str
    step: str          # "create_account" | "send_email" | "done"
    payload: dict
    created_at: datetime
    retries: int = 0

• Implémenter la compensation (rollback) pour chaque étape modifiant un système externe :

executed_steps = []
try:
    create_order(order); executed_steps.append("create_order")
    charge_payment(order); executed_steps.append("charge_payment")
    send_confirmation(order)
except Exception:
    for step in reversed(executed_steps):
        compensators[step](order)  # rollback dans l'ordre inverse
    raise

• Sous-workflows : encapsuler les blocs réutilisables (ex : KYCWorkflow, NotificationWorkflow).

6. Intégrer la décision intelligente (LLM)

• Utiliser le LLM uniquement pour les cas non déterministes : classification, extraction de texte libre, cas ambigus.
• Définir un seuil de confiance : en dessous → escalade humaine.

result = llm_classify(document_text)
if result.confidence < 0.85:
    escalate_to_human(document_id, reason="confiance insuffisante")
else:
    route_automatically(result.category)

• Toujours valider la sortie LLM avec un schéma Pydantic avant de l'injecter dans le workflow.

7. Gérer les erreurs et la résilience

• Retry policy : exponentiel avec jitter pour éviter le thundering herd.
• Dead-letter queue : messages non traités après N tentatives → DLQ avec alerte.
• Timeout par étape et timeout global du workflow.
• Idempotence : chaque action doit être réexécutable sans effet de bord. Utiliser un idempotency_key (UUID de la commande, hash de l'événement).

def process_payment(order_id: str, idempotency_key: str):
    if db.exists(f"payment:{idempotency_key}"):
        return db.get(f"payment:{idempotency_key}")  # résultat déjà calculé
    result = payment_gateway.charge(order_id)
    db.set(f"payment:{idempotency_key}", result, ex=86400)
    return result

8. Monitorer et améliorer

KPIs à collecter :

• Taux de succès : succeeded / total par workflow et par étape.
• Temps de cycle moyen et p95.
• Taux d'escalade humaine : indicateur de maturité du modèle de décision.
• Volume de la DLQ : indicateur de problèmes systémiques.

Stack recommandée 2026 : OpenTelemetry (traces) + Prometheus (métriques) + Grafana (dashboards).

from opentelemetry import trace
tracer = trace.get_tracer("workflow-agent")

with tracer.start_as_current_span("step.create_order") as span:
    span.set_attribute("order.id", order_id)
    create_order(order_id)

Garde-fous et anti-patterns

Bonnes pratiques 2026

• Contract-first : définir le contrat d'interface (schéma Pydantic / OpenAPI) avant d'implémenter les connecteurs.
• Infrastructure as Code : déclarer les queues, topics et schedulers dans Terraform/Bicep, pas en CLI.
• Versioning de workflow : suffixer les définitions (OnboardingWorkflow_v2) pour coexistence avec les instances en cours.
• Secrets : injecter via vault (Azure Key Vault, HashiCorp Vault) — jamais en variable d'environnement dans le code.
• Tests : unit-tester chaque activité isolément ; integration-tester le workflow avec un Temporal test server ou mocks.
• Observabilité dès le départ : ne pas ajouter les traces après coup ; instrumenter dès la première ligne de code.