CrewAI - Multi-Agent AI Framework

Table of Contents

Introduction

CrewAI is a cutting-edge framework for orchestrating autonomous AI agents that work together as a collaborative team. Inspired by real-world crew dynamics, it enables developers to create sophisticated multi-agent systems where each agent has a specific role, expertise, and set of tools.

Key features of CrewAI:

• Role-based agents: Define agents with specific roles, goals, and backstories
• Task management: Create structured tasks with expected outputs and dependencies
• Flexible orchestration: Sequential, parallel, and hierarchical execution modes
• Tool ecosystem: Rich set of built-in tools plus custom tool creation
• Memory systems: Short-term, long-term, and entity memory for context persistence
• Delegation: Agents can delegate tasks to other agents when appropriate

Multi-agent systems outperform single-agent approaches for complex tasks because they allow specialization, parallel processing, and collaborative problem-solving — much like a human team.

In this tutorial, you will learn how to create agents, define tasks, orchestrate crews, integrate tools, and build real-world multi-agent applications.

Prerequisites

• Python 3.10 or higher
• An OpenAI API key (or other supported LLM provider)
• Basic understanding of Large Language Models (LLMs)
• Familiarity with Python async patterns (for parallel execution)

Installation and Setup

# Install CrewAI with all tools
pip install crewai[tools]

Or minimal installation
pip install crewai

Set up environment variables
export OPENAIAPIKEY="your-openai-api-key"

Optional: for other LLM providers
export ANTHROPICAPIKEY="your-anthropic-key"

Verify the installation:

import crewai
print(f"CrewAI version: {crewai.version}")

from crewai import Agent, Task, Crew
print("All core components imported successfully")

Core Concepts

CrewAI is built around three core abstractions:

from crewai import Agent, Task, Crew, Process

The simplest possible crew
agent = Agent(
    role="Greeting Agent",
    goal="Greet the user warmly",
    backstory="You are a friendly agent that welcomes users.",
    verbose=True,
)

task = Task(
    description="Greet the user and wish them a productive day.",
    expectedoutput="A warm, personalized greeting message.",

    agent=agent,
)

