Building AI agents is fun until your LLM decides to pass a string into an integer field or hallucinates a non-existent key inside a tool call. I spent three hours last week debugging a pipeline because the OpenAI API returned "age": "twenty-five" instead of an integer. When building production-ready agents, generic function calling isn't enough. You need strict schema validation and an automatic self-correction loop to catch malformed payloads before they crash your backend.
In this guide, I will show you how to combine Pydantic schemas with a dynamic retry loop. This pattern ensures your agents recover from invalid JSON outputs gracefully, dramatically improving agent reliability.
Most LLM providers offer function calling features. You give the model a JSON schema, and it attempts to return structured data matching that schema.
However, models still make mistakes. They might pass invalid emails, out-of-range numbers, or fail to populate required parameters when handling complex tasks. If your application attempts to parse this data directly, your backend throws an exception and the user task fails.
To fix this, we need three distinct layers:
- Strict Schema Validation: Using Pydantic to enforce exact types, ranges, and patterns.
- Runtime Error Capture: Catching validation failures instantly.
- Feedback-Based Retry Logic: Feeding the exact Pydantic error message back to the LLM so it can correct its own output.
If you are building complex systems like workflow automation pipelines, this extra layer of safety is the difference between an agent that works in demo videos and one that runs reliably in production.
Step 1: Define Your Strict Pydantic Schema
We will start by defining the tool parameters using Pydantic V2. Instead of relying on basic JSON types, we can add field validation rules, regular expressions, and custom validators.
Here is an example schema for a financial transfer execution tool:
from pydantic import BaseModel, Field, field_validator
from typing import Literal
class TransferFundsSchema(BaseModel):
account_id: str = Field(
...,
pattern=r"^ACC-\d{6}$",
description="The account ID, formatted like ACC-123456"
)
amount: float = Field(
...,
gt=0,
le=10000,
description="Transfer amount in USD. Must be greater than 0 and at most 10,000"
)
currency: Literal["USD", "EUR", "GBP"] = Field(
default="USD",
description="3-letter currency code"
)
memo: str = Field(
...,
min_length=5,
max_length=100,
description="Reason for transfer"
)
@field_validator("memo")
@classmethod
def memo_must_not_contain_sql(cls, value: str) -> str:
forbidden = ["DROP", "SELECT", "DELETE", "UPDATE"]
if any(word in value.upper() for word in forbidden):
raise ValueError("Memo contains forbidden SQL keywords.")
return value
Look at how explicit this schema is. We enforce exact string patterns, strict monetary bounds, explicit currency options, and even basic sanitization. If an LLM sends "amount": -50 or an invalid account string like "12345", Pydantic immediately throws a ValidationError.
Step 2: Extract OpenAI Structured Outputs JSON Schema
Modern versions of OpenAI allow you to pass Pydantic models directly using response formats or tool definitions. But under the hood, Pydantic converts your model to a standard JSON Schema dictionary.
Here is how you extract the JSON schema to send to your model:
import json
def get_tool_definition():
return {
"type": "function",
"function": {
"name": "transfer_funds",
"description": "Executes a secure financial transfer.",
"parameters": TransferFundsSchema.model_json_schema()
}
}
This guarantees your LLM receives the exact structure defined by your Python codebase.
Step 3: Build the Self-Correcting Retry Wrapper
This is where the magic happens. When the LLM generates arguments for transfer_funds, we intercept the arguments, parse them using TransferFundsSchema.model_validate_json(), and check for errors.
If parsing succeeds, we execute the action. If it fails, we catch the ValidationError, format the human-readable error messages, and send them back to the LLM inside the tool message history.
import openai
from pydantic import ValidationError
client = openai.OpenAI()
def run_agent_with_retry(user_prompt: str, max_retries: int = 3):
messages = [
{"role": "system", "content": "You are a helpful assistant. Call tools accurately based on user requests."},
{"role": "user", "content": user_prompt}
]
tools = [get_tool_definition()]
for attempt in range(max_retries):
print(f"\n--- Attempt {attempt + 1} ---")
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message
messages.append(message)
# Check if the LLM requested a tool call
if not message.tool_calls:
print("Agent responded with text instead of a tool call:")
print(message.content)
return message.content
tool_call = message.tool_calls[0]
tool_name = tool_call.function.name
raw_args = tool_call.function.arguments
print(f"LLM called tool '{tool_name}' with raw args: {raw_args}")
if tool_name == "transfer_funds":
try:
# Validate raw JSON against our Pydantic model
validated_data = TransferFundsSchema.model_validate_json(raw_args)
print("Validation Successful!")
# Execute your real business logic here
return execute_transfer(validated_data)
except ValidationError as e:
# Format validation errors clearly for the model
error_details = e.errors(include_url=False)
clean_errors = [
f"Field '{'->'.join(str(loc) for loc in err['loc'])}': {err['msg']}"
for err in error_details
]
error_message = "Schema validation failed:\n" + "\n".join(clean_errors)
print(f"Validation Error caught: {error_message}")
# Append the validation error as a tool response
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": f"ERROR: Your input arguments were invalid.\n{error_message}\nPlease fix the arguments and try calling the tool again."
})
raise RuntimeError("Agent failed to provide valid tool inputs within the retry limit.")
def execute_transfer(data: TransferFundsSchema):
return f"Successfully transferred ${data.amount} {data.currency} to {data.account_id}."
Step 4: Test the Resilience of the Retry Logic
Let's test this system with an ambiguous prompt that forces the LLM to guess fields incorrectly on purpose.
if __name__ == "__main__":
# The prompt deliberately provides bad data format to trigger validation failures
prompt = "Transfer 15000 dollars to account 999111 for contract payment standard UPDATE query."
try:
result = run_agent_with_retry(prompt)
print("\nFinal Result:", result)
except Exception as err:
print("\nExecution Failed:", err)
What Happens Behind the Scenes:
-
Attempt 1: The LLM sees "account 999111" and sends
"account_id": "999111". It also passes"amount": 15000. -
Pydantic Triggers Errors:
-
account_idfails the regex^ACC-\d{6}$. -
amountfails the maximum boundary check (le=10000). -
memocontains forbidden SQL word"UPDATE".
-
- Error Serialization: The detailed error string goes straight back to the LLM message array.
-
Attempt 2: The LLM reads the errors, fixes the formatting to
"ACC-999111", adjusts the amount down to$10,000, fixes the memo, and submits valid JSON. - Success: Parsing succeeds, and execution completes.
By combining strict schema validation with natural language feedback loops, you eliminate almost all runtime payload errors.
Next Steps for Production Systems
If you want to scale this pattern in production:
- Limit Max Retries: Set retries to 2 or 3. If an agent fails 3 times in a row, fall back to human intervention.
- Token Cost Optimization: Retries consume tokens. Ensure your validation error messages are direct and minimal.
- Log Failures: Store repeated validation failures in your logging system. They usually reveal flaws in your prompt descriptions or schema design.
When building production-ready AI agent development workflows, handling real-world edge cases is essential. If your team needs help scaling fault-tolerant AI solutions or custom LLM infrastructure, Gaper provides experienced software engineers to help build resilient enterprise applications.
Top comments (0)