# CrewAI

Role-based, process-driven multi-agent framework. Optimized for structured team workflows.

---

## Core Concepts

### Agents = Roles
```python
from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Senior Research Analyst",
    goal="Find cutting-edge information on {topic}",
    backstory="""You're an expert researcher with 10 years experience
    in technology trend analysis. You know how to find the signal
    in the noise.""",
    tools=[search_tool, scrape_tool],
    llm="gpt-4o",
    verbose=True,
    allow_delegation=True,
    max_iter=5,
    memory=True
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear, engaging content from research",
    backstory="""You transform complex research into accessible
    technical articles. You excel at structure and clarity.""",
    tools=[],
    llm="gpt-4o",
    verbose=True
)
```

### Tasks = Work Items
```python
research_task = Task(
    description="Research the latest developments in {topic}",
    expected_output="Comprehensive report with 5 key findings, sources, and quotes",
    agent=researcher,
    async_execution=False,  # Sequential by default
    context=["previous_research"]  # Dependencies
)

writing_task = Task(
    description="Write a technical blog post based on research",
    expected_output="2000-word blog post with intro, sections, conclusion",
    agent=writer,
    context=[research_task]  # Depends on research
)
```

### Crews = Teams + Processes
```python
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,  # or Process.hierarchical
    verbose=2,
    memory=True,
    cache=True,
    max_rpm=100,  # Rate limiting
    embedder={
        "provider": "openai",
        "model": "text-embedding-3-small"
    }
)

result = crew.kickoff(inputs={"topic": "AI Agents in 2024"})
```

---

## Processes

### Sequential (Default)
```
Task 1 → Task 2 → Task 3
```
- Each task waits for previous
- Context passed automatically
- Simple, predictable

### Hierarchical
```
Manager Agent
    ├── Worker 1 → Task 1
    ├── Worker 2 → Task 2
    └── Worker 3 → Task 3
```
- Manager agent decomposes, delegates, reviews
- Workers execute in parallel where possible
- Manager reviews and approves

```python
crew = Crew(
    agents=[manager, worker1, worker2],
    tasks=[task1, task2],
    process=Process.hierarchical,
    manager_llm="gpt-4o",  # Separate LLM for manager
    manager_agent=manager  # Or explicit manager agent
)
```

### Process Comparison
| Aspect | Sequential | Hierarchical |
|---|---|---|
| **Control** | Fixed order | Manager decides |
| **Parallelism** | None | Automatic |
| **Flexibility** | Low | High |
| **Complexity** | Simple | Manager prompt critical |
| **Debugging** | Easy | Harder |

---

## Agent Configuration

### Full Agent Parameters
```python
Agent(
    role="Role Title",
    goal="Objective",
    backstory="Character context",
    
    # Tools
    tools=[tool1, tool2],
    tool_call_limit=10,
    
    # LLM
    llm="gpt-4o",
    function_calling_llm="gpt-4o-mini",  # For tools
    
    # Behavior
    verbose=True,
    allow_delegation=True,  # Can delegate to other agents
    max_iter=15,  # Max reasoning loops
    max_rpm=60,   # Rate limit
    
    # Memory
    memory=True,
    embedder={"provider": "openai", "model": "text-embedding-3-small"},
    
    # Callbacks
    step_callback=lambda step: log_step(step),
    
    # Constraints
    max_execution_time=300,  # Seconds
    respect_context_window=True
)
```

---

## Tools

### Built-in Tools
```python
from crewai_tools import (
    SerperDevTool,      # Google search
    ScrapeWebsiteTool,  # Web scraping
    FileReadTool,       # Read files
    FileWriteTool,      # Write files
    DirectoryReadTool,  # List directory
    CSVSearchTool,      # Query CSV
    JSONSearchTool,     # Query JSON
    EXASearchTool,      # Exa.ai search
    YouTubeChannelTool, # YouTube data
    GithubSearchTool,   # GitHub search
)
```

### Custom Tools
```python
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class MyToolInput(BaseModel):
    query: str = Field(description="Search query")

class MyTool(BaseTool):
    name: str = "my_custom_tool"
    description: str = "Description for LLM"
    args_schema: type[BaseModel] = MyToolInput
    
    def _run(self, query: str) -> str:
        # Implementation
        return result

# Use
agent = Agent(tools=[MyTool()], ...)
```

### Tool Calling
- Uses OpenAI function calling format
- Supports parallel tool calls
- Automatic retry on failure

---

## Memory

### Short-term (Conversation)
- Automatic within crew execution
- Passed via task context

### Long-term (Embeddings)
```python
crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,
    embedder={
        "provider": "openai",
        "model": "text-embedding-3-small",
        "config": {"api_key": "..."}
    }
)
# Stores: task outputs, agent reasoning, tool results
# Retrieves: relevant context for new tasks
```

