CrewAI hierarchical crews for complex research tasks

CrewAI hierarchical crew research tasks solve a real problem: single-pass agents hallucinate, miss context, and cannot self-correct. A hierarchical process introduces a manager agent that plans, delegates, reviews, and iterates — turning a fragile prompt chain into a resilient workflow. This tutorial walks you through building a three-layer research crew that produces cited, structured reports you can actually ship.

Step 1: Define the research scope and output contract

Before writing any agent code, specify what “done” looks like. A research crew needs a concrete deliverable: a Markdown report with executive summary, findings grouped by theme, source citations, and a confidence score per claim. Write this contract as a Pydantic model so the manager agent can validate output before returning it.

# models.py
from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum

class ConfidenceLevel(str, Enum):
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

class Citation(BaseModel):
    source_id: str
    title: str
    url: str
    accessed_at: str
    snippet: str

class Finding(BaseModel):
    theme: str
    claim: str
    evidence: List[str]
    citations: List[Citation]
    confidence: ConfidenceLevel
    contradictions: List[str] = Field(default_factory=list)

class ResearchReport(BaseModel):
    topic: str
    executive_summary: str
    findings: List[Finding]
    gaps: List[str]
    generated_at: str
    total_sources: int

This model becomes the output_json schema for the final writer agent. The manager agent uses it to reject incomplete drafts and request revisions.

Step 2: Configure the LLM gateway and shared settings

Use a single OpenAI-compatible endpoint so you can swap models per agent without rewriting client code. Set temperature low for research agents (0.1–0.3) and slightly higher for the writer (0.4) to allow synthesis. If you route through n4n.ai, you get automatic fallback when a provider degrades and per-token metering across all agents — useful for cost attribution in multi-agent runs.

# config.py
import os
from langchain_openai import ChatOpenAI

def get_llm(model: str, temperature: float = 0.2) -> ChatOpenAI:
    return ChatOpenAI(
        model=model,
        temperature=temperature,
        api_key=os.getenv("OPENROUTER_API_KEY"),
        base_url="https://openrouter.n4n.ai/v1",  # or your preferred gateway
        max_tokens=4000,
        request_timeout=120,
    )

# Model assignments
MANAGER_MODEL = "anthropic/claude-3.5-sonnet"
RESEARCHER_MODEL = "google/gemini-1.5-pro"
ANALYST_MODEL = "anthropic/claude-3.5-sonnet"
WRITER_MODEL = "openai/gpt-4o"

Keep model assignments in one place. When a new model beats the current one on a specific task (e.g., Gemini for long-context retrieval), you change one line.

Step 3: Build the toolset — search, fetch, and extract

Agents need deterministic tools, not vague “browse the web” instructions. Wrap a search API (Serper, Tavily, or Exa) and a content extractor (trafilatura or newspaper3k) into CrewAI tools with strict input/output types.

# tools.py
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type, List, Optional
import requests
import trafilatura
import json
from datetime import datetime

class SearchInput(BaseModel):
    query: str = Field(..., description="Precise search query, use quotes for exact phrases")
    num_results: int = Field(default=10, ge=5, le=20)
    recency_days: Optional[int] = Field(default=365, description="Limit results to last N days")

class SearchResult(BaseModel):
    title: str
    url: str
    snippet: str
    position: int

class SearchTool(BaseTool):
    name: str = "web_search"
    args_schema: Type[BaseModel] = SearchInput

    def _run(self, query: str, num_results: int = 10, recency_days: int = 365) -> str:
        api_key = os.getenv("SERPER_API_KEY")
        payload = {
            "q": query,
            "num": num_results,
            "tbs": f"qdr:d{recency_days}" if recency_days <= 365 else ""
        }
        headers = {"X-API-KEY": api_key, "Content-Type": "application/json"}
        resp = requests.post("https://google.serper.dev/search", json=payload, headers=headers)
        data = resp.json()
        results = []
        for i, item in enumerate(data.get("organic", [])[:num_results]):
            results.append(SearchResult(
                title=item.get("title", ""),
                url=item.get("link", ""),
                snippet=item.get("snippet", ""),
                position=i + 1
            ).model_dump())
        return json.dumps(results, indent=2)

