Python for AI

Why AI code looks like this

OpenAI, Anthropic, Google, Hugging Face, LangChain, and LlamaIndex all ship Python first. Early on, a stand-in function can pretend to be the model so you learn the program with no API key. Swap the stand-in for a real client at the end. Use a virtual environment and pin packages. AI libraries move fast.

Starter

tickets = [
    {"id": "T1", "text": "Charged twice for order A-1001"},
    {"id": "T2", "text": "Package has not moved in 6 days"},
]
print(len(tickets), "tickets ready")

OpenAI, Anthropic, Google, Hugging Face, LangChain, and LlamaIndex all ship Python first. The same five tickets can teach the whole slice: read them from a file, ask a model to label each one, check the label, save the result.

tickets = [
    {"id": "T1", "text": "Charged twice for order A-1001"},
    {"id": "T2", "text": "Package has not moved in 6 days"},
    {"id": "T3", "text": "How do I change my email?"},
]

Early on, a stand-in function can pretend to be the model so you learn the program with no API key. Swap in a real client at the end.

Strings, lists, and dictionaries

You will write this comprehension again to drop low-score search hits. Index with strings (ticket['id']), not with hope. If a key might be missing, use .get('priority', 'low') so one bad row does not crash the batch.

Starter

tickets = [{"id": "T1", "text": "refund"}, {"id": "T2", "text": "hello"}]
hot = [t for t in tickets if "refund" in t["text"]]
print(hot[0]["id"])

A ticket is a dictionary. A batch is a list of dictionaries. Model JSON comes back as text, then becomes a dictionary you index with strings.

ticket = tickets[0]
print(ticket["id"], ticket["text"])

labels = []
for ticket in tickets:
    labels.append({"id": ticket["id"], "topic": "billing"})

# same idea, written as a comprehension
ids = [t["id"] for t in tickets if "order" in t["text"].lower()]

Functions are how tools are born

Default arguments show up everywhere: `def search(query: str, k: int = 4)`. The default is what you get when the model forgets to pass k. Keep the docstring honest. Later, that docstring is what the model reads to decide whether to call the tool.

Starter

def classify(text: str) -> dict:
    if "refund" in text.lower():
        return {"topic": "billing", "priority": "high"}
    return {"topic": "other", "priority": "low"}

print(classify("Please refund order A17"))

A tool is a function with a clear name, arguments, and a return value. Write the plain function first. Frameworks only wrap it.

def classify(text: str) -> dict:
    lowered = text.lower()
    if "charged" in lowered or "refund" in lowered:
        return {"topic": "billing", "priority": "high"}
    if "package" in lowered or "shipping" in lowered:
        return {"topic": "shipping", "priority": "medium"}
    return {"topic": "other", "priority": "low"}

print(classify("Charged twice for order A-1001"))

Default arguments show up constantly: def search(query: str, k: int = 4). Keyword arguments keep call sites readable when a client has ten options.

Files and JSON

json.loads reads a string. json.load reads a file object. Always pass encoding utf-8. Model output often arrives wrapped in a markdown fence. Strip the fence or, better, ask for structured output so you never parse by hand.

Starter

import json
raw = '{"id": "T1", "text": "refund"}'
ticket = json.loads(raw)
print(ticket["id"])
print(json.dumps(ticket))

Goldens, configs, and model replies live in JSON. Always set the encoding, and use a with block so the file closes.

import json
from pathlib import Path

raw = Path("tickets.json").read_text(encoding="utf-8")
tickets = json.loads(raw)

result = {"id": "T1", "topic": "billing", "priority": "high"}
Path("out.json").write_text(json.dumps(result, indent=2), encoding="utf-8")

Errors, timeouts, retries

Backoff means 0.5s, then 1s, then 2s, so you do not stampede a rate limit. Catch a narrow error. A 400 from a bad prompt will fail forever and burn quota if you retry it. After the last try, raise a new error that names how many attempts you made.

Starter

def call():
    raise TimeoutError("slow model")

try:
    call()
except TimeoutError as exc:
    print("retry this one:", exc)
# do not retry ValueError from a broken prompt; it fails the same way every time

Model calls fail: network, rate limit, bad JSON, empty content. Catch the narrow error, wait, try again, then give up with a message a human can read.

import time

