design-task

intent

task design is the primary lever for agent reliability. spend 80% of effort here, 20% on agent tuning. a well-designed task with a mediocre agent outperforms a brilliant agent hamstrung by vague instructions. this skill covers the anatomy of effective tasks (description + expected_output), single-purpose principle, dependency wiring, guardrails, and YAML configuration. use this when building new tasks, debugging agent output quality, or restructuring workflows.

inputs

• CrewAI project structure: agents.yaml and tasks.yaml config files (or inline Task() calls in crew.py)
• agent assignments: which agent owns each task (required in sequential/hierarchical crews)
• upstream task outputs: prior task results passed via context=[task_name] (for dependent tasks)
• llm instance: LLM model assigned to agents (used by LLM-based guardrails)
• external tools: SerperDevTool, ScrapeWebsiteTool, or custom tools (if task requires data fetching)
• pydantic models: optional, for structured output parsing (e.g., ResearchReport schema)
• variable inputs: runtime values passed to crew.kickoff(inputs={...}) that populate {variable} placeholders

procedure

1. write the task description (what + how + context + constraints)

description is the agent's instruction set. include:

• what to do: core action (research, analyze, write, etc.)
• how to do it: specific numbered steps or approach
• context: why this matters, what downstream tasks depend on it
• constraints: scope limits, things to avoid, acceptable sources only

example:

description: >
  Conduct thorough research about {topic} for the year {current_year}.
  Your research should:
  1. Identify the top 5 key trends and breakthroughs
  2. For each trend, find at least 2 credible sources
  3. Note any controversies or competing viewpoints
  4. Assess potential industry impact (high/medium/low)

  Focus on developments from the last 6 months.
  Do NOT include speculation or unverified claims.
  The output will feed into a report for {target_audience}.

2. write the expected_output (success criteria as human-readable string)

expected_output tells the agent what "done" looks like. specify:

• format: bullet points, paragraphs, JSON, markdown table, structured sections
• structure: headings, order, number of sections
• length: approximate word count or number of items
• quality markers: citations required, confidence levels, specific fields

example:

expected_output: >
  A structured research brief with 5 sections, one per trend.
  Each section includes: trend name, 2-3 paragraph summary,
  source citations, impact assessment (high/medium/low),
  and a confidence level for your findings.

do not reference pydantic class names in expected_output. keep it human-readable so the agent understands what fields to populate.

3. assign agent and declare single purpose

each task must have exactly one objective. no "god tasks" combining research + writing + editing.

research_task = Task(
    description="...",
    expected_output="...",
    agent=researcher,  # required in sequential/hierarchical crews
)

if your initial task covers multiple operations (research + write + edit), split into separate tasks with explicit dependencies:

• research_task (agent: researcher)
• writing_task (agent: writer, context=[research_task])
• editing_task (agent: editor, context=[writing_task])

4. configure output format (optional but recommended)

choose one or combine:

a) structured output with pydantic

from pydantic import BaseModel

class ResearchReport(BaseModel):
    trends: list[str]
    confidence: float
    sources: list[str]

research_task = Task(
    description="...",
    expected_output="A structured report with trends, confidence score, and sources.",
    agent=researcher,
    output_pydantic=ResearchReport,  # model goes here, NOT in expected_output
)

access via result.pydantic or all_outputs[0].pydantic.

b) file output

Task(
    description="...",
    expected_output="...",
    agent=writer,
    output_file="output/report.md",
    create_directory=True,  # auto-create if missing
)

c) markdown formatting

Task(
    description="...",
    expected_output="...",
    agent=writer,
    markdown=True,  # auto-instructs agent to use proper headers, lists, emphasis
)

5. set up task dependencies (if needed)

sequential process (default)

in Process.sequential, tasks run in order. each task automatically receives all prior task outputs as implicit context. do not use context= unless you need non-linear flow:

# task_b auto-receives task_a output
task_a = Task(description="...", agent=researcher)
task_b = Task(description="...", agent=analyst)
# implicit: task_b sees task_a's result

explicit context for non-linear dependencies

# task_c depends on task_a but NOT task_b
task_c = Task(
    description="...",
    agent=analyst,
    context=[task_a],  # only receives task_a, skips task_b
)

