Lesson 4 — classification with structured output¶
What you will build¶
The first model call, and the smallest one that is useful: a category and a confidence, over the closed schema subset the AI path's lesson 5 describes. The model answers; the service decides whether the answer may move a ticket. That order is the whole design in one sentence.
Two things in the transcript are the lesson:
- a reply that is valid JSON but the wrong shape leaves the ticket exactly where it was, and the client is told nothing about what the model said;
- every provider failure looks the same from outside. The typed error is for the application, not for the caller.
What you need¶
Nothing new: the AI facade is built into the package, and the fake provider is what these lessons use.
The classification request carries response_schema, and check_schema accepts only the keywords the
package implements: type, properties, required, and additionalProperties. oneOf, $ref, and
the rest of the JSON Schema vocabulary are refused at build time rather than half-honoured at runtime.
examples/support_desk/triage.py
"""The triage workflow: classify, ground, draft — and stop before anything is sent.
This is the composition the path is about. Each step does one thing and writes one row; the edges
say what may follow; and the graph **ends at a draft**. There is no step that sends, no tool that
sends, and no path from a model's output to a customer's inbox: sending is a separate call that a
person with the right role has to make, which is the boundary the whole design protects.
Each capability is also a method, so the same code serves a single call from a route and a step in
the graph. A composition that only exists inside a workflow is a composition nobody can test in
isolation.
"""
from __future__ import annotations
from collections.abc import Mapping
from typing import Annotated, Any
from support_desk.knowledge import KnowledgeBase
from support_desk.security import Principal
from support_desk.service import CATEGORIES, CONFIDENCES, TicketService
from support_desk.usage import UsageLog
from xyberos import Inject, Retry, StepContext, Workflow, seconds
from xyberos.ai import AIFacade, ChatRequest, Message, Usage
CLASSIFICATION_SCHEMA: dict[str, object] = {
"type": "object",
"properties": {
"category": {"type": "string", "enum": list(CATEGORIES)},
"confidence": {"type": "string", "enum": list(CONFIDENCES)},
},
"required": ["category", "confidence"],
"additionalProperties": False,
}
CLASSIFY_PROMPT = "Classify a ticket: the category that fits best, and how confident you are."
DRAFT_PROMPT = (
"Draft a short reply to the customer. Use the quoted policy if one is given, and promise "
"nothing the policy does not say. Write only the reply."
)
#: Below this confidence the desk asks a person instead of drafting: a draft costs money and
#: attention, and a wrong category is worse than a slow one.
LOW_CONFIDENCE = "low"
class Triage:
"""Runs the desk's triage, and owns none of the transitions it performs.
The transitions belong to the service, which keeps this module about *composing* a model call
with an application rule rather than about deciding who may do what.
"""
def __init__(
self,
tickets: Annotated[TicketService, Inject()],
knowledge: Annotated[KnowledgeBase, Inject()],
assistant: Annotated[AIFacade, Inject()],
usage: Annotated[UsageLog, Inject()],
) -> None:
self._tickets = tickets
self._knowledge = knowledge
self._assistant = assistant
self._usage = usage
self._workflow = self._build()
@property
def workflow(self) -> Workflow:
"""Return the graph, which is a value: it can be rendered and inspected without running."""
return self._workflow
async def classify(self, principal: Principal, ticket_id: int) -> dict[str, object]:
"""Ask the model what a ticket is, then let the service record the answer."""
ticket = await self._tickets.get_ticket(principal, ticket_id)
response = await self._assistant.chat(
ChatRequest(
messages=(
Message.system(CLASSIFY_PROMPT),
Message.user(f"{ticket['subject']}\n\n{ticket['body']}"),
),
response_schema=CLASSIFICATION_SCHEMA,
)
)
payload = response.structured
if not isinstance(payload, Mapping):
raise AssertionError("a response schema that validated produced no payload")
classified = await self._tickets.classify(
principal,
ticket_id,
category=payload["category"],
confidence=payload["confidence"],
)
await self._record(principal, ticket_id, "classify", response.model, response.usage)
return classified
async def ground(self, principal: Principal, ticket_id: int) -> str:
"""Return the passage that answers a ticket best, or an empty string when none matches."""
ticket = await self._tickets.get_ticket(principal, ticket_id)
hits = await self._knowledge.search(f"{ticket['subject']}\n{ticket['body']}", limit=1)
return str(hits[0]["text"]) if hits else ""
async def draft(
self, principal: Principal, ticket_id: int, *, grounding: str = ""
) -> dict[str, object]:
"""Write a draft and store it, which sends nothing and waits for a person."""
ticket = await self._tickets.get_ticket(principal, ticket_id)
response = await self._assistant.chat(
ChatRequest(
messages=(
Message.system(DRAFT_PROMPT),
Message.user(
f"{ticket['subject']}\n\n{ticket['body']}\n\n"
f"Policy: {grounding or 'none found'}"
),
)
)
)
suggestion = await self._tickets.attach_suggestion(
principal, ticket_id, body=response.message.content, model=response.model
)
await self._record(principal, ticket_id, "draft", response.model, response.usage)
return suggestion
async def run(self, principal: Principal, ticket_id: int) -> dict[str, object]:
"""Run the workflow for one ticket, and return where it finished."""
result: Any = await self._workflow.run(
input={"principal": principal, "ticket": ticket_id}, timeout=seconds(30)
)
state = await self._tickets.state_of(principal, ticket_id)
return {
"ticket": ticket_id,
"state": state,
"steps": {step: outcome.status.value for step, outcome in result.steps.items()},
}
def _build(self) -> Workflow:
"""Define the graph once, at construction, so a malformed workflow fails at startup."""
workflow = (
Workflow("triage", version="1", input_type=dict)
.step("classify", self._classify_step, retry=Retry(max_attempts=2), timeout=seconds(5))
.step("ground", self._ground_step, timeout=seconds(5))
.step("draft", self._draft_step, retry=Retry(max_attempts=2), timeout=seconds(10))
.step("flag", self._flag_step)
.edge("classify", "ground", when=lambda value: value["confidence"] != LOW_CONFIDENCE)
.edge("classify", "flag", when=lambda value: value["confidence"] == LOW_CONFIDENCE)
.edge("ground", "draft")
)
workflow.validate()
return workflow
async def _classify_step(self, value: dict[str, Any], context: StepContext) -> dict[str, Any]:
"""The classification capability as a step: the model answers, the service records it."""
principal, ticket_id = _call(value)
classified = await self.classify(principal, ticket_id)
return {
**value,
"category": classified["category"],
"confidence": classified["confidence"],
}
async def _ground_step(self, value: dict[str, Any], context: StepContext) -> dict[str, Any]:
"""The retrieval capability as a step, so the draft can quote a passage."""
principal, ticket_id = _call(value)
return {**value, "grounding": await self.ground(principal, ticket_id)}
async def _draft_step(self, value: dict[str, Any], context: StepContext) -> dict[str, Any]:
"""The drafting capability as a step: it writes a suggestion and nothing else."""
principal, ticket_id = _call(value)
suggestion = await self.draft(
principal, ticket_id, grounding=str(value.get("grounding", ""))
)
return {**value, "suggestion": suggestion["id"]}
async def _flag_step(self, value: dict[str, Any], context: StepContext) -> dict[str, Any]:
"""Leave the ticket classified and say why, instead of guessing at a draft."""
principal, ticket_id = _call(value)
detail = f"confidence={value['confidence']}: a person classifies this one"
await self._tickets.record_event(principal, ticket_id, "flagged", detail)
return value
async def _record(
self,
principal: Principal,
ticket_id: int,
step: str,
model: str,
usage: Usage,
) -> None:
"""Write one usage row for a call the provider reported numbers for."""
await self._usage.record(
tenant=principal.tenant,
ticket_id=ticket_id,
step=step,
model=model,
input_tokens=usage.input_tokens,
output_tokens=usage.output_tokens,
)
def _call(value: Mapping[str, Any]) -> tuple[Principal, int]:
"""Return the principal and ticket id a step was given, refusing anything else."""
principal = value["principal"]
if not isinstance(principal, Principal):
raise AssertionError("a workflow step was given something other than a principal")
return principal, int(value["ticket"])
examples/learn_hybrid/hybrid04_classification.py
"""Lesson 4 — classification with structured output over the closed schema subset.
Run it with:
python -m learn_hybrid.hybrid04_classification
"""
from __future__ import annotations
import asyncio
import tempfile
from pathlib import Path
from support_desk.app import create_app
from support_desk.migrate import migrate
from support_desk.security import Principal, PrincipalVerifier
from support_desk.service import CATEGORIES, CONFIDENCES, TicketService
from support_desk.triage import CLASSIFICATION_SCHEMA, Triage
from xyberos.ai import FakeModelProvider, ModelCapabilities, ModelReply, Usage, check_schema
from xyberos.database import SqliteSettings
from xyberos.errors import XyberosError
from xyberos.http import HttpAdapter
from xyberos.http.testing import create_test_client
SECRET = "a-secret-the-deployment-keeps"
VERIFIER = PrincipalVerifier(SECRET)
ADA = VERIFIER.headers("ada@example.com", "acme", ["requester"])
SAM = VERIFIER.headers("sam@desk", "acme", ["support_agent"])
LEE = VERIFIER.headers("lee@desk", "acme", ["support_lead"])
TICKET = {"subject": "Where is my refund?", "body": "Order 42 came back on Monday."}
#: The three answers a provider can give, and the one thing they all have to do: leave the ticket
#: exactly where it was.
CASES = (
("a valid answer", ModelReply.json_payload({"category": "refund", "confidence": "high"})),
("the wrong shape", ModelReply.wrong_shape({"category": "banana", "confidence": "high"})),
("not JSON at all", ModelReply.malformed(content="I think it is about a refund.")),
)
def classify(reply: ModelReply) -> tuple[int, str, dict[str, object]]:
"""Classify one ticket through the route, and return what a client can see."""
with tempfile.TemporaryDirectory() as directory:
settings = SqliteSettings(path=Path(directory) / "support.db")
migrate(settings)
provider = FakeModelProvider(
[reply], capabilities=ModelCapabilities(structured_output=True)
)
app = create_app(settings=settings, secret=SECRET, provider=provider)
with create_test_client(HttpAdapter(app)) as client:
client.post("/tickets", json=TICKET, headers=ADA)
response = client.post("/tickets/1/classify", headers=SAM)
ticket: dict[str, object] = client.get("/tickets/1", headers=SAM).json()
return response.status_code, response.text, ticket
async def typed_failure() -> None:
"""The same failure seen from inside the application, which is where the type matters."""
with tempfile.TemporaryDirectory() as directory:
settings = SqliteSettings(path=Path(directory) / "support.db")
migrate(settings)
provider = FakeModelProvider(
[ModelReply.wrong_shape({"category": "banana", "confidence": "high"})],
capabilities=ModelCapabilities(structured_output=True),
)
app = create_app(settings=settings, secret=SECRET, provider=provider)
agent = Principal("sam@desk", "acme", frozenset({"support_agent"}))
async with app.lifespan():
tickets = await app.resolve(TicketService)
triage = await app.resolve(Triage)
await tickets.open_ticket(agent, TICKET["subject"], TICKET["body"])
try:
await triage.classify(agent, 1)
except XyberosError as error:
print("inside the application:", type(error).__name__, "-", error)
def main() -> None:
"""One schema, one ticket, and the three things a provider can answer with."""
check_schema(CLASSIFICATION_SCHEMA, location="hybrid lesson 4")
print("accepted keywords:", sorted(CLASSIFICATION_SCHEMA))
print("categories: ", list(CATEGORIES), "confidences:", list(CONFIDENCES))
for label, reply in CASES:
status, text, ticket = classify(reply)
leaked = "banana" in text
print(
f"{label:<16} {status} ticket={ticket['state']:<10} leaked the model's text: {leaked}"
)
if status == 200:
print(f" -> {ticket['category']}/{ticket['confidence']}")
print(
"every provider failure looks the same to a client:",
"the request could not be completed" in text,
)
asyncio.run(typed_failure())
without_structured = FakeModelProvider(
[ModelReply.json_payload({"category": "refund", "confidence": "high"})],
capabilities=ModelCapabilities(structured_output=False),
)
with tempfile.TemporaryDirectory() as directory:
settings = SqliteSettings(path=Path(directory) / "support.db")
migrate(settings)
app = create_app(settings=settings, secret=SECRET, provider=without_structured)
with create_test_client(HttpAdapter(app)) as client:
client.post("/tickets", json=TICKET, headers=ADA)
response = client.post("/tickets/1/classify", headers=SAM)
print("a model without structured output ->", response.status_code)
print(
" and the ticket is still:", client.get("/tickets/1", headers=SAM).json()["state"]
)
priced = FakeModelProvider(
[ModelReply.json_payload({"category": "question", "confidence": "low"}, usage=Usage(31, 7))]
)
with tempfile.TemporaryDirectory() as directory:
settings = SqliteSettings(path=Path(directory) / "support.db")
migrate(settings)
app = create_app(settings=settings, secret=SECRET, provider=priced)
with create_test_client(HttpAdapter(app)) as client:
client.post(
"/tickets", json={"subject": "Is there an API?", "body": "Asking."}, headers=ADA
)
client.post("/tickets/1/classify", headers=SAM)
steps = client.get("/usage", headers=LEE).json()["steps"]
print("the call is recorded:", steps[0]["step"], steps[0]["input_tokens"], "tokens in")
if __name__ == "__main__":
main()
How to run it¶
What it prints¶
accepted keywords: ['additionalProperties', 'properties', 'required', 'type']
categories: ['refund', 'question', 'complaint', 'abuse'] confidences: ['low', 'medium', 'high']
a valid answer 200 ticket=classified leaked the model's text: False
-> refund/high
the wrong shape 500 ticket=new leaked the model's text: False
not JSON at all 500 ticket=new leaked the model's text: False
every provider failure looks the same to a client: True
inside the application: AISchemaError - the model response at $.category must be one of: 'refund', 'question', 'complaint', 'abuse'; got 'banana'
a model without structured output -> 500
and the ticket is still: new
the call is recorded: classify 31 tokens in
The 422's absence is the point: a model's bad answer is not the client's mistake, so it is a 500
whose message says only that the request could not be completed. The model's text does not reach the
caller — leaked the model's text: False — because a response body is a place secrets and prompt
content travel by accident.
Read the inside the application line next to it: the same failure, seen where it matters, is a typed
AISchemaError that names the path ($.category) and the value that failed. An application logs that;
a client never sees it.
Common mistakes¶
- Treating a model's answer as input to validate. It is output to check: the schema is the contract, and a wrong shape is a failure of the dependency, not of the caller.
- Letting the model perform the transition.
classifyasks for a value;TicketService.classifywrites the row. If the model could write, there would be no place to enforce the matrix. structured()when you also need the usage.AIFacade.structuredreturns the payload alone.chatwith aresponse_schemareturns the payload and the tokens, which is what the usage row in lesson 10 needs.- A free-text prompt for a category. An enum in the schema is cheaper than parsing prose, and it fails loudly instead of silently inventing a category.
- Catching the provider's error and returning it to the client. Log the type; answer generically.
Next¶
Lesson 5 gives the draft something to quote: passages you store, vectors you compute, and a similarity query you write.