crewai-developer

What this skill does

# CrewAI Developer Guide

## Overview

CrewAI is a lean, lightning-fast Python framework for building collaborative AI agent teams and structured workflows. It empowers developers to create autonomous AI agents with specific roles, tools, and goals that work together to tackle complex tasks. This skill covers Crews (autonomous collaboration), Flows (structured orchestration), agents, tasks, and enterprise deployment.

## Core Concepts

### Agents: Specialized Team Members

Agents are autonomous AI units with specific roles, goals, and capabilities.

```python
from crewai import Agent

# Create a research agent
researcher = Agent(
    role='Senior Research Analyst',
    goal='Uncover cutting-edge developments in AI and data science',
    backstory="""You are an expert at a leading tech think tank.
    Your expertise lies in identifying emerging trends and technologies in AI,
    data science, and machine learning.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, scrape_tool]
)

# Create a writer agent
writer = Agent(
    role='Tech Content Strategist',
    goal='Craft compelling content on tech advancements',
    backstory="""You are a renowned content strategist, known for
    your insightful and engaging articles on technology and innovation.
    You transform complex concepts into compelling narratives.""",
    verbose=True,
    allow_delegation=True,
    tools=[write_tool]
)
```

#### Agent Key Properties

```python
agent = Agent(
    role='Role Name',              # The agent's job title
    goal='Specific objective',     # What the agent aims to achieve
    backstory='Background story',  # Context and expertise
    verbose=True,                  # Enable detailed logging
    allow_delegation=False,        # Can delegate tasks to other agents
    tools=[tool1, tool2],         # Available tools
    llm=custom_llm,               # Custom LLM configuration
    max_iter=15,                  # Maximum iterations for task
    max_rpm=10,                   # Rate limit (requests per minute)
    memory=True,                  # Enable memory
    cache=True,                   # Enable response caching
    system_template="template",   # Custom system prompt template
    prompt_template="template",   # Custom prompt template
    response_template="template"  # Custom response template
)
```

### Tasks: Individual Assignments

Tasks define specific work to be completed by agents.

```python
from crewai import Task

# Research task
research_task = Task(
    description="""Conduct a comprehensive analysis of the latest advancements in AI.
    Identify key trends, breakthrough technologies, and potential industry impacts.
    Compile your findings in a detailed report.""",
    expected_output='A comprehensive 3-paragraph report on AI advancements',
    agent=researcher,
    tools=[search_tool],
    output_file='research_report.md'
)

# Writing task
write_task = Task(
    description="""Using the research analyst's report, develop an engaging blog post
    highlighting the most significant AI advancements.
    Make it accessible and engaging for a general audience.""",
    expected_output='A 4-paragraph blog post about AI advancements',
    agent=writer,
    context=[research_task],  # Depends on research_task output
    output_file='blog_post.md'
)
```

#### Task Key Properties

```python
task = Task(
    description='Detailed task description',
    expected_output='Clear output format',
    agent=agent_instance,
    tools=[tool1, tool2],           # Task-specific tools
    context=[previous_task],        # Dependencies
    async_execution=False,          # Run asynchronously
    output_json=OutputClass,        # Structured output (Pydantic)
    output_pydantic=OutputClass,    # Pydantic validation
    output_file='result.txt',       # Save output to file
    callback=callback_function,     # Callback on completion
    human_input=False              # Request human feedback
)
```

### Crews: Organizing Agent Teams

Crews orchestrate agents working together toward a common goal.

```python
from crewai import Crew, Process

# Create a crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,  # or Process.hierarchical
    verbose=True,
    memory=True,
    cache=True,
    max_rpm=10,
    share_crew=False
)

# Kickoff the crew
result = crew.kickoff()
print(result)

# Kickoff with custom inputs
result = crew.kickoff(inputs={
    'topic': 'Artificial Intelligence',
    'audience': 'developers'
})
```

#### Process Types

```python
# Sequential process (tasks run one after another)
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    process=Process.sequential
)

# Hierarchical process (manager delegates to agents)
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    process=Process.hierarchical,
    manager_llm='gpt-4'  # Required for hierarchical
)
```

### Flows: Structured Workflow Orchestration

Flows provide event-driven, deterministic control over execution paths.

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

class BlogPostFlow(Flow):

    @start()
    def fetch_topic(self):
        """Entry point - fetch the topic to write about"""
        print("Starting blog post generation")
        return "AI advancements in 2024"

    @listen(fetch_topic)
    def research_topic(self, topic):
        """Research the topic"""
        print(f"Researching: {topic}")
        # Integrate with Crew for autonomous research
        research_crew = Crew(
            agents=[researcher],
            tasks=[research_task]
        )
        result = research_crew.kickoff(inputs={'topic': topic})
        return result

    @listen(research_topic)
    def write_blog_post(self, research_data):
        """Write the blog post"""
        print("Writing blog post...")
        write_crew = Crew(
            agents=[writer],
            tasks=[write_task]
        )
        result = write_crew.kickoff(inputs={'research': research_data})
        return result

    @listen(write_blog_post)
    def finalize(self, blog_post):
        """Finalize and save"""
        print("Blog post completed!")
        return blog_post

# Execute flow
flow = BlogPostFlow()
result = flow.kickoff()
```

#### Flow State Management

```python
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel

class ArticleState(BaseModel):
    topic: str = ""
    research: str = ""
    draft: str = ""
    final: str = ""

class ArticleFlow(Flow[ArticleState]):

    @start()
    def set_topic(self):
        self.state.topic = "AI Ethics"
        return self.state.topic

    @listen(set_topic)
    def research(self, topic):
        # Research logic
        self.state.research = "Research findings..."
        return self.state.research

    @listen(research)
    def write_draft(self, research):
        self.state.draft = "Draft content..."
        return self.state.draft

# Access state
flow = ArticleFlow()
flow.kickoff()
print(flow.state.topic)
print(flow.state.research)
```

#### Router Pattern

```python
from crewai.flow.flow import Flow, listen, start, router

class ContentFlow(Flow):

    @start()
    def categorize_content(self):
        return "technical"  # or "marketing", "blog"

    @router(categorize_content)
    def route_content(self, category):
        if category == "technical":
            return "write_technical"
        elif category == "marketing":
            return "write_marketing"
        else:
            return "write_blog"

    @listen("write_technical")
    def write_technical_doc(self):
        return "Technical documentation..."

    @listen("write_marketing")
    def write_marketing_copy(self):
        return "Marketing content..."

    @listen("write_blog")
    def write_blog_post(self):
        return "Blog post..."
```

## Tools: Extending Agent Capabilities

### Built-in Tools

```python
from crewai_tools import (
    SerperDevTool,