CrewAI Multi-Agents · Lesson 9 of 16
Defining Tasks with Expected Output
Tasks Are the Unit of Work
In CrewAI, an Agent is who does the work. A Task is what work is done. The quality of your tasks — specifically how clearly you define description and expected_output — determines the quality of your crew's output more than any other single factor.
This lesson covers every Task parameter, explains why each one matters, and shows a complete example of a research-to-writing pipeline with four tasks.
Full Task Constructor
Python
from crewai import Task, Agent
from pydantic import BaseModel
# Assume agents are already defined
researcher: Agent = ...
writer: Agent = ...
task = Task(
# --- Core ---
description=(
"Research the clinical pharmacology of tirzepatide, including "
"mechanism of action, PK/PD properties, and approved indications."
),
expected_output=(
"A structured summary with four sections: "
"1) Mechanism of Action (2-3 paragraphs explaining the dual GIP/GLP-1 agonism), "
"2) Pharmacokinetics (half-life, volume of distribution, metabolism), "
"3) Pharmacodynamics (dose-response, effects on HbA1c and body weight), "
"4) Approved Indications (by region: US, EU, UK). "
"Each section must be at least 100 words."
),
# --- Assignment ---
agent=researcher,
# --- Dependencies ---
context=[], # List of Task instances whose output to inject
# --- Output options ---
output_file="tirzepatide_research.md", # Save output to this file
output_pydantic=None, # Parse output into this Pydantic model
output_json=False, # Force JSON output
# --- Execution ---
async_execution=False, # Run concurrently with other tasks?
# --- Callbacks ---
callback=None, # Function called when task completes
)description: What to Do
The description is the task prompt. It should be clear, specific, and actionable.
What Makes a Good Description
1. Specify the scope
Python
# Too broad
description="Research tirzepatide."
# Scoped — agent knows what to look for
description=(
"Research the clinical trial evidence for tirzepatide in type 2 diabetes. "
"Focus specifically on the SURPASS trial program (SURPASS-1 through SURPASS-5). "
"Collect: primary endpoints (HbA1c reduction, body weight change), "
"key safety findings (GI events, thyroid findings), "
"and comparator outcomes where a comparator arm was included."
)2. Tell the agent what sources to prioritize
Python
description=(
"Search for the latest evidence on GLP-1 receptor agonists in heart failure. "
"Prioritize: randomized controlled trials published after 2023, "
"ACC/AHA guidelines, and ESC Heart Failure guidelines. "
"Note the quality of evidence (RCT vs observational) for each finding."
)3. Use input variables for reusable tasks
Python
description=(
"Research the clinical evidence for {drug} in treating {indication}. "
"Focus on Phase 3 RCTs. Include efficacy and safety data."
)
# At kickoff time:
crew.kickoff(inputs={"drug": "semaglutide", "indication": "obesity"})expected_output: The Contract
expected_output is the single most impactful parameter in CrewAI. It is injected into the agent's prompt as the definition of a successfully completed task. The LLM uses it to know when it is done and what form the output should take.
The Cost of Vague Expected Output
Python
# Vague — agent decides what "good" looks like
expected_output="A useful summary."
# Result: inconsistent length, format, depth. Useless for downstream processing.
# Specific — agent has a clear target
expected_output=(
"A markdown document with exactly four numbered sections: "
"1) Clinical Background (one paragraph, max 80 words), "
"2) Efficacy Evidence (a markdown table with columns: Trial, Comparator, "
" HbA1c Reduction, Weight Loss, Sample Size), "
"3) Safety Profile (bullet list of adverse events above 5% incidence, "
" with incidence percentage for each), "
"4) Clinical Recommendation (two sentences: one summary, one for whom to consider). "
"Total length: 300-500 words. No promotional language."
)The specific version will produce a consistent, structured document every time. The vague version will produce a different format every run.
context: Receiving Another Task's Output
The context parameter is how tasks pass data to each other. When you set context=[task_a], CrewAI prepends the output of task_a to the agent's context window before it starts work on the current task.
Python
# Task A: research
research_task = Task(
description="Research the mechanism of action of SGLT2 inhibitors.",
expected_output="A technical summary of the mechanism, including renal glucose handling.",
agent=pharmacologist,
)
# Task B: writing — receives task A's output
writing_task = Task(
description=(
"Write a patient-friendly explanation of how SGLT2 inhibitors work. "
"Base your explanation on the technical research provided."
),
expected_output=(
"A 150-word patient-facing explanation. "
"No medical jargon. Use an analogy to explain the mechanism."
),
agent=medical_writer,
context=[research_task], # <-- research output prepended to writing agent's context
)You can pass multiple task outputs into a single task:
Python
synthesis_task = Task(
description="Synthesize the research and the safety analysis into a clinical brief.",
expected_output="A structured clinical brief covering efficacy and safety.",
agent=lead_writer,
context=[research_task, safety_analysis_task], # both outputs provided
)output_file: Saving Results to Disk
The output_file parameter saves the task's output to a file when the task completes:
Python
final_report_task = Task(
description="Write the final pharmacovigilance safety report.",
expected_output="A formatted Word-compatible markdown safety report.",
agent=report_writer,
output_file="safety_report_2026_Q1.md", # Saved automatically on completion
)This is useful for:
- Audit trails in regulated environments
- Passing outputs to non-CrewAI systems
- Human review of individual task outputs
output_pydantic: Structured Output for Downstream Code
When downstream code needs to parse task output, use output_pydantic to force the LLM output into a Pydantic model:
Python
from pydantic import BaseModel, Field
from typing import List, Optional
class ClinicalTrialResult(BaseModel):
trial_name: str = Field(description="Name of the clinical trial")
comparator: str = Field(description="Comparator arm or 'placebo'")
hba1c_reduction: float = Field(description="HbA1c reduction in percentage points")
weight_change_kg: float = Field(description="Body weight change in kilograms")
sample_size: int = Field(description="Number of participants in the trial")
class EfficacySummary(BaseModel):
drug_name: str
trials: List[ClinicalTrialResult]
overall_conclusion: str
efficacy_task = Task(
description=(
"Extract efficacy data from the SURPASS trial program for tirzepatide. "
"Include SURPASS-1, SURPASS-2, SURPASS-3, SURPASS-4, and SURPASS-5."
),
expected_output="Structured efficacy data from the SURPASS trials.",
agent=data_extractor,
output_pydantic=EfficacySummary, # Force output into this model
)
crew = Crew(
agents=[data_extractor],
tasks=[efficacy_task],
process=Process.sequential,
)
result = crew.kickoff()
# Access structured output
summary: EfficacySummary = result.pydantic
for trial in summary.trials:
print(f"{trial.trial_name}: HbA1c -{trial.hba1c_reduction}%, Weight {trial.weight_change_kg}kg")async_execution: Parallel Tasks
Tasks with async_execution=True start immediately without waiting for the previous task to complete. This is covered in depth in the async tasks lesson. The key point here is the declaration:
Python
task_a = Task(
description="Research Drug A.",
expected_output="Research summary for Drug A.",
agent=researcher_1,
async_execution=True, # Runs concurrently
)
task_b = Task(
description="Research Drug B.",
expected_output="Research summary for Drug B.",
agent=researcher_2,
async_execution=True, # Also runs concurrently
)
# This task waits for both async tasks above
synthesis_task = Task(
description="Compare Drug A and Drug B based on the research.",
expected_output="A head-to-head comparison.",
agent=lead_analyst,
context=[task_a, task_b], # Waits for both to complete
async_execution=False, # Synchronization point
)callback: Monitoring Task Completion
Python
from crewai.tasks import TaskOutput
def on_task_complete(output: TaskOutput) -> None:
print(f"Task completed: {output.description[:50]}...")
print(f"Output length: {len(output.raw)} characters")
# Could: log to database, send Slack notification, trigger downstream process
research_task = Task(
description="Research the safety profile of rivaroxaban.",
expected_output="A structured safety summary.",
agent=safety_analyst,
callback=on_task_complete,
)Complete Example: Research → Analysis → Writing → Review
Python
from crewai import Agent, Task, Crew, Process, LLM
from crewai_tools import SerperDevTool
from pydantic import BaseModel, Field
from typing import List
llm = LLM(model="gpt-4o")
search_tool = SerperDevTool()
# --- Agents ---
researcher = Agent(
role="Clinical Research Analyst",
goal="Find high-quality clinical evidence for pharmaceutical questions",
backstory=(
"You are a clinical pharmacologist with expertise in diabetes therapy. "
"You evaluate evidence by GRADE criteria and always note evidence quality."
),
tools=[search_tool],
llm=llm,
verbose=True,
)
analyst = Agent(
role="Clinical Data Analyst",
goal="Extract and structure quantitative data from clinical research",
backstory=(
"You specialize in extracting clean, structured data from clinical summaries. "
"You produce tables and metrics, not prose."
),
llm=llm,
verbose=True,
)
writer = Agent(
role="Medical Communications Specialist",
goal="Write clear, audience-appropriate medical content",
backstory=(
"You have 10 years of medical writing experience. "
"You write for HCP audiences, avoiding promotional language."
),
llm=llm,
verbose=True,
)
reviewer = Agent(
role="Medical Affairs Reviewer",
goal="Ensure accuracy and compliance of all medical communications",
backstory=(
"You are a physician with medical affairs experience. "
"You apply medical-legal-regulatory review standards. "
"You approve only what is fully supported by the provided evidence."
),
llm=llm,
verbose=True,
)
# --- Pydantic model for structured analysis ---
class EfficacyDataPoint(BaseModel):
endpoint: str = Field(description="Clinical endpoint measured")
value: str = Field(description="Measured value with units")
comparator: str = Field(description="What this is compared against")
class StructuredAnalysis(BaseModel):
drug: str
indication: str
efficacy_data: List[EfficacyDataPoint]
key_safety_signals: List[str]
# --- Tasks ---
research_task = Task(
description=(
"Research the clinical efficacy and safety of {drug} for {indication}. "
"Focus on Phase 3 trial data and FDA label information."
),
expected_output=(
"A detailed research summary covering: "
"key Phase 3 trials (names, endpoints, results), "
"safety profile (adverse events above 5% incidence with rates), "
"FDA approval status and approved indications."
),
agent=researcher,
)
analysis_task = Task(
description=(
"Extract structured quantitative data from the research summary. "
"Organize efficacy endpoints and safety signals into a clean format."
),
expected_output="Structured clinical data for downstream use.",
agent=analyst,
context=[research_task],
output_pydantic=StructuredAnalysis,
)
writing_task = Task(
description=(
"Write a 400-word HCP-facing clinical brief on {drug} for {indication} "
"using the research and structured analysis provided. "
"Target audience: specialists. No promotional language."
),
expected_output=(
"A 400-word clinical brief with: Title, Clinical Background, "
"Efficacy Summary (with specific trial references), "
"Safety Profile, and Clinical Use Considerations."
),
agent=writer,
context=[research_task, analysis_task],
output_file="{drug}_clinical_brief.md",
)
review_task = Task(
description=(
"Review the clinical brief for medical accuracy and compliance. "
"Verify all claims against the provided research. "
"Apply medical-legal-regulatory standards."
),
expected_output=(
"A review decision: Approved / Revise Required / Rejected. "
"If not Approved: numbered list of required changes, "
"each citing the specific unsupported claim and required correction."
),
agent=reviewer,
context=[research_task, writing_task],
)
# --- Crew ---
crew = Crew(
agents=[researcher, analyst, writer, reviewer],
tasks=[research_task, analysis_task, writing_task, review_task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff(inputs={
"drug": "empagliflozin",
"indication": "heart failure with reduced ejection fraction",
})
print("=== REVIEW DECISION ===")
print(result.raw)Summary
| Parameter | Required | Purpose |
|-----------|----------|---------|
| description | Yes | What the agent should do |
| expected_output | Yes | What a successful completion looks like |
| agent | Yes | Which agent performs the task |
| context | No | Which previous task outputs to inject |
| output_file | No | Save task output to a file |
| output_pydantic | No | Parse output into a Pydantic model |
| output_json | No | Force JSON-formatted output |
| async_execution | No | Run this task concurrently |
| callback | No | Function to call when task completes |
The most important investment you can make in a CrewAI system is writing precise expected_output definitions. Everything else is configuration.