Структуриран изход: JSON по схема в Python

Проблем

Кодът, който чака JSON

Пощата на поддръжката получава стотици писма на ден. Програма ги разпределя по опашки: доставки, връщания, плащания, общи. Тя пита модел за темата на писмото и чака JSON с полето intent. Ето три отговора на модела за едно и също писмо:

answers = [
    'Разбира се! Ето данните:\n```json\n{"intent": "delivery", "order_id": "1042", "urgent": true}\n```',
    '{"intent": "доставка", "order_id": "1042", "urgent": "да", "summary": "Куриерът не е дошъл."}',
    '{"intent": "delivery", "order_id": "1042", "urgent": true, "summary": "Куриерът не е дошъл, клиентът заминава утре."}',
]
for answer in answers:
    data = parse_json(answer)
    if data is None:
        print(f"не е JSON: {answer.splitlines()[0]}")
        continue
    queue = QUEUES.get(data["intent"])
    if queue is None:
        print(f'няма опашка за "{data["intent"]}"')
    else:
        print(f"опашка: {queue}")

не е JSON: Разбира се! Ето данните:
няма опашка за "доставка"
опашка: Доставки

Само третият става. Първият е учтив, но е текст около JSON и програмата не го чете. Вторият е JSON, но темата е на български и няма такава опашка. В урока за промпта видя, че примерите помагат. Но програмата трябва да работи и в нощта, в която моделът реши да е учтив.

Идеята

Схема, проверка, втори опит

Когато отговорът отива в код, а не при човек, той има договор. Договорът е схема: кои полета има, какъв е типът на всяко и кои стойности са разрешени.

Промптът казва схемата. Полетата, типовете и разрешените стойности са в инструкцията.
Програмата чете JSON-а. Ако не е JSON, това е грешка, а не срив.
Програмата проверява по схемата: има ли всяко поле, верен ли е типът, от разрешените ли е стойността, няма ли излишни полета.
Грешките отиват обратно към модела като ново съобщение, и той опитва пак.
Опитите са най-много три. После писмото отива при човек.

Отговор по схема, който програмата проверява, е структуриран изход (structured output).

Изходът на модела е недоверен вход като всичко, което идва отвън: програмата не вярва, че е по договора, а го проверява.

Виж го

Четене и проверка

Схемата на писмото: четири полета, intent с четири разрешени стойности:

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

Четенето връща обекта или None, ако текстът не е точно един JSON обект:

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

Модулът json е в стандартната библиотека. json.loads хвърля JSONDecodeError за текст, който не е JSON.

Типът на стойността в думите на JSON:

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"

Проверката не хвърля изключение. Тя връща списък с грешките, защото програмата има какво да прави с тях:

def validate(data, schema):
    errors = []
    for name, rule in schema.items():
        if name not in data:
            errors.append(f'липсва "{name}"')
            continue
        value = data[name]
        kind = json_type(value)
        if kind != rule["type"]:
            errors.append(f'"{name}" е {kind}, а трябва да е {rule["type"]}')
        elif "enum" in rule and value not in rule["enum"]:
            allowed = ", ".join(rule["enum"])
            errors.append(f'"{name}": "{value}" не е от {allowed}')
    for key in data:
        if key not in schema:
            errors.append(f'излишно поле "{key}"')
    return errors

Четири отговора на модела:

answers = [
    '{"intent": "delivery", "order_id": "1042", "urgent": true, "summary": "Куриерът не е дошъл, клиентът заминава утре."}',
    '{"intent": "доставка", "order_id": "1042", "urgent": "да", "summary": "Куриерът не е дошъл."}',
    '{"intent": "return", "urgent": false, "summary": "Иска да върне яке.", "mood": "спокоен"}',
    'Разбира се! Ето данните:\n```json\n{"intent": "delivery", "order_id": "1042", "urgent": true}\n```',
]
for answer in answers:
    data = parse_json(answer)
    errors = ["не е JSON"] if data is None else validate(data, TICKET)
    print("\n".join(errors) if errors else "ок")
    print("---")

ок
---
"intent": "доставка" не е от delivery, return, payment, other
"urgent" е string, а трябва да е boolean
---
липсва "order_id"
излишно поле "mood"
---
не е JSON
---

Пробвай

Предвиди изхода

Какво печата проверката за този отговор?

