CrewAI

Overview

Why monitor CrewAI?

Multi-agent pipelines are harder to debug than single chains. An agent might time out waiting for a tool. The LLM might return output that fails the next agent's input validation. The crew might finish but produce empty results — technically successful, actually broken.

Common scenarios where NotiLens catches problems your logs miss: a Researcher agent calls a web search tool that rate-limits mid-run → the Writer agent gets no context → produces generic output. Or a crew that usually takes 45 seconds suddenly takes 8 minutes — something is retrying silently.

Auto-Patch

Installation

pip install notilens crewai crewai-tools

Setup

Add patch=True — NotiLens hooks into CrewAI's execution callbacks and tracks every agent's LLM calls automatically.

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

nl = notilens.init(
    name="content-crew",
    token="YOUR_TOKEN",
    secret="YOUR_SECRET",
    patch=True   # hooks into CrewAI callbacks
)

researcher = Agent(
    role="Senior Researcher",
    goal="Find accurate, up-to-date information on the given topic",
    backstory="You are an expert at finding and synthesising information.",
    tools=[SerperDevTool()],
    verbose=True,
)

writer = Agent(
    role="Content Writer",
    goal="Write engaging, accurate content based on research findings",
    backstory="You write compelling content that drives engagement.",
    verbose=True,
)

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

# NotiLens tracks every LLM call: which agent, tokens, cost, latency
result = crew.kickoff(inputs={"topic": "AI trends in 2026"})

What gets tracked automatically

Token and cost metrics are per-agent — so you can see which agent in your crew is consuming the most tokens and driving your LLM bill.

Real-world example — content creation crew

A 3-agent crew: Researcher finds sources, Writer drafts the article, Editor polishes it. With patch=True, NotiLens tracks every LLM call per agent, so you can see if the Researcher is consuming 70% of your tokens (context too large) or if the Editor is taking 3x longer than usual (prompt issue).

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

nl = notilens.init(
    name="article-crew",
    token="YOUR_TOKEN",
    secret="YOUR_SECRET",
    patch=True
)

# 3-agent content pipeline
researcher = Agent(
    role="Researcher",
    goal="Find 5 recent, credible sources on the topic",
    backstory="Expert at web research and source validation.",
    tools=[SerperDevTool()],
)
writer = Agent(
    role="Writer",
    goal="Write a 600-word article from the research",
    backstory="Experienced tech writer — clear, engaging, accurate.",
)
editor = Agent(
    role="Editor",
    goal="Polish tone, fix grammar, ensure factual accuracy",
    backstory="Senior editor with attention to detail.",
)

research_task = Task(
    description="Research: {topic}. Find 5 sources with key data points.",
    agent=researcher, expected_output="5 bullet-point summaries with sources",
)
write_task = Task(
    description="Write a 600-word article using the research findings.",
    agent=writer, expected_output="Full article draft",
)
edit_task = Task(
    description="Review and polish the article draft. Fix issues.",
    agent=editor, expected_output="Final polished article",
)

crew   = Crew(agents=[researcher, writer, editor],
              tasks=[research_task, write_task, edit_task],
              process=Process.sequential)
result = crew.kickoff(inputs={"topic": "The future of edge AI in manufacturing"})

Manual

Manual Instrumentation

Wrap the crew execution in a NotiLens task run when you need lifecycle events, per-crew metrics, or a human review gate after the crew finishes.

Real-world use: a competitor analysis crew that finishes and needs a product manager to review findings before the report is sent to leadership.

import notilens
from crewai import Agent, Task, Crew

nl  = notilens.init(name="competitor-crew", token="YOUR_TOKEN", secret="YOUR_SECRET")
run = nl.task("competitor-analysis")
run.start()

try:
    analyst = Agent(
        role="Competitive Analyst",
        goal="Analyse 3 competitors: pricing, features, positioning",
        backstory="Expert in SaaS competitive intelligence.",
        tools=[SerperDevTool()],
    )
    reporter = Agent(
        role="Report Writer",
        goal="Summarise findings into a 1-page executive brief",
        backstory="Clear, concise business writer.",
    )

    analysis_task = Task(
        description="Analyse competitors: {competitors}. Focus on pricing and features.",
        agent=analyst, expected_output="Structured analysis per competitor",
    )
    report_task = Task(
        description="Write an executive brief from the analysis.",
        agent=reporter, expected_output="One-page executive summary",
    )

    run.progress("Crew running competitor analysis...")
    crew   = Crew(agents=[analyst, reporter], tasks=[analysis_task, report_task])
    result = crew.kickoff(inputs={"competitors": "Datadog, Grafana, New Relic"})

    run.metric("pages_analysed", 3)
    run.output_generated("Competitor brief ready — 3 companies analysed")

    # Flag for PM review before sending to leadership
    run.input_required("PM review needed before sending to leadership")

    if wait_for_approval() == "approved":
        run.input_approved("Approved — sending to leadership")
        send_report(result.raw)
        run.complete("Report distributed")
    else:
        run.input_rejected("More detail needed on pricing section")

except Exception as e:
    run.fail(str(e))

Multi-Agent Tracking

Tracking individual agent progress

For crews where you want per-agent visibility, use CrewAI's step_callback to fire NotiLens events as each agent completes its task:

Useful when you have long-running crews with 5+ agents and need to know which step is stalling in production.

import notilens
from crewai import Agent, Task, Crew

nl  = notilens.init(name="research-crew", token="YOUR_TOKEN", secret="YOUR_SECRET",
                    patch=True)
run = nl.task("deep-research")
run.start()

completed_agents = []

def on_step_complete(step_output):
    agent_role = step_output.agent if hasattr(step_output, 'agent') else 'agent'
    completed_agents.append(agent_role)
    run.progress(f"{agent_role} completed — {len(completed_agents)} of 4 agents done")

crew = Crew(
    agents=[researcher, analyst, writer, editor],
    tasks=[...],
    step_callback=on_step_complete,
)

result = crew.kickoff(inputs={"topic": "..."})
run.output_generated("Research complete — 4 agents, full report ready")
run.complete("Done")