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() -> Optional[TelemetryCollector]
Returns:
  • Optional[TelemetryCollector]: The current telemetry collector instance, or None if telemetry is disabled
Example: 
telemetry = get_telemetry()
if telemetry:
    print(f"Telemetry enabled: {telemetry.enabled}")
    print(f"Collection level: {telemetry.level}")

​
enable_telemetry

Enables telemetry collection with specified configuration. 
def enable_telemetry(
    level: str = "minimal",
    custom_collector: Optional[TelemetryCollector] = None,
    export_endpoint: Optional[str] = None,
    export_interval: int = 60,
    include_prompts: bool = False,
    include_results: bool = False
) -> TelemetryCollector
Parameters:
  • level (str, optional): Telemetry level (“minimal”, “basic”, “detailed”). Defaults to “minimal”
  • custom_collector (TelemetryCollector, optional): Custom telemetry collector instance
  • export_endpoint (str, optional): Endpoint for exporting telemetry data
  • export_interval (int, optional): Export interval in seconds. Defaults to 60
  • include_prompts (bool, optional): Include prompt content in telemetry. Defaults to False
  • include_results (bool, optional): Include result content in telemetry. Defaults to False
Returns:
  • TelemetryCollector: The enabled telemetry collector instance
Telemetry Levels:
  • "minimal": Basic metrics only (counts, timings)
  • "basic": Includes error information and performance metrics
  • "detailed": Full telemetry including traces and detailed metrics
Example: 
# Enable minimal telemetry
telemetry = enable_telemetry(level="minimal")

# Enable detailed telemetry with custom endpoint
telemetry = enable_telemetry(
    level="detailed",
    export_endpoint="http://localhost:4317",
    export_interval=30,
    include_prompts=False,
    include_results=False
)

​
disable_telemetry

Disables telemetry collection completely. 
def disable_telemetry() -> None
Example: 
# Disable all telemetry
disable_telemetry()

# Verify telemetry is disabled
telemetry = get_telemetry()
assert telemetry is None or not telemetry.enabled

​
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 can be configured via environment variables:
  • PRAISONAI_TELEMETRY_ENABLED: Enable/disable telemetry (“true”/“false”)
  • PRAISONAI_TELEMETRY_LEVEL: Set telemetry level (“minimal”/“basic”/“detailed”)
  • PRAISONAI_TELEMETRY_ENDPOINT: Set export endpoint
  • PRAISONAI_TELEMETRY_EXPORT_INTERVAL: Set export interval in seconds
  • PRAISONAI_TELEMETRY_INCLUDE_PROMPTS: Include prompts in telemetry (“true”/“false”)
  • PRAISONAI_TELEMETRY_INCLUDE_RESULTS: Include results in telemetry (“true”/“false”)

​
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
)
import os

# Configure via environment
os.environ["PRAISONAI_TELEMETRY_LEVEL"] = "detailed"
os.environ["PRAISONAI_TELEMETRY_ENDPOINT"] = "http://localhost:4317"

# Enable telemetry
telemetry = 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.run("Analyze the sales data")

# Access telemetry data
if telemetry:
    # Custom metric
    telemetry.record_metric(
        "custom_analysis_score",
        value=0.95,
        labels={"category": "sales"}
    )
    
    # Custom event
    telemetry.record_event(
        "analysis_completed",
        attributes={
            "agent": "AnalysisAgent",
            "data_points": 1000
        }
    )

# Disable telemetry when done
disable_telemetry()

​
Integration with Observability Platforms

PraisonAI telemetry is compatible with OpenTelemetry and can export to:
  • Prometheus: Metrics collection and alerting
  • Jaeger: Distributed tracing
  • Grafana: Visualization and dashboards
  • Datadog: Full-stack monitoring
  • New Relic: Application performance monitoring
  • CloudWatch: AWS monitoring service

​
Example: Prometheus Integration

# Enable telemetry with Prometheus endpoint
telemetry = enable_telemetry(
    level="detailed",
    export_endpoint="http://localhost:9090/metrics"
)

# Metrics will be automatically exported in Prometheus format

​
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