pip install crewai crewai-tools

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
import os

search_tool = SerperDevTool(api_key=os.environ["SERPER_API_KEY"])

researcher = Agent(
    role="AI Researcher",
    goal="Find accurate, up-to-date information on the given topic.",
    backstory="You are an experienced researcher skilled at finding credible sources.",
    tools=[search_tool],
    verbose=True,
)

writer = Agent(
    role="Technical Writer",
    goal="Transform research findings into a clear, engaging summary.",
    backstory="You write concise technical content for developer audiences.",
    verbose=True,
)

research_task = Task(
    description="Research the key capabilities of LlamaIndex as of 2026.",
    expected_output="A structured list of LlamaIndex features with brief descriptions.",
    agent=researcher,
)

write_task = Task(
    description="Write a 200-word blog introduction based on the research findings.",
    expected_output="A polished 200-word blog introduction paragraph.",
    agent=writer,
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result.raw)

from crewai import Agent, Task, Crew, Process
import os

# Use Anthropic Claude
researcher = Agent(
    role="Research Analyst",
    goal="Gather comprehensive technical information on the assigned topic.",
    backstory=(
        "You are a meticulous research analyst who synthesises information "
        "from multiple sources into structured, factual reports."
    ),
    llm="claude-sonnet-4-6",
    verbose=True,
    max_iter=5,
)

writer = Agent(
    role="Technical Writer",
    goal="Produce clear, developer-friendly documentation from research input.",
    backstory=(
        "You are a senior technical writer who turns dense research into "
        "readable guides with examples and callouts."
    ),
    llm="claude-sonnet-4-6",
    verbose=True,
)

qa = Agent(
    role="Quality Reviewer",
    goal="Ensure the written content is accurate, complete, and follows best practices.",
    backstory=(
        "You are a QA engineer who reviews technical documentation for "
        "accuracy, missing edge cases, and clarity."
    ),
    llm="claude-sonnet-4-6",
    verbose=True,
)

task1 = Task(
    description=(
        "Research polars, the Python DataFrame library. Cover: "
        "architecture, performance characteristics, and the lazy API."
    ),
    expected_output=(
        "A structured report with sections: Overview, Performance, LazyFrame API, "
        "and 3 code examples."
    ),
    agent=researcher,
)

task2 = Task(
    description="Write a developer guide section on polars based on the research report.",
    expected_output=(
        "A 400-word guide section with an introduction, 2 code blocks with output, "
        "and a common pitfalls callout."
    ),
    agent=writer,
    context=[task1],   # receives task1's output as additional context
)

task3 = Task(
    description=(
        "Review the written guide for technical accuracy and completeness. "
        "Identify any missing information or errors."
    ),
    expected_output=(
        "A review report listing: Approved items (bullet list), "
        "Issues found (numbered list), and Suggested changes."
    ),
    agent=qa,
    context=[task2],
)

crew = Crew(
    agents=[researcher, writer, qa],
    tasks=[task1, task2, task3],
    process=Process.sequential,
    verbose=True,
    memory=True,   # shared long-term memory across the crew
)

result = crew.kickoff()
print(result.raw)

from crewai import Agent
from crewai_tools import FileReadTool, CodeInterpreterTool

agent = Agent(
    role="Data Scientist",
    goal="Analyse datasets and produce insights with Python code.",
    backstory="You are an expert data scientist who writes clean, well-commented analysis code.",

    llm="gpt-4o",              # or "claude-sonnet-4-6", "gemini/gemini-1.5-flash"
    tools=[FileReadTool(), CodeInterpreterTool()],

    verbose=True,              # print agent reasoning to stdout
    memory=True,               # give agent access to crew memory
    max_iter=10,               # max self-correction attempts per task
    max_rpm=20,                # max LLM requests per minute (rate limit guard)
    allow_delegation=False,    # prevent agent from delegating to others
)

from crewai import Task

task = Task(
    description=(
        "Analyse the sales CSV at data/sales_2026.csv and produce "
        "a monthly revenue summary with the top 3 products by revenue."
    ),
    expected_output=(
        "A markdown table with columns: Month, Total Revenue, Top Product. "
        "Include a brief 2-sentence insight below the table."
    ),
    agent=data_scientist,

    context=[previous_task],   # list of Task objects whose output feeds this task
    output_file="results/summary.md",  # write output to file
    human_input=False,         # set True to pause and ask for human approval
)

