Skip to content

operonx.telemetry

V3 tracing surface — automatic op-level recording, Consumer subclasses convert the recorded WorkflowTrace into whatever target format they want. See Operon(pipeline, trace=…) for how to wire consumers.

Consumer base

consumer

Consumer base class for V3 workflow-trace consumers.

A Consumer reads a WorkflowTrace (attached to ExecutionHandle.trace after a run) and converts it into whatever target-specific form it wants — a directory tree on disk, a batched HTTP POST to Langfuse, a Loki stream, a PDF report, a dict for a custom UI, and so on.

Base contract is deliberately small:

  • subclasses override :meth:consume, called once the run ends
  • a live consumer also overrides :meth:on_start and/or :meth:on_execution, called as the run starts and as each execution lands, so a store can show a run while it goes
  • :meth:sanitize, :meth:offload_media, :meth:truncate are shared utilities every consumer will want but nothing forces you to use them

No I/O, no state, no coupling to any specific backend. Consumers that DO have state (buffered clients, TTL sweepers, whatever) manage it themselves in their subclass — the base is pure.

See docs/TRACING_V3_DESIGN.md §3-5 for the full design.

Classes

Consumer 

Consumer(config: Optional[Dict[str, Any]] = None)

Bases: ABC

Base class for any V3 workflow-trace consumer.

Subclasses override :meth:consume. The base offers three shared utilities that every real consumer tends to need:

  • :meth:sanitize — strip non-JSON-serialisable values so the payload can round-trip through json.dumps / Parquet / Langfuse's ingest API.
  • :meth:offload_media — replace large binary payloads (audio, numpy arrays, model outputs) with content-addressed refs and write the raw bytes to a media directory. Dedup is automatic.
  • :meth:truncate — cheap string truncation with a hint of the dropped length.

Rules for the base:

  • No I/O, no side effects — every helper is a pure function.
  • No shared state on the instance beyond self.config (a plain dict handed in at __init__).
  • Every helper is opt-in — a minimal consumer can override consume alone and ignore the rest.

Example — a trivial in-memory consumer::

class DictConsumer(Consumer):
    def consume(self, trace):
        return {n.op_id: n.op_name for n in trace.nodes}
Source code in operonx/telemetry/consumer.py 
71
72
def __init__(self, config: Optional[Dict[str, Any]] = None) -> None:
    self.config: Dict[str, Any] = dict(config or {})
Attributes
live property 
live: bool

True when this consumer overrides :meth:on_start or :meth:on_execution: the engine then hands it the run as it goes.

Methods:
on_start 
on_start(trace: WorkflowTrace) -> None

A run began; trace is its live trace (no nodes yet). Optional.

Called on the event loop, inside Operon.start: queue the work and return — a database call here would stall every run in the process. An exception is logged and never reaches the run.

Source code in operonx/telemetry/consumer.py 
74
75
76
77
78
79
80
def on_start(self, trace: WorkflowTrace) -> None:
    """A run began; ``trace`` is its live trace (no nodes yet). Optional.

    Called on the event loop, inside ``Operon.start``: queue the work
    and return — a database call here would stall every run in the
    process. An exception is logged and never reaches the run.
    """
on_execution 
on_execution(trace: WorkflowTrace, execution: OpExecution) -> None

One execution was recorded. Optional.

Called on the event loop right after execution is appended to trace.nodes, so its index there — what a store keys the row by, and what the final write keys it by too — is len(trace.nodes) - 1 while this runs. Queue and return, as for :meth:on_start. consume still gets the whole trace at the end.

Source code in operonx/telemetry/consumer.py 
82
83
84
85
86
87
88
89
90
def on_execution(self, trace: WorkflowTrace, execution: OpExecution) -> None:
    """One execution was recorded. Optional.

    Called on the event loop right after ``execution`` is appended to
    ``trace.nodes``, so its index there — what a store keys the row by,
    and what the final write keys it by too — is
    ``len(trace.nodes) - 1`` while this runs. Queue and return, as for
    :meth:`on_start`. ``consume`` still gets the whole trace at the end.
    """
consume abstractmethod 
consume(trace: WorkflowTrace) -> Any

Convert trace to a target-specific artefact.

Return whatever makes sense for the target — a Path (files written), URL (posted to a service), dict (in-memory view), bytes (a rendered blob). The caller decides what to do with it.

Should raise on unrecoverable errors so the caller can log + continue with the next consumer.

Source code in operonx/telemetry/consumer.py 
101
102
103
104
105
106
107
108
109
110
111
@abstractmethod
def consume(self, trace: WorkflowTrace) -> Any:
    """Convert `trace` to a target-specific artefact.

    Return whatever makes sense for the target — a `Path` (files
    written), URL (posted to a service), dict (in-memory view),
    bytes (a rendered blob). The caller decides what to do with it.

    Should raise on unrecoverable errors so the caller can log +
    continue with the next consumer.
    """
