Applied AI · Avancé 🔴 · Session 4
✏️ Exercices
← Retour au programme 📄 Source .md

Applied AI — Niveau Avancé — Session 4

Exercices : Architecture multi-agents

Instructeur : Yann Isola Durée totale : 90 minutes Modalité : individuel ou binôme — les corrigés sont destinés à l’instructeur

Exercice 1 — Choix de stratégie de décomposition (25 min)

Contexte

Vous êtes architecte solutions chez un courtier en instruments financiers tokenisés. Trois projets arrivent simultanément. Pour chacun, vous devez choisir une stratégie de décomposition (verticale, horizontale, récursive, hybride — ou aucune décomposition) et la justifier avec les trois critères de décision (seuil de complexité, budget de tokens, bénéfice de spécialisation).

Cas A — Générateur de fiches produit

Une tâche répétitive : à partir d’une fiche technique de 2 pages, générer une description commerciale de 300 mots dans un ton défini. Volume : 500 fiches/jour. La fiche technique + le prompt tiennent largement dans le contexte.

Cas B — Due diligence d’une contrepartie

Pour chaque nouvelle contrepartie : analyse juridique (structure sociale, licences), analyse financière (bilans, ratios), analyse réputationnelle (presse, sanctions), analyse technique (infrastructure de conservation des actifs). Puis production d’un mémo de synthèse avec recommandation GO/NO-GO. Chaque volet mobilise des documents volumineux et un vocabulaire spécialisé distinct.

Cas C — Migration documentaire d’un fonds

Un fonds transmet une arborescence de ~4 000 documents (profondeur et structure imprévisibles : dossiers imbriqués, formats hétérogènes, langues mixtes). Chaque document doit être classifié, résumé, indexé et rattaché à un référentiel réglementaire. La structure de l’arborescence n’est connue qu’à l’exécution.

Livrables attendus (par cas)

  1. Stratégie choisie (une phrase).
  2. Justification par les 3 critères (tableau).
  3. Schéma d’architecture (ASCII ou dessin) : agents, flux, points de convergence.
  4. Pour le cas C uniquement : les trois garde-fous obligatoires de votre architecture et leurs valeurs proposées.

Corrigé instructeur — Exercice 1

Cas A — Aucune décomposition.

Cas B — Hybride : horizontale pour l’analyse, verticale pour l’ensemble.

Cas C — Récursive.

Barème indicatif (sur 20) : Cas A : 5 pts (dont 2 pour avoir résisté à la sur-ingénierie) · Cas B : 8 pts (dont 3 pour l’hybride correctement articulé) · Cas C : 7 pts (dont 3 pour les garde-fous chiffrés).

Exercice 2 — Conception d’une politique d’escalade (30 min)

Contexte

Un assistant multi-agents de service client opère pour une plateforme de conservation d’actifs numériques. Il peut : répondre aux questions (produits, frais, procédures), modifier des informations de profil non sensibles, initier des procédures de récupération de compte, et préparer (sans exécuter) des instructions de retrait.

Trois incidents récents motivent une refonte de la politique d’escalade :

Travail demandé

  1. Matrice de déclencheurs (8 pts) — Pour chacune des 4 capacités de l’agent, définissez les déclencheurs d’escalade selon les trois familles (confiance sous seuil / hors périmètre / niveau de risque). Présentez sous forme de tableau. Chaque incident doit être couvert par au moins un déclencheur.
  2. Choix des patrons (6 pts) — Pour chaque déclencheur, associez le patron approprié (pause-and-ask / queue-for-review / fallback-to-human) et justifiez en une phrase.
  3. Politique de SLA (6 pts) — Concevez la réponse à l’incident 3 : niveaux de priorité, délais cibles, action de timeout par niveau (fail-safe / fail-operational / ré-escalade), et deux métriques de pilotage.

Contrainte

