Chapter 21 · Observability and Monitoring Integration

Chapter goals: Understand every field in usage_metrics, integrate with Langtrace and Langfuse two third-party observability platforms, and export data to a custom monitoring system using event listeners.

21.1 Understand usage_metrics First: Cost Calculation Without External Platforms

Every kickoff() return carries token_usage (i.e., usage metrics) — this is the zero-dependency cost-accounting starting point:

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

llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
)

researcher = Agent(role="Researcher", goal="Research topic", backstory="Industry analyst", llm=llm)
task = Task(
    description="Research market landscape of {topic}",
    expected_output="300-word Chinese briefing",
    agent=researcher,
)
crew = Crew(agents=[researcher], tasks=[task])

result = crew.kickoff(inputs={"topic": "Domestic Large Models"})

usage = result.token_usage
print(f"Successful requests:   {usage.successful_requests}")
print(f"Prompt tokens:         {usage.prompt_tokens}")
print(f"Completion tokens:     {usage.completion_tokens}")
print(f"Total tokens:          {usage.total_tokens}")
# Multiply by DeepSeek pricing to derive per-run cost;
# average over multiple runs to estimate cost per thousand calls

Meanings and usage of the four fields:

• successful_requests: number of LLM calls made — often far exceeds task count (one Agent execution = at least one call + another round after each tool call); this is the first indicator for spotting "Agent looping";
• prompt_tokens / completion_tokens: input and output billed separately; output-side unit price is usually several times higher — prioritize cutting output length;
• total_tokens: grand total; multiply by unit price to get cost directly.

Task-level metrics

Individual TaskOutput objects also carry usage information; paired with the task_callback from Chapter 14, you can attribute cost to each环节.

21.2 Integrating Langtrace: Full-Stack Tracing with One-Line Init

Langtrace is an open-source LLM observability platform. The official docs describe a very lightweight integration. Key ordering detail: init must happen before importing CrewAI:

pip install langtrace-python-sdk

from langtrace_python_sdk import langtrace

# Must be placed before all crewai imports to intercept the full call chain
langtrace.init(api_key=os.environ["LANGTRACE_API_KEY"])

# Then write CrewAI code as usual
from crewai import Agent, Task, Crew  # noqa: E402

After setting the project type to CrewAI in the Langtrace console, you get three views: Token & Cost Tracking (cost per Agent/tool interaction), Execution Step Trace Graph (who called which tool, how long it took), Performance Regression Comparison (cross-version run horizontal comparison).

21.3 Integrating Langfuse: OpenTelemetry Standard Path

Langfuse is another mainstream open-source platform. The official recommendation is to integrate via OpenTelemetry (through the OpenLit SDK) — the benefit is your instrumentation is not locked to any vendor:

pip install langfuse openlit crewai crewai_tools

import os

# Langfuse project credentials (obtained from the console settings page)
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"  # or US region

# Initialize OpenLit's OpenTelemetry instrumentation; traces auto-report to Langfuse
import openlit
openlit.init()

from langfuse import get_client
langfuse = get_client()
assert langfuse.auth_check(), "Langfuse credential verification failed"

# Then create and run Crews normally; traces appear in the Langfuse console

Selection guide: if you just want to quickly see "how much it cost, which step is slow", Langtrace's one-line init is the easiest; if your team already has OTel infrastructure or needs self-hosted Langfuse, the OpenLit/OpenTelemetry path is more portable. Both can coexist for canary switching.

21.4 Event Listeners: Exporting to Custom Monitoring

When you do not want to depend on third-party platforms, you can use CrewAI's event bus to build a custom exporter. Core three-piece set: CrewAIEventsBus (singleton event bus), event classes (e.g., CrewKickoffStartedEvent), BaseEventListener (listener基类):

import time
from collections import defaultdict

from crewai.events import (
    BaseEventListener,
    CrewKickoffStartedEvent,
    CrewKickoffCompletedEvent,
    AgentExecutionCompletedEvent,
)


