Lesson 5 — structured output

What you will build

A model answer an application can act on: a payload validated against a schema you wrote, with three distinct refusals for the three ways it can go wrong. This is the lesson that turns "the model said something" into a value a program may branch on.

What you need

Nothing new.

examples/learn_ai/ai05_structured.py

"""Lesson 5 — structured output over a closed schema subset.

Run it with:

    python -m learn_ai.ai05_structured
"""

from __future__ import annotations

import asyncio
from collections.abc import Mapping

from xyberos.ai import (
    AIFacade,
    ChatRequest,
    FakeModelProvider,
    Message,
    ModelCapabilities,
    ModelReply,
    check_schema,
    describe_schema,
)
from xyberos.errors import AIError, AISchemaError

CATEGORY_SCHEMA = {
    "type": "object",
    "properties": {
        "category": {"type": "string", "enum": ["refund", "question", "complaint"]},
        "confidence": {"type": "string", "enum": ["low", "medium", "high"]},
    },
    "required": ["category", "confidence"],
    "additionalProperties": False,
}

UNSUPPORTED_SCHEMA = {
    "type": "object",
    "properties": {"category": {"type": "string"}},
    "oneOf": [{"required": ["category"]}],
}


async def classify(provider: FakeModelProvider) -> object:
    """Ask one provider to classify a ticket, and return what the facade accepted."""
    assistant = AIFacade(provider)
    return await assistant.structured(
        ChatRequest(
            messages=(Message.system("Classify the ticket."), Message.user("My badge stopped.")),
            response_schema=CATEGORY_SCHEMA,
        )
    )


async def main() -> None:
    """One schema, three answers: accepted, wrong, and unparseable."""
    check_schema(CATEGORY_SCHEMA, location="lesson")
    print("the schema is usable:", describe_schema(CATEGORY_SCHEMA))

    capable = ModelCapabilities(structured_output=True)
    good = FakeModelProvider(
        [ModelReply.json_payload({"category": "complaint", "confidence": "high"})],
        capabilities=capable,
    )
    accepted = await classify(good)
    assert isinstance(accepted, Mapping)
    print("accepted:", dict(accepted))

    wrong = FakeModelProvider(
        [ModelReply.wrong_shape({"category": "banana", "confidence": "high"})],
        capabilities=capable,
    )
    try:
        await classify(wrong)
    except AISchemaError as error:
        print("refused:", error)

    unparseable = FakeModelProvider(
        [ModelReply.malformed(content="I think it is a complaint.")],
        capabilities=capable,
    )
    try:
        await classify(unparseable)
    except AIError as error:
        print("unparseable:", type(error).__name__, error)

    try:
        check_schema(UNSUPPORTED_SCHEMA, location="lesson")
    except AIError as error:
        print("a schema outside the subset:", type(error).__name__, error)


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

Three checks are in play, and they happen at different times:

Check	When	What it prevents
check_schema(schema, location=...)	at startup, or in a test	a schema outside the supported subset, discovered at 3am
the capability declaration (lesson 2)	before the call	asking a provider for something it cannot do
validate_payload / AISchemaError	on the response	acting on a payload that does not match what you asked for

The subset is closed on purpose. It supports type, properties, required, enum, items, additionalProperties, the length/range keywords, and minimum/maximum — and it names the keywords it does not support in the error rather than approximating them.

How to run it

PowerShellPOSIX

$env:PYTHONPATH = "src;examples"
python -m learn_ai.ai05_structured

PYTHONPATH=src:examples python -m learn_ai.ai05_structured

What it prints

the schema is usable: object
accepted: {'category': 'complaint', 'confidence': 'high'}
refused: the model response at $.category must be one of: 'refund', 'question', 'complaint'; got 'banana'
unparseable: AIInvalidPayloadError the model returned content that is not JSON (Expecting value); retry the call or adjust the prompt
a schema outside the subset: AIRequestError lesson at $ uses unsupported keyword(s): oneOf; supported keywords are: additionalProperties, enum, items, maxItems, maxLength, maximum, minItems, minLength, minimum, properties, required, type

Four lines, four different outcomes:

• accepted: the payload satisfied the schema, and the facade returned it as a mapping.
• refused: valid JSON that breaks the schema — 'banana' is not one of the three categories — is an AISchemaError. This is the failure a retry can fix and an if cannot.
• unparseable: text instead of JSON is an AIInvalidPayloadError, which the taxonomy treats as transient: trying again often succeeds, because the model's next answer may be well-formed.
• a schema outside the subset: oneOf is refused at the point you declare it, before a single call.

Common mistakes

• Parsing the content yourself. json.loads(response.message.content) throws away the validation and the error taxonomy; the facade already did both.
• Validating with a schema you never checked. A schema with an unsupported keyword is not a stricter schema, it is a different one; check_schema at startup is one line.
• Trusting a valid payload to be a correct one. A payload can satisfy the schema and still be wrong — the model chose a category that passes the enum and is not the right one. Validation buys shape, not correctness, which is lesson 16's subject.
• Adding additionalProperties: true to be flexible. It lets a model invent fields the application then ignores; keep schemas closed and version them when they change.
• Treating a schema error as fatal. It is a retryable transport failure by construction — retry with the same prompt, or with the error appended, before giving up.

Next

Lesson 6 is about everything that can go wrong on the way to that payload: timeouts, rate limits, refusals, and which of them are worth retrying.