diamond dependency pattern

task_a = Task(...)                          # entry point
task_b = Task(..., context=[task_a])        # depends on a
task_c = Task(..., context=[task_a])        # also depends on a (parallel-friendly)
task_d = Task(..., context=[task_b, task_c])  # depends on both b and c

6. add tools to task (if data fetching required)

task-level tools override agent defaults for that specific task:

from crewai_tools import SerperDevTool, ScrapeWebsiteTool

Task(
    description="Search for and scrape the top 5 articles about {topic}...",
    expected_output="...",
    agent=researcher,
    tools=[SerperDevTool(), ScrapeWebsiteTool()],
)

use task-level tools when:

• task needs tools the agent doesn't normally have
• you want to restrict an agent to specific tools for this task only
• different tasks by same agent need different tool sets

7. add guardrails (quality control)

guardrails validate output before it passes downstream. if validation fails, agent retries (up to guardrail_max_retries).

function-based guardrails (deterministic checks)

def validate_word_count(output) -> tuple[bool, Any]:
    """ensure output is between 500-2000 words."""
    word_count = len(output.raw.split())
    if word_count < 500:
        return (False, f"Output too short ({word_count} words). Expand to at least 500 words.")
    if word_count > 2000:
        return (False, f"Output too long ({word_count} words). Condense to under 2000 words.")
    return (True, output)

Task(
    description="...",
    expected_output="...",
    agent=writer,
    guardrail=validate_word_count,
    guardrail_max_retries=3,
)

return format: (bool, Any) where bool is pass/fail, Any is result (on success) or error message (on failure).

llm-based guardrails (subjective quality checks)

Task(
    description="...",
    expected_output="...",
    agent=researcher,
    guardrail="Verify the output contains at least 3 source citations and no speculative claims.",
)

string guardrails use the agent's llm to evaluate. good for subjective checks.

chaining multiple guardrails

Task(
    description="...",
    expected_output="...",
    agent=researcher,
    guardrails=[
        validate_word_count,  # function: deterministic
        validate_no_pii,      # function: deterministic
        "Ensure tone is professional and suitable for business audience.",  # llm: subjective
    ],
    guardrail_max_retries=3,
)

guardrails execute sequentially. each receives output of previous guardrail.

8. configure async execution (optional)

Task(
    description="...",
    expected_output="...",
    agent=researcher,
    async_execution=True,  # runs in background, does not block next task
)

use for tasks that can run in parallel. downstream tasks will wait for async result if you add context=[async_task].

9. enable human review (optional)

Task(
    description="...",
    expected_output="...",
    agent=writer,
    human_input=True,  # agent pauses and waits for human feedback before finalizing
)

use for critical outputs requiring human approval. do not conflate with conversational flows (each user message in a flow should call handle_turn(), not spawn a new crew task).

10. add callbacks (optional)

def log_completion(output):
    print(f"Task completed: {output.description[:50]}...")
    save_to_database(output.raw)

Task(
    description="...",
    expected_output="...",
    agent=researcher,
    callback=log_completion,  # called after task completion
)

11. use YAML config (recommended for multi-task crews)

tasks.yaml:

research_task:
  description: >
    Conduct thorough research about {topic} for {current_year}.
    Identify key trends, breakthrough technologies,
    and potential industry impacts.
    Focus on the last 6 months of developments.
  expected_output: >
    A structured research brief with 5 sections.
    Each section: trend name, 2-3 paragraph summary,
    source citations, and impact assessment.
  agent: researcher

analysis_task:
  description: >
    Analyze the research findings and create actionable recommendations
    for {target_audience}.
  expected_output: >
    A prioritized list of 5 recommendations with:
    rationale, estimated effort, and expected impact.
  agent: analyst
  context:
    - research_task

report_task:
  description: >
    Compile a final report combining research and analysis for {target_audience}.
  expected_output: >
    A polished markdown report with executive summary,
    detailed findings, recommendations, and appendices.
  agent: writer
  output_file: output/report.md

wire in crew.py:

@CrewBase
class ResearchCrew:
    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    @task
    def research_task(self) -> Task:
        return Task(config=self.tasks_config["research_task"])

    @task
    def analysis_task(self) -> Task:
        return Task(
            config=self.tasks_config["analysis_task"],
            context=[self.research_task()],
        )

    @task
    def report_task(self) -> Task:
        return Task(
            config=self.tasks_config["report_task"],
            output_file="output/report.md",
        )

critical: method name (def research_task) must match YAML key (research_task:).

12. variable interpolation for reusable tasks

use {variable} placeholders in YAML:

research_task:
  description: >
    Research {topic} trends for {current_year},
    targeting {target_audience}.

replace at runtime:

crew.kickoff(inputs={
    "topic": "AI Agents",
    "current_year": "2025",
    "target_audience": "developers",
})

common mistakes: missing variable in inputs (literal {variable} appears in prompt), using {{ }} jinja2 syntax (crewaI uses single braces), unused variables in inputs (silently ignored).

decision points

should i split this task into multiple tasks?

if the description includes multiple distinct operations (research AND write AND edit), split it. one task = one objective. sequential flow passes outputs automatically between tasks.

do i need structured output (pydantic)?

yes, if downstream code needs to parse and access specific fields (e.g., trends list, confidence score). use output_pydantic=ResearchReport. if task output is just text for human consumption, skip it.

should i use context= explicitly in sequential process?

only if you need non-linear dependencies. in Process.sequential, each task auto-receives all prior outputs. use context=[task_a] only if task_c needs task_a but NOT task_b.

how strict should my expected_output be?

specific but achievable. if expected_output is impossible to match, agent loops and wastes retries. specify format (sections, headers, word count), not overly rigid formatting rules. lower guardrail_max_retries=1 to fail faster on impossible criteria.

should i use llm-based or function-based guardrails?

function-based for deterministic checks (word count, field presence, PII detection). llm-based for subjective checks (tone, professionalism, factual accuracy). mix both for comprehensive validation.

do i add tools to the task or the agent?

task-level tools override agent defaults. use if the task needs unique tools or you want to restrict an agent's tool set for a specific task. otherwise, assign tools to the agent config.

when do i use async_execution=True?

for tasks that can run in parallel without blocking the next task. pair with context=[async_task] on downstream tasks to wait for result. skip if task output feeds immediately into the next task.

should i use human_input=True or a conversational flow?

human_input=True is for approving/correcting task output before it flows downstream in a batch workflow. do not use for conversational chatbots. in flows, each user message calls handle_turn(message, session_id=...), not crew.kickoff().

output contract

a well-designed task produces:

• task output object with properties: raw (string), pydantic (if output_pydantic set), description, agent
• file output (if output_file specified): markdown, json, or text saved to disk
• structured data (if output_pydantic set): parsed pydantic model accessible via result.pydantic or all_outputs[N].pydantic
• validation status: guardrails pass or retry loop exhausted
• execution metadata: completion time, agent used, tool calls made

for crews with multiple tasks, result object contains:

• result.raw: raw output of last task
• result.pydantic: parsed last task output (if output_pydantic set)
• result.tasks_output: list of all TaskOutput objects
• result.tasks_output[0].pydantic: first task's parsed output

outcome signal

the task worked when:

• agent produces output matching expected_output format and structure (sections, headers, fields, quality markers present)
• guardrails pass without retry loops
• downstream tasks receive complete, usable context from prior task outputs
• file output appears at specified path with correct format (if output_file set)
• structured output parses cleanly into pydantic model without type errors (if output_pydantic set)
• human review feedback applied successfully (if human_input=True)
• workflow dependencies respect order (upstream tasks complete before downstream tasks read their outputs)
• no vague or fabricated data in output (agent followed constraints, used real sources, avoided speculation)
• total execution time stays reasonable (no retry loops spinning indefinitely on impossible criteria)

test by running crew.kickoff() with sample inputs and inspecting result.raw, result.tasks_output, and file outputs. verify each section, citation, field, and quality marker is present before passing to downstream tasks or users.

---

credits: original guide by crewaiinc (https://skills.sh). enriched per implexa quality standards with explicit decision logic, edge cases, validation patterns, and outcome verification steps.