class ExtractInput(BaseModel):
    url: str = Field(..., description="URL to fetch and extract main content from")
    max_chars: int = Field(default=8000, description="Truncate extracted text to this length")

class ExtractTool(BaseTool):
    name: str = "extract_content"
    args_schema: Type[BaseModel] = ExtractInput

    def _run(self, url: str, max_chars: int = 8000) -> str:
        downloaded = trafilatura.fetch_url(url)
        if not downloaded:
            return f"ERROR: Failed to fetch {url}"
        text = trafilatura.extract(downloaded, include_comments=False, include_tables=True)
        if not text:
            return f"ERROR: No extractable content at {url}"
        return text[:max_chars]

Register these tools globally so any agent can invoke them. The search tool returns structured JSON; the extract tool returns clean text. Both are deterministic and retryable.

Step 4: Create the specialized researcher agents

In a crewai hierarchical crew research tasks setup, you want at least two researcher personas: a broad scanner that casts a wide net, and a deep diver that follows up on specific threads. Give each a distinct system prompt and tool budget.

# agents.py
from crewai import Agent
from config import get_llm, RESEARCHER_MODEL, ANALYST_MODEL
from tools import SearchTool, ExtractTool

search_tool = SearchTool()
extract_tool = ExtractTool()

broad_researcher = Agent(
    role="Broad Research Scanner",
    goal="Identify authoritative sources covering all facets of the topic, prioritizing primary sources, recent publications, and diverse viewpoints",
    backstory=(
        "You are a systematic literature reviewer. You cast a wide net using precise search queries, "
        "then filter for credibility: peer-reviewed papers, government reports, established journalism, "
        "and recognized industry publications. You avoid blogs, forums, and unattributed content. "
        "You output a structured source list with relevance scores."
    ),
    tools=[search_tool, extract_tool],
    llm=get_llm(RESEARCHER_MODEL, temperature=0.1),
    max_iter=5,
    verbose=True,
    allow_delegation=False,
)

deep_researcher = Agent(
    role="Deep-Dive Researcher",
    goal="Extract detailed evidence, data points, and direct quotes from high-priority sources identified by the scanner",
    backstory=(
        "You take a curated source list and read each source thoroughly. You pull exact figures, "
        "methodology details, conflicting statements, and contextual nuances. You cite every claim "
        "with a source ID and character offset where possible. You flag paywalled or inaccessible content."
    ),
    tools=[extract_tool],
    llm=get_llm(RESEARCHER_MODEL, temperature=0.1),
    max_iter=8,
    verbose=True,
    allow_delegation=False,
)

analyst = Agent(
    role="Research Analyst",
    goal="Synthesize raw findings into thematically grouped claims, identify contradictions, assess confidence, and surface gaps",
    backstory=(
        "You receive extracted evidence from multiple sources. You cluster claims by theme, "
        "resolve conflicts by weighing source credibility and recency, assign confidence levels, "
        "and explicitly list what remains unknown. You output structured Finding objects."
    ),
    llm=get_llm(ANALYST_MODEL, temperature=0.2),
    max_iter=4,
    verbose=True,
    allow_delegation=False,
)

Notice allow_delegation=False — these are leaf agents. The manager handles delegation.

Step 5: Define the manager agent with explicit process control

The manager is the brain of the crewai hierarchical crew research tasks workflow. It receives the topic, creates a research plan, assigns tasks to researchers, reviews their output, requests revisions, and finally commissions the report. Give it a detailed prompt that encodes your quality bar.

# manager.py
from crewai import Agent, Task, Crew, Process
from config import get_llm, MANAGER_MODEL, WRITER_MODEL
from agents import broad_researcher, deep_researcher, analyst
from models import ResearchReport
import json
from datetime import datetime

