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
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
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
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
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.
View the code here
[
{
"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
python triage.py
cat results.jsonSorted 5 tickets. For a person: [4]
[
{
"id": 1,
"category": "billing",
"priority": 4,
"checked": true
},
{
"id": 2,
"category": "shipping",
"priority": 3,
"checked": true
},
{
"id": 3,
"category": "billing",
"priority": 4,
"checked": true
},
{
"id": 4,
"category": null,
"priority": null,
"checked": false
},
{
"id": 5,
"category": "shipping",
"priority": 3,
"checked": true
}
]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
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 NoneThis 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.
pytest -q.. [100%] 2 passed in 1.05s
The lessons behind triage.py
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.
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.contentINSTRUCTIONS 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.
import asyncio
from triage import main
asyncio.run(main())
with open("results.json") as f:
print(f.read())Sorted 5 tickets. For a person: []
[
{
"id": 1,
"category": "billing",
"priority": 4,
"checked": true
},
{
"id": 2,
"category": "shipping",
"priority": 4,
"checked": true
},
{
"id": 3,
"category": "billing",
"priority": 3,
"checked": true
},
{
"id": 4,
"category": "other",
"priority": 2,
"checked": true
},
{
"id": 5,
"category": "shipping",
"priority": 3,
"checked": true
}
]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
otherwith 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-in | openai/gpt-oss-120b on Groq | |
|---|---|---|
| Needs | Nothing | GROQ_API_KEY and pip install openai |
| Ticket 4, the password question | A sentence, so a person sorts it | other, priority 2, in the run above |
| Same answers every run | Yes | Not 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.
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:
| Topic | What it is for |
|---|---|
| Tuples and sets | A tuple is a list that cannot change; a set holds each value once and checks membership fast. |
while, break and match | Loops that run until a condition changes, leaving a loop early, and matching a value against several shapes. |
Docstrings and *args | Describing a function inside it, and collecting any number of positional arguments. |
pathlib | Working with file paths that behave the same on Windows, macOS and Linux. |
logging | Recording what a program did, with levels, instead of print. |
| Environment variables | Keeping API keys out of your code, read with os.environ as the real model section does. |
Generators and yield | Producing values one at a time, which is how streamed model output arrives. |
| Writing decorators | You have used @dataclass and @pytest.mark.parametrize; writing your own comes later. |
| HTTP requests | Calling any web API with httpx or requests. |
| Regular expressions | Finding patterns in text with the re module. |
Related
- Previous: pytest.raises and parametrize
- Next course: APIs for AI
- Reference: The Python Tutorial
- Add retries: copy
ask_with_retriesfrom Timeouts and retries, changeawait model.ask(text)toawait model(text)so it takes a function, and callawait ask_with_retries(ask_model, ticket["text"])intriage. - Add a semaphore so no more than two tickets are asked at once, and time the run.
- Add a test that loads
results.jsonafter runningmainand checks it has five rows.
Slow is fine. Stopping is the only problem.