Observability Hooks

from praisonaiagents import Agent

agent = Agent(
    name="observable-agent",
    instructions="Run with observability hooks for tracing.",
)
agent.start("Process this request and emit trace events.")

​

Quick Start

# From https://app.agentops.ai
export AGENTOPS_API_KEY=...

from praisonai.agents_generator import AgentsGenerator

# init_observability and finalize_observability are called for you
gen = AgentsGenerator("agents.yaml", "crewai", config_list=[...])
gen.generate_crew_and_kickoff()

from praisonai.framework_adapters.base import BaseFrameworkAdapter
from praisonai.observability.hooks import init_observability

class MyAdapter(BaseFrameworkAdapter):
    name = "myframework"

    def setup(self, *, framework_tag: str) -> None:
        # Add extra tags for this run
        init_observability(framework_tag, tags=["tenant=acme", "experiment=foo"])

from praisonai.framework_adapters.base import BaseFrameworkAdapter
from praisonai.observability.hooks import observability_session

class MyAdapter(BaseFrameworkAdapter):
    name = "myframework"

    def run(self, *args, **kwargs):
        with observability_session(self.name, tags=["tenant=acme"]):
            return my_framework.execute(...)

from praisonai.framework_adapters.base import BaseFrameworkAdapter
from praisonai.observability.hooks import init_observability, finalize_observability

class MyStandaloneRunner:
    name = "myframework"

    def run(self, *args, **kwargs):
        init_observability(self.name, tags=["tenant=acme"])
        try:
            result = my_framework.execute(...)
            return result
        except Exception:
            raise
        finally:
            import sys
            status = "Failure" if sys.exc_info()[0] is not None else "Success"
            finalize_observability(self.name, status=status)

from praisonai.observability.hooks import is_agentops_available

if is_agentops_available():
    ...  # do extra setup

​

How It Works

• Auto-call site: the generator opens observability_session(adapter.name) and runs _run_adapter_setup(adapter) inside the session, so setup events and any setup/import failure are recorded and finalized instead of slipping outside observability. The call sequence is:
  1. _prepare_for_run(config) → validates, resolves adapter; does not run setup, does not init observability
  2. with observability_session(adapter.name):
     • _run_adapter_setup(adapter) (calls adapter.setup(framework_tag=adapter.name))
     • adapter.run(...) / await adapter.arun(...)
• AgentOps init guard: agentops.init(...) only fires if both (a) is_agentops_available() returns true, and (b) AGENTOPS_API_KEY is set in the env. Init is centralised here — agents_generator no longer double-inits AgentOps (PR #2062).
• Failure mode: ImportError (no agentops) is logged at DEBUG; any other exception is logged at WARNING and never propagated
• is_agentops_available() lazy function — prefer over the removed eager AGENTOPS_AVAILABLE constant in this module

• Auto-call site: AgentsGenerator.generate_crew_and_kickoff (sync) and agenerate_crew_and_kickoff (async) bracket the entire adapter lifecycle — setup + run/arun — in observability_session(adapter.name). On context exit the session finalizes with status="Failure" when an exception is propagating and status="Success" otherwise. Adapters must not finalize themselves. This closes the AutoGen leak: the AutoGen adapter previously never called finalize_observability, so every AutoGen run left an ObservabilityRun, a _swapped_runs entry, and sink handles alive.
• AgentOps end guard: agentops.end_session(...) only fires if agentops is importable
• Failure mode: ImportError returns silently; any other exception is logged at WARNING and never propagated
• Why symmetric calls matter: without finalize_observability, AgentOps dashboard sessions stay stuck “in progress”

​

Configuration

​

init_observability

​

finalize_observability

​

observability_session

​

Concurrent runs

​

discover_observability_sinks

​

Third-party sink plugins

​

Register a sink (plugin authors)

# pyproject.toml in your plugin package
[project.entry-points."praisonai.observability_sinks"]
my_sink = "my_package.sinks:my_sink_factory"

​

Discover registered sinks

from praisonai.observability.hooks import discover_observability_sinks

for factory in discover_observability_sinks():
    sink = factory()
    ...

​

Invalidating the cache

from praisonai.observability.hooks import reset_observability_sink_discovery

# After a dynamic plugin install, or between tests:
reset_observability_sink_discovery()

​

Best Practices

from praisonai.observability.hooks import observability_session

class MyStandaloneRunner:
    name = "myframework"

    def run(self, *args, **kwargs):
        with observability_session(self.name, tags=["tenant=acme"]):
            return my_framework.execute(...)

def setup(self, *, framework_tag: str) -> None:
    # Good - adds run-specific context
    init_observability(framework_tag, tags=[
        f"tenant={self.tenant_id}",
        f"experiment={self.experiment_name}"
    ])

# ✅ Good - use the hook or check availability
from praisonai.observability.hooks import is_agentops_available, init_observability

if is_agentops_available():
    # Safe to do AgentOps-specific setup
    pass

# ❌ Bad - direct import can fail
import agentops  # May fail if not installed

# Future providers will be added automatically
def init_observability(framework_tag, *, tags=None):
    _init_agentops(framework_tag, tags or [])
    # _init_langfuse(framework_tag, tags)    # Future
    # _init_wandb(framework_tag, tags)       # Future

​

Related