Structured output: JSON by a schema in Python

Problem

The code that expects JSON

The support inbox gets hundreds of emails a day. A program sorts them into queues: deliveries, returns, payments, general. It asks a model for the email’s topic and expects JSON with the field intent. Here are three answers of the model for the same email:

answers = [
    'Sure! Here is the data:\n```json\n{"intent": "delivery", "order_id": "1042", "urgent": true}\n```',
    '{"intent": "shipping", "order_id": "1042", "urgent": "yes", "summary": "The courier did not come."}',
    '{"intent": "delivery", "order_id": "1042", "urgent": true, "summary": "The courier did not come, the customer leaves tomorrow."}',
]
for answer in answers:
    data = parse_json(answer)
    if data is None:
        print(f"not JSON: {answer.splitlines()[0]}")
        continue
    queue = QUEUES.get(data["intent"])
    if queue is None:
        print(f'no queue for "{data["intent"]}"')
    else:
        print(f"queue: {queue}")

not JSON: Sure! Here is the data:
no queue for "shipping"
queue: Deliveries

Only the third one works. The first is polite, but it is text around JSON and the program cannot read it. The second is JSON, but the topic is a word the program does not know and there is no such queue. In the lesson about the prompt you saw that examples help. But the program has to work on the night the model decides to be polite.

The idea

A schema, a check, a second attempt

When the answer goes into code and not to a person, it has a contract. The contract is a schema: which fields there are, the type of each and which values are allowed.

The prompt states the schema. The fields, the types and the allowed values are in the instruction.
The program reads the JSON. If it is not JSON, that is an error, not a crash.
The program checks it against the schema: is every field there, is the type right, is the value allowed, are there no extra fields.
The errors go back to the model as a new message, and it tries again.
There are at most three attempts. Then the email goes to a person.

An answer by a schema that the program checks is structured output.

The model’s output is untrusted input like everything that comes from outside: the program does not trust that it keeps the contract, it checks.

See it

Reading and checking

The email’s schema: four fields, intent with four allowed values:

TICKET = {
    "intent": {"type": "string", "enum": ["delivery", "return", "payment", "other"]},
    "order_id": {"type": "string"},
    "urgent": {"type": "boolean"},
    "summary": {"type": "string"},
}

Reading returns the object, or None if the text is not exactly one JSON object:

def parse_json(text):
    try:
        data = json.loads(text)
    except json.JSONDecodeError:
        return None
    return data if isinstance(data, dict) else None

The json module is in the standard library. json.loads raises JSONDecodeError for text that is not JSON.

The type of a value in JSON’s own words:

def json_type(value):
    if value is None:
        return "null"
    if isinstance(value, bool):
        return "boolean"
    if isinstance(value, (int, float)):
        return "number"
    if isinstance(value, str):
        return "string"
    return "array" if isinstance(value, list) else "object"

The check does not throw an exception. It returns a list of errors, because the program has something to do with them:

def validate(data, schema):
    errors = []
    for name, rule in schema.items():
        if name not in data:
            errors.append(f'missing "{name}"')
            continue
        value = data[name]
        kind = json_type(value)
        if kind != rule["type"]:
            errors.append(f'"{name}" is {kind}, must be {rule["type"]}')
        elif "enum" in rule and value not in rule["enum"]:
            allowed = ", ".join(rule["enum"])
            errors.append(f'"{name}": "{value}" is not one of {allowed}')
    for key in data:
        if key not in schema:
            errors.append(f'extra field "{key}"')
    return errors

Four answers of the model:

answers = [
    '{"intent": "delivery", "order_id": "1042", "urgent": true, "summary": "The courier did not come, the customer leaves tomorrow."}',
    '{"intent": "shipping", "order_id": "1042", "urgent": "yes", "summary": "The courier did not come."}',
    '{"intent": "return", "urgent": false, "summary": "Wants to return a jacket.", "mood": "calm"}',
    'Sure! Here is the data:\n```json\n{"intent": "delivery", "order_id": "1042", "urgent": true}\n```',
]
for answer in answers:
    data = parse_json(answer)
    errors = ["not JSON"] if data is None else validate(data, TICKET)
    print("\n".join(errors) if errors else "ok")
    print("---")

ok
---
"intent": "shipping" is not one of delivery, return, payment, other
"urgent" is string, must be boolean
---
missing "order_id"
extra field "mood"
---
not JSON
---

Try it

Predict the output

