Sequential vs Hierarchical Processes

1"""
2Sequential Process Flow:
3
4Task 1 ──▶ Task 2 ──▶ Task 3 ──▶ Output
5  │          │          │
6  ▼          ▼          ▼
7Agent 1   Agent 2   Agent 3
8
9Key characteristics:
10- Tasks execute in defined order
11- Each task's output becomes context for next task
12- Deterministic flow - same order every time
13- Simple to understand and debug
14"""
15
16from crewai import Crew, Agent, Task, Process
17from langchain_openai import ChatOpenAI
18
19llm = ChatOpenAI(model="gpt-4o")
20
21# Define agents
22researcher = Agent(
23    role="Researcher",
24    goal="Gather comprehensive information",
25    backstory="Expert at finding reliable information",
26    llm=llm
27)
28
29analyst = Agent(
30    role="Analyst",
31    goal="Analyze and interpret data",
32    backstory="Expert at extracting insights",
33    llm=llm
34)
35
36writer = Agent(
37    role="Writer",
38    goal="Create clear, engaging content",
39    backstory="Expert at technical writing",
40    llm=llm
41)
42
43# Define tasks in order
44research_task = Task(
45    description="Research AI agent frameworks",
46    expected_output="Research findings",
47    agent=researcher
48)
49
50analysis_task = Task(
51    description="Analyze the research findings",
52    expected_output="Analysis report",
53    agent=analyst,
54    context=[research_task]  # Uses output from research
55)
56
57writing_task = Task(
58    description="Write a blog post based on analysis",
59    expected_output="Blog post",
60    agent=writer,
61    context=[research_task, analysis_task]  # Uses both previous outputs
62)
63
64# Create sequential crew
65crew = Crew(
66    agents=[researcher, analyst, writer],
67    tasks=[research_task, analysis_task, writing_task],
68    process=Process.sequential,
69    verbose=True
70)
71
72result = crew.kickoff()

1"""
2When Sequential Falls Short:
3
41. No parallelism - tasks run one at a time
52. Fixed order - can't adapt to changing needs
63. No dynamic routing - can't skip or repeat tasks
74. Bottlenecks - slow task blocks all subsequent tasks
8"""
9
10# Example: Sequential bottleneck
11slow_research = Task(
12    description="Comprehensive research (takes 5 minutes)",
13    expected_output="Research",
14    agent=researcher
15)
16
17quick_analysis = Task(
18    description="Quick analysis (takes 30 seconds)",
19    expected_output="Analysis",
20    agent=analyst,
21    context=[slow_research]
22)
23
24# Analysis must wait 5 minutes even though it's quick
25# In hierarchical, manager could start other work while waiting

1"""
2Hierarchical Process Flow:
3
4                ┌─────────────┐
5                │   Manager   │
6                │   (LLM)     │
7                └──────┬──────┘
8                       │
9         ┌─────────────┼─────────────┐
10         │             │             │
11         ▼             ▼             ▼
12    ┌─────────┐  ┌─────────┐  ┌─────────┐
13    │ Agent 1 │  │ Agent 2 │  │ Agent 3 │
14    └─────────┘  └─────────┘  └─────────┘
15
16Key characteristics:
17- Manager LLM orchestrates all agents
18- Dynamic task assignment based on context
19- Can delegate, re-delegate, and adapt
20- More flexible but less predictable
21"""
22
23from crewai import Crew, Process
24from langchain_openai import ChatOpenAI
25
26# Create hierarchical crew with manager
27hierarchical_crew = Crew(
28    agents=[researcher, analyst, writer],
29    tasks=[research_task, analysis_task, writing_task],
30    process=Process.hierarchical,
31    manager_llm=ChatOpenAI(model="gpt-4o"),  # Required for hierarchical
32    verbose=True
33)
34
35result = hierarchical_crew.kickoff()

1from crewai import Agent, Crew, Process
2
3# Define a custom manager agent
4manager = Agent(
5    role="Project Manager",
6    goal="Coordinate the team to deliver high-quality results efficiently",
7    backstory="""You are an experienced project manager who excels at
8    breaking down complex projects, assigning the right people to tasks,
9    and ensuring quality deliverables. You know when to push for speed
10    and when to prioritize quality.""",
11    allow_delegation=True,
12    verbose=True
13)
14
15# Use custom manager
16crew_with_custom_manager = Crew(
17    agents=[researcher, analyst, writer],
18    tasks=[research_task, analysis_task, writing_task],
19    process=Process.hierarchical,
20    manager_agent=manager,  # Custom manager instead of manager_llm
21    verbose=True
22)
23
24# The manager agent will:
25# 1. Review all tasks and agent capabilities
26# 2. Decide which agent handles each task
27# 3. Coordinate handoffs and context sharing
28# 4. Ensure quality and completeness

1"""
2Hierarchical Limitations:
3
41. Higher cost - Manager LLM adds token usage
52. Less predictable - Dynamic routing varies
63. More complex - Harder to debug
74. Manager dependency - Manager quality affects all
85. Potential loops - Manager might reassign repeatedly
9"""
10
11# Cost comparison example
12"""
13Sequential:
14- Agent 1: ~1000 tokens
15- Agent 2: ~1000 tokens
16- Agent 3: ~1000 tokens
17Total: ~3000 tokens
18
19Hierarchical:
20- Manager decisions: ~500 tokens × N decisions
21- Agent 1: ~1000 tokens
22- Agent 2: ~1000 tokens
23- Agent 3: ~1000 tokens
24- Manager coordination: ~500 tokens
25Total: ~4500+ tokens (50% more)
26"""

