CrewAI

Minimal Crew: Researcher -> Writer, sequential process

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

search_tool = SerperDevTool()

researcher = Agent(
    role="Senior Research Analyst",
    goal="Find and summarise the latest developments on {topic}",
    backstory="Expert at distilling complex information into clear insights.",
    tools=[search_tool],
    verbose=True,
)

writer = Agent(
    role="Content Writer",
    goal="Write a concise, accurate report based on the research findings",
    backstory="Skilled at turning research into readable prose.",
    verbose=True,
)

research_task = Task(
    description="Research the current state of {topic}. Identify key trends.",
    expected_output="A bullet-point summary of 5-10 key findings with sources.",
    agent=researcher,
)

write_task = Task(
    description="Write a 3-paragraph report based on the research findings.",
    expected_output="A polished 3-paragraph report, ready to publish.",
    agent=writer,
    context=[research_task],  # output of research_task feeds this task
)

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

result = crew.kickoff(inputs={"topic": "CrewAI multi-agent frameworks"})

context=[research_task] is the key wiring: the Writer receives the Researcher's output as context. Without it, agents run in isolation and the Writer has no research to draw from.

Hierarchical Crew with manager LLM and memory

from crewai import Agent, Task, Crew, Process
import os

manager_llm = "claude-sonnet-4-5"  # or any LLM identifier

analyst = Agent(
    role="Data Analyst",
    goal="Extract structured insights from raw data",
    backstory="Precise, detail-oriented data specialist.",
)

validator = Agent(
    role="Quality Validator",
    goal="Verify accuracy and completeness of analysis outputs",
    backstory="Critical thinker who catches errors before they ship.",
)

writer = Agent(
    role="Report Writer",
    goal="Produce a clear, stakeholder-ready report",
    backstory="Expert at translating technical findings into business language.",
)

analyse_task = Task(
    description="Analyse the dataset at {data_path} and extract key metrics.",
    expected_output="Structured JSON with metrics: totals, trends, anomalies.",
)

validate_task = Task(
    description="Check the analysis output for accuracy and completeness.",
    expected_output="Validation report: pass/fail per metric with notes.",
)

report_task = Task(
    description="Write an executive summary based on validated analysis.",
    expected_output="2-page executive summary with charts described in text.",
)

crew = Crew(
    agents=[analyst, validator, writer],
    tasks=[analyse_task, validate_task, report_task],
    process=Process.hierarchical,
    manager_llm=manager_llm,
    memory=True,  # LanceDB-backed memory across calls
    verbose=True,
)

result = crew.kickoff(inputs={"data_path": "/data/q2-report.csv"})

In hierarchical mode, the manager LLM assigns tasks and validates outputs before proceeding. memory=True activates LanceDB-backed memory so agents recall context from earlier calls. Watch the extra LLM call cost per task delegation.

Flow: event-driven routing between Crews with typed state

from crewai.flow.flow import Flow, listen, start, router
from crewai import Crew, Process
from pydantic import BaseModel

class ReportState(BaseModel):
    topic: str = ""
    research_summary: str = ""
    needs_deep_dive: bool = False
    final_report: str = ""

class ReportFlow(Flow[ReportState]):

    @start()
    def kick_off_research(self):
        research_crew = build_research_crew()
        result = research_crew.kickoff(inputs={"topic": self.state.topic})
        self.state.research_summary = str(result)

    @router(kick_off_research)
    def route_by_depth(self):
        # Send to deep dive if summary flagged gaps
        if "insufficient data" in self.state.research_summary.lower():
            return "deep_dive"
        return "write"

    @listen("deep_dive")
    def run_deep_dive(self):
        deep_crew = build_deep_dive_crew()
        result = deep_crew.kickoff(
            inputs={"summary": self.state.research_summary}
        )
        self.state.research_summary += " " + str(result)

    @listen("write")
    @listen("run_deep_dive")
    def write_report(self):
        write_crew = build_write_crew()
        result = write_crew.kickoff(
            inputs={"research": self.state.research_summary}
        )
        self.state.final_report = str(result)

flow = ReportFlow()
flow.kickoff(inputs={"topic": "AI agent orchestration trends 2026"})

@router returns a label that determines which @listen branch executes next, enabling conditional branching without rewriting Crews. Use Flows when the coordination logic (branching, state passing) outgrows what a single Crew process can express.