answers = [
    '{"intent": "refund", "order_id": 1042, "urgent": false, "summary": "Иска парите си обратно."}',
]
for answer in answers:
    data = parse_json(answer)
    errors = ["не е JSON"] if data is None else validate(data, TICKET)
    print("\n".join(errors) if errors else "ок")
    print("---")

Отговор

"intent": "refund" не е от delivery, return, payment, other
"order_id" е number, а трябва да е string
---

ок
---

"intent": "refund" не е от delivery, return, payment, other
---

Пробвай

Попълни

Ред от проверката: типът е верен, остава стойността да е от разрешените. Какво липсва?

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

Идеята

Грешките обратно към модела

Какво правиш с отговор, който не минава? Ако зададеш въпроса още веднъж, моделът често повтаря същата грешка. По-добре е да му кажеш какво не е наред.

Отговорът на модела влиза в историята с роля assistant, а след него ново съобщение user с грешките: „Отговорът не е по схемата: intent… Върни само JSON по схемата.“ Моделът вижда своя опит и точните грешки и поправя тях. Това е втори опит с обратна връзка.

Всеки опит праща цялата история. Вторият е по-скъп от първия, а третият от втория: от урока за контекста знаеш, че API-то няма памет и всичко тръгва наново. Затова опитите са малко, а след тях писмото отива при човек.

Виж го

Три писма, три съдби

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 = ["не е JSON"]
        else:
            errors = validate(data, TICKET)
        verdict = "; ".join(errors) if errors else "ок"
        tokens_in = reply["input_tokens"]
        print(f"  опит {attempt} (~{tokens_in} токена): {verdict}")
        if not errors:
            return Ticket(**data)
        messages.append({"role": "assistant", "content": reply["text"]})
        messages.append({"role": "user",
                         "content": "Отговорът не е по схемата: " + verdict + ". Върни само JSON по схемата."})
    return None

def validate(data, schema):
    errors = []
    for name, rule in schema.items():
        if name not in data:
            errors.append(f'липсва "{name}"')
            continue
        value = data[name]
        kind = json_type(value)
        if kind != rule["type"]:
            errors.append(f'"{name}" е {kind}, а трябва да е {rule["type"]}')
        elif "enum" in rule and value not in rule["enum"]:
            allowed = ", ".join(rule["enum"])
            errors.append(f'"{name}": "{value}" не е от {allowed}')
    for key in data:
        if key not in schema:
            errors.append(f'излишно поле "{key}"')
    return errors

Стъпка по стъпка, с думи

Писмото отива при модела с инструкция: върни само JSON с четири полета.
Отговорът е валиден JSON и се прочита. Това още не значи, че е верен.
Всяко поле се сверява със схемата: има ли го, верен ли е типът, от разрешените стойности ли е.
Две грешки: intent и urgent. Те са списък от текстове, не изключение: програмата решава какво да прави.
Грешките отиват обратно към модела като ново съобщение. Опит 2 праща цялата история: ~97 токена вместо ~46.
Вторият отговор минава всички проверки. Едва сега става Ticket.
Писмото отива в опашката за доставки като спешно. Най-много 3 опита, после поема човек.

Инструкцията и колко опита има програмата:

SYSTEM = ("Разпредели писмото от клиента. Върни само JSON с полетата "
          "intent (delivery, return, payment или other), order_id (текст), "
          "urgent (true или false) и summary (едно изречение).")
MAX_ATTEMPTS = 3

Едно писмо: питане, четене, проверка и при грешка нов опит с грешките. Когато всичко мине, речникът става dataclass Ticket, и оттук нататък кодът работи с полета, не с текст:

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 = ["не е JSON"]
        else:
            errors = validate(data, TICKET)
        verdict = "; ".join(errors) if errors else "ок"
        tokens_in = reply["input_tokens"]
        print(f"  опит {attempt} (~{tokens_in} токена): {verdict}")
        if not errors:
            return Ticket(**data)
        messages.append({"role": "assistant", "content": reply["text"]})
        messages.append({"role": "user",
                         "content": "Отговорът не е по схемата: " + verdict + ". Върни само JSON по схемата."})
    return None

