Telemetry Module

Telemetry and observability functions for monitoring agent performance

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

• Telemetry Feature Documentation
• Monitoring with AgentOps
• Latency Tracking
• Display Module