### Entity Memory
```python
# Tracks entities across tasks
entity_memory = True  # Remembers people, companies, concepts
```

---

## Flows (Structured Processes)

```python
from crewai import Flow, Start, Listen, Router

class ResearchFlow(Flow):
    @Start()
    def start_research(self):
        return {"topic": self.inputs["topic"]}
    
    @Listen("start_research")
    def deep_research(self, topic):
        # Parallel research agents
        return {"findings": parallel_research(topic)}
    
    @Router("deep_research")
    def route_by_finding(self, findings):
        if findings["needs_deep_dive"]:
            return "deep_dive"
        return "write"
    
    @Listen("deep_dive")
    def deep_dive(self, findings):
        return {"enhanced": deep_research(findings)}
    
    @Listen("write")
    def write_report(self, findings):
        return {"report": write(findings)}

flow = ResearchFlow()
result = flow.kickoff(inputs={"topic": "AI Agents"})
```

---

## Human-in-the-Loop

### Human Tools
```python
from crewai_tools import HumanInputTool

human_tool = HumanInputTool()

agent = Agent(
    role="Analyst",
    tools=[human_tool],
    # When agent needs human input, calls tool
)
```

### Task-Level Approval
```python
task = Task(
    description="...",
    agent=analyst,
    human_input=True  # Pauses for human review before execution
)
```

---

## Flows (CrewAI Flows - New)

```python
from crewai.flow import Flow, Start, Listen, Router

class MyFlow(Flow):
    @Start()
    def start(self):
        return {"data": "initial"}
    
    @Listen("start")
    def step1(self, state):
        return {"result": process(state["data"])}
    
    @Router("step1")
    def route(self, state):
        if state["result"].needs_review:
            return "review"
        return "finalize"
    
    @Listen("review")
    def human_review(self, state):
        return {"approved": human_approve(state["result"])}
    
    @Listen("finalize")
    def finalize(self, state):
        return {"output": finalize(state["result"])}

flow = MyFlow()
result = flow.kickoff()
```

---

## Deployment

### CrewAI Enterprise
```python
# Hosted platform with:
# - RBAC, audit logs
# - Scaling, queue management
# - Analytics, cost tracking
# - SSO, compliance

from crewai import Crew
crew = Crew(..., enterprise=True)
```

### Self-Hosted
```dockerfile
FROM python:3.11-slim
RUN pip install crewai crewai-tools
COPY . .
CMD ["python", "main.py"]
```

### Kubernetes
```yaml
# Deploy as job or deployment
# Use Redis for shared memory/cache
# Use PostgreSQL for persistence
```

---

## Best Practices

### Agent Design
1. **Specific roles** — "Senior Python Dev" > "Developer"
2. **Clear goals** — Measurable, time-bound
3. **Rich backstories** — Drives persona consistency
4. **Minimal tools** — Only what's needed
5. **Test prompts** — Iterate on backstory/goal

### Task Design
1. **Explicit expected_output** — Enables validation
2. **Single responsibility** — One task, one outcome
3. **Context dependencies** — Use `context=[task]` not implicit
3. **Async where possible** — `async_execution=True` for independent

### Crew Design
1. **Start sequential** — Move to hierarchical only when needed
2. **Limit agents** — 3-5 typical, >7 gets unwieldy
3. **Monitor costs** — `max_rpm`, token tracking
4. **Enable memory** — For multi-session continuity

---

## Debugging

### Verbose Levels
```python
crew = Crew(..., verbose=0)  # Silent
crew = Crew(..., verbose=1)  # Task starts/ends
crew = Crew(..., verbose=2)  # Full reasoning + tool calls
```

### Step Callbacks
```python
def log_step(step):
    print(f"Agent: {step.agent.role}")
    print(f"Action: {step.action}")
    print(f"Result: {step.result}")

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

### Debug Mode
```python
import os
os.environ["CREWAI_DEBUG"] = "1"
# Logs all prompts, responses, tool calls
```

---

## Related Notes

- [[Frameworks/Framework-Comparison.md]] — vs LangGraph, AutoGen
- [[Architectures/MultiAgent-Patterns.md]] — Role-based patterns
- [[MultiAgent/Coordination-Patterns.md]] — Sequential, hierarchical
- [[Frameworks/LangGraph.md]] — Alternative for stateful
- [[Frameworks/AutoGen.md]] — Alternative for conversational
- [[Architectures/Human-in-the-Loop.md]] — Human tools
- [[ContextEngineering/Memory-Systems.md]] — Memory patterns
- [[Production/Observability.md]] — Monitoring

---

*Sources: CrewAI docs (docs.crewai.com), CrewAI GitHub, CrewAI Enterprise docs, CrewAI Flows docs, crewai-tools package, community examples, production case studies.*