# Exercises — Advanced Level, Session 1
# “Claude API: Deep Dive”

**Program:** Applied AI — Yann Isola
**Audience:** Solutions architects (preparation *Claude Certified Architect*)
**Estimated total duration:** 2:30 – 3 hours (outside of class or in a supervised workshop)
**Prerequisites:** Python 3.10+, SDK (SDK = Software Development Kit) `anthropic` installed, test API (API = Application Programming Interface) key.

> ⚠ All model names, prices and limits cited are **volatile**: check the official documentation (docs.anthropic.com) before hardcoding anything.

---

## Exercise 1 — API request construction: the robust client (60 min)

### Context

You are an architect at a tokenized instruments broker. The product team wants a Python module `claude_client.py` reusable by all internal departments. Your mission: design the reference call function, with all parameters controlled and all `stop_reason` managed.

### Instructions

**Part A — The complete request (20 min)**

Write a function `appeler_claude()` that:

1. Accepts: `question: str`, `persona: str`, `format_json: bool = False`, `max_tokens: int = 1024`.
2. Construct the query with:
- a **system prompt** (privileged channel: persona + exit rules — never in a `user` message);
- `temperature=0.2` (technical task);
- if `format_json=True`: a **prefilling** `assistant` with `"{"` to force the JSON opening (JSON = JavaScript Object Notation), and a `stop_sequences=["```"]` de sécurité.
3. Retourne un objet structuré `ReponseClaude(texte, stop_reason, input_tokens, output_tokens, cout_estime)`.

Squelette de départ :

```python
from dataclasses import dataclass
import anthropic

PRICE_INPUT_PAR_MTOK = $3.00 # / Mtok ⚠ volatile
PRICE_OUTPUT_PAR_MTOK = 15.00 # $ / Mtok ⚠ volatile

@dataclass
class ReplyClaude:
text: str
stop_reason:str
input_tokens: int
output_tokens: int
estimated_cost: float

def call_claude(question: str, persona: str,
format_json: bool = False,
max_tokens: int = 1024) -> ReponseClaude:
client = anthropic.Anthropic()
# ... to be completed ...```

**Partie B — La garde stop_reason (20 min)**

Complétez la fonction pour traiter **chacun** des quatre `stop_reason` :

- `end_turn` → retour nominal ;
- `max_tokens` → lever `ReponseTronqueeError` en incluant le texte partiel ET le compte de tokens (le client appelant décidera de relancer) ;
- `stop_sequence` → retour nominal + log de la séquence rencontrée (`response.stop_sequence`) ;
- `tool_use` → lever `NotImplementedError("tool loop hors périmètre session 1")`.

N'oubliez pas : si `format_json=True`, **re-préfixez** le `"{"` du prefill (il n'est pas inclus dans la réponse).

**Partie C — Le pré-comptage (20 min)**

Avant l'appel réel, utilisez l'endpoint `client.messages.count_tokens(...)` pour :

1. Compter les tokens d'entrée de la requête complète (system + messages).
2. Si `input_tokens + max_tokens > 200_000` ⚠ (fenêtre de contexte), lever `ContexteDebordeError` **sans consommer d'appel de génération**.
3. Logger l'estimation de coût **avant** l'appel.

### Livrables

- `claude_client.py` complet et exécutable.
- Un bloc de tests manuels (`if __name__ == "__main__":`) démontrant : un appel nominal, un appel JSON avec prefill, une troncature provoquée (`max_tokens=20`).

### Critères d'évaluation

| Critère | Points |
|---------|--------|
| Requête complète et paramètres justifiés (commentaires) | /6 |
| Les 4 `stop_reason` traités correctement | /6 |
| Prefill JSON re-préfixé côté client | /3 |
| Pré-comptage + garde de fenêtre de contexte | /3 |
| Qualité du code (typage, dataclass, logs) | /2 |
| **Total** | **/20** |

### Piège à éviter (indice)

`stop_reason: "max_tokens"` arrive avec un **HTTP 200**. Si votre gestion d'erreur ne regarde que les exceptions HTTP, vous livrerez des JSON tronqués en production.

---

## Exercice 2 — Implémentation du streaming SSE (50 min)

### Contexte

Le front-end de votre plateforme affiche les réponses de Claude en temps réel. Vous devez implémenter le consommateur de flux côté serveur, en traitant les **événements bruts** SSE (SSE = Server-Sent Events) — pas seulement le helper haut niveau — car la certification l'exige et votre équipe front a besoin des métadonnées fines.

### Consignes

**Partie A — Le consommateur d'événements (25 min)**

Implémentez `streamer_reponse()` qui consomme le flux bas niveau et maintient un état complet :

```python
def streamer_response(client, question: str) -> dict:
"""Consumes the SSE stream and returns the final state:
{text, stop_reason, output_tokens, events_recus (list of types), ttft_ms}
"""
import time
state = {"text": "", "stop_reason": None, "output_tokens": None,
"events_received": [], "ttft_ms": None}
start = time.monotonic()

with client.messages.stream(
model="claude-sonnet-4-5", # ⚠
max_tokens=800,
messages=[{"role": "user", "content": question}],
) as stream:
for event in stream:
status["events_received"].append(event.type)
# ... to complete: process each type of event ...
return state```

Exigences :

