Why this matters

For a senior full-stack engineer, the Python gap is rarely syntax. It is confidence under operational pressure: provider SDKs returning semi-structured payloads, ingestion scripts that need to be reruns weeks later, evaluation loops where a silent failure quietly corrupts a benchmark, and debugging sessions where the real failure was shape mismatch not model quality.

The habits in this lesson are about closing that gap fast.

Core concepts

Pydantic at the boundary

Every AI system has at least three categories of external input: provider API responses, documents or files being ingested, and configuration or environment values. Each one can fail silently if you let raw dicts propagate inward.

Pydantic's job is to be the gatekeeper at those entry points.

from pydantic import BaseModel, Field, field_validator


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

    @field_validator("content")
    @classmethod
    def content_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("LLM returned empty content")
        return v.strip()

If the provider sends {"content": null}, you find out at the boundary instead of discovering it three layers downstream as an AttributeError on NoneType.

Boundary layers as an architectural habit

Think of your system in four layers:

1. Boundary layer — provider responses, uploaded files, queue payloads, environment config. This is where Pydantic lives.
2. Transformation layer — cleaning, scoring, filtering, reshaping. Use simple dataclasses or TypedDicts here.
3. Orchestration layer — retries, sequencing, branching, persistence decisions.
4. Review layer — logs, traces, metrics, evaluation output.

The rule: keep transformation and orchestration logic ignorant of raw external shapes. If a provider changes their response schema, only the boundary model needs to change.

Idempotent scripts

Most high-leverage AI work happens in scripts: ingestion backfills, benchmark generation, dataset cleanup, prompt experiment exports. If a script cannot be safely rerun, it is not ready for real use.

Idempotent script checklist:

• write to a new output file or use an explicit --overwrite flag
• skip rows that were already processed rather than reprocessing blindly
• print summary counts at the end: processed, skipped, failed
• capture failures per row rather than aborting the entire run

from pathlib import Path


def run_pipeline(input_path: Path, output_path: Path, overwrite: bool = False) -> None:
    if output_path.exists() and not overwrite:
        raise FileExistsError(f"{output_path} already exists. Pass overwrite=True to replace it.")

    rows = [json.loads(line) for line in input_path.read_text().splitlines() if line.strip()]
    processed, skipped, failed = [], [], []

    for row in rows:
        try:
            result = transform(row)
            processed.append(result)
        except Exception as exc:
            failed.append({"id": row.get("id"), "error": str(exc)})

    output_path.write_text("\n".join(json.dumps(r) for r in processed))
    print(f"Done: {len(processed)} processed, {len(skipped)} skipped, {len(failed)} failed")

Logging for AI debugging

Standard application logging covers request/response cycles. AI debugging needs more:

• which model and which prompt version produced the output
• the retry count and whether a fallback was used
• how long the provider call took
• why a specific chunk was retrieved or excluded
• what the validation failure reason was

import logging
import time

logger = logging.getLogger(__name__)


async def call_provider(request: LLMRequest) -> LLMResponse:
    start = time.monotonic()
    try:
        raw = await provider_client.generate(request.model_dump())
        response = LLMResponse.model_validate(raw)
        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,
            },
        )
        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

This pattern gives you structured log events that are queryable. When something goes wrong at 2 AM, you want to filter by request_id or model, not scan freeform strings.

Working example

A full boundary + logging pattern for a document ingestion step:

import json
import logging
from pathlib import Path
from pydantic import BaseModel, Field

logger = logging.getLogger(__name__)


class RawDocument(BaseModel):
    doc_id: str = Field(min_length=1)
    title: str
    body: str = Field(min_length=10)
    source_url: str


def ingest_documents(raw_path: Path, output_path: Path) -> None:
    lines = raw_path.read_text().splitlines()
    good, bad = [], []

    for i, line in enumerate(lines):
        try:
            doc = RawDocument.model_validate_json(line)
            good.append({"id": doc.doc_id, "title": doc.title, "body": doc.body})
        except Exception as exc:
            bad.append({"line": i, "error": str(exc)})
            logger.warning("doc_parse_failed", extra={"line": i, "error": str(exc)})

    output_path.write_text("\n".join(json.dumps(d) for d in good))
    logger.info("ingestion_complete", extra={"ok": len(good), "failed": len(bad)})
    print(f"Ingested {len(good)} docs, {len(bad)} failures")

Common mistakes

Validating too late. Passing raw dicts through multiple functions before checking for required fields means failures surface far from their origin. Validate immediately at the boundary.

Logs that only describe success. logger.info("Request completed") tells you nothing useful when diagnosing production issues. Log the identifiers, model, latency, and failure reason every time.

Scripts that require manual cleanup after partial failures. If a script processes 80% of rows then crashes, a non-idempotent version leaves you guessing which rows were already processed. Capture per-row failures and write outputs atomically.

Using print instead of structured logging in production code. Print is fine in notebooks. In services and scripts that run on a schedule, structured logs give you the queryability you need to debug at scale.

Try it yourself

1. Take any AI script you have and add per-row failure capture. Run it on a dataset with one intentionally malformed row and verify the failure is captured without aborting the run.
2. Add structured logging to one provider call. Include request ID, model, latency, and any validation failure reason.
3. Write a Pydantic model for a provider response you use. Add a validator that catches at least one real edge case you have seen in the wild.