Your personal operating system for career transition.

Python patterns every AI engineer should know

AI engineering Python is not about algorithms. It is about reliability under operational pressure: provider APIs that change their response shapes, ingestion scripts that need to run safely on a schedule, async pipelines that hit rate limits, and debugging sessions where the real question is "which chunk was retrieved and why?"

The patterns below are the ones that separate Python code that works in demos from Python code that works in production.

Pattern 1: Pydantic at every external boundary

Every point where data enters your system from outside — provider API responses, uploaded files, queue payloads, environment variables — is a boundary that should be validated with Pydantic.

from pydantic import BaseModel, Field, model_validator

class LLMResponse(BaseModel):
    request_id: str = Field(min_length=1)
    content: str
    model: str
    input_tokens: int = Field(ge=0)
    output_tokens: int = Field(ge=0)
    finish_reason: str = "stop"

    @model_validator(mode="after")
    def content_not_empty(self) -> "LLMResponse":
        if not self.content.strip():
            raise ValueError("LLM returned empty content")
        return self

The rule: validate at the boundary, then pass typed objects inward. If a provider changes their response schema, only the boundary model changes.

Pattern 2: Async with bounded concurrency

LLM calls are I/O-bound. Processing 50 documents serially takes 10× longer than processing them concurrently with a semaphore:

import asyncio

async def process_batch(items, client, max_concurrent=5):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def _one(item):
        async with semaphore:
            return await asyncio.wait_for(client.generate(item), timeout=30.0)

    results = await asyncio.gather(*[_one(i) for i in items], return_exceptions=True)
    successes = [r for r in results if not isinstance(r, Exception)]
    failures = [r for r in results if isinstance(r, Exception)]
    return successes, failures

return_exceptions=True prevents one failure from cancelling all remaining tasks. asyncio.wait_for prevents a hung provider from blocking the event loop indefinitely.

Pattern 3: Generators for memory-efficient pipelines

List comprehensions materialize all results in memory. For large corpora, use generator pipelines:

from pathlib import Path
from typing import Iterator
import json

def read_jsonl(path: Path) -> Iterator[dict]:
    with path.open() as f:
        for line in f:
            if line.strip():
                yield json.loads(line)

def validate(records: Iterator[dict]) -> Iterator[dict]:
    for r in records:
        if r.get("body") and len(r["body"]) >= 100:
            yield r

def normalize(records: Iterator[dict]) -> Iterator[dict]:
    for r in records:
        yield {**r, "body": r["body"].strip()}

# Compose: constant memory regardless of file size
pipeline = normalize(validate(read_jsonl(Path("docs.jsonl"))))

Each stage is lazy: it pulls one item from the previous stage when the next stage asks for one. A 10GB file and a 10MB file use the same peak memory.

Pattern 4: Structured logging for AI debugging

Standard application logging records what happened. AI debugging needs to record what the model saw and why it made the decisions it made:

import logging
import time

logger = logging.getLogger(__name__)

async def call_provider(request):
    start = time.monotonic()
    try:
        response = await client.generate(request.model_dump())
        logger.info("provider_call_ok", extra={
            "request_id": request.request_id,
            "model": request.model,
            "latency_ms": int((time.monotonic() - start) * 1000),
            "finish_reason": response.finish_reason,
            "output_tokens": response.output_tokens,
        })
        return response
    except Exception as exc:
        logger.error("provider_call_failed", extra={
            "request_id": request.request_id,
            "model": request.model,
            "latency_ms": int((time.monotonic() - start) * 1000),
            "error": str(exc),
        })
        raise

Structured log events (extra={} dict) are queryable. When debugging at 2 AM, you want filter by request_id, not grep for strings.

Pattern 5: Idempotent scripts

Scripts that run on a schedule, process datasets, or run evaluations must be safely re-runnable:

• Write to a new output file or use an explicit --overwrite flag
• Skip already-processed rows rather than reprocessing blindly
• Capture per-row failures instead of aborting the run
• Print a summary count at the end: processed, skipped, failed

The underlying principle

These patterns are all the same idea applied to different layers: explicit boundaries, explicit errors, explicit state. Python that hides failures, propagates raw dicts, or side-steps structure is hard to debug. Python that makes boundaries explicit and validates at them is debuggable when something goes wrong.

Reference

• Pydantic v2 docs
• asyncio documentation
• Python typing module