operonx.telemetry

Tracing backends — Langfuse, OpenTelemetry, and a local file-based tracer.

tracers

Backwards-compat shim for the legacy tracer constructor.

Pre-T2: each backend had its own Tracer subclass. Post-T2.12: LangfuseTracer is a TracePipeline subclass that wires up a LangfuseTreeExporter internally. New code should use operonx.telemetry.exporters.LangfuseTreeExporter (or LangfuseGroupedTimelineExporter) inside an explicit TracePipeline.

Classes

LangfuseTracer

LangfuseTracer(
    config: Optional["LangfuseConfig"] = None,
    resource: Optional[str] = None,
    tags: Optional[List[str]] = None,
    trace_filter: Optional["TraceFilter"] = None,
)

Bases: TracePipeline

Pipeline-shaped Langfuse tracer with the legacy constructor signature.

Constructor accepts the same args as the pre-T2 tracer. Internally constructs a LangfuseTreeExporter plus the processor chain derived from trace_filter (via trace_filter_to_processors).

Parameters:

Name	Type	Description	Default
config	Optional['LangfuseConfig']	Direct LangfuseConfig (mutually exclusive with resource).	None
resource	Optional[str]	"langfuse:..." ResourceHub key.	None
tags	Optional[List[str]]	Static tags attached to every trace.	None
trace_filter	Optional['TraceFilter']	Legacy TraceFilter config; auto-loaded from the resource's YAML config block if None.	None

Use LangfuseTracer.as_pipeline(...) to get a plain TracePipeline without the legacy attributes — recommended for new code.

Source code in operonx/telemetry/tracers/langfuse.py

def __init__(
    self,
    config: Optional["LangfuseConfig"] = None,
    resource: Optional[str] = None,
    tags: Optional[List[str]] = None,
    trace_filter: Optional["TraceFilter"] = None,
) -> None:
    import warnings

    warnings.warn(
        "LangfuseTracer(...) is deprecated and will be removed in a future "
        "release. Migrate to TracePipeline + LangfuseTreeExporter directly, "
        "or use LangfuseTracer.as_pipeline(...) as a one-line transitional "
        "shim. See docs/TRACING_REDESIGN_PLAN.md §10 (educa migration).",
        DeprecationWarning,
        stacklevel=2,
    )

    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'")

    # Auto-load trace_filter from resource YAML if not supplied — same
    # behavior as the pre-T2 ConfigurableTracer base class.
    if trace_filter is None and resource is not None:
        trace_filter = self._auto_load_trace_filter_from_resource(resource)

    # Build the new exporter + processor chain
    from operonx.telemetry.exporters import LangfuseTreeExporter

    exporter = LangfuseTreeExporter(
        config=config,
        resource=resource,
        tags=tags,
    )
    processors = trace_filter_to_processors(trace_filter) if trace_filter else []

    # Initialize as a TracePipeline so the engine routes through the new path
    TracePipeline.__init__(
        self,
        processors=processors,
        exporters=[exporter],
    )

    # Legacy public attrs — kept so existing callers reading .tags /
    # .resource / .trace_filter / ._get_client() keep working.
    self._config = config
    self._resource = resource
    self.tags = list(tags or [])
    self._trace_filter = trace_filter
    self._exporter = exporter

Functions

to_config_dict

to_config_dict() -> Optional[Dict[str, Any]]

Return Langfuse config dict for the Rust backend.

None when constructed from a ResourceHub key (in which case the Rust side resolves the resource itself).

Source code in operonx/telemetry/tracers/langfuse.py

def to_config_dict(self) -> Optional[Dict[str, Any]]:
    """Return Langfuse config dict for the Rust backend.

    ``None`` when constructed from a ResourceHub key (in which case the
    Rust side resolves the resource itself).
    """
    if self._config is None:
        return None
    return {
        "public_key": self._config.public_key,
        "secret_key": self._config.secret_key,
        "host": self._config.host,
    }

as_pipeline classmethod

as_pipeline(
    config: Optional["LangfuseConfig"] = None,
    resource: Optional[str] = None,
    tags: Optional[List[str]] = None,
    trace_filter: Optional["TraceFilter"] = None,
    *,
    grouped_timeline: bool = False,
) -> TracePipeline

Return a plain TracePipeline configured equivalently.

Use this when you want the new pipeline shape without the legacy attributes (.tags, .resource, etc.). The constructor provides the same wiring plus those attributes for back-compat.

Set grouped_timeline=True for educa's turn-grouped shape (uses LangfuseGroupedTimelineExporter — pair with GroupBy + Aggregate processors).

Source code in operonx/telemetry/tracers/langfuse.py

@classmethod
def as_pipeline(
    cls,
    config: Optional["LangfuseConfig"] = None,
    resource: Optional[str] = None,
    tags: Optional[List[str]] = None,
    trace_filter: Optional["TraceFilter"] = None,
    *,
    grouped_timeline: bool = False,
) -> TracePipeline:
    """Return a plain ``TracePipeline`` configured equivalently.

    Use this when you want the new pipeline shape without the legacy
    attributes (``.tags``, ``.resource``, etc.). The constructor
    provides the same wiring plus those attributes for back-compat.

    Set ``grouped_timeline=True`` for educa's turn-grouped shape (uses
    ``LangfuseGroupedTimelineExporter`` — pair with ``GroupBy`` +
    ``Aggregate`` processors).
    """
    from operonx.telemetry.exporters import (
        LangfuseGroupedTimelineExporter,
        LangfuseTreeExporter,
    )

    if trace_filter is None and resource is not None:
        trace_filter = cls._auto_load_trace_filter_from_resource(resource)

    processors = trace_filter_to_processors(trace_filter) if trace_filter else []

    exporter_cls = LangfuseGroupedTimelineExporter if grouped_timeline else LangfuseTreeExporter
    exporter = exporter_cls(config=config, resource=resource, tags=tags)
    return TracePipeline(processors=processors, exporters=[exporter])