Schema как контракт для модели: JSON Schema, починка ответа, потоковый разбор

senior~80 мин

открытый урокЭтот раздел читается без входа. Войди, чтобы отмечать прогресс, вести заметки и решать задачи в редакторе. войти

Schema как контракт для модели: JSON Schema, починка ответа, потоковый разбор

Сцена · собеседник, который иногда врёт про формат

Ты просишь языковую модель разобрать текст инцидента и вернуть JSON. В промпте написано: “верни объект с полями title, severity, affectedServices”. В девяти случаях из десяти приходит ровно то, что просили. В десятом приходит вот это:

Конечно! Вот результат:

{"title": "Сервис лежит", "severity": "очень плохо"}

Три беды в одном ответе. Вокруг JSON налипла болтовня, поле severity содержит свободный текст вместо одного из трёх допустимых значений, а affectedServices вообще нет.

Обычная реакция это JSON.parse в try/catch плюс ручные проверки полей. Так делать не надо, потому что ты уже умеешь лучше: у тебя есть схема, которая одновременно описывает контракт, проверяет ответ и, как выяснится через минуту, может объяснить модели, чего от неё хотят.

Идея урока в одном предложении: схема это единственный источник истины, из которого выводятся и запрос к модели, и разбор ответа, и текст замечаний при починке.

Карта урока · что заберёшь домой

Семь разделов:

Раздел 1, Schema.toJsonSchemaDocument, схема превращается в спецификацию для модели.
Раздел 2, аннотации как часть промпта: описания, примеры, перечисления.
Раздел 3, разбор ответа: вытащить JSON из болтовни и декодировать.
Раздел 4, починка: ошибку разбора возвращаем модели человеческим языком.
Раздел 5, потоковый разбор построчного JSON.
Раздел 6, частичные результаты и прогрессивный показ.
Раздел 7, границы: бюджет попыток, метрики, когда сдаваться.

Раздел 1 · Схема как спецификация

Schema.toJsonSchemaDocument(schema) превращает любую схему в JSON Schema, то есть в тот самый формат, который принимают современные API моделей.

import { Schema } from 'effect';

const Severity = Schema.Literals(['info', 'warning', 'critical']);

const Incident = Schema.Struct({
  title: Schema.String,
  severity: Severity,
  affectedServices: Schema.Array(Schema.String),
  rootCause: Schema.NullOr(Schema.String),
});

const document = Schema.toJsonSchemaDocument(Incident);
console.log(JSON.stringify(document.schema, null, 2));

Реальный вывод:

{
  "type": "object",
  "required": ["title", "severity", "affectedServices", "rootCause"],
  "properties": {
    "title": { "type": "string" },
    "severity": { "enum": ["info", "warning", "critical"] },
    "affectedServices": { "type": "array", "items": { "type": "string" } },
    "rootCause": { "anyOf": [{ "type": "string" }, { "type": "null" }] }
  },
  "additionalProperties": false
}

Возвращается не голая спецификация, а документ: у него есть dialect (по умолчанию draft 2020-12), сама schema и пул definitions для вложенных и рекурсивных структур. Разделение полезное, потому что провайдеры хотят разные диалекты, и рядом лежат готовые переводчики: JsonSchema.toDocumentDraft07(document) и JsonSchema.toMultiDocumentOpenApi3_1(...).

Обрати внимание на три вещи, которые получились сами.

Schema.Literals([...]) превратился в enum. Модель, которой отдали такую спецификацию, не имеет права придумать очень плохо.

additionalProperties: false запрещает лишние поля, и это поведение по умолчанию. Важно: без запрета модель охотно добавляет confidence, notes и summary, потому что ей так удобнее. Если лишние поля нужны, разрешай их явно опцией { additionalProperties: true }.

required перечисляет все поля. Если какое-то поле действительно необязательное, это надо сказать схемой (Schema.optional из 21 · Schema, второй заход), а не надеяться, что модель догадается.

Итого: одна схема, два применения. Ту же самую Incident ты отдаёшь в запрос как спецификацию и ей же декодируешь ответ. Рассинхронизироваться они физически не могут.

Раздел 2 · Аннотации это часть промпта

