Validate and Repair Loop

The design rule: the quality of the repair is bounded by the quality of the error message. A validator that says "invalid" is useless here; one that says which field, what was expected, and what was received usually gets a correct answer on the first repair.

Format errors for a model, not for a log

Raw validator output is written for developers and is often terse to the point of uselessness out of context. Translate it: field path, what was expected, what was received, and where possible a hint. Cap the list — a model handed forty errors fixes the first few and drifts on the rest.

MAX_ERRORS_SHOWN = 8


def format_errors(e: ValidationError) -> str:
    lines = []
    for err in e.errors()[:MAX_ERRORS_SHOWN]:
        path = ".".join(str(p) for p in err["loc"]) or "(root)"
        got = repr(err.get("input"))[:80]
        lines.append(f"- {path}: {err['msg']} (received {got})")
    if len(e.errors()) > MAX_ERRORS_SHOWN:
        lines.append(f"- ...and {len(e.errors()) - MAX_ERRORS_SHOWN} more")
    return "\n".join(lines)

The bounded loop

Two repair attempts is the right default. Track whether the error set is shrinking: if attempt two produces the same errors as attempt one, the model cannot satisfy the constraint by reformatting and further attempts are waste. Escalate instead — a stronger model, a simplified schema, or a clean failure.

REPAIR = """Your previous output failed validation.

Your output:
{previous}

Errors:
{errors}

Return the corrected output. Fix only what the errors identify — keep every other
field exactly as it was. If a field cannot be determined from the source, use null
rather than inventing a value."""


def generate_validated(prompt: str, schema: type[T], max_repairs: int = 2) -> T:
    raw = call_model(prompt)
    previous_errors: set[str] = set()

    for attempt in range(max_repairs + 1):
        try:
            return schema.model_validate_json(raw)
        except ValidationError as e:
            current = {f"{'.'.join(map(str, x['loc']))}:{x['type']}" for x in e.errors()}

            if attempt == max_repairs:
                raise UnrepairableOutput(raw, e)
            if current == previous_errors:
                # Same failures twice: reformatting will not fix this.
                raise UnrepairableOutput(raw, e, reason="not_converging")

            previous_errors = current
            metrics.increment("repair.attempt", tags={"attempt": attempt + 1})
            raw = call_model(REPAIR.format(previous=raw, errors=format_errors(e)))

Validators beyond schemas

The loop generalizes to any deterministic checker that produces actionable errors. Generated SQL goes through a parser and an EXPLAIN; generated code goes through a type checker and a test run; generated configuration goes through the tool's own validator. The pattern is identical — run the checker, feed its errors back, bound the loop — and it is what makes model-generated code usable in a pipeline.

def generate_sql(question: str, schema_ddl: str, max_repairs: int = 2) -> str:
    sql = call_model(sql_prompt(question, schema_ddl))

    for _ in range(max_repairs):
        problems = (
            syntax_errors(sql)
            or unknown_identifiers(sql, schema_ddl)   # catches invented columns
            or explain_errors(sql)                     # planner rejects it
            or forbidden_operations(sql)               # writes in a read-only path
        )
        if not problems:
            return sql
        sql = call_model(REPAIR.format(previous=sql, errors=bullets(problems)))

    raise UnrepairableOutput(sql)

• Error messages must name the field, the expectation, and the received value
• Cap the number of errors shown; long lists degrade repair quality
• Instruct the model to change only what failed, or correct fields will drift
• Detect non-convergence by comparing error sets between attempts
• Track repair rate per constraint — a constraint repaired constantly is a prompt bug