Introduction to Observability in CrewAI
As AI agent systems grow in complexity, understanding what happens inside your CrewAI crews becomes critical. A single crew might involve multiple agents, several LLM calls, tool executions, and inter-agent handoffs — all of which can fail silently or produce unexpected results. Observability and tracing give you visibility into these internal operations, allowing you to debug issues, optimize performance, and monitor production behavior.
CrewAI, being a framework for orchestrating role-playing autonomous AI agents, benefits enormously from observability. Without it, you're essentially flying blind when a crew produces a poor output or hangs indefinitely. With proper tracing, you can see exactly which agent took too long, which tool call failed, or which prompt produced an unexpected response.
Why Observability Matters for Agent Systems
Traditional software observability focuses on logs, metrics, and traces. For AI agent systems, the stakes are higher because the behavior is non-deterministic. The same input can produce different outputs across runs, making debugging significantly harder. Here's why observability is essential:
- Non-deterministic execution: LLMs produce variable outputs, so you need detailed traces to understand what happened in each run.
- Cost tracking: Every LLM call costs money. Observability helps you track token usage and identify expensive operations.
- Latency diagnosis: Agent crews can take seconds or minutes. Tracing reveals where time is spent — is it the LLM, a tool, or agent reasoning?
- Failure attribution: When a crew fails, you need to know which agent or tool caused the failure, not just that "it didn't work."
- Quality monitoring: In production, you need to monitor whether agent outputs meet quality thresholds over time.
Understanding CrewAI's Tracing Architecture
CrewAI provides built-in support for tracing through its telemetry and callback systems. At a high level, tracing in CrewAI works by instrumenting key operations — agent execution, task completion, tool usage, and LLM calls — and emitting structured events that can be consumed by observability platforms.
The framework supports several approaches to observability:
- Built-in verbose logging and print statements
- Callback handlers for custom instrumentation
- Integration with LangSmith for LLM call tracing
- Integration with OpenTelemetry for distributed tracing
- Third-party platforms like Phoenix (Arize), Langfuse, and Weights & Biases
Setting Up Basic Verbose Logging
The simplest form of observability in CrewAI is enabling verbose mode. This gives you console output showing what each agent is doing during execution. While basic, it's invaluable during development.
from crewai import Agent, Task, Crew, Process
# Create agents with verbose enabled
researcher = Agent(
role="Research Analyst",
goal="Find comprehensive information about the topic",
backstory="You are an expert researcher with attention to detail.",
verbose=True,
allow_delegation=False
)
writer = Agent(
role="Content Writer",
goal="Write clear, engaging content based on research",
backstory="You are a skilled writer who simplifies complex topics.",
verbose=True,
allow_delegation=False
)
# Create tasks
research_task = Task(
description="Research the impact of observability in AI systems.",
expected_output="A detailed research report with key findings.",
agent=researcher
)
writing_task = Task(
description="Write a blog post based on the research findings.",
expected_output="A well-structured blog post of 500-800 words.",
agent=writer
)
# Create crew with verbose output
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
process=Process.sequential,
verbose=True
)
result = crew.kickoff()
print("Final Result:", result)
With verbose=True set on both agents and the crew, you'll see detailed console output including which agent is working on which task, the intermediate thoughts, and the final outputs. This is your first line of defense when debugging.
Using CrewAI's Built-in Telemetry
CrewAI includes a telemetry system that collects anonymous usage data by default. You can extend this concept by building custom event listeners. The key is to hook into CrewAI's execution lifecycle.
import logging
import time
from crewai import Agent, Task, Crew
from crewai.utilities import Printer
# Set up structured logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('crew_execution.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger("CrewAI-Observability")
class ObservableCrew:
"""Wrapper class that adds observability to crew execution."""
def __init__(self, crew: Crew):
self.crew = crew
self.execution_log = []
def kickoff(self, inputs=None):
start_time = time.time()
logger.info("Crew execution started")
logger.info(f"Number of agents: {len(self.crew.agents)}")
logger.info(f"Number of tasks: {len(self.crew.tasks)}")
try:
result = self.crew.kickoff(inputs=inputs)
elapsed = time.time() - start_time
logger.info(f"Crew execution completed in {elapsed:.2f}s")
logger.info(f"Result length: {len(str(result))} characters")
self.execution_log.append({
"status": "success",
"duration": elapsed,
"result_length": len(str(result))
})
return result
except Exception as e:
elapsed = time.time() - start_time
logger.error(f"Crew execution failed after {elapsed:.2f}s: {e}")
self.execution_log.append({
"status": "failed",
"duration": elapsed,
"error": str(e)
})
raise
# Usage
crew = Crew(agents=[researcher, writer], tasks=[research_task, writing_task])
observable_crew = ObservableCrew(crew)
result = observable_crew.kickoff()
Integrating with LangSmith for LLM Tracing
LangSmith is one of the most popular tracing platforms for LLM applications. Since CrewAI uses LangChain's LLM abstractions under the hood, you can leverage LangSmith for detailed tracing of every LLM call, including prompts, responses, token counts, and latencies.
import os
from crewai import Agent, Task, Crew
# Set LangSmith environment variables BEFORE importing crewai
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "crewai-observability-demo"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
# Now create your crew
analyst = Agent(
role="Data Analyst",
goal="Analyze the provided data and extract insights",
backstory="You are a meticulous data analyst.",
verbose=True
)
analysis_task = Task(
description="Analyze sales data trends for Q4 2024.",
expected_output="A summary of key trends and anomalies.",
agent=analyst
)
crew = Crew(
agents=[analyst],
tasks=[analysis_task],
verbose=True
)
# Every LLM call will now be traced in LangSmith
result = crew.kickoff()
Once configured, you can visit the LangSmith dashboard to see a complete trace of your crew's execution. Each trace shows the full prompt sent to the LLM, the response received, token counts, latency, and the hierarchical relationship between calls. This is particularly useful for understanding how agents reason and where they might be wasting tokens.
Integrating with OpenTelemetry
For production-grade distributed tracing, OpenTelemetry is the industry standard. It allows you to trace CrewAI execution alongside the rest of your infrastructure — databases, APIs, message queues, and more. Here's how to instrument CrewAI with OpenTelemetry:
import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
# Set up OpenTelemetry
resource = Resource.create({
"service.name": "crewai-service",
"service.version": "1.0.0",
})
provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(
OTLPSpanExporter(endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"))
)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("crewai.tracer")
from crewai import Agent, Task, Crew
class TracedAgent(Agent):
"""Agent subclass that adds OpenTelemetry spans."""
def execute_task(self, task, context=None, tools=None):
with tracer.start_as_current_span(f"agent.{self.role}") as span:
span.set_attribute("agent.role", self.role)
span.set_attribute("agent.goal", self.goal)
span.set_attribute("task.description", task.description[:200])
try:
result = super().execute_task(task, context, tools)
span.set_attribute("execution.status", "success")
span.set_attribute("result.length", len(str(result)))
return result
except Exception as e:
span.set_attribute("execution.status", "error")
span.record_exception(e)
raise
class TracedTask(Task):
"""Task subclass that adds OpenTelemetry spans."""
def execute(self, agent, context=None, tools=None):
with tracer.start_as_current_span(f"task.{self.description[:50]}") as span:
span.set_attribute("task.description", self.description)
span.set_attribute("task.agent", agent.role)
result = super().execute(agent, context, tools)
span.set_attribute("task.result_length", len(str(result)))
return result
# Usage with traced components
researcher = TracedAgent(
role="Researcher",
goal="Gather information",
backstory="Expert researcher"
)
task = TracedTask(
description="Research AI observability tools",
expected_output="List of tools with pros and cons",
agent=researcher
)
with tracer.start_as_current_span("crew.execution") as span:
span.set_attribute("crew.agent_count", 1)
span.set_attribute("crew.task_count", 1)
crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()
span.set_attribute("crew.result_length", len(str(result)))
This approach creates a hierarchical trace where the crew execution is the root span, each task is a child span, and each agent execution is a nested child. You can view these traces in any OpenTelemetry-compatible backend like Jaeger, Zipkin, Grafana Tempo, or Datadog.
Integrating with Arize Phoenix
Arize Phoenix is an open-source observability tool specifically designed for LLM applications. It provides excellent visualization for agent traces and is easy to set up locally.
import os
from crewai import Agent, Task, Crew
# Install phoenix and set up tracing
# pip install arize-phoenix openinference-instrumentation-openai
import phoenix as px
from openinference.instrumentation.openai import OpenAIInstrumentor
# Start Phoenix server (runs on localhost:6006)
px.launch_app()
# Instrument OpenAI calls
OpenAIInstrumentor().instrument()
# Now all CrewAI LLM calls will be traced in Phoenix
researcher = Agent(
role="Market Researcher",
goal="Analyze market trends and competitor strategies",
backstory="You are a seasoned market analyst.",
verbose=True
)
strategist = Agent(
role="Strategy Consultant",
goal="Develop actionable business strategies",
backstory="You are a McKinsey-level strategy consultant.",
verbose=True
)
research_task = Task(
description="Analyze the competitive landscape for AI observability tools.",
expected_output="Competitor analysis with strengths and weaknesses.",
agent=researcher
)
strategy_task = Task(
description="Based on the research, propose a go-to-market strategy.",
expected_output="Strategic recommendations with prioritized actions.",
agent=strategist
)
crew = Crew(
agents=[researcher, strategist],
tasks=[research_task, strategy_task],
verbose=True
)
result = crew.kickoff()
# View traces at http://localhost:6006
print("Check Phoenix UI at http://localhost:6006 for detailed traces")
Integrating with Langfuse
Langfuse is another popular open-source LLM observability platform that offers self-hosting and cloud options. It provides rich tracing, evaluation, and prompt management features.
import os
from crewai import Agent, Task, Crew, LLM
# Set Langfuse credentials
os.environ["LANGFUSE_PUBLIC_KEY"] = "your-public-key"
os.environ["LANGFUSE_SECRET_KEY"] = "your-secret-key"
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"
from langfuse.openai import openai
from langfuse.callback import CallbackHandler
# Create a Langfuse callback handler
langfuse_handler = CallbackHandler()
# Create your crew
llm = LLM(model="gpt-4", temperature=0.7)
analyst = Agent(
role="Financial Analyst",
goal="Analyze financial data and provide investment recommendations",
backstory="You are a CFA charterholder with 15 years of experience.",
llm=llm,
verbose=True
)
analysis_task = Task(
description="Analyze the financial health of a tech startup.",
expected_output="Financial analysis report with risk assessment.",
agent=analyst
)
crew = Crew(
agents=[analyst],
tasks=[analysis_task],
verbose=True
)
# Pass the Langfuse handler to kickoff for tracing
result = crew.kickoff()
Building Custom Callback Handlers
If you need fine-grained control over what gets traced, you can build custom callback handlers. CrewAI leverages LangChain's callback system, so you can create handlers that respond to specific events.
from typing import Any, Dict, List, Optional
from uuid import UUID
import json
import time
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from langchain_core.agents import AgentAction, AgentFinish
class CrewAICustomHandler(BaseCallbackHandler):
"""Custom callback handler for detailed CrewAI tracing."""
def __init__(self):
self.llm_calls = []
self.tool_calls = []
self.current_chain_id = None
self.start_times = {}
def on_llm_start(
self, serialized: Dict[str, Any], prompts: List[str],
run_id: UUID, **kwargs
) -> None:
self.start_times[run_id] = time.time()
model_name = serialized.get("name", "unknown")
print(f"[LLM START] Model: {model_name}")
print(f"[LLM START] Prompt preview: {prompts[0][:200]}...")
def on_llm_end(
self, response: LLMResult, run_id: UUID, **kwargs
) -> None:
elapsed = time.time() - self.start_times.get(run_id, time.time())
token_usage = response.llm_output.get("token_usage", {}) if response.llm_output else {}
call_info = {
"run_id": str(run_id),
"duration_seconds": round(elapsed, 3),
"prompt_tokens": token_usage.get("prompt_tokens", 0),
"completion_tokens": token_usage.get("completion_tokens", 0),
"total_tokens": token_usage.get("total_tokens", 0),
}
self.llm_calls.append(call_info)
print(f"[LLM END] Duration: {elapsed:.2f}s, Tokens: {call_info['total_tokens']}")
def on_llm_error(self, error: Exception, run_id: UUID, **kwargs) -> None:
print(f"[LLM ERROR] {error}")
def on_tool_start(
self, serialized: Dict[str, Any], input_str: str,
run_id: UUID, **kwargs
) -> None:
self.start_times[run_id] = time.time()
tool_name = serialized.get("name", "unknown")
print(f"[TOOL START] Tool: {tool_name}, Input: {input_str[:100]}")
def on_tool_end(self, output: str, run_id: UUID, **kwargs) -> None:
elapsed = time.time() - self.start_times.get(run_id, time.time())
print(f"[TOOL END] Duration: {elapsed:.2f}s, Output: {output[:100]}...")
self.tool_calls.append({
"run_id": str(run_id),
"duration_seconds": round(elapsed, 3),
"output_preview": output[:200]
})
def on_tool_error(self, error: Exception, run_id: UUID, **kwargs) -> None:
print(f"[TOOL ERROR] {error}")
def on_agent_action(self, action: AgentAction, run_id: UUID, **kwargs) -> None:
print(f"[AGENT ACTION] Tool: {action.tool}, Input: {action.tool_input[:100]}")
def on_agent_finish(self, finish: AgentFinish, run_id: UUID, **kwargs) -> None:
print(f"[AGENT FINISH] Output: {finish.return_values.get('output', '')[:200]}...")
def get_summary(self) -> dict:
total_tokens = sum(c["total_tokens"] for c in self.llm_calls)
total_llm_time = sum(c["duration_seconds"] for c in self.llm_calls)
total_tool_time = sum(c["duration_seconds"] for c in self.tool_calls)
return {
"total_llm_calls": len(self.llm_calls),
"total_tool_calls": len(self.tool_calls),
"total_tokens_used": total_tokens,
"total_llm_time_seconds": round(total_llm_time, 2),
"total_tool_time_seconds": round(total_tool_time, 2),
"llm_calls": self.llm_calls,
"tool_calls": self.tool_calls,
}
# Usage
handler = CrewAICustomHandler()
llm = LLM(
model="gpt-4",
temperature=0.7,
callbacks=[handler]
)
agent = Agent(
role="Research Assistant",
goal="Find and summarize information",
backstory="You are a helpful research assistant.",
llm=llm,
verbose=True
)
task = Task(
description="Research the benefits of observability in distributed systems.",
expected_output="A summary of 3-5 key benefits.",
agent=agent
)
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()
# Print observability summary
print("\n" + "="*50)
print("OBSERVABILITY SUMMARY")
print("="*50)
summary = handler.get_summary()
print(json.dumps(summary, indent=2))
Tracking Token Usage and Costs
One of the most important aspects of observability for AI agent systems is tracking token usage and associated costs. CrewAI crews can make dozens of LLM calls per execution, and without tracking, costs can spiral out of control.
import os
from crewai import Agent, Task, Crew, LLM
from typing import Dict, List
import json
class CostTracker:
"""Track and estimate costs for LLM calls."""
# Pricing per 1K tokens (update with current rates)
PRICING = {
"gpt-4": {"input": 0.03, "output": 0.06},
"gpt-4-turbo": {"input": 0.01, "output": 0.03},
"gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
"gpt-4o": {"input": 0.005, "output": 0.015},
"gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
}
def __init__(self):
self.calls: List[Dict] = []
def record_call(self, model: str, input_tokens: int, output_tokens: int, agent_name: str):
pricing = self.PRICING.get(model, {"input": 0.01, "output": 0.03})
input_cost = (input_tokens / 1000) * pricing["input"]
output_cost = (output_tokens / 1000) * pricing["output"]
total_cost = input_cost + output_cost
self.calls.append({
"model": model,
"agent": agent_name,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_tokens": input_tokens + output_tokens,
"input_cost": round(input_cost, 6),
"output_cost": round(output_cost, 6),
"total_cost": round(total_cost, 6),
})
def get_report(self) -> Dict:
total_input = sum(c["input_tokens"] for c in self.calls)
total_output = sum(c["output_tokens"] for c in self.calls)
total_cost = sum(c["total_cost"] for c in self.calls)
by_agent = {}
for call in self.calls:
agent = call["agent"]
if agent not in by_agent:
by_agent[agent] = {"calls": 0, "tokens": 0, "cost": 0.0}
by_agent[agent]["calls"] += 1
by_agent[agent]["tokens"] += call["total_tokens"]
by_agent[agent]["cost"] += call["total_cost"]
return {
"total_calls": len(self.calls),
"total_input_tokens": total_input,
"total_output_tokens": total_output,
"total_tokens": total_input + total_output,
"total_cost": round(total_cost, 4),
"by_agent": by_agent,
}
def print_report(self):
report = self.get_report()
print("\n" + "="*60)
print("COST TRACKING REPORT")
print("="*60)
print(f"Total LLM Calls: {report['total_calls']}")
print(f"Total Input Tokens: {report['total_input_tokens']:,}")
print(f"Total Output Tokens: {report['total_output_tokens']:,}")
print(f"Total Tokens: {report['total_tokens']:,}")
print(f"Total Cost: ${report['total_cost']:.4f}")
print("-"*60)
print("Breakdown by Agent:")
for agent, stats in report["by_agent"].items():
print(f" {agent}: {stats['calls']} calls, "
f"{stats['tokens']:,} tokens, ${stats['cost']:.4f}")
print("="*60)
# Integration with CrewAI using callbacks
cost_tracker = CostTracker()
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
class CostTrackingCallback(BaseCallbackHandler):
def __init__(self, tracker: CostTracker, agent_name: str, model: str):
self.tracker = tracker
self.agent_name = agent_name
self.model = model
def on_llm_end(self, response: LLMResult, **kwargs) -> None:
if response.llm_output and "token_usage" in response.llm_output:
usage = response.llm_output["token_usage"]
self.tracker.record_call(
model=self.model,
input_tokens=usage.get("prompt_tokens", 0),
output_tokens=usage.get("completion_tokens", 0),
agent_name=self.agent_name
)
# Create agents with cost tracking
model_name = "gpt-4o-mini"
researcher = Agent(
role="Researcher",
goal="Conduct thorough research",
backstory="Expert researcher",
llm=LLM(
model=model_name,
callbacks=[CostTrackingCallback(cost_tracker, "Researcher", model_name)]
)
)
writer = Agent(
role="Writer",
goal="Write compelling content",
backstory="Expert writer",
llm=LLM(
model=model_name,
callbacks=[CostTrackingCallback(cost_tracker, "Writer", model_name)]
)
)
research_task = Task(
description="Research the history of observability in software engineering.",
expected_output="A timeline of key milestones.",
agent=researcher
)
write_task = Task(
description="Write an article based on the research.",
expected_output="A 500-word article.",
agent=writer
)
crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
result = crew.kickoff()
# Print cost report
cost_tracker.print_report()
Monitoring Crew Execution with Structured Logging
For production systems, structured logging is preferable to print statements. Structured logs are machine-readable and can be ingested by log aggregation systems like Elasticsearch, Splunk, or CloudWatch.
import logging
import json
import sys
from datetime import datetime
from crewai import Agent, Task, Crew
class StructuredFormatter(logging.Formatter):
"""JSON formatter for structured logging."""
def format(self, record):
log_entry = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
# Add custom fields from extra
if hasattr(record, 'crew_id'):
log_entry["crew_id"] = record.crew_id
if hasattr(record, 'agent_role'):
log_entry["agent_role"] = record.agent_role
if hasattr(record, 'task_description'):
log_entry["task_description"] = record.task_description
if hasattr(record, 'duration_ms'):
log_entry["duration_ms"] = record.duration_ms
if hasattr(record, 'token_count'):
log_entry["token_count"] = record.token_count
if record.exc_info:
log_entry["exception"] = self.formatException(record.exc_info)
return json.dumps(log_entry)
# Configure structured logging
logger = logging.getLogger("crewai.observability")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(StructuredFormatter())
logger.addHandler(handler)
# Also log to file
file_handler = logging.FileHandler('crewai_structured.log')
file_handler.setFormatter(StructuredFormatter())
logger.addHandler(file_handler)
# Example usage with structured logging
logger.info(
"Crew execution started",
extra={
"crew_id": "crew-001",
}
)
researcher = Agent(
role="Research Analyst",
goal="Analyze data patterns",
backstory="Data analysis expert",
verbose=False # Disable verbose to rely on structured logs
)
task = Task(
description="Analyze customer churn data and identify key factors.",
expected_output="Analysis report with top 5 churn factors.",
agent=researcher
)
crew = Crew(agents=[researcher], tasks=[task], verbose=False)
import time
start = time.time()
try:
result = crew.kickoff()
duration_ms = int((time.time() - start) * 1000)
logger.info(
"Crew execution completed",
extra={
"crew_id": "crew-001",
"duration_ms": duration_ms,
"token_count": len(str(result)) // 4, # Rough estimate
}
)
except Exception as e:
duration_ms = int((time.time() - start) * 1000)
logger.error(
"Crew execution failed",
extra={
"crew_id": "crew-001",
"duration_ms": duration_ms,
},
exc_info=True
)
raise
Building a Comprehensive Observability Wrapper
Let's put everything together into a comprehensive observability wrapper that you can use across all your CrewAI projects. This wrapper combines structured logging, cost tracking, execution timing, and error handling.
import time
import json
import logging
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field, asdict
from crewai import Agent, Task, Crew
@dataclass
class AgentMetrics:
role: str
tasks_completed: int = 0
total_tokens: int = 0
total_duration_seconds: float = 0.0
errors: int = 0
delegations: int = 0
@dataclass
class CrewMetrics:
crew_id: str
start_time: str = ""
end_time: str = ""
total_duration_seconds: float = 0.0
total_tokens: int = 0
estimated_cost: float = 0.0
status: str = "pending"
agent_metrics: Dict[str, AgentMetrics] = field(default_factory=dict)
errors: List[str] = field(default_factory=list)
task_results: List[Dict] = field(default_factory=list)
class CrewObservabilityWrapper:
"""Comprehensive observability wrapper for CrewAI crews."""
def __init__(self, crew_id: str, enable_logging: bool = True):
self.crew_id = crew_id
self.metrics = CrewMetrics(crew_id=crew_id)
self.logger = logging.getLogger(f"crewai.obs.{crew_id}")
if enable_logging:
self.logger.setLevel(logging.INFO)
if not self.logger.handlers:
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter('%(asctime)s [%(name)s] %(levelname)s: %(message)s')
)
self.logger.addHandler(handler)
def observe(self, crew: Crew, inputs: Optional[Dict] = None) -> Any:
"""Execute crew with full observability."""
from datetime import datetime
self.metrics.start_time = datetime.utcnow().isoformat() + "Z"
self.logger.info(f"Starting crew execution: {self.crew_id}")
self.logger.info(f"Agents: {[a.role for a in crew.agents]}")
self.logger.info(f"Tasks: {len(crew.tasks)}")
# Initialize agent metrics
for agent in crew.agents:
self.metrics.agent_metrics[agent.role] = AgentMetrics(role=agent.role)
start_time = time.time()
try:
result = crew.kickoff(inputs=inputs)
self.metrics.status = "success"
self.metrics.total_duration_seconds = time.time() - start_time
self.metrics.end_time = datetime.utcnow().isoformat() + "Z"
self.logger.info(
f"Crew completed successfully in "
f"{self.metrics.total_duration_seconds:.2f}s"
)
# Record task results
for i, task in enumerate(crew.tasks):
self.metrics.task_results.append({
"task_index": i,
"description": task.description[:100],
"agent": task.agent.role if task.agent else "unassigned",
"completed": True,
})
return result
except Exception as e:
self.metrics.status = "failed"
self.metrics.total_duration_seconds = time.time() - start_time
self.metrics.end_time = datetime.utcnow().isoformat() + "Z"
self.metrics.errors.append(str(e))
self.logger.error(f"Crew failed: {e}", exc_info=True)
raise
finally:
self._save_metrics()
def _save_metrics(self):
"""Save metrics to a JSON file."""
filename = f"crew_metrics_{self.crew_id}.json"
with open(filename, 'w') as f:
json.dump(asdict(self.metrics), f, indent=2)
self.logger.info(f"Metrics saved to {filename}")
def print_summary(self):
"""Print a human-readable summary of the execution."""
m = self.metrics
print("\n" + "="*60)
print(f"CREW EXECUTION SUMMARY: {m.crew_id}")
print("="*60)
print(f"Status: {m.status}")
print(f"Duration: {m.total_duration_seconds:.2f}s")
print(f"Start: {m.start_time}")
print(f"End: {m.end_time}")
print(f"Total Tokens: {m.total_tokens:,}")
print(f"Est. Cost: ${m.estimated_cost:.4f}")
print(f"Errors: {len(m.errors)}")
print("-"*60)
print("Agent Performance:")
for role, am in m.agent_metrics.items():
print(f" {role}:")
print(f" Tasks: {am.tasks_completed}, "
f"Tokens: {am.total_tokens:,}, "
f"Duration: {am.total_duration_seconds:.2f}s, "
f"Errors: {am.errors}")
print("-"*60)
print("Task Results:")
for tr in m.metrics.task_results:
status = "✓" if tr["completed"] else "✗"
print(f" {status} Task {tr['task_index']}: {tr['description']}...")
print("="*60)
# Usage
if __name__ == "__main__":
# Create agents
researcher = Agent(
role="Research Analyst",
goal="Find comprehensive information",
backstory="Expert researcher with 10 years of experience.",
verbose=False
)
analyst = Agent(
role="Data Analyst",
goal="Analyze and interpret data",
backstory="Senior data analyst specializing in trends.",
verbose=False
)
# Create tasks
research_task = Task(
description="Research current trends in AI observability tools.",
expected_output="A list of 5 trending tools with descriptions.",
agent=researcher
)
analysis_task = Task(
description="Analyze the research and identify the top 3 tools.",
expected_output="Ranked analysis of top 3 tools with justification.",
agent=analyst
)
# Create crew
crew = Crew(
agents=[researcher, analyst],
tasks=[research_task, analysis_task],
verbose=False
)
# Execute with observability
observer = CrewObservabilityWrapper("ai-obs-research-001")
result = observer.observe(crew)
# Print summary
observer.print_summary()
print(f"\nFinal Result:\n{result}")
Best Practices for CrewAI Observability
Based on the techniques covered above, here are the key best practices to follow when implementing observability for your CrewAI applications:
- Start with verbose mode in development: Always set
verbose=Trueduring development. It's the fastest way to understand what your agents are doing without any setup. - Use structured logging in production: Replace print statements with structured JSON logs that can be ingested by log aggregation systems. This enables searching, filtering, and alerting.
- Track token usage from day one: Token costs can accumulate quickly with multi-agent crews. Implement cost tracking early so you catch expensive patterns before they reach production.
- Choose the right tracing platform: Use LangSmith if you're already in the LangChain ecosystem, Phoenix for local development and open-source needs, or OpenTelemetry for enterprise distributed tracing.
- Trace the full hierarchy: Ensure your traces capture the crew → task → agent → LLM call hierarchy. Flat traces are much harder to debug than hierarchical ones.
- Monitor for failures and retries: Agents sometimes fail and retry silently. Your observability system should flag repeated failures and excessive retries.
- Log tool inputs and outputs: Tool calls are a common source of failures. Always log what was passed to a tool and what it returned, including error cases.
- Set up alerts on latency: If a crew that normally takes 30 seconds suddenly takes 5 minutes, you need to know immediately. Set up latency alerts in your monitoring system.
- Version your prompts and configurations: Include version information in your traces so you can correlate performance changes with configuration changes.
- Respect privacy and compliance: LLM traces may contain sensitive user data. Ensure your observability platform handles PII appropriately and complies with regulations like GDPR.
- Sample in production: If you have high traffic, consider sampling traces (e.g., 10% of executions