sanitize 
sanitize(payload: Any) -> Any

Recursively strip non-JSON-serialisable values.

Anything that isn't str/int/float/bool/None/dict/list/tuple or a bytes-like object (which offload_media handles) gets replaced with {"$unserializable": "<type_name>"}. Nested dicts/lists are walked in place.

Numpy arrays go through unchanged — offload_media will recognise + offload them. Callers that don't want numpy inline should run offload_media FIRST.

Source code in operonx/telemetry/consumer.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def sanitize(self, payload: Any) -> Any:
    """Recursively strip non-JSON-serialisable values.

    Anything that isn't `str/int/float/bool/None/dict/list/tuple`
    or a bytes-like object (which `offload_media` handles) gets
    replaced with ``{"$unserializable": "<type_name>"}``. Nested
    dicts/lists are walked in place.

    Numpy arrays go through unchanged — `offload_media` will
    recognise + offload them. Callers that don't want numpy inline
    should run `offload_media` FIRST.
    """
    if isinstance(payload, dict):
        return {k: self.sanitize(v) for k, v in payload.items()}
    if isinstance(payload, list):
        return [self.sanitize(v) for v in payload]
    if isinstance(payload, tuple):
        return [self.sanitize(v) for v in payload]
    if isinstance(payload, (str, int, float, bool)) or payload is None:
        return payload
    if isinstance(payload, _BYTES_TYPES):
        # Leave for offload_media OR downstream serializer to handle
        # explicitly — sanitize deliberately doesn't strip these.
        return payload
    # numpy arrays / Media objects / anything with array-like buffer
    # protocol — leave for offload_media.
    if _is_ndarray(payload) or _is_media(payload):
        return payload
    return {"$unserializable": type(payload).__name__}
offload_media 
offload_media(payload: Any, media_dir: Path, threshold: int = 1024) -> Any

Recursively offload large binary values to media_dir.

Any value that's bytes / bytearray / memoryview / numpy.ndarray / operonx.core.media.Media above threshold bytes gets:

  1. serialised to bytes,
  2. hashed with SHA-256,
  3. written to media_dir/<sha256>.<ext> (natural dedup — same bytes → same file, atomic open("xb") with EEXIST tolerated),
  4. replaced in the payload with {"$media_ref": "media/<sha256>.<ext>", "size": <bytes>}.

Payloads smaller than threshold stay inline so the trace remains greppable for small state dicts. Returns a new payload (input is not mutated).

Source code in operonx/telemetry/consumer.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def offload_media(
    self,
    payload: Any,
    media_dir: Path,
    threshold: int = 1024,
) -> Any:
    """Recursively offload large binary values to `media_dir`.

    Any value that's `bytes` / `bytearray` / `memoryview` /
    `numpy.ndarray` / `operonx.core.media.Media` above `threshold`
    bytes gets:

    1. serialised to bytes,
    2. hashed with SHA-256,
    3. written to ``media_dir/<sha256>.<ext>`` (natural dedup — same
       bytes → same file, atomic `open("xb")` with `EEXIST`
       tolerated),
    4. replaced in the payload with
       ``{"$media_ref": "media/<sha256>.<ext>", "size": <bytes>}``.

    Payloads smaller than `threshold` stay inline so the trace
    remains greppable for small state dicts. Returns a new payload
    (input is not mutated).
    """
    media_dir.mkdir(parents=True, exist_ok=True)
    return self._offload_walk(payload, media_dir, threshold)
truncate 
truncate(s: str, limit: int = 500) -> str

Cap s at limit chars, appending a length hint if dropped.

Source code in operonx/telemetry/consumer.py 
174
175
176
177
178
179
def truncate(self, s: str, limit: int = 500) -> str:
    """Cap `s` at `limit` chars, appending a length hint if dropped."""
    if len(s) <= limit:
        return s
    dropped = len(s) - limit
    return f"{s[:limit]}…(+{dropped})"

Concrete consumers

local

LocalConsumer — generic disk-based V3 workflow-trace consumer.

Writes each run to its own directory under <root>, filed by where the run came from (its origin, see :mod:operonx.app.origin), in a layout designed to be read by humans (view.txt) and machines (nodes.jsonl) with zero tooling:

<root>/                              default: <project>/.operonx/runs
  services/<service>/<YYYY-MM-DD>/<trace_id>/
  jobs/<job>/<job_run>/<trace_id>/
  evals/<job>/<job_run>/<trace_id>/
  playground/<YYYY-MM-DD>/<trace_id>/
  adhoc/<workflow>/<YYYY-MM-DD>/<trace_id>/
    meta.json         — workflow name, timings, metadata (origin…),
                        status ("ok"/"error") and the run's errors
    nodes.jsonl       — source of truth (one OpExecution per line,
                        media offloaded to refs; a generator's later
                        records name the first in ``inputs_from``
                        instead of repeating its inputs)
    view.txt          — human-readable chronological rendering
                        (regeneratable from nodes.jsonl at any time)
    media/            — content-addressed offload store
      <sha256>.<ext>
  latest -> <the most recent run's directory>

layout: flat keeps the old <root>/<trace_id>/ shape; any other string is a template over {origin}, {name}, {group}, {day}, {trace_id} and the run's metadata keys. Days are UTC.

Nothing here is callbot-specific — a subclass (CallbotLocalConsumer) overrides _render_view to add turn-grouped headers derived from operonx ctx. See docs/TRACING_V3_DESIGN.md §4-5.

Concurrency / safety:

  • Each trace writes to a per-trace subdirectory → no shared file, no locking.
  • Content-hashed media dedups across traces (create-exclusive on the hash filename, ignore FileExistsError).
  • Writes go to <trace_id>.tmp/ first; atomic rename on success → callers either see a complete directory or nothing.
  • TTL cleanup is external (systemd timer or cron) — the consumer just writes, it doesn't own retention policy.

Classes

LocalConsumer 

LocalConsumer(config: Optional[Dict[str, Any]] = None)

Bases: Consumer

Writes one directory per run under root.

Config keys (all optional, sensible defaults):

  • root (str | Path) — base directory. Unset (or empty), it is $OPERONX_RUNS_DIR, else <project>/.operonx/runs — the project an Application bootstrapped, or the nearest operonx.toml at or above the working directory (a script run inside a project) — else /tmp/operonx_traces. A relative root resolves against the project root.
  • layout (str) — origin (default), flat, or a template; see the module docstring.
  • media_threshold (int) — bytes; payloads at or above this get offloaded to media/. Defaults to 1024.
  • write_view_txt (bool) — set False to skip the human-readable render (raw nodes.jsonl only). Defaults to True.
  • arrow_formatters (dict[str, Formatter]) — per-op summary formatters merged over the built-in FORMATTERS. Any op-name not in the map falls back to :func:default_arrow.

Returns the path to the final directory on success.

Source code in operonx/telemetry/consumer.py 
71
72
def __init__(self, config: Optional[Dict[str, Any]] = None) -> None:
    self.config: Dict[str, Any] = dict(config or {})

LocalConsumerConfig

Bases: YamlModel

YAML-configurable :class:LocalConsumer.

Functions:

resolve_root 

resolve_root(configured: Any = '') -> Path

The directory runs are written under — see root in :class:LocalConsumer.

Source code in operonx/telemetry/consumers/local.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def resolve_root(configured: Any = "") -> Path:
    """The directory runs are written under — see ``root`` in
    :class:`LocalConsumer`."""
    import os

    from operonx.core.workflow_trace import active_project

    project = active_project()
    if configured:
        path = Path(str(configured)).expanduser()
        if not path.is_absolute() and project is not None:
            path = Path(project) / path
        return path
    env = os.environ.get("OPERONX_RUNS_DIR")
    if env:
        return Path(env).expanduser()
    if project is not None:
        return Path(project) / ".operonx" / "runs"
    return Path(_FALLBACK_ROOT)

run_path 

run_path(trace: WorkflowTrace, layout: str = 'origin') -> Path

The run's directory relative to the root, from its metadata.

Source code in operonx/telemetry/consumers/local.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def run_path(trace: WorkflowTrace, layout: str = "origin") -> Path:
    """The run's directory relative to the root, from its metadata."""
    import time as _time

    meta = trace.metadata or {}
    origin = str(meta.get("origin") or "adhoc")
    wall = trace.wall_started_at or _time.time()
    day = _time.strftime("%Y-%m-%d", _time.gmtime(wall))
    if origin in ("job", "eval"):
        name, group = meta.get("job") or trace.workflow_name, meta.get("job_run") or day
    elif origin == "service":
        name, group = meta.get("service") or trace.workflow_name, day
    elif origin == "playground":
        name, group = meta.get("service") or trace.workflow_name, day
    else:
        name, group = trace.workflow_name, day
    if layout == "flat":
        return Path(_safe(trace.trace_id))
    if layout == "origin":
        folder = ORIGIN_DIRS.get(origin, _safe(origin))
        if origin == "playground":
            parts = [folder, group, trace.trace_id]
        else:
            parts = [folder, name, group, trace.trace_id]
        return Path(*[_safe(p) for p in parts])
    fields = {k: v for k, v in meta.items() if not isinstance(v, (dict, list))}
    fields.update(origin=origin, name=name, group=group, day=day, trace_id=trace.trace_id)
    try:
        rendered = layout.format(**fields)
    except (KeyError, IndexError, ValueError):
        rendered = f"{origin}/{name}/{group}/{trace.trace_id}"
    return Path(*[_safe(p) for p in rendered.split("/") if p])

