windlass.interfaces.tracer

tracer

The observability interface.

Windlass instruments every meaningful operation — a model call, a retrieval, a tool execution, an ingestion run — as a :class:Span. A :class:Tracer decides what to do with those spans: drop them, print them, or ship them to LangSmith or Langfuse.

Instrumentation is always on and always cheap. With no tracer configured, spans resolve to :class:NullTracer, whose span() is a context manager that does nothing measurable.

Implementers override one method, :meth:Tracer.start_span.

Example

from windlass.providers.observability.console import ConsoleTracer tracer = ConsoleTracer(enabled=False) with tracer.span("demo", kind="llm") as span: ... _ = span.set_output("done") span.name 'demo'

Span

Span(
    name: str,
    *,
    kind: str = "chain",
    parent_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    trace_id: str | None = None
)

One timed, attributed unit of work.

Spans are created by :meth:Tracer.span and finished automatically when the context manager exits. Exceptions are recorded before propagating, so a failed run is still fully traced.

Parameters:

Name	Type	Description	Default
name	str	Human readable operation name.	required
kind	str	One of :data:SPAN_KINDS.	'chain'
parent_id	str | None	Enclosing span's id, when nested.	None
metadata	dict[str, Any] | None	Static attributes attached at creation.	None
trace_id	str | None	Groups spans belonging to one top-level request.	None

Attributes:

Name	Type	Description
id		Unique span id.
started_at		Monotonic start time.
ended_at	float | None	Monotonic end time, set on :meth:end.
error	str | None	Error message, when the span failed.
usage	Usage | None	Token accounting, for model spans.

Source code in src\windlass\interfaces\tracer.py

def __init__(
    self,
    name: str,
    *,
    kind: str = "chain",
    parent_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    trace_id: str | None = None,
) -> None:
    self.id = uuid.uuid4().hex[:16]
    self.trace_id = trace_id or self.id
    self.parent_id = parent_id
    self.name = name
    self.kind = kind
    self.metadata: dict[str, Any] = dict(metadata or {})
    self.inputs: Any = None
    self.outputs: Any = None
    self.usage: Usage | None = None
    self.error: str | None = None
    self.started_at = time.perf_counter()
    self.ended_at: float | None = None
    self._native: Any = None

duration_ms property

duration_ms: float

Elapsed time in milliseconds — live until the span ends.

set_input

set_input(value: Any) -> Span

Record the operation's input. Returns self for chaining.

Source code in src\windlass\interfaces\tracer.py

def set_input(self, value: Any) -> Span:
    """Record the operation's input. Returns ``self`` for chaining."""
    self.inputs = value
    return self

set_output

set_output(value: Any) -> Span

Record the operation's output. Returns self for chaining.

Source code in src\windlass\interfaces\tracer.py

def set_output(self, value: Any) -> Span:
    """Record the operation's output. Returns ``self`` for chaining."""
    self.outputs = value
    return self

set_usage

set_usage(usage: Usage) -> Span

Record token accounting. Returns self for chaining.

Source code in src\windlass\interfaces\tracer.py

def set_usage(self, usage: Usage) -> Span:
    """Record token accounting. Returns ``self`` for chaining."""
    self.usage = usage
    return self

set_metadata

set_metadata(**fields: Any) -> Span

Merge extra attributes. Returns self for chaining.

Source code in src\windlass\interfaces\tracer.py

def set_metadata(self, **fields: Any) -> Span:
    """Merge extra attributes. Returns ``self`` for chaining."""
    self.metadata.update(fields)
    return self

set_error

set_error(error: BaseException | str) -> Span

Mark the span as failed. Returns self for chaining.

Source code in src\windlass\interfaces\tracer.py

def set_error(self, error: BaseException | str) -> Span:
    """Mark the span as failed. Returns ``self`` for chaining."""
    self.error = str(error)
    return self

end

end() -> Span

Stop the clock. Idempotent.

