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.

The user starts an orchestrated run; init and finalize hooks emit traces to AgentOps and other providers.

praisonai.observability.hooks.init_observability(framework_tag, *, tags=None) and finalize_observability(_framework_tag, *, status=...) are public hooks. The orchestrator calls both automatically; custom framework adapters should use observability_session() as the recommended pattern.

---

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.

   Alternative: manual try/except pattern

   from praisonai.framework_adapters.base import BaseFrameworkAdapter
   from praisonai.observability.hooks import init_observability, finalize_observability
   
   class MyAdapter(BaseFrameworkAdapter):
       name = "myframework"
   
       def setup(self, *, framework_tag: str) -> None:
           init_observability(framework_tag, tags=["tenant=acme"])
   
       def run(self, *args, **kwargs):
           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: orchestrator calls init_observability(adapter.name) immediately after assert_framework_available(...) and before adapter.setup(...)
• 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: built-in adapters (praisonai_adapter, crewai_adapter, AutoGen v0.4, AG2) now wrap run()/arun() in a try: ... finally: block. finalize_observability(self.name, status=status) is called from that finally, with status derived from sys.exc_info(). AgentOps sessions are tagged "Success" on the happy path and "Failure" whenever an exception is propagating (including KeyboardInterrupt, LLM errors, rate limits, cancellation).
• 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

Parameter	Type	Default	Description
framework_tag	str	required	Primary tag (e.g. "crewai", "autogen_v4"). Becomes the first entry in default_tags passed to agentops.init.
tags	list[str] | None	None	Extra tags appended after framework_tag.

finalize_observability

Parameter	Type	Default	Description
_framework_tag	str	required	Framework name for context (reserved for future observability providers — currently unused, but pass framework_tag for forward-compat).
status	str (keyword-only)	"Success"	Session status passed to agentops.end_session(...). Conventional values: "Success", "Failure".

observability_session

Parameter	Type	Default	Description
framework_tag	str	required	Framework tag passed to init_observability and finalize_observability.
tags	list[str] | None	None	Extra tags forwarded to init_observability.

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

discover_observability_sinks

Signature	Returns	Description
() -> list[Callable]	List of factory callables	Returns entry-point-registered sink factories. Lazy + best-effort: broken plugins are logged at DEBUG and never break a run.

---

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

---

Best Practices

Use observability_session for custom adapters

observability_session is the recommended pattern for custom adapters. It guarantees finalize_observability always runs — on success and on any failure — with the correct status derived from sys.exc_info(). This prevents AgentOps/other sessions from being orphaned in an “in progress” state on error, KeyboardInterrupt, or rate-limit paths.

from praisonai.observability.hooks import observability_session

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

Status convention

Use status="Success" for the happy path and status="Failure" in exception cases. The string is passed verbatim to agentops.end_session(...); future providers may map other values. When using observability_session, status is derived automatically.

Use for run-scoped tags only

The orchestrator calls init_observability(adapter.name) once per run. If you call it again from setup(), you’ll re-init with your tags (last call wins for AgentOps). Use this for run-scoped tags only:

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

Don't import agentops directly

Don’t import agentops at the top of your adapter — gate it behind is_agentops_available() or rely on the hook to no-op silently:

# ✅ 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-proof for new providers

New providers (Langfuse, W&B, etc.) will be added inside _init_<provider> helpers in praisonai/observability/hooks.py — calling init_observability(...) will automatically pick them up; you don’t need to update adapter code:

# 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

AgentOps

AgentOps integration documentation

Framework Adapter Plugins

How to create custom framework adapters

Custom Tracing

ContextTraceSink protocol and third-party sink plugins

Gateway Tracing Hook

Emit OpenTelemetry spans across each gateway pipeline stage