What does the check print for this answer?

answers = [
    '{"intent": "refund", "order_id": 1042, "urgent": false, "summary": "Wants the money back."}',
]
for answer in answers:
    data = parse_json(answer)
    errors = ["not JSON"] if data is None else validate(data, TICKET)
    print("\n".join(errors) if errors else "ok")
    print("---")

Answer

"intent": "refund" is not one of delivery, return, payment, other
---

"intent": "refund" is not one of delivery, return, payment, other
"order_id" is number, must be string
---

ok
---

Try it

Fill in the blank

A line from the check: the type is right, now the value must be one of the allowed ones. What is missing?

elif "enum" in rule and value  rule["enum"]:

The idea

The errors back to the model

What do you do with an answer that does not pass? If you ask the question once more, the model often repeats the same error. It is better to tell it what is wrong.

The model’s answer goes into the history with the role assistant, followed by a new user message with the errors: “The answer does not match the schema: intent… Return only JSON that matches the schema.” The model sees its attempt and the exact errors and fixes them. This is a second attempt with feedback.

Every attempt sends the whole history. The second costs more than the first, and the third more than the second: from the lesson about context you know the API has no memory and everything starts again. That is why the attempts are few, and after them the email goes to a person.

See it

Three emails, three fates

def triage(client, email):
    messages = [{"role": "user", "content": email}]
    for attempt in range(1, MAX_ATTEMPTS + 1):
        reply = client.create(SYSTEM, messages)
        data = parse_json(reply["text"])
        if data is None:
            errors = ["not JSON"]
        else:
            errors = validate(data, TICKET)
        verdict = "; ".join(errors) if errors else "ok"
        tokens_in = reply["input_tokens"]
        print(f"  attempt {attempt} (~{tokens_in} tokens): {verdict}")
        if not errors:
            return Ticket(**data)
        messages.append({"role": "assistant", "content": reply["text"]})
        messages.append({"role": "user",
                         "content": "The answer does not match the schema: " + verdict + ". Return only JSON that matches the schema."})
    return None

def validate(data, schema):
    errors = []
    for name, rule in schema.items():
        if name not in data:
            errors.append(f'missing "{name}"')
            continue
        value = data[name]
        kind = json_type(value)
        if kind != rule["type"]:
            errors.append(f'"{name}" is {kind}, must be {rule["type"]}')
        elif "enum" in rule and value not in rule["enum"]:
            allowed = ", ".join(rule["enum"])
            errors.append(f'"{name}": "{value}" is not one of {allowed}')
    for key in data:
        if key not in schema:
            errors.append(f'extra field "{key}"')
    return errors

Step by step, in words

The email goes to the model with an instruction: return only JSON with four fields.
The answer is valid JSON and is read. That does not mean it is right.
Every field is checked against the schema: is it there, is the type right, is the value one of the allowed ones.
Two errors: intent and urgent. They are a list of strings, not an exception: the program decides what to do.
The errors go back to the model as a new message. Attempt 2 sends the whole history: ~108 tokens instead of ~53.
The second answer passes every check. Only now does it become a Ticket.
The email goes to the delivery queue, marked urgent. At most 3 attempts; after that a person takes over.

The instruction and how many attempts the program has:

SYSTEM = ("Sort the customer email. Return only JSON with the fields "
          "intent (delivery, return, payment or other), order_id (text), "
          "urgent (true or false) and summary (one sentence).")
MAX_ATTEMPTS = 3

One email: ask, read, check, and on an error a new attempt with the errors. When everything passes, the map becomes dataclass Ticket, and from then on the code works with fields, not with text:

def triage(client, email):
    messages = [{"role": "user", "content": email}]
    for attempt in range(1, MAX_ATTEMPTS + 1):
        reply = client.create(SYSTEM, messages)
        data = parse_json(reply["text"])
        if data is None:
            errors = ["not JSON"]
        else:
            errors = validate(data, TICKET)
        verdict = "; ".join(errors) if errors else "ok"
        tokens_in = reply["input_tokens"]
        print(f"  attempt {attempt} (~{tokens_in} tokens): {verdict}")
        if not errors:
            return Ticket(**data)
        messages.append({"role": "assistant", "content": reply["text"]})
        messages.append({"role": "user",
                         "content": "The answer does not match the schema: " + verdict + ". Return only JSON that matches the schema."})
    return None

