# Exercices — Session 3 (Avancé) : Claude Agent SDK > **Programme :** Applied AI — Yann Isola > **Public :** architectes solutions — préparation *Claude Certified Architect* > **Prérequis techniques :** Python ≥ 3.10, `pip install claude-agent-sdk` ⚠ (vérifier le nom exact du paquet dans la documentation officielle), clé API (Application Programming Interface — interface de programmation applicative) valide. Chaque exercice comporte : contexte métier, cahier des charges, contraintes, critères d'évaluation, et corrigé commenté. **Ne lisez le corrigé qu'après votre tentative.** --- ## Exercice 1 — Construire un agent avec outils (≈ 45 min) ### Contexte Vous équipez le support interne d'un courtier en instruments financiers tokenisés. L'agent doit répondre aux questions des opérateurs sur l'état des règlements-livraisons (settlement) en interrogeant deux systèmes internes (simulés ici par des fonctions Python). ### Cahier des charges 1. Créer deux outils avec le décorateur `@tool` : - `statut_reglement(reference: str) -> str` — retourne le statut d'une opération (simulez avec un dictionnaire : `"SETL-001": "réglée"`, `"SETL-002": "en attente de collatéral"`, `"SETL-003": "suspens — incident dépositaire"`). - `lister_suspens(gravite: str = "toutes") -> str` — liste les opérations en suspens, filtrables par gravité (`"critique"` / `"mineure"` / `"toutes"`). 2. Créer un `Agent` nommé `ops-settlement` avec des `instructions` imposant : réponse en français, citation systématique de la référence d'opération, refus poli de toute question hors périmètre règlement-livraison. 3. Exécuter via `Runner.run()` sur trois requêtes de test : - « Où en est SETL-002 ? » - « Liste-moi les suspens critiques. » - « Quel temps fait-il à Genève ? » (doit être refusée) ### Contraintes de qualité (évaluées) - **Docstrings d'outils** : chaque docstring doit indiquer *quand* utiliser l'outil et décrire chaque paramètre. Une docstring d'une ligne vague = échec du critère. - **Typage complet** : annotations sur tous les paramètres et le retour. - **Bornage** : `lister_suspens` doit valider `gravite` et retourner un message d'erreur textuel actionnable si la valeur est invalide (pas d'exception pour une erreur métier). ### Critères d'évaluation (10 pts) | Critère | Pts | |---|---| | Outils fonctionnels, schémas corrects | 3 | | Docstrings de qualité « prompt engineering » | 2 | | Instructions : périmètre + refus hors-sujet effectif | 2 | | Erreur métier retournée en texte (pas levée) | 2 | | Les 3 requêtes de test se comportent comme attendu | 1 | ### Corrigé commenté ```python from claude_agent_sdk import Agent, Runner, tool REGLEMENTS = { "SETL-001": {"statut": "réglée", "gravite": None}, "SETL-002": {"statut": "en attente de collatéral", "gravite": "mineure"}, "SETL-003": {"statut": "suspens — incident dépositaire", "gravite": "critique"}, } @tool def statut_reglement(reference: str) -> str: """Retourne le statut d'une opération de règlement-livraison. À utiliser dès que l'utilisateur mentionne une référence d'opération (format SETL-XXX) et demande son état. Args: reference: référence de l'opération, ex. "SETL-002". """ op = REGLEMENTS.get(reference.strip().upper()) if op is None: # Erreur MÉTIER → texte actionnable, pas d'exception : # le modèle peut rebondir (demander la bonne référence). return (f"Aucune opération trouvée pour '{reference}'. " f"Vérifier le format (SETL-XXX) ou la date de valeur.") return f"{reference} : {op['statut']}" @tool def lister_suspens(gravite: str = "toutes") -> str: """Liste les opérations en suspens (non réglées). À utiliser pour toute demande de vue d'ensemble des suspens, éventuellement filtrée par gravité. Args: gravite: "critique", "mineure" ou "toutes" (défaut). """ if gravite not in ("critique", "mineure", "toutes"): return ("Valeur de gravité invalide. " "Valeurs acceptées : critique, mineure, toutes.") lignes = [ f"{ref} : {op['statut']} (gravité : {op['gravite']})" for ref, op in REGLEMENTS.items() if op["gravite"] and (gravite == "toutes" or op["gravite"] == gravite) ] return "\n".join(lignes) if lignes else "Aucun suspens pour ce filtre." agent_ops = Agent( name="ops-settlement", model="claude-sonnet-4-5", # ⚠ identifiant de modèle volatile instructions=( "Tu es l'assistant des opérateurs règlement-livraison d'un courtier. " "Règles impératives :\n" "1. Réponds exclusivement en français.\n" "2. Cite TOUJOURS la référence d'opération (SETL-XXX) dans ta réponse.\n" "3. Ton périmètre est STRICTEMENT le règlement-livraison. Pour toute " "autre demande, décline poliment en une phrase et rappelle ton périmètre." ), tools=[statut_reglement, lister_suspens], ) for question in ["Où en est SETL-002 ?", "Liste-moi les suspens critiques.", "Quel temps fait-il à Genève ?"]: print(Runner.run(agent_ops, question).final_output, "\n") ``` **Points de correction à discuter :** - La validation de `gravite` retourne un texte : le modèle peut alors reformuler sa demande — comportement auto-correctif impossible avec une exception brute. - Le refus hors-périmètre est porté par les `instructions` ; en production on ajouterait un output guardrail (exercice 3) pour le garantir. --- ## Exercice 2 — Implémenter des handoffs (≈ 60 min) ### Contexte Guichet unique de support d'une plateforme de tokenisation : les demandes arrivent mélangées (technique, conformité/KYC — Know Your Customer, connaissance client —, facturation). Vous devez construire un **agent de triage** qui route par handoff vers trois spécialistes. ### Cahier des charges 1. Trois agents spécialistes : `tech`, `conformite`, `facturation` — chacun avec des `instructions` propres et **au moins un outil factice** pertinent. 2. Un agent `triage` : - modèle léger (ex. `claude-haiku-4-5` ⚠), - `handoffs` vers les trois spécialistes, - instructions interdisant explicitement de résoudre lui-même. 3. Un callback `on_handoff` (via l'option du handoff ou un hook) qui journalise chaque transfert : `triage → conformite (raison)`. 4. Tests : trois demandes, une par spécialité, plus une demande **ambiguë** (« Ma vérification d'identité bloque et en plus j'ai été facturé deux fois ») — observer et commenter le choix du modèle. ### Contraintes - L'agent cible doit exploiter l'historique : le spécialiste ne doit PAS redemander ce que l'utilisateur a déjà dit (vérifiez dans la sortie). - Prévoir un **handoff retour** : chaque spécialiste peut retransférer au triage s'il est hors de son périmètre. - Borner le run (`max_turns`) pour éviter le ping-pong triage ↔ spécialiste. ### Critères d'évaluation (10 pts) | Critère | Pts | |---|---| | Routage correct des 3 demandes claires | 3 | | Journalisation des handoffs opérationnelle | 2 | | Historique exploité (pas de re-questionnement) | 2 | | Handoff retour + bornage anti-boucle | 2 | | Analyse écrite du cas ambigu (5–10 lignes) | 1 | ### Corrigé commenté (extraits clés) ```python from claude_agent_sdk import Agent, Runner, handoff, tool @tool def verifier_dossier_kyc(client_id: str) -> str: """Vérifie l'état du dossier KYC (Know Your Customer) d'un client. Args: client_id: identifiant client, ex. "C-1024". """ return f"Dossier {client_id} : pièce d'identité expirée, à renouveler." def log_handoff(source: str, cible: str): def _cb(ctx): print(f"[HANDOFF] {source} → {cible}") return _cb agent_conformite = Agent( name="conformite", model="claude-sonnet-4-5", # ⚠ volatile instructions=( "Spécialiste conformité/KYC. Tu disposes de l'historique complet : " "ne redemande jamais une information déjà fournie. " "Si la demande sort de la conformité, retransfère au triage." ), tools=[verifier_dossier_kyc], # handoff retour — ajouté après création du triage (voir note) ) # ... agents tech et facturation analogues ... agent_triage = Agent( name="triage", model="claude-haiku-4-5", # ⚠ volatile — router = tâche simple, modèle léger instructions=( "Tu es un aiguilleur. Analyse la demande et transfère au bon " "spécialiste : tech (bugs, API), conformite (KYC, identité), " "facturation (paiements, factures). Tu ne résous JAMAIS toi-même. " "Si plusieurs sujets coexistent, choisis le plus bloquant pour " "le client et mentionne l'autre sujet dans ton transfert." ), handoffs=[ handoff(agent_tech, on_handoff=log_handoff("triage", "tech")), handoff(agent_conformite, on_handoff=log_handoff("triage", "conformite")), handoff(agent_facturation, on_handoff=log_handoff("triage", "facturation")), ], ) # Handoff retour : câblé après coup pour éviter la référence circulaire. agent_conformite.handoffs = [handoff(agent_triage, on_handoff=log_handoff("conformite", "triage"))] res = Runner.run(agent_triage, "Ma vérification d'identité bloque et j'ai été facturé deux fois.", max_turns=12) # ⬅ anti-boucle print(res.final_output) ``` **Points de correction :** - **Référence circulaire** : triage ↔ spécialistes s'entre-référencent ; on câble le handoff retour *après* l'instanciation. Question d'architecture classique. - **Cas ambigu** : il n'y a pas de « bonne » réponse unique — ce qu'on évalue, c'est que le candidat a *anticipé* l'ambiguïté dans les instructions du triage (règle « le plus bloquant d'abord, mentionner l'autre sujet »). Alternative défendable : demander une clarification à l'utilisateur avant de router. - **`max_turns=12`** : sans borne, un spécialiste mal instruit peut retransférer indéfiniment. --- ## Exercice 3 — Concevoir des guardrails (≈ 60 min) ### Contexte Votre agent `facturation` de l'exercice 2 part en production dans un contexte régulé. Direction des risques : « aucune donnée de carte ne doit entrer, aucune promesse d'engagement financier ne doit sortir, et tout doit être auditable ». ### Cahier des charges 1. **Input guardrail** `bloquer_pan` : détecte un PAN (Primary Account Number — numéro de carte bancaire, 16 chiffres éventuellement séparés par espaces/tirets) dans le message entrant et déclenche le tripwire avec un message d'orientation vers le canal sécurisé. 2. **Output guardrail** `bloquer_engagement` : bloque toute sortie contenant une promesse ferme de remboursement ou un montant garanti. Deux implémentations à livrer et comparer : - v1 : règles/regex (mots déclencheurs : « je vous garantis », « vous serez remboursé de », montants + « sous X jours »…) ; - v2 : LLM-as-judge — un modèle léger classifie la sortie (`engagement_ferme` / `information`), avec un prompt de juge que vous rédigez. 3. **Hooks d'audit** : classe `RunHooks` journalisant `on_tool_start/end` et les déclenchements de guardrails (horodatage, nom d'agent, motif) dans un fichier JSONL (JSON Lines — un objet JSON par ligne). 4. **Gestion applicative** : intercepter les exceptions tripwire et retourner à l'utilisateur un message de repli propre (jamais de stack trace). 5. **Jeu de tests** : ≥ 6 cas — 2 entrées avec PAN (formats différents), 1 entrée saine, 2 sorties engageantes, 1 sortie informative. Mesurer faux positifs / faux négatifs de v1 vs v2. ### Critères d'évaluation (10 pts) | Critère | Pts | |---|---| | Input guardrail : détection des 2 formats de PAN, entrée saine passante | 2 | | Output guardrail v1 (règles) fonctionnel | 2 | | Output guardrail v2 (LLM-as-judge) : prompt de juge rigoureux | 2 | | Comparaison v1/v2 chiffrée (tableau FP/FN) + recommandation argumentée | 2 | | Hooks d'audit JSONL + repli applicatif propre | 2 | ### Corrigé commenté (extraits clés) ```python import re, json, datetime from claude_agent_sdk import (Agent, Runner, RunHooks, input_guardrail, output_guardrail, GuardrailTripwire, InputGuardrailTripwireTriggered, OutputGuardrailTripwireTriggered) PAN_RE = re.compile(r"\b(?:\d[ -]?){15}\d\b") # 16 chiffres, séparateurs tolérés @input_guardrail def bloquer_pan(ctx, agent, message: str) -> GuardrailTripwire: """Refuse tout message contenant un numéro de carte (PAN).""" if PAN_RE.search(message): return GuardrailTripwire( triggered=True, message=("Par sécurité, ne transmettez jamais de numéro de carte ici. " "Utilisez le portail sécurisé : Espace client → Paiements."), ) return GuardrailTripwire(triggered=False) # ---- Output v1 : règles ---- MOTIFS_ENGAGEMENT = [ r"je vous garantis", r"vous serez rembours", r"\d+ ?(€|EUR|CHF).{0,40}sous \d+ jours", ] @output_guardrail def bloquer_engagement_v1(ctx, agent, sortie: str) -> GuardrailTripwire: """Bloque toute promesse ferme d'engagement financier (règles).""" hit = any(re.search(p, sortie, re.IGNORECASE) for p in MOTIFS_ENGAGEMENT) return GuardrailTripwire(triggered=hit) # ---- Output v2 : LLM-as-judge ---- PROMPT_JUGE = """Tu es un contrôleur conformité. Classe le texte suivant : - "engagement_ferme" : promesse contraignante (remboursement garanti, montant précis dû, délai ferme engageant la société). - "information" : explication de procédure, statut, conditionnel ("sous réserve de validation", "généralement"). Réponds par UN seul mot. Texte : \"\"\"{sortie}\"\"\"""" @output_guardrail def bloquer_engagement_v2(ctx, agent, sortie: str) -> GuardrailTripwire: """Bloque les engagements fermes (juge LLM léger).""" verdict = appel_modele_leger(PROMPT_JUGE.format(sortie=sortie)) # ⚠ coût/latence return GuardrailTripwire(triggered=verdict.strip() == "engagement_ferme") # ---- Hooks d'audit JSONL ---- class AuditHooks(RunHooks): def _log(self, **champs): champs["ts"] = datetime.datetime.utcnow().isoformat() with open("audit.jsonl", "a") as f: f.write(json.dumps(champs, ensure_ascii=False) + "\n") async def on_tool_start(self, ctx, agent, tool): self._log(evt="tool_start", agent=agent.name, tool=tool.name) async def on_tool_end(self, ctx, agent, tool, result): self._log(evt="tool_end", agent=agent.name, tool=tool.name) # ---- Repli applicatif ---- def repondre(agent, message): try: return Runner.run(agent, message, hooks=AuditHooks()).final_output except InputGuardrailTripwireTriggered as e: return e.guardrail_message # message d'orientation, pas de trace except OutputGuardrailTripwireTriggered: return ("Votre demande nécessite une validation manuelle. " "Un conseiller vous recontacte sous 24 h ouvrées.") ``` **Comparaison attendue v1 vs v2 (exemple de résultat) :** | | Faux positifs | Faux négatifs | Latence | Coût | |---|---|---|---|---| | v1 règles | Moyens (bloque « vous serez remboursé *si* la fraude est confirmée ») | Élevés (paraphrases non listées) | ~0 ms | 0 | | v2 juge LLM | Faibles | Faibles | +200–800 ms ⚠ | ~1 appel léger/réponse ⚠ | **Recommandation type (à argumenter) :** v1 en première ligne (rapide, gratuite) **ET** v2 en seconde ligne sur les sorties que v1 laisse passer — défense en profondeur, coût du juge payé seulement quand nécessaire. Toute réponse justifiant un autre arbitrage (ex. v2 seul pour un flux à faible volume) est acceptable si le raisonnement coût/latence/risque est explicite. --- ## Barème global - Exercice 1 : /10 — Exercice 2 : /10 — Exercice 3 : /10. - ≥ 24/30 : niveau attendu pour la certification sur ce domaine. - Entre 18 et 23 : revoir handoffs et guardrails (séquences 3–4 du cours). - < 18 : refaire l'exercice 1 avec le corrigé masqué, puis progresser.