CrewAI hierarchical process tutorial: manager LLM setup

The CrewAI hierarchical process manager llm setup lets you delegate task planning to a dedicated manager agent instead of hardcoding execution order. This tutorial walks through building a working hierarchical crew from scratch: defining specialized agents, configuring the manager LLM, wiring tasks with dependencies, and running the crew end to end. You’ll see exactly what each component does and what output to expect at every checkpoint.

Prerequisites

Python 3.10+
An OpenAI API key (or compatible endpoint)
crewai and crewai-tools installed

pip install crewai crewai-tools python-dotenv

Create a .env file in your project root:

OPENAI_API_KEY=sk-...
# Optional: if you route through a gateway like n4n.ai
# OPENAI_BASE_URL=https://api.n4n.ai/v1

Project structure

hierarchical_crew/
├── main.py
├── agents.py
├── tasks.py
└── tools.py

Define a custom tool

Start with a simple tool so agents have something concrete to work with. This keeps the example focused on the hierarchical mechanics rather than tool complexity.

# tools.py
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field


class MarketSizeInput(BaseModel):
    market: str = Field(..., description="Market segment to estimate")


class MarketSizeTool(BaseTool):
    name: str = "market_size_estimator"
    args_schema: Type[BaseModel] = MarketSizeInput

    def _run(self, market: str) -> str:
        # Stubbed data — replace with real API calls in production
        estimates = {
            "ai coding assistants": {"tam": 12.4, "sam": 3.1, "som": 0.4},
            "devops automation": {"tam": 8.7, "sam": 2.2, "som": 0.3},
            "llm observability": {"tam": 4.2, "sam": 1.1, "som": 0.15},
        }
        key = market.lower()
        if key in estimates:
            e = estimates[key]
            return f"TAM: ${e['tam']}B, SAM: ${e['sam']}B, SOM: ${e['som']}B"
        return f"No data for '{market}'. Known markets: {list(estimates.keys())}"

Create specialized agents

Each agent gets a focused role, backstory, and the tools it needs. The manager agent will coordinate them.

# agents.py
from crewai import Agent
from tools import MarketSizeTool

market_tool = MarketSizeTool()

researcher = Agent(
    role="Market Research Analyst",
    goal="Gather accurate market sizing data for target segments",
    backstory=(
        "You specialize in bottom-up market sizing for B2B SaaS. "
        "You know how to break down TAM into SAM and SOM using "
        "public filings, analyst reports, and comparable companies."
    ),
    tools=[market_tool],
    verbose=True,
    allow_delegation=False,
)

strategist = Agent(
    role="Go-to-Market Strategist",
    goal="Translate market data into actionable GTM recommendations",
    backstory=(
        "You turn market sizing into channel strategy, pricing models, "
        "and prioritized target segments. You think in terms of "
        "sales motion, customer acquisition cost, and expansion paths."
    ),
    tools=[],
    verbose=True,
    allow_delegation=False,
)

financial_analyst = Agent(
    role="Financial Analyst",
    goal="Model revenue potential and unit economics for each segment",
    backstory=(
        "You build bottom-up revenue models from market sizing inputs. "
        "You calculate ARR trajectories, payback periods, and LTV/CAC "
        "ratios under different pricing scenarios."
    ),
    tools=[],
    verbose=True,
    allow_delegation=False,
)

Configure the manager LLM

This is the core of the crewai hierarchical process manager llm setup. The manager agent doesn’t execute tasks directly — it plans, delegates, and synthesizes. Give it a capable model and clear instructions.

# agents.py (continued)
from crewai import Agent
from langchain_openai import ChatOpenAI

manager_llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.1,
    # If using a gateway that forwards provider cache hints:
    # model_kwargs={"extra_headers": {"Cache-Control": "no-cache"}}
)