1. `message_start` → capturer l'`id` du message.
2. `content_block_delta` (sous-type `text_delta`) → accumuler le texte ; au **premier** delta, enregistrer le TTFT (TTFT = Time To First Token) en millisecondes.
3. `message_delta` → capturer `stop_reason` et `output_tokens` (rappel : ils n'arrivent QUE dans cet événement).
4. `message_stop` → clore proprement.
5. Ignorer les `ping` sans planter ; sur un événement `error`, lever une exception avec le détail.

**Partie B — L'assertion de séquence (15 min)**

Écrivez `verifier_sequence(evenements: list[str]) -> bool` qui valide l'ordre canonique :

```message_start
→ (content_block_start → content_block_delta* → content_block_stop)+
→ message_delta
→ message_stop```

Testez-la sur la liste `evenements_recus` de la partie A. Cette fonction sert de test d'intégration : si Anthropic change le protocole ou si votre parseur perd des événements, elle échoue bruyamment.

**Partie C — Question d'architecture (10 min, rédigé)**

En 10 lignes max : votre front-end est derrière un proxy inverse (nginx) qui **bufferise** les réponses HTTP. Quel est l'impact sur votre streaming, quel symptôme observera l'utilisateur, et quelles directives de configuration corrigent le problème ? (Indice : `proxy_buffering`, `X-Accel-Buffering`, `text/event-stream`.)

### Critères d'évaluation

| Critère | Points |
|---------|--------|
| Tous les types d'événements traités (y compris ping/error) | /7 |
| TTFT mesuré au bon endroit (premier text_delta) | /3 |
| stop_reason + output_tokens extraits de message_delta | /4 |
| Validateur de séquence correct | /4 |
| Question proxy : buffering identifié + correctifs | /2 |
| **Total** | **/20** |

### Piège à éviter (indice)

Chercher `stop_reason` dans `message_start` ou `message_stop` = 0 point sur le critère 3. Relisez la séquence.

---

## Exercice 3 — Conception d'un traitement par lots (Batches API) (60 min)

### Contexte

Votre société doit classifier **80 000 courriels clients** archivés (conformité) : catégorie, sentiment, présence de réclamation réglementaire. Pas d'exigence de latence — le rapport est mensuel. Budget serré. C'est un cas d'école pour la **Batches API** : traitement asynchrone, **−50 % ⚠** sur les coûts, SLA (SLA = Service Level Agreement) de **24 h ⚠**, jusqu'à **100 000 requêtes ⚠** par batch.

### Consignes

**Partie A — Document de conception (25 min, rédigé)**

Produisez une note d'architecture (1–2 pages) couvrant :

1. **Découpage** : un seul batch de 80k ou plusieurs ? Justifiez (limites ⚠ : 100k requêtes / ~256 Mo ⚠ par batch ; granularité de reprise sur erreur).
2. **Schéma de `custom_id`** : proposez un format traçable (ex. `mail-{lot}-{id_source}`) et expliquez pourquoi la corrélation par `custom_id` est **obligatoire** (les résultats ne reviennent pas dans l'ordre).
3. **Choix de modèle** : quel modèle pour de la classification simple à 80k exemplaires, et pourquoi ? (coût vs capacité)
4. **Cumul caching + batch** : votre prompt de classification contient 3 000 tokens de taxonomie identiques pour les 80k requêtes. Expliquez comment `cache_control` se combine avec le batch et estimez l'économie supplémentaire (hits non garantis ⚠).
5. **Gestion des états finaux** : `succeeded`, `errored`, `expired`, `canceled` — politique de rejeu pour chacun.
6. **Estimation de coût complète** : avec entrée ~3 300 tokens/requête et sortie ~150 tokens/requête, chiffrez le coût total avec et sans batch, avec et sans cache (tarifs ⚠ de la grille du jour, montrez vos calculs).

**Partie B — Implémentation du pipeline (35 min)**

Codez `pipeline_batch.py` avec quatre fonctions :

```python
def build_requests(emails: list[dict]) -> list[dict]:
"""Generates batch requests with traceable custom_id,
taxonomy system prompt marked cache_control,
and prefill helper '{' to force the JSON."""

def submit(client, requests: list[dict]) -> str:
"""Creates the batch, returns its id. Cut into several
batches if > 100_000 requests.""" # ⚠

def monitor(client, batch_id: str, interval_s: int = 60) -> None:
"""Polling processing_status until 'ended'.
Log request_counts on each iteration.
Backoff: don't hammer the API."""

def harvest(client, batch_id: str) -> tuple[list, list]:
"""Iterates the JSONL results (JSONL = JSON Lines).
Returns (successes, failures). Failures include custom_id
+ error type for targeted replay."""
```

Requirements:

1. `errored` results are written to `rejeu.jsonl`, ready to be resubmitted in a patch batch.
2. Each `succeeded` response is validated: parsable JSON AND `stop_reason == "end_turn"` (a classification truncated by `max_tokens` is a **silent failure** to catch).
3. Polling uses an increasing interval (60 s → 120 s → 300 s max): a batch can last for hours, no need to poll every second.

### Evaluation criteria

| Criterion | Points |
|--------|--------|
| Architectural note: the 6 points treated with figures | /8 |
| trackable custom_id + correct correlation | /3 |
| cache_control combined with batch | /3 |
| Stop_reason validation on each result | /3 |
| Chess Replay Pipeline | /3 |
| **Total** | **/20** |

### Trap to avoid (hint)

Two classic pitfalls: (1) assuming that the results come back in order of submission; (2) count only HTTP errors and pass truncated classifications (`stop_reason: "max_tokens"`) as successes.

---

## Overall scale

| Exercise | Weight |
|---------|-------|
| 1 — Robust client | 35% |
| 2 — SSE Streaming | 30% |
| 3 — Batch design | 35% |

**Session validation threshold:** 60%. The standard answers are provided by the trainer after submission.