Introduction to Structured Output Validation with CrewAI
When building AI agent systems, one of the most persistent challenges developers face is getting consistent, predictable data from language models. LLMs are inherently probabilistic — they generate text based on likelihood, not strict schemas. CrewAI addresses this challenge head-on with its structured output validation features, allowing you to enforce schemas, validate responses, and build reliable agent pipelines that integrate cleanly with downstream systems.
In this guide, we'll explore everything you need to know about structured output validation in CrewAI: what it is, why it matters, how to implement it using Pydantic models and instructor-style validation, and the best practices that will keep your agent workflows production-ready.
What Is Structured Output Validation?
Structured output validation is the process of constraining an LLM's response to conform to a predefined data schema, then verifying that the output matches that schema before accepting it. In CrewAI, this is primarily achieved through Pydantic models, which serve as both the instruction template for the model and the validation layer for the response.
Instead of receiving free-form text and parsing it with fragile regex or custom logic, you define a Python class that describes exactly what fields you expect, their types, and any constraints. CrewAI then instructs the underlying LLM to produce output matching that schema and validates the result automatically.
How It Works Under the Hood
CrewAI leverages the litellm library and, where supported, native function calling or JSON mode from providers like OpenAI, Anthropic, and Google. When you attach a Pydantic model to a task, CrewAI:
- Generates a JSON schema from your Pydantic model
- Injects schema instructions into the prompt sent to the LLM
- Requests the response in JSON format (or via tool/function calling)
- Parses the response and validates it against the Pydantic model
- Retries or raises an error if validation fails
Why Structured Output Validation Matters
Without structured output validation, agent systems become brittle. A single unexpected field name, a missing value, or a malformed date can break your entire pipeline. Here are the core reasons structured validation is essential:
- Predictability: Downstream code can rely on the shape and types of the data.
- Type Safety: Pydantic enforces types at runtime, catching errors early.
- Composability: One agent's output can feed directly into another agent's input without manual transformation.
- Reduced Parsing Logic: No more regex hacks or try/except blocks around JSON parsing.
- Better Prompting: The schema itself acts as documentation for what the model should produce.
- Validation Rules: You can enforce business logic — value ranges, string formats, enum memberships — directly in the model.
Setting Up Your Environment
Before diving into code, make sure you have the necessary packages installed. CrewAI handles most dependencies, but you'll want the latest versions for the best structured output support.
pip install crewai crewai-tools pydantic
Set your API key as an environment variable:
export OPENAI_API_KEY="your-api-key-here"
For this tutorial, we'll use OpenAI's models, but the same patterns apply to Anthropic, Google, and other providers supported by CrewAI.
Defining Your First Structured Output Model
The foundation of structured output validation in CrewAI is the Pydantic model. Let's start with a simple example: an agent that analyzes a product review and returns structured sentiment data.
from pydantic import BaseModel, Field
from typing import Literal, Optional
from datetime import datetime
class ReviewAnalysis(BaseModel):
"""Structured output for product review analysis."""
sentiment: Literal["positive", "negative", "neutral"] = Field(
...,
description="The overall sentiment of the review"
)
confidence_score: float = Field(
...,
ge=0.0,
le=1.0,
description="Confidence score between 0 and 1"
)
key_themes: list[str] = Field(
...,
description="List of main themes mentioned in the review"
)
summary: str = Field(
...,
max_length=200,
description="A concise summary of the review, max 200 characters"
)
would_recommend: bool = Field(
...,
description="Whether the reviewer would recommend the product"
)
mentioned_issues: Optional[list[str]] = Field(
default=None,
description="Any specific issues mentioned, if applicable"
)
Notice how each field includes a description. CrewAI uses these descriptions to guide the LLM, so clear, specific descriptions dramatically improve output quality. The Field constraints like ge, le, and max_length are enforced by Pydantic during validation.
Attaching Structured Output to a CrewAI Task
Once you have your Pydantic model, you attach it to a task using the output_pydantic parameter. Let's build a complete crew that uses our ReviewAnalysis model.
from crewai import Agent, Task, Crew, Process
# Define the agent
review_analyst = Agent(
role="Product Review Analyst",
goal="Analyze customer reviews and extract structured insights",
backstory=(
"You are an expert product analyst who has reviewed thousands "
"of customer feedback entries. You excel at identifying sentiment, "
"themes, and actionable insights from reviews."
),
verbose=True,
allow_delegation=False
)
# Define the task with structured output
analysis_task = Task(
description=(
"Analyze the following product review and provide a structured "
"analysis including sentiment, confidence score, key themes, "
"a concise summary, recommendation status, and any issues mentioned.\n\n"
"Review: {review_text}"
),
expected_output="A structured JSON analysis of the review",
agent=review_analyst,
output_pydantic=ReviewAnalysis
)
# Create and run the crew
crew = Crew(
agents=[review_analyst],
tasks=[analysis_task],
process=Process.sequential,
verbose=True
)
result = crew.kickoff(
inputs={
"review_text": (
"I bought this laptop three months ago. The battery life is "
"amazing and the screen is gorgeous, but the keyboard feels "
"cheap and the fan gets loud under load. Overall I'd recommend "
"it for the price, just be aware of the keyboard quality."
)
}
)
# Access the structured output
structured_result = result.pydantic
print(f"Sentiment: {structured_result.sentiment}")
print(f"Confidence: {structured_result.confidence_score}")
print(f"Themes: {structured_result.key_themes}")
print(f"Summary: {structured_result.summary}")
print(f"Would Recommend: {structured_result.would_recommend}")
print(f"Issues: {structured_result.mentioned_issues}")
The result.pydantic attribute gives you a fully validated instance of your ReviewAnalysis model. If the LLM's response fails validation, CrewAI will attempt retries before raising an error.
Working with Nested Models
Real-world use cases often require nested structures. Pydantic makes this straightforward — you can compose models within models. Let's build a more complex example: a research agent that produces a structured report with nested sections.
from pydantic import BaseModel, Field
from typing import Optional
class Citation(BaseModel):
"""A citation reference in a research report."""
source: str = Field(..., description="The name or title of the source")
url: Optional[str] = Field(None, description="URL if available")
relevance: str = Field(..., description="Why this source is relevant")
class KeyFinding(BaseModel):
"""A key finding in the research."""
title: str = Field(..., description="Short title for the finding")
description: str = Field(..., description="Detailed description")
confidence: Literal["high", "medium", "low"] = Field(
..., description="Confidence level in this finding"
)
supporting_citations: list[Citation] = Field(
default_factory=list,
description="Citations supporting this finding"
)
class ResearchReport(BaseModel):
"""A complete structured research report."""
topic: str = Field(..., description="The research topic")
executive_summary: str = Field(
..., max_length=500, description="Executive summary, max 500 chars"
)
key_findings: list[KeyFinding] = Field(
..., description="List of key findings"
)
open_questions: list[str] = Field(
default_factory=list,
description="Unanswered questions for further research"
)
sources_consulted: list[Citation] = Field(
default_factory=list,
description="All sources consulted during research"
)
Now let's create a crew that uses this nested model:
researcher = Agent(
role="Senior Research Analyst",
goal="Conduct thorough research and produce a structured report",
backstory=(
"You are a meticulous research analyst with 15 years of experience "
"in technology and market research. You always cite your sources "
"and clearly distinguish between high and low confidence findings."
),
verbose=True,
allow_delegation=False
)
research_task = Task(
description=(
"Research the following topic and produce a comprehensive "
"structured report with key findings, citations, and open questions.\n\n"
"Topic: {topic}\n\n"
"Focus on recent developments and provide citations for all claims."
),
expected_output="A structured research report with findings and citations",
agent=researcher,
output_pydantic=ResearchReport
)
research_crew = Crew(
agents=[researcher],
tasks=[research_task],
process=Process.sequential,
verbose=True
)
result = research_crew.kickoff(
inputs={"topic": "The impact of large language models on software development"}
)
report = result.pydantic
print(f"Topic: {report.topic}")
print(f"\nExecutive Summary: {report.executive_summary}")
print(f"\nKey Findings ({len(report.key_findings)}):")
for i, finding in enumerate(report.key_findings, 1):
print(f" {i}. [{finding.confidence.upper()}] {finding.title}")
print(f" {finding.description[:100]}...")
print(f" Citations: {len(finding.supporting_citations)}")
print(f"\nOpen Questions: {len(report.open_questions)}")
Using Output JSON for Simpler Cases
In addition to output_pydantic, CrewAI offers output_json, which accepts a JSON schema dictionary instead of a Pydantic model. This is useful when you want lightweight schema enforcement without defining a full model class, or when integrating with systems that provide JSON schemas directly.
json_schema = {
"type": "object",
"properties": {
"product_name": {"type": "string"},
"price": {"type": "number"},
"in_stock": {"type": "boolean"},
"tags": {
"type": "array",
"items": {"type": "string"}
}
},
"required": ["product_name", "price", "in_stock"]
}
extraction_task = Task(
description="Extract product information from: {product_description}",
expected_output="Structured product data in JSON",
agent=extraction_agent,
output_json=json_schema
)
However, for most production use cases, output_pydantic is preferred because it provides type safety, validation rules, and a clean Python interface for working with the results.
Chaining Tasks with Structured Outputs
One of the most powerful patterns in CrewAI is chaining tasks where one agent's structured output becomes another agent's input. This creates reliable multi-step pipelines.
class ContentBrief(BaseModel):
"""Output from the planning agent."""
title: str = Field(..., description="Proposed content title")
target_audience: str = Field(..., description="Target audience description")
key_points: list[str] = Field(..., description="Key points to cover")
tone: Literal["professional", "casual", "technical"] = Field(
..., description="Writing tone"
)
word_count_target: int = Field(
..., ge=100, le=5000, description="Target word count"
)
class FinishedContent(BaseModel):
"""Output from the writing agent."""
title: str = Field(..., description="Final content title")
body: str = Field(..., description="The full content body in markdown")
word_count: int = Field(..., ge=50, description="Actual word count")
meta_description: str = Field(
..., max_length=160, description="SEO meta description, max 160 chars"
)
# Agent 1: Plans the content
planner = Agent(
role="Content Strategist",
goal="Create detailed content briefs from topic requests",
backstory="You are a content strategist who excels at planning engaging articles.",
verbose=True
)
# Agent 2: Writes the content
writer = Agent(
role="Content Writer",
goal="Write high-quality content based on briefs",
backstory="You are a skilled writer who follows briefs precisely.",
verbose=True
)
planning_task = Task(
description="Create a content brief for the topic: {topic}",
expected_output="A structured content brief",
agent=planner,
output_pydantic=ContentBrief
)
writing_task = Task(
description=(
"Write content based on the following brief:\n\n"
"Title: {title}\n"
"Audience: {target_audience}\n"
"Tone: {tone}\n"
"Target word count: {word_count_target}\n"
"Key points to cover:\n{key_points}\n\n"
"Produce the final article and an SEO meta description."
),
expected_output="A finished article with metadata",
agent=writer,
output_pydantic=FinishedContent,
context=[planning_task] # This passes the planning output as context
)
content_crew = Crew(
agents=[planner, writer],
tasks=[planning_task, writing_task],
process=Process.sequential,
verbose=True
)
result = content_crew.kickoff(
inputs={"topic": "Getting started with Kubernetes for beginners"}
)
finished = result.pydantic
print(f"Title: {finished.title}")
print(f"Word Count: {finished.word_count}")
print(f"Meta: {finished.meta_description}")
print(f"\nBody preview:\n{finished.body[:500]}...")
The context=[planning_task] parameter ensures the writing agent has access to the planning task's output. CrewAI automatically handles passing the structured data between tasks.
Handling Validation Errors and Retries
Even with the best schemas, LLMs occasionally produce invalid output. CrewAI provides configuration options to handle these cases gracefully. You can control retry behavior at the task level.
from crewai import Task
robust_task = Task(
description="Extract structured data from: {input_text}",
expected_output="Validated structured data",
agent=data_agent,
output_pydantic=ReviewAnalysis,
max_retries=3 # Retry up to 3 times on validation failure
)
For more advanced error handling, you can wrap your crew execution in a try/except block and implement custom fallback logic:
from pydantic import ValidationError
try:
result = crew.kickoff(inputs={"review_text": review})
structured = result.pydantic
# Process the validated data
save_to_database(structured)
except ValidationError as e:
print(f"Validation failed after retries: {e}")
# Fallback: use raw text output or log for manual review
raw_output = result.raw if result else None
log_for_manual_review(raw_output)
except Exception as e:
print(f"Unexpected error: {e}")
# Handle other failures (API errors, timeouts, etc.)
Custom Validators with Pydantic
Pydantic v2 supports custom validators that let you enforce complex business rules beyond simple type checking. This is invaluable for ensuring data quality from LLM outputs.
from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional
from datetime import date
class FinancialReport(BaseModel):
"""Structured financial report with custom validation."""
company_name: str = Field(..., min_length=1, description="Company name")
report_date: date = Field(..., description="Date of the report")
revenue: float = Field(..., ge=0, description="Total revenue in USD")
expenses: float = Field(..., ge=0, description="Total expenses in USD")
net_profit: float = Field(..., description="Net profit (revenue - expenses)")
currency: str = Field(default="USD", description="Currency code")
@field_validator("currency")
@classmethod
def validate_currency(cls, v):
valid_currencies = {"USD", "EUR", "GBP", "JPY", "CAD", "AUD"}
if v.upper() not in valid_currencies:
raise ValueError(f"Currency must be one of {valid_currencies}")
return v.upper()
@model_validator(mode="after")
def validate_profit_calculation(self):
"""Ensure net_profit equals revenue minus expenses."""
expected_profit = round(self.revenue - self.expenses, 2)
if abs(self.net_profit - expected_profit) > 0.01:
raise ValueError(
f"net_profit ({self.net_profit}) does not match "
f"revenue - expenses ({expected_profit})"
)
return self
@field_validator("report_date")
@classmethod
def validate_date_not_future(cls, v):
from datetime import date as today_date
if v > today_date.today():
raise ValueError("Report date cannot be in the future")
return v
These validators run automatically when CrewAI parses the LLM's response. If the model produces a net profit that doesn't match revenue minus expenses, the validation fails and CrewAI retries with an error message guiding the model to correct its output.
Best Practices for Structured Output Validation
1. Write Descriptive Field Descriptions
The description parameter in Pydantic fields is your primary tool for guiding the LLM. Be specific about format, allowed values, and intent.
# Bad - vague description
score: float = Field(..., description="A score")
# Good - specific and actionable
score: float = Field(
...,
ge=0.0,
le=10.0,
description="Relevance score from 0.0 (irrelevant) to 10.0 (highly relevant)"
)
2. Use Enums and Literal Types for Constrained Values
When a field can only take specific values, use Literal or Python enums. This prevents the model from inventing new categories.
from enum import Enum
class Priority(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class Issue(BaseModel):
title: str
priority: Priority
# Or use Literal:
# priority: Literal["low", "medium", "high", "critical"]
3. Keep Models Focused and Reasonably Sized
Large, deeply nested models can confuse LLMs and increase the chance of validation failures. If you need complex output, consider breaking it into multiple tasks with smaller models, then assembling the results programmatically.
4. Provide Examples in Task Descriptions
While the schema guides structure, including an example in the task description dramatically improves output quality, especially for complex fields.
task = Task(
description=(
"Analyze the customer feedback and return structured data.\n\n"
"Example output structure:\n"
'{{"sentiment": "positive", "score": 0.85, "tags": ["quality", "price"]}}\n\n'
"Feedback: {feedback}"
),
expected_output="Structured analysis",
agent=analyst,
output_pydantic=FeedbackAnalysis
)
5. Use Optional Fields Judiciously
Mark fields as Optional only when the LLM might legitimately not have information to fill them. Overusing optional fields can lead to lazy outputs where the model skips fields it should populate.
6. Test with Edge Cases
Always test your crews with edge-case inputs: empty strings, very long text, ambiguous content, and inputs in different languages. This reveals where your schema might be too strict or too loose.
7. Log Raw Outputs for Debugging
When validation fails, the raw LLM output is invaluable for debugging. Always capture result.raw alongside the structured result.
result = crew.kickoff(inputs=inputs)
if result.pydantic:
process_data(result.pydantic)
else:
# Log raw output for debugging
with open("failed_outputs.log", "a") as f:
f.write(f"---\nInput: {inputs}\nRaw: {result.raw}\n")
8. Choose the Right Model
Structured output quality varies significantly between models. GPT-4o, Claude 3.5 Sonnet, and Gemini 1.5 Pro generally handle complex schemas well. Smaller or older models may struggle with nested structures or strict constraints. Match your model choice to your schema complexity.
Using Output File for Persistence
CrewAI can automatically save structured outputs to files, which is useful for auditing and downstream processing.
analysis_task = Task(
description="Analyze review: {review_text}",
expected_output="Structured analysis",
agent=analyst,
output_pydantic=ReviewAnalysis,
output_file="output/review_analysis.json" # Saves validated JSON
)
The output is saved as valid JSON conforming to your Pydantic model's schema, ready for ingestion by other systems.
Conclusion
Structured output validation is one of CrewAI's most valuable features for building production-grade AI agent systems. By combining Pydantic's robust validation with CrewAI's task orchestration, you get type-safe, predictable, and composable agent pipelines that integrate seamlessly with the rest of your application stack. The key to success lies in writing clear field descriptions, choosing appropriate constraints, testing with edge cases, and handling validation failures gracefully. Start with simple schemas, iterate based on real outputs, and gradually introduce custom validators and nested models as your use cases demand. With these practices in place, you can confidently build agent systems that deliver reliable structured data every time.