Skip to main content

​
Telemetry Module

The Telemetry Module provides comprehensive monitoring and observability features for PraisonAI agents.

​
Installation

pip install praisonaiagents

# For full observability features
pip install "praisonai[tools]"
It allows you to track performance metrics, usage patterns, and system behavior while maintaining privacy.

​
Import

from praisonaiagents import (
    get_telemetry,
    enable_telemetry,
    disable_telemetry,
    MinimalTelemetry,
    TelemetryCollector
)
Telemetry features require installation with the telemetry extra: pip install praisonaiagents[telemetry]

​
Functions

​
get_telemetry

Retrieves the current telemetry instance. 
def get_telemetry() -> MinimalTelemetry
Returns:
  • MinimalTelemetry: The global telemetry instance. Always returns an instance (never None). Check .enabled to see if telemetry is active.
Example: 
telemetry = get_telemetry()
print(f"Telemetry enabled: {telemetry.enabled}")

​
enable_telemetry

Programmatically enables telemetry. If an environment opt-out variable is set, this call is a no-op. 
def enable_telemetry() -> None
Returns: None
Precedence: Explicit environment opt-out always wins over programmatic calls. Setting DO_NOT_TRACK=true (or any *_DISABLED flag) makes enable_telemetry() a no-op. Otherwise, telemetry is off by default until enable_telemetry() is called or an opt-in env var is exported.
Example: 
from praisonaiagents import enable_telemetry, get_telemetry

# Turn telemetry on (no-op if an env opt-out is set)
enable_telemetry()
assert get_telemetry().enabled is True

​
disable_telemetry

Programmatically disables telemetry. Instrumentation hooks stop installing after this call. 
def disable_telemetry() -> None
Returns: None Example: 
from praisonaiagents import disable_telemetry, get_telemetry

# Disable all telemetry
disable_telemetry()
assert get_telemetry().enabled is False

​
Classes

​
MinimalTelemetry

A lightweight telemetry collector that captures essential metrics only. 
class MinimalTelemetry:
    def __init__(self)
Methods:

​
record_metric

Records a custom metric. 
def record_metric(
    name: str,
    value: Union[int, float],
    unit: Optional[str] = None,
    labels: Optional[Dict[str, str]] = None
) -> None
Parameters:
  • name (str): Metric name
  • value (Union[int, float]): Metric value
  • unit (str, optional): Unit of measurement
  • labels (Dict[str, str], optional): Additional labels for the metric

​
record_event

Records a custom event. 
def record_event(
    name: str,
    attributes: Optional[Dict[str, Any]] = None
) -> None
Parameters:
  • name (str): Event name
  • attributes (Dict[str, Any], optional): Event attributes
Example: 
telemetry = MinimalTelemetry()

# Record metrics
telemetry.record_metric("tokens_processed", 1500, unit="tokens")
telemetry.record_metric("response_time", 0.234, unit="seconds", labels={"model": "gpt-4"})

# Record events
telemetry.record_event("task_completed", attributes={"task_id": "123", "agent": "DataAgent"})

​
TelemetryCollector

Full-featured telemetry collector with advanced capabilities. 
class TelemetryCollector:
    def __init__(
        self,
        service_name: str = "praisonai",
        service_version: Optional[str] = None,
        export_endpoint: Optional[str] = None,
        export_interval: int = 60,
        enable_traces: bool = True,
        enable_metrics: bool = True,
        enable_logs: bool = True
    )
Parameters:
  • service_name (str, optional): Name of the service. Defaults to “praisonai”
  • service_version (str, optional): Service version
  • export_endpoint (str, optional): OpenTelemetry export endpoint
  • export_interval (int, optional): Export interval in seconds
  • enable_traces (bool, optional): Enable trace collection. Defaults to True
  • enable_metrics (bool, optional): Enable metrics collection. Defaults to True
  • enable_logs (bool, optional): Enable log collection. Defaults to True
Methods:

​
start_span

Starts a new telemetry span for tracing. 
@contextmanager
def start_span(
    name: str,
    kind: str = "internal",
    attributes: Optional[Dict[str, Any]] = None
) -> Span
Parameters:
  • name (str): Span name
  • kind (str, optional): Span kind (“internal”, “client”, “server”). Defaults to “internal”
  • attributes (Dict[str, Any], optional): Span attributes