client = FakeClient()
inbox = [
    "Куриерът не дойде вчера, поръчка 1042. Спешно е, заминавам утре!",
    "Искам да върна поръчка 1187, якето ми е голямо.",
    "здр, нещо не става",
]
for email in inbox:
    print(f"писмо: {email}")
    ticket = triage(client, email)
    if ticket is None:
        print("  → към човек")
    else:
        urgent = ", спешно" if ticket.urgent else ""
        print(f"  → {ticket.intent}, {ticket.order_id}{urgent}")

писмо: Куриерът не дойде вчера, поръчка 1042. Спешно е, заминавам утре!
  опит 1 (~46 токена): "intent": "доставка" не е от delivery, return, payment, other; "urgent" е string, а трябва да е boolean
  опит 2 (~97 токена): ок
  → delivery, 1042, спешно
писмо: Искам да върна поръчка 1187, якето ми е голямо.
  опит 1 (~45 токена): ок
  → return, 1187
писмо: здр, нещо не става
  опит 1 (~39 токена): не е JSON
  опит 2 (~66 токена): не е JSON
  опит 3 (~93 токена): не е JSON
  → към човек

Писмото за 1042: първият отговор има две грешки, вторият минава. Вторият опит струва ~97 токена вместо ~46.
Писмото за 1187: минава от първия път.
„здр, нещо не става“: моделът не намира тема и трите пъти отговаря с изречение. Програмата не измисля категория, а дава писмото на човек.

Моделът тук е фалшив клиент като в урока за промпта: отговорите са записани и всеки път са същите. Неговото правило е в „По-точно“.

Пробвай

Намери грешката

Колега пише по-кратка проверка. Отговорът {"intent": "return", "urgent": false, "summary": "…"} минава като „ок“, а след минута програмата пада някъде далеч, при търсенето на поръчката. На кой ред е грешката?

Избери реда с грешката.

Идеята

Схемата в самото API

Новите API-та приемат схемата като част от заявката. Тогава моделът генерира само текст, който отговаря на нея: всеки следващ токен се избира измежду тези, които пазят JSON-а по схемата. Учтивото „Разбира се!“ и "urgent": "да" стават невъзможни.

Проверката в кода обаче остава:

Отказ или отрязан отговор. Ако моделът откаже (stop_reason е refusal) или стигне max_tokens, текстът може да не е по схемата.
Правилата на магазина. Схемата казва „текст“, но не знае дали поръчка 1187 съществува и дали е на този клиент. Това е проверка в програмата.
Друг модел или доставчик. Кодът, който проверява, не зависи от това кой е генерирал JSON-а.

Пусни при теб. Тази програма не е в проверката на урока: трябва ти профил при доставчик на модели, ключ и пари за всяко извикване. Ключът е в LLM_API_KEY, никога в кода. Доставчиците имат официални SDK-та за Java, C#, Python и JavaScript със същата форма (съобщения, роли, токени, инструменти); точните имена на класовете и методите са в документацията на доставчика. Заявката като JSON: схемата е в полето output_config, а името на модела е заместител:

{
  "model": "<името на модела>",
  "max_tokens": 1024,
  "system": "Разпредели писмото от клиента. Върни само JSON с полетата intent (delivery, return, payment или other), order_id (текст), urgent (true или false) и summary (едно изречение).",
  "messages": [
    {
      "role": "user",
      "content": "Куриерът не дойде вчера, поръчка 1042. Спешно е, заминавам утре!"
    }
  ],
  "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
      }
    }
  }
}

Примерен отговор. JSON-ът по схемата е текст в текстовия блок и програмата пак го чете и проверява:

{
  "content": [
    {
      "type": "text",
      "text": "{\"intent\": \"delivery\", \"order_id\": \"1042\", \"urgent\": true, \"summary\": \"Куриерът не е дошъл, клиентът заминава утре.\"}"
    }
  ],
  "stop_reason": "end_turn",
  "usage": {"input_tokens": 46, "output_tokens": 19}
}

Без SDK заявката се праща със стандартния HTTP клиент на езика. Адресът е заместител:

# request_json е JSON-ът на заявката отгоре.
# Точната заглавка за ключа е в документацията на доставчика.
import os
import urllib.request

request = urllib.request.Request(
    "https://<доставчик>/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"))

Всеки обект в схемата има additionalProperties: false: така излишно поле като mood изобщо не може да се появи.

Идеята

Кога вреди