manager = Agent(
    role="Research Manager",
    goal="Produce a comprehensive, well-cited research report on the given topic by orchestrating a team of specialized agents",
    backstory=(
        "You are an experienced research director. You break complex topics into researchable questions, "
        "delegate to scanners and deep-divers, review their work for completeness and citation quality, "
        "send work back for gaps or weak evidence, and only approve when the report meets the output contract. "
        "You track source count, theme coverage, and confidence distribution. You never accept hallucinated citations."
    ),
    llm=get_llm(MANAGER_MODEL, temperature=0.15),
    max_iter=12,
    verbose=True,
    allow_delegation=True,  # Critical: enables hierarchical delegation
)

# The manager's first task: create a research plan
planning_task = Task(
    description=(
        "Given the research topic: '{topic}', create a detailed research plan with:\n"
        "1. 5-8 specific research questions that decompose the topic\n"
        "2. For each question: suggested search queries (2-3 per question)\n"
        "3. Source type priorities (academic, government, industry, news)\n"
        "4. Expected themes for final grouping\n"
        "5. Success criteria: minimum sources per theme, recency requirements\n"
        "Output as JSON."
    ),
    expected_output="Valid JSON research plan",
    agent=manager,
    output_json=None,  # Manager validates manually
)

# Scanner task — delegated by manager
scan_task = Task(
    description=(
        "Execute the research plan. For each research question, run the suggested searches. "
        "Collect 15-25 unique, credible sources. Deduplicate by URL. Score each source 1-10 on credibility. "
        "Return a JSON list of sources with: source_id, title, url, snippet, credibility_score, question_addressed."
    ),
    expected_output="JSON array of 15-25 scored sources",
    agent=broad_researcher,
    context=[planning_task],
)

# Deep-dive task — delegated by manager
deep_dive_task = Task(
    description=(
        "For each high-priority source (credibility >= 7) from the scanner, extract full content. "
        "Pull exact quotes, data tables, methodology descriptions, and author conclusions. "
        "Return JSON: source_id, key_claims[], quotes[], data_points[], methodology_notes, limitations."
    ),
    expected_output="JSON array of detailed extractions",
    agent=deep_researcher,
    context=[scan_task],
)

# Analyst task — delegated by manager
analysis_task = Task(
    description=(
        "Synthesize extractions into Finding objects per the ResearchReport schema. "
        "Group claims by theme. For each claim: list supporting evidence, citations, confidence, "
        "and any contradictory evidence. Identify gaps where evidence is thin or missing. "
        "Output must validate against ResearchReport schema (findings array only)."
    ),
    expected_output="JSON array of Finding objects",
    agent=analyst,
    context=[deep_dive_task],
    output_json=None,  # Will validate in manager review
)

# Final writer task — produces the deliverable
writer = Agent(
    role="Report Writer",
    goal="Transform structured findings into a polished, publication-ready Markdown report",
    backstory=(
        "You write clear, executive-ready research reports. You craft an executive summary that "
        "stands alone, organize findings by theme with clear headings, embed citations as footnotes, "
        "and include a gaps section. You never invent claims not in the findings."
    ),
    llm=get_llm(WRITER_MODEL, temperature=0.4),
    max_iter=3,
    verbose=True,
    allow_delegation=False,
)

writing_task = Task(
    description=(
        "Using the validated findings from the analyst, produce the final ResearchReport. "
        "Write executive_summary (2-3 paragraphs), format findings with Markdown headings per theme, "
        "render citations as numbered footnotes with URLs, include gaps and metadata. "
        "Output must be valid JSON matching ResearchReport schema."
    ),
    expected_output="Complete ResearchReport JSON",
    agent=writer,
    context=[analysis_task],
    output_json=ResearchReport,
)

Step 6: Assemble the hierarchical crew and add the review loop

