Validation and Retry Strategies

Validation: The Last Line of Defense Before Production

Even with Custom Structured Outputs, validation remains essential. The schema guarantees structure, not semantic coherence. A price field might contain -999.0: a valid float that perfectly matches the schema, yet makes no sense for a price.

The distinction is clear. Pydantic and Zod handle structural validation automatically: required fields, field types, enum constraints, and nested structures. You don’t need to write any code for this. Semantic validation, however, is entirely your responsibility. This includes numeric bounds (a price greater than zero, a score between 0 and 1), cross-field consistency (the total must equal the sum of line items), string formats (a genuinely valid email, a correctly formatted date), and business logic (an due date later than the invoice date).

Embedding Business Rules in the Schema

Pydantic lets you embed these checks directly into the model, making them inseparable from the structure. @field_validator handles individual fields—for example, ensuring a quantity is positive or a unit price isn’t negative. @model_validator(mode="after") runs once all fields are populated, enabling cross-field checks, such as verifying that the stated total matches quantity × unit price within one cent.

from pydantic import BaseModel, Field, field_validator, model_validator

class InvoiceLine(BaseModel):
    description: str
    quantity: float
    unit_price: float
    total: float

    @field_validator("quantity")
    @classmethod
    def quantity_positive(cls, v):
        if v <= 0:
            raise ValueError(f"Quantity must be positive, got: {v}")
        return v

    @field_validator("unit_price")
    @classmethod
    def price_non_negative(cls, v):
        if v < 0:
            raise ValueError(f"Unit price cannot be negative: {v}")
        return v

    @model_validator(mode="after")
    def check_total(self):
        expected = round(self.quantity * self.unit_price, 2)
        if abs(self.total - expected) > 0.01:
            raise ValueError(
                f"Inconsistent total: {self.total} != {self.quantity} x {self.unit_price} = {expected}"
            )
        return self

Zod expresses the same logic using built-in constraints like .positive() and .nonnegative() for fields, and .refine() for cross-object rules applied to the complete object.

import { z } from "zod";

const InvoiceLineSchema = z
  .object({
    description: z.string(),
    quantity: z.number().positive("Quantity must be positive"),
    unitPrice: z.number().nonnegative("Price cannot be negative"),
    total: z.number(),
  })
  .refine(
    (data) => Math.abs(data.total - data.quantity * data.unitPrice) < 0.01,
    { message: "Total does not match quantity x unit price" }
  );

Retry with Feedback

When validation fails, the most effective pattern is to send the error back to the model so it can correct itself. The function below chains three checks in order of importance: complete response, non-null object, and business rule compliance. On failure, it appends the faulty response and the error message to the conversation, giving the next attempt precise diagnostic context. Note the copy of the message list at the start of the function—without it, you’d pollute the caller’s conversation history.

from mistralai.client import Mistral
from pydantic import BaseModel, ValidationError
import json

def extract_with_retry(
    client: Mistral,
    messages: list,
    schema,
    max_retries: int = 3,
    model: str = "mistral-large-latest"
):
    """Extraction with retry and structured error feedback."""
    conversation = list(messages)  # Copy to avoid modifying the original

    for attempt in range(max_retries):
        try:
            response = client.chat.parse(
                model=model,
                messages=conversation,
                response_format=schema,
                temperature=0,
                max_tokens=2048
            )

            # Check that the response is complete
            if response.choices[0].finish_reason != "stop":
                raise ValueError(
                    f"Incomplete response (finish_reason={response.choices[0].finish_reason})"
                )

            parsed = response.choices[0].message.parsed
            if parsed is None:
                raise ValueError("Parsed response is None")

            # Additional semantic validation
            validate_business_rules(parsed)

            return parsed

        except (ValidationError, ValueError) as e:
            error_msg = str(e)
            print(f"Attempt {attempt + 1}/{max_retries} failed: {error_msg}")

            if attempt < max_retries - 1:
                # Add feedback to context
                raw = response.choices[0].message.content if response else ""
                conversation.append({"role": "assistant", "content": raw})
                conversation.append({
                    "role": "user",
                    "content": f"Your response contains an error: {error_msg}. Please correct it."
                })
            else:
                raise RuntimeError(
                    f"Failed after {max_retries} attempts. Last error: {error_msg}"
                )

