Span, SpanType, TokenUsage

Overview

Why it exists

Every exporter in this package (and any exporter you write) needs to consume a single, backend-agnostic shape of "one unit of traced work." Span is that shape. It's a plain @dataclass, not a context manager — the context-manager behavior (open/close/track/export) lives on Tracer, so Span itself stays a simple, inspectable value object.

When to use it directly

Most code never constructs a Span directly — you get one from with get_tracer().span(...) as span: and call its mutator methods (set_input, set_output, etc.) on the yielded instance. Construct a Span directly only if you're writing infrastructure that bypasses Tracer.span() entirely (which is effectively what Tracer.record_external_span() does internally).

SpanType

class SpanType(str, Enum):
    CHAIN = "chain"
    LLM = "llm"
    TOOL = "tool"
    RETRIEVER = "retriever"
    EMBEDDING = "embedding"
    VECTOR_SEARCH = "vector_search"
    CONTEXT = "context"
    RERANK = "rerank"
    AGENT = "agent"
    PLANNING = "planning"
    TOOL_CALL = "tool_call"
    TOOL_RESULT = "tool_result"
    ROUTER = "router"
    RESPONSE = "response"
    GENERIC = "generic"

A str subclass Enum of well-known span categories. Because it subclasses str, a SpanType member compares equal to and can be used anywhere a plain string is expected (e.g. SpanType.LLM == "llm" is True). Span.type and Tracer.span(..., type=...) both accept either a SpanType member, SpanType.LLM.value, or any custom string — the type annotation on Span.type is str, and nothing in the source validates it against the enum. LangSmithExporter is the one place that treats unknown type strings specially: any type not present in its _RUN_TYPE_MAP falls back to "chain".

TokenUsage

@dataclass
class TokenUsage:
    prompt_tokens: Optional[int] = None
    completion_tokens: Optional[int] = None
    total_tokens: Optional[int] = None

    def to_dict(self) -> Dict[str, Any]: ...

Token accounting for a single LLM call. All three fields default to None. to_dict() returns a plain dict with the same three keys, regardless of whether the values are set — used by Span.to_dict() and by LangSmithExporter.export() (under extra["token_usage"]).

You don't normally construct TokenUsage yourself — call Span.set_token_usage(), which builds one for you and fills in total_tokens if you only supplied the two component counts.

Span

@dataclass
class Span:
    name: str
    type: str = SpanType.GENERIC.value
    id: str = field(default_factory=lambda: uuid.uuid4().hex)
    parent_id: Optional[str] = None
    trace_id: str = field(default_factory=lambda: uuid.uuid4().hex)
    start_time: float = field(default_factory=time.time)
    end_time: Optional[float] = None
    duration: Optional[float] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    input: Any = None
    output: Any = None
    error: Optional[str] = None
    stack_trace: Optional[str] = None
    token_usage: Optional[TokenUsage] = None
    model: Optional[str] = None
    provider: Optional[str] = None

Attributes

[!NOTE] id and trace_id are independently random by default Constructing Span(name="x") directly gives it a new, unrelated trace_id — it is not automatically grouped with any other span. Grouping under one trace_id only happens through Tracer.span(), which copies the parent's trace_id onto the child (new_span.trace_id = parent.trace_id), or by passing trace_id explicitly to Tracer.record_external_span().

Properties

success

@property
def success(self) -> bool

True when error is None. There is no separate "failed" flag — a span is considered failed purely because mark_error() (or finish(error=...)) set self.error.

Methods

set_input

def set_input(self, value: Any) -> None

Sets self.input = value. No copying, validation, or size limiting happens here — truncation for export happens later, in to_dict().

set_output

def set_output(self, value: Any) -> None

Sets self.output = value. Same no-validation behavior as set_input.

set_metadata

def set_metadata(self, **kwargs: Any) -> None

