LLM APIs fundamentals.

Setup & API Keys

Each provider issues an API key from its own console. Treat every one of these exactly like a database password — never hardcode it, never commit it, never log it.

Load keys from environment variables, not from source code — this is the Python equivalent of keeping secrets out of local.properties / BuildConfig in an Android project.

pip install openai anthropic google-genai python-dotenv

OPENAI_API_KEY=sk-...
GEMINI_API_KEY=AIza...
ANTHROPIC_API_KEY=sk-ant-...

from dotenv import load_dotenv
import os

load_dotenv()  # reads .env into the process environment
api_key = os.environ["OPENAI_API_KEY"]  # raises KeyError if missing

Anatomy of a Chat Completion Request

OpenAI's chat.completions shape became the de facto pattern the rest of the industry converged on. Learn it first; everything else in this lesson is a variation on it.

from openai import OpenAI

client = OpenAI()  # reads OPENAI_API_KEY from the environment automatically

response = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[
        {"role": "system", "content": "You are a concise Android engineering mentor."},
        {"role": "user", "content": "Explain recursion in one sentence."},
    ],
    max_tokens=100,
)

print(response.choices[0].message.content)
print(response.usage)  # prompt_tokens, completion_tokens, total_tokens

Three Providers, One Shape

Same prompt, same intent, three SDKs. The differences are mostly naming: where the system prompt goes, what the response field is called, and how token usage is reported.

from openai import OpenAI
client = OpenAI()

r = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[
        {"role": "system",
         "content": "Be concise."},
        {"role": "user",
         "content": "Explain recursion "
                     "in one sentence."},
    ],
)
print(r.choices[0].message.content)

from google import genai
client = genai.Client()

r = client.models.generate_content(
    model="gemini-2.5-flash",
    config={"system_instruction":
            "Be concise."},
    contents="Explain recursion "
             "in one sentence.",
)
print(r.text)

from anthropic import Anthropic
client = Anthropic()

r = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=100,
    system="Be concise.",
    messages=[
        {"role": "user",
         "content": "Explain recursion "
                     "in one sentence."},
    ],
)
print(r.content[0].text)

Streaming Responses

A non-streamed call waits for the entire response before returning anything — fine for short answers, bad for a chat UI where the user should see text appear as it's generated. Streaming is conceptually the same problem Kotlin's Flow<String> solves: emit values as they become available instead of collecting them all first.

fun streamAnswer(prompt: String): Flow<String> = flow {
    val chunks = fakeLlmStream(prompt)
    chunks.collect { chunk -> emit(chunk) }
}

// caller
streamAnswer("Explain recursion").collect { chunk ->
    print(chunk)
}

stream = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[{"role": "user", "content": "Explain recursion"}],
    stream=True,
)

for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

Under the hood, this is Server-Sent Events (SSE): one HTTP response, kept open, with the server pushing small JSON chunks as the model generates tokens. The SDK turns that into a plain Python iterator so you never touch raw SSE parsing.

Tokens & Context Windows

A token is not a word or a character — it's a chunk of text the model's tokenizer produces, roughly 4 characters of English on average. "Recursion" might be one token; "antidisestablishmentarianism" might be six. Every request's system + history + user message + expected response must fit inside the model's context window — its maximum token budget.

import tiktoken

enc = tiktoken.encoding_for_model("gpt-4.1-mini")
tokens = enc.encode("Explain recursion in one sentence.")
print(len(tokens))   # 8 — an approximate cost/size check before you call the API

A running conversation eats into that budget with every turn, since you resend the whole history each time (Section 1). Here's what a growing conversation looks like against a 128K-token window:

Cost Math

Providers bill separately for input tokens (what you send) and output tokens (what the model generates), usually priced per million tokens, with output priced higher than input. Always check each provider's live pricing page before estimating a real budget — the numbers below are illustrative, not current pricing.

def estimate_cost(prompt_tokens: int, completion_tokens: int,
                   price_per_1m_in: float, price_per_1m_out: float) -> float:
    input_cost = (prompt_tokens / 1_000_000) * price_per_1m_in
    output_cost = (completion_tokens / 1_000_000) * price_per_1m_out
    return round(input_cost + output_cost, 6)

# after a real call, using the SDK's reported usage:
cost = estimate_cost(
    prompt_tokens=response.usage.prompt_tokens,
    completion_tokens=response.usage.completion_tokens,
    price_per_1m_in=0.40,
    price_per_1m_out=1.60,
)
print(f"${cost:.6f} for this call")

Parameters That Matter

The biggest adjustment coming from deterministic APIs: the same input can produce a different output on every call. These parameters control how much.

Errors & Retries

This is Topic 1 Section 5 (error handling) applied to a specific, very real failure mode: LLM APIs rate-limit you, time out under load, and occasionally return server errors that have nothing to do with your request. Catch the SDK's specific exception types, not a bare except:.

import time
from openai import OpenAI, RateLimitError, APIStatusError, APITimeoutError

client = OpenAI()

def ask_with_retry(prompt: str, retries: int = 3) -> str:
    for attempt in range(retries):
        try:
            r = client.chat.completions.create(
                model="gpt-4.1-mini",
                messages=[{"role": "user", "content": prompt}],
                timeout=30,
            )
            return r.choices[0].message.content
        except RateLimitError:
            wait = 2 ** attempt  # exponential backoff: 1s, 2s, 4s
            print(f"rate limited, retrying in {wait}s")
            time.sleep(wait)
        except APITimeoutError:
            print("timed out, retrying")
        except APIStatusError as e:
            if e.status_code >= 500:
                print(f"server error {e.status_code}, retrying")
            else:
                raise  # a 400 won't fix itself on retry — don't loop on it
    raise RuntimeError(f"failed after {retries} attempts")

Capstone: A Provider-Agnostic Ask Function

Build a single function that streams a response from whichever provider you pass in, prints tokens as they arrive, and reports the estimated cost when it's done. This exact shape — one call site, swappable providers — is what Topic 14's capstone product builds on top of.

import os
from dataclasses import dataclass
from openai import OpenAI
from anthropic import Anthropic

# illustrative pricing per 1M tokens — verify against each provider's current pricing page
PRICING = {
    "openai": {"in": 0.40, "out": 1.60},
    "claude": {"in": 3.00, "out": 15.00},
}

@dataclass
class AskResult:
    text: str
    prompt_tokens: int
    completion_tokens: int

    def cost(self, provider: str) -> float:
        rate = PRICING[provider]
        return round(
            (self.prompt_tokens / 1_000_000) * rate["in"]
            + (self.completion_tokens / 1_000_000) * rate["out"],
            6,
        )


def ask_openai(prompt: str) -> AskResult:
    client = OpenAI()
    chunks: list[str] = []
    stream = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        stream_options={"include_usage": True},
    )
    usage = None
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            delta = chunk.choices[0].delta.content
            print(delta, end="", flush=True)
            chunks.append(delta)
        if chunk.usage:
            usage = chunk.usage
    print()
    return AskResult("".join(chunks), usage.prompt_tokens, usage.completion_tokens)


def ask_claude(prompt: str) -> AskResult:
    client = Anthropic()
    chunks: list[str] = []
    with client.messages.stream(
        model="claude-sonnet-4-5",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
            chunks.append(text)
        final = stream.get_final_message()
    print()
    return AskResult("".join(chunks), final.usage.input_tokens, final.usage.output_tokens)


def ask(provider: str, prompt: str) -> None:
    result = ask_openai(prompt) if provider == "openai" else ask_claude(prompt)
    print(f"\n[{provider}] tokens: {result.prompt_tokens} in / "
          f"{result.completion_tokens} out — est. ${result.cost(provider):.6f}")


if __name__ == "__main__":
    ask("openai", "Explain recursion in one sentence, then give a Kotlin example.")