crew = Crew(
    agents=[agent],
    tasks=[task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result)

from crewai import Agent
from crewaitools import SerperDevTool, WebsiteSearchTool

Research analyst agent
researchanalyst = Agent(

    role="Senior Research Analyst",
    goal="Discover and analyze the latest trends and developments "
         "in the given topic with meticulous attention to detail",
    backstory=(
        "You are a veteran research analyst with 15 years of experience "
        "in technology and market research. You are known for your ability "
        "to find obscure but critical information and synthesize it into "
        "clear, actionable insights. You always verify information from "
        "multiple sources before drawing conclusions."
    ),
    tools=[SerperDevTool(), WebsiteSearchTool()],
    verbose=True,
    allowdelegation=True,
    maxiter=10,

    maxrpm=15,  # Rate limit: max requests per minute
)

Technical writer agent
technicalwriter = Agent(

    role="Senior Technical Writer",
    goal="Transform complex research findings into clear, engaging, "
         "and well-structured technical content",
    backstory=(
        "You are an award-winning technical writer who has authored "
        "numerous articles for top tech publications. You excel at "
        "making complex topics accessible to a broad audience while "
        "maintaining technical accuracy. Your writing is known for "
        "its clarity, logical flow, and practical examples."
    ),
    verbose=True,
    allowdelegation=False,
    maxiter=5,

)

Quality assurance agent
qareviewer = Agent(
    role="Quality Assurance Reviewer",
    goal="Ensure all content meets the highest standards of accuracy, "
         "completeness, and readability",
    backstory=(
        "You are a meticulous QA specialist who has reviewed thousands "
        "of technical documents. You have an eye for detail and can "
        "spot factual errors, logical inconsistencies, and unclear "
        "explanations. You provide constructive feedback that helps "
        "improve the overall quality of deliverables."
    ),
    verbose=True,
    allowdelegation=True,

    maxiter=5,
)

Configuring Agent LLMs

from crewai import Agent, LLM

Use different LLMs for different agents
gpt4llm = LLM(

    model="openai/gpt-4o",
    temperature=0.7,
)

claudellm = LLM(
    model="anthropic/claude-sonnet-4-20250514",
    temperature=0.5,
)

Cost-effective agent for simple tasks
simpleagent = Agent(

    role="Data Formatter",
    goal="Format raw data into structured outputs",
    backstory="You are efficient at data formatting.",
    llm=LLM(model="openai/gpt-4o-mini", temperature=0.1),
)

Premium agent for complex reasoning
reasoningagent = Agent(
    role="Strategic Advisor",
    goal="Provide deep strategic analysis",
    backstory="You are a world-class strategic thinker.",
    llm=gpt4llm,

)

from crewai import Task

Research task with detailed instructions
researchtask = Task(
    description=(
        "Conduct a comprehensive research analysis on '{topic}'. "
        "Your research should cover:\n"
        "1. Current state of the technology/field\n"
        "2. Key players and their contributions\n"
        "3. Recent breakthroughs and developments (last 6 months)\n"
        "4. Market size and growth projections\n"
        "5. Challenges and limitations\n"
        "6. Future outlook and predictions\n\n"
        "Use at least 5 different sources. Cross-reference claims "
        "across multiple sources for accuracy."
    ),
    expectedoutput=(

        "A detailed research report in markdown format with:\n"
        "- Executive summary (200 words)\n"
        "- Findings organized by the 6 areas above\n"
        "- Data tables where applicable\n"
        "- Source citations\n"
        "- Confidence ratings for each finding (High/Medium/Low)"
    ),
    agent=researchanalyst,
    outputfile="researchreport.md",
)

Writing task that depends on research
writingtask = Task(

    description=(
        "Using the research findings provided, write a comprehensive "
        "technical article about '{topic}'. The article should:\n"
        "- Be 2000-3000 words\n"
        "- Include an engaging introduction\n"
        "- Present information in a logical flow\n"
        "- Include practical examples and code snippets where applicable\n"
        "- End with actionable conclusions\n"
        "- Be suitable for a technical blog audience"
    ),
    expectedoutput=(
        "A polished, publication-ready technical article in markdown "
        "format, with proper headings, code blocks, and a natural flow."
    ),
    agent=technicalwriter,

    context=[researchtask],  # This task uses output from researchtask

    outputfile="article.md",
)

Review task
reviewtask = Task(

    description=(
        "Review the technical article for:\n"
        "1. Factual accuracy against the research report\n"
        "2. Logical consistency and flow\n"
        "3. Grammar and readability\n"
        "4. Completeness of coverage\n"
        "5. Technical accuracy of code examples\n\n"
        "Provide a detailed review with specific suggestions for improvement."
    ),
    expectedoutput=(
        "A structured review document with:\n"
        "- Overall quality score (1-10)\n"
        "- List of issues found (categorized by severity)\n"
        "- Specific improvement suggestions\n"
        "- Final recommendation (Approve/Revise/Reject)"
    ),
    agent=qareviewer,

    context=[researchtask, writingtask],

    outputfile="review.md",
)

Tasks with Callbacks

def ontaskcomplete(output):
    """Callback fired when a task completes."""
    print(f"Task completed!")
    print(f"Output length: {len(str(output))} characters")

    # Save metrics, send notifications, etc.
    with open("tasklog.txt", "a") as f:

        f.write(f"Completed at: {output.metadata.get('timestamp', 'N/A')}\n")

researchtask = Task(
    description="Research the topic thoroughly.",
    expectedoutput="A comprehensive research report.",

    agent=researchanalyst,
    callback=ontaskcomplete,
)

Crew Orchestration

The Crew brings agents and tasks together and manages execution.

from crewai import Crew, Process

Sequential crew: tasks run one after another
sequentialcrew = Crew(

    agents=[researchanalyst, technicalwriter, qareviewer],
    tasks=[researchtask, writingtask, reviewtask],

    process=Process.sequential,
    verbose=True,
    memory=True,           # Enable memory
    maxrpm=20,            # Global rate limit
    sharecrew=False,      # Don't share data with CrewAI

)

Kick off the crew with input variables
result = sequentialcrew.kickoff(
    inputs={"topic": "Large Language Model Agents in 2026"}
)

print(f"Final output:\n{result}")
print(f"Token usage: {result.tokenusage}")



Hierarchical crew: a manager agent coordinates others
hierarchicalcrew = Crew(
    agents=[researchanalyst, technicalwriter, qareviewer],

    tasks=[researchtask, writingtask, reviewtask],
    process=Process.hierarchical,
    managerllm=LLM(model="openai/gpt-4o"),

    verbose=True,
)

result = hierarchicalcrew.kickoff(
    inputs={"topic": "Multi-Agent Systems for Enterprise"}
)

Running Crews Asynchronously

import asyncio

async def runmultiplecrews():
    """Run multiple crews concurrently."""
    topics = [
        "AI in Healthcare 2026",
        "Quantum Computing Applications",
        "Autonomous Robotics Trends",
    ]

    tasksasync = []

    for topic in topics:
        crew = Crew(
            agents=[researchanalyst, technicalwriter],

            tasks=[researchtask, writingtask],

            process=Process.sequential,
            verbose=False,
        )
        tasksasync.append(
            crew.kickoffasync(inputs={"topic": topic})

        )

    results = await asyncio.gather(tasksasync)

    for topic, result in zip(topics, results):
        print(f"\n{topic}: {len(str(result))} chars generated")

    return results

Run it
results = asyncio.run(runmultiplecrews())

from crewaitools import (
    SerperDevTool,
    WebsiteSearchTool,
    FileReadTool,
    DirectoryReadTool,
    CodeInterpreterTool,
)
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type


Built-in tools
searchtool = SerperDevTool()              # Google search
websitetool = WebsiteSearchTool()          # Search within websites

filetool = FileReadTool()                  # Read files
dirtool = DirectoryReadTool()              # Read directories

codetool = CodeInterpreterTool()           # Execute Python code


Custom tool: Database Query
class DatabaseQueryInput(BaseModel):
    """Input schema for database query tool."""
    query: str = Field(..., description="SQL query to execute")
    database: str = Field(default="main", description="Database name")


class DatabaseQueryTool(BaseTool):
    name: str = "Database Query Tool"
    description: str = (
        "Execute SQL queries against the database. Use this to fetch "
        "data, statistics, or any structured information stored in "
        "our databases."
    )
    argsschema: Type[BaseModel] = DatabaseQueryInput


    def run(self, query: str, database: str = "main") -> str:
        """Execute the database query and return results."""
        import sqlite3

        conn = sqlite3.connect(f"{database}.db")
        cursor = conn.cursor()

        try:
            cursor.execute(query)
            results = cursor.fetchall()
            columns = [desc[0] for desc in cursor.description] if cursor.description else []

            if not results:
                return "Query returned no results."

            # Format as markdown table
            header = "| " + " | ".join(columns) + " |"
            separator = "| " + " | ".join(["---"]  len(columns)) + " |"
            rows = []
            for row in results:
                rows.append("| " + " | ".join(str(v) for v in row) + " |")

            return f"{header}\n{separator}\n" + "\n".join(rows)

        except Exception as e:
            return f"Query error: {str(e)}"
        finally:
            conn.close()


Custom tool: API Client
class APIClientTool(BaseTool):
    name: str = "API Client"
    description: str = "Make HTTP requests to external APIs."

    def run(self, url: str, method: str = "GET", payload: str = None) -> str:

        import requests
        import json

        try:
            if method.upper() == "GET":
                response = requests.get(url, timeout=30)
            elif method.upper() == "POST":
                data = json.loads(payload) if payload else {}
                response = requests.post(url, json=data, timeout=30)
            else:
                return f"Unsupported method: {method}"

            return json.dumps(response.json(), indent=2)

        except Exception as e:
            return f"API request failed: {str(e)}"


Assign tools to agents
dataanalyst = Agent(
    role="Data Analyst",
    goal="Analyze data from databases and APIs to extract insights",
    backstory="You are an expert data analyst skilled in SQL and data interpretation.",
    tools=[DatabaseQueryTool(), APIClientTool(), codetool],

    verbose=True,
)

from crewai import Crew, Process, Agent, Task

Sequential: tasks run in order, each building on the previous
def createsequentialpipeline():

    """Pipeline where each step depends on the previous."""
    researcher = Agent(
        role="Researcher",
        goal="Gather comprehensive information",
        backstory="Expert researcher with attention to detail.",
    )

    analyst = Agent(
        role="Analyst",
        goal="Analyze data and extract patterns",
        backstory="Senior data analyst with statistical expertise.",
    )

    writer = Agent(
        role="Report Writer",
        goal="Create clear, well-structured reports",
        backstory="Technical writer with publication experience.",
    )

    task1 = Task(
        description="Research {topic} thoroughly.",
        expectedoutput="Raw research data and findings.",
        agent=researcher,
    )

    task2 = Task(
        description="Analyze the research findings for key patterns and insights.",
        expectedoutput="Analytical summary with key insights.",

        agent=analyst,
        context=[task1],
    )

    task3 = Task(
        description="Write a final report combining research and analysis.",
        expectedoutput="A polished final report.",
        agent=writer,
        context=[task1, task2],
    )

    crew = Crew(
        agents=[researcher, analyst, writer],
        tasks=[task1, task2, task3],
        process=Process.sequential,
        verbose=True,
    )

    return crew


Parallel-like: independent tasks that don't depend on each other
def createparallelresearchcrew():

    """Multiple researchers working on different sub-topics simultaneously."""
    marketresearcher = Agent(
        role="Market Researcher",
        goal="Analyze market dynamics and competitive landscape",
        backstory="Market research specialist.",
    )

    techresearcher = Agent(

        role="Technical Researcher",
        goal="Analyze technical capabilities and innovations",
        backstory="Deep tech research specialist.",
    )

    synthesizer = Agent(
        role="Research Synthesizer",
        goal="Combine multiple research streams into a cohesive report",
        backstory="Expert at synthesizing diverse information sources.",
    )

    markettask = Task(
        description="Research the market landscape for {topic}.",
        expectedoutput="Market analysis report.",

        agent=marketresearcher,
    )

    techtask = Task(

        description="Research the technical aspects of {topic}.",
        expectedoutput="Technical analysis report.",
        agent=techresearcher,

    )

    synthesistask = Task(
        description="Synthesize the market and technical research into one report.",
        expectedoutput="Comprehensive combined report.",

        agent=synthesizer,
        context=[markettask, techtask],

    )

    crew = Crew(
        agents=[marketresearcher, techresearcher, synthesizer],

        tasks=[markettask, techtask, synthesistask],
        process=Process.sequential,
        verbose=True,
    )

    return crew

Memory and Context Management

CrewAI supports multiple memory types for maintaining context across interactions.

from crewai import Crew, Process

Enable all memory types
crewwithmemory = Crew(
    agents=[researchanalyst, technicalwriter],
    tasks=[researchtask, writingtask],
    process=Process.sequential,
    memory=True,              # Enable memory system
    embedder={                # Configure embedding model for memory
        "provider": "openai",
        "config": {
            "model": "text-embedding-3-small",
        }
    },
    verbose=True,
)

Memory types explained:
1. Short-term memory: Context within a single crew execution
- Automatically shares relevant context between agents
- Tracks task outputs for dependent tasks

2. Long-term memory: Persists across multiple executions
- Stores successful patterns and strategies
- Helps agents improve over time

3. Entity memory: Tracks information about specific entities
- Remembers details about people, companies, concepts
- Builds a knowledge graph over time


Using memory in a multi-run scenario
def iterativeresearch(topic, numiterations=3):
    """Run the crew multiple times, building on previous knowledge."""
    crew = Crew(
        agents=[researchanalyst, technicalwriter],
        tasks=[researchtask, writingtask],
        process=Process.sequential,
        memory=True,
        verbose=True,
    )

    results = []
    for i in range(numiterations):

        print(f"\n{'='50}")

        print(f"Iteration {i+1}/{numiterations}")
        print(f"{'='50}")

        refinedtopic = f"{topic} - Iteration {i+1}: "

        if i == 0:
            refinedtopic += "Initial broad research"
        elif i == 1:
            refinedtopic += "Deep dive into key findings"

        else:
            refinedtopic += "Final synthesis and recommendations"

        result = crew.kickoff(inputs={"topic": refinedtopic})

        results.append(result)

    return results

# Enable delegation for collaborative problem-solving
projectmanager = Agent(
    role="Project Manager",
    goal="Coordinate the team to deliver high-quality results on time",
    backstory=(
        "You are a seasoned project manager who excels at breaking "
        "down complex projects into manageable tasks. You know when "
        "to delegate work to specialists and when to handle things yourself."
    ),
    allowdelegation=True,  # Can delegate to other agents

    verbose=True,
)

frontenddev = Agent(
    role="Frontend Developer",
    goal="Build responsive and accessible user interfaces",
    backstory="Expert React/TypeScript developer with 8 years of experience.",
    allowdelegation=False,  # Specialist: handles own work

    verbose=True,
)

backenddev = Agent(
    role="Backend Developer",
    goal="Design and implement robust APIs and data systems",
    backstory="Senior Python/FastAPI developer with distributed systems expertise.",
    allowdelegation=False,

    verbose=True,
)

The PM can delegate parts of a complex task to specialists
planningtask = Task(
    description=(
        "Plan and coordinate the development of a new feature: {feature}. "
        "Break it down into frontend and backend components. "
        "Delegate the technical design to the appropriate specialists. "
        "Synthesize their inputs into a comprehensive implementation plan."
    ),
    expectedoutput=(

        "A detailed implementation plan with:\n"
        "- Architecture overview\n"
        "- Frontend components and their specifications\n"
        "- Backend API endpoints and data models\n"
        "- Integration points\n"
        "- Timeline estimate"
    ),
    agent=projectmanager,
)

crew = Crew(
    agents=[projectmanager, frontenddev, backenddev],

    tasks=[planningtask],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(
    inputs={"feature": "Real-time collaborative document editor"}
)

Real-World Use Case: Research Agent Team

A complete research team that can investigate any topic and produce a publication-ready report.

from crewai import Agent, Task, Crew, Process
from crewaitools import SerperDevTool, WebsiteSearchTool



def createresearchteam(topic: str):

    """Create a full research team for comprehensive topic analysis."""

    # Tools
    search = SerperDevTool()
    web = WebsiteSearchTool()

    # Agents
    leadresearcher = Agent(
        role="Lead Research Scientist",
        goal=f"Conduct thorough primary research on {topic}",
        backstory=(
            "You lead a research team at a top think tank. You are known "
            "for exhaustive research methodology and finding primary sources."
        ),
        tools=[search, web],
        allowdelegation=True,

        maxiter=15,
        verbose=True,
    )

    factchecker = Agent(

        role="Fact Checker",
        goal="Verify all claims and statistics with authoritative sources",
        backstory=(
            "You are a professional fact-checker who has worked for "
            "major news organizations. You never let unverified claims "
            "pass through."
        ),
        tools=[search, web],
        verbose=True,
    )

    editor = Agent(
        role="Senior Editor",
        goal="Produce a polished, publication-ready research report",
        backstory=(
            "You are a senior editor at a leading research publication. "
            "You transform raw research into clear, compelling narratives."
        ),
        verbose=True,
    )

    # Tasks
    researchtask = Task(
        description=(
            f"Conduct comprehensive research on: {topic}\n\n"
            "Requirements:\n"
            "- Identify at least 10 relevant sources\n"
            "- Cover historical context, current state, and future outlook\n"
            "- Include quantitative data (statistics, market figures)\n"
            "- Note contrasting viewpoints and debates in the field\n"
            "- Flag any claims that need fact-checking"
        ),
        expectedoutput="Detailed research notes with source citations.",

        agent=leadresearcher,
    )

    factchecktask = Task(
        description=(
            "Review all claims in the research findings. For each claim:\n"
            "1. Verify with at least one additional independent source\n"
            "2. Rate confidence: Confirmed / Likely / Unverified / Disputed\n"
            "3. Note any corrections needed\n"
            "4. Add missing context where necessary"
        ),
        expectedoutput="Fact-check report with verification status for each claim.",

        agent=factchecker,
        context=[researchtask],

    )

    editingtask = Task(
        description=(
            "Using the verified research, create a final report with:\n"
            "1. Executive Summary (300 words)\n"
            "2. Introduction and Background\n"
            "3. Key Findings (organized thematically)\n"
            "4. Analysis and Implications\n"
            "5. Future Outlook\n"
            "6. References\n\n"
            "Ensure all disputed or unverified claims are either "
            "removed or clearly marked as such."
        ),
        expectedoutput="Publication-ready research report in markdown.",

        agent=editor,
        context=[researchtask, factchecktask],
        outputfile=f"research{topic.replace(' ', '').lower()}.md",

    )

    crew = Crew(
        agents=[leadresearcher, factchecker, editor],

        tasks=[researchtask, factchecktask, editingtask],

        process=Process.sequential,
        memory=True,
        verbose=True,
    )

    return crew


Usage
researchcrew = createresearchteam("AI Agents in Enterprise Automation 2026")
result = researchcrew.kickoff()

print(f"Report generated: {len(str(result))} characters")

from crewai import Agent, Task, Crew, Process
from crewaitools import CodeInterpreterTool, FileReadTool, DirectoryReadTool


def createcodingteam(projectdescription: str):

    """Create a coding team for software development tasks."""

    codetool = CodeInterpreterTool()
    filetool = FileReadTool()

    dirtool = DirectoryReadTool()

    # Architect
    architect = Agent(
        role="Software Architect",
        goal="Design scalable, maintainable software architectures",
        backstory=(
            "You are a principal software architect with 20 years of "
            "experience. You design systems that are elegant, performant, "
            "and easy to maintain. You favor SOLID principles, clean "
            "architecture, and pragmatic design decisions."
        ),
        verbose=True,
        allowdelegation=True,

    )

    # Developer
    developer = Agent(
        role="Senior Software Developer",
        goal="Write clean, efficient, and well-tested code",
        backstory=(
            "You are a senior developer who writes production-quality "
            "code. You follow TDD, write comprehensive docstrings, and "
            "handle edge cases. You prefer simple, readable code over "
            "clever abstractions."
        ),
        tools=[codetool, filetool],

        verbose=True,
    )

    # Code reviewer
    reviewer = Agent(
        role="Code Reviewer",
        goal="Ensure code quality, security, and adherence to best practices",
        backstory=(
            "You are a tech lead known for thorough code reviews. You "
            "catch bugs, security vulnerabilities, and performance issues. "
            "Your reviews are constructive and educational."
        ),
        tools=[codetool],
        verbose=True,
    )

    # Tasks
    designtask = Task(

        description=(
            f"Design the architecture for: {projectdescription}\n\n"
            "Provide:\n"
            "1. High-level architecture diagram (text-based)\n"
            "2. Module/package structure\n"
            "3. Key interfaces and data models\n"
            "4. Technology choices with justification\n"
            "5. Error handling strategy\n"
            "6. Testing strategy"
        ),
        expectedoutput="Architecture design document with all specified sections.",

        agent=architect,
    )

    implementationtask = Task(
        description=(
            "Implement the system based on the architecture design.\n\n"
            "Requirements:\n"
            "- Follow the architecture exactly\n"
            "- Write complete, runnable Python code\n"
            "- Include type hints on all functions\n"
            "- Write docstrings for all public APIs\n"
            "- Include unit tests for core functionality\n"
            "- Handle errors gracefully"
        ),
        expectedoutput="Complete implementation with source code and tests.",

        agent=developer,
        context=[designtask],
    )

    reviewtask = Task(

        description=(
            "Review the implementation for:\n"
            "1. Adherence to the architecture design\n"
            "2. Code quality (readability, DRY, SOLID)\n"
            "3. Security vulnerabilities\n"
            "4. Performance issues\n"
            "5. Test coverage and quality\n"
            "6. Error handling completeness\n\n"
            "Provide specific code suggestions for each issue found."
        ),
        expectedoutput=(
            "Code review report with issues, severity levels, "
            "and specific improvement suggestions."
        ),
        agent=reviewer,
        context=[designtask, implementationtask],
    )

    crew = Crew(
        agents=[architect, developer, reviewer],
        tasks=[designtask, implementationtask, reviewtask],

        process=Process.sequential,
        memory=True,
        verbose=True,
    )

    return crew


Usage
codingcrew = createcodingteam(
    "A REST API for a task management system with user authentication, "
    "task CRUD operations, team collaboration, and real-time notifications "
    "using FastAPI and SQLAlchemy"
)

result = codingcrew.kickoff()

print(f"Development output: {len(str(result))} characters")

# BAD: Vague role
agent = Agent(
    role="Helper",
    goal="Help with things",
    backstory="You help people.",
)

GOOD: Specific, well-defined role
agent = Agent(
    role="Financial Data Analyst",
    goal="Analyze financial datasets to identify trends, anomalies, "
         "and actionable investment insights",
    backstory=(
        "You have a CFA certification and 10 years of experience "
        "at top investment firms. You specialize in quantitative "
        "analysis and have a track record of identifying profitable "
        "market opportunities through data-driven approaches."
    ),
)

# BAD: Vague task
task = Task(
    description="Analyze the data.",
    expectedoutput="Analysis results.",
    agent=analyst,
)

GOOD: Detailed, structured task
task = Task(
    description=(
        "Analyze the Q4 2025 sales data to determine:\n"
        "1. Top 5 products by revenue\n"
        "2. Month-over-month growth rates\n"
        "3. Regional performance comparison\n"
        "4. Customer segment analysis\n\n"
        "Use statistical methods to identify significant trends. "
        "Flag any anomalies that warrant investigation."
    ),
    expectedoutput=(

        "A structured analysis report with:\n"
        "- Data summary table\n"
        "- Key findings (numbered list)\n"
        "- Visualizations described in text\n"
        "- Recommendations for Q1 2026"