Read-side helpers

The trace_reader submodule is the inverse of the write-side decorators: it pulls structured information back out of a finished Langfuse trace. It exists so every pod that evaluates a turn against its own trace shares one polling strategy and one set of JSON coercions, rather than each writing their own.

Why polling is necessary

OTel spans flushed by the running application reach Langfuse asynchronously. At the moment an HTTP turn returns, the trace is often not yet queryable. Without polling, a predictor would either sleep blindly for a fixed duration or write its own retry loop. fetch_trace eliminates both by running a backoff schedule and returning as soon as the trace has observations.

fetch_trace

from semantic_tracing.trace_reader import fetch_trace

trace = await fetch_trace(trace_id)

fetch_trace polls Langfuse with exponential backoff (DEFAULT_BACKOFF_SECONDS = (2.0, 4.0, 8.0, 16.0, 32.0, 64.0)) and returns once the trace has at least one observation. Pass a custom backoff tuple to adjust the budget:

trace = await fetch_trace(trace_id, backoff=(0.5, 1.0, 2.0, 4.0))

fetch_trace returns a LangfuseTrace, assembled by read_trace(trace_id, client=None). read_trace reads every observation of the trace through Langfuse's v2 observations API (cursor paged) and returns None when there are none yet.

Field	Meaning
id, name	The trace id and the root observation's trace name
input, output, metadata, tags	The root observation's. In Langfuse a trace has no input or output of its own. Input and output are JSON-decoded where they parse
latency	Seconds from the first observation start to the last end, ignoring observations in the sdk-experiment environment. None when no observation has both times
total_cost	Sum of the observations' costs; None when none has a cost
observations	The Langfuse ObservationV2 objects, with input and output decoded

select_root(observations) picks the trace's root. An offline evaluation adds a bookkeeping span onto a finished trace in the sdk-experiment environment (EXPERIMENT_ENVIRONMENT). It is a second root, and readers ignore it for the root, latency, input and output.

The v2 API truncates a metadata value longer than 200 characters, so a long value arrives cut off.

If no trace with observations appears within the backoff budget, fetch_trace raises TimeoutError with the last underlying error chained via raise ... from. Predictors typically catch this and record it in the prediction record rather than letting the exception propagate.

find_observations

from semantic_tracing.trace_reader import find_observations

obs = find_observations(trace, name="route_request")
obs = find_observations(trace, name_in={"route_request", "orchestrate"})

find_observations filters a trace's observation list by name, preserving the original order. Pass name for an exact single match or name_in for an allow-set. The two parameters are mutually exclusive.

as_dict and coerce_text

Langfuse stores observation inputs and outputs as JSON that may arrive as either a dict or a serialized string depending on how it was set. These helpers normalize the inconsistency:

from semantic_tracing.trace_reader import as_dict, coerce_text

args = as_dict(obs.input)          # always returns a dict or raises
text = coerce_text(obs.output)     # always returns a string

coerce_text accepts an optional prefer_keys tuple. When the value is a dict, it tries each key in order and returns the first string value found:

text = coerce_text(obs.output, prefer_keys=("content", "text", "answer"))

Pod composition pattern

Pod-specific extractors should compose these helpers rather than reach into Langfuse directly. Pod code provides the domain knowledge — which observation names are relevant, which output fields contain the data — while the SDK handles the polling, coercion, and backoff:

from semantic_tracing.trace_reader import (
    as_dict, coerce_text, fetch_trace, find_observations,
)

TOOL_NAMES = frozenset({"read_policy", "create_ticket", "lookup_user"})

def extract_tool_calls(trace):
    return [
        ToolCall(
            name=obs.name,
            args=as_dict(obs.input),
            output=obs.output,
        )
        for obs in find_observations(trace, name_in=TOOL_NAMES)
    ]

This keeps pod-specific knowledge (the tool name set, the output container keys) in the pod and the generic polling and coercion plumbing in the SDK.