Debugging CrewAI hierarchical process delegation

Hierarchical crews in CrewAI introduce a manager agent that decides which specialist agent handles each task. When delegation breaks — tasks stall, wrong agents are picked, or the manager loops — you need systematic crewai hierarchical process delegation debugging rather than guesswork. This walkthrough gives you a reproducible process: enable the right logs, instrument the manager’s decision path, add guardrails, and verify the fix with a minimal test case.

Step 1: Reproduce with a minimal crew

Start by stripping your crew to the smallest configuration that still shows the symptom. Remove custom tools, memory, and external APIs. Keep one manager and two specialists with clearly distinct roles.

# minimal_crew.py
from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Researcher",
    goal="Find one fact about the topic",
    backstory="You retrieve a single verifiable fact.",
    allow_delegation=False,
    verbose=True,
)

writer = Agent(
    role="Writer",
    goal="Summarize the fact in one sentence",
    backstory="You write concise summaries.",
    allow_delegation=False,
    verbose=True,
)

manager = Agent(
    role="Manager",
    goal="Decide which specialist handles the request",
    backstory="You route tasks to Researcher or Writer.",
    allow_delegation=True,
    verbose=True,
)

research_task = Task(
    description="Research the boiling point of water at sea level",
    expected_output="One sentence with the temperature and unit",
    agent=researcher,
)

write_task = Task(
    description="Summarize the research result",
    expected_output="One sentence summary",
    agent=writer,
)

crew = Crew(
    agents=[manager, researcher, writer],
    tasks=[research_task, write_task],
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True,
)

if __name__ == "__main__":
    result = crew.kickoff()
    print(result)

Run it and confirm the failure mode: does the manager pick the wrong agent, repeat the same delegation, or hang?

python minimal_crew.py

Verify success: the script finishes in under 30 seconds and prints a coherent result. If it loops or errors, you have a reproducible case.

Step 2: Enable full manager reasoning logs

CrewAI’s verbose=True shows task outputs but not the manager’s internal chain-of-thought. Patch the manager’s LLM call to capture the raw prompt and completion.

# debug_logging.py
import json
from crewai import Agent, Task, Crew, Process
from crewai.llm import LLM
from langchain_core.callbacks import BaseCallbackHandler

