CrewAI in Practice

from crewai.tools import tool

@tool("web_search")
def web_search(query: str) -> str:
    """Search the web for information on a query.
    Returns a list of results with title, url, and snippet.
    Use this to find relevant sources for a research question."""
    result = _web_search_impl(query)  # underlying implementation from Lesson 3
    # CrewAI tools must return a string - serialize the result
    import json
    return json.dumps(result, indent=2)

@tool("fetch_document")
def fetch_document(url: str) -> str:
    """Retrieve the full text content of a document by URL.
    Use this after web_search to read a specific page in detail."""
    result = _fetch_document_impl(url)  # underlying implementation from Lesson 3
    return json.dumps(result, indent=2)

from crewai import Agent
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-opus-4-8", temperature=0)

researcher = Agent(
    role="Senior Research Specialist",
    goal=(
        "Find accurate, well-sourced information on the research question "
        "using web search and document retrieval. Gather at least three "
        "credible sources with specific, quotable claims."
    ),
    backstory=(
        "You are an experienced researcher with a systematic approach: "
        "search broadly first, then read the most relevant sources deeply. "
        "You always note when information is incomplete or contradictory."
    ),
    tools=[web_search, fetch_document],
    llm=llm,
    verbose=True,  # logs agent reasoning steps
    max_iter=10,   # circuit breaker: max reasoning iterations
)

report_writer = Agent(
    role="Research Report Writer",
    goal=(
        "Synthesize research findings into a structured, readable report "
        "that directly answers the question with sourced claims and "
        "explicitly flags any information gaps."
    ),
    backstory=(
        "You write clear, precise research reports for technical audiences. "
        "Every claim in your reports traces to a specific source. "
        "You are honest about what the research did and did not find."
    ),
    llm=llm,
    verbose=True,
)

from crewai import Task

def create_research_task(question: str) -> Task:
    return Task(
        description=f"""Research the following question thoroughly:

{question}

Use web_search to find relevant sources, then fetch_document to read
the most relevant pages in full. Collect at least three specific,
quotable claims with their source URLs.""",
        expected_output=(
            "A structured list of findings: each finding is a claim, "
            "its source URL, and a brief note on reliability. "
            "Flag any areas where sources conflict or information is missing."
        ),
        agent=researcher,
    )

def create_report_task(question: str, research_task: Task) -> Task:
    return Task(
        description=f"""Write a research report that answers:

{question}

Use the research findings from the previous task. Structure the report with:
1. Direct answer to the question
2. Supporting evidence (at least 3 sourced claims with URLs)
3. Information gaps or caveats""",
        expected_output=(
            "A structured research report in markdown format with: "
            "an executive summary, sourced claims section, and gaps section."
        ),
        agent=report_writer,
        context=[research_task],  # report task has access to research task output
    )

from crewai import Crew, Process

def run_research_crew(question: str) -> str:
    research_task = create_research_task(question)
    report_task = create_report_task(question, research_task)

    crew = Crew(
        agents=[researcher, report_writer],
        tasks=[research_task, report_task],
        process=Process.sequential,  # tasks run in order, second sees first's output
        verbose=True,
        # memory=True,  # enable for cross-run memory (requires embedding model)
    )

    result = crew.kickoff(inputs={"question": question})
    return result.raw  # the final task's output as a string

# Run it
report = run_research_crew(
    "What are the main approaches to AI agent memory?"
)
print(report)

from crewai import Crew, Process, Agent

manager = Agent(
    role="Research Manager",
    goal="Coordinate the research team and synthesize their findings",
    backstory="An experienced manager who delegates effectively and synthesizes diverse inputs",
    llm=llm,
    allow_delegation=True,  # can assign tasks to other agents
)

parallel_crew = Crew(
    agents=[researcher, report_writer],
    tasks=[research_task, report_task],
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True,
)