Exercises — Advanced Level, Session 1

“Claude API: Deep Dive”

Program : Applied AI — Yann Isola Audience : Solution 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 services. Your mission: design the reference call function, with all the parameters controlled and all the stop_reason managed.

Instructions

Part A — The complete request (20 min)

Write a function appeler_claude() Who :

1. Accept: question: str , persona: str , format_json: bool = False , max_tokens: int = 1024.
2. Build the query with:
   • A system prompt (privileged channel: persona + exit rules — never in a message user ) ;
   • temperature=0.2 (technical task);
   • if format_json=True : A prefilling assistant with "{" to force the opening of JSON (JSON = JavaScript Object Notation), and a stop_sequences=["```"] security.
3. Return a structured object ReponseClaude(texte, stop_reason, input_tokens, output_tokens, cout_estime).

Starting skeleton:

from dataclasses import dataclass
import anthropic

PRIX_INPUT_PAR_MTOK = 3.00    # $ / Mtok ⚠ volatil
PRIX_OUTPUT_PAR_MTOK = 15.00  # $ / Mtok ⚠ volatil

@dataclass
class ReponseClaude:
    texte: str
    stop_reason: str
    input_tokens: int
    output_tokens: int
    cout_estime: float

def appeler_claude(question: str, persona: str,
                   format_json: bool = False,
                   max_tokens: int = 1024) -> ReponseClaude:
    client = anthropic.Anthropic()
    # ... à compléter ...

Part B — The guard stop_reason (20 min)

Complete the function to process each of the four stop_reason :

• end_turn → nominal return;
• max_tokens → lift ReponseTronqueeError by including the partial text AND the token count (the calling client will decide to restart);
• stop_sequence → nominal return + log of the sequence encountered (response.stop_sequence ) ;
• tool_use → lift NotImplementedError("tool loop hors périmètre session 1").

Remember: if format_json=True , re-prefix THE "{" of the prefill (it is not included in the response).

Part C — The pre-count (20 min)

Before the actual call, use the endpoint client.messages.count_tokens(...) For :

1. Count the input tokens of the complete request (system + messages).
2. If input_tokens + max_tokens > 200_000 ⚠ (context window), raise ContexteDebordeError without consuming a generation call .
3. Log the cost estimate Before the call.

Deliverables

• claude_client.py complete and executable.
• A block of manual tests (if __name__ == "__main__":) demonstrating: a roll call, a JSON call with prefill, a truncation caused (max_tokens=20).

Evaluation criteria

Criteria	Points
Complete query and justified parameters (comments)	/6
The 4 stop_reason treated correctly	/6
Client-side re-prefixed JSON prefill	/3
Pre-count + context window guard	/3
Code quality (typing, dataclass, logs)	/2
Total	/20

Trap to avoid (hint)

stop_reason: "max_tokens" arrives with a HTTP200. If your error handling only looks at HTTP exceptions, you will deliver truncated JSON to production.

---

Exercise 2 — Implementing SSE streaming (50 min)

Context

The front-end of your platform displays responses from Claude in real time. You must implement the stream consumer on the server side, processing the raw events SSE (SSE = Server-Sent Events) — not just the high-level helper — because the certification requires it and your front-end team needs fine-grained metadata.

Instructions

Part A — The Event Consumer (25 min)

Implement streamer_reponse() which consumes the low-level flow and maintains a complete state:

def streamer_reponse(client, question: str) -> dict:
    """Consomme le flux SSE et retourne l'état final :
    {texte, stop_reason, output_tokens, evenements_recus (liste des types), ttft_ms}
    """
    import time
    etat = {"texte": "", "stop_reason": None, "output_tokens": None,
            "evenements_recus": [], "ttft_ms": None}
    debut = 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:
            etat["evenements_recus"].append(event.type)
            # ... à compléter : traiter chaque type d'événement ...
    return etat

Requirements:

1. message_start → capture the id of the message.
2. content_block_delta (subtype text_delta ) → accumulate the text; At first delta, record the TTFT (TTFT = Time To First Token) in milliseconds.
3. message_delta → capture stop_reason And output_tokens (reminder: they ONLY happen in this event).
4. message_stop → close properly.
5. Ignore the ping without planting; on an event error , throw an exception with the detail.

Part B — The sequence assertion (15 min)

Write verifier_sequence(evenements: list[str]) -> bool which validates the canonical order:

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

Test it on the list evenements_recus from part A. This function serves as an integration test: if Anthropic changes the protocol or if your parser loses events, it fails loudly.

Part C — Question of architecture (10 min, written)

In 10 lines max: your front-end is behind a reverse proxy (nginx) which buffer HTTP responses. What is the impact on your streaming, what symptom will the user observe, and what configuration guidelines correct the problem? (Hint : proxy_buffering , X-Accel-Buffering , text/event-stream .)

Evaluation criteria

Criteria	Points
All event types processed (including ping/error)	/7
TTFT measured in the right place (first text_delta)	/3
stop_reason + output_tokens taken from message_delta	/4
Correct Sequence Validator	/4
Proxy question: buffering identified + fixes	/2
Total	/20

Trap to avoid (hint)

Seek stop_reason In message_start Or message_stop = 0 points on criterion 3. Reread the sequence.

---

Exercise 3 — Designing a batch process (Batches API) (60 min)

Context

Your company must classify 80,000 customer emails archived (compliance): category, sentiment, presence of regulatory complaint. No latency requirements — reporting is monthly. Tight budget. This is a textbook case for API batches : asynchronous processing, −50 % ⚠ on costs, SLA (SLA = Service Level Agreement) of 24 hours ⚠, until 100,000 requests ⚠ per batch.

Instructions

Part A — Design document (25 min, written)

Produce an architectural note (1–2 pages) covering:

1. Cutting : a single batch of 80k or several? Justify (limits ⚠: 100k requests / ~256 MB ⚠ per batch; error recovery granularity).
2. Diagram of custom_id : propose a traceable format (e.g. mail-{lot}-{id_source}) and explain why the correlation by custom_id East OBLIGATORY (results do not return in order).
3. Model choice : which model for simple classification with 80k copies, and why? (cost vs capacity)
4. Cumulative caching + batch : your classification prompt contains 3,000 identical taxonomy tokens for the 80k queries. Explain how cache_control combines with the batch and estimate the additional saving (hits not guaranteed ⚠).
5. End state management : succeeded , errored , expired , canceled — replay policy for everyone.
6. Complete cost estimate : with input ~3,300 tokens/request and output ~150 tokens/request, calculate the total cost with and without batch, with and without cache (prices ⚠ from today's grid, show your calculations).

Part B — Pipeline Implementation (35 min)

Code pipeline_batch.py with four functions:

def construire_requetes(emails: list[dict]) -> list[dict]:
    """Génère les requêtes batch avec custom_id traçables,
    system prompt de taxonomie marqué cache_control,
    et prefill assistant '{' pour forcer le JSON."""

def soumettre(client, requetes: list[dict]) -> str:
    """Crée le batch, retourne son id. Découpe en plusieurs
    batches si > 100_000 requêtes."""  # ⚠

def surveiller(client, batch_id: str, intervalle_s: int = 60) -> None:
    """Polling de processing_status jusqu'à 'ended'.
    Logge request_counts à chaque itération.
    Backoff : ne pas marteler l'API."""

def recolter(client, batch_id: str) -> tuple[list, list]:
    """Itère les résultats JSONL (JSONL = JSON Lines).
    Retourne (succes, echecs). Les echecs incluent custom_id
    + type d'erreur pour rejeu ciblé."""

Requirements:

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

Evaluation criteria

Criteria	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 the order of submission; (2) count only HTTP errors and let truncated classifications pass (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.