AI EngineeringZero to ProductionHome·About·Contact
Part I · Chapter 1

Environment & Your First API Call

Before any RAG or agent, you need a rock-solid understanding of the single API call — the atom every larger system is built from. By the end you'll have a working environment and will understand every field of a request and response.

⏱️ ~45 min🧪 5 labs💻 Python 3.10+🎯 Beginner
⚙️ To run this for realThe code here is complete and correct as written. To actually execute it you'll need:
  • an Anthropic API key (ANTHROPIC_API_KEY) + pip install anthropic
Reading and learning works without any of this — run when you're ready. IDs/ARNs in examples are placeholders; swap in your own.

Learning objectives

  • Install the SDK and configure credentials the way production teams do.
  • Make a first request and read back token usage and stop reasons.
  • Explain every field in a request: model, max_tokens, system, messages, thinking, effort.
  • Stream responses and handle errors like a real service.

Why we start with one call essential

It's tempting to jump straight to agents. Resist that. A RAG pipeline is many single calls plus retrieval; an agent is many single calls in a loop. If you don't deeply understand one request — what you pay for, why it stops, how it streams, how it fails — you'll be debugging blind later. This chapter is the foundation the whole course stands on.

Mental modelAn LLM call is a stateless function: you send the entire conversation every time, and you get back text (and/or tool requests). The API remembers nothing between calls. Everything "stateful" you build later is you re-sending history.

Lab 1.1 · Set up the environment essential