1"""
2Use SEQUENTIAL when:
3
4✓ Tasks have clear, fixed order
5✓ Each task depends on previous output
6✓ Predictability is important
7✓ Cost optimization is priority
8✓ Simple, linear workflows
9✓ Debugging/tracing is important
10
11Examples:
12- Research → Analyze → Write pipeline
13- Extract → Transform → Load (ETL)
14- Draft → Review → Publish workflow
15"""
16
17"""
18Use HIERARCHICAL when:
19
20✓ Task assignment should be dynamic
21✓ Multiple agents could handle same task
22✓ Need intelligent error recovery
23✓ Complex, non-linear workflows
24✓ Tasks may need re-ordering
25✓ Quality control is critical
26
27Examples:
28- Customer support with specialists
29- Complex problem-solving teams
30- Projects with unclear requirements
31- Teams with overlapping skills
32"""

1"""
2Pattern: Hierarchical coordination of sequential sub-teams
3
4                    ┌──────────────┐
5                    │   Manager    │
6                    └──────┬───────┘
7                           │
8         ┌─────────────────┼─────────────────┐
9         │                 │                 │
10         ▼                 ▼                 ▼
11   ┌───────────┐    ┌───────────┐    ┌───────────┐
12   │ Research  │    │ Analysis  │    │ Writing   │
13   │   Team    │    │   Team    │    │   Team    │
14   │ (Seq.)    │    │ (Seq.)    │    │ (Seq.)    │
15   └───────────┘    └───────────┘    └───────────┘
16"""
17
18# Sub-crew for research
19research_crew = Crew(
20    agents=[primary_researcher, fact_checker],
21    tasks=[initial_research, verification],
22    process=Process.sequential
23)
24
25# Sub-crew for analysis
26analysis_crew = Crew(
27    agents=[data_analyst, strategic_analyst],
28    tasks=[data_analysis, strategic_analysis],
29    process=Process.sequential
30)
31
32# Main crew coordinates sub-crews
33# (Implementation depends on specific needs)

1from crewai import Agent, Task, Crew, Process
2
3def create_research_team():
4    """Create a specialized research sub-team."""
5
6    searcher = Agent(
7        role="Web Searcher",
8        goal="Find relevant sources",
9        backstory="Expert at web research"
10    )
11
12    reader = Agent(
13        role="Content Reader",
14        goal="Extract key information from sources",
15        backstory="Expert at reading comprehension"
16    )
17
18    search_task = Task(
19        description="Search for information on {topic}",
20        expected_output="List of relevant sources",
21        agent=searcher
22    )
23
24    read_task = Task(
25        description="Extract key points from sources",
26        expected_output="Summary of findings",
27        agent=reader,
28        context=[search_task]
29    )
30
31    return Crew(
32        agents=[searcher, reader],
33        tasks=[search_task, read_task],
34        process=Process.sequential
35    )
36
37
38def run_multi_team_project(topic: str):
39    """Coordinate multiple teams on a project."""
40
41    # Run research team
42    research_crew = create_research_team()
43    research_result = research_crew.kickoff(inputs={"topic": topic})
44
45    # Pass results to analysis team
46    analysis_agent = Agent(
47        role="Analyst",
48        goal="Analyze research findings",
49        backstory="Expert analyst"
50    )
51
52    analysis_task = Task(
53        description=f"Analyze these findings: {research_result}",
54        expected_output="Analysis report",
55        agent=analysis_agent
56    )
57
58    analysis_crew = Crew(
59        agents=[analysis_agent],
60        tasks=[analysis_task]
61    )
62
63    return analysis_crew.kickoff()

1from crewai import Crew, Process
2
3
4def create_adaptive_crew(complexity: str, agents: list, tasks: list):
5    """Select process based on task complexity."""
6
7    if complexity == "simple":
8        # Linear tasks - use sequential
9        return Crew(
10            agents=agents,
11            tasks=tasks,
12            process=Process.sequential,
13            verbose=True
14        )
15
16    elif complexity == "complex":
17        # Dynamic coordination needed - use hierarchical
18        return Crew(
19            agents=agents,
20            tasks=tasks,
21            process=Process.hierarchical,
22            manager_llm=ChatOpenAI(model="gpt-4o"),
23            verbose=True
24        )
25
26    else:
27        # Default to sequential for safety
28        return Crew(
29            agents=agents,
30            tasks=tasks,
31            process=Process.sequential,
32            verbose=True
33        )
34
35
36# Usage based on task analysis
37def analyze_complexity(tasks: list) -> str:
38    """Analyze task complexity to choose process."""
39
40    # Simple heuristics
41    if len(tasks) <= 3:
42        return "simple"
43
44    # Check for task dependencies
45    has_complex_deps = any(
46        len(getattr(t, 'context', []) or []) > 2
47        for t in tasks
48    )
49
50    if has_complex_deps:
51        return "complex"
52
53    return "simple"
54
55
56# Automatic process selection
57complexity = analyze_complexity(tasks)
58crew = create_adaptive_crew(complexity, agents, tasks)