CrewAI agents excel at reasoning, but their raw text outputs are brittle for downstream systems. The crewai pydantic task output model pattern solves this by forcing agents to emit validated JSON that your application can trust. This guide walks through defining models, wiring them into tasks, handling validation failures, and verifying the pipeline end to end.
Step 1: Define your output contract with Pydantic
Start by modeling what a successful task completion looks like. Use Pydantic v2 for its stricter validation and better error messages. Keep models flat where possible — nested structures increase token usage and failure rates.
# models/task_outputs.py
from pydantic import BaseModel, Field, field_validator
from typing import Literal
from datetime import datetime
class CompetitorAnalysis(BaseModel):
"""Structured output for competitive research task."""
company_name: str = Field(..., min_length=1, max_length=200)
pricing_tier: Literal["free", "starter", "professional", "enterprise"]
key_features: list[str] = Field(..., min_length=1, max_length=20)
market_position: Literal["leader", "challenger", "niche", "emerging"]
last_updated: datetime = Field(default_factory=datetime.utcnow)
@field_validator("key_features")
@classmethod
def dedupe_features(cls, v: list[str]) -> list[str]:
seen = set()
return [x for x in v if not (x in seen or seen.add(x))]
class ResearchReport(BaseModel):
"""Top-level container for the full research task."""
competitors: list[CompetitorAnalysis] = Field(..., min_length=1, max_length=10)
summary: str = Field(..., min_length=50, max_length=2000)
confidence_score: float = Field(..., ge=0.0, le=1.0)The validators catch common agent hallucinations: duplicate features, out-of-range confidence scores, missing required fields. The Literal enums constrain vocabulary so you don’t get “premium” when you expected “professional.”
Step 2: Create a task that returns the model
CrewAI’s Task class accepts an output_pydantic parameter. The agent receives the model’s JSON schema in its system prompt and must conform.
# tasks/research_tasks.py
from crewai import Task, Agent
from models.task_outputs import ResearchReport
def build_competitor_research_task(agent: Agent) -> Task:
return Task(
description=(
"Research the top 5 competitors in the AI code assistant market. "
"For each, identify: company name, pricing tier (free/starter/professional/enterprise), "
"3-5 key differentiating features, and market position (leader/challenger/niche/emerging). "
"Produce a concise executive summary and a confidence score (0.0-1.0) for your findings."
),
expected_output=(
"A JSON object matching the ResearchReport schema with competitors array, "
"summary string, and confidence_score float."
),
agent=agent,
output_pydantic=ResearchReport,
)Note the expected_output field — it’s not decorative. CrewAI includes this in the prompt context. Be specific about the schema shape so the agent doesn’t guess.
Step 3: Configure the agent for structured output
Agents need explicit instruction to emit JSON only. Use the system_template or role/goal/backstory to reinforce the contract.
# agents/research_agents.py
from crewai import Agent
from llm import get_llm # your LLM wrapper
def build_research_analyst() -> Agent:
return Agent(
role="Senior Market Research Analyst",
goal="Produce accurate, structured competitive intelligence reports",
backstory=(
"You are a methodical analyst who always outputs valid JSON matching "
"the provided schema. You never include commentary, markdown, or "
"explanatory text outside the JSON structure."
),
llm=get_llm(temperature=0.1), # low temperature for deterministic structure
verbose=True,
allow_delegation=False,
)Temperature matters here. Values above 0.3 noticeably increase schema violations. If your LLM wrapper supports response_format={"type": "json_object"}, enable it — CrewAI will pass this through to providers that honor it.
Step 4: Handle validation failures with a retry loop
Agents will violate the schema. Build a retry mechanism that feeds validation errors back to the agent.
# utils/structured_task_runner.py
from crewai import Crew, Task, Agent
from pydantic import ValidationError
from typing import Type, TypeVar
import logging
T = TypeVar("T", bound=BaseModel)
logger = logging.getLogger(__name__)
def run_structured_task(
task: Task,
agent: Agent,
output_model: Type[T],
max_retries: int = 3,
) -> T:
"""
Execute a task with Pydantic validation and automatic retry on failure.
"""
last_error: str | None = None
for attempt in range(1, max_retries + 1):
crew = Crew(
agents=[agent],
tasks=[task],
verbose=True,
)
try:
raw_result = crew.kickoff()
# CrewAI returns the parsed model when output_pydantic is set
if isinstance(raw_result, output_model):
logger.info(f"Task succeeded on attempt {attempt}")
return raw_result
# Fallback: parse manually if CrewAI returns string
parsed = output_model.model_validate_json(raw_result)
logger.info(f"Task succeeded on attempt {attempt} (manual parse)")
return parsed
except ValidationError as e:
last_error = str(e)
logger.warning(f"Validation failed (attempt {attempt}/{max_retries}): {e}")
# Inject error feedback into task description for retry
task.description = (
f"{task.description}\n\n"
f"PREVIOUS ATTEMPT FAILED VALIDATION:\n{last_error}\n\n"
"Correct the output to match the schema exactly. Output JSON only."
)
except Exception as e:
logger.error(f"Unexpected error on attempt {attempt}: {e}")
last_error = str(e)
raise RuntimeError(
f"Task failed after {max_retries} attempts. Last error: {last_error}"
)This pattern — re-injecting validation errors into the task description — is the most reliable way to get compliance without fine-tuning. The agent sees its own mistakes and self-corrects.
Step 5: Wire it together in a runnable script
# main.py
import logging
from agents.research_agents import build_research_analyst
from tasks.research_tasks import build_competitor_research_task
from models.task_outputs import ResearchReport
from utils.structured_task_runner import run_structured_task
logging.basicConfig(level=logging.INFO)
def main() -> ResearchReport:
analyst = build_research_analyst()
task = build_competitor_research_task(analyst)
report = run_structured_task(
task=task,
agent=analyst,
output_model=ResearchReport,
max_retries=3,
)
# At this point, `report` is a fully validated ResearchReport instance
print(f"Analyzed {len(report.competitors)} competitors")
print(f"Confidence: {report.confidence_score:.2f}")
print(f"Summary: {report.summary[:200]}...")
# Safe to serialize for APIs, databases, message queues
print(report.model_dump_json(indent=2))
return report
if __name__ == "__main__":
main()Run it: python main.py. You should see the validation loop in the logs, then a clean JSON dump.
Step 6: Verify success with automated checks
Don’t rely on eyeballing output. Add a verification script that runs in CI.
# tests/test_structured_output.py
import pytest
from models.task_outputs import ResearchReport, CompetitorAnalysis
from main import main
def test_research_report_structure():
"""Integration test: full pipeline produces valid schema."""
report = main()
# Schema validation already passed, but assert business invariants
assert isinstance(report, ResearchReport)
assert 1 <= len(report.competitors) <= 10
assert 0.0 <= report.confidence_score <= 1.0
assert len(report.summary) >= 50
# Verify each competitor has required fields populated
for comp in report.competitors:
assert isinstance(comp, CompetitorAnalysis)
assert comp.company_name.strip()
assert comp.pricing_tier in ("free", "starter", "professional", "enterprise")
assert 1 <= len(comp.key_features) <= 20
assert comp.market_position in ("leader", "challenger", "niche", "emerging")
def test_no_duplicate_features():
"""Custom validator works end-to-end."""
report = main()
for comp in report.competitors:
assert len(comp.key_features) == len(set(comp.key_features))Run with pytest tests/test_structured_output.py -v. This catches regressions when you swap models or update prompts.
Step 7: Optimize token usage for complex schemas
Large schemas consume context window and confuse smaller models. Two techniques help:
1. Split into multiple tasks with intermediate models
# Instead of one 50-field model, chain tasks:
# Task 1 -> CompetitorList (names only)
# Task 2 -> DetailedCompetitor (per company, parallelizable)
# Task 3 -> ResearchReport (aggregation)2. Use model_json_schema() to inject only required fields
# prompts/schema_hints.py
from models.task_outputs import ResearchReport
def get_schema_hint() -> str:
schema = ResearchReport.model_json_schema()
# Strip descriptions, examples, metadata — keep only type/required/enum
return json.dumps(schema, separators=(",", ":"), indent=None)Then reference {schema_hint} in your task description. This keeps the prompt lean while giving the agent exact constraints.
Common failure modes and fixes
| Symptom | Cause | Fix |
|---|---|---|
| Agent wraps JSON in markdown fences | Model trained on code blocks | Add “Output raw JSON only, no markdown” to backstory |
| Missing required fields | Schema too complex for context window | Split task, reduce fields, use model_json_schema() hint |
| Enum values hallucinated | Agent doesn’t see enum constraints | Use Literal in Pydantic; verify schema hint includes enum arrays |
| Confidence scores always 0.99 | Agent ignores ge/le validators |
Add explicit instruction: “Confidence must reflect actual certainty” |
| Retry loop exhausts without success | Temperature too high or model too small | Lower temperature, upgrade model, or simplify schema |
When to use this pattern
The crewai pydantic task output model approach shines when:
- Downstream systems consume the output programmatically (APIs, ETL, databases)
- You need audit trails with guaranteed schema compliance
- Multiple teams depend on the output contract
- Regulatory or compliance requirements mandate structured logs
Avoid it for exploratory tasks, creative writing, or human-facing chat where flexibility matters more than structure.
Production considerations
Observability: Log the raw agent output alongside the parsed model. Validation errors are debugging gold.
Cost control: Each retry is another LLM call. Cap retries at 2-3 and monitor the retry rate. If it exceeds 20%, simplify the schema or upgrade the model.
Versioning: Treat Pydantic models like API contracts. Version them (ResearchReportV1, ResearchReportV2) and migrate tasks explicitly. Never mutate a model in place that running tasks depend on.
Provider fallback: If you route through a gateway like n4n.ai that supports automatic fallback across 240+ models, configure the retry loop to escalate to a more capable model on repeated validation failures — cheaper models for the first attempt, stronger reasoning models for recovery.
The pattern is straightforward: define the contract, enforce it in the task, retry on violation, verify in CI. Once wired, you get typed, validated data structures instead of fragile string parsing. Your downstream services will thank you.