Python Refresher for AI Engineers: The 20% You Actually Need

Python Refresher for AI Engineers: The 20% You Actually Need

You do not need to master Python to build with LLMs. Most AI applications are IO glue: fetch context, call an API, parse the response, stream tokens somewhere, retry when things fail, log everything. The language features that matter are the ones that make those six things clean.

Here is where each pattern below shows up in a typical LLM application, so you know why it earns its place:

The Data Shapes You Will Fight#

Everything arriving from or going to a model API is JSON, which in Python means nested dicts and lists of dicts. Fluency here is non-negotiable: safe access with .get(), comprehensions for transforming lists, json.loads and json.dumps round trips, and sorting by key.

import json

raw = '{"messages": [{"role": "user", "content": "hi"}, {"role": "assistant", "content": "Hello!"}]}'
data = json.loads(raw)

user_turns = [
    m["content"].strip().lower()
    for m in data["messages"]
    if m["role"] == "user"
]

longest = max(data["messages"], key=lambda m: len(m["content"]), default=None)

Two details worth internalizing from that snippet. The comprehension reads as "for each message, if it is from the user, take cleaned content," which is most of your daily transformation code. And max(..., default=None) returns None on empty input instead of raising, which matters when the model returns nothing usable at 3am.

Type Hints Pay Rent#

Type hints stopped being optional decoration the moment pydantic became the standard boundary for anything touching external JSON. Use them as contracts: validate untrusted data at the edges of your system, then trust types inside it.

from pydantic import BaseModel

class Ticket(BaseModel):
    id: str
    category: str
    urgency: int
    summary: str

ticket = Ticket.model_validate_json(api_response)
ticket.urgency

If the API response is malformed, this fails loudly at the boundary instead of producing a mystery KeyError three functions later. Rules of thumb: pydantic models for data crossing trust boundaries (API responses, model outputs), plain dataclasses for internal records, TypedDict only when you genuinely must keep a dict shape. Anything the LLM produced is untrusted by definition; parse it through a schema every time.

async/await Without the Panic#

Async has one job in this domain: waiting on networks efficiently. A model API takes half a second to ten seconds to respond, during which your process does nothing. If you have ten independent summarizations to run, sequential execution wastes minutes.

import asyncio
import anthropic

client = anthropic.AsyncAnthropic()

async def summarize(text: str) -> str:
    msg = await client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=300,
        messages=[{"role": "user", "content": f"Summarize in two sentences:\n{text}"}],
    )
    return msg.content[0].text

async def main():
    summaries = await asyncio.gather(*(summarize(t) for t in documents))

asyncio.run(main())

asyncio.gather runs all calls concurrently and gives you results in order. Pass return_exceptions=True if you want failures collected alongside successes instead of aborting the batch. The decision of when to bother:

One warning: do not sprinkle async on CPU-bound work. It helps waiting, not computing.

Generators Are How Streaming Works#

Every time you watch a chatbot type word by word, a generator is doing the work. A generator is a function with yield in it: calling it returns an iterator that produces values lazily, pausing between them. Model providers stream responses as chunks precisely because generators let you consume them as they arrive instead of buffering everything.

That makes wrapping streams a core skill. Here is a passthrough that regroups raw token chunks into whole words:

def word_stream(token_chunks):
    pending = ""
    for chunk in token_chunks:
        pending += chunk
        while " " in pending:
            word, pending = pending.split(" ", 1)
            yield word + " "
    if pending:
        yield pending

And wiring it to a real OpenAI stream:

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Explain vector search briefly."}],
    stream=True,
)

chunks = (event.choices[0].delta.content or "" for event in stream)
for word in word_stream(chunks):
    print(word, end="", flush=True)

Once streaming is just iterators, useful tricks become obvious: tap the same generator to count tokens, redact patterns before display, or tee output into a transcript log. No framework required.

Fail Politely#

Providers throttle aggressively. Rate limits are not an edge case; they are Tuesday afternoon under load. Every production call needs three things: a timeout so hangs cannot accumulate, retries with exponential backoff plus jitter so thundering herds do not stampede, and specific handling for rate-limit errors rather than generic catch-alls.

The tenacity library turns that policy into a decorator:

from tenacity import retry, stop_after_attempt, wait_random_exponential

@retry(wait=wait_random_exponential(min=1, max=30), stop=stop_after_attempt(5))
def embed(texts: list[str]) -> list[list[float]]:
    resp = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return [d.embedding for d in resp.data]

Five attempts, waits that grow randomly up to thirty seconds. The retry flow it implements:

Also learn context managers (with blocks) properly while you are here, because both major SDKs use them for client lifecycles and file handling around document ingestion. They are how Python guarantees cleanup, and cleanup is what keeps long-running ingestion jobs from leaking connections.

The Rapid-Fire Round#

Small things that come up weekly: f-strings everywhere (f"{len(tokens)} tokens"); pathlib.Path over string path surgery; enumerate() and zip() instead of index arithmetic; collections.Counter for frequency questions and defaultdict(list) for grouping; re-raising with context using raise RuntimeError("embedding failed") from err; the if __name__ == "__main__": guard on any script you will import from elsewhere; and logging over bare print in anything that survives past the prototype.

None of this is deep. All of it is load-bearing.

Practice That Sticks#

Take any script that calls a model API and upgrade it in one sitting: add a pydantic model at the response boundary, wrap the network call in a retry decorator, and switch printing to streamed words through your own generator. If the result survives a flaky network and malformed output without crashing, you have genuinely learned the twenty percent. Everything else in Python you can look up when you need it.