CrewAI

The mental model

A crew of specialists, not one generalist.

CrewAI models a job as a small team. Each Agent is defined in natural language: a role, a goal, and a backstory that get injected straight into the system prompt and shape how the model behaves. You then write Tasks, each owned by one agent, and hand them to a Crew that runs them under a Process (sequential or hierarchical). The pitch is speed: you describe the roles and the work in prose, and the framework handles the collaboration loop, passing one task's output as context to the next.

That prose-first convenience is also the trade-off. A Crew decides a lot for you, so you own less of the control flow than you would in LangGraph. CrewAI answers this with Flows, a newer event-driven layer where you write plain Python methods wired by @start and @listen, thread explicit state, and call Crews only where fuzziness actually helps. Crews are the autonomous team; Flows are the deterministic conductor. Real systems use both.

flowchart TB
  IN(["inputs: topic"]) --> C{{"Crew
process sequential"}}
  C --> A1["Agent
Researcher"]
  A1 --> T1["Task
research"]
  T1 -->|"output as context"| T2["Task
write"]
  T2 --> A2["Agent
Writer"]
  A2 --> OUT(["CrewOutput"])
  classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B;
  classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B;
  class A1,A2 model;
  class C,T1,T2 code;
    

Install & setup

One package, one API key.

CrewAI is Python-only (3.10 to 3.13 as of the 1.x line). Install the core, and add the tools extra for the prebuilt toolkit. CrewAI talks to models through LiteLLM, so a provider is just a string like anthropic/claude-sonnet-4-5 or openai/gpt-4o, and the matching env key is all it needs.

# core + the prebuilt tool library
pip install crewai "crewai[tools]"

# set the key for whichever provider your model string names
export ANTHROPIC_API_KEY="sk-ant-..."
# or OPENAI_API_KEY=... for openai/ models, plus SERPER_API_KEY for web search

CrewAI also ships a scaffolding CLI (crewai create crew my_project) that generates a project with YAML files for agents and tasks. This page teaches the pure-Python API instead, because seeing every object constructed in code is the fastest way to understand what the YAML is doing for you.

Hello world

One agent, one task, one crew.

The smallest useful CrewAI program is three objects: an Agent, a Task that agent owns, and a Crew that runs it. Curly-brace placeholders in the prose are filled from the inputs you pass to kickoff().

from crewai import Agent, Task, Crew

researcher = Agent(
    role="Senior Researcher",
    goal="Explain {topic} clearly and accurately",
    backstory="You are a meticulous analyst who prizes primary sources.",
    llm="anthropic/claude-sonnet-4-5",   # LiteLLM-style model string
)

brief = Task(
    description="Write a 5-bullet primer on {topic} for a busy engineer.",
    expected_output="Exactly 5 concise bullet points, no preamble.",
    agent=researcher,
)

crew = Crew(agents=[researcher], tasks=[brief])

result = crew.kickoff(inputs={"topic": "vector databases"})
print(result.raw)   # the final text; result is a CrewOutput

role, goal, and backstory are not comments, they become the system prompt

The core · Agent

An agent is a persona plus tools.

The three prose fields do the heavy lifting. role is who the agent is, goal is what it optimises for, and backstory is the context that colours its judgement. Beyond those, the arguments you reach for most are llm, tools, verbose, and allow_delegation. Keep allow_delegation=False unless an agent genuinely needs to hand work to a teammate, because delegation multiplies model calls fast.

from crewai import Agent, LLM
from crewai_tools import SerperDevTool

# the LLM class gives you temperature, max_tokens, etc.
llm = LLM(model="anthropic/claude-sonnet-4-5", temperature=0.2)

analyst = Agent(
    role="Market Analyst",
    goal="Surface hard numbers on {sector} demand",
    backstory="Ex-equity researcher who never states a figure without a source.",
    llm=llm,
    tools=[SerperDevTool()],   # web search; needs SERPER_API_KEY
    allow_delegation=False,     # do NOT let it spawn sub-tasks by default
    max_iter=15,               # cap the reason-act loop (default 20)
    verbose=True,             # print the agent's thinking to the console
)

The core · Task

Describe the work and pin the output shape.

A Task is a unit of work owned by one agent. description is the instruction and expected_output is your acceptance test in prose: the more concrete it is, the more predictable the result. The most valuable production argument is output_pydantic, which forces the result into a typed model you can trust downstream instead of parsing free text. Use context to feed earlier task outputs into a later task explicitly.

from crewai import Task
from pydantic import BaseModel

class Finding(BaseModel):
    headline: str
    evidence: list[str]
    confidence: float

research = Task(
    description="Research {sector} demand for 2026 and cite each claim.",
    expected_output="A headline, 3 to 5 evidence strings, and a 0-1 confidence.",
    agent=analyst,
    output_pydantic=Finding,      # result.pydantic is a typed Finding
)

summary = Task(
    description="Turn the research into a one-paragraph exec summary.",
    expected_output="A single tight paragraph, no bullet points.",
    agent=writer,
    context=[research],           # pass the research output in as context
    output_file="summary.md",       # also write it to disk
)

After a run, a task's result is on its output object: research.output.pydantic.headline gives you the typed field, and research.output.raw the plain text. In a sequential crew, each task's output is automatically threaded to the next even without context, but naming it explicitly makes the dependency readable.

The core · Crew & Process

The Crew runs the tasks. The Process decides how.

A Crew binds agents and tasks together and executes them under a Process. There are two. Sequential runs tasks top to bottom, each output feeding the next; it is predictable and the right default. Hierarchical adds a manager that plans, delegates each task to the best agent, and reviews the results; you must supply a manager_llm or a custom manager_agent, and CrewAI creates the manager for you.

from crewai import Crew, Process

# sequential: a straight line, task1 -> task2 -> task3
crew = Crew(
    agents=[analyst, writer],
    tasks=[research, summary],
    process=Process.sequential,   # the default; explicit here for clarity
    verbose=True,
)

# hierarchical: a manager delegates and reviews. tasks need no fixed agent.
managed = Crew(
    agents=[analyst, writer],
    tasks=[research, summary],
    process=Process.hierarchical,
    manager_llm="anthropic/claude-sonnet-4-5",   # REQUIRED for hierarchical
)

result = crew.kickoff(inputs={"sector": "solar storage"})
print(result.raw)                 # final output
print(result.token_usage)         # prompt/completion tokens for the whole run

flowchart TB
  subgraph SEQ["Process sequential"]
    direction LR
    s1["task 1"] --> s2["task 2"] --> s3["task 3"]
  end
  subgraph HIER["Process hierarchical"]
    direction TB
    M{"Manager
plans and reviews"}
    M -->|"delegate"| w1["Agent A"]
    M -->|"delegate"| w2["Agent B"]
    w1 -->|"result"| M
    w2 -->|"result"| M
  end
  classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B;
  classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B;
  class s1,s2,s3 code;
  class M,w1,w2 model;
    

To run the same crew over many inputs, use crew.kickoff_for_each(inputs=[{...}, {...}]), or the native async await crew.kickoff_async(...). Each call returns a fresh CrewOutput; a Crew is stateless by default and runs once to completion.

Tools

Prebuilt tools, or your own function.

Tools are how agents touch the outside world. The crewai_tools package ships dozens ready to use, like SerperDevTool for web search, ScrapeWebsiteTool, FileReadTool, and PDFSearchTool. For your own logic, the fastest path is the @tool decorator: the docstring becomes the tool description the model reads.

from crewai.tools import tool

@tool("Order lookup")
def lookup_order(order_id: str) -> dict:
    """Look up an order status by its ID. Use for any order question."""
    # validate + authorize here; the model only PROPOSES the call
    return {"id": order_id, "status": "shipped"}

agent = Agent(role="Support Agent", goal="Resolve order questions",
              backstory="Patient, precise, never guesses an order state.",
              tools=[lookup_order])

When you need typed, validated arguments, subclass BaseTool and declare an args_schema. This is the production-grade shape: the schema rejects malformed calls before your code runs.

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

class RefundInput(BaseModel):
    order_id: str = Field(..., description="The order to refund")
    amount: float = Field(..., description="Amount in GBP, must be > 0")

class RefundTool(BaseTool):
    name: str = "Issue refund"
    description: str = "Refund an order. Only for verified, approved refunds."
    args_schema: Type[BaseModel] = RefundInput

    def _run(self, order_id: str, amount: float) -> str:
        if amount <= 0:
            return "Rejected: amount must be positive."
        # side effect gated behind your own checks
        return f"Refunded {amount} for {order_id}."

Memory

Turn it on with one flag.

By default a Crew is stateless: each run starts blank. Pass memory=True and CrewAI adds three complementary layers. Short-term memory holds the current run's context via embeddings, so later tasks recall what earlier ones found. Long-term memory persists across runs in a local SQLite store, so a crew improves as it sees more work. Entity memory tracks the people, places, and things mentioned, so references stay consistent. Memory needs an embedder; it defaults to OpenAI, so set the key or point it elsewhere.

crew = Crew(
    agents=[analyst, writer],
    tasks=[research, summary],
    memory=True,                 # short-term + long-term + entity
    embedder={                    # optional: override the default OpenAI embedder
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"},
    },
)

CrewAI is consolidating these into a single unified Memory API in the 1.x line, but the memory=True switch remains the simplest way to enable recall, and it is what most projects still use.

Flows

The deterministic conductor.

A Crew is autonomous: you point it at a goal and it decides the moves. When you need to own the sequence, branch on results, or loop, reach for a Flow. A Flow is a plain Python class whose methods are wired by decorators. @start marks an entry point, @listen(x) runs a method when x finishes and receives its return value, and @router branches on a result. State lives on self.state, best typed with a Pydantic model, and it threads through every step. You call .kickoff() to run the whole thing.

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

class ReportState(BaseModel):
    topic: str = ""
    draft: str = ""
    approved: bool = False

class ReportFlow(Flow[ReportState]):
    @start()
    def gather(self):
        # a Crew can run INSIDE a flow step where fuzziness helps
        self.state.draft = research_crew.kickoff(
            inputs={"topic": self.state.topic}).raw

    @router(gather)
    def gate(self):
        # deterministic branch: code owns the decision
        return "ok" if len(self.state.draft) > 200 else "retry"

    @listen("ok")
    def publish(self):
        self.state.approved = True
        return f"Published: {self.state.draft[:60]}..."

flow = ReportFlow()
result = flow.kickoff(inputs={"topic": "tidal energy"})

Use or_(a, b) to fire when either upstream method finishes and and_(a, b) to wait for both. Because state is explicit and the wiring is code, a Flow is testable and debuggable in a way a bare Crew is not; it is CrewAI's answer to the control question that pushes many teams toward LangGraph.

flowchart TB
  ST(["kickoff"]) --> G["gather
runs a Crew"]
  G --> R{"router gate"}
  R -->|"ok"| P["publish"]
  R -->|"retry"| G
  P --> E(["done"])
  classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B;
  classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B;
  class G model;
  class R,P code;
    

Crews vs Flows

Autonomy versus control, pick per step.

This is the single most important distinction in CrewAI, and the one beginners miss. A Crew hands the model the wheel; a Flow keeps it in code and calls a Crew only where open-ended reasoning earns its cost. They compose: a Flow step can kick off a Crew, and that is the recommended production shape.

Putting it together

A research-and-write crew, end to end.

Two agents with sharp, non-overlapping roles. A researcher gathers cited facts into a typed Finding; a writer turns that into a short brief. The research output is threaded to the writer as context, memory is on so entities stay consistent, and the whole thing runs from a single kickoff.

from crewai import Agent, Task, Crew, Process, LLM
from crewai_tools import SerperDevTool
from pydantic import BaseModel

llm = LLM(model="anthropic/claude-sonnet-4-5", temperature=0.3)

class Finding(BaseModel):
    headline: str
    evidence: list[str]

researcher = Agent(
    role="Senior Researcher",
    goal="Find cited, current facts about {topic}",
    backstory="You never state a claim without a source and a date.",
    llm=llm, tools=[SerperDevTool()], allow_delegation=False,
)

writer = Agent(
    role="Technical Writer",
    goal="Turn findings into a crisp brief for an engineer",
    backstory="You write tight, jargon-light prose and cut every spare word.",
    llm=llm, allow_delegation=False,
)

research = Task(
    description="Research {topic}. Cite each claim with a source.",
    expected_output="A headline and 3 to 5 evidence strings.",
    agent=researcher,
    output_pydantic=Finding,
)

write = Task(
    description="Write a 150-word brief on {topic} from the research.",
    expected_output="One tight paragraph, no headings.",
    agent=writer,
    context=[research],           # feed the typed findings in
    output_file="brief.md",
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research, write],
    process=Process.sequential,
    memory=True,
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "on-device LLM inference"})
print(result.raw)                         # the final brief
print(research.output.pydantic.headline)  # typed access to step 1
print(result.token_usage)                 # cost accounting for the run

two roles, one line each of real judgement, and the framework runs the relay

Production

From notebook to service.

• Wrap Crews in a Flow so control flow, retries, and branching live in code you can test, not in agent improvisation.
• Force typed output with output_pydantic on any task another system consumes, so you validate instead of parse.
• Watch cost: read result.token_usage after every run, cap the loop with max_iter and max_rpm, and keep allow_delegation=False unless a role truly needs it.
• Trace runs with a callback or an observability integration (CrewAI supports OpenTelemetry-style tracing and tools like Langfuse and AgentOps) so you can see each agent step, its tokens, and its latency.
• Gate side effects inside the tool. Validate arguments and check authorization in _run before anything touches money or data you cannot un-send.
• Pin the version. The 1.x API is stable but moving; pin crewai==1.15.* and read the release notes before you bump.

Gotchas

What trips people up.

Reach for it when

• A task genuinely splits into distinct roles, like a researcher, a writer, and a critic.
• You value time-to-first-run and want a collaborating team standing today.
• You are prototyping a multi-agent idea and need to test the shape cheaply before investing in something heavier.
• You want autonomy in a few steps but determinism around them: a Flow with Crews dropped in.

Skip it when

• It is really a single agent calling one or two tools: a vendor SDK or Pydantic AI is a lighter path.
• You need fine-grained control and durable, checkpointed state with human-in-the-loop: LangGraph gives you the explicit graph.
• You need predictable cost: multi-agent multiplies model calls, and a Crew gives you less control over each one than a graph does.

The principle it teaches

Where it fits your labs

A Week 4 tool, and a useful cautionary contrast back in Week 1. When you sketch a multi-agent architecture, CrewAI is the fast way to try it; put it next to LangGraph, which you met in Week 1 as the control-first alternative. CrewAI optimises for how quickly a team stands up, LangGraph for how much of the flow you own. The course asks you to prove the crew earns its extra cost against a single agent before it ships.