Source code in src\windlass\interfaces\tracer.py

def end(self) -> Span:
    """Stop the clock. Idempotent."""
    if self.ended_at is None:
        self.ended_at = time.perf_counter()
    return self

attach_native

attach_native(obj: Any) -> Span

Associate the backend's own span object, for Level 3 access.

Source code in src\windlass\interfaces\tracer.py

def attach_native(self, obj: Any) -> Span:
    """Associate the backend's own span object, for Level 3 access."""
    self._native = obj
    return self

native

native() -> Any

Return the backend's span object, if the tracer created one.

Source code in src\windlass\interfaces\tracer.py

def native(self) -> Any:
    """Return the backend's span object, if the tracer created one."""
    return self._native

to_dict

to_dict() -> dict[str, Any]

Return a JSON-safe representation of the span.

Source code in src\windlass\interfaces\tracer.py

def to_dict(self) -> dict[str, Any]:
    """Return a JSON-safe representation of the span."""
    return {
        "id": self.id,
        "trace_id": self.trace_id,
        "parent_id": self.parent_id,
        "name": self.name,
        "kind": self.kind,
        "duration_ms": round(self.duration_ms, 2),
        "metadata": self.metadata,
        "inputs": _clip(self.inputs),
        "outputs": _clip(self.outputs),
        "usage": self.usage.model_dump() if self.usage else None,
        "error": self.error,
    }

Tracer