manager = Agent(
    role="Project Manager",
    goal=(
        "Orchestrate the research, strategy, and financial analysis "
        "to produce a complete market entry assessment"
    ),
    backstory=(
        "You coordinate cross-functional analyses. You break down "
        "complex requests into discrete tasks, assign them to the "
        "right specialists, and synthesize their outputs into a "
        "coherent deliverable. You track dependencies and ensure "
        "no gaps in the final assessment."
    ),
    tools=[],
    verbose=True,
    allow_delegation=True,  # Critical: enables hierarchical delegation
    llm=manager_llm,
)

Checkpoint 1 — verify agents load:

python -c "from agents import researcher, strategist, financial_analyst, manager; print('Agents loaded:', [a.role for a in [researcher, strategist, financial_analyst, manager]])"

Expected output:

Agents loaded: ['Market Research Analyst', 'Go-to-Market Strategist', 'Financial Analyst', 'Project Manager']

Define tasks with explicit dependencies

In a hierarchical crew, tasks declare context (which prior task outputs they need) and the manager decides execution order. Each task assigns a single responsible agent.

# tasks.py
from crewai import Task
from agents import researcher, strategist, financial_analyst

research_task = Task(
    description=(
        "Estimate TAM, SAM, and SOM for 'AI coding assistants' "
        "and 'LLM observability' markets. Use the market_size_estimator "
        "tool for each. Return a structured summary with all three "
        "metrics per market."
    ),
    expected_output=(
        "A markdown table with columns: Market, TAM ($B), SAM ($B), SOM ($B). "
        "Include both markets."
    ),
    agent=researcher,
)

strategy_task = Task(
    description=(
        "Using the market sizing data, recommend which market to prioritize "
        "for initial entry. Consider: market maturity, competitive density, "
        "sales motion fit (PLG vs sales-led), and expansion potential. "
        "Provide a clear recommendation with rationale."
    ),
    expected_output=(
        "A prioritized recommendation (1-2 paragraphs) naming the "
        "primary target market and 3-4 bullet points of rationale."
    ),
    agent=strategist,
    context=[research_task],  # Depends on research output
)

financial_task = Task(
    description=(
        "Build a 3-year revenue model for the recommended primary market. "
        "Assume: Year 1 = 0.5% SOM capture, Year 2 = 1.5%, Year 3 = 3%. "
        "Price points: $500/mo (SMB), $2,000/mo (mid-market), $10,000/mo (enterprise). "
        "Mix: 60/30/10. Show ARR per year and cumulative."
    ),
    expected_output=(
        "A markdown table: Year, SMB ARR, Mid-market ARR, Enterprise ARR, "
        "Total ARR, Cumulative ARR. Plus 2-3 sentences on unit economics."
    ),
    agent=financial_analyst,
    context=[strategy_task],  # Depends on strategy output
)

synthesis_task = Task(
    description=(
        "Synthesize all prior outputs into a one-page executive summary "
        "for leadership. Include: market opportunity, recommended entry "
        "market, 3-year revenue projection, key risks, and next steps. "
        "Format as a clean markdown document."
    ),
    expected_output=(
        "A complete executive summary in markdown with clear sections: "
        "Opportunity, Recommendation, Financial Projection, Risks, Next Steps."
    ),
    agent=manager,  # Manager synthesizes — doesn't delegate this one
    context=[research_task, strategy_task, financial_task],
)

Wire the hierarchical crew

Now assemble the crew with process="hierarchical" and assign the manager agent. The manager receives all tasks and decides delegation.

# main.py
from crewai import Crew, Process
from agents import researcher, strategist, financial_analyst, manager
from tasks import research_task, strategy_task, financial_task, synthesis_task

crew = Crew(
    agents=[researcher, strategist, financial_analyst, manager],
    tasks=[research_task, strategy_task, financial_task, synthesis_task],
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True,
    memory=False,  # Enable for multi-run context retention
    planning=True,  # Let manager create a plan before execution
)

if __name__ == "__main__":
    result = crew.kickoff()
    print("\n" + "=" * 60)
    print("FINAL OUTPUT")
    print("=" * 60)
    print(result)

