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
: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

def __init__(self, config: Optional[Dict[str, Any]] = None) -> None:
    self.config: Dict[str, Any] = dict(config or {})

Methods:¶

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

@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

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:

serialised to bytes,
hashed with SHA-256,
written to media_dir/<sha256>.<ext> (natural dedup — same bytes → same file, atomic open("xb") with EEXIST tolerated),
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

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

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 <root>/<trace_id>/ in a layout designed to be read by humans (view.txt) and machines (nodes.jsonl) with zero tooling:

<root>/
  <trace_id>/
    meta.json         — workflow name, timings, tags
    nodes.jsonl       — source of truth (one OpExecution per line,
                        media offloaded to refs)
    view.txt          — human-readable chronological rendering
                        (regeneratable from nodes.jsonl at any time)
    media/            — content-addressed offload store
      <sha256>.<ext>
  latest -> <trace_id> — symlink to the most-recent call

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; defaults to /tmp/operonx_traces.
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

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:¶

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

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 — batch-ship a WorkflowTrace to Langfuse at end of call.

Converts each :class:OpExecution in the trace into one Langfuse span observation, walks upstreams to pick a parent (first upstream wins, matching the tree-flattening trade-off explicit in the config), and POSTs the whole batch via :meth:LangfuseClient.ingest. Runs post-call so it never sits in the WS hot path — the trade-off (documented in :doc:TRACING_V3_DESIGN §5) is no live streaming during the call.

Media offload uses the base :meth:Consumer.offload_media walk to replace big payloads with local $media_ref tokens; those aren't Langfuse's media type — the media stays local. For real Langfuse media uploads set upload_media=True (backlog — the current v1 keeps media local only).

Example resources.yaml::

consumer_langfuse:
  edupia:
    client_resource: langfuse:edupia    # reference existing client
    workflow_name:   callbot            # sets Langfuse trace name
    parent_strategy: first_upstream     # or "root_only" / "sequential"

Then hand the resource key to :class:Operon::

engine = Operon(pipeline, trace="consumer_langfuse:edupia")

Classes¶

LangfuseConsumer ¶

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

Bases: Consumer

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

Config keys (all optional, sensible defaults):

client (LangfuseClient instance) — REQUIRED at construct time; typically injected by the ResourceHub factory.
workflow_name (str) — Langfuse trace name; defaults to the run's workflow_name.
parent_strategy (str) — how to pick ONE parent for the tree-only Langfuse span model. Choices:
- "first_upstream" (default) — first UpstreamRef wins.
- "root_only" — no parents; every node hangs off the trace root.
- "sequential" — parent = previous node by start_time.
media_threshold (int) — bytes above which payloads get replaced with a $media_ref token; defaults to 1024.
media_dir (str | Path) — where to write offloaded media blobs; defaults to a per-run temp dir. Media stays local for now.

Returns the Langfuse trace URL on success (via client.trace_url).

Source code in operonx/telemetry/consumer.py

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.

Functions:¶

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

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

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

POST /api/public/media with metadata → get mediaId + presigned URL.
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

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

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

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

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

@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

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

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

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

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)

operonx.telemetry¶

Consumer base¶

consumer ¶

Classes¶

Consumer ¶

Methods:¶

consume abstractmethod ¶

sanitize ¶

offload_media ¶

truncate ¶

Concrete consumers¶

local ¶

Classes¶

LocalConsumer ¶

LocalConsumerConfig ¶

Functions:¶

default_arrow ¶

langfuse ¶

Classes¶

LangfuseConsumer ¶

LangfuseConsumerConfig ¶

Functions:¶

Langfuse backend¶

backends ¶

Classes¶

LangfuseClient ¶

Methods:¶

ingest ¶

upload_media ¶

trace_url ¶

fetch_trace ¶

auth_check ¶

LangfuseConfig ¶

Methods:¶

from_env classmethod ¶

LangfusePromptManager ¶

Methods:¶

get_prompt ¶

get_prompt_text ¶

format_prompt ¶

consume `abstractmethod` ¶

from_env `classmethod` ¶