default_arrow 

default_arrow(n: OpExecution) -> str

Fallback per-op arrow when nothing custom is registered.

Just shows the input / output key names — enough to see the shape of the op without spamming values.

Source code in operonx/telemetry/consumers/local.py 
150
151
152
153
154
155
156
def default_arrow(n: OpExecution) -> str:
    """Fallback per-op arrow when nothing custom is registered.

    Just shows the input / output key names — enough to see the shape
    of the op without spamming values.
    """
    return f"{list(n.inputs)}→{list(n.outputs)}"

langfuse

LangfuseConsumer — ship a WorkflowTrace to Langfuse as the tree the run actually had.

The record is flat: one :class:OpExecution per execution, with the scheduler's ctx and every upstream edge. Langfuse is a tree viewer, so this consumer builds the tree that already exists in ctx — a generator's yield record carries the ctx it dispatched, which makes it the container of everything that ran for that item — and never guesses one from data edges (see docs/TRACING_CTX_TREE_PLAN.md §2).

Parent of a record, first hit wins:

  1. ctx main (session-long ops): the trace itself.
  2. a level-1 yield (main.[i]) with no upstream that resolves to a record: the trace itself. (Refs to the root graph's inputs point at engine#main, which is never a record.)
  3. the upstream producer whose ctx is a prefix of, or equal to, this ctx and that started first.
  4. the yield record at this ctx, then at each shorter prefix down to depth 2 — this is what catches an op whose edges stop at a GraphOp boundary.
  5. else a stand-in span for the unrecorded yield main.[i], named after the root-level transient stream (audio_in [357]). Transient streams write no per-item record, so this is the only synthetic node a run needs besides graph containers.

Members of a GraphOp (engine.agent_turn.*) nest under one container span per (graph, ctx). Names are op names; a yield is synthesize [2]. Observation ids are f"{run_id}/{op_id}" — op_id alone is the same string in every run of the same graph, and Langfuse ids are unique per project, so a bare op_id let each call overwrite the spans of the call before it. Times come from the trace's wall-clock anchor. An LLMOp record is a generation (model, usage); everything else a span. Runs post-call, one batch; per-event errors in Langfuse's 207 reply are logged, not dropped.

Example resources.yaml::

trace_langfuse:
  edupia:
    client_resource: langfuse:edupia
    workflow_name:   callbot

Classes

LangfuseConsumer 

LangfuseConsumer(config: Optional[Dict[str, Any]] = None)

Bases: Consumer

Ship a whole :class:WorkflowTrace to Langfuse as one batch.

Config keys:

  • client (LangfuseClient) — REQUIRED; injected by the ResourceHub factory from client_resource.
  • workflow_name — the Langfuse trace name; defaults to the run's.
  • media_threshold — bytes above which a payload becomes a local $media_ref token (default 1024). media_dir — where those go.

Returns the trace URL.

Source code in operonx/telemetry/consumer.py 
71
72
def __init__(self, config: Optional[Dict[str, Any]] = None) -> None:
    self.config: Dict[str, Any] = dict(config or {})

LangfuseConsumerConfig

Bases: YamlModel

YAML-configurable :class:LangfuseConsumer.

parent_strategy is accepted and ignored: the tree comes from ctx now. It stays in the schema so an existing resources.yaml keeps loading; remove it at leisure.

Functions:

build_tree 

build_tree(trace: WorkflowTrace) -> Dict[str, Dict[str, Any]]

Every observation of the run, keyed by id, with its parent set.

Source code in operonx/telemetry/consumers/langfuse.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
def build_tree(trace: WorkflowTrace) -> Dict[str, Dict[str, Any]]:
    """Every observation of the run, keyed by id, with its parent set."""
    rows = sorted(trace.nodes, key=lambda n: n.start_time)
    by_id = {n.op_id: n for n in rows}
    by_ctx: Dict[Tuple[str, ...], List[OpExecution]] = defaultdict(list)
    for n in rows:
        by_ctx[tuple(n.ctx)].append(n)
    stream = next(
        (n.op_name for n in rows if len(n.ctx) == 1 and (n.outputs or {}).get("_transient_stream")),
        None,
    )
    # A run started inside an op body records under that op's step: its own
    # record (op_type "graph") is the root its ops hang under.
    roots: Dict[str, List[OpExecution]] = defaultdict(list)
    for n in rows:
        if n.op_type == NESTED_RUN:
            roots[n.op_full_name].append(n)

    def nested_of(r: OpExecution) -> Optional[Tuple[OpExecution, int]]:
        if not roots:
            return None
        return nested_owner(r.op_full_name, tuple(r.ctx), roots)

    def owner_of(r: OpExecution) -> Optional[str]:
        """The record a child execution was recorded under, walking up past
        any ancestor not in the trace (a run still being written)."""
        full, ctx = r.op_full_name, tuple(r.ctx)
        while True:
            owner = child_parent_id(full, ctx)
            if owner is None:
                return None
            # a child of retried attempt n hangs under that attempt's record
            for cand in (f"{owner}@{r.attempt}", owner):
                if cand in by_id:
                    return cand
            full, ctx = full.rsplit(".", 1)[0], ctx[:-1]

    def parent_of(r: OpExecution) -> Tuple[Optional[str], str]:
        ctx = tuple(r.ctx)
        owner = owner_of(r)
        if owner is not None:
            return owner, "child"
        # Inside a nested run its record is the root: the same rules, with
        # its ctx as the floor instead of ("main",).
        nested = nested_of(r)
        top = nested[0].op_id if nested is not None else None
        floor = len(nested[0].ctx) if nested is not None else 1
        if len(ctx) == floor:
            return (top, "nested run") if nested is not None else (None, "root")
        resolved = [by_id[u.from_op_id] for u in r.upstreams if u.from_op_id in by_id]
        if len(ctx) == floor + 1 and r.is_yield and not resolved:
            return top, "level-1 yield"
        for c in resolved:
            cc = tuple(c.ctx)
            if (
                floor < len(cc) <= len(ctx)
                and ctx[: len(cc)] == cc
                and c.start_time <= r.start_time
            ):
                return c.op_id, "fed by it"
        for k in range(len(ctx), floor, -1):
            cands = [
                c
                for c in by_ctx.get(ctx[:k], [])
                if c is not r and c.start_time <= r.start_time and (k < len(ctx) or c.is_yield)
            ]
            if cands:
                return cands[-1].op_id, "same-ctx yield" if k == len(ctx) else "ctx prefix"
        if nested is not None:
            return top, "nested run"
        return f"stand-in:{format_ctx(ctx[:2])}", "stand-in"

    nodes: Dict[str, Dict[str, Any]] = {}
    for r in rows:
        parent, rule = parent_of(r)
        ctx = tuple(r.ctx)
        if rule == "stand-in" and parent not in nodes:
            nodes[parent] = {
                "id": parent,
                "kind": "stand-in",
                "parent": None,
                "rule": "stand-in",
                "name": f"{stream or 'stream'} [{ctx[1].strip('[]')}]",
                "start": r.start_time,
                "end": r.end_time,
                "record": None,
                "ctx": format_ctx(ctx[:2]),
            }
        # GraphOp members nest under one container per (graph path, ctx),
        # placed where the first member's parent is; a member fed by a
        # sibling member keeps that sibling as its parent.
        parts = r.op_full_name.split(".")
        # the graphs between the run's root (a nested run's record, for its
        # ops) and the op; a nested run's are named by full path
        nested = nested_of(r) if child_parent_id(r.op_full_name, ctx) is None else None
        base = nested[1] if nested is not None else 1
        graphs = parts[base:-1]
        # A child execution sits under the op that ran it, never in a
        # container named after that op.
        if graphs and child_parent_id(r.op_full_name, ctx) is None:
            parent_rec = by_id.get(parent) if parent else None
            sibling = (
                parent_rec is not None and parent_rec.op_full_name.split(".")[base:-1] == graphs
            )
            if not sibling:
                above = parent
                for depth in range(1, len(graphs) + 1):
                    named = parts[: base + depth] if nested is not None else graphs[:depth]
                    cid = f"graph:{'.'.join(named)}#{format_ctx(ctx)}"
                    if cid not in nodes:
                        nodes[cid] = {
                            "id": cid,
                            "kind": "container",
                            "parent": above,
                            "rule": "graph container",
                            "name": graphs[depth - 1],
                            "start": r.start_time,
                            "end": r.end_time,
                            "record": None,
                            "ctx": format_ctx(ctx),
                        }
                    above = cid
                parent = above
        nodes[r.op_id] = {
            "id": r.op_id,
            "kind": "record",
            "parent": parent,
            "rule": rule,
            "name": f"{r.op_name} [{_idx(ctx)}]" if r.is_yield else r.op_name,
            "start": r.start_time,
            "end": r.end_time,
            "record": r,
            "ctx": format_ctx(ctx),
        }
    # synthetic nodes span their descendants
    children: Dict[Optional[str], List[str]] = defaultdict(list)
    for n in nodes.values():
        children[n["parent"]].append(n["id"])

    def extent(nid: str) -> Tuple[float, float]:
        n = nodes[nid]
        s, e = n["start"], n["end"]
        for k in children.get(nid, []):
            ks, ke = extent(k)
            s, e = min(s, ks), max(e, ke)
        if n["kind"] != "record":
            n["start"], n["end"] = s, e
        return s, e

    for nid in children[None]:
        extent(nid)
    return nodes

Langfuse backend

Low-level Langfuse HTTP client + prompt manager — reused by LangfuseConsumer for shipping traces, and available standalone for prompt fetching.

backends

Observability backends.

Each backend provides
  • Config class: registered to ResourceHub via the YAML config layer
  • Client class: HTTP/grpc transport, used by exporters
Available backends
  • langfuse: Langfuse observability platform

Classes

LangfuseClient 

LangfuseClient(config: LangfuseConfig)

Langfuse client using the public REST API for tracing.

Pure HTTP — no SDK dependency. Used by LangfuseTracer for trace ingestion.

Example 
from operonx.core.registry import ResourceHub

client = ResourceHub.instance().get("langfuse:default")
client.ingest([{"id": "...", "type": "trace-create", "body": {...}}])
Source code in operonx/telemetry/backends/langfuse/client.py 
34
35
36
37
def __init__(self, config: LangfuseConfig):
    self._config = config
    self._auth = base64.b64encode(f"{config.public_key}:{config.secret_key}".encode()).decode()
    self._ingest_url = f"{config.host.rstrip('/')}/api/public/ingestion"
Methods:
ingest 
ingest(batch: List[Dict[str, Any]], timeout: int = 30) -> Dict[str, Any]

Send a batch of events to Langfuse ingestion API.

Parameters:

Name Type Description Default
batch List[Dict[str, Any]]

List of ingestion events (trace-create, span-create, etc.)

 required
timeout int

Request timeout in seconds

 30

Returns:

Type Description
Dict[str, Any]

Response dict with 'successes' and 'errors' lists
Source code in operonx/telemetry/backends/langfuse/client.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def ingest(self, batch: List[Dict[str, Any]], timeout: int = 30) -> Dict[str, Any]:
    """Send a batch of events to Langfuse ingestion API.

    Args:
        batch: List of ingestion events (trace-create, span-create, etc.)
        timeout: Request timeout in seconds

    Returns:
        Response dict with 'successes' and 'errors' lists
    """
    metadata = {
        "batch_size": len(batch),
        "sdk_integration": "default",
        "sdk_name": "python",
        "sdk_version": "operonx",
        "public_key": self._config.public_key,
    }
    body = json.dumps({"batch": batch, "metadata": metadata}, default=str).encode("utf-8")
    req = urllib.request.Request(
        self._ingest_url,
        data=body,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Basic {self._auth}",
            "X-Langfuse-Sdk-Name": "python",
            "X-Langfuse-Sdk-Version": "operonx",
            "X-Langfuse-Public-Key": self._config.public_key,
        },
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read())
upload_media 
upload_media(
    *,
    trace_id: str,
    field: str,
    content_type: str,
    content: bytes,
    observation_id: Optional[str] = None,
    timeout: int = 30,
) -> Optional[str]

