Problem 117 · LLM Systems

JSON-Schema-Lite Validator with a Repair Retry Loop

MediumAI engineering

01Problem

Write `validate(value, schema, path="")` supporting `type` (object, array, string, integer, number, boolean, null), `properties`, `required`, `additionalProperties: false`, `items`, `enum`, `minimum`/`maximum`, `minLength`/`maxLength`. Return a list of errors like `"/items/1/qty: expected integer"`. Then `generate_json(llm, schema, messages, max_attempts)` calls `llm(messages) -> str`, strips a Markdown code fence if present, parses and validates, and on failure appends the bad output and a user message listing the errors before retrying. Return `(obj, attempts)` or raise after `max_attempts`.

02Key insight

Even with native structured output you validate, because semantic constraints (ranges, enums, cross-field rules) are not all enforced by the decoder and providers differ in what they guarantee. In Python `True` is an `int`, so a naive `isinstance` check accepts `true` for an integer field; type checks must exclude bool explicitly. The retry loop is only useful if it feeds back **specific** errors with JSON-pointer paths; resending the same prompt mostly reproduces the same failure. Cap attempts and surface the last errors, because unbounded repair loops are a cost leak.

03Solution

solution.pypython71 lines

import json
import re
from typing import Any, Callable

FENCE = re.compile(r"^\s*" + chr(96) * 3 + r"(?:json)?\s*(.*?)\s*" + chr(96) * 3 + r"\s*$", re.S)

_TYPES: dict[str, Callable[[Any], bool]] = {
    "object": lambda v: isinstance(v, dict),
    "array": lambda v: isinstance(v, list),
    "string": lambda v: isinstance(v, str),
    "integer": lambda v: isinstance(v, int) and not isinstance(v, bool),
    "number": lambda v: isinstance(v, (int, float)) and not isinstance(v, bool),
    "boolean": lambda v: isinstance(v, bool),
    "null": lambda v: v is None,
}


def validate(value: Any, schema: dict, path: str = "") -> list[str]:
    where = path or "/"
    t = schema.get("type")
    if t and not _TYPES[t](value):
        return [f"{where}: expected {t}"]
    errors: list[str] = []
    if "enum" in schema and value not in schema["enum"]:
        errors.append(f"{where}: must be one of {schema['enum']}")
    if isinstance(value, (int, float)) and not isinstance(value, bool):
        if "minimum" in schema and value < schema["minimum"]:
            errors.append(f"{where}: below minimum {schema['minimum']}")
        if "maximum" in schema and value > schema["maximum"]:
            errors.append(f"{where}: above maximum {schema['maximum']}")
    if isinstance(value, str):
        if len(value) < schema.get("minLength", 0):
            errors.append(f"{where}: shorter than {schema['minLength']}")
        if "maxLength" in schema and len(value) > schema["maxLength"]:
            errors.append(f"{where}: longer than {schema['maxLength']}")
    if isinstance(value, dict):
        props = schema.get("properties", {})
        for key in schema.get("required", []):
            if key not in value:
                errors.append(f"{path}/{key}: required")
        for key, sub in value.items():
            if key in props:
                errors.extend(validate(sub, props[key], f"{path}/{key}"))
            elif schema.get("additionalProperties") is False:
                errors.append(f"{path}/{key}: not allowed")
    if isinstance(value, list) and "items" in schema:
        for i, item in enumerate(value):
            errors.extend(validate(item, schema["items"], f"{path}/{i}"))
    return errors


def generate_json(llm: Callable[[list[dict]], str], schema: dict, messages: list[dict],
                  max_attempts: int = 3) -> tuple[Any, int]:
    convo = list(messages)
    errors: list[str] = []
    for attempt in range(1, max_attempts + 1):
        raw = llm(convo)
        m = FENCE.match(raw)
        body = m.group(1) if m else raw
        try:
            obj = json.loads(body)
            errors = validate(obj, schema)
        except json.JSONDecodeError as e:
            errors = [f"invalid JSON: {e.msg} at char {e.pos}"]
        if not errors:
            return obj, attempt
        convo = convo + [
            {"role": "assistant", "content": raw},
            {"role": "user", "content": "Fix these errors and reply with JSON only:\n" + "\n".join(errors)},
        ]
    raise ValueError("schema validation failed: " + "; ".join(errors))

Try it first.

TimeO(size of value) per validation; O(attempts) LLM calls

SpaceO(depth) recursion plus O(errors); conversation grows per attempt

tests.py · 12 passing

tests.pypython36 lines

schema = {
    "type": "object",
    "required": ["sku", "items"],
    "additionalProperties": False,
    "properties": {
        "sku": {"type": "string", "minLength": 3},
        "status": {"type": "string", "enum": ["open", "closed"]},
        "items": {"type": "array", "items": {
            "type": "object", "required": ["qty"],
            "properties": {"qty": {"type": "integer", "minimum": 1}}}},
    },
}
assert validate({"sku": "ABC", "items": [{"qty": 2}]}, schema) == []
assert validate({"sku": "ABC", "items": [{"qty": 2}, {"qty": True}]}, schema) == ["/items/1/qty: expected integer"]
assert validate({"items": []}, schema) == ["/sku: required"]
assert validate({"sku": "AB", "items": [{"qty": 0}], "x": 1}, schema) == [
    "/sku: shorter than 3", "/items/0/qty: below minimum 1", "/x: not allowed"]
assert validate({"sku": "ABC", "items": [], "status": "lost"}, schema)[0].startswith("/status: must be one of")
assert validate([1], schema) == ["/: expected object"]
assert validate(3.5, {"type": "number"}) == [] and validate(3.5, {"type": "integer"}) != []
fence = chr(96) * 3
replies = iter(["Sure! {sku: ABC}", '{"sku": "ABC", "items": [{"qty": "2"}]}',
                fence + 'json\n{"sku": "ABC", "items": [{"qty": 2}]}\n' + fence])
seen = []
def llm(msgs):
    seen.append(msgs)
    return next(replies)
obj, attempts = generate_json(llm, schema, [{"role": "user", "content": "extract"}])
assert attempts == 3 and obj["items"][0]["qty"] == 2
assert "invalid JSON" in seen[1][-1]["content"]
assert "/items/0/qty: expected integer" in seen[2][-1]["content"]
try:
    generate_json(lambda m: "nope", schema, [], max_attempts=2)
    assert False
except ValueError as e:
    assert "invalid JSON" in str(e)

04Staff-level follow-up

Prefer constrained decoding (grammar or JSON-schema-guided generation) when the serving stack supports it; then this validator only catches semantic rules. Track first-attempt validity and retries per call as metrics, since a prompt or model change that drops first-pass validity from 99% to 90% silently doubles latency at the tail. Also decide what the repair turn costs: often a cheaper model can fix syntax.