Checkpoint 2 — run the crew:

python main.py

Expected output (abridged — actual LLM output will vary):

[DEBUG] == Working Agent: Project Manager
[DEBUG] == Starting Task: Estimate TAM, SAM, and SOM for 'AI coding assistants'...
[DEBUG] == Working Agent: Market Research Analyst
[DEBUG] == Tool market_size_estimator returned: TAM: $12.4B, SAM: $3.1B, SOM: $0.4B
[DEBUG] == Tool market_size_estimator returned: TAM: $4.2B, SAM: $1.1B, SOM: $0.15B
[DEBUG] == Completed Task: Market Research Analyst

[DEBUG] == Working Agent: Project Manager
[DEBUG] == Starting Task: Using the market sizing data, recommend which market...
[DEBUG] == Working Agent: Go-to-Market Strategist
[DEBUG] == Completed Task: Go-to-Market Strategist

[DEBUG] == Working Agent: Project Manager
[DEBUG] == Starting Task: Build a 3-year revenue model for the recommended...
[DEBUG] == Working Agent: Financial Analyst
[DEBUG] == Completed Task: Financial Analyst

[DEBUG] == Working Agent: Project Manager
[DEBUG] == Starting Task: Synthesize all prior outputs into a one-page...
[DEBUG] == Completed Task: Project Manager

============================================================
FINAL OUTPUT
============================================================
# Executive Summary: Market Entry Assessment

## Opportunity
The AI coding assistant market presents a $12.4B TAM with $3.1B SAM...
[full markdown summary follows]

What the manager actually does

With planning=True, the manager first emits a plan. You can inspect it by adding a callback:

# main.py (add before kickoff)
from crewai import Crew, Process
from agents import researcher, strategist, financial_analyst, manager
from tasks import research_task, strategy_task, financial_task, synthesis_task

def print_plan(plan):
    print("\n=== MANAGER PLAN ===")
    for step in plan:
        print(f"  {step['task']} -> {step['agent']}")

crew = Crew(
    agents=[researcher, strategist, financial_analyst, manager],
    tasks=[research_task, strategy_task, financial_task, synthesis_task],
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True,
    planning=True,
    planning_callback=print_plan,
)

result = crew.kickoff()

Sample plan output:

=== MANAGER PLAN ===
  Estimate TAM, SAM, and SOM for 'AI coding assistants'... -> Market Research Analyst
  Using the market sizing data, recommend which market... -> Go-to-Market Strategist
  Build a 3-year revenue model for the recommended... -> Financial Analyst
  Synthesize all prior outputs into a one-page... -> Project Manager

The manager respects context dependencies automatically — it won’t schedule the strategy task before research completes.

Tuning the manager LLM

The manager’s reasoning quality directly affects delegation correctness. Three levers matter most:

Model capability — Use the strongest model you can afford for the manager. gpt-4o or claude-3.5-sonnet handle multi-step planning reliably. Weaker models hallucinate dependencies or assign tasks to wrong agents.

Temperature — Keep it low (0.1–0.3). The manager is a planner, not a creative writer. Deterministic delegation beats variety.

System prompt via backstory — The backstory field is the manager’s system prompt. Be explicit about:

How to break down requests
When to delegate vs. execute
How to handle missing information
Output format expectations

# Stronger manager backstory example
manager = Agent(
    role="Project Manager",
    goal="...",
    backstory=(
        "You are a senior technical program manager. "
        "Planning rules:\n"
        "1. Decompose the user request into atomic, non-overlapping tasks.\n"
        "2. Assign each task to exactly one specialist agent by role.\n"
        "3. Respect context dependencies — never schedule a task before "
        "its dependencies complete.\n"
        "4. If a task requires information not in context, either "
        "delegate a research sub-task or note the gap in the plan.\n"
        "5. The final synthesis task is YOURS — do not delegate it.\n"
        "Output a JSON plan with fields: task, agent, dependencies."
    ),
    llm=manager_llm,
    allow_delegation=True,
)

