Stop Prompting, Start Schemas: The Real Fix for Unreliable AI Agents

Every developer who has shipped an AI agent has hit the same wall. You craft a prompt that looks bulletproof. You instruct the model to "return JSON only." You add "do not wrap in markdown." You even threaten it. And then, in production, the agent returns:

Here's your result:
{ "status": "success", "data": { ... }

Trailing backticks. A missing brace. A field that's null when the schema says string. The model "understands" the task, but it can't reliably emit the contract you asked for. The fix isn't a better prompt — it's a schema.

This is a deep dive into why schema-constrained generation — not prompt engineering — is the real fix for unreliable AI agents. We'll look at how it works under the hood, why it beats prompt-based parsing, and how to build agents that fail loudly, validate hard, and never silently corrupt data. If you're building production agents with LLMs, this is the mental model you need.

• 1. The Problem: Why Prompting Alone Fails
• 2. The Core Idea: Constrain the Output Space
• 3. How Schema-Constrained Generation Works Under the Hood
• 4. Choosing a Schema Layer
• 5. Hands-On: Building a Reliable Agent with Structured Outputs
• 6. Beyond JSON: Tool Calling and Multi-Step Agents
• 7. Production Best Practices
• 8. When Schemas Aren't Enough
• Frequently Asked Questions

1. The Problem: Why Prompting Alone Fails

Prompting is a soft constraint. It tells the model what you want, but it doesn't give the model a mechanism to guarantee it. The model is a next-token predictor sampling from a probability distribution over the entire vocabulary — and at every step, it can choose a token that breaks your contract.

When you say "return JSON," you're asking the model to satisfy a grammar it has only statistically learned. The model has seen thousands of JSON examples in training, so it's likely to produce something JSON-shaped. But "likely" is not "guaranteed." The failure modes are predictable and familiar:

• Format drift: The model wraps output in ```json fenced blocks, adds prose before or after, or uses single quotes instead of double quotes.
• Truncation: Long outputs hit the max_tokens limit mid-object, producing invalid JSON.
• Hallucinated fields: The model invents keys that weren't in your schema, or returns null for required fields.
• Type coercion: A field declared as integer comes back as "42" (a string), or a list comes back as a single object.
• Semantic nonsense: The JSON parses perfectly, but the values are wrong — a date in the wrong timezone, an enum value that doesn't exist, a status that contradicts the data.

Each of these is a distributional failure. The model's weights encode a tendency, not a rule. And this is the key insight: you cannot prompt your way out of a probability distribution. You can nudge the distribution, but you can't force it to respect a grammar.

The second problem is that prompt-based approaches defer all the work to post-hoc parsing. You generate text, then you try to parse it with json.loads() or a regex. When parsing fails, you either retry the whole generation (expensive, non-deterministic) or you fall back to a default (silently wrong). Both are bad. Retrying is a band-aid; it doesn't fix the root cause. It just rolls the dice again.

The root cause is that you're asking a probabilistic system to produce a deterministic artifact, and you're only enforcing the contract after the fact.

Schemas flip this around. Instead of generating freely and hoping, you constrain the generation process itself so that the model physically cannot emit invalid output. That's a fundamentally different architecture.

2. The Core Idea: Constrain the Output Space

A schema is a formal specification of the output's shape: its fields, types, allowed values, nesting, and constraints. JSON Schema is the canonical example, but the same idea applies to Pydantic models, TypeScript interfaces, Zod schemas, or even a context-free grammar.

The fix is to feed that schema into the generation loop, not into the prompt. Instead of letting the model pick any token from the vocabulary at each step, you mask the vocabulary to only the tokens that keep the output valid under the schema.

This is called constrained decoding, structured outputs, or grammar-constrained generation. It's the difference between asking a model to "draw a circle" and handing it a compass. With a compass, the circle is guaranteed — the tool enforces the constraint.

Here's the conceptual shift:

With schema-constrained generation, the contract is enforced during decoding. The model can only choose tokens that advance toward a valid completion. You get what you asked for, every time.

3. How Schema-Constrained Generation Works Under the Hood

To understand why this is so much more reliable, you need to know what's happening at the token level. Let's walk through the mechanics.

3.1 The logit mask

At every decoding step, the model produces a probability distribution over the vocabulary — a vector of logits, one per token. Normally, you sample from this full distribution. With constrained decoding, you build a mask of allowed tokens and set the logits of disallowed tokens to -infinity (or a very large negative number), so they can never be sampled.

# Conceptual — not runnable as-is, but shows the mechanism.
import torch

logits = model(input_ids)  # shape: [1, vocab_size]
allowed_mask = build_allowed_token_mask(state, schema)  # shape: [1, vocab_size]

masked_logits = logits.masked_fill(~allowed_mask, float('-inf'))
next_token = sample(masked_logits)

The mask is the entire trick. It's not a heuristic; it's a hard constraint. The model cannot produce a token that would violate the schema, because that token has zero probability.

3.2 The state machine

But how do you know which tokens are allowed at each step? You need to track the current state of the partial output and derive the allowed tokens from it. This is where the schema becomes a finite-state machine (or a pushdown automaton for nested structures).

Imagine the schema:

{
  "type": "object",
  "properties": {
    "name": { "type": "string" },
    "age": { "type": "integer", "minimum": 0 }
  },
  "required": ["name", "age"]
}

After the model emits { "name": "Alice", , the parser state knows:

• We're inside an object.
• name is already satisfied.
• The next key must be age (since required includes it and it's not yet present).
• The next token must be "age" (or whitespace, or } if age weren't required — but it is, so } is not allowed yet).

The state machine encodes all the grammar rules of JSON Schema: object key ordering (or unordered keys), string escapes, number formatting, enum membership, array lengths, minItems/maxItems, additionalProperties, and so on. As the model generates, the state advances, and the allowed token set is recomputed from the new state.

3.3 Grammar-based generation

For arbitrary JSON Schema, implementations typically compile the schema into a context-free grammar (CFG) and then use a parser (like a GLR or Earley parser) to track the valid continuations. Libraries such as Outlines and Instructor do exactly this: they convert a Pydantic model or JSON Schema into a grammar, then run constrained decoding against any compatible backend (OpenAI, Anthropic, local models via transformers, llama.cpp, vLLM, etc.).

# Outlines — schema-constrained generation against a local model.
import outlines
from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int

model = outlines.models.transformers("meta-llama/Meta-Llama-3-8B-Instruct")
generator = outlines.generate.json(model, User)

result = generator("Extract the user from: Alice is 30 years old.")
print(result)  # User(name='Alice', age=30) — always valid, no parsing needed

The key takeaway: the schema is not documentation. It is the generator. The model never has a chance to emit invalid JSON, because invalid tokens are unreachable.

3.4 Provider-native structured outputs

Since late 2024, major providers have shipped this natively. OpenAI's Structured Outputs and Anthropic's tool-use with strict schemas both implement constrained decoding on their side. You pass a JSON Schema, and the API guarantees the response matches it. This is the easiest adoption path for most teams.

from openai import OpenAI
from pydantic import BaseModel

client = OpenAI()

class User(BaseModel):
    name: str
    age: int

response = client.chat.completions.create(
    model="gpt-4o",
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "user_extraction",
            "strict": True,
            "schema": User.model_json_schema()
        }
    },
    messages=[{"role": "user", "content": "Alice is 30."}]
)

user = User.model_validate_json(response.choices[0].message.content)

Notice what changed: there's no json.loads() call, no try/except, no regex cleanup, no retry loop. The response is guaranteed to be valid JSON matching the schema, so you can deserialize it directly. If the API can't satisfy the schema, it returns an error — it doesn't silently return garbage.

4. Choosing a Schema Layer

Before you build, you need to pick your schema source of truth. The choice matters because it affects how you define, validate, and version your contracts.

My recommendation for most teams: define your contract once as a Pydantic model (or Zod schema), and derive the JSON Schema from it. This gives you type safety in code and a provider-compatible schema for constrained decoding. You never hand-write JSON Schema by hand — that's how drift happens.

from pydantic import BaseModel, Field
from typing import Literal

class Order(BaseModel):
    order_id: str = Field(description="Unique order identifier")
    status: Literal["pending", "paid", "shipped", "cancelled"]
    total_cents: int = Field(ge=0, description="Total in cents, non-negative")
    items: list[str] = Field(min_length=1, max_length=50)

Now Order.model_json_schema() gives you a JSON Schema that both OpenAI and Anthropic can consume directly. And Order.model_validate_json() gives you a validated Python object on the way out. One source of truth, two directions.

5. Hands-On: Building a Reliable Agent with Structured Outputs

Let's build a concrete, production-shaped example: an order-processing agent that extracts structured data from a natural-language email and decides whether to approve it. This is the kind of task that breaks naive prompt-based agents constantly.

Step 1 — Define the schema

from pydantic import BaseModel, Field
from typing import Literal, Optional

class OrderItem(BaseModel):
    sku: str
    quantity: int = Field(ge=1, le=1000)
    unit_price_cents: int = Field(ge=0)

class Extraction(BaseModel):
    """Extracted order data from a customer email."""
    customer_email: str
    order_items: list[OrderItem] = Field(min_length=1)
    shipping_address: str
    priority: Literal["standard", "express"] = "standard"
    notes: Optional[str] = None

Step 2 — Configure the client with structured outputs

from openai import OpenAI

client = OpenAI()

SYSTEM_PROMPT = """
You are an order extraction agent. Extract order information from customer emails.
Return ONLY the structured data. Never invent fields that are not in the schema.
If a required field is missing from the email, use your best judgment to infer it,
but never fabricate a customer email address.
"""

def extract_order(email_body: str) -> Extraction:
    response = client.chat.completions.create(
        model="gpt-4o",
        response_format={
            "type": "json_schema",
            "json_schema": {
                "name": "order_extraction",
                "strict": True,
                "schema": Extraction.model_json_schema()
            }
        },
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": email_body}
        ],
        temperature=0,  # deterministic, best for extraction
    )

    # No json.loads(), no try/except, no regex cleanup.
    # The API guarantees this matches Extraction's schema.
    return Extraction.model_validate_json(response.choices[0].message.content)

Step 3 — Validate on the way out anyway

Even with structured outputs, you should still validate. Why? Two reasons: (1) it catches semantic issues that JSON Schema can't express, and (2) it protects you if you ever switch to a provider that doesn't support strict schemas. Validation is cheap; silent corruption is expensive.

class ExtractionError(Exception):
    """Raised when an extraction fails semantic validation."""
    def __init__(self, message: str, details: dict):
        super().__init__(message)
        self.details = details

def validate_semantics(extraction: Extraction) -> None:
    """Business rules that JSON Schema can't enforce."""
    if not extraction.customer_email.endswith("@"):
        raise ExtractionError("Customer email is malformed", {
            "field": "customer_email",
            "value": extraction.customer_email
        })

    total = sum(i.unit_price_cents * i.quantity for i in extraction.order_items)
    if total <= 0:
        raise ExtractionError("Order total must be positive", {
            "field": "order_items",
            "value": total
        })

Step 4 — Wire it into a resilient pipeline

Now the full agent loop. Notice the retry logic is not for parsing failures — those are impossible now. It's for semantic failures and provider errors, which is a much smaller and more predictable surface.

import logging
from tenacity import retry, stop_after_attempt, wait_exponential

logger = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    reraise=True,
)
def safe_extract(email_body: str) -> Extraction:
    extraction = extract_order(email_body)
    validate_semantics(extraction)
    return extraction

# Usage
try:
    order = safe_extract("Hi, please ship 2x SKU-1234 and 1x SKU-5678 to 123 Main St.")
    print(f"Processed order for {order.customer_email}: {len(order.order_items)} items")
except ExtractionError as e:
    logger.error("Semantic validation failed: %s", e.details)
    # Route to human review, don't silently drop
    raise

This is the real fix. The prompt still matters — it guides the model's reasoning — but it no longer bears the burden of enforcing the format. That burden is on the schema, where it belongs.

6. Beyond JSON: Tool Calling and Multi-Step Agents

JSON schemas are just the beginning. The same principle extends to tool calling and multi-step agent loops, which is where most production agents actually live.

6.1 Tool schemas as contracts

When an agent calls a tool, it's not just returning data — it's selecting an action. That selection is a structured decision, and it should be schema-constrained too. Provider-native tool calling already does this: you declare each tool's parameters as a JSON Schema, and the model must emit a valid call.

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": "Get the current status of an order.",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string", "description": "The order ID"}
                },
                "required": ["order_id"],
                "additionalProperties": False
            }
        }
    }
]

With strict: True and additionalProperties: False, the provider guarantees the tool call is well-formed. The agent can't call get_order_status without an order_id, and it can't pass extra parameters. This eliminates an entire class of "tool call hallucination" bugs.

6.2 The ReAct loop with schema enforcement

A ReAct (Reasoning + Acting) agent interleaves reasoning steps and tool calls. The classic bug in naive ReAct implementations is that the model emits a malformed action, the parser fails, and the whole loop crashes. Schema enforcement fixes this at the source.

from pydantic import BaseModel
from typing import Union

class Think(BaseModel):
    thought: str

class Act(BaseModel):
    tool_name: str
    tool_args: dict

class ReActStep(BaseModel):
    """One step in the ReAct loop. The model must choose exactly one action."""
    step: Union[Think, Act] = Field(description="Either a reasoning step or a tool action")
    done: bool = Field(description="True if the task is complete")

By making step a discriminated union, you force the model to produce either a thought or an action — not both, not neither, not a malformed hybrid. The loop then becomes deterministic:

def run_agent(task: str) -> str:
    history = [{"role": "user", "content": task}]
    for _ in range(MAX_STEPS):
        response = client.chat.completions.create(
            model="gpt-4o",
            response_format={
                "type": "json_schema",
                "json_schema": {
                    "name": "react_step",
                    "strict": True,
                    "schema": ReActStep.model_json_schema()
                }
            },
            messages=history,
        )
        step = ReActStep.model_validate_json(response.choices[0].message.content)

        if step.done:
            return final_answer(step)

        if isinstance(step.step, Act):
            result = execute_tool(step.step.tool_name, step.step.tool_args)
        else:
            result = None  # reasoning step, no tool call

        history.append({"role": "assistant", "content": response.choices[0].message.content})
        history.append({"role": "user", "content": str(result or step.step.thought)})

    raise MaxStepsExceeded()

Notice there's no json.loads(), no if "tool" in text heuristic, no regex extraction. The schema does all the structural work. The model's reasoning is still free — that's what the prompt guides — but the structure is locked.

7. Production Best Practices

Schemas make agents reliable, but reliability in production requires more than just enabling structured outputs. Here's what I've learned shipping these systems.

7.1 Version your schemas

A schema is an API contract. When you change it, you break every stored response and every downstream consumer. Version it explicitly:

class ExtractionV2(BaseModel):
    """V2: adds currency field and drops legacy notes field."""
    customer_email: str
    order_items: list[OrderItem]
    shipping_address: str
    currency: Literal["USD", "EUR", "GBP"] = "USD"
    priority: Literal["standard", "express"] = "standard"

Keep the old version around for replay and migration. Never silently mutate a schema in place.

7.2 Fail loudly, not silently

The worst thing an agent can do is return null for a required field and have your code treat it as valid. Always validate on the way out, even with structured outputs. If validation fails, raise an error that gets routed to a human — don't log and continue.

try:
    extraction = safe_extract(email)
except ExtractionError as e:
    # Route to human review queue, not to a silent default
    review_queue.enqueue(email, error=e.details)
    raise

7.3 Use temperature 0 for extraction, low for reasoning

Extraction is a deterministic task. temperature=0 gives you the most reproducible results. For open-ended reasoning or creative tasks, you might want a bit more variety — but even then, keep the structure constrained and vary the content, not the shape.

7.4 Measure the right things

Track these metrics, and you'll see the difference schemas make:

• Schema violation rate: should be near zero with structured outputs.
• Post-hoc parse failure rate: should drop to zero; if it doesn't, your provider isn't enforcing.
• Semantic validation failure rate: the real remaining signal — this tells you where the model's reasoning is weak.
• Retry rate: should drop dramatically, since you're no longer retrying on format errors.

7.5 Know the limits of JSON Schema

JSON Schema can't express everything. It can't say "this date must be in the future" or "this email must exist in our database" or "these two fields must be consistent with each other." Those are semantic constraints, and they belong in your validation layer (Pydantic validators, business rules, etc.). Schema-constrained generation handles syntax; your code handles semantics. Don't conflate the two.

8. When Schemas Aren't Enough

Let's be honest about the limits. Schema-constrained generation is a massive upgrade, but it's not a silver bullet. There are two classes of failure it doesn't fix:

1. Reasoning failures. The model produces perfectly valid JSON, but the content is wrong. It extracts the wrong date, picks the wrong enum value, or infers a fact that isn't in the source. Schema constraints can't fix bad reasoning — they can only make sure the bad reasoning is well-formatted. This is where you need better prompts, better context, retrieval, or a more capable model.

2. Semantic inconsistency. Two fields that are individually valid but jointly impossible (e.g., total_cents doesn't match the sum of items). No schema can express cross-field invariants, so you need validation logic and, when it fails, a recovery strategy: retry with the validation error fed back into the prompt, or escalate to a human.

The schema guarantees the shape. It does not guarantee the truth. Those are two different problems, and you need both.

For failure #2, a good pattern is to feed validation errors back into the model as a corrective message:

def extract_with_self_correction(email_body: str, max_attempts: int = 2) -> Extraction:
    messages = [{"role": "user", "content": email_body}]
    for attempt in range(max_attempts):
        response = client.chat.completions.create(
            model="gpt-4o",
            response_format={"type": "json_schema", "json_schema": {...}},
            messages=messages,
        )
        try:
            extraction = Extraction.model_validate_json(response.choices[0].message.content)
            validate_semantics(extraction)
            return extraction
        except (ValidationError, ExtractionError) as e:
            if attempt == max_attempts - 1:
                raise
            messages.append({"role": "assistant", "content": response.choices[0].message.content})
            messages.append({"role": "user", "content": f"Your output failed validation: {e}. Please correct it."})

This keeps the format guaranteed by the schema, while giving the model a chance to fix its reasoning. It's the best of both worlds.

Frequently Asked Questions

Why do I still get malformed JSON even with structured outputs enabled?

If you're using a provider-native structured outputs API (e.g., response_format with json_schema), malformed JSON should be impossible — the provider enforces it during decoding. If you still see it, you're likely (a) not passing strict: True, (b) not setting additionalProperties: False on all objects, or (c) using a provider that only suggests rather than enforces the schema. Some providers return a schema that's a subset of JSON Schema and silently degrade. Always validate on the way out as a safety net.

What's the difference between structured outputs and function calling?

Function calling is a specific form of structured output where the schema describes a tool's parameters, and the model's output is a tool invocation. Structured outputs is the broader concept: any JSON Schema can constrain the response. Function calling is often the easiest way to get structured outputs on older models, but the dedicated json_schema response format is now more general and more reliable.

Should I use Pydantic or raw JSON Schema?

Use Pydantic (or Zod in TypeScript) as your source of truth and derive JSON Schema from it. Hand-writing JSON Schema is error-prone and leads to drift between your code types and your API contract. With Pydantic, you get type safety, validation, and auto-generated schemas in one place. Only use raw JSON Schema when you need a language-agnostic contract shared across services.

If you're building agents that touch production data, make the schema the contract — not the prompt. The prompt guides reasoning; the schema enforces structure. That division of labor is what separates a demo from a system you can actually ship. For more on engineering reliable AI systems, see Tamiz's Insights and the engineering guides at tamiz.pro.

Beyond Validation: Composable Schemas

Validation is only the first layer of the schema-first approach. The real power emerges when you treat schemas as composable building blocks rather than isolated constraints. Think of it this way: a single schema tells you what a valid response looks like, but a composition of schemas tells you how a system should behave across multiple turns, tools, and failure modes.

Let's move from simple validation to a more sophisticated pattern—schema-driven state machines. Consider a customer support agent that needs to handle a multi-step refund process. Instead of writing a prompt that says "handle the refund flow," you define a schema for each state and a transition map between them.

from pydantic import BaseModel, Field
from typing import Literal, Union
from enum import Enum

class RefundState(str, Enum):
    AWAITING_ORDER_ID = "awaiting_order_id"
    AWAITING_REASON = "awaiting_reason"
    AWAITING_APPROVAL = "awaiting_approval"
    COMPLETED = "completed"

class OrderIDRequest(BaseModel):
    intent: Literal["provide_order_id"] = "provide_order_id"
    order_id: str = Field(..., pattern=r"^ORD-\d{6}$")

class RefundReasonRequest(BaseModel):
    intent: Literal["provide_reason"] = "provide_reason"
    reason: str = Field(..., min_length=10, max_length=500)
    urgency: Literal["low", "medium", "high"] = "medium"

class ApprovalDecision(BaseModel):
    intent: Literal["approve_refund"] = "approve_refund"
    approved: bool
    notes: str | None = None

class RefundResponse(BaseModel):
    state: RefundState
    data: Union[OrderIDRequest, RefundReasonRequest, ApprovalDecision]

# Define the transition function
def next_state(current: RefundState, response: RefundResponse) -> RefundState:
    if current == RefundState.AWAITING_ORDER_ID:
        if isinstance(response.data, OrderIDRequest):
            return RefundState.AWAITING_REASON
        raise ValueError("Expected order ID")
    
    if current == RefundState.AWAITING_REASON:
        if isinstance(response.data, RefundReasonRequest):
            return RefundState.AWAITING_APPROVAL
        raise ValueError("Expected refund reason")
    
    if current == RefundState.AWAITING_APPROVAL:
        if isinstance(response.data, ApprovalDecision):
            return RefundState.COMPLETED if response.data.approved else RefundState.AWAITING_REASON
        raise ValueError("Expected approval decision")
    
    return RefundState.COMPLETED

This pattern gives you something that pure prompting can never provide: provable state transitions. You can unit-test every path, enforce invariants at compile time, and guarantee that the agent cannot skip a step or hallucinate a state that doesn't exist.

The Execution Layer: Wiring Schemas into the Agent Loop

Now that you have schemas, you need an execution layer that actually uses them. The naive approach is to parse the model output and hope it matches. The robust approach is to validate, retry, and fall back through a structured pipeline.

Here's a production-ready agent loop that treats schemas as first-class citizens:

import json
from typing import Type, TypeVar, Generic
from pydantic import BaseModel, ValidationError

T = TypeVar('T', bound=BaseModel)

class SchemaAgent(Generic[T]):
    def __init__(
        self,
        model: callable,  # LLM callable: (prompt, schema) -> str
        schema: Type[T],
        max_retries: int = 3,
        fallback: T | None = None,
    ):
        self.model = model
        self.schema = schema
        self.max_retries = max_retries
        self.fallback = fallback
    
    def run(self, prompt: str, context: dict | None = None) -> T:
        last_error = None
        
        for attempt in range(self.max_retries):
            try:
                # Pass the schema as a structured constraint, not just a prompt hint
                raw_output = self.model(
                    prompt=prompt,
                    schema=self.schema.model_json_schema(),
                    context=context or {},
                )
                
                # Validate and coerce
                validated = self.schema.model_validate_json(raw_output)
                return validated
                
            except ValidationError as e:
                last_error = e
                # Feed the error back into the prompt for self-correction
                prompt = self._augment_prompt(prompt, e)
                continue
            except json.JSONDecodeError as e:
                last_error = e
                prompt = self._augment_prompt(prompt, "Response was not valid JSON")
                continue
        
        if self.fallback is not None:
            return self.fallback
        
        raise RuntimeError(
            f"Agent failed after {self.max_retries} attempts. Last error: {last_error}"
        )
    
    def _augment_prompt(self, prompt: str, error: Exception) -> str:
        return f"""
{prompt}

Your previous response was invalid for the following reason:
{error}

Please provide a corrected response that strictly adheres to the schema.
"""

Notice what's happening here. The schema isn't just appended to the prompt as text—it's passed as a structured parameter to the model. This forces the model to generate output that can be parsed and validated deterministically. The retry loop includes the validation error in the next prompt, which gives the model a concrete correction target rather than a vague "try again."

Patterns for Production: Multi-Step Workflows

Real-world agents rarely do one thing. They orchestrate tools, call external APIs, and maintain conversational context. The schema-first approach shines in these scenarios because it gives you a way to define the contract for each step.

Here's a pattern for a multi-step agent that handles a complete user request:

from typing import List, Dict, Any
from pydantic import BaseModel

class ToolCall(BaseModel):
    tool_name: str
    arguments: Dict[str, Any]

class AgentStep(BaseModel):
    step_id: int
    tool_calls: List[ToolCall]
    expected_output_schema: Dict[str, Any]
    next_step: int | None = None

class WorkflowDefinition(BaseModel):
    steps: Dict[int, AgentStep]
    entry_point: int = 0

class WorkflowExecutor:
    def __init__(self, workflow: WorkflowDefinition, tools: Dict[str, callable]):
        self.workflow = workflow
        self.tools = tools
    
    def execute(self, initial_input: Dict[str, Any]) -> Dict[str, Any]:
        current_step = self.workflow.entry_point
        context = initial_input
        
        while current_step is not None:
            step = self.workflow.steps[current_step]
            
            # Execute each tool call in sequence
            for tool_call in step.tool_calls:
                if tool_call.tool_name not in self.tools:
                    raise ValueError(f"Unknown tool: {tool_call.tool_name}")
                
                result = self.tools[tool_call.tool_name](**tool_call.arguments)
                
                # Validate the result against the expected schema
                self._validate_result(result, step.expected_output_schema)
                
                # Merge result into context
                context.update(result)
            
            current_step = step.next_step
        
        return context
    
    def _validate_result(self, result: Any, schema: Dict[str, Any]) -> None:
        # Use a lightweight validator here, or integrate with Pydantic
        # This is where you'd enforce the schema contract
        pass

This pattern is especially valuable when you're building agents that call external APIs. Instead of hoping the LLM correctly formats a function call, you define the exact schema for each tool's input and output. The workflow executor enforces those schemas at runtime, so a malformed tool call fails fast rather than propagating through the system.

Error Recovery: The Schema-First Safety Net

One of the most underrated benefits of schema-first design is that it gives you a structured way to handle errors. When an agent produces invalid output, you don't just retry—you can analyze why it failed and take corrective action.

from pydantic import BaseModel, ValidationError
from typing import Literal, Union

class RecoveryAction(BaseModel):
    action: Literal["retry", "clarify", "escalate"]
    message: str
    suggested_schema: Dict[str, Any] | None = None

class ErrorAnalyzer:
    def analyze(self, error: ValidationError, attempts: int) -> RecoveryAction:
        if attempts < 2:
            return RecoveryAction(
                action="retry",
                message="Please correct the output to match the schema.",
                suggested_schema=error.schema,
            )
        
        if self._is_ambiguity_error(error):
            return RecoveryAction(
                action="clarify",
                message="The request is ambiguous. Please provide more specific information.",
            )
        
        return RecoveryAction(
            action="escalate",
            message="Unable to produce valid output. Escalating to human review.",
        )
    
    def _is_ambiguity_error(self, error: ValidationError) -> bool:
        # Heuristic: ambiguity errors often involve missing fields or wrong types
        return any("missing" in str(e) or "type" in str(e).lower() for e in error.errors())

This error analyzer gives you a decision tree for handling failures. Instead of blindly retrying, you can escalate to a human when the model consistently fails, or ask for clarification when the input is ambiguous. This is the kind of robustness that makes agents actually deployable in production.

Conclusion: The Schema-First Mindset

We started this article with a provocative claim: prompting is not the fix for unreliable AI agents. By now, I hope you see why. Prompts are suggestions; schemas are contracts. Suggestions can be ignored; contracts must be honored.

The shift to schema-first development changes how you build agents in fundamental ways:

1. You design contracts before prompts. The schema defines what the system must produce, and the prompt becomes just one mechanism for achieving that.
2. You test determinism, not vibes. With schemas, you can write unit tests that verify the agent's behavior in every state, not just check if the output "looks right."
3. You fail fast and recover gracefully. Invalid outputs are caught at the boundary, analyzed, and corrected through a structured pipeline rather than silently propagating through the system.
4. You compose agents like software. Schemas make agents composable because they define clear interfaces between components.

The tools I've shown here—Pydantic for validation, state machines for workflow control, and structured error recovery—are not exotic. They're the same patterns you'd use for any production software. The only difference is that now you're applying them to systems that generate code and text dynamically.

Stop prompting. Start schemas. Your agents will thank you.

For more on engineering reliable AI systems, see Tamiz's Insights and the engineering guides at tamiz.pro.