Tracer(
    *,
    enabled: bool = True,
    project: str | None = None,
    tags: tuple[str, ...] = (),
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract tracing backend.

Parameters:

Name	Type	Description	Default
enabled	bool	Master switch. Disabled tracers still create spans (so code paths are identical) but never export them.	True
project	str | None	Project / session name shown in the backend's UI.	None
tags	tuple[str, ...]	Tags applied to every span.	()
name	str | None	Component name.	None
**config	Any	Backend-specific options.	{}

Example

Implementing a tracer takes one method::

class PrintTracer(Tracer):
    provider_name = "print"

    def start_span(self, span): ...
    def end_span(self, span):
        print(span.to_dict())

Source code in src\windlass\interfaces\tracer.py

def __init__(
    self,
    *,
    enabled: bool = True,
    project: str | None = None,
    tags: tuple[str, ...] = (),
    name: str | None = None,
    **config: Any,
) -> None:
    from windlass.core.config import settings

    super().__init__(
        name=name or self.provider_name,
        enabled=enabled,
        project=project or settings().project,
        tags=tags,
        **config,
    )
    self.enabled = enabled
    self.project = project or settings().project
    self.tags = tuple(tags)
    self._stack: list[Span] = []

start_span abstractmethod

start_span(span: Span) -> None

Called when a span begins.

Parameters:

Name	Type	Description	Default
span	Span	The span that just started. Attach a backend object with :meth:Span.attach_native if the backend has one.	required

Source code in src\windlass\interfaces\tracer.py

@abc.abstractmethod
def start_span(self, span: Span) -> None:
    """Called when a span begins.

    Args:
        span: The span that just started. Attach a backend object with
            :meth:`Span.attach_native` if the backend has one.
    """

end_span

end_span(span: Span) -> None

Called when a span ends, successfully or not.

Parameters:

Name	Type	Description	Default
span	Span	The finished span.	required

Source code in src\windlass\interfaces\tracer.py

def end_span(self, span: Span) -> None:
    """Called when a span ends, successfully or not.

    Args:
        span: The finished span.
    """

flush

flush() -> None

Force any buffered spans to be exported.

Backends that batch should override this; a process that exits without flushing loses its last traces.

Source code in src\windlass\interfaces\tracer.py

def flush(self) -> None:
    """Force any buffered spans to be exported.

    Backends that batch should override this; a process that exits without
    flushing loses its last traces.
    """

span

span(
    name: str, *, kind: str = "chain", inputs: Any = None, **metadata: Any
) -> Iterator[Span]

Open a span around a block of work.

Spans nest automatically: a span opened inside another becomes its child, which is what produces a readable trace tree from plain code.

Parameters:

Name	Type	Description	Default
name	str	Operation name.	required
kind	str	One of :data:SPAN_KINDS.	'chain'
inputs	Any	The operation's input, recorded immediately.	None
**metadata	Any	Extra attributes.	{}

Yields:

Type	Description
Span	The live :class:Span.

Raises:

Type	Description
Exception	Anything the block raises, after recording it.

Example

from windlass.providers.observability.console import ConsoleTracer with ConsoleTracer(enabled=False).span("work", kind="tool") as s: ... s.set_output(42)

Source code in src\windlass\interfaces\tracer.py

@contextmanager
def span(
    self,
    name: str,
    *,
    kind: str = "chain",
    inputs: Any = None,
    **metadata: Any,
) -> Iterator[Span]:
    """Open a span around a block of work.

    Spans nest automatically: a span opened inside another becomes its
    child, which is what produces a readable trace tree from plain code.

    Args:
        name: Operation name.
        kind: One of :data:`SPAN_KINDS`.
        inputs: The operation's input, recorded immediately.
        **metadata: Extra attributes.

    Yields:
        The live :class:`Span`.

    Raises:
        Exception: Anything the block raises, after recording it.

    Example:
        >>> from windlass.providers.observability.console import ConsoleTracer
        >>> with ConsoleTracer(enabled=False).span("work", kind="tool") as s:
        ...     s.set_output(42)
        <Span tool:work ...>
    """
    parent = self._stack[-1] if self._stack else None
    current = Span(
        name,
        kind=kind,
        parent_id=parent.id if parent else None,
        trace_id=parent.trace_id if parent else None,
        metadata={**dict.fromkeys(self.tags, True), **metadata},
    )
    if inputs is not None:
        current.set_input(inputs)

    self._stack.append(current)
    if self.enabled:
        try:
            self.start_span(current)
        except Exception as exc:
            self._log.debug("Tracer %s failed to start a span: %s", self.name, exc)
    try:
        yield current
    except BaseException as exc:
        current.set_error(exc)
        raise
    finally:
        current.end()
        self._stack.pop()
        if self.enabled:
            try:
                self.end_span(current)
            except Exception as exc:
                self._log.debug("Tracer %s failed to end a span: %s", self.name, exc)

current_span

current_span() -> Span | None

Return the innermost open span, if any.

Source code in src\windlass\interfaces\tracer.py

def current_span(self) -> Span | None:
    """Return the innermost open span, if any."""
    return self._stack[-1] if self._stack else None

event

event(name: str, **fields: Any) -> None

Record a zero-duration event on the current span.

Parameters:

Name	Type	Description	Default
name	str	Event name.	required
**fields	Any	Event attributes.	{}

Source code in src\windlass\interfaces\tracer.py

def event(self, name: str, **fields: Any) -> None:
    """Record a zero-duration event on the current span.

    Args:
        name: Event name.
        **fields: Event attributes.
    """
    span = self.current_span()
    if span is not None:
        span.metadata.setdefault("events", []).append({"name": name, **fields})

NullTracer

NullTracer(**config: Any)

Bases: Tracer

A tracer that discards everything.

Used whenever observability is not configured, so instrumented code paths never need a if tracer is not None: guard.

Example

with NullTracer().span("x") as span: ... span.set_output(1)

Source code in src\windlass\interfaces\tracer.py

def __init__(self, **config: Any) -> None:
    super().__init__(enabled=False, **config)

start_span

start_span(span: Span) -> None

Discards the span.

Source code in src\windlass\interfaces\tracer.py

def start_span(self, span: Span) -> None:
    """Discards the span."""

end_span

end_span(span: Span) -> None

Discards the span.

Source code in src\windlass\interfaces\tracer.py

def end_span(self, span: Span) -> None:
    """Discards the span."""