JSON Schema умеет нести описания, и модели их читают. Значит, документация полей переезжает из промпта в схему:

const Severity = Schema.Literals(['info', 'warning', 'critical']).annotate({
  description: 'Насколько серьёзен инцидент',
});

const Incident = Schema.Struct({
  title: Schema.String.annotate({
    description: 'Краткий заголовок инцидента, не длиннее 80 символов',
    examples: ['Сервис оплаты отвечает 500'],
  }),
  severity: Severity,
  affectedServices: Schema.Array(Schema.String).annotate({
    description: 'Список затронутых сервисов по именам',
  }),
  rootCause: Schema.NullOr(Schema.String).annotate({
    description: 'Причина, если её удалось определить, иначе null',
  }),
}).annotate({
  identifier: 'Incident',
  description: 'Структурированный разбор инцидента',
});

Метод называется annotate, в единственном числе: в v4 у аннотаций один вход вместо пары annotations и annotationsFromSelf.

Всё это доезжает до спецификации без потерь:

"title": {
  "type": "string",
  "description": "Краткий заголовок инцидента, не длиннее 80 символов",
  "examples": ["Сервис оплаты отвечает 500"]
}

Практические выводы.

identifier задаёт имя структуры в пуле definitions. Без него там будет анонимная запись, и читать спецификацию (а её будет читать и человек, и модель) станет неудобно.

description пиши как инструкцию, а не как комментарий для коллеги. Не “заголовок”, а “краткий заголовок инцидента, не длиннее 80 символов”.

examples работают лучше, чем длинные объяснения. Один правдоподобный пример экономит три предложения описания.

И главное следствие: когда правило живёт в схеме, оно автоматически попадает и в спецификацию для модели, и в проверку ответа. Промпт и валидация больше не расходятся.

E22.1★

Возьми схему, которую ты уже используешь в проекте, добавь identifier, описания и по одному примеру на поле, и сгенерируй Schema.toJsonSchemaDocument. Прочитай результат глазами: понятно ли из него, что заполнять, если ты не видел исходного кода.

Раздел 3 · Разбор ответа

Даже с жёсткой спецификацией ответ приходит строкой, и её надо превратить в значение домена.

import { Effect, Schema } from 'effect';

const decodeIncident = Schema.decodeUnknownEffect(Schema.fromJsonString(Incident), {
  errors: 'all',
});

Две детали, обе важные.

Schema.fromJsonString(Incident) разбирает строку и сразу проверяет структуру. Отдельного JSON.parse в try/catch не нужно: битый JSON даст ту же SchemaError, что и неверная форма.

{ errors: 'all' } собирает все расхождения. Это не косметика: в Разделе 4 мы отправим их модели, и одно замечание вместо трёх означает три круга починки вместо одного.

Отдельно про болтовню вокруг JSON. Если API умеет гарантировать формат ответа (а современные это умеют), проблема не возникает. Если работаешь через свободный текст, нужен маленький хелпер:

const extractJson = (raw: string): string => {
  const start = raw.indexOf('{');
  const end = raw.lastIndexOf('}');
  return start >= 0 && end > start ? raw.slice(start, end + 1) : raw;
};

Он нарочно тупой: берёт кусок от первой фигурной скобки до последней. Умнее делать не надо, потому что правильное решение это не улучшать эвристику, а включить гарантированный формат на стороне API.

Раздел 4 · Починка: верни модели её же ошибки

Самое ценное в этом уроке. Если ответ не прошёл схему, у тебя на руках SchemaError, а из неё форматтером получается список конкретных замечаний. Их можно отдать модели дословно.

import { Context, Effect, Schema, SchemaIssue } from 'effect';

class Model extends Context.Service<
  Model,
  { readonly complete: (prompt: string) => Effect.Effect<string> }
>()('Model') {}

const format = SchemaIssue.makeFormatterStandardSchemaV1();

