windlass.providers.observability.console

console

Console and in-memory tracers — dependency-free observability.

:class:ConsoleTracer prints an indented trace tree as your pipeline runs. It is the fastest way to answer "what is this thing actually doing?" without signing up for anything::

Windlass.rag().observe("console")

:class:MemoryTracer keeps spans in a list, which is what you want in tests: assert that retrieval ran once, that the guardrail fired, that no LLM call was made on the cached path.

Example

tracer = MemoryTracer() with tracer.span("retrieve", kind="retriever") as span: ... span.set_output(["a", "b"]) [s.name for s in tracer.spans]['retrieve']

ConsoleTracer

ConsoleTracer(
    *,
    stream: TextIO | None = None,
    show_io: bool = False,
    show_usage: bool = True,
    max_value_length: int = 160,
    colour: bool | None = None,
    enabled: bool = True,
    **config: Any
)

Bases: Tracer

Prints spans as an indented tree.

Parameters:

Name	Type	Description	Default
stream	TextIO | None	Where to write. Defaults to sys.stderr, so traces do not contaminate a program's real output.	None
show_io	bool	Print each span's inputs and outputs.	False
show_usage	bool	Print token usage for model spans.	True
max_value_length	int	Truncate printed values at this length.	160
colour	bool | None	Emit ANSI colour. Auto-detected from the stream when None.	None
enabled	bool	Master switch.	True
**config	Any	Forwarded to :class:~windlass.interfaces.tracer.Tracer.	{}

Example

import io buffer = io.StringIO() tracer = ConsoleTracer(stream=buffer, colour=False) with tracer.span("work", kind="tool"): ... pass "work" in buffer.getvalue() True

Source code in src\windlass\providers\observability\console.py

def __init__(
    self,
    *,
    stream: TextIO | None = None,
    show_io: bool = False,
    show_usage: bool = True,
    max_value_length: int = 160,
    colour: bool | None = None,
    enabled: bool = True,
    **config: Any,
) -> None:
    super().__init__(enabled=enabled, **config)
    self.stream = stream or sys.stderr
    self.show_io = show_io
    self.show_usage = show_usage
    self.max_value_length = max_value_length
    self.colour = colour if colour is not None else _supports_colour(self.stream)
    self._depth = 0
    self._lock = threading.RLock()

start_span

start_span(span: Span) -> None

Print the span's opening line and indent.

Source code in src\windlass\providers\observability\console.py

def start_span(self, span: Span) -> None:
    """Print the span's opening line and indent."""
    with self._lock:
        icon = _ICONS.get(span.kind, "•")
        self._write(f"{icon} {self._paint(span.name, '1;36')}  ({span.kind})")
        if self.show_io and span.inputs is not None:
            self._depth += 1
            self._write(f"in:  {self._render(span.inputs)}", dim=True)
            self._depth -= 1
        self._depth += 1

end_span

end_span(span: Span) -> None

Print the span's result line and dedent.

Source code in src\windlass\providers\observability\console.py

def end_span(self, span: Span) -> None:
    """Print the span's result line and dedent."""
    with self._lock:
        self._depth = max(0, self._depth - 1)
        self._depth += 1
        if span.error:
            self._write(f"✗ {self._paint(span.error, '1;31')}")
        else:
            if self.show_io and span.outputs is not None:
                self._write(f"out: {self._render(span.outputs)}", dim=True)
            bits = [f"{span.duration_ms:.0f}ms"]
            if self.show_usage and span.usage and span.usage.total_tokens:
                bits.append(f"{span.usage.total_tokens} tok")
                if span.usage.cost_usd:
                    bits.append(f"${span.usage.cost_usd:.4f}")
            self._write(self._paint("· " + "  ".join(bits), "2"))
        self._depth = max(0, self._depth - 1)

MemoryTracer

MemoryTracer(*, max_spans: int = 10000, enabled: bool = True, **config: Any)