A Cascade of Fallback Plans

Retry doesn’t solve everything: sometimes a document is too atypical for the schema. Instead of failing abruptly, degrade gracefully. The first attempt uses Custom Structured Outputs, the most reliable option. If that fails, fall back to JSON Mode, which is more permissive, then pass the output through model_validate()—this retains schema constraints without relying on the model to generate them perfectly. As a last resort, the function returns None, leaving the caller to decide between manual queueing or abandonment.

def extract_with_fallback(client, text, schema, system_prompt):
    """Extraction with fallback cascade."""

    # Attempt 1: Custom Structured Output (best)
    try:
        return extract_with_retry(
            client,
            [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": text}
            ],
            schema,
            max_retries=2,
            model="mistral-large-latest"
        )
    except RuntimeError:
        print("Custom parse failed, falling back to JSON Mode")

    # Attempt 2: JSON Mode (more flexible)
    try:
        response = client.chat.complete(
            model="mistral-large-latest",
            messages=[
                {"role": "system", "content": system_prompt + "\nReturn valid JSON."},
                {"role": "user", "content": text}
            ],
            response_format={"type": "json_object"},
            temperature=0
        )
        raw = json.loads(response.choices[0].message.content)
        return schema.model_validate(raw)  # Attempt Pydantic validation
    except Exception:
        print("JSON Mode failed, falling back to manual extraction")

    # Attempt 3: Return partial object or error
    return None

Cut Off the Tap in Time

In a high-frequency pipeline, an API outage turns retry into an amplifier: every document fails three times, your bill triples, and latency spikes. A circuit breaker monitors failures over a sliding window and blocks calls once a threshold is crossed—five failures in sixty seconds in the configuration shown here. A success closes the circuit and clears the counter.

from collections import deque
from time import time

class CircuitBreaker:
    def __init__(self, failure_threshold=5, window_seconds=60):
        self.failures = deque()
        self.threshold = failure_threshold
        self.window = window_seconds
        self.is_open = False

    def record_failure(self):
        now = time()
        self.failures.append(now)
        # Clean up errors outside the window
        while self.failures and self.failures[0] < now - self.window:
            self.failures.popleft()
        if len(self.failures) >= self.threshold:
            self.is_open = True

    def record_success(self):
        self.is_open = False
        self.failures.clear()

    def can_proceed(self) -> bool:
        if not self.is_open:
            return True
        # Check if the window has passed
        now = time()
        while self.failures and self.failures[0] < now - self.window:
            self.failures.popleft()
        if len(self.failures) < self.threshold:
            self.is_open = False
            return True
        return False

# Usage
breaker = CircuitBreaker(failure_threshold=5, window_seconds=60)

def safe_extract(client, text, schema):
    if not breaker.can_proceed():
        raise RuntimeError("Circuit open—too many recent errors")
    try:
        result = extract_with_retry(client, [...], schema)
        breaker.record_success()
        return result
    except RuntimeError:
        breaker.record_failure()
        raise

Make It Visible

All this machinery stays blind without traces. Log every call with its duration and the schema involved, whether it succeeds or fails. This data lets you later determine whether degradation stems from a model issue, a document type, or a prompt change.

import logging
import time

logger = logging.getLogger("structured_output")

def monitored_extract(client, text, schema, **kwargs):
    start = time.time()
    try:
        result = extract_with_retry(client, [...], schema, **kwargs)
        duration = time.time() - start
        logger.info(f"Extraction succeeded in {duration:.2f}s — schema={schema.__name__}")
        return result
    except Exception as e:
        duration = time.time() - start
        logger.error(f"Extraction failed in {duration:.2f}s — schema={schema.__name__} — {e}")
        raise

Key Takeaways

• Structural validation (types) is automatic; semantic validation (logic) is your responsibility
• Retry with feedback is the most effective pattern: send the error back to the model
• Implement a fallback cascade: Custom > JSON Mode > default value
• In high-frequency production, use a circuit breaker to prevent error cascades
• Log every call for monitoring and diagnostics