const askWithRepair = (
  prompt: string,
  attemptsLeft: number,
): Effect.Effect<typeof Incident.Type, Schema.SchemaError, Model> =>
  Effect.gen(function* () {
    const model = yield* Model;
    const raw = yield* model.complete(prompt);

    return yield* decodeIncident(extractJson(raw)).pipe(
      Effect.catchTag('SchemaError', (error) => {
        if (attemptsLeft <= 0) return Effect.fail(error);

        const issues = format(error.issue)
          .issues.map((issue) => `поле ${issue.path.join('.')}: ${issue.message}`)
          .join('; ');

        const repaired = `${prompt}\n\nПредыдущий ответ не прошёл проверку. Исправь: ${issues}. Верни только JSON.`;

        return askWithRepair(repaired, attemptsLeft - 1);
      }),
    );
  });

Живой прогон на подменённой модели, которая с первого раза отвечает "severity": "очень плохо":

после починки с обратной связью = {"title":"Сервис лежит","severity":"critical"}

Разберём, почему это работает лучше голого повтора.

Замечание конкретное. Не “попробуй ещё раз”, а “поле severity: Expected info, actual очень плохо”. Модель понимает, что именно править.

Замечания приходят все сразу, потому что мы включили errors: 'all'. Один круг вместо трёх.

Текст замечаний сгенерирован из схемы, а не написан руками. Поменяешь схему, замечания поменяются сами.

Ошибка остаётся типизированной. Если попытки кончились, наружу уходит честная SchemaError, а не выдуманное исключение. Вызывающий решает, что делать: отдать пользователю заглушку, записать в очередь на ручной разбор, упасть.

Сравни с обычным Effect.retry: он повторит запрос с тем же промптом, и модель с приличной вероятностью повторит ту же ошибку. Ретрай без обратной связи лечит флаки, а не непонимание.

E22.2★★

Добавь в цикл починки счётчик и залогируй, на какой попытке ответ прошёл. Прогони десяток запросов и посмотри распределение: если больше половины требует починки, проблема не в модели, а в схеме или описаниях.

Раздел 5 · Потоковый разбор

Ждать целиком ответ, в котором сто позиций, скучно. Приём простой и надёжный: попросить модель отдавать по одному JSON-объекту на строку, а поток разбирать построчно.

import { Effect, Schema, Stream } from 'effect';

const parseStream = (chunks: Stream.Stream<string>) =>
  chunks.pipe(
    Stream.splitLines,
    Stream.mapEffect(Schema.decodeUnknownEffect(Schema.fromJsonString(Incident))),
  );

Проверим на потоке, где объект разорван между чанками (именно так и приходят токены):

const fakeStream = Stream.make(
  '{"title":"первый","severity":"info"}\n{"title":"втор',
  'ой","severity":"warning"}\n',
);

// [{"title":"первый","severity":"info"},{"title":"второй","severity":"warning"}]

Stream.splitLines сам склеивает границы чанков и отдаёт целые строки. Это ровно то, чего обычно не хватает при ручной сборке буфера: разрыв посреди слова второй не сломал ничего.

Почему построчный JSON, а не один большой массив. Массив можно разобрать только целиком, и первый элемент ты увидишь в конце. Построчный формат даёт инкрементальный результат: первый инцидент показан пользователю, пока модель ещё пишет второй. Плюс схема проверяет каждый элемент по отдельности, и одна битая строка не отменяет остальные.

Теперь про битую строку, и тут есть ловушка. Первое, что приходит в голову, это Stream.result:

// так поток остановится на первой же битой строке
const wrong = parseStream(chunks).pipe(Stream.result);

Stream.result действительно переводит ошибку в значение (Result.failure вместо падения), но поток на этом заканчивается: на входе из трёх строк, где вторая битая, до конца доедет ровно одна. Ошибка в потоке это терминальное событие, и result его не отменяет, он только меняет форму.

Чтобы пропускать битые элементы и продолжать, ошибку надо гасить внутри обработки элемента, до того как она станет ошибкой потока:

import { Effect, Option, Schema, Stream } from 'effect';

const decodeLine = Schema.decodeUnknownEffect(Schema.fromJsonString(Incident));

const resilient = chunks.pipe(
  Stream.splitLines,
  Stream.mapEffect((line) =>
    decodeLine(line).pipe(
      Effect.map(Option.some),
      Effect.catch(() =>
        Effect.logWarning(`строка не прошла схему: ${line}`).pipe(Effect.as(Option.none())),
      ),
    ),
  ),
  Stream.filterMap((option) => option),
);

