CrewAI Comprehensive Refresher

1. Foundations

Philosophy

CrewAI is designed to mimic a human organization. You don't just write a prompt; you hire an "Agent" with a job title (Role), a mission (Goal), and a personality (Backstory). This constraint reduces hallucinations because the LLM stays "in character".

When to use?

Automating Jobs: "I need a Researcher, a Writer, and an Editor."
Known Processes: You know the steps (A->B->C), you just need AI to do them intelligently.
Tool Delegation: You want agents to autonomously use search, scraping, or internal APIs.

2. Agents & Roles

Agents are the distinct workers. Defining a good agent requires "Prompt Engineering" in the form of a Backstory.

from crewai import Agent

researcher = Agent(
    role='Senior Researcher',
    goal='Uncover groundbreaking developments in {topic}',
    # Backstory is crucial: It tells the LLM HOW to behave and what tone to use.
    backstory="You are a veteran analyst. You are skeptical of hype and prefer data.",
    verbose=True,      # Logs thinking process to console
    memory=True,       # Enables RAG memory (Short/Long term)
    allow_delegation=False, # Can this agent ask others for help?
    tools=[],          # List of tools available to this agent
    max_iter=5         # Safety limit to prevent infinite loops
)

3. Tasks & Outputs

Tasks are the specific assignments. A good task is descriptive and result-oriented.

from crewai import Task

report_task = Task(
    description='Research {topic}. Focus on Q3 trends.',
    # Expected Output is MANDATORY. It guides the LLM on format.
    expected_output='A bulleted list of trends with 3 citations each.',
    agent=researcher,
    output_file='report.md' # Automatically saves the result to a file
)

Advanced: Context Passing

By default, CrewAI's `Process.sequential` passes the output of Task 1 to Task 2. But what if Task 3 needs the output of Task 1 AND Task 2, or if you are skipping tasks?

Use the context parameter to explicit pass outputs from specific previous tasks.

# 1. Define previous tasks
research_task = Task(...)
analysis_task = Task(...)

# 2. Define downstream task with EXPLICIT context
writing_task = Task(
    description="Write a summary combining research and analysis.",
    expected_output="A summary paragraph.",
    agent=writer,
    # The writer will receive the outputs of BOTH tasks as input context
    context=[research_task, analysis_task] 
)

Advanced: Structured Output

For production, you rarely want just text. You want JSON or Python objects. CrewAI integrates Pydantic for this.

from pydantic import BaseModel

class Trend(BaseModel):
    name: str
    impact_score: int

class MarketReport(BaseModel):
    summary: str
    trends: list[Trend]

# The agent will FORCE the output to match this schema
task = Task(
    ...,
    output_pydantic=MarketReport 
    # OR use output_json=True for a raw dict
)

# Accessing the result
result = crew.kickoff()
print(result.pydantic.trends[0].name) # Type-safe access!

Advanced: Async Execution

Tasks usually block. If you have two independent research tasks, run them in parallel to save time.

# These two run simultaneously
task_1 = Task(..., async_execution=True)
task_2 = Task(..., async_execution=True)

# This task waits for both to finish (if it uses them as context)
task_3 = Task(..., context=[task_1, task_2])

Technique: Hierarchical Process

Instead of a relay race (Sequential), use a Boss (Manager). The Manager takes the Goal and breaks it down, assigning tasks to workers dynamically.

from crewai import Process, Crew
from langchain_openai import ChatOpenAI

# The Manager needs a 'smart' model (GPT-4 class) to plan effectively
manager_llm = ChatOpenAI(model="gpt-4")

crew = Crew(
    agents=[researcher, writer],
    tasks=[task1, task2],
    process=Process.hierarchical,        
    manager_llm=manager_llm
)

Technique: Planning Mode

Enabling planning=True adds a special "Plan" step before execution starts. The Crew analyzes all tasks and agents to create a global execution strategy.

crew = Crew(
    agents=[...],
    tasks=[...],
    planning=True, # <--- Adds a planning step at the start
    verbose=True
)

Technique: Memory Systems

CrewAI has a sophisticated memory layer to prevent redundancy.

Short-term: RAG over the current run's outputs.
Long-term: SQLite/ChromaDB storage of past run insights.
Entity: Extractions of key subjects (e.g., "Tesla", "Elon").

crew = Crew(
    ...,
    memory=True, 
    embedder={
        "provider": "openai",
        "config": {"model": 'text-embedding-3-small'}
    }
)

Technique: Callbacks

Hooks for observability or side-effects (e.g., logging to a database).

task_callback: Runs after a specific Task finishes. Good for saving intermediate results.

step_callback: Runs after every single Agent step (thought/action). Good for detailed debugging.

def log_step(step_output):
    print(f"Agent took a step: {step_output}")

agent = Agent(..., step_callback=log_step)

Technique: CrewAI Flows (v0.80+)

For building complex applications, simple Crews aren't enough. Flows allow you to chain Crews, run python logic, and handle state.

from crewai.flow.flow import Flow, listen, start

class ContentPipeline(Flow):
    
    @start()
    def generate_topics(self):
        return ["AI", "Space"]

    @listen(generate_topics)
    def research_topic(self, topics):
        # You can spin up a Crew dynamically here
        crew = Crew(...)
        return crew.kickoff(inputs={"topic": topics[0]})

flow = ContentPipeline()
flow.kickoff()

CrewAI Refresher