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
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
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
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
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
python triage.py
cat results.jsonAll 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
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 NoneOne test for a ticket that sorts, one for the failure. asyncio.run calls the async triage from an ordinary test function.
pytest -qWhere each piece came from
Things to add
- Wrap the call in
ask_with_retriesfrom lesson 28, and give the stand-in aFlakyModel-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.jsonafter runningmainand checks it has five rows.
What this course left out
These are worth learning next, 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. |
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 a real API with httpx or requests. |
| Regular expressions | Finding patterns in text with the re module. |
You understood something today that you didn't yesterday.