client = FakeClient()
inbox = [
    "The courier did not come yesterday, order 1042. It is urgent, I leave tomorrow!",
    "I want to return order 1187, the jacket is too big.",
    "hi, something is wrong",
]
for email in inbox:
    print(f"email: {email}")
    ticket = triage(client, email)
    if ticket is None:
        print("  → to a person")
    else:
        urgent = ", urgent" if ticket.urgent else ""
        print(f"  → {ticket.intent}, {ticket.order_id}{urgent}")

email: The courier did not come yesterday, order 1042. It is urgent, I leave tomorrow!
  attempt 1 (~53 tokens): "intent": "shipping" is not one of delivery, return, payment, other; "urgent" is string, must be boolean
  attempt 2 (~108 tokens): ok
  → delivery, 1042, urgent
email: I want to return order 1187, the jacket is too big.
  attempt 1 (~49 tokens): ok
  → return, 1187
email: hi, something is wrong
  attempt 1 (~40 tokens): not JSON
  attempt 2 (~71 tokens): not JSON
  attempt 3 (~102 tokens): not JSON
  → to a person

The email about 1042: the first answer has two errors, the second passes. The second attempt costs ~108 tokens instead of ~53.
The email about 1187: passes the first time.
“hi, something is wrong”: the model finds no topic and answers with a sentence all three times. The program does not invent a category; it gives the email to a person.

The model here is a fake client, as in the lesson about the prompt: the answers are recorded and the same every time. Its rule is in “More detail”.

Try it

Find the bug

A colleague writes a shorter check. The answer {"intent": "return", "urgent": false, "summary": "…"} passes as “ok”, and a minute later the program crashes somewhere far away, when it looks up the order. Which line has the bug?

Pick the line with the bug.

The idea

The schema in the API itself

Newer APIs accept the schema as part of the request. Then the model generates only text that matches it: every next token is chosen among those that keep the JSON by the schema. The polite “Sure!” and "urgent": "yes" become impossible.

The check in the code stays, though:

A refusal or a cut-off answer. If the model refuses (stop_reason is refusal) or reaches max_tokens, the text may not match the schema.
The shop’s rules. The schema says “text”, but it does not know whether order 1187 exists and whether it belongs to this customer. That is a check in the program.
Another model or provider. The code that checks does not depend on who generated the JSON.

Run it at your place. This program is not part of the lesson’s check: you need an account with a model provider, a key and money for every call. The key is in LLM_API_KEY, never in the code. Providers ship official SDKs for Java, C#, Python and JavaScript with the same shape (messages, roles, tokens, tools); the exact class and method names are in the provider’s documentation. The request as JSON: the schema is in the field output_config, and the model name is a placeholder:

{
  "model": "<model name>",
  "max_tokens": 1024,
  "system": "Sort the customer email. Return only JSON with the fields intent (delivery, return, payment or other), order_id (text), urgent (true or false) and summary (one sentence).",
  "messages": [
    {
      "role": "user",
      "content": "The courier did not come yesterday, order 1042. It is urgent, I leave tomorrow!"
    }
  ],
  "output_config": {
    "format": {
      "type": "json_schema",
      "schema": {
        "type": "object",
        "properties": {
          "intent": {
            "type": "string",
            "enum": ["delivery", "return", "payment", "other"]
          },
          "order_id": {"type": "string"},
          "urgent": {"type": "boolean"},
          "summary": {"type": "string"}
        },
        "required": ["intent", "order_id", "urgent", "summary"],
        "additionalProperties": false
      }
    }
  }
}

An example answer. The JSON by the schema is text in the text block, and the program still reads and checks it:

{
  "content": [
    {
      "type": "text",
      "text": "{\"intent\": \"delivery\", \"order_id\": \"1042\", \"urgent\": true, \"summary\": \"The courier did not come, the customer leaves tomorrow.\"}"
    }
  ],
  "stop_reason": "end_turn",
  "usage": {"input_tokens": 53, "output_tokens": 21}
}

Without an SDK, the request goes out with the language’s standard HTTP client. The address is a placeholder:

# request_json is the request JSON above.
# The exact key header is in the provider’s documentation.
import os
import urllib.request

request = urllib.request.Request(
    "https://<provider>/messages",
    data=request_json.encode("utf-8"),
    headers={"Authorization": "Bearer " + os.environ["LLM_API_KEY"],
             "Content-Type": "application/json"},
)
with urllib.request.urlopen(request) as response:
    print(response.read().decode("utf-8"))