Votre politique doit rester exploitable : si tout escalade en P1 vers un humain, vous avez échoué. Indiquez explicitement ce qui reste autonome.

Corrigé instructeur — Exercice 2

1. Matrice de déclencheurs — éléments attendus :

Capacité Confiance Hors périmètre Risque
Questions produits/frais Juge distinct, seuil calibré (ex. 0,75) → escalade si en dessous Conseil fiscal/juridique/investissement détecté → escalade (couvre incident 2) Faible → autonome par défaut
Modification profil non sensible — Demande glissant vers données sensibles (IBAN, 2FA) → escalade Faible/réversible → autonome + journalisation
Récupération de compte Incohérences dans les vérifications → escalade — Élevé par nature : détecteur de signaux de fraude (urgence, pression, incohérences) → escalade obligatoire (couvre incident 1)
Préparation de retrait — — Élevé/irréversible en aval → approbation humaine préalable systématique au-delà d’un seuil de montant

Points clés à vérifier : (a) l’incident 2 est traité par hors périmètre (conseil fiscal réglementé), pas seulement par la confiance — une réponse inventée avec assurance a précisément une confiance auto-déclarée élevée, d’où la nécessité du classifieur de périmètre et/ou d’un juge distinct ; (b) l’incident 1 exige un déclencheur de risque indépendant de la confiance.

2. Patrons attendus :

3. SLA — réponse type :

Exercice 3 — Implémentation de la gestion d’erreurs (35 min)

Contexte

Vous recevez le squelette d’un orchestrateur qui appelle trois sous-agents en parallèle puis fusionne leurs résultats. En production, l’API de l’agent enrichisseur connaît des pannes intermittentes qui font actuellement échouer des missions entières.

# --- CODE EXISTANT (défaillant) ---
def run_mission(task):
    r1 = agent_analyste.run(task)        # fiable
    r2 = agent_enrichisseur.run(task)    # pannes intermittentes !
    r3 = agent_redacteur_notes.run(task) # fiable
    return merge(r1, r2, r3)             # KeyError si r2 manque

Travail demandé

Réécrivez run_mission et son infrastructure pour intégrer, dans cet ordre :

  1. (6 pts) Typologie + retry avec backoff — Classez les erreurs (réessayables / fatales). Implémentez un retry (max 3 tentatives) avec backoff exponentiel et jitter, appliqué uniquement aux erreurs réessayables. L’appel à agent_enrichisseur est-il rejouable sans précaution ? Justifiez en une phrase dans un commentaire (indice : idempotence).
  2. (6 pts) Circuit breaker — Implémentez un disjoncteur (seuil : 5 échecs, refroidissement : 30 s) autour de agent_enrichisseur uniquement. Les trois états doivent être explicites dans le code.
  3. (4 pts) Dégradation gracieuse — Si l’enrichisseur est indisponible (circuit ouvert ou retries épuisés), la mission doit aboutir quand même avec un résultat partiel, porteur d’une mention explicite du volet manquant. merge ne doit plus jamais lever KeyError.
  4. (4 pts) DLQ + corrélation — Toute sous-tâche définitivement échouée part en file de lettres mortes avec : correlation ID, nombre de tentatives, dernière erreur, référence de checkpoint. Le correlation ID traverse tous les logs de la mission.

Contraintes

Corrigé instructeur — Exercice 3 (solution de référence)

import time, random, uuid, json

# ---------- 1. Typologie d'erreurs ----------
class AgentError(Exception): retryable = False
class TransientError(AgentError): retryable = True    # timeout, rate limit
class FatalError(AgentError): retryable = False       # entrée invalide, policy
class CircuitOpenError(AgentError): retryable = False

def backoff(attempt):
    return min(8, 2 ** attempt) * random.uniform(0.5, 1.5)

