Trace and evaluate a CrewAI research crew

CrewAIInstrumentor wraps CrewAI’s internal execution methods so every agent run, tool call, and task produces a span automatically. Because this project is registered as ProjectType.OBSERVE, you configure evals as a platform Eval Task attached to those spans later, rather than calling an evaluator from code.

from crewai import LLM, Agent, Crew, Process, Task
from crewai_tools import SerperDevTool, FileReadTool
from fi_instrumentation import register, FITracer
from fi_instrumentation.fi_types import ProjectType
from traceai_crewai import CrewAIInstrumentor

trace_provider = register(
    project_type=ProjectType.OBSERVE,
    project_name="crewai-research-team",
    set_global_tracer_provider=True,
)

# Auto-instruments crewai.Task._execute_core and crewai.Crew.kickoff.
# No manual spans needed for the crew's own execution.
CrewAIInstrumentor().instrument(tracer_provider=trace_provider)

tracer = FITracer(trace_provider.get_tracer(__name__))

You should see no output here beyond a clean exit. The instrumentor is now active for every CrewAI call in the process.

Four specialized agents, each with a narrow role. allow_delegation=False keeps the sequential process predictable: each agent runs once, in order.

llm = LLM(model="gpt-4o", temperature=0.7, max_tokens=2000)

market_researcher = Agent(
    role="Senior Market Research Analyst",
    goal="Research and analyze emerging technology trends and market dynamics",
    backstory=(
        "You are a market research analyst with 15 years in technology markets. "
        "You favor data-backed claims over speculation."
    ),
    llm=llm,
    tools=[SerperDevTool()],
    allow_delegation=False,
)

competitive_analyst = Agent(
    role="Competitive Intelligence Specialist",
    goal="Analyze competitive landscapes and identify market opportunities",
    backstory=(
        "You analyze competitor strategies and market positioning to find gaps "
        "a new entrant could exploit."
    ),
    llm=llm,
    tools=[SerperDevTool()],
    allow_delegation=False,
)

report_writer = Agent(
    role="Technical Report Writer",
    goal="Create a comprehensive, well-structured research report",
    backstory="You turn raw research into executive summaries and recommendations.",
    llm=llm,
    tools=[FileReadTool()],
    allow_delegation=False,
)

quality_analyst = Agent(
    role="Research Quality Assurance Specialist",
    goal="Verify accuracy and completeness of the research findings",
    backstory="You fact-check claims and flag logical gaps before a report ships.",
    llm=llm,
    allow_delegation=False,
)

You should see the four Agent objects construct without error. Nothing runs yet.

Each task names its agent and its expected output. report_generation_task and quality_assurance_task don’t yet declare context=, so CrewAI has no explicit link telling them which upstream task output to build on. The next step shows why that matters.

def create_research_tasks(research_topic: str) -> list[Task]:
    market_research_task = Task(
        description=(
            f"Research the market for: {research_topic}. Cover market size, "
            "growth drivers, major players, and regulatory landscape. Cite sources."
        ),
        agent=market_researcher,
        expected_output="A market research summary with cited data points",
    )

    competitive_analysis_task = Task(
        description=(
            f"Analyze the competitive landscape for: {research_topic}. "
            "Cover top competitors, positioning, and market gaps."
        ),
        agent=competitive_analyst,
        expected_output="A competitive analysis with named competitors",
    )

    report_generation_task = Task(
        description=(
            f"Write a research report on: {research_topic} with an executive "
            "summary, market overview, competitive landscape, and recommendations."
        ),
        agent=report_writer,
        expected_output="A structured research report",
    )

    quality_assurance_task = Task(
        description=(
            "Review the report for factual accuracy, logical consistency, and "
            "completeness. List any issues found."
        ),
        agent=quality_analyst,
        expected_output="A quality review with a pass/fail verdict",
    )

    return [
        market_research_task,
        competitive_analysis_task,
        report_generation_task,
        quality_assurance_task,
    ]

You should see a list of four Task objects. Process.sequential in the next step runs them in this order.

def run_research_crew(research_topic: str) -> str:
    tasks = create_research_tasks(research_topic)

    research_crew = Crew(
        agents=[market_researcher, competitive_analyst, report_writer, quality_analyst],
        tasks=tasks,
        process=Process.sequential,
        memory=True,
    )

    result = research_crew.kickoff()
    return str(result)


if __name__ == "__main__":
    topic = "Generative AI in Healthcare: Market Opportunities and Challenges"
    report = run_research_crew(topic)
    print(report[:500])

Output shape (illustrative):

## Executive Summary
Generative AI is reshaping healthcare diagnostics and...

You should see the crew’s four agents run in sequence in your terminal (CrewAI’s own verbose logging), followed by the printed report excerpt. Once the Eval Task from the next step is attached, open the report_generation_task span in Observe: without an explicit link to the upstream research, report_writer only sees its own task description as input, so groundedness scores low. There’s nothing in that span’s captured context for the eval to check the report against.

Fix it by wiring the dependency explicitly, so CrewAI feeds the upstream outputs into the report task and quality-assurance task instead of leaving them to infer it:

    report_generation_task = Task(
        description=(
            f"Write a research report on: {research_topic} with an executive "
            "summary, market overview, competitive landscape, and recommendations."
        ),
        agent=report_writer,
        expected_output="A structured research report",
        context=[market_research_task, competitive_analysis_task],
    )

    quality_assurance_task = Task(
        description=(
            "Review the report for factual accuracy, logical consistency, and "
            "completeness. List any issues found."
        ),
        agent=quality_analyst,
        expected_output="A quality review with a pass/fail verdict",
        context=[market_research_task, competitive_analysis_task, report_generation_task],
    )

Rerun the same topic. The report_generation_task span now carries the market and competitive findings as input, and the groundedness score on that span moves up from the first run: the report has actual source material to be grounded against instead of just the topic string.

In the dashboard, define evals as an Eval Task and attach them to the crew’s span types rather than calling an evaluator from code. This scores every run as it’s generated, including the two you just made.

Open app.futureagi.com → Observe → the crewai-research-team project → Eval Task, and attach Completeness, Groundedness, and Context Relevance to the task span type. Each of the three resolves on the built-in evals reference.

You should see the new Eval Task listed against the span type you selected, and it starts scoring the next trace that hits that span.

Open the crewai-research-team project. Each run appears as a trace with one span per agent task.

The trace view shows the four agent tasks in execution order, with tool calls nested under the market researcher and competitive analyst spans

Open the report_generation_task span to see the three eval scores attached to it directly by the Eval Task.

Each eval score sits on the span it was computed from, so you can correlate a low score with the exact agent output that produced it