Python for AIPython 3.10+ · Pydantic 2.12
0%
1
Curious builder0 XP earned · 300 to level 2
0 daysFinish a lesson to begin
Badge collection0 of 6 unlocked
33 small wins to finish your path

The ticket triage program, end to end

Lesson 0 promised a program that reads support tickets, asks a model to sort each one, checks every answer and saves the results. Here it is.

Make a folder with a virtual environment, as in lesson 29, and install pydantic and pytest into it. Put tickets.json from lesson 16 in the folder. Then write triage.py in four pieces.

The shape of an answer

Exampletriage.py, part 1
import asyncio
import json
from typing import Literal

from pydantic import BaseModel, Field, ValidationError


class Triage(BaseModel):
    category: Literal["billing", "shipping", "other"]
    priority: int = Field(ge=1, le=5)

The imports, then the Triage model from lesson 24: three allowed categories and a priority from 1 to 5.

The model

Exampletriage.py, part 2
async def ask_model(text: str) -> str:
    await asyncio.sleep(0.5)
    text = text.lower()
    if "charged" in text or "refund" in text:
        return '{"category": "billing", "priority": 4}'
    if "parcel" in text or "arrived" in text:
        return '{"category": "shipping", "priority": 3}'
    return "I am not sure how to sort this one."

The stand-in from lesson 17, now written with async def and a half-second wait, as in lesson 26, so it behaves like a call over the network.

One ticket

Exampletriage.py, part 3
async def triage(ticket: dict) -> dict:
    answer = await ask_model(ticket["text"])
    try:
        result = Triage.model_validate_json(answer)
    except ValidationError:
        return {"id": ticket["id"], "category": None, "priority": None, "checked": False}
    return {"id": ticket["id"], **result.model_dump(), "checked": True}

Ask, then check with model_validate_json from lesson 25. An answer that fails the check is not guessed at: the row records checked: False and no category, so nothing downstream treats it as sorted.

**result.model_dump() inside the dictionary copies every field of the checked answer into it. It is the same pair of stars as **settings in lesson 12, used to spread a dictionary out instead of collecting one.

Every ticket

Exampletriage.py, part 4
async def main():
    with open("tickets.json") as f:
        tickets = json.load(f)

    results = await asyncio.gather(*[triage(ticket) for ticket in tickets])

    with open("results.json", "w") as f:
        json.dump(results, f, indent=2)

    unchecked = [row["id"] for row in results if not row["checked"]]
    print(f"Sorted {len(results)} tickets. For a person: {unchecked}")


if __name__ == "__main__":
    asyncio.run(main())

Read the file (lesson 16), sort every ticket at once with gather (lesson 27), write the results, and list the ids a person should look at.

if __name__ == "__main__": is true only when the file is run directly with python triage.py. When the tests import triage, main does not run, so importing the file does not start sorting tickets.

Run it

Example
python triage.py
cat results.json

All five tickets in about half a second. Ticket 4, the password question, is the one the model could not sort, and it is listed for a person instead of being filed as something it is not.

Prove it

Exampletest_triage.py
import asyncio

from triage import triage


def test_billing_ticket_is_checked():
    row = asyncio.run(triage({"id": 1, "text": "I was charged twice"}))
    assert row["category"] == "billing"
    assert row["checked"] is True


def test_unclear_ticket_goes_to_a_person():
    row = asyncio.run(triage({"id": 4, "text": "How do I change my password?"}))
    assert row["checked"] is False
    assert row["category"] is None

One test for a ticket that sorts, one for the failure. asyncio.run calls the async triage from an ordinary test function.

Example
pytest -q

Where each piece came from

triage.py, by lesson
Datalists of dictionaries, lesson 9JSON files, lesson 16The modelthe stand-in, lesson 17async def, lesson 26CheckingPydantic, lesson 24model_validate_json, lesson 25try and except, lesson 18Speedgather, lesson 27Proofpytest, lesson 30testing failures, lesson 31triage.py

Things to add

Try it yourself
  • Wrap the call in ask_with_retries from lesson 28, and give the stand-in a FlakyModel-style failure.
  • Add a semaphore so no more than two tickets are asked at once, and time the run.
  • Add a test that loads results.json after running main and checks it has five rows.

What this course left out

These are worth learning next, when a project needs them:

TopicWhat it is for
Tuples and setsA tuple is a list that cannot change; a set holds each value once and checks membership fast.
while, break and matchLoops that run until a condition changes, leaving a loop early, and matching a value against several shapes.
Docstrings and *argsDescribing a function inside it, and collecting any number of positional arguments.
pathlibWorking with file paths that behave the same on Windows, macOS and Linux.
loggingRecording what a program did, with levels, instead of print.
Environment variablesKeeping API keys out of your code, read with os.environ.
Generators and yieldProducing values one at a time, which is how streamed model output arrives.
Writing decoratorsYou have used @dataclass and @pytest.mark.parametrize; writing your own comes later.
HTTP requestsCalling a real API with httpx or requests.
Regular expressionsFinding patterns in text with the re module.

You understood something today that you didn't yesterday.