Bases: Tracer

Records every span in memory.

Parameters:

Name	Type	Description	Default
max_spans	int	Ceiling on retained spans; the oldest are dropped. Prevents a long-running process from growing without bound.	10000
enabled	bool	Master switch.	True
**config	Any	Forwarded to :class:~windlass.interfaces.tracer.Tracer.	{}

Attributes:

Name	Type	Description
spans	list[Span]	Finished spans in completion order.

Example

tracer = MemoryTracer() with tracer.span("llm-call", kind="llm"): ... pass tracer.count("llm") 1

Source code in src\windlass\providers\observability\console.py

def __init__(self, *, max_spans: int = 10_000, enabled: bool = True, **config: Any) -> None:
    super().__init__(enabled=enabled, **config)
    self.max_spans = max_spans
    self.spans: list[Span] = []
    self._lock = threading.RLock()

start_span

start_span(span: Span) -> None

No-op: spans are recorded when they finish.

Source code in src\windlass\providers\observability\console.py

def start_span(self, span: Span) -> None:
    """No-op: spans are recorded when they finish."""

end_span

end_span(span: Span) -> None

Record the finished span.

Source code in src\windlass\providers\observability\console.py

def end_span(self, span: Span) -> None:
    """Record the finished span."""
    with self._lock:
        self.spans.append(span)
        if len(self.spans) > self.max_spans:
            del self.spans[: len(self.spans) - self.max_spans]

count

count(kind: str | None = None) -> int

Return how many spans were recorded.

Parameters:

Name	Type	Description	Default
kind	str | None	Restrict the count to one span kind.	None

Returns:

Type	Description
int	The number of matching spans.

Source code in src\windlass\providers\observability\console.py

def count(self, kind: str | None = None) -> int:
    """Return how many spans were recorded.

    Args:
        kind: Restrict the count to one span kind.

    Returns:
        The number of matching spans.
    """
    with self._lock:
        if kind is None:
            return len(self.spans)
        return sum(1 for s in self.spans if s.kind == kind)

by_kind

by_kind(kind: str) -> list[Span]

Return every recorded span of one kind.

Source code in src\windlass\providers\observability\console.py

def by_kind(self, kind: str) -> list[Span]:
    """Return every recorded span of one kind."""
    with self._lock:
        return [s for s in self.spans if s.kind == kind]

errors

errors() -> list[Span]

Return every span that failed.

Source code in src\windlass\providers\observability\console.py

def errors(self) -> list[Span]:
    """Return every span that failed."""
    with self._lock:
        return [s for s in self.spans if s.error]

total_usage

total_usage() -> dict[str, int]

Return summed token usage across every recorded span.

Returns:

Type	Description
dict[str, int]	A dict with prompt_tokens, completion_tokens and
dict[str, int]	total_tokens.

Source code in src\windlass\providers\observability\console.py

def total_usage(self) -> dict[str, int]:
    """Return summed token usage across every recorded span.

    Returns:
        A dict with ``prompt_tokens``, ``completion_tokens`` and
        ``total_tokens``.
    """
    with self._lock:
        prompt = sum(s.usage.prompt_tokens for s in self.spans if s.usage)
        completion = sum(s.usage.completion_tokens for s in self.spans if s.usage)
    return {
        "prompt_tokens": prompt,
        "completion_tokens": completion,
        "total_tokens": prompt + completion,
    }

clear

clear() -> None

Discard every recorded span.

Source code in src\windlass\providers\observability\console.py

def clear(self) -> None:
    """Discard every recorded span."""
    with self._lock:
        self.spans.clear()

to_list

to_list() -> list[dict[str, Any]]

Return every span as a JSON-safe dict.

Source code in src\windlass\providers\observability\console.py

def to_list(self) -> list[dict[str, Any]]:
    """Return every span as a JSON-safe dict."""
    with self._lock:
        return [s.to_dict() for s in self.spans]