Project: ticket triage

The ticket triage project is a small Python codebase that reads support tickets, asks a model to sort each one, checks every answer with Pydantic, logs what it did and saves the results.

Last updated: 13 Oct, 2026 · Python 3.12 · Pydantic 2.12 · pytest 9.1

The Logging lesson recorded what a program does. This final project puts logging, type-checked data, async calls and tests together into the layout a real AI project uses: a src/ package for the code, a tests/ folder, a main.py to run it, and a requirements.txt. Make a folder with a virtual environment, as in Virtual environments and pip, install pydantic and pytest, and put tickets.json from JSON in the folder.

The project layout

Six files in two folders. The code lives in src/, split so the part you would swap for a real model is on its own:

• requirements.txt: the packages to install.
• src/__init__.py: an empty file that marks src as a package you can import from.
• src/model.py: the stand-in model, the one file you replace for production.
• src/pipeline.py: the Triage schema and the sorting logic, with logging.
• main.py: the entry point: set up logging, read the tickets, run, save the results.
• tests/test_pipeline.py: the tests.

Listing the packages

Pin the two direct dependencies, as in Virtual environments and pip:

pydantic==2.12.0
pytest==9.1.1

The stand-in model

The model stands in for a real call, so the project runs with no key, as in Stand-in model. The half-second wait with async def, from async and await, makes it behave like a call over the network. It is one function on its own, so swapping in a real model later changes this file alone.

import asyncio


async def ask_model(text: str) -> str:
    """Stand-in model. Swap this one function for a real client in production."""
    await asyncio.sleep(0.1)
    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}'
    # a deliberate miss: plain text, not json, so the pipeline has to cope
    return "I am not sure how to sort this one."

The Triage schema

The Pydantic models schema, plus a named logger from Logging that the next part writes to:

import asyncio
import logging
from typing import Literal

from pydantic import BaseModel, Field, ValidationError

from src.model import ask_model

logger = logging.getLogger("triage")


class Triage(BaseModel):
    # Literal pins the category to three values; priority has to sit in 1..5
    category: Literal["billing", "shipping", "other"]
    priority: int = Field(ge=1, le=5)

Sorting one ticket

triage asks the model and checks the answer with model_validate_json from model_validate_json. An answer that fails the check is not guessed at: the logger records that the ticket needs a person, and the row carries checked: False. run_all sorts every ticket at once with gather from asyncio.gather.

async def triage(ticket: dict) -> dict:
    answer = await ask_model(ticket["text"])
    try:
        result = Triage.model_validate_json(answer)
    except ValidationError:
        # a bad answer is set aside for a person rather than crashing the batch
        logger.warning("ticket %s needs a person", ticket["id"])
        return {"id": ticket["id"], "category": None, "priority": None, "checked": False}
    logger.info("ticket %s sorted as %s", ticket["id"], result.category)
    return {"id": ticket["id"], **result.model_dump(), "checked": True}


async def run_all(tickets: list) -> list:
    # fire every ticket off at once instead of waiting for each in turn
    return await asyncio.gather(*[triage(t) for t in tickets])

logger.warning("ticket %s needs a person", ticket["id"]) leaves the %s for logging to fill in, so the id is only turned into text if the line is written. **result.model_dump() spreads the checked answer's fields into the row, the same stars as **settings in Function arguments.

Writing main.py

main.py sets up logging to a file, reads tickets.json from JSON, runs the pipeline, writes results.json, and prints a one-line summary. logging.basicConfig runs once here, at the top, as Logging shows.

import asyncio
import json
import logging

from src.pipeline import run_all

# log to a file, overwriting it on each run
logging.basicConfig(filename="run.log", filemode="w", level=logging.INFO,
                    format="%(name)s %(levelname)s %(message)s")


async def main():
    with open("tickets.json") as f:
        tickets = json.load(f)
    results = await run_all(tickets)
    with open("results.json", "w") as f:
        json.dump(results, f, indent=2)
    # the ones the model could not sort go to a person
    unchecked = [r["id"] for r in results if not r["checked"]]
    print(f"Sorted {len(results)} tickets. For a person: {unchecked}")


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

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

[
  {
    "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"
  }
]

Running the project

import runpy

runpy.run_path("main.py", run_name="__main__")
print(open("results.json").read())
print("--- run.log ---")
print(open("run.log").read())

Sorted 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
  }
]
--- run.log ---
triage INFO ticket 1 sorted as billing
triage INFO ticket 2 sorted as shipping
triage INFO ticket 3 sorted as billing
triage WARNING ticket 4 needs a person
triage INFO ticket 5 sorted as shipping

All five tickets are sorted in about half a second. results.json holds a row for each, and ticket 4, the password question, is marked checked: false and listed for a person instead of filed as something it is not. run.log shows the named logger's record of the run, written by src/pipeline.py.

The tests

Two tests, as in pytest: one for a ticket that sorts, one for the failure. asyncio.run calls the async triage from an ordinary test function.

import asyncio

from src.pipeline import triage


def test_billing_ticket_is_checked():
    # asyncio.run drives the async pipeline from a plain sync test
    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

Running the tests

import subprocess
import sys

result = subprocess.run([sys.executable, "-m", "pytest", "-q"],
                        capture_output=True, text=True)
print(result.stdout)

..                                                                       [100%]
2 passed in 0.26s

Running python -m pytest from the project root finds the tests in tests/ and runs them against the code in src/. Both pass. Run this after any change, and a broken parser or a wrong category shows up as an F at once.

The lessons behind the project

Swapping in a real modelOptional

The stand-in answers the same way every run. A real model is one replacement of src/model.py; every other file stays as it is. 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

# groq speaks the openai api, so the same client works against its url
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 src/pipeline.py refuses anything else. Shown as the drop-in swap, not run here: with a real model ticket 4 would be sorted instead of sent to a person, so the stand-in is kept for the tests above.

Python topics to learn next

The course left these parts of Python out. Learn them when a project needs them:

Related

• Previous: Logging
• Next course: APIs for AI
• Reference: The Python Tutorial