continuum-observability

name	continuum-observability
description	Trace agent runs with Langfuse, decorate functions with @observe, collect latency/token/error metrics, and report errors. Invoke when the user asks about "see what the LLM was prompted with", "Langfuse traces", "track latency", "metrics dashboard", "error reporting", or "instrument my function".

Continuum Observability Skill

Authoritative source: docs/observability.md.

---

Quick start

Tracing is on by default if Langfuse is configured.

docker compose --profile observability up -d              # brings up Langfuse on :3000

LANGFUSE_ENABLED=true
LANGFUSE_HOST=http://localhost:3000
LANGFUSE_PUBLIC_KEY=…                                      # from Langfuse UI
LANGFUSE_SECRET_KEY=…

Open http://localhost:3000 — every agent run, LLM call, and tool call shows up automatically.

To disable: LANGFUSE_ENABLED=false.

---

Trace your own functions

from orchestrator.observability import observe, SpanLevel

@observe(name="my-pipeline", capture_input=True, capture_output=True,
         metadata={"version": "v2"}, level=SpanLevel.DEFAULT)
async def run_pipeline(data):
    ...

Specialized variants:

from orchestrator.observability import trace_tool, trace_agent

@trace_tool(name="search_db", tool_type="database")
def search_db(query): ...

@trace_agent
async def my_custom_agent_runner(...): ...

---

Manual spans

from orchestrator.core.container import get_container

mgr = get_container().tracing_manager

with mgr.trace(name="onboarding", user_id="u1", session_id="s1",
               metadata={"tier": "pro"}, tags=["beta"]) as trace:
    with mgr.span(name="step-1", input={"foo": 1}) as span:
        span.event(name="cache-hit")
        span.score(name="quality", value=0.92)

Generation spans (LLM calls):

with mgr.span(name="step") as span:
    gen = span.generation(name="llm-call", model="gpt-4o-mini",
                          input=messages)
    # … call the LLM …
    gen.end(output=response_text,
            usage_prompt_tokens=120, usage_completion_tokens=80, usage_total_tokens=200)

---

Async-safe trace context

from orchestrator.observability import (
    set_trace_context, restore_trace_context,
    get_current_trace_id, get_current_session_id, get_current_user_id,
)

token = set_trace_context(trace_id="abc", user_id="u1", session_id="s1")
try:
    await client.chat(messages)                # inherits the trace context
finally:
    restore_trace_context(token)

---

Metrics

from orchestrator.observability import (
    get_metrics_collector, get_metrics_summary, reset_metrics,
)

mc = get_metrics_collector()

# Latency
with mc.track_latency("rag.retrieve") as m:
    docs = await retrieve(query)
print(m.duration_ms)

# Or record directly
mc.record_latency("db.query", 12.4, metadata={"table": "users"})

summary = get_metrics_summary()      # {latency: {...}, tokens: {...}, errors: {...}}

---

Error reporting

from orchestrator.observability import (
    report_error, report_exception, flush_errors,
    enable_error_reporting, disable_error_reporting,
)

try:
    ...
except SomeError as e:
    # `context` is a short label string; structured data goes in `metadata=`.
    report_error(e, context="db_query", user_id="u1", metadata={"table": "users"})

Every OrchestratorError with should_report=True is reported automatically. The reporter has a thread-safe queue (max 1000 entries) and an auto-flush every 5 seconds.

---

Debug tip: log the assembled prompt

LOG_FULL_PROMPT=true

The runner prints the entire message list it sends to the LLM — indispensable for debugging memory / RAG / handoff flows.

---

ObservabilityConfig (when wiring providers manually)

from orchestrator.observability import ObservabilityConfig, initialize_observability

cfg = ObservabilityConfig(
    providers=["langfuse"],
    enabled=True,
    public_key="pk-…", secret_key="sk-…",
    host="http://localhost:3000",
    sample_rate=1.0, flush_interval=1, flush_at=15,
    environment="production", default_tags=["app:billing"],
)
mgr = initialize_observability(cfg)

The lifecycle manager normally calls this for you.

---

Custom provider

from orchestrator.observability import (
    ObservabilityProvider, ProviderCapabilities, register_provider,
)

class MyProvider(ObservabilityProvider):
    def __init__(self): super().__init__(name="my", config={})
    def supports_feature(self, feat): return feat == ProviderCapabilities.TRACE
    def trace(self, name, **kw): ...
    def span(self, *, trace_id, name, **kw): ...
    # …generation, event, score, flush, shutdown

register_provider("my", MyProvider())

---

Don't

• Don't expect to see traces if LANGFUSE_ENABLED=false or the keys are missing.
• Don't rely on @observe to capture huge payloads — truncate_data caps each field at 10 KB by default.
• Don't make @observe-decorated functions sync if the rest of your code is async — the contextvars magic works best in fully-async code.
• Don't flush_langfuse() in the middle of a request loop — it's expensive. Lifecycle shutdown handles it.
• Don't forget to set SHARED_SERVICES_ENABLED=false if your process is the sole owner of Langfuse — otherwise traces don't flush on shutdown.