JSON за всичко. Отговорът към клиента е текст. Схема трябва там, където отговорът отива в код: опашка, база, друг сервиз.
Дълбока схема. Пет нива вложени обекти и двайсет полета дават повече грешки и по-скъпи опити. По-добре две прости заявки, отколкото една сложна.
Кръпки вместо проверка. „Изрежи от първата { до последната }“ работи до първия отговор с две {. Ако API-то поддържа схема, ползвай нея. Ако не, проверявай и опитвай пак.
Вярната форма не е верен отговор. "order_id": "1043" минава схемата, но моделът може да е сбъркал цифра: моделите измислят правдоподобно. Номерът се сверява с базата, преди да се ползва.
Опитите имат цена. Всеки опит е още токени и още секунди за клиента. Колко точно струва това и как се пази бюджет, е в урока за цената и бързината.
Схемата и промптът се менят заедно. Нова категория в схемата без нова категория в промпта значи модел, който никога не я казва. Версията на схемата е в кода, до промпта.

По-точно, за любопитните

Думите на JSON Schema. Схемата в API-то е JSON Schema: type (object, string, number, boolean, array, null), properties с полетата, required със задължителните, enum с разрешените стойности и additionalProperties: false срещу излишните. Нашата проверка прави подмножество от това, колкото да се види механизмът. В проект се ползва готова библиотека за JSON Schema.

От текст към тип. След проверката речникът става Ticket и останалият код не проверява нищо повече: компилаторът или полетата казват какво има. Това е идеята „прочети и провери на входа, а навътре работи с типове“. Проверката е на едно място, а не разпиляна из програмата.

Инструментите ползват същото. В следващия урок моделът иска програмата да изпълни функция. Аргументите на функцията също идват като JSON по схема (input_schema) и също се проверяват.

Фалшивият клиент отвътре. Отговорите му са записани по номера на поръчката в писмото: за 1042 първият е грешен, а след поправка е верен; за 1187 е верен веднага; за всичко друго е изречение. Токените са думите × 1,3.

# Стои на мястото на истинския модел: записани отговори,
# без мрежа и всеки път едни и същи.
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": "Куриерът не е дошъл, клиентът заминава утре."}'
                    if corrected else '{"intent": "доставка", "order_id": "1042", "urgent": "да", "summary": "Куриерът не е дошъл."}')
        elif "1187" in email:
            text = '{"intent": "return", "order_id": "1187", "urgent": false, "summary": "Иска да върне якето, голямо е."}'
        else:
            text = "Не мога да разбера какво иска клиентът."
        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)}


# Грубо: думите × 1,3, закръглено нагоре.
def tokens(text):
    return (len(text.split()) * 13 + 9) // 10

Упражнение

Номерът е четири цифри

Реши задачата в редактора си.

Добави в проверката правило: order_id е точно 4 цифри. Ако не е, грешката е "order_id": "10420" не е 4 цифри. Пусни проверката за трите отговора, които се различават само по номера: 1042, 10420 и №1042.

ок
---
"order_id": "10420" не е 4 цифри
---
"order_id": "№1042" не е 4 цифри
---

Още задачи

Празно резюме. summary не може да е празен текст или само интервали. Добави правилото и проба с "summary": " ".
Ограничен брой токени. Спри опитите и когато общите токени на писмото минат 100, дори да не са минали три опита. Колко опита прави тогава писмото „здр, нещо не става“?

Тест

Провери се

1. Моделът върна валиден JSON. Значи ли това, че програмата може да го ползва?Да: щом JSON-ът се чете, данните са верниНе: трябва да се провери по схемата, защото полета може да липсват или да са с грешен тип и стойностДа, ако в промпта пише „върни само JSON“

2. Защо при втория опит грешките се пращат на модела, вместо въпросът да се зададе още веднъж?Така моделът вижда какво точно е сгрешил и поправя негоЗащото API-то не позволява един и същ въпрос два пътиЗащото така е по-евтино

3. Писмото е неразбираемо и моделът три пъти връща изречение вместо JSON. Какво прави програмата?Спира след третия опит и дава писмото на човекСлага писмото в опашката „Общи“ с празни полетаПита, докато получи JSON

4. API-то вече гарантира JSON по схемата. Остава ли проверката в кода?Само за Java, защото няма JSON в стандартната библиотекаДа: отговорът може да е отказ или да е отрязан, а и схемата не знае правилата на магазинаНе: схемата в API-то я прави излишна

Структуриран изход: JSON по схема

Бисквитки