Example code for learning — review, test, and adapt it before running against real or production systems. Commands can create, change, or delete resources. See the Terms & Disclaimer.
Lab 1.1
  1. Create an isolated project. A virtual environment keeps this course's packages away from the rest of your system.
    terminalmkdir llm-course && cd llm-course
    python3 -m venv .venv
    source .venv/bin/activate       # Windows: .venv\Scripts\activate
  2. Install the SDK (plus a few packages we'll use in later chapters).
    terminalpip install anthropic pydantic python-dotenv
  3. Verify it imported.
    terminalpython -c "import anthropic; print(anthropic.__version__)"

    If that prints a version number, you're ready.

Lab 1.2 · API keys, the safe way essential

Your API key is a credential that spends real money. Never hardcode it in a .py file, never commit it to git. The standard pattern is an environment variable loaded from a .env file that's git-ignored.

Lab 1.2
  1. Get a key from your provider's console and copy it.
  2. Create a .env file in your project root:
    .envANTHROPIC_API_KEY=sk-ant-your-key-here
  3. Git-ignore it so it never leaves your machine:
    .gitignore.env
    .venv/
    __pycache__/
  4. Load it in Python. The SDK reads ANTHROPIC_API_KEY automatically once it's in the environment.
    config.pyfrom dotenv import load_dotenv
    load_dotenv()   # reads .env into os.environ
Never do thisclient = Anthropic(api_key="sk-ant-abc123...") — a hardcoded key gets committed, ends up in logs, and leaks. Also: never put a key in a prompt or message; prompts are logged.

Lab 1.3 · Your first call intermediate

Lab 1.3

Create first_call.py and type it out (don't paste — typing builds memory):

first_call.pyfrom dotenv import load_dotenv
from anthropic import Anthropic

load_dotenv()
client = Anthropic()                       # reads the key from the env

resp = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=300,
    system="You are a concise technical tutor.",
    messages=[
        {"role": "user", "content": "Explain what an API token is in 2 sentences."},
    ],
)

for block in resp.content:            # content is a LIST of blocks
    if block.type == "text":
        print(block.text)

print("---")
print("stop reason:", resp.stop_reason)
print("input tokens:", resp.usage.input_tokens)
print("output tokens:", resp.usage.output_tokens)

Run it:

terminalpython first_call.py

Expected shape of the output:

An API token is a unit of text... (a couple of sentences)
---
stop reason: end_turn
input tokens: 24
output tokens: 41
▶ How this works

This is the whole shape of every call you'll ever make to a language model: build a client, send a request describing what you want, then read the reply. If you understand these ~15 lines, you understand the API.

  1. load_dotenv() loads your secret API key from the .env file into the environment. client = Anthropic() creates the object that talks to the API — it picks up the key automatically, so the key never appears in your code.
  2. client.messages.create(...) sends the request. model picks which Claude to use; max_tokens caps how long the reply can be; system sets the assistant's role; messages is the conversation — here one user message.
  3. The reply's resp.content is a list of blocks, not a plain string. We loop over it and print only the text blocks. (Later, tool-use replies add other block types — that's why it's a list.)
  4. resp.stop_reason tells you why the model stopped (normally "end_turn"), and resp.usage reports how many tokens you were charged for — input and output separately.

What the output means: The model's 2-sentence explanation prints, then a separator, then the stop reason and the input/output token counts you'll use to compute cost.

Try this: Change the user content to your own question and re-run. Then lower max_tokens to 20 and watch stop_reason become "max_tokens" — the model was cut off, not finished.

Notice three things
  • content is a list of blocks, not a string. Always loop and check block.type — later there will be thinking and tool_use blocks too.
  • stop_reason: "end_turn" means the model finished naturally. Other values matter — see §Reading the response.
  • You're billed on input + output tokens. That 24/41 split is your cost model.

Anatomy of a request intermediate

the API is stateless — you send the whole conversation every call REQUEST model: "claude-opus-4-8" max_tokens: 1024 system: "You are…" messages: [ …history… ] thinking: adaptive POST /v1/messages RESPONSE content: [ {text…} ] stop_reason: "end_turn" usage: {in, out tokens} append reply → resend to continue next turn One request in, one response out — and no server-side memory. The model only knows what's in messages, so a multi-turn chat means appending each reply and resending the whole transcript. That statelessness is why "memory" is something you manage (Ch 4).
🗺️ How to read this diagram

This picture explains the single most surprising thing about the API: it has no memory. Each call is independent.

  • The left box (REQUEST) is everything you send: the model name, limits, your system prompt, and the messages list — the entire conversation so far.
  • The arrow is the network call (POST /v1/messages) — your request going up to Anthropic's servers.
  • The right box (RESPONSE) is what comes back: the content blocks, the stop reason, and token usage.
  • The key idea in the caption: to continue a conversation you must append the reply and resend the whole thing next time. The server didn't remember your last turn — you carry the history.

In short: "Stateless" means you own the conversation history, not the server. Every multi-turn chat you build is just this loop: send history → get reply → add it to history → send again.

Every field you sent, explained:

FieldWhat it isGuidance
modelWhich model serves the requestUse the exact ID string. Frontier for reasoning, small/fast for classification.
max_tokensHard cap on output lengthToo low → truncated mid-sentence. Default ~4096 for chat; higher needs streaming.
systemThe role/rules the model followsStable instructions. Kept separate from the conversation.
messagesThe conversation so farAlternating user/assistant. First must be user.
thinkingWhether the model reasons before answering{"type":"adaptive"} lets it decide depth. Great default for hard tasks.
output_config.effortHow hard the model workslow/medium/high/max. Trades cost/latency for quality.

The messages array is a transcript

To have a multi-turn conversation, you append to messages and resend the whole thing:

multi_turn.pymessages = [
    {"role": "user",      "content": "My name is Sam."},
    {"role": "assistant", "content": "Nice to meet you, Sam!"},
    {"role": "user",      "content": "What's my name?"},   # model sees the history
]
# The API is stateless — this whole list is sent every call.

Reading the response — stop reasons intermediate

Before you use content, check why the model stopped. This one habit prevents a whole class of production bugs.

stop_reasonMeaningWhat to do
end_turnFinished naturallyUse the content.
max_tokensHit the output cap — truncatedRaise max_tokens or stream; don't treat as complete.
tool_useModel wants to call a toolExecute it, send the result back (Chapter 4).
refusalDeclined for safetyDon't read content[0] blindly — it may be empty. Handle gracefully.
pause_turnLong server-side tool run pausedRe-send to resume.
Habit to build nowWrite if resp.stop_reason == "refusal": ... and == "max_tokens" branches from day one. Code that assumes content[0].text always exists will crash on refusals.

Lab 1.4 · Streaming advanced

For anything that produces more than a sentence or two, stream. It shows tokens as they arrive (better UX) and avoids HTTP timeouts on long outputs.

Lab 1.4
streaming.pyfrom dotenv import load_dotenv
from anthropic import Anthropic
load_dotenv()
client = Anthropic()

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=500,
    messages=[{"role":"user", "content":"Write a haiku about databases, then explain it."}],
) as stream:
    for text in stream.text_stream:      # yields text chunks as they arrive
        print(text, end="", flush=True)

    final = stream.get_final_message()     # full Message once done
    print("\n---\ntokens:", final.usage.output_tokens)