Merges kwargs into self.metadata via dict.update — existing keys not passed in this call are preserved; keys passed again overwrite the previous value. Example: span.set_metadata(user_id="u_123", retries=2).

set_model

def set_model(self, model: Optional[str] = None, provider: Optional[str] = None) -> None

Sets self.model and/or self.provider, but only for arguments that are not None — passing nothing (set_model()) is a no-op; passing only model="gpt-4o" leaves provider untouched (it does not reset it to None).

set_token_usage

def set_token_usage(
    self,
    prompt_tokens: Optional[int] = None,
    completion_tokens: Optional[int] = None,
    total_tokens: Optional[int] = None,
) -> None

Builds a TokenUsage and assigns it to self.token_usage, replacing any previous value (unlike set_model, this is not a merge). If total_tokens is omitted but both prompt_tokens and completion_tokens are given, total_tokens is computed as their sum. If any other combination is given (e.g. only prompt_tokens), total_tokens stays None.

span.set_token_usage(prompt_tokens=120, completion_tokens=45)
span.token_usage.total_tokens  # 165

mark_error

def mark_error(self, exc: BaseException) -> None

Sets self.error to f"{type(exc).__name__}: {exc}" and self.stack_trace to the full formatted traceback (via traceback.format_exception). Does not set end_time or duration — that's finish()'s job. Calling this directly is only useful if you're managing a span's lifecycle yourself, outside Tracer.span().

finish

def finish(self, error: Optional[BaseException] = None) -> None

Sets end_time = time.time(), computes duration = max(0.0, end_time - start_time), and — if error is given — calls mark_error(error). This is what Tracer.span() calls in its finally-adjacent logic when a with block exits, whether normally or via exception.

to_dict

def to_dict(self) -> Dict[str, Any]

Returns every field as a plain dict, JSON-serialization-friendly except that token_usage is converted via TokenUsage.to_dict() (or None), and input/output are passed through the module-level _safe_repr() helper rather than included verbatim. Also includes the computed success property. This is what JSONExporter calls to build each JSONL record.

_safe_repr (internal)

def _safe_repr(value: Any, limit: int = 2000) -> Any

Not part of the public API (module-private, no leading class), but worth understanding because it silently reshapes what ends up in Span.to_dict()["input"] / ["output"]:

• None, int, float, bool pass through unchanged.
• Everything else is converted with str(value) if it's already a str, or repr(value) otherwise.
• If the resulting text is longer than limit (2000 characters, not configurable from to_dict()), it's truncated to limit characters with a suffix like "... [truncated 483 chars]".

Practical effect: span.input / span.output as live attributes on the Span object keep whatever you set (e.g. a full dict or a large DataFrame) — the truncation only happens when to_dict() is called, i.e. at JSONExporter export time. Exporters that don't call to_dict() — ConsoleExporter and LangSmithExporter — read span.input / span.output directly and are not subject to this 2000-character limit.

Thread safety

Span itself has no locking. It's designed to be mutated by exactly one logical owner (the code inside a with tracer.span(...) block) up until it's hand off to the export queue; nothing in the source suggests concurrent mutation of a single Span instance from multiple threads is supported or guarded against.

Complete example

from zerda.observability import get_tracer, SpanType

with get_tracer().span("Chat completion", type=SpanType.LLM.value, model="gpt-4o", provider="openai") as span:
    span.set_input({"messages": [{"role": "user", "content": "hi"}]})
    try:
        reply = call_llm(...)
    except Exception as exc:
        span.mark_error(exc)
        raise
    span.set_output(reply.text)
    span.set_token_usage(prompt_tokens=reply.usage.prompt, completion_tokens=reply.usage.completion)

(span.mark_error(exc) here is redundant with what Tracer.span() already does on an uncaught exception — shown only to illustrate the method; in normal use you'd just let the exception propagate and let Tracer.span() call finish(error=exc) for you.)

Related: Tracer · BaseExporter · Architecture