The hierarchical process in CrewAI means the manager agent dynamically delegates tasks based on its reasoning. But you still need a quality gate: the manager must review each delegated output and either approve or send back with specific feedback. Implement this as a custom callback or by giving the manager a review task that gates progression.

# crew.py
from crewai import Crew, Process
from manager import (
    manager, planning_task, scan_task, deep_dive_task,
    analysis_task, writing_task, writer
)
from models import ResearchReport
import json
from datetime import datetime

class ResearchCrew:
    def __init__(self, topic: str):
        self.topic = topic
        self.crew = Crew(
            agents=[manager, writer],  # Only top-level agents; others are delegated
            tasks=[planning_task, writing_task],  # Manager expands the rest
            process=Process.hierarchical,
            manager_agent=manager,
            verbose=True,
            memory=True,  # Enables cross-task context
            max_rpm=30,   # Rate limit protection
        )
        self.state = {}

    def run(self) -> ResearchReport:
        # Kick off with the topic injected into the planning task
        result = self.crew.kickoff(inputs={"topic": self.topic})
        
        # The hierarchical process returns the final task's output
        # Validate against schema
        if isinstance(result, str):
            report_data = json.loads(result)
        else:
            report_data = result
        
        report = ResearchReport(**report_data)
        report.generated_at = datetime.utcnow().isoformat() + "Z"
        return report

    def run_with_review(self, max_cycles: int = 3) -> ResearchReport:
        """
        Manual review loop for higher quality. The manager reviews each stage
        and can request rework. This bypasses pure hierarchical mode may skip review.
        """
        # This pattern gives you explicit control over the delegation flow
        # Use when you need audit trails or human-in-the-loop checkpoints
        pass  # Implementation left as exercise; see notes below

Step 7: Execute and verify

Run the crew with a test topic. The hierarchical process will show the manager’s reasoning as it delegates, reviews, and iterates.

# run.py
import os
from dotenv import load_dotenv
from crew import ResearchCrew
from models import ResearchReport

load_dotenv()

if __name__ == "__main__":
    topic = "Impact of retrieval-augmented generation on hallucination rates in production LLM systems"
    
    crew = ResearchCrew(topic)
    report = crew.run()
    
    # Verification checks
    print(f"Topic: {report.topic}")
    print(f"Sources: {report.total_sources}")
    print(f"Findings: {len(report.findings)}")
    print(f"Themes: {set(f.theme for f in report.findings)}")
    print(f"Confidence distribution: "
          f"{ {c: sum(1 for f in report.findings if f.confidence == c) for c in ['high','medium','low']} }")
    print(f"Gaps identified: {len(report.gaps)}")
    
    # Save full report
    with open(f"research_{topic[:40].replace(' ', '_')}.json", "w") as f:
        f.write(report.model_dump_json(indent=2))
    
    # Render Markdown for reading
    md = f"# Research Report: {report.topic}\n\n"
    md += f"**Generated:** {report.generated_at}  \n"
    md += f"**Total Sources:** {report.total_sources}\n\n"
    md += f"## Executive Summary\n{report.executive_summary}\n\n"
    md += "## Findings\n"
    for finding in report.findings:
        md += f"### {finding.theme}\n"
        md += f"**Claim:** {finding.claim}  \n"
        md += f"**Confidence:** {finding.confidence.value}  \n"
        md += f"**Evidence:**\n"
        for ev in finding.evidence:
            md += f"- {ev}\n"
        md += f"**Citations:**\n"
        for cit in finding.citations:
            md += f"- [{cit.source_id}] {cit.title} ({cit.url})\n"
        if finding.contradictions:
            md += f"**Contradictions:**\n"
            for c in finding.contradictions:
                md += f"- {c}\n"
        md += "\n"
    md += "## Gaps\n"
    for gap in report.gaps:
        md += f"- {gap}\n"
    
    with open(f"report_{topic[:40].replace(' ', '_')}.md", "w") as f:
        f.write(md)
    
    print("\nReport saved. Open the .md file to verify quality.")

