> ## Documentation Index > Fetch the complete documentation index at: https://praison.ai/docs/llms.txt > Use this file to discover all available pages before exploring further. # Observability Hooks > Centralized observability init and finalize (AgentOps and future providers) for PraisonAI and custom adapters ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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. The **sync** path now runs `adapter.run()` inside a per-call `scoped_bridge()` (async unchanged), so a stuck coroutine in one run cannot park the shared loop for others. See [Async Bridge → Sync `generate_crew_and_kickoff` auto-scoping](/docs/features/async-bridge#sync-generate_crew_and_kickoff-auto-scoping). ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} graph LR A[🚀 AgentsGenerator] --> B[🪝 observability_session] B --> S[🛠️ adapter.setup] S --> D[🤖 adapter.run / arun] D -->|success| ES[finalize: Success]:::success D -->|exception| EF[finalize: Failure]:::failure ES --> F[✅ AgentOps session end] EF --> F classDef orch fill:#8B0000,stroke:#7C90A0,color:#fff classDef hook fill:#F59E0B,stroke:#7C90A0,color:#fff classDef setup fill:#189AB4,stroke:#7C90A0,color:#fff classDef success fill:#10B981,stroke:#7C90A0,color:#fff classDef failure fill:#8B0000,stroke:#7C90A0,color:#fff classDef run fill:#189AB4,stroke:#7C90A0,color:#fff class A orch class B hook class S setup class D run class ES success class EF failure class F success ``` `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 ```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # From https://app.agentops.ai export AGENTOPS_API_KEY=... ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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() ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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"]) ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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. 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. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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) ``` ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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: 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 `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. * **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. ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} sequenceDiagram participant Gen as AgentsGenerator participant Hook as observability.hooks participant Adapter as Framework Adapter participant AOps as AgentOps Gen->>Hook: enter observability_session(adapter.name) Hook->>AOps: init_observability → agentops.init(...) Gen->>Adapter: _run_adapter_setup(adapter) (adapter.setup) Gen->>Adapter: adapter.run(...) / await adapter.arun(...) Gen->>Hook: exit observability_session (auto-finalize) Hook->>AOps: finalize_observability(status derived from sys.exc_info()) ``` *** ## 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. ### 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) ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} sequenceDiagram participant RunA as observability_session A participant RunB as observability_session B participant AOps as AgentOps RunA->>AOps: start_session → handle A RunB->>AOps: start_session → handle B RunA->>AOps: end_session(handle A) RunB->>AOps: end_session(handle B) Note over RunA,RunB: Separate handles — no cross-finalize, no tag leak ``` ### `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. | | Cached after first call | Same list on subsequent calls | The entry-point walk is process-static, so it is memoized after the first call. Per-run isolation is preserved because sink *instances* still come from calling the cached factories. Call `reset_observability_sink_discovery()` to invalidate. | *** ## 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. ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} graph LR A[🔌 Plugin Package] -->|registers| B[praisonai.observability_sinks entry point] B -->|discovered by| C[discover_observability_sinks]:::tool C -->|returns| D[sink factory callables]:::process D -->|called| E[✅ Active Sinks]:::success classDef plugin fill:#6366F1,stroke:#7C90A0,color:#fff classDef tool fill:#189AB4,stroke:#7C90A0,color:#fff classDef process fill:#F59E0B,stroke:#7C90A0,color:#fff classDef success fill:#10B981,stroke:#7C90A0,color:#fff class A plugin class B plugin class C tool class D process class E success ``` ### Register a sink (plugin authors) ```toml theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # 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 ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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 For custom adapters that are called **directly** (not via `AgentsGenerator`), `observability_session` is required. 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. When invoked via `AgentsGenerator`, the generator's own session already covers the run — an adapter that opens its own inner `observability_session` will nest / double-init and should not. ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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(...) ``` 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. The generator opens `observability_session(adapter.name)` once per run, which calls `init_observability(adapter.name)`. Because `setup()` runs **inside** that session, calling `init_observability` again from `setup()` re-inits with your tags (last call wins for AgentOps). Use this for run-scoped tags only: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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` at the top of your adapter — gate it behind `is_agentops_available()` or rely on the hook to no-op silently: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # ✅ 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 ``` You no longer need to serialize concurrent runs to keep tags clean. Each `observability_session(...)` starts its own AgentOps session via `start_session` and ends only that run's handle. If you were adding a wait/lock around parallel `AgentsGenerator.generate_crew_and_kickoff()` calls specifically to avoid AgentOps cross-contamination, you can remove it. New providers (Langfuse, W\&B, etc.) will be added inside `_init_` helpers in `praisonai/observability/hooks.py` — calling `init_observability(...)` will automatically pick them up; you don't need to update adapter code: ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} # 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 integration documentation How to create custom framework adapters ContextTraceSink protocol and third-party sink plugins Emit OpenTelemetry spans across each gateway pipeline stage