def call_with_retry(fn, attempts=3):
    last_error = None
    for n in range(attempts):
        try:
            return fn()
        except (TimeoutError, ValueError) as exc:
            last_error = exc
            time.sleep(0.5 * (2 ** n))  # 0.5s, 1s, 2s
    raise RuntimeError(f"gave up after {attempts} tries") from last_error

Type hints and Pydantic

Hints alone do not check anything at runtime. Pydantic does. A missing field should fail here, not three services later. `model_dump()` is the plain dict you save. Field descriptions later become instructions to the model when you pass the class to a structured-output API.

Starter

from pydantic import BaseModel

class TicketLabel(BaseModel):
    topic: str
    priority: str

label = TicketLabel(topic="billing", priority="high")
print(label.model_dump())

A type hint documents the shape. Pydantic checks it at runtime, which is what you want when a model fills the fields.

from pydantic import BaseModel, Field

class TicketLabel(BaseModel):
    topic: str = Field(description="billing, shipping, or other")
    priority: str = Field(description="low, medium, or high")

label = TicketLabel(topic="billing", priority="high")
print(label.model_dump())

If the model returns priority: 1 and you declared str, Pydantic can coerce or reject it depending on the version and config. Rejection is what you want for labels. Later LangChain pages pass this same class to with_structured_output.

Async when you wait on many calls

Twenty calls of one second each take about twenty seconds in a line and about one second side by side. Cap concurrency with a semaphore of 5 so you do not hit the rate limit. Do not mix time.sleep into async code. It blocks the whole loop. Use asyncio.sleep.

Starter

import asyncio

async def label_one(ticket_id: str) -> str:
    await asyncio.sleep(0.01)  # stand-in for the network
    return ticket_id

async def main():
    done = await asyncio.gather(label_one("T1"), label_one("T2"))
    print(done)

asyncio.run(main())

A model call is mostly waiting. async lets you wait on several at once instead of in a line.

import asyncio

async def label_one(ticket):
    await asyncio.sleep(0.1)  # stand-in for the network
    return {"id": ticket["id"], **classify(ticket["text"])}

async def main():
    tasks = [label_one(t) for t in tickets]
    done = await asyncio.gather(*tasks)
    print(done)

asyncio.run(main())

Rule of thumb: if you label 20 tickets and each call takes 1 second, a loop takes about 20 seconds. gather takes about 1 second plus a little overhead. Cap concurrency (a semaphore of 5) so you do not stampede the rate limit.

Put it together: ticket triage

Keep the stand-in until the shape is right. Then replace only `stand_in` with a real client. Failures should land in an errors file, not disappear into a default label, once you care about quality. Pin the packages in requirements.txt.

Starter

import json
from pydantic import BaseModel, ValidationError

class TicketLabel(BaseModel):
    topic: str
    priority: str

def stand_in(text: str) -> str:
    topic = "billing" if "refund" in text.lower() else "other"
    return json.dumps({"topic": topic, "priority": "low"})

try:
    print(TicketLabel.model_validate_json(stand_in("refund me")))
except ValidationError as exc:
    print("bad model output", exc.error_count())

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

class TicketLabel(BaseModel):
    topic: str = Field(pattern="^(billing|shipping|other)$")
    priority: str = Field(pattern="^(low|medium|high)$")

def stand_in_model(text: str) -> str:
    guess = classify(text)
    return json.dumps(guess)

def triage(path: str) -> list[dict]:
    tickets = json.loads(Path(path).read_text(encoding="utf-8"))
    saved = []
    for ticket in tickets:
        try:
            label = TicketLabel.model_validate_json(stand_in_model(ticket["text"]))
        except ValidationError:
            label = TicketLabel(topic="other", priority="low")
        saved.append({"id": ticket["id"], **label.model_dump()})
    Path("labels.json").write_text(json.dumps(saved, indent=2), encoding="utf-8")
    return saved

That is the whole shape of an AI program: load work, call a model, validate, persist. Frameworks add tracing and tools around this loop. They do not replace it.

Practice

• Add a fourth ticket and a rule so “password reset” becomes topic account. Update the regex.
• Save failed validations to errors.json instead of silently relabeling them.
• Rewrite triage with asyncio.gather and a semaphore of 3.
• Explain model_dump() versus printing the Pydantic object.