Common failure modes and fixes

Symptom	Cause	Fix
Manager assigns task to wrong agent	Vague role/goal descriptions	Make `role` and `goal` distinct and specific
Circular dependency error	`context` references form a cycle	Ensure DAG — each task depends only on earlier tasks
Manager executes task itself instead of delegating	`allow_delegation=False` on manager	Set `allow_delegation=True` on manager only
Plan ignores a task	Task not in `crew.tasks` list	All tasks must be in the crew’s `tasks` array
Output missing sections	`expected_output` too vague	Specify exact format (markdown table, sections, bullet count)

Sequential vs hierarchical: when to use which

Dimension	Sequential	Hierarchical
Execution order	Fixed, defined by task list	Dynamic, planned by manager
Flexibility	Low — change code to reorder	High — manager adapts to request
Debugging	Trivial — linear trace	Harder — need planning callback
Token usage	Predictable	Higher — planning + delegation overhead
Best for	Known, repeatable workflows	Open-ended, variable requests

Use hierarchical when the task graph isn’t known at code-author time — e.g., “analyze this market” where the manager decides which sub-markets to research.

Production considerations

Observability — Wrap crew.kickoff() with logging for: manager plan, each agent’s raw output, tool calls, token counts. This is essential for debugging delegation errors.

Rate limits — The manager makes sequential execution safer. If you hit provider limits, the crew stalls. A gateway with automatic fallback (like n4n.ai) keeps the crew moving across provider degradations.

Caching — Enable cache=True on the crew for repeated runs with identical inputs. The manager’s plan and agent outputs cache separately.

crew = Crew(
    # ...
    cache=True,
    cache_dir=".crew_cache",
)

Memory — Set memory=True and provide a memory_config if you want the manager to recall prior runs. Useful for iterative refinement (“now add competitive analysis to the previous assessment”).

Full runnable example

All files together:

# tools.py
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

class MarketSizeInput(BaseModel):
    market: str = Field(..., description="Market segment to estimate")

class MarketSizeTool(BaseTool):
    name: str = "market_size_estimator"
    args_schema: Type[BaseModel] = MarketSizeInput

    def _run(self, market: str) -> str:
        estimates = {
            "ai coding assistants": {"tam": 12.4, "sam": 3.1, "som": 0.4},
            "devops automation": {"tam": 8.7, "sam": 2.2, "som": 0.3},
            "llm observability": {"tam": 4.2, "sam": 1.1, "som": 0.15},
        }
        key = market.lower()
        if key in estimates:
            e = estimates[key]
            return f"TAM: ${e['tam']}B, SAM: ${e['sam']}B, SOM: ${e['som']}B"
        return f"No data for '{market}'. Known markets: {list(estimates.keys())}"

# agents.py
from crewai import Agent
from langchain_openai import ChatOpenAI
from tools import MarketSizeTool

market_tool = MarketSizeTool()

researcher = Agent(
    role="Market Research Analyst",
    goal="Gather accurate market sizing data for target segments",
    backstory=(
        "You specialize in bottom-up market sizing for B2B SaaS. "
        "You know how to break down TAM into SAM and SOM using "
        "public filings, analyst reports, and comparable companies."
    ),
    tools=[market_tool],
    verbose=True,
    allow_delegation=False,
)

strategist = Agent(
    role="Go-to-Market Strategist",
    goal="Translate market data into actionable GTM recommendations",
    backstory=(
        "You turn market sizing into channel strategy, pricing models, "
        "and prioritized target segments. You think in terms of "
        "sales motion, customer acquisition cost, and expansion paths."
    ),
    tools=[],
    verbose=True,
    allow_delegation=False,
)