from crewai import Crew, Process

# Sequential — tasks run in order; each receives previous task outputs
crew = Crew(
    agents=[researcher, writer, qa],
    tasks=[task1, task2, task3],
    process=Process.sequential,
    verbose=True,
    memory=True,
    max_rpm=30,
    output_log_file="crew_log.txt",
)

# Hierarchical — a manager agent routes tasks dynamically
crew = Crew(
    agents=[researcher, writer, qa],
    tasks=[task1, task2, task3],
    process=Process.hierarchical,
    manager_llm="gpt-4o",     # manager agent uses this LLM
    verbose=True,
)

# Synchronous
result = crew.kickoff()
print(result.raw)          # final output string
print(result.tasks_output) # list of TaskOutput objects
print(result.token_usage)  # total token counts

# With inputs (interpolated into task descriptions via {variable})
result = crew.kickoff(inputs={"topic": "LangChain", "year": "2026"})

# Async (for FastAPI, asyncio)
import asyncio
result = asyncio.run(crew.kickoff_async())

# Batch — run the same crew for a list of inputs
inputs_list = [{"topic": "LangChain"}, {"topic": "LlamaIndex"}, {"topic": "AutoGen"}]
results = crew.kickoff_for_each(inputs=inputs_list)

from crewai_tools import (
    SerperDevTool,      # Google search via Serper API
    WebsiteSearchTool,  # RAG over a specific URL
    FileReadTool,       # read local files
    DirectoryReadTool,  # read all files in a directory
    CodeInterpreterTool, # execute Python code
    CSVSearchTool,      # semantic search over a CSV
    PDFSearchTool,      # semantic search over a PDF
)

# Custom tool with @tool decorator
from crewai.tools import tool

@tool("Database Query Tool")
def query_database(sql: str) -> str:
    """Execute a read-only SQL query and return results as a string."""
    import sqlite3
    conn = sqlite3.connect("app.db")
    cursor = conn.execute(sql)
    rows = cursor.fetchmany(20)
    return "\n".join(str(row) for row in rows)

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer],
    tasks=[task1, task2],
    process=Process.sequential,
    memory=True,               # enables all memory types
    embedder={
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"},
    },
)

# config/agents.yaml
researcher:
  role: Research Analyst
  goal: Find accurate technical information on {topic}.
  backstory: You are a meticulous researcher skilled at synthesising complex sources.

writer:
  role: Technical Writer
  goal: Write clear developer documentation based on research.
  backstory: You write concise technical guides for developer audiences.

# config/tasks.yaml
research_task:
  description: Research {topic} and produce a structured technical report.
  expected_output: A structured report with sections and 3 code examples.
  agent: researcher

write_task:
  description: Write a developer guide on {topic} based on the research.
  expected_output: A 400-word guide with code blocks and a pitfalls callout.
  agent: writer
  context: [research_task]

from crewai import Agent, Task, Crew
from crewai.project import CrewBase, agent, task, crew
import yaml

@CrewBase
class ResearchCrew:
    agents_config = "config/agents.yaml"
    tasks_config  = "config/tasks.yaml"

    @agent
    def researcher(self) -> Agent:
        return Agent(config=self.agents_config["researcher"], llm="claude-sonnet-4-6")

    @agent
    def writer(self) -> Agent:
        return Agent(config=self.agents_config["writer"], llm="claude-sonnet-4-6")

    @task
    def research_task(self) -> Task:
        return Task(config=self.tasks_config["research_task"])

    @task
    def write_task(self) -> Task:
        return Task(config=self.tasks_config["write_task"])

    @crew
    def crew(self) -> Crew:
        return Crew(agents=self.agents, tasks=self.tasks, process=Process.sequential, verbose=True)

result = ResearchCrew().crew().kickoff(inputs={"topic": "ChromaDB"})
print(result.raw)

result = crew.kickoff()

# Final output
print(result.raw)

# Per-task output
for task_output in result.tasks_output:
    print(f"Task: {task_output.name}")
    print(f"Agent: {task_output.agent}")
    print(f"Output: {task_output.raw[:200]}")

# Token usage
usage = result.token_usage
print(f"Total tokens: {usage.total_tokens}")
print(f"Prompt: {usage.prompt_tokens}, Completion: {usage.completion_tokens}")