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

Project: ticket triage

The ticket triage program is a Python script that reads support tickets, asks a model to sort each one, checks every answer with Pydantic and saves the results.

Last updated: 30 Sep, 2026 · Python 3.14 · Pydantic 2.12 · pytest 9.1

This is the program described in Python for AI overview, built from the pieces of every part. Make a folder with a virtual environment, as in Virtual environments and pip, and install pydantic and pytest into it. Put tickets.json from JSON in the folder. Then write triage.py in four pieces; it replaces the small triage.py from pytest.

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 Pydantic models: 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 Stand-in model, now written with async def and a half-second wait, as in async and await, so it behaves like a call over the network.

Sorting 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 model_validate_json. 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 Function arguments, used to spread a dictionary out instead of collecting one.

Sorting 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 (JSON), sort every ticket at once with gather (asyncio.gather), 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.

Project files used on this pageThis lesson builds on a project from earlier lessons. The code below imports this file. Click a file to see its code, or follow the link to the lesson that wrote it. To run the code yourself, keep it in the same folder.
View the code here
tickets.json
[
  {
    "id": 1,
    "customer": "Asha",
    "text": "I was charged twice for one order"
  },
  {
    "id": 2,
    "customer": "Ben",
    "text": "My parcel has not arrived"
  },
  {
    "id": 3,
    "customer": "Chen",
    "text": "Can I get a refund for the blue mug?"
  },
  {
    "id": 4,
    "customer": "Dara",
    "text": "How do I change my password?"
  },
  {
    "id": 5,
    "customer": "Eli",
    "text": "The parcel arrived but the box was crushed"
  }
]

Sorting the five tickets

Example
python triage.py
cat results.json

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

Testing the program

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

This test_triage.py replaces the one from pytest: 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

The lessons behind triage.py

Five tickets through triage.py
triage(ticket), for each ticketanswertickets.jsonfive ticketsasyncio.gatherall five at onceask_modelthe stand-in, half a secondTriage.model_validate_jsonno JSON: checked is Falseresults.jsonand the ids for a person
Hover or tap a piece to see what it is and which lesson built it.
Trace a run

Pick one to watch it run, step by step.

Swapping in a real modelOptional

The stand-in answers the same way every run. A real model is one replacement of part 2 of triage.py; parts 1, 3 and 4 stay as they are. It needs the free Groq key from Installation and setup and one more package: pip install openai. Groq accepts the OpenAI client, pointed at its own address.

python
import os

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.groq.com/openai/v1",
    api_key=os.environ["GROQ_API_KEY"],
)
INSTRUCTIONS = ("Sort the support ticket into billing, shipping or other, with a priority "
                'from 1 (low) to 5 (urgent). Reply with JSON only, like {"category": "billing", "priority": 4}.')


async def ask_model(text: str) -> str:
    reply = await client.chat.completions.create(
        model="openai/gpt-oss-120b",
        messages=[{"role": "system", "content": INSTRUCTIONS},
                  {"role": "user", "content": text}],
    )
    return reply.choices[0].message.content

INSTRUCTIONS is the system message, the standing instruction from Lists of dictionaries. It names the three categories and asks for JSON only, because the check in part 3 refuses anything else. reply.choices[0].message.content is the text of the answer, the same kind of string the stand-in returned.

Save this as run.py next to triage.py, or run python triage.py and open results.json. Keep the stand-in version of part 2 for pytest: test_triage.py expects the password question to go to a person, and the real model sorts it instead.

ExampleAPI keyrun.py
import asyncio

from triage import main

asyncio.run(main())

with open("results.json") as f:
    print(f.read())

What the real model sent back

  • All five answers passed the check. Every reply was JSON with an allowed category and a priority from 1 to 5, so the list for a person is empty.
  • Ticket 4 was sorted. The stand-in answered the password question with a sentence; the real model filed it as other with priority 2.
  • Some priorities differ from the stand-in's. Ticket 2 got 4 instead of 3, and ticket 3 got 3 instead of 4. The model chooses these itself, and another run can choose differently.
  • Parts 1, 3 and 4 ran unchanged. The same check and the same results file work with either model.

Stand-in vs real model in triage.py

Stand-inopenai/gpt-oss-120b on Groq
NeedsNothingGROQ_API_KEY and pip install openai
Ticket 4, the password questionA sentence, so a person sorts itother, priority 2, in the run above
Same answers every runYesNot guaranteed

Where this program's pattern shows up

  • Every agent framework later in the roadmap: a model call, a check on what came back, and a decision about what to do with a bad answer.
  • Batch jobs over support tickets, emails or documents, with a person for whatever the check refuses.
Watch out. Never paste the key into triage.py. os.environ["GROQ_API_KEY"] reads it from the terminal, so the file can be shared; without the key set, that line raises a KeyError naming it.

Python topics to learn next

The course skipped these parts of Python. Learn them 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 as the real model section does.
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 any web API with httpx or requests.
Regular expressionsFinding patterns in text with the re module.
Try it yourself
  • Add retries: copy ask_with_retries from Timeouts and retries, change await model.ask(text) to await model(text) so it takes a function, and call await ask_with_retries(ask_model, ticket["text"]) in triage.
  • 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.

Slow is fine. Stopping is the only problem.