1.1 — Python for AI Engineers

1. What is it?

This chapter is not "Python from scratch." It's the specific subset of Python engineering that shows up constantly in AI systems and that most tutorials skip: asyncio (because every LLM call is I/O-bound and production AI services live or die on concurrency), typing and pydantic (because structured LLM output, tool schemas, and API contracts all run through them), packaging/dependency management with uv, and the failure-handling patterns you need when a chunk of your "business logic" is a network call to a non-deterministic third-party model.

2. Why does it exist?

Python won AI engineering not because it's fast (it isn't) but because its ecosystem (NumPy → PyTorch → Hugging Face → LangChain) accumulated first, and because its dynamic typing and REPL-driven style made it the fastest language to prototype ML code in. The cost of that history: Python's concurrency model, packaging story, and type system were all bolted on after the fact, and an AI engineer who doesn't understand why they work the way they do will write code that quietly serializes what should be concurrent, or silently passes malformed data through an untyped boundary until it hits an LLM and produces garbage three services downstream.

3. What problem does it solve?

An AI application spends most of its wall-clock time waiting: waiting for an LLM API to respond, waiting for a vector DB query, waiting for a database round trip. If your code handles that waiting badly, your AI product will feel slow and cost more than it should, regardless of how good your prompts are. The tools in this chapter — asyncio, connection pooling, typed boundaries — exist to make an I/O-bound system fast, correct, and debuggable.

4. How does it work internally?

The GIL and why it doesn't matter for LLM apps

Python's Global Interpreter Lock (GIL) prevents two threads from executing Python bytecode at the same time in one process. This makes Python bad at CPU-bound parallelism (e.g., running two tokenizers on two cores inside one process gains you little without multiprocessing). It does not make Python bad at I/O-bound concurrency: a thread or an asyncio task waiting on a network response releases the GIL while it waits, so hundreds of concurrent LLM calls can be in flight from one Python process. This is why asyncio — not multiprocessing — is the default concurrency model for AI backends.

asyncio's execution model

Event loop (single thread)
┌─────────────────────────────────────────────────────┐
│  while True:                                          │
│      ready = tasks that can make progress right now   │
│      run each ready task until it hits `await`        │
│      (control returns to the loop; task is parked)    │
│      poll: has any I/O completed? → mark that task ready │
└─────────────────────────────────────────────────────┘

await is the handoff point: "I'm blocked on I/O, run something else until this resolves." Calling a blocking function (e.g., the synchronous requests.get, or CPU-heavy work like local reranking) from inside an async def without await freezes the entire event loop — every other in-flight request in your FastAPI process stalls. This is the single most common production bug in AI backends: someone imports the sync SDK for a vector DB or calls a synchronous tokenizer inside an async route handler, and the whole service's p99 latency spikes.

Finding the blocking call: profiling, including on a live process

Section 11 claims Python's per-call overhead matters and that a blocking call stalls the whole event loop — but claiming a bottleneck exists and finding it empirically are different skills.

For a standalone script or a suspected CPU-bound hot path, cProfile gives a function-level breakdown:

bash

python -m cProfile -o profile.out -m myapp.batch_embed
python -c "import pstats; pstats.Stats('profile.out').sort_stats('cumulative').print_stats(15)"

This works fine for offline scripts, but it requires running the code under the profiler from the start — not useful for "our production Uvicorn worker is stalling right now and we can't restart it without dropping in-flight requests." For that, use py-spy, which attaches to an already-running process from the outside (reading its stack via the OS, not by injecting code into it), so a live production process never needs to be modified, restarted, or paused for more than a few milliseconds:

bash

py-spy top --pid 4821          # live, refreshing view of which functions are hot right now
py-spy dump --pid 4821         # one-shot stack trace of every thread at this instant

This is the production-relevant tool specifically because you cannot attach a normal debugger (pdb) to a running production process the way you would locally — pdb expects to pause execution and hand you an interactive prompt inline in that process's own terminal, which isn't available for a backgrounded Uvicorn worker serving live traffic, and pausing it that way would itself cause the outage you're trying to diagnose. py-spy dump on a hung worker will typically show a stack frame sitting inside a synchronous call (a sync DB driver, requests.get, blocking file I/O) instead of an await — which is the smoking gun for the exact failure mode described in the asyncio section above: one blocking call, caught in the act, freezing everything else on that event loop.

A complementary, always-on safety net: set PYTHONASYNCIODEBUG=1 in a staging/pre-production environment. It makes asyncio log a warning whenever a callback (including a route handler) runs for longer than its slow-callback threshold (100ms by default) without yielding — surfacing the exact "someone called a blocking function inside async def" bug from section 4 in a log line during testing, before it becomes the Monday-morning-rush incident described in section 20.

Typed boundaries with pydantic

Pydantic models validate and coerce data at runtime, which matters enormously at the boundaries where an AI system talks to something non-deterministic:

python

from pydantic import BaseModel, Field, field_validator