class ManagerLogger(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print("\n=== MANAGER PROMPT ===")
        for p in prompts:
            print(p[:2000])
    
    def on_llm_end(self, response, **kwargs):
        print("\n=== MANAGER RESPONSE ===")
        for gen in response.generations:
            for chunk in gen:
                print(chunk.text[:2000])

logged_llm = LLM(
    model="gpt-4o-mini",
    callbacks=[ManagerLogger()],
)

manager = Agent(
    role="Manager",
    goal="Decide which specialist handles the request",
    backstory="You route tasks to Researcher or Writer.",
    allow_delegation=True,
    verbose=True,
    llm=logged_llm,
)

Re-run the minimal crew. The manager’s prompt will show the exact context it receives: available agents, their descriptions, task history, and the current request. Look for:

• Missing or truncated agent descriptions
• Task history that grows without bound
• Ambiguous delegation instructions in the system prompt

Verify success: you can see the manager’s reasoning for each delegation decision in the console output.

Step 3: Instrument delegation outcomes

Add a wrapper that records every delegation attempt — which agent was chosen, the task description, and whether the delegated task succeeded.

# instrumentation.py
from dataclasses import dataclass, field
from typing import List
from crewai import Agent, Task, Crew, Process

@dataclass
class DelegationRecord:
    step: int
    requested_agent: str
    actual_agent: str
    task_description: str
    success: bool
    error: str = ""

delegation_log: List[DelegationRecord] = []
step_counter = 0

class InstrumentedCrew(Crew):
    def _execute_task(self, task, agent, context=None):
        global step_counter
        step_counter += 1
        record = DelegationRecord(
            step=step_counter,
            requested_agent=agent.role,
            actual_agent=agent.role,
            task_description=task.description[:200],
            success=False,
        )
        try:
            result = super()._execute_task(task, agent, context)
            record.success = True
            return result
        except Exception as e:
            record.error = str(e)
            raise
        finally:
            delegation_log.append(record)

# Use InstrumentedCrew instead of Crew
crew = InstrumentedCrew(
    agents=[manager, researcher, writer],
    tasks=[research_task, write_task],
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True,
)

if __name__ == "__main__":
    result = crew.kickoff()
    print("\n=== DELEGATION LOG ===")
    for r in delegation_log:
        status = "OK" if r.success else f"FAIL: {r.error}"
        print(f"Step {r.step}: {r.requested_agent} -> {r.actual_agent} | {status}")
        print(f"  Task: {r.task_description}")

Verify success: after run completion, the delegation log shows a clean sequence with no repeated steps and all success=True.

Step 4: Add explicit delegation rules to the manager

The most common failure is an underspecified manager backstory. Give the manager a structured decision framework in its system prompt.

manager = Agent(
    role="Manager",
    goal="Route each task to exactly one specialist",
    backstory=(
        "You are a routing manager. Follow these rules strictly:\n"
        "1. If the task asks for facts, data, or research -> delegate to Researcher\n"
        "2. If the task asks for writing, summarizing, or formatting -> delegate to Writer\n"
        "3. Never delegate to yourself.\n"
        "4. Never delegate the same task twice.\n"
        "5. After a specialist completes a task, consider the workflow done unless a new task appears."
    ),
    allow_delegation=True,
    verbose=True,
    llm=logged_llm,
)

Test again. The manager should now route deterministically.

Verify success: delegation log shows Researcher for the research task, Writer for the write task, no self-delegation, no repeats.

Step 5: Bound the manager’s context window

Hierarchical crews accumulate the full conversation history in the manager’s prompt. With many tasks, this exceeds the model’s context limit and truncates agent descriptions. Implement a sliding window that keeps only the last N interactions.

# context_window.py
from crewai import Crew
from typing import List, Dict

class BoundedContextCrew(Crew):
    MAX_HISTORY = 6  # keep last 3 task cycles (request + response)
    
    def _get_manager_prompt(self, task, agents, history):
        # history is a list of dicts with 'task', 'agent', 'output'
        trimmed = history[-self.MAX_HISTORY:]
        return super()._get_manager_prompt(task, agents, trimmed)

crew = BoundedContextCrew(
    agents=[manager, researcher, writer],
    tasks=[research_task, write_task],
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True,
)

For longer workflows, increase MAX_HISTORY or summarize older entries instead of dropping them.

Verify success: manager prompt size stays stable across 10+ task cycles (check via the ManagerLogger from Step 2).

Step 6: Handle rate limits and provider failures

When the manager’s LLM call fails — rate limit, timeout, or provider degradation — the crew can hang or retry indefinitely. Wrap the LLM with a retry policy and fallback.

# resilient_llm.py
from crewai.llm import LLM
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

class ResilientLLM(LLM):
    @retry(
        wait=wait_exponential_jitter(initial=1, max=30),
        stop=stop_after_attempt(3),
        reraise=True,
    )
    def call(self, messages, **kwargs):
        return super().call(messages, **kwargs)

resilient_llm = ResilientLLM(model="gpt-4o-mini", callbacks=[ManagerLogger()])
manager = Agent(..., llm=resilient_llm)

If you route through a gateway that exposes multiple providers, you can also implement provider-level fallback. For example, n4n.ai forwards x-provider routing hints and returns x-fallback-provider headers when the primary is degraded — your client can read that header and switch automatically on the next request.

Verify success: simulate a rate limit (set a low RPM quota on your API key) and confirm the crew completes after retries without manual intervention.

Step 7: Write a regression test

Encode the expected delegation sequence as a test that fails if the manager deviates.

# test_delegation.py
import pytest
from minimal_crew import crew, delegation_log

def test_hierarchical_delegation_sequence():
    result = crew.kickoff()
    
    assert len(delegation_log) == 2, f"Expected 2 delegations, got {len(delegation_log)}"
    
    # Step 1: research task -> Researcher
    assert delegation_log[0].actual_agent == "Researcher"
    assert delegation_log[0].success is True
    
    # Step 2: write task -> Writer
    assert delegation_log[1].actual_agent == "Writer"
    assert delegation_log[1].success is True
    
    # No self-delegation
    for record in delegation_log:
        assert record.actual_agent != "Manager"
    
    print("Regression test passed")

Run with pytest test_delegation.py -v. This catches regressions when you upgrade CrewAI, change models, or modify agent descriptions.

Verify success: test passes consistently across 5 consecutive runs.

Step 8: Profile token usage per delegation

Delegation loops often stem from the manager re-explaining the same context. Track tokens per manager turn to spot bloat.

# token_tracker.py
from crewai.llm import LLM
from dataclasses import dataclass

@dataclass
class TokenSnapshot:
    step: int
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int

token_history: List[TokenSnapshot] = []

class TokenTrackingLLM(LLM):
    def call(self, messages, **kwargs):
        response = super().call(messages, **kwargs)
        usage = response.usage if hasattr(response, 'usage') else None
        if usage:
            token_history.append(TokenSnapshot(
                step=len(token_history) + 1,
                prompt_tokens=usage.prompt_tokens,
                completion_tokens=usage.completion_tokens,
                total_tokens=usage.total_tokens,
            ))
        return response

tracking_llm = TokenTrackingLLM(model="gpt-4o-mini", callbacks=[ManagerLogger()])
manager = Agent(..., llm=tracking_llm)

After a run, print the history:

for t in token_history:
    print(f"Step {t.step}: {t.prompt_tokens} prompt + {t.completion_tokens} completion = {t.total_tokens} total")

Verify success: prompt tokens stay roughly constant across steps (within 20%). Growth indicates accumulating context that needs the sliding window from Step 5.

Step 9: Validate with a multi-turn scenario

Extend the minimal crew to three sequential tasks that require alternating specialists. This exposes handoff bugs that single-task runs miss.

# multiturn_crew.py
from crewai import Agent, Task, Crew, Process

tasks = [
    Task(description="Research the capital of France", expected_output="City name", agent=researcher),
    Task(description="Write a one-sentence travel blurb for that city", expected_output="One sentence", agent=writer),
    Task(description="Research the population of that city", expected_output="Number with year", agent=researcher),
    Task(description="Summarize both facts in a tweet", expected_output="Under 280 chars", agent=writer),
]

crew = InstrumentedCrew(
    agents=[manager, researcher, writer],
    tasks=tasks,
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True,
)

if __name__ == "__main__":
    result = crew.kickoff()
    print("\n=== FINAL RESULT ===")
    print(result)
    print("\n=== DELEGATION LOG ===")
    for r in delegation_log:
        status = "OK" if r.success else f"FAIL: {r.error}"
        print(f"Step {r.step}: {r.actual_agent} | {status} | {r.task_description[:80]}")

Expected delegation sequence: Researcher, Writer, Researcher, Writer.

Verify success: log matches expected sequence exactly, no repeated research or write steps, total runtime under 60 seconds.

Step 10: Document the debug checklist

Create a DEBUGGING.md in your repo so the next engineer (or future you) follows the same path.

# CrewAI Hierarchical Delegation Debug Checklist

1. **Reproduce minimally** — `minimal_crew.py` with 2 specialists, 2 tasks
2. **Capture manager reasoning** — `ManagerLogger` callback on manager's LLM
3. **Log every delegation** — `InstrumentedCrew` records agent, task, success
4. **Explicit routing rules** — manager backstory with numbered decision rules
5. **Bound context window** — `BoundedContextCrew` keeps last N history entries
6. **Resilient LLM** — retry with exponential backoff, provider fallback
7. **Regression test** — `test_delegation.py` asserts exact delegation sequence
8. **Token profiling** — `TokenTrackingLLM` watches for prompt bloat
9. **Multi-turn validation** — 4+ task alternating sequence
10. **CI gate** — run regression test on every PR

Common failure patterns and fixes

When to escalate beyond the crew

If the manager consistently misroutes despite explicit rules, the model may lack the reasoning capacity for your routing complexity. Options:

• Use a stronger model for the manager only (keep specialists on cheaper models)
• Replace hierarchical process with a deterministic router function that selects the agent, then call Crew(process=Process.sequential) with that agent
• Add a classification step before the crew: a lightweight LLM call that tags the task type, then map tags to agents in code

The hierarchical process is convenient but not magic. Treat the manager as a component you instrument, test, and version like any other service.