Observability Hooks

Centralized observability init and finalize (AgentOps and future providers) for PraisonAI and custom adapters

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.")

Observability Hooks provide a centralized entry and exit point for observability providers (AgentOps, future Langfuse, W&B) in PraisonAI and custom framework adapters.

As of PR #3152, AgentsGenerator owns the observability lifecycle. Both the sync (generate_crew_and_kickoff) and async (agenerate_crew_and_kickoff) run paths bracket adapter.setup() and adapter.run()/adapter.arun() in a single observability_session. Adapters no longer initialise or finalize observability themselves — this closes the AutoGen leak (the v0.4 adapter never called finalize, so every run leaked a session) and makes finalize impossible to forget for any future adapter.

praisonai.observability.hooks.init_observability(framework_tag, , tags=None) and finalize_observability(_framework_tag, , status=...) are public hooks. The generator drives both automatically through observability_session(); custom adapters called directly (not via AgentsGenerator) should use observability_session() themselves.

Quick Start

1. Simple Usage

   # 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()

2. With Configuration

   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"])

3. Pair init with finalize in custom adapters

   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(...)

   observability_session calls init_observability on entry and finalize_observability on exit. Status is derived from sys.exc_info() automatically, so success/failure is tagged correctly with no boilerplate.

   Standalone: manual finalize for callers not going through the generator

   This pattern is valid only for standalone callers that invoke an adapter directly, outside AgentsGenerator. Do not copy it into a custom adapter that will be called by AgentsGenerator — the generator already owns init+finalize via observability_session, and a self-finalizing adapter would double-finalize.

   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)

4. Branch on availability

   from praisonai.observability.hooks import is_agentops_available
   
   if is_agentops_available():
       ...  # do extra setup

How It Works

init_observability(framework_tag, *, tags=None) centralizes observability initialization:

• 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:
• _prepare_for_run(config) → validates, resolves adapter; does not run setup, does not init observability
• 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

finalize_observability(_framework_tag, *, status=...) closes observability sessions symmetrically:

• 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. For workflow YAMLs, the session is tagged with the config-resolved framework (registry default when omitted) rather than a hardcoded "praisonai" label — a fix landed in PraisonAI PR #5039.
• 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”

The hook also leaves room for future providers (the source already has placeholder comments for _init_langfuse and _init_wandb), so users may want to know the surface area.

Configuration

init_observability

finalize_observability

observability_session

Returns a context manager (ContextManager[None]). Status is auto-derived from sys.exc_info() — no status kwarg needed.

Concurrent runs

Two overlapping observability_session(...) blocks (parallel agents, nested crews) get their own AgentOps session each — tags don’t leak, and finalizing one no longer ends the other. finalize_observability resolves the owning ObservabilityRun before it calls agentops.end_session(...), so a run whose start_session returned no handle ends nothing instead of tearing down the package-global session a concurrent run still depends on. (PraisonAI #3492)

Per-task isolation under asyncio

Each asyncio task gets its own run stack, so the void-signature finalize_observability(tag) is safe even when N concurrent arun() tasks share one worker thread.

The run stack lives in a contextvars.ContextVar, not threading.local(). asyncio copies the context per task (asyncio.create_task, asyncio.to_thread, copy_context), so _run_stack.get() in each task returns that task’s own stack — task A’s finalize_observability("crewai") pops task A’s run, never task B’s. A thread-local slot would be shared by every task on the thread, letting task A close task B’s sinks mid-run and restore another tenant’s emitter.

The value is treated as immutable (copy-on-write): _push_run, _pop_run, and _remove_run always set a brand-new list, and _get_run_stack() returns a copy. A ContextVar copied into a child context shares the same underlying list object, so mutating it in place would let a spawned task pop from its parent’s stack.

Prefer the explicit-handle form finalize_observability(tag, run=my_run) wherever you hold the handle; the void form is the compatibility shim for legacy call sites. Both are now concurrency-safe — the void form’s _pop_run() reaches only this context’s stack.

The regression suite test_observability_run_stack_isolation.py is the canonical guarantee: _push does not mutate a copied context, _pop in a copied context leaves the parent intact, _get_run_stack returns a copy, two concurrent asyncio tasks get isolated stacks, and _remove_run is copy-on-write. Fixed in PR #4948 (fixes #4945).

discover_observability_sinks

Third-party sink plugins

Third-party packages can register an observability sink factory under the praisonai.observability_sinks entry-point group. PraisonAI discovers them lazily via discover_observability_sinks() — broken plugins are logged at DEBUG and never break a run.

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"

Your factory is a zero-arg (or framework-tag-aware) callable that returns a sink implementing the core SDK’s TraceSinkProtocol.

Discover registered sinks

from praisonai.observability.hooks import discover_observability_sinks

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

Invalidating the cache

The factory list is memoized after the first discovery, so a dynamic plugin install or a test needs to invalidate it.

from praisonai.observability.hooks import reset_observability_sink_discovery

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

Cache invalidation only affects the factory list — already-running observability sessions keep their existing sinks.

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