Теперь до конца доезжают обе валидные строки, а битая уходит в лог. Правило, которое стоит запомнить шире этого урока: решение “продолжать или падать” принимается на уровне элемента, а не на уровне потока.

Раздел 6 · Частичные результаты

Иногда показывать хочется до того, как объект дописан. Значит, нужна копия схемы, в которой все поля необязательные. Отдельного Schema.partial в v4 нет, потому что он не нужен: это обычное преобразование полей из 21 · Schema, второй заход.

import { Schema, Struct } from 'effect';

const PartialIncident = Incident.mapFields(Struct.map(Schema.optional));

const draft = yield* Schema.decodeUnknownEffect(PartialIncident)({ title: 'ещё пишу' });
// { title: 'ещё пишу' }

Читается прямо: “к каждому полю применить Schema.optional”. Если нужен точный вариант, где ключа может просто не быть, подставь Schema.optionalKey. А если необязательными должны стать не все поля, а два конкретных, есть Struct.mapPick(['title', 'severity'], Schema.optional).

Правило же простое и жёсткое: частичная схема живёт только в слое отображения. Ниже, в домен, попадает результат полной схемы. Иначе весь выигрыш parse-don't-validate испарится: по коду поедут title?: string, и каждый обработчик обзаведётся проверкой на undefined.

Практическая схема выглядит так:

поток кусков разбирается частичной схемой и рисуется в интерфейсе с пометкой “черновик”;
по завершении ответ целиком разбирается полной схемой;
только этот результат уходит в бизнес-логику и в хранилище.

Раздел 7 · Границы и метрики

Цикл починки без ограничителя это способ потратить бюджет за ночь. Ограничивай двумя способами сразу:

const guarded = askWithRepair(prompt, 2).pipe(
  Effect.timeout('30 seconds'),
  Effect.trackDuration(llmDuration),
  Effect.tapCause((cause) => Effect.logError('модель не уложилась в контракт', cause)),
);

По числу попыток и по времени, потому что это разные отказы: первое ловит модель, которая упорно отвечает не по схеме, второе ловит зависший запрос.

И обязательно измеряй. Метрики из 18 · Observability здесь дают прямой сигнал качества промпта:

const firstTry = Metric.counter('llm_schema_ok_first_try', { incremental: true });
const repaired = Metric.counter('llm_schema_repaired', { incremental: true });
const failed = Metric.counter('llm_schema_failed', { incremental: true });

Как читать эти три числа:

растёт repaired, значит описания полей недостаточно понятные. Чини схему, а не модель;
растёт failed, значит контракт слишком жёсткий для задачи. Возможно, поле стоит сделать NullOr вместо обязательного;
всё в firstTry, значит контракт хороший, и цикл починки можно оставить как страховку, не думая о нём.

Что сказать про сам API

Схема даёт спецификацию, дальше её нужно передать модели. Механизм зависит от провайдера, но идея одна: JSON Schema уходит в запрос, а API гарантирует форму ответа. У Anthropic это output_config.format в Messages API либо инструмент со строгой схемой (strict: true); модель в примерах ниже это claude-opus-5.

Оборачивать вызов имеет смысл в сервис из 04 · Services и Layer, с HttpClient из 19 · HttpClient и обвязка API внутри:

import { Context, Effect, Layer, Schema } from 'effect';
import { FetchHttpClient, HttpClient, HttpClientResponse } from 'effect/unstable/http';

class Model extends Context.Service<Model, ModelShape>()('Model') {
  static readonly layer = Layer.effect(
    Model,
    Effect.gen(function* () {
      const client = yield* HttpClient.HttpClient;

      const structured = <A, I>(schema: Schema.Codec<A, I>, prompt: string) =>
        Effect.gen(function* () {
          const spec = Schema.toJsonSchemaDocument(schema);
          const body = yield* buildRequestBody({ prompt, spec }); // форма зависит от провайдера
          const response = yield* client.post('/v1/messages', { body });
          const envelope = yield* HttpClientResponse.schemaBodyJson(ModelEnvelope)(response);
          return yield* Schema.decodeUnknownEffect(Schema.fromJsonString(schema), {
            errors: 'all',
          })(envelope.text);
        });

      return Model.of({ structured });
    }),
  ).pipe(Layer.provide(FetchHttpClient.layer));
}

