16. AI systems engineering in Python

An LLM application is an ordinary networked Python service whose slowest, least reliable and most expensive dependency is the model API. Engineer it like one: typed boundaries, async I/O, timeouts, retries, caching, logging and evals.

[embedded content: LLM service architecture · request path]

Arrows show the request path, and responses travel back the same way. Dashed boxes sit outside your process, so each one can be slow or fail; the sections below cover how each box handles that.

What a model API call really is

• An SDK call is an HTTPS POST with a JSON body (model, messages, parameters, tools) that returns JSON (content blocks, a stop reason, token usage).
• The SDK adds authentication headers, retries some errors and parses the response into typed objects. Knowing this lets you debug with raw HTTP, log exact payloads and switch providers.
• The API is stateless: you resend the whole conversation every turn. Cost and latency grow with the history you send.
• Read the token usage on every response and log it. Cost is tokens multiplied by price.

import anthropic

client = anthropic.Anthropic()        # reads ANTHROPIC_API_KEY from the environment
resp = client.messages.create(
    model="claude-sonnet-5-5",        # current model names are in the provider's docs
    max_tokens=1024,
    messages=[{"role": "user", "content": "Summarize this: ..."}],
)
print(resp.content[0].text)
print(resp.usage)                     # input and output token counts

Other providers’ SDKs have the same shape. Check the current documentation for exact parameter names, streaming events and structured-output options.

Streaming

• Ask for a stream and iterate over events as they arrive (Part 6), or use async for with an async client (Part 12).
• Streaming lowers time to first token for users; the total cost is the same.
• Handle disconnects mid-stream, and accumulate the full text for logging.

Reliability

• Timeouts on every call. A missing timeout is how services hang forever.
• Retries with exponential backoff and jitter (Part 8) on rate limits (429), overload and server errors (5xx) and connection failures. Do not retry 400-class errors: the request itself is wrong.
• Respect retry-after headers when the provider sends them.
• Concurrency limits with a semaphore (Part 12), plus a client-side budget if the provider limits tokens per minute.
• Idempotency: retrying a call that triggered side effects, such as a tool that sends email, can repeat them.
• Fallbacks: a smaller model, a cached answer, or a clear error instead of a hang.

Structured outputs

Define the output as a Pydantic model, give its schema to the model through the provider’s structured-output or tool feature, validate the reply, and repair on failure (Part 10). Treat model output exactly like user input from the internet: untrusted until validated.

Tool calling from first principles

1. You describe each tool: a name, a description, and a JSON Schema for its arguments.
2. The model replies with a tool-use request: a tool name plus arguments as JSON.
3. Your code validates the arguments, runs the real function, and sends back a tool-result message.
4. Repeat until the model returns a final answer instead of another tool request.

That loop is an “agent”. Frameworks add structure around it, but the loop is the core.

TOOLS = {}

def tool(fn):                         # a registry decorator (Part 5)
    TOOLS[fn.__name__] = fn
    return fn

@tool
def get_weather(city: str) -> str:
    """Current weather for a city."""     # this docstring becomes prompt text
    ...

def run_tool_call(name: str, args: dict) -> str:
    if name not in TOOLS:
        return f"error: unknown tool {name}"
    try:
        return str(TOOLS[name](**args))
    except Exception as e:            # a deliberate boundary: report the error to the model
        return f"error: {type(e).__name__}: {e}"

Security: tool arguments come from the model, and the model can be steered by text hidden in retrieved documents or web pages (prompt injection). Validate arguments, scope permissions narrowly, require confirmation for destructive actions, and never pass model output to eval or a shell.

Retrieval-augmented generation (RAG), mechanically

1. Ingest: load documents, split them into chunks, embed each chunk in batches, and store the vectors with their text and metadata.
2. Retrieve: embed the question, find the nearest chunks by cosine similarity (Part 13), and optionally rerank them.
3. Generate: put the top chunks into the prompt and ask for an answer with citations.

The Python pieces: generators for ingestion (Part 6), NumPy or a vector store (FAISS, pgvector, Qdrant, Chroma and others), async for parallel embedding calls, and Pydantic for chunk records. Most RAG quality problems are data problems (chunking, missing metadata, weak retrieval), so measure retrieval separately from generation.

Serving with FastAPI

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class AskRequest(BaseModel):
    question: str
    top_k: int = 5

class AskResponse(BaseModel):
    answer: str
    sources: list[str]

@app.post("/ask", response_model=AskResponse)
async def ask(req: AskRequest) -> AskResponse:
    chunks = await retrieve(req.question, req.top_k)
    answer = await generate(req.question, chunks)
    return AskResponse(answer=answer, sources=[c.id for c in chunks])

Run it with uvicorn app:app --reload. FastAPI validates requests against AskRequest, serializes the response, and serves interactive docs at /docs.

• async def endpoints share one event loop, so never block inside them (Part 12). Plain def endpoints run in a thread pool.
• Create heavy resources (models, API clients) once at startup in a lifespan handler, not per request.
• Stream to the browser with StreamingResponse and an async generator.
• To self-host models, run a dedicated inference server (vLLM, TGI, or Ollama locally) and call it over HTTP. Batching and KV-cache management are its job, not your web handler’s.

Configuration and secrets

• Load secrets from environment variables or a secrets manager, never from code or git. Keep a local .env file in .gitignore.
• Build one typed settings object (pydantic-settings) at startup and pass it down, instead of scattered os.environ reads.
• Version prompts like code: in files, under git, referenced by name and version in your logs.

Logging and observability

• Use the logging module, not print. Configure it once at the entry point and create log = logging.getLogger(__name__) in each module.
• Log structured records: request id, model, latency, token usage, cost and error type.
• Use tracing (OpenTelemetry or an LLM-specific tracer) to see each step of an agent loop.
• Never log secrets, and decide deliberately how much user content you log. GDPR applies to prompts too.

Caching

• Deterministic work, such as embeddings, can be cached under a hash of model, input and parameters.
• Use functools.cache within a process, and diskcache, SQLite or Redis across runs.
• Provider-side prompt caching cuts the cost of repeated long prefixes. Put the static part of the prompt (instructions, documents) first.

Reproducibility

• Record dependency locks, model names and versions, prompt versions, seeds and dataset hashes with every experiment.
• Seed everything: random.seed, np.random.default_rng(seed), torch.manual_seed. Full GPU determinism needs extra flags and costs speed.
• LLM APIs are not perfectly deterministic even at temperature 0. Evaluate on sets of examples, not single samples.

Questions to ask:

• What happens when the model API is slow, down, rate-limited, or returns invalid JSON?
• What does one request cost, and where is that logged?
• Which of this tool’s inputs came from the model, and are they validated?
• Could I reproduce this result next month?