▶ How this works

Streaming shows the reply as it's generated, word by word, instead of waiting for the whole thing. It's the difference between a chatbot that types live and one that freezes then dumps a paragraph.

  1. with client.messages.stream(...) as stream: opens a streaming connection. The with block makes sure the connection is closed cleanly when you're done.
  2. for text in stream.text_stream: receives small chunks of text as the model produces them. print(text, end="", flush=True) prints each chunk immediately with no newline, so they flow together like live typing.
  3. stream.get_final_message() gives you the complete assembled reply once streaming finishes — handy for reading the total token usage.

What the output means: You see the haiku and its explanation appear progressively, then a token count on the final line.

Try this: Remove flush=True on some terminals and the text may appear in bursts instead of smoothly — flush forces each chunk to the screen right away.

When to streamAlways for user-facing chat, and for any request with large max_tokens. Use get_final_message() to still get token counts and stop reason even while streaming.

Lab 1.5 · Handling errors like a service advanced

Networks fail, rate limits trip, servers overload. The SDK auto-retries 429/5xx, but you should still catch and classify errors.

Lab 1.5
safe_call.pyimport anthropic
from anthropic import Anthropic
client = Anthropic()

def ask(prompt: str) -> str:
    try:
        resp = client.messages.create(
            model="claude-opus-4-8", max_tokens=500,
            messages=[{"role":"user","content":prompt}],
        )
    except anthropic.RateLimitError as e:      # 429
        return "Busy right now — try again shortly."
    except anthropic.APIStatusError as e:      # other non-2xx
        if e.status_code >= 500:
            return "Service issue — please retry."
        raise
    except anthropic.APIConnectionError:          # network
        return "Network problem — check your connection."

    if resp.stop_reason == "refusal":
        return "I can't help with that request."
    return next((b.text for b in resp.content if b.type=="text"), "")

print(ask("Give me one tip for learning to code."))
▶ How this works

Real services fail sometimes — rate limits, server hiccups, dropped network. This function wraps a call so that instead of crashing, it returns a friendly message for each kind of failure. This is the difference between a demo and something you'd let real users touch.

  1. try: attempts the API call. If it succeeds, we skip all the except blocks and go to the checks at the bottom.
  2. Each except catches a specific error: RateLimitError (too many requests, HTTP 429), APIStatusError (other bad responses — we only soften 500-level server errors and re-raise the rest), and APIConnectionError (network problem). Catching specific errors, not a blanket except, is good practice.
  3. After a successful call, if resp.stop_reason == "refusal" handles the model declining, and the final next((b.text ...), "") safely pulls the text out, defaulting to an empty string if there's no text block.

What the output means: For a normal prompt you get the model's tip. If the API were rate-limited, you'd get "Busy right now — try again shortly." instead of a stack trace.

Try this: This try/except pattern (from P4) is used all over the course's production code. Notice it catches named errors — you always want to know which failure happened.

Catch specific firstOrder your except blocks most-specific → least-specific. A single broad except Exception hides retryable-vs-fatal distinctions you need in production.

Exercises advanced

Exercise 1.1 — Token accountant

Context: Token counts are only useful once you turn them into money. Pricing the call is the smallest possible step from “it works” to “I know what it costs.”

Your task: Modify first_call.py to print the estimated cost of the call, assuming input costs $5 per 1M tokens and output $25 per 1M tokens.