class MetricsListener(BaseEventListener):
    """Push crew run metrics to any monitoring system (example: print + in-memory aggregation)"""

    def __init__(self):
        super().__init__()
        self._start_ts = {}
        self.stats = defaultdict(float)

    def setup_listeners(self, crewai_event_bus):
        @crewai_event_bus.on(CrewKickoffStartedEvent)
        def on_started(source, event):
            self._start_ts[event.crew_name] = time.time()
            print(f"[metrics] crew '{event.crew_name}' started")

        @crewai_event_bus.on(AgentExecutionCompletedEvent)
        def on_agent_done(source, event):
            print(f"[metrics] agent '{event.agent.role}' completed")
            self.stats["agent_completions"] += 1
            # In production, push to Prometheus pushgateway / StatsD / Kafka here ...

        @crewai_event_bus.on(CrewKickoffCompletedEvent)
        def on_completed(source, event):
            cost = time.time() - self._start_ts.get(event.crew_name, time.time())
            self.stats["crew_wall_seconds"] += cost
            print(f"[metrics] crew completed in {cost:.1f}s")
            # Can write result.token_usage to a time-series DB here


# Critical: must create an instance and ensure it is imported (otherwise GC'd and listener失效)
metrics_listener = MetricsListener()

Registration: import the module in your crew/flow definition file and instantiate the listener (or包 it into a listeners/ package for unified引入). Events are broadcast — the same bus can挂 both a Langtrace exporter and your Prometheus exporter simultaneously, no conflict.

21.5 What to Monitor on a Production Dashboard

Metric	Source	Anomaly Signal
Per-kickoff cost	token_usage.total_tokens × unit price	Week-over-week spike → prompt inflation or infinite loop
successful_requests	usage metrics	Far exceeds task count → Agent looping / delegation storm
P95 end-to-end latency	event time delta / platform trace	Tail degradation → tool timeout retries piling up
Task success rate	callback/event stats	Rising failure rate → upstream API changes
Tool error distribution	trace tool spans	Concentrated on one tool → that dependency is unstable

Practice advice: start with 21.1's raw metrics + logs to get "visibility"; after data is flowing, add Langtrace/Langfuse for trace details; finally use 21.4's event exports to沉淀 the core four or five metrics into your company's unified monitoring/alerting system.

Chapter Summary

• result.token_usage provides four fields: successful_requests/prompt_tokens/completion_tokens/total_tokens — zero-dependency cost starting point;
• Langtrace: pip install langtrace-python-sdk + langtrace.init() before importing CrewAI;
• Langfuse: via OpenTelemetry, one-line openlit.init() auto-reports traces; self-hostable, vendor-agnostic;
• BaseEventListener + crewai_event_bus.on(EventClass) can export any event to custom monitoring; note the instance must stay alive;
• Dashboard four pillars: cost, call count, P95 latency, success rate; plus tool error distribution to定位 dependency issues.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. A Crew has 3 tasks but token_usage shows successful_requests=17. Most likely cause?

2. When integrating Langtrace per official docs, what is the most common pitfall?

3. What is the officially recommended "CrewAI + Langfuse" integration path?

4. After defining a custom BaseEventListener, the listener receives no events. Most likely cause?

🛠️ Hands-On Exercises

1. Write a CostReporter callback for the marketing Crew from Chapter 23: print total_tokens and RMB cost (converted by DeepSeek unit price) after each kickoff; aggregate averages over 10 runs.
2. Register a local Langtrace account (or self-deploy OpenLit + Langfuse); add tracing to an existing Crew; screenshot to find "which Agent spent the most tokens."
3. Extend the 21.4 MetricsListener to capture output length from AgentExecutionCompletedEvent; aggregate "average output chars per Agent" and print alongside kickoff completion.

Observability is ready; the last mile is getting the service running: Chapter 22 · Deployment and Service.