Upload a media blob and return its Langfuse media reference token.

Two-step flow per Langfuse docs
  1. POST /api/public/media with metadata → get mediaId + presigned URL.
  2. PUT the raw bytes to the presigned URL with the required headers.

Parameters:

Name Type Description Default
trace_id str

Trace this media belongs to.

 required
field str

Where the media lives — "input", "output", or "metadata".

 required
content_type str

MIME type, e.g. "image/png".

 required
content bytes

Raw bytes to upload.

 required
observation_id Optional[str]

Optional observation (span / generation) ID.

 None
timeout int

HTTP timeout per request.

 30

Returns:

Type Description
Optional[str]

The reference token string
Optional[str]

@@@langfuseMedia:type=<mime>|id=<mediaId>|source=bytes@@@,
Optional[str]

or None if the upload failed.
Source code in operonx/telemetry/backends/langfuse/client.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def upload_media(
    self,
    *,
    trace_id: str,
    field: str,
    content_type: str,
    content: bytes,
    observation_id: Optional[str] = None,
    timeout: int = 30,
) -> Optional[str]:
    """Upload a media blob and return its Langfuse media reference token.

    Two-step flow per Langfuse docs:
      1. POST /api/public/media with metadata → get mediaId + presigned URL.
      2. PUT the raw bytes to the presigned URL with the required headers.

    Args:
        trace_id: Trace this media belongs to.
        field: Where the media lives — ``"input"``, ``"output"``, or
            ``"metadata"``.
        content_type: MIME type, e.g. ``"image/png"``.
        content: Raw bytes to upload.
        observation_id: Optional observation (span / generation) ID.
        timeout: HTTP timeout per request.

    Returns:
        The reference token string
        ``@@@langfuseMedia:type=<mime>|id=<mediaId>|source=bytes@@@``,
        or ``None`` if the upload failed.
    """
    content_length = len(content)
    sha256 = base64.b64encode(hashlib.sha256(content).digest()).decode()

    body_dict: Dict[str, Any] = {
        "traceId": trace_id,
        "field": field,
        "contentType": content_type,
        "contentLength": content_length,
        "sha256Hash": sha256,
    }
    if observation_id:
        body_dict["observationId"] = observation_id

    url = f"{self._config.host.rstrip('/')}/api/public/media"
    req = urllib.request.Request(
        url,
        data=json.dumps(body_dict).encode("utf-8"),
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Basic {self._auth}",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            init = json.loads(resp.read())
    except Exception as e:
        LOGGER.warning("Langfuse media init failed: %s", e)
        return None

    upload_url = init.get("uploadUrl")
    media_id = init.get("mediaId")
    if not media_id:
        LOGGER.warning("Langfuse media init returned no mediaId: %r", init)
        return None

    # If the server returned a null uploadUrl the blob is already stored
    # (dedup by sha256). Skip the PUT and return the token.
    if upload_url:
        upload_method = init.get("uploadHttpMethod") or "PUT"
        # Langfuse's own SDK ignores uploadHttpHeaders from the init
        # response and constructs headers client-side. The presigned S3
        # URL is signed for exactly these headers, so we must send them
        # verbatim:
        #   Content-Type             — matches declared contentType
        #   x-amz-checksum-sha256    — S3 integrity check
        #   x-ms-blob-type           — Azure Blob (ignored by S3)
        upload_headers = {
            "Content-Type": content_type,
            "x-amz-checksum-sha256": sha256,
            "x-ms-blob-type": "BlockBlob",
        }

        put_req = urllib.request.Request(
            upload_url,
            data=content,
            headers=upload_headers,
            method=upload_method,
        )
        try:
            with urllib.request.urlopen(put_req, timeout=timeout) as put_resp:
                if put_resp.status >= 300:
                    LOGGER.warning("Langfuse media PUT returned %d", put_resp.status)
                    return None
                # Some providers require an explicit PATCH to confirm the
                # upload (see below).
        except urllib.error.HTTPError as e:
            body = ""
            try:
                body = e.read().decode(errors="replace")[:500]
            except Exception:
                pass
            LOGGER.warning(
                "Langfuse media PUT failed: %s (headers=%s body=%r)",
                e,
                list(upload_headers.keys()),
                body,
            )
            return None
        except Exception as e:
            LOGGER.warning("Langfuse media PUT failed: %s", e)
            return None

        # Langfuse expects a PATCH to /api/public/media/{mediaId} to mark
        # the upload complete with the status code + timing. Without this
        # the blob is orphaned and the reference token won't resolve.
        self._confirm_media_upload(media_id, 200, timeout)

    return f"@@@langfuseMedia:type={content_type}|id={media_id}|source=bytes@@@"
trace_url 
trace_url(trace_id: str) -> str

Build the Langfuse UI URL for a trace.

Source code in operonx/telemetry/backends/langfuse/client.py 
221
222
223
def trace_url(self, trace_id: str) -> str:
    """Build the Langfuse UI URL for a trace."""
    return f"{self._config.host.rstrip('/')}/trace/{trace_id}"
fetch_trace 
fetch_trace(trace_id: str, timeout: int = 10) -> Optional[Dict[str, Any]]

GET /api/public/traces/{traceId} — fetch a posted trace for verification.

Returns the parsed JSON body (trace + nested observations) on success, or None if the trace is not found / unauthorized / unreachable. Used by integration tests to verify trace structure landed correctly.

Source code in operonx/telemetry/backends/langfuse/client.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def fetch_trace(self, trace_id: str, timeout: int = 10) -> Optional[Dict[str, Any]]:
    """GET /api/public/traces/{traceId} — fetch a posted trace for verification.

    Returns the parsed JSON body (trace + nested observations) on success,
    or None if the trace is not found / unauthorized / unreachable. Used
    by integration tests to verify trace structure landed correctly.
    """
    url = f"{self._config.host.rstrip('/')}/api/public/traces/{trace_id}"
    req = urllib.request.Request(
        url,
        headers={"Authorization": f"Basic {self._auth}"},
        method="GET",
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            if resp.status != 200:
                return None
            body = resp.read().decode("utf-8")
            return json.loads(body)
    except urllib.error.HTTPError as e:
        if e.code == 404:
            return None  # not yet ingested or wrong id
        raise
    except Exception:
        return None
auth_check 
auth_check() -> bool

Check authentication by hitting the health endpoint.

Returns:

Type Description
bool

True if authentication is successful
Source code in operonx/telemetry/backends/langfuse/client.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def auth_check(self) -> bool:
    """Check authentication by hitting the health endpoint.

    Returns:
        True if authentication is successful
    """
    url = f"{self._config.host.rstrip('/')}/api/public/health"
    req = urllib.request.Request(
        url,
        headers={"Authorization": f"Basic {self._auth}"},
        method="GET",
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            return resp.status == 200
    except Exception:
        return False

LangfuseConfig

Bases: YamlModel

Configuration for Langfuse observability backend.

This config is registered to ResourceHub and used to create LangfuseClient.

Attributes:

Name Type Description
public_key str

Public API key for Langfuse authentication
secret_key str

Secret API key for Langfuse authentication
host str

Langfuse server URL (default: cloud.langfuse.com)
no_proxy Optional[str]

Proxy bypass setting for internal networks
enabled bool

Whether tracing is enabled
sample_rate float

Sampling rate for traces (0.0 to 1.0)
Example 
# resources.yaml
langfuse:default:
  public_key: pk-...
  secret_key: sk-...
  host: https://cloud.langfuse.com

from operonx.core.registry import ResourceHub

client = ResourceHub.instance().get("langfuse:default")
Methods:
from_env classmethod 
from_env() -> LangfuseConfig

Create config from environment variables.

Environment variables
  • LANGFUSE_PUBLIC_KEY
  • LANGFUSE_SECRET_KEY
  • LANGFUSE_HOST (optional)
  • NO_PROXY (optional)
Source code in operonx/telemetry/backends/langfuse/config.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@classmethod
def from_env(cls) -> "LangfuseConfig":
    """Create config from environment variables.

    Environment variables:
        - LANGFUSE_PUBLIC_KEY
        - LANGFUSE_SECRET_KEY
        - LANGFUSE_HOST (optional)
        - NO_PROXY (optional)
    """
    import os

    return cls(
        public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
        secret_key=os.environ["LANGFUSE_SECRET_KEY"],
        host=os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com"),
        no_proxy=os.environ.get("NO_PROXY"),
    )

LangfusePromptManager 

LangfusePromptManager(
    config: Optional[LangfuseConfig] = None, resource: Optional[str] = None
)

Manages Langfuse prompts via the official SDK.

Requires the langfuse package (pip install operonx-telemetry[langfuse]).

Example 
from operonx.telemetry import LangfusePromptManager, LangfuseConfig

pm = LangfusePromptManager(config=LangfuseConfig.from_env())
text = pm.get_prompt_text("my-prompt")
formatted = pm.format_prompt("my-prompt", name="Alice")

# Bracket notation
text = pm["my-prompt"]
Source code in operonx/telemetry/backends/langfuse/prompt_manager.py 
34
35
36
37
38
39
40
41
42
43
44
45
def __init__(
    self,
    config: Optional[LangfuseConfig] = None,
    resource: Optional[str] = None,
):
    if config is None and resource is None:
        raise ValueError("Must provide either 'config' or 'resource'")
    if config is not None and resource is not None:
        raise ValueError("Cannot provide both 'config' and 'resource'")
    self._config = config
    self._resource = resource
    self._sdk_client = None
Methods:
get_prompt 
get_prompt(name: str, version: Optional[int] = None, **kwargs)

Get a prompt object from Langfuse.

Source code in operonx/telemetry/backends/langfuse/prompt_manager.py 
71
72
73
74
75
def get_prompt(self, name: str, version: Optional[int] = None, **kwargs):
    """Get a prompt object from Langfuse."""
    if version:
        return self._langfuse.get_prompt(name, version=version, **kwargs)
    return self._langfuse.get_prompt(name, **kwargs)
get_prompt_text 
get_prompt_text(name: str, version: Optional[int] = None) -> str

Get prompt text content.

Source code in operonx/telemetry/backends/langfuse/prompt_manager.py 
77
78
79
80
def get_prompt_text(self, name: str, version: Optional[int] = None) -> str:
    """Get prompt text content."""
    prompt = self.get_prompt(name, version=version)
    return prompt.prompt
format_prompt 
format_prompt(name: str, **variables) -> str

Get and format a prompt with variables.

Source code in operonx/telemetry/backends/langfuse/prompt_manager.py 
82
83
84
85
def format_prompt(self, name: str, **variables) -> str:
    """Get and format a prompt with variables."""
    prompt_text = self.get_prompt_text(name)
    return prompt_text.format(**variables)