Requirements:

  • Read both resp.usage.input_tokens and resp.usage.output_tokens
  • Apply (in*5 + out*25) / 1_000_000
  • Print the cost formatted to enough decimals to see a fractional-cent figure
  • Keep input and output priced at their separate rates

💡 Hint: Divide by 1_000_000 once at the end; an f-string like f"${cost:.6f}" shows the small numbers clearly.

Show solution

Illustrative fragment — defines demo values / files are needed before this runs standalone.

cost = (resp.usage.input_tokens * 5 +
        resp.usage.output_tokens * 25) / 1_000_000
print(f"cost: ${cost:.6f}")

Exercise 1.2 — Break it on purpose

Context: The fastest way to internalise stop_reason is to force a truncation on purpose and watch what the API reports — a lesson in never assuming a reply is complete.

Your task: Set max_tokens=10 and ask for a long essay, then inspect resp.stop_reason. Report what you get and what it implies for defensive coding.

Requirements:

  • Use a deliberately tiny max_tokens against a long request
  • Observe stop_reason become "max_tokens" with text cut off mid-sentence
  • State the lesson: check the stop reason before using content downstream
  • Note that a truncated reply must not be treated as a complete answer

💡 Hint: The cut-off text plus stop_reason == "max_tokens" is the whole point — the model was interrupted, not finished.

Show solution

stop_reason will be "max_tokens" and the text cuts off mid-sentence. Lesson: never assume a response is complete — check the stop reason before using the content downstream.

Exercise 1.3 — Mini chat loop

Context: A stateless API means “memory” is just you re-sending history. Building a terminal chat loop makes that trick concrete: append every turn and resend the whole list.

Your task: Build a terminal chat that keeps the conversation going — append each user input and each assistant reply to a messages list and resend it every turn.

Requirements:

  • Loop reading user input until a sentinel like quit ends it
  • Append each user message as {"role":"user",...} before the call
  • Resend the full messages list every turn — the API remembers nothing
  • Extract the assistant text safely from the block list and print it
  • Append the assistant reply back so it's part of the next turn's history

💡 Hint: The “memory” is entirely in your list: send history → get reply → append it → send again. Needs ANTHROPIC_API_KEY to actually run.

Show solution

Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.

messages = []
while True:
    user = input("you> ")
    if user == "quit": break
    messages.append({"role":"user","content":user})
    resp = client.messages.create(model="claude-opus-4-8",
             max_tokens=500, messages=messages)
    reply = next(b.text for b in resp.content if b.type=="text")
    print("bot>", reply)
    messages.append({"role":"assistant","content":reply})

Notice memory = you resending history. That's the whole trick.

🪜 Practice ladder beginner → industry

Six graded exercises, easy to real-world. Try each before opening its solution.

Exercise 1 · Read the token countsBeginner

Context: Every LLM call is billed on input plus output tokens, and the response object carries both counts. Reading them is the first habit toward understanding what a call actually costs.

Your task: Given a response object resp, print the input tokens, the output tokens, and their total, using the resp.usage fields from the lesson.

Requirements:

  • Read resp.usage.input_tokens and resp.usage.output_tokens
  • Total is the sum of the two
  • Keep it runnable with a tiny stand-in usage object (no API key)
  • Print all three values with clear labels

💡 Hint: You don't need a live call — a small class exposing input_tokens/output_tokens is enough to exercise the logic.

Show solution

The lesson shows resp.usage.input_tokens and resp.usage.output_tokens. Total is just their sum. Runnable with a tiny stand-in for the usage object:

class Usage:
    input_tokens = 24
    output_tokens = 41
class Resp:
    usage = Usage()

resp = Resp()
print("input:", resp.usage.input_tokens)
print("output:", resp.usage.output_tokens)
print("total:", resp.usage.input_tokens + resp.usage.output_tokens)
Exercise 2 · Extract text safely, then cost itIntermediate

Context: Two lesson facts combine here: resp.content is a list of blocks (not a string), and usage lets you price the call. Getting text out safely is what keeps code from crashing on tool-use or refusal replies.