financial_analyst = Agent(
    role="Financial Analyst",
    goal="Model revenue potential and unit economics for each segment",
    backstory=(
        "You build bottom-up revenue models from market sizing inputs. "
        "You calculate ARR trajectories, payback periods, and LTV/CAC "
        "ratios under different pricing scenarios."
    ),
    tools=[],
    verbose=True,
    allow_delegation=False,
)

manager_llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

manager = Agent(
    role="Project Manager",
    goal=(
        "Orchestrate the research, strategy, and financial analysis "
        "to produce a complete market entry assessment"
    ),
    backstory=(
        "You coordinate cross-functional analyses. You break down "
        "complex requests into discrete tasks, assign them to the "
        "right specialists, and synthesize their outputs into a "
        "coherent deliverable. You track dependencies and ensure "
        "no gaps in the final assessment."
    ),
    tools=[],
    verbose=True,
    allow_delegation=True,
    llm=manager_llm,
)

# tasks.py
from crewai import Task
from agents import researcher, strategist, financial_analyst, manager

research_task = Task(
    description=(
        "Estimate TAM, SAM, and SOM for 'AI coding assistants' "
        "and 'LLM observability' markets. Use the market_size_estimator "
        "tool for each. Return a structured summary with all three "
        "metrics per market."
    ),
    expected_output=(
        "A markdown table with columns: Market, TAM ($B), SAM ($B), SOM ($B). "
        "Include both markets."
    ),
    agent=researcher,
)

strategy_task = Task(
    description=(
        "Using the market sizing data, recommend which market to prioritize "
        "for initial entry. Consider: market maturity, competitive density, "
        "sales motion fit (PLG vs sales-led), and expansion potential. "
        "Provide a clear recommendation with rationale."
    ),
    expected_output=(
        "A prioritized recommendation (1-2 paragraphs) naming the "
        "primary target market and 3-4 bullet points of rationale."
    ),
    agent=strategist,
    context=[research_task],
)

financial_task = Task(
    description=(
        "Build a 3-year revenue model for the recommended primary market. "
        "Assume: Year 1 = 0.5% SOM capture, Year 2 = 1.5%, Year 3 = 3%. "
        "Price points: $500/mo (SMB), $2,000/mo (mid-market), $10,000/mo (enterprise). "
        "Mix: 60/30/10. Show ARR per year and cumulative."
    ),
    expected_output=(
        "A markdown table: Year, SMB ARR, Mid-market ARR, Enterprise ARR, "
        "Total ARR, Cumulative ARR. Plus 2-3 sentences on unit economics."
    ),
    agent=financial_analyst,
    context=[strategy_task],
)

synthesis_task = Task(
    description=(
        "Synthesize all prior outputs into a one-page executive summary "
        "for leadership. Include: market opportunity, recommended entry "
        "market, 3-year revenue projection, key risks, and next steps. "
        "Format as a clean markdown document."
    ),
    expected_output=(
        "A complete executive summary in markdown with clear sections: "
        "Opportunity, Recommendation, Financial Projection, Risks, Next Steps."
    ),
    agent=manager,
    context=[research_task, strategy_task, financial_task],
)

# main.py
from crewai import Crew, Process
from agents import researcher, strategist, financial_analyst, manager
from tasks import research_task, strategy_task, financial_task, synthesis_task

def print_plan(plan):
    print("\n=== MANAGER PLAN ===")
    for step in plan:
        print(f"  {step['task']} -> {step['agent']}")

crew = Crew(
    agents=[researcher, strategist, financial_analyst, manager],
    tasks=[research_task, strategy_task, financial_task, synthesis_task],
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True,
    planning=True,
    planning_callback=print_plan,
    cache=True,
    cache_dir=".crew_cache",
)

if __name__ == "__main__":
    result = crew.kickoff()
    print("\n" + "=" * 60)
    print("FINAL OUTPUT")
    print("=" * 60)
    print(result)

Run it:

python main.py

You now have a working crewai hierarchical process manager llm setup that plans, delegates, and synthesizes. Swap the tools, agents, and tasks for your domain — the delegation mechanics stay the same.