Orchestrating Complex Workflows

Sequential Process

Tasks execute in order, each receiving the output of the previous:

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,
)

Hierarchical Process

A manager agent delegates and coordinates:

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4o"),
)

Task Dependencies

Define explicit dependencies between tasks:

analysis_task = Task(
    description="Analyze the research data",
    expected_output="Key insights and trends",
    agent=analyst,
    context=[research_task],  # Depends on research_task output
)

Conditional Logic

Use task callbacks and conditional routing:

def check_quality(output):
    if "insufficient data" in output.raw.lower():
        return "retry"
    return "continue"

research_task = Task(
    description="Research AI trends",
    expected_output="Comprehensive report",
    agent=researcher,
    callback=check_quality,
)

Workflow Patterns

• Pipeline: A → B → C (sequential processing)
• Fan-out: A → [B, C, D] (parallel research)
• Fan-in: [B, C, D] → E (synthesis)
• Loop: A → B → A (iterative refinement)

Common Pitfall: Not Passing Context Between Dependent Tasks

By default, each task receives only its own description and the agent's instructions. If task B needs the output of task A, you must explicitly pass it using the context parameter. Forgetting to do this causes agents to work from incomplete information or hallucinate data that should have come from a previous step:

# Wrong: analysis_task has no access to research results
analysis_task = Task(description="Analyze the research.", agent=analyst)

# Correct: analysis_task receives research_task's output
analysis_task = Task(
    description="Analyze the research findings below.",
    context=[research_task],  # Output of research_task is injected into the context
    agent=analyst,
)

Best Practice: Use Async Kickoff for Long Crews

Long-running crews (10+ minutes) should run asynchronously so your application can respond to other requests while the crew works:

import asyncio

async def run_crew():
    result = await crew.kickoff_async()
    return result

# Store result in database when done
asyncio.create_task(run_crew())

Named Product Reference: CrewAI Enterprise

CrewAI offers an enterprise cloud platform that provides hosted crew execution, visual workflow building, observability dashboards, and scheduled crew runs. For teams that need to run crews on a schedule (e.g., daily content generation, weekly competitive analysis), the enterprise platform eliminates the need to manage a task queue and worker infrastructure.