Every object in the schema has additionalProperties: false: that way an extra field like mood cannot appear at all.

The idea

When it hurts

JSON for everything. The answer to a customer is text. A schema belongs where the answer goes into code: a queue, a database, another service.
A deep schema. Five levels of nested objects and twenty fields give more errors and more expensive attempts. Two simple requests are better than one complicated one.
Patches instead of a check. “Cut from the first { to the last }” works until the first answer with two {. If the API supports a schema, use it. If not, check and try again.
The right shape is not the right answer. "order_id": "1043" passes the schema, but the model may have got a digit wrong: models invent plausibly. The number is checked against the database before it is used.
Attempts have a price. Every attempt is more tokens and more seconds for the customer. How much exactly, and how to keep a budget, is in the lesson about cost and speed.
The schema and the prompt change together. A new category in the schema without a new category in the prompt means a model that never says it. The schema’s version is in the code, next to the prompt.

More precisely, if you are curious

The words of JSON Schema. The schema in the API is JSON Schema: type (object, string, number, boolean, array, null), properties with the fields, required with the mandatory ones, enum with the allowed values and additionalProperties: false against extra ones. Our check does a subset of that, enough to show the mechanism. A project uses a ready library for JSON Schema.

From text to a type. After the check the map becomes Ticket, and the rest of the code checks nothing more: the compiler or the fields say what is there. That is the idea “read and check at the entrance, and work with types inside”. The check is in one place, not scattered across the program.

Tools use the same. In the next lesson the model asks the program to run a function. The function’s arguments also arrive as JSON by a schema (input_schema) and are also checked.

The fake client inside. Its answers are recorded by the order number in the email: for 1042 the first one is wrong, and after the correction it is right; for 1187 it is right at once; for anything else it is a sentence. The tokens are words × 1.3.

# Stands in for the real model: recorded answers,
# with no network and the same every time.
class FakeClient:
    def create(self, system, messages):
        email = messages[0]["content"]
        corrected = len(messages) > 1
        if "1042" in email:
            text = ('{"intent": "delivery", "order_id": "1042", "urgent": true, "summary": "The courier did not come, the customer leaves tomorrow."}'
                    if corrected else '{"intent": "shipping", "order_id": "1042", "urgent": "yes", "summary": "The courier did not come."}')
        elif "1187" in email:
            text = '{"intent": "return", "order_id": "1187", "urgent": false, "summary": "Wants to return the jacket, it is too big."}'
        else:
            text = "I cannot tell what the customer wants."
        input_tokens = tokens(system) + sum(tokens(m["content"]) for m in messages)
        return {"text": text, "stop_reason": "end_turn",
                "input_tokens": input_tokens, "output_tokens": tokens(text)}


# Roughly: words × 1.3, rounded up.
def tokens(text):
    return (len(text.split()) * 13 + 9) // 10

Exercise

The number is four digits

Solve it in your editor.

Add a rule to the check: order_id is exactly 4 digits. If not, the error is "order_id": "10420" is not 4 digits. Run the check for the three answers that differ only in the number: 1042, 10420 and #1042.

ok
---
"order_id": "10420" is not 4 digits
---
"order_id": "#1042" is not 4 digits
---

More tasks

An empty summary. summary cannot be empty text or only spaces. Add the rule and a test with "summary": " ".
A limited number of tokens. Stop the attempts also when the email’s total tokens pass 100, even before three attempts. How many attempts does the email “hi, something is wrong” make then?

Test

Test yourself

1. The model returned valid JSON. Does that mean the program can use it?Yes: if the JSON can be read, the data is rightYes, if the prompt says “return only JSON”No: it must be checked against the schema, because fields can be missing or have the wrong type and value

2. Why are the errors sent to the model on the second attempt, instead of asking the question once more?That way the model sees exactly what it got wrong and fixes thatBecause the API does not allow the same question twiceBecause it is cheaper

3. The email makes no sense and the model returns a sentence instead of JSON three times. What does the program do?It keeps asking until it gets JSONIt puts the email into the “General” queue with empty fieldsIt stops after the third attempt and gives the email to a person

4. The API now guarantees JSON by the schema. Does the check in the code stay?Yes: the answer can be a refusal or cut off, and the schema does not know the shop’s rulesOnly for Java, because it has no JSON in the standard libraryNo: the schema in the API makes it unnecessary

Structured output: JSON by a schema

Cookies