class ExtractedInvoice(BaseModel):
    """Represents structured fields extracted from an invoice document by an LLM."""
    vendor_name: str
    invoice_number: str
    total_amount_usd: float = Field(gt=0)
    line_items: list[str] = Field(default_factory=list)

    @field_validator("invoice_number")
    @classmethod
    def not_empty(cls, v: str) -> str:
        """Rejects an empty invoice number rather than letting it flow downstream silently."""
        if not v.strip():
            raise ValueError("invoice_number must not be empty")
        return v

When an LLM's structured output is parsed into this model, malformed output raises immediately and loudly, at the boundary — instead of an empty string quietly reaching a billing system three function calls later.

Typed configuration with `pydantic-settings`

The same "validate at the boundary, fail loudly" principle applies to configuration, not just LLM output. An app that reads os.environ["DATABASE_URL"] deep inside some function only discovers a missing variable the first time that code path runs — which in production might be hours after deploy, mid-request, for a customer. pydantic_settings.BaseSettings turns configuration into a typed, validated boundary that's checked once, at startup:

python

from typing import Literal

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    """Typed application configuration, loaded from environment variables and .env files."""

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")

    environment: Literal["dev", "staging", "prod"] = "dev"
    anthropic_api_key: str
    database_url: str
    max_concurrent_llm_calls: int = Field(default=20, gt=0)


settings = Settings()  # instantiated once, at import time — this is the fail-fast point

Constructing Settings() at module import time (typically in a small config.py imported by main.py before the app object is even built) means a missing anthropic_api_key or a malformed max_concurrent_llm_calls raises a pydantic.ValidationError and crashes the process at boot, before it starts accepting traffic — not at the first request that happens to touch that field. Crashing loudly at startup is strictly better than a mysterious KeyError three hours into production traffic.

For dev/staging/prod layering, point env_file at the right file per environment rather than hand-editing .env between deploys:

python

import os

env_name = os.environ.get("APP_ENV", "dev")
settings = Settings(_env_file=f".env.{env_name}")

This chapter covers only the app-level typed-config pattern; where the underlying secret values actually live (env vars injected by the platform, a secrets manager, etc.) is Part 9.5's job, not this one.

Packaging and dependency management with `uv`

uv is a Rust-based Python package/project manager that replaces the slow, fragmented pip + venv + pip-tools workflow with one fast tool and one lockfile.

bash

uv init my-ai-service          # creates pyproject.toml, a src layout, and a .python-version
cd my-ai-service
uv add fastapi httpx anthropic pydantic-settings   # adds to [project.dependencies], updates uv.lock
uv add --dev pytest pytest-asyncio ruff            # dev-only tools, a separate dependency group
uv add --optional docs mkdocs                      # an optional extra, installed only on request
uv run uvicorn main:app --reload                   # runs inside the project's managed venv

uv add writes two things: the human-edited constraint in pyproject.toml (e.g. "fastapi>=0.115,<1.0") and an exact, fully-resolved snapshot of every dependency (direct and transitive) with pinned versions and hashes in uv.lock. This distinction is the entire point of a lockfile: pyproject.toml says what's acceptable, uv.lock says what was actually installed and tested — commit both, so a teammate or a CI runner or a production build three months from now resolves to the exact same dependency graph you tested against, instead of picking up a transitive dependency's breaking minor release.

toml

# pyproject.toml (excerpt)
[project]
dependencies = [
    "fastapi>=0.115,<1.0",
    "httpx>=0.27",
]

[dependency-groups]
dev = ["pytest>=8.0", "pytest-asyncio", "ruff"]

[project.optional-dependencies]
docs = ["mkdocs"]

Pinning strategy is situational, not universal: for a production Docker build, use uv sync --frozen --no-dev — --frozen refuses to re-resolve or touch uv.lock even if pyproject.toml changed (fail the build loudly rather than silently drift), and --no-dev excludes the dev dependency group so test tooling never ships in the production image. For a library other teams import (an internal SDK, a shared tool package), pin looser ranges in pyproject.toml (>=2.1,<3.0, not ==2.1.4) so it doesn't force a specific transitive version on every consumer — the lockfile discipline above is for applications you deploy, not packages you publish for others to depend on.

5. Simple mental model

Think of an AI backend as a restaurant kitchen with one chef and many slow-cooking dishes. The chef (your single-threaded event loop) doesn't stand and stare at the oven (the LLM API) waiting for one dish to finish — they start dish A, put it in the oven, immediately start prepping dish B while A cooks, get interrupted by dish A's timer, plate it, move to dish C. One chef, many dishes in flight, because the slow part (the oven / the LLM) isn't the chef's attention — it's external. The moment the chef ignores this and stands staring at one oven (a blocking call), every other order backs up.

6. Real-world example

A customer support AI backend receives 200 concurrent chat requests. Each request needs: an embedding call (~50ms), a vector DB query (~30ms), an LLM completion (~2000ms). Done synchronously, 200 requests would serialize to 200 × 2080ms ≈ 7 minutes for the batch. Done with asyncio and a connection pool, all 200 can be in flight concurrently (bounded by the LLM provider's rate limit), so the batch finishes in roughly the time of the slowest single request plus queuing — often under 10 seconds.