Your task: Write a function that returns the concatenated text of all text blocks in resp.content, then print the call's cost at $5/1M input and $25/1M output tokens.

Requirements:

  • Iterate resp.content and keep only blocks where block.type == "text"
  • Never index content[0] blindly — join all text blocks
  • Apply the cost formula: (in*5 + out*25) / 1_000_000
  • Print the extracted text and the cost formatted to a few decimals
  • Runnable offline with fake block/usage objects

💡 Hint: A one-line generator inside "".join(...) filtered on block.type does the extraction; reuse Exercise 1.1's cost formula.

Show solution

Loop the content list and keep only block.type == "text" (never assume content[0]). Then apply the cost formula from Exercise 1.1. Runnable with fakes:

class Block:
    def __init__(self, type, text=""):
        self.type = type; self.text = text
class Usage:
    input_tokens = 24; output_tokens = 41
class Resp:
    content = [Block("text", "An API token is a chunk of text. "),
               Block("text", "Models are billed per token.")]
    usage = Usage()

def text_of(resp):
    return "".join(b.text for b in resp.content if b.type == "text")

resp = Resp()
print(text_of(resp))
cost = (resp.usage.input_tokens * 5 + resp.usage.output_tokens * 25) / 1_000_000
print(f"cost: ${cost:.6f}")
Exercise 3 · Dispatch on every stop_reasonAdvanced

Context: Before touching content, production code checks why the model stopped — this one habit prevents a whole class of bugs, from crashing on refusals to treating truncated output as complete.

Your task: Write handle(resp) that returns the right action string for each of the five stop_reason values (end_turn, max_tokens, tool_use, refusal, pause_turn) and does NOT read content[0].text on a refusal.

Requirements:

  • Branch on resp.stop_reason first, before reading any content
  • max_tokens must be flagged as truncated, not treated as complete
  • refusal path must not touch content (it may be empty)
  • Only pull text on the safe paths, via a helper that defaults to ""
  • Include a fallback for an unknown stop reason
  • Runnable with fake response objects, including one with empty content

💡 Hint: A first_text() helper using next((b.text ...), "") lets the refusal and empty-content cases pass through without an IndexError.

Show solution

Branch on the stop reason first, then only touch content on the safe paths. Note the max_tokens case must flag the text as truncated, not complete:

def first_text(resp):
    return next((b.text for b in resp.content if b.type == "text"), "")

def handle(resp):
    sr = resp.stop_reason
    if sr == "end_turn":
        return "use: " + first_text(resp)
    if sr == "max_tokens":
        return "TRUNCATED (raise max_tokens or stream): " + first_text(resp)
    if sr == "tool_use":
        return "run the requested tool, send result back"
    if sr == "refusal":
        return "handle refusal gracefully (content may be empty)"
    if sr == "pause_turn":
        return "re-send to resume"
    return "unknown stop reason"

class B:
    def __init__(self, t, x=""): self.type=t; self.text=x
class R:
    def __init__(self, sr, content): self.stop_reason=sr; self.content=content

print(handle(R("end_turn", [B("text","done.")])))
print(handle(R("max_tokens", [B("text","half a sen")])))
print(handle(R("refusal", [])))  # note: empty content, must not crash
Exercise 4 · Correct multi-turn historyExpert

Context: Because the API is stateless, a multi-turn chat is you re-sending the whole transcript — and the API rejects a first non-user message or two same-role turns in a row. Enforcing that invariant yourself catches bugs before the network does.

Your task: Write append_turn(messages, role, text) that appends a turn while asserting the roles strictly alternate user/assistant and that the first message is user — the invariant the lesson's messages table requires.

Requirements:

  • Assert role is one of user/assistant
  • First appended message must be user
  • Reject a turn whose role equals the previous message's role (no repeats)
  • Append {"role": role, "content": text} and return the list
  • Demonstrate both a valid transcript and a rejected first-assistant turn

💡 Hint: Two asserts cover it: an empty list requires user; otherwise role != messages[-1]["role"].

Show solution

The subtle correctness point: the API rejects a first non-user message and rejects two same-role messages in a row. Enforce it yourself:

def append_turn(messages, role, text):
    assert role in ("user", "assistant")
    if not messages:
        assert role == "user", "first message must be user"
    else:
        assert role != messages[-1]["role"], "roles must alternate"
    messages.append({"role": role, "content": text})
    return messages

m = []
append_turn(m, "user", "My name is Sam.")
append_turn(m, "assistant", "Nice to meet you, Sam!")
append_turn(m, "user", "What's my name?")
print(len(m), "turns; first role =", m[0]["role"])

for bad in [("assistant","x")]:  # first must be user
    try:
        append_turn([], *bad)
    except AssertionError as e:
        print("rejected:", e)
Exercise 5 · A resilient ask() with retry + classified errorsProfessional

Context: Networks fail, rate limits trip, servers overload. Production code wraps the SDK call in an explicit, observable layer that retries transient failures and still guards against refusals and truncation before trusting the text.

Your task: Harden the lesson's safe_call.py: wrap client.messages.create so it retries transient failures (429 / 5xx) with exponential backoff, catches each error type specifically, and checks stop_reason == "refusal" before reading text. (Needs an API key to run.)

Requirements:

  • Bounded retry loop with exponential backoff (e.g. 2 ** attempt)
  • Catch RateLimitError, APIStatusError, and APIConnectionError specifically, most- to least-specific
  • Retry 429 / 5xx; re-raise 4xx as fatal rather than hiding it
  • Return a friendly message on refusal and log/flag a max_tokens truncation
  • Extract text safely, defaulting to ""; add structured logging

💡 Hint: Treat only 429, 5xx, and connection errors as retryable; a 4xx is a caller bug you want to surface, not swallow.

Show solution

Real SDK code (the SDK already retries internally; this adds an explicit, observable layer). Needs an API key to run — it is not offline-runnable:

import time, logging
import anthropic
from anthropic import Anthropic

client = Anthropic()
log = logging.getLogger("ask")

def ask(prompt: str, retries: int = 3) -> str:
    for attempt in range(retries):
        try:
            resp = client.messages.create(
                model="claude-opus-4-8", max_tokens=500,
                messages=[{"role": "user", "content": prompt}],
            )
            if resp.stop_reason == "refusal":
                return "I can't help with that request."
            if resp.stop_reason == "max_tokens":
                log.warning("truncated output")
            return next((b.text for b in resp.content if b.type == "text"), "")
        except anthropic.RateLimitError:            # 429 — retryable
            wait = 2 ** attempt
            log.warning("rate limited; backoff %ss", wait); time.sleep(wait)
        except anthropic.APIStatusError as e:       # other non-2xx
            if e.status_code >= 500 and attempt < retries - 1:
                time.sleep(2 ** attempt); continue    # 5xx — retryable
            raise                                     # 4xx — fatal, don't hide
        except anthropic.APIConnectionError:        # network — retryable
            time.sleep(2 ** attempt)
    return "Service unavailable after retries."

print(ask("Give me one tip for learning to code."))

Key production points: specific exceptions ordered most->least specific, bounded retries with backoff, structured logging for observability, and the refusal / truncation guards from the lesson so downstream code never trusts an incomplete answer.

Exercise 6 · Cost-and-usage meter for a shared team keyIndustry scenario

Context: A team sharing one ANTHROPIC_API_KEY can't answer “what did RAG-search cost last month?” without per-call accounting. The single call is the unit of cost, so a thin meter at that seam turns usage into a finance report.

Your task: Design a wrapper around messages.create that tags every call with a feature label and accumulates input/output tokens and dollar cost per label. Provide a runnable accounting core and describe how you'd wire it to the real SDK.

Requirements:

  • Keep a per-feature ledger of input tokens, output tokens, and USD
  • Take price rates per model as input — do not hard-code a single model's price
  • Compute cost from resp.usage after each call: (in*pin + out*pout) / 1_000_000
  • Provide a report() that lists features, sorted by spend
  • Show the real SDK seam (a metered_call that reads resp.usage) as commented code needing a key
  • Accounting core runs offline; note that rates drift so versions/dates should be stamped