​
record_agent_execution

Records agent execution metrics. 
def record_agent_execution(
    agent_name: str,
    duration: float,
    status: str,
    error: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None
) -> None
Parameters:
  • agent_name (str): Name of the agent
  • duration (float): Execution duration in seconds
  • status (str): Execution status (“success”, “failure”, “partial”)
  • error (str, optional): Error message if failed
  • metadata (Dict[str, Any], optional): Additional metadata

​
record_tool_usage

Records tool usage metrics. 
def record_tool_usage(
    tool_name: str,
    agent_name: str,
    duration: float,
    success: bool,
    parameters: Optional[Dict[str, Any]] = None
) -> None
Parameters:
  • tool_name (str): Name of the tool
  • agent_name (str): Name of the agent using the tool
  • duration (float): Tool execution duration
  • success (bool): Whether the tool execution was successful
  • parameters (Dict[str, Any], optional): Tool parameters (if include_prompts is enabled)

​
export_metrics

Manually triggers metric export. 
def export_metrics() -> None

​
shutdown

Gracefully shuts down the telemetry collector. 
def shutdown() -> None
Example: 
# Create custom telemetry collector
collector = TelemetryCollector(
    service_name="my-ai-service",
    service_version="1.0.0",
    export_endpoint="http://otel-collector:4317",
    enable_traces=True,
    enable_metrics=True
)

# Use with tracing
with collector.start_span("process_request", attributes={"request_id": "456"}):
    # Record agent execution
    collector.record_agent_execution(
        agent_name="ProcessorAgent",
        duration=1.23,
        status="success",
        metadata={"input_size": 1024}
    )
    
    # Record tool usage
    collector.record_tool_usage(
        tool_name="web_search",
        agent_name="ProcessorAgent",
        duration=0.45,
        success=True
    )

# Shutdown when done
collector.shutdown()

​
Environment Variables

Telemetry behavior is controlled by these environment variables: Opt-in (enables telemetry when set to true):
  • PRAISONAI_TELEMETRY_ENABLED=true
  • PRAISONAI_PERFORMANCE_ENABLED=true
Opt-out (disables telemetry; always wins over programmatic calls):
  • PRAISONAI_TELEMETRY_DISABLED=true
  • PRAISONAI_DISABLE_TELEMETRY=true
  • PRAISONAI_PERFORMANCE_DISABLED=true
  • DO_NOT_TRACK=true
Precedence: env opt-out > programmatic override > cached env-default

​
Privacy and Security

  1. No PII by Default: Telemetry does not collect personally identifiable information
  2. Opt-in for Content: Prompt and result content is only collected if explicitly enabled
  3. Local First: Telemetry data stays local unless an export endpoint is configured
  4. Minimal by Default: Default configuration collects only essential metrics

​
Example: Complete Telemetry Setup

from praisonaiagents import (
    Agent,
    enable_telemetry,
    get_telemetry,
    disable_telemetry
)

# Enable telemetry programmatically (no-op if env opt-out is set)
enable_telemetry()

# Create agents - telemetry is automatically integrated
agent = Agent(
    name="AnalysisAgent",
    instructions="Analyze data and provide insights"
)

# Run agent - metrics are automatically collected
result = agent.start("Analyze the sales data")

# Access telemetry data
telemetry = get_telemetry()
if telemetry.enabled:
    # Manually track events (usually automatic)
    telemetry.track_agent_execution("AnalysisAgent", success=True)
    telemetry.track_task_completion("AnalyzeData", success=True)
    telemetry.track_tool_usage("web_search", success=True)
    telemetry.track_error("ValueError")
    telemetry.track_feature_usage("custom_feature")

# Disable telemetry when done
disable_telemetry()
assert get_telemetry().enabled is False

​
Integration with Observability Platforms

For OpenTelemetry, Langfuse, and Prometheus integrations, see the Observability documentation.

​
Best Practices

  1. Start Minimal: Begin with minimal telemetry and increase as needed
  2. Respect Privacy: Never enable prompt/result collection without user consent
  3. Monitor Performance: Use telemetry to identify bottlenecks and optimize
  4. Set Up Alerts: Configure alerts for critical metrics
  5. Regular Reviews: Periodically review collected metrics for insights
  6. Graceful Shutdown: Always call shutdown() in production applications

​
See Also