Verify success: Open the generated Markdown. Check that:

1. Every claim in findings has at least one citation with a real URL
2. Confidence levels match evidence strength (high = multiple independent sources)
3. Contradictions are explicitly noted, not papered over
4. Gaps section is non-empty — honest about what’s unknown
5. Executive summary reads standalone without the full report

Step 8: Harden for production

The tutorial version works for one-offs. Production crews need observability, cost control, and failure recovery.

Add structured logging: Wrap each agent’s execution in a span that logs prompt tokens, completion tokens, latency, and tool calls. Ship to your observability stack (LangSmith, Langfuse, or custom).

Implement circuit breakers: If the search API returns 429 or 5xx, the manager should wait and retry with exponential backoff, not crash the crew. Use tenacity on tool _run methods.

Cache extractions: Content extraction is idempotent. Cache by URL hash (SHA256) with a 7-day TTL. Saves tokens and avoids re-fetching when the manager requests a re-review.

Human review gate: For high-stakes topics, insert a human_input=True task between analysis and writing. The manager presents findings in a dashboard; an analyst approves or requests changes before the writer runs.

Cost attribution: Tag each LLM call with agent_role and task_name. Aggregate per-run to know exactly what a deep-dive costs vs. a scan.

# production_hardening.py (sketch)
import hashlib
import sqlite3
from functools import wraps
from tenacity import retry, stop_after_attempt, wait_exponential

DB_PATH = "research_cache.sqlite"

def init_cache():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS extractions (
            url_hash TEXT PRIMARY KEY,
            url TEXT,
            content TEXT,
            extracted_at TEXT,
            chars INTEGER
        )
    """)
    conn.commit()
    return conn

def cached_extract(url: str, max_chars: int = 8000) -> str:
    conn = init_cache()
    url_hash = hashlib.sha256(url.encode()).hexdigest()
    row = conn.execute("SELECT content FROM extractions WHERE url_hash = ?", (url_hash,)).fetchone()
    if row:
        return row[0][:max_chars]
    # ... fetch and extract ...
    conn.execute("INSERT INTO extractions VALUES (?, ?, ?, datetime('now'), ?)",
                 (url_hash, url, content, len(content)))
    conn.commit()
    return content[:max_chars]

@retry(wait=wait_exponential(multiplier=1, min=2, max=30), stop=stop_after_attempt(3))
def resilient_search(query: str, **kwargs) -> str:
    # Your search tool _run logic here
    pass

Common failure modes and fixes

Scaling patterns

Once the single-topic crew works, you’ll want batch processing. Two patterns:

Parallel topics: Spin up multiple ResearchCrew instances with asyncio.gather, each with its own manager. Share the search/extract cache. Rate-limit at the gateway level.

Recursive decomposition: For massive topics (e.g., “state of AI alignment research”), the manager’s planning task outputs sub-topics. Each sub-topic spawns a child crew. The parent manager aggregates child reports. This is where hierarchical shines — the same manager logic recurses.

# recursive_manager.py (concept)
def decompose_and_conquer(topic: str, depth: int = 0, max_depth: int = 2) -> ResearchReport:
    if depth >= max_depth:
        return ResearchCrew(topic).run()
    
    # Manager creates sub-topics
    subtopics = manager_plan_subtopics(topic)
    
    # Run child crews in parallel
    child_reports = asyncio.gather(*[decompose_and_conquer(st, depth+1) for st in subtopics])
    
    # Parent manager synthesizes
    return manager_synthesize(topic, child_reports)

You now have a working crewai hierarchical crew research tasks pipeline that produces auditable, cited research reports. The hierarchical process isn’t magic — it’s a disciplined delegation loop with a manager that enforces quality gates. Start with the single-topic version, verify the output contract on real topics, then add the production hardening and recursive patterns as your workload demands.