Praison AI

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

bash
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

python
from praisonaiagents import (
    get_telemetry,
    enable_telemetry,
    disable_telemetry,
    MinimalTelemetry,
    TelemetryCollector
)

Functions

get_telemetry

Retrieves the current telemetry instance.

python
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:

python
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.

python
def enable_telemetry() -> None

Returns: None

Example:

python
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.

python
def disable_telemetry() -> None

Returns: None

Example:

python
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.

python
class MinimalTelemetry:
    def __init__(self)

Methods:

record_metric

Records a custom metric.

python
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.

python
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:

python
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.

python
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.

python
@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.

python
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.

python
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.

python
def export_metrics() -> None

shutdown

Gracefully shuts down the telemetry collector.

python
def shutdown() -> None

Example:

python
# 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

python
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