def with_retry(fn, cid, name, max_attempts=3):
    # L'agent enrichisseur est un appel de LECTURE (pas d'effet de bord) :
    # il est donc idempotent par nature → retry sûr. S'il créait des
    # ressources, il faudrait une clé d'idempotence avant tout retry.
    last = None
    for attempt in range(max_attempts):
        try:
            return fn()
        except TransientError as e:
            last = e
            log(cid, f"{name} tentative {attempt+1} échouée : {e}")
            time.sleep(backoff(attempt) * 0.01)  # ×0.01 pour le test en classe
    raise last

# ---------- 2. Circuit breaker ----------
class CircuitBreaker:
    def __init__(self, threshold=5, cooldown=30 * 0.01):  # cooldown réduit pour la démo
        self.threshold, self.cooldown = threshold, cooldown
        self.failures, self.state, self.opened_at = 0, "closed", None
        self.opens = 0

    def call(self, fn):
        if self.state == "open":
            if time.time() - self.opened_at >= self.cooldown:
                self.state = "half_open"
            else:
                raise CircuitOpenError("circuit ouvert")
        try:
            result = fn()
        except TransientError:
            self.failures += 1
            if self.state == "half_open" or self.failures >= self.threshold:
                self.state, self.opened_at = "open", time.time()
                self.opens += 1
            raise
        self.failures, self.state = 0, "closed"
        return result

# ---------- Stubs ----------
class FlakyAgent:
    def __init__(self, name, fail_rate=0.0):
        self.name, self.fail_rate = name, fail_rate
    def run(self, task):
        if random.random() < self.fail_rate:
            raise TransientError(f"{self.name} : timeout simulé")
        return {"agent": self.name, "content": f"résultat({task})"}

def log(cid, msg): print(f"[{cid}] {msg}")

# ---------- 3 & 4. Orchestrateur avec dégradation + DLQ ----------
DLQ = []
breaker = CircuitBreaker()

def run_mission(task):
    cid = f"corr-{uuid.uuid4().hex[:8]}"
    results, missing = {}, []

    results["analyste"] = agent_analyste.run(task)          # fiable (simplifié)
    results["notes"] = agent_redacteur_notes.run(task)

    try:
        results["enrichisseur"] = with_retry(
            lambda: breaker.call(lambda: agent_enrichisseur.run(task)),
            cid, "enrichisseur")
    except AgentError as e:
        missing.append("enrichisseur")
        DLQ.append({
            "task_id": task, "correlation_id": cid,
            "attempts": 3 if not isinstance(e, CircuitOpenError) else 0,
            "last_error": type(e).__name__,
            "checkpoint_ref": f"ckpt://{task}/partial",
        })
        log(cid, f"enrichisseur abandonné → DLQ ({type(e).__name__})")

    return merge(results, missing, cid)

def merge(results, missing, cid):
    out = {"correlation_id": cid, "sections": results, "complete": not missing}
    if missing:
        out["avertissement"] = (f"Volets indisponibles : {', '.join(missing)} "
                                "(service en panne) — résultat partiel.")
    return out

# ---------- Test : 20 missions ----------
agent_analyste = FlakyAgent("analyste")
agent_enrichisseur = FlakyAgent("enrichisseur", fail_rate=0.4)
agent_redacteur_notes = FlakyAgent("notes")

full = degraded = 0
for i in range(20):
    m = run_mission(f"task-{i}")
    full += m["complete"]; degraded += not m["complete"]

print(f"\nComplets: {full} | Dégradés: {degraded} | DLQ: {len(DLQ)} | "
      f"Ouvertures circuit: {breaker.opens}")

Points de notation à vérifier :

Prolongement pour les rapides : ajouter un checkpoint réel (dict sérialisé en JSON) et une fonction replay_dlq() qui rejoue les entrées après « réparation » (fail_rate mis à 0) — démontre le cycle complet DLQ → correction → rejeu.

Fin des exercices — Session 4, niveau avancé. Les corrigés ne doivent pas être distribués avant la séance de restitution.