Обрати внимание на сигнатуру structured: она принимает любую схему и возвращает значение её типа. Один метод на все структурированные запросы, никакого дублирования на каждый новый тип ответа. И, что важнее, в тестах этот сервис подменяется слоем, как в 19, поэтому весь цикл починки тестируется без единого сетевого вызова и без единого потраченного токена.

E22.3★★

Собери тестовую реализацию Model, которая на первый запрос отдаёт мусор, на второй валидный ответ, и напиши тест: результат корректный, попыток ровно две, счётчик llm_schema_repaired увеличился на единицу.

Финал · чек-лист

Чек-листготово

CHK-1

Генерируешь спецификацию для модели из той же схемы, которой потом разбираешь ответ, а не пишешь её руками в промпте.

CHK-2

Держишь описания и примеры полей в аннотациях схемы, зная, что они доедут и до спецификации, и до сообщений об ошибках.

CHK-3

Разбираешь ответ через Schema.fromJsonString с errors: 'all', а не JSON.parse в try/catch.

CHK-4

При невалидном ответе возвращаешь модели конкретные замечания из форматтера SchemaIssue, а не повторяешь тот же промпт.

CHK-5

Понимаешь разницу: ретрай лечит флаки, обратная связь лечит непонимание контракта.

CHK-6

Просишь построчный JSON для длинных ответов и разбираешь его через Stream.splitLines, не собирая буфер руками.

CHK-7

Держишь частичную схему (mapFields со Struct.map(Schema.optional)) только в слое отображения, а в домен пускаешь результат полной схемы.

CHK-8

Ограничиваешь цикл починки и по числу попыток, и по времени, и меришь долю ответов, прошедших с первого раза.

ДЗ

Задания делаются в отдельном репозитории llm-schema-<nick> или в ветке Pulse, на выбор. Тег PR: lesson-22.

HW-EFF-22-JSONSCHEMA★★

Опиши схемой структуру разбора инцидента (минимум пять полей, включая перечисление, массив и допускающее null поле). Добавь identifier, описания и примеры. Сгенерируй Schema.toJsonSchemaDocument и приложи вывод к PR. Проверь тестом, что в спецификации есть enum для поля с перечислением и additionalProperties: false.

HW-EFF-22-REPAIR★★★

Реализуй askWithRepair из Раздела 4. Тестовая модель: первый ответ содержит две ошибки одновременно (неверное значение перечисления и отсутствующее поле), второй ответ корректный. Проверь, что в промпт второй попытки попали оба замечания, и что при исчерпании попыток наружу уходит SchemaError, а не что-то другое.

HW-EFF-22-STREAM★★★

Собери разбор построчного JSON из Раздела 5. Тест: поток из трёх чанков, где один объект разорван между чанками, а одна строка заведомо битая. Ожидаемый результат: валидные объекты разобраны, битая строка залогирована и пропущена, поток завершился нормально.

HW-EFF-22-PARTIAL★★★

Сделай прогрессивный показ: частичная схема для промежуточного состояния, полная для финального. Напиши тест, показывающий, что функция бизнес-слоя принимает только полный тип (достаточно проверки компиляции: попытка передать частичное значение не должна собираться). В README.md объясни двумя предложениями, почему граница проведена именно здесь.

HW-EFF-22-BUDGET★★★

Добавь три счётчика из Раздела 7 и ограничение по времени. Прогони сто запросов на подменённой модели, которая отвечает неверно с вероятностью, зависящей от номера попытки, и приложи к PR получившееся распределение. Одним абзацем: что бы ты поменял в схеме, если бы доля починок оказалась выше трети.

Дальше

23 · Sink и продвинутый Stream. Разбор потока мы сделали, а вот куда его сложить (файл, база, очередь, с ретраями и запасным приёмником) это следующий урок.

Полезно перечитать:

21 · Schema, второй заход, откуда взялись форматтер SchemaIssue и errors: 'all', на которых держится вся починка.
10 · Batching, если запросов к модели много: RequestResolver и кеш с TTL экономят и время, и деньги.