💡 Hint: Make the wrapper the only sanctioned entry point so every caller must pass a feature; a defaultdict keyed by label is the whole ledger.

Show solution

Design. The single call is the unit of cost, so the meter lives at exactly that seam. Wrap create(), read resp.usage after each call, and fold the numbers into a per-label ledger keyed by feature. Persist the ledger (DB / metrics backend) keyed by month; expose a report. Model tier is a per-call price so the meter must take rates as input, not hard-code one model.

The accounting core is pure Python and runnable; the commented line shows the real SDK seam (needs an API key there):

from collections import defaultdict

# price per 1M tokens, per model: (input, output)
PRICES = {
    "claude-opus-4-8":  (5.0, 25.0),
    "claude-haiku-4-5": (1.0,  5.0),
}

class CostMeter:
    def __init__(self):
        self.ledger = defaultdict(lambda: {"in": 0, "out": 0, "usd": 0.0})

    def record(self, feature, model, in_tok, out_tok):
        pin, pout = PRICES[model]
        usd = (in_tok * pin + out_tok * pout) / 1_000_000
        row = self.ledger[feature]
        row["in"] += in_tok; row["out"] += out_tok; row["usd"] += usd
        return usd

    def report(self):
        for feat, r in sorted(self.ledger.items(), key=lambda kv: -kv[1]["usd"]):
            print(f"{feat:16} {r['in']:>7} in {r['out']:>7} out  ${r['usd']:.4f}")

    # def metered_call(self, feature, **kw):
    #     resp = client.messages.create(**kw)     # real SDK — needs API key
    #     self.record(feature, kw["model"],
    #                 resp.usage.input_tokens, resp.usage.output_tokens)
    #     return resp

m = CostMeter()
m.record("rag-search",  "claude-haiku-4-5", 120_000, 30_000)
m.record("rag-search",  "claude-haiku-4-5",  80_000, 20_000)
m.record("chat",        "claude-opus-4-8",   40_000, 60_000)
m.report()

Tradeoffs. Tagging at the wrapper is cheap and accurate but relies on every caller passing a feature; enforce it by making the wrapper the only sanctioned entry point. Prices drift, so store the rate table with a version/date and stamp each ledger row so old months re-cost correctly. For high volume, emit each call as a metric (feature, model, tokens) to your observability stack and aggregate there instead of an in-process dict.

✓ Checkpoint — you can move on when you can…

  • Make a streaming and non-streaming call from a clean environment.
  • Explain model, max_tokens, system, messages, thinking, effort without looking.
  • Name what end_turn, max_tokens, and refusal mean and handle each.
  • Explain why the API being stateless means you manage conversation history.
🏗️ Toward the capstoneEverything you just learned is the atom of the AI DevOps Engineer you'll build. That agent is thousands of these calls in a loop — every time it runs kubectl get or terraform plan, it's one call like first_call.py. And checking stop_reason before acting? That habit becomes critical when the "output" is a decision to touch infrastructure. See the build-along plan →

Knowledge check check yourself

✓ Knowledge check

The lesson calls an LLM call a 'stateless function.' What practical consequence does that statelessness have for building a multi-turn conversation?

Show answer
The API remembers nothing between calls, so every turn you must resend the entire conversation in the messages array. Any 'memory' you build is just you appending each reply to the history and re-sending the whole transcript on the next call.
✓ Knowledge check

Why does the chapter insist you inspect stop_reason before using resp.content, and what bug does assuming content[0].text always exists cause?

Show answer
stop_reason tells you why the model stopped: end_turn (finished), max_tokens (truncated mid-answer, not complete), tool_use (wants a tool), or refusal (may have empty content). Code that reads content[0].text blindly will crash on a refusal and will treat a truncated max_tokens response as a complete answer downstream.
© 2026 studybydoing.in · AI Engineering: Zero to Production · All rights reserved. · About · Privacy Policy · Terms · Contact
Educational content, provided as-is and without warranty. Code samples are examples — review, test, and adapt them before using in production. See the Terms of Use & Disclaimer. Use at your own risk.
© studybydoing.in