Chapter 1

Foundations · Python for AI engineering

📖 2 min read · 6 sections · May 2026

Why this matters

Most AI engineering bugs are not model bugs. They're plumbing bugs — silent failures when a webhook times out mid-call, a streaming response that drops tokens because someone forgot to flush, a memory leak because the LLM client wasn't closed. Python fluency is the floor; everything above it leaks through it.

Async/await — the most important skill

LLM calls take seconds. Vector lookups take milliseconds. Tool calls run external APIs. Everything is I/O-bound. Synchronous Python here would serve maybe ~5 requests/second from a single worker; async pushes it well past 100.

# Wrong — blocks the event loop
import requests
def fetch_pages(urls):
    return [requests.get(u).json() for u in urls]   # serial, 10× slower

# Right
import asyncio, httpx
async def fetch_pages(urls):
    async with httpx.AsyncClient() as client:
        return await asyncio.gather(*[client.get(u) for u in urls])

Three rules that catch 90% of async bugs:

Never call sync I/O inside an async def. If you must (legacy lib), wrap with await asyncio.to_thread(fn).
Always pair async with for client/session lifecycle. Leaked HTTPX clients = file descriptor leaks.
Use asyncio.gather() with a concurrency limit (asyncio.Semaphore) — unbounded fan-out trashes downstream rate limits.

Typing, Pydantic, and why you want it

LLMs hand you back strings. Strings are lies. Every successful AI app I've seen treats the LLM's output as untrusted input and parses it through a strict schema before doing anything with it. Pydantic is the dominant choice.

from pydantic import BaseModel, Field
class TripPlan(BaseModel):
    destination: str
    duration_days: int = Field(ge=1, le=30)
    budget_inr: int | None = None

# Most modern LLM SDKs let you pass a schema and get back a validated instance.
plan = client.chat.completions.parse(model='...', response_format=TripPlan, ...)

Generators, streaming, and backpressure

LLM responses stream token by token. Streaming to the user is a perceived-latency win (first token in ~200ms vs 3s for full response). Done wrong, you'll OOM under load — store-all-tokens-then-send patterns get you there fast.

async def stream_completion(prompt):
    async for chunk in llm.stream(prompt):
        yield chunk.text
        # Don't accumulate. Let the consumer decide what to keep.

Error handling as architecture

Retries, timeouts, and partial failures are not afterthoughts in LLM code — they're the design. The provider will return 429 (rate limit), 500, 502, 503, network blips, and content-filter blocks. Wrap every call.

from tenacity import retry, wait_exponential_jitter, stop_after_attempt, retry_if_exception_type

@retry(
  wait=wait_exponential_jitter(initial=1, max=10),
  stop=stop_after_attempt(4),
  retry=retry_if_exception_type((TimeoutError, RateLimitError, ServerError)),
)
async def safe_complete(prompt):
    return await llm.complete(prompt, timeout=30)

Q: Why prefer Pydantic over typing.TypedDict for LLM outputs?

TypedDict is static — it doesn't validate at runtime. The LLM will eventually hand you a payload with a missing field, an out-of-range value, or a stringified number. Pydantic catches that at the boundary; TypedDict catches it three function calls deeper when something else explodes.

Cross-Q: What about dataclass? A: Same problem — no runtime validation by default. pydantic.dataclasses.dataclass gives you the validation. Vanilla @dataclass doesn't.

Takeaway

Treat the LLM as an unreliable network service that occasionally returns gibberish. Everything in your code should look like it would if you were calling a flaky third-party API: timeouts, retries with backoff, schema validation, structured logging, idempotent operations where possible.

Further reading: asyncio docs · Pydantic docs · tenacity for retries