Français
Applied AI · Advanced 🔴 · Session 4
✏️ Exercises
← Return to program 📄 Source .md

Applied AI — Advanced Level — Session 4

Exercises: Multi-agent architecture

Instructor: Yann Isola Total duration: 90 minutes Modality: individual or pair — the answers are intended for the instructor

Exercise 1 — Choice of decomposition strategy (25 min)

Context

You are a solutions architect at a broker in tokenized financial instruments. Three projects are arriving simultaneously. For each , you must choose a decomposition strategy (vertical, horizontal, recursive, hybrid — or no decomposition ) and justify it with the three decision criteria (complexity threshold, token budget, specialization benefit).

Case A — Product sheet generator

A repetitive task: from a 2-page technical sheet, generate a 300-word commercial description in a defined tone. Volume: 500 files/day. The technical sheet + the prompt are largely in context.

Case B — Due diligence of a counterparty

For each new counterparty: legal analysis (social structure, licenses), financial analysis (balance sheets, ratios), reputational analysis (press, sanctions), technical analysis (asset conservation infrastructure). Then production of a summary memo with GO/NO-GO recommendation. Each component requires voluminous documents and a distinct specialized vocabulary.

Case C — Documentary migration of a fund

A collection transmits a tree structure of ~4,000 documents (unpredictable depth and structure: nested files, heterogeneous formats, mixed languages). Each document must be classified, summarized, indexed and linked to a regulatory framework. The structure of the tree is only known at runtime.

Expected deliverables (per case)

  1. Strategy chosen (one sentence).
  2. Justification by the 3 criteria (table).
  3. Architectural diagram (ASCII or drawing): agents, flows, points of convergence.
  4. For case C only: the three guardrails requirements of your architecture and their proposed values.

Instructor answer key — Exercise 1

Case A — No decomposition.

Case B — Hybrid: horizontal for the analysis, vertical for the whole.

Case C — Recursive.

Indicative scale (out of 20): Case A: 5 pts (including 2 for resisting over-engineering) · Case B: 8 pts (including 3 for the correctly articulated hybrid) · Case C: 7 pts (including 3 for the encrypted guardrails).

Exercise 2 — Designing an escalation policy (30 min)

Context

A multi-agent customer service assistant operates for a digital asset custody platform. It may: answer questions (products, fees, procedures), modify non-sensitive profile information, initiate account recovery procedures, and prepare (but not execute) withdrawal instructions.

Three recent incidents motivate an overhaul of the escalation policy:

Work requested

  1. Trigger Matrix (8 pts) — For each of the agent's 4 capabilities, define the escalation triggers according to the three families (trust below threshold / outside perimeter / risk level). Present in table form. Each incident must be covered by at least one trigger.
  2. Choice of patterns (6 pts) — For each trigger, associate the appropriate pattern (pause-and-ask / queue-for-review / fallback-to-human) and justify in one sentence.
  3. SLA policy (6 pts) — Design the response to incident 3: priority levels, target deadlines, timeout action by level (fail-safe / fail-operational / re-escalation), and two metrics piloting.

Constraint

Your policy must remain actionable: if everything escalates in P1 to a human, you have failed. Explicitly state what remains autonomous .

Instructor answer key — Exercise 2

1. Trigger matrix – expected elements:

Ability Trust Outside the perimeter Risk
Product/fee questions Separate judge, calibrated threshold (e.g. 0.75) → escalation if below Tax/legal/investment advice detected → escalation (covers incident 2) Low → standalone by default
Non-sensitive profile modification — Request sliding to sensitive data (IBAN, 2FA) → escalation Low/reversible → standalone + logging
Account Recovery Inconsistencies in verifications → escalation — Raised by nature : fraud signal detector (urgency, pressure, inconsistencies) → mandatory escalation (covers incident 1)
Preparing for withdrawal — — High/irreversible downstream → systematic prior human approval beyond an amount threshold

Key points to check: (a) Incident 2 is handled by outside the scope (regulated tax advice), not just by trust — a confidently invented answer precisely has high self-reported trust, hence the need for the scope classifier and/or a separate judge; (b) Incident 1 requires a trigger risk independent of trust.

2. Expected patterns:

3. SLA — typical answer:

Exercise 3 — Implementing error handling (35 min)

Context

You receive the skeleton of an orchestrator that calls three subagents in parallel and then merges their results. In production, the agent API enrichisseur is experiencing intermittent outages that are currently causing entire missions to fail.

# --- 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

Work requested

Rewrite run_mission and its infrastructure to integrate, in this order:

  1. (6 pts) Typology + retry with backoff — Classify errors (retryable / fatal). Implement a retry (max 3 attempts) with exponential backoff and jitter, applied uniquely to retryable errors. The call to agent_enrichisseur can it be replayed without precaution? Justify in one sentence in a comment (hint: idempotence).
  2. (6 pts) Circuit breaker — Implement a circuit breaker (threshold: 5 failures, cooling: 30 s) around agent_enrichisseur uniquely. All three states must be explicit in the code.
  3. (4 pts) Graceful degradation — If the enricher is unavailable (open circuit or exhausted retries), the mission must still succeed with a partial result, carrying an explicit mention of the missing component. merge must never rise again KeyError .
  4. (4 pts) DLQ + correlation — Any definitively failed subtask goes into a queue of dead letters with: correlation ID, number of attempts, last error, checkpoint reference. The correlation ID crosses all mission logs.

Constraints

Instructor answer key — Exercise 3 (reference solution)

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}")

Scoring points to check:

Extension for the rapids: add a real checkpoint (dict serialized in JSON) and a function replay_dlq() which replays inputs after “repair” (fail_rate set to 0) — demonstrates the complete DLQ → fix → replay cycle.

End of exercises — Session 4, advanced level. Answer keys should not be distributed before the feedback session.