windlass.providers.observability.platforms

platforms

LangSmith and Langfuse tracers.

Both give you a hosted trace viewer: nested spans, prompts, completions, token counts and latency, searchable across runs. That is the difference between "users say it got worse" and "the retriever started returning the wrong section on Tuesday".

Install with::

pip install "windlass[observability]"

Neither tracer is ever allowed to break your application. Every export path swallows its own errors and logs a warning, and every flush runs under a deadline — a vendor client that blocks on an internal queue must not be able to hang the process it is observing, and an exception handler cannot catch a hang.

Example

from windlass import Windlass # doctest: +SKIP rag = Windlass.rag().observe("langfuse") # doctest: +SKIP

LangSmithTracer

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

Bases: Tracer

Sends spans to LangSmith.

Parameters:

Name	Type	Description	Default
api_key	str | None	Credential. Falls back to LANGSMITH_API_KEY / LANGCHAIN_API_KEY.	None
project	str | None	LangSmith project name.	None
api_url	str | None	Endpoint override for self-hosted deployments.	None
tags	tuple[str, ...]	Tags applied to every run.	()
enabled	bool	Master switch.	True
**config	Any	Forwarded to :class:~windlass.interfaces.tracer.Tracer.	{}

Raises:

Type	Description
MissingDependencyError	When langsmith is not installed.
ConfigurationError	When no API key can be found.

Performance

The LangSmith client batches and sends in the background, so tracing adds negligible latency to the request path. Call :meth:flush before a short-lived process exits or you will lose the tail of your traces.

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

def __init__(
    self,
    *,
    api_key: str | None = None,
    project: str | None = None,
    api_url: str | None = None,
    tags: tuple[str, ...] = (),
    enabled: bool = True,
    **config: Any,
) -> None:
    super().__init__(enabled=enabled, project=project, tags=tags, **config)
    langsmith = require("langsmith", extra="observability", feature="The LangSmith tracer")
    key = api_key or settings().secret("langsmith_api_key")
    if not key:
        raise ConfigurationError(
            "No API key configured for LangSmith.",
            hint="Set LANGSMITH_API_KEY, or pass "
            "Windlass.tracer('langsmith', api_key='...').",
        )
    self._client = langsmith.Client(api_key=key, api_url=api_url)
    self._runs: dict[str, Any] = {}

native

native() -> Any

Return the underlying langsmith.Client (Level 3 access).

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

def native(self) -> Any:
    """Return the underlying ``langsmith.Client`` (Level 3 access)."""
    return self._client

start_span

start_span(span: Span) -> None

Create a LangSmith run for the span.

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

def start_span(self, span: Span) -> None:
    """Create a LangSmith run for the span."""
    import uuid

    try:
        run_id = uuid.UUID(span.id.ljust(32, "0")[:32])
        parent_id = uuid.UUID(span.parent_id.ljust(32, "0")[:32]) if span.parent_id else None
        self._client.create_run(
            id=run_id,
            name=span.name,
            run_type=_LANGSMITH_KINDS.get(span.kind, "chain"),
            inputs=_as_dict(span.inputs),
            parent_run_id=parent_id,
            project_name=self.project,
            tags=list(self.tags),
            extra={"metadata": span.metadata},
        )
        self._runs[span.id] = run_id
        span.attach_native(run_id)
    except Exception as exc:
        self._log.debug("LangSmith create_run failed: %s", exc)

end_span

end_span(span: Span) -> None

Close the LangSmith run.

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

def end_span(self, span: Span) -> None:
    """Close the LangSmith run."""
    run_id = self._runs.pop(span.id, None)
    if run_id is None:
        return
    try:
        self._client.update_run(
            run_id,
            outputs=_as_dict(span.outputs),
            error=span.error,
            end_time=None,
            extra={
                "metadata": {
                    **span.metadata,
                    "duration_ms": round(span.duration_ms, 2),
                    **(span.usage.model_dump() if span.usage else {}),
                }
            },
        )
    except Exception as exc:
        self._log.debug("LangSmith update_run failed: %s", exc)

flush

flush() -> None

Wait for queued runs to be sent, giving up after FLUSH_TIMEOUT.

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

def flush(self) -> None:
    """Wait for queued runs to be sent, giving up after ``FLUSH_TIMEOUT``."""
    flusher = getattr(self._client, "flush", None)
    if callable(flusher):
        _bounded_flush(flusher, label="LangSmith", timeout=FLUSH_TIMEOUT, log=self._log)

aclose async

aclose() -> None

Flush, then shut the client's background worker down.

Without the shutdown, the LangSmith SDK's worker thread outlives the application and tries to spawn a new thread from its own interpreter shutdown hook — which Python refuses once teardown has begun. The result is a traceback printed from inside langsmith after the program has already finished successfully, which looks like a crash and is not one.

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

async def aclose(self) -> None:
    """Flush, then shut the client's background worker down.

    Without the shutdown, the LangSmith SDK's worker thread outlives the
    application and tries to spawn a *new* thread from its own interpreter
    shutdown hook — which Python refuses once teardown has begun. The result
    is a traceback printed from inside ``langsmith`` after the program has
    already finished successfully, which looks like a crash and is not one.
    """
    self.flush()
    for method in ("cleanup", "shutdown", "close"):
        closer = getattr(self._client, method, None)
        if callable(closer):
            _bounded_flush(
                closer, label=f"LangSmith {method}", timeout=FLUSH_TIMEOUT, log=self._log
            )
            return

LangfuseTracer

LangfuseTracer(
    *,
    public_key: str | None = None,
    secret_key: str | None = None,
    host: str | None = None,
    project: str | None = None,
    tags: tuple[str, ...] = (),
    enabled: bool = True,
    flush_timeout: float = FLUSH_TIMEOUT,
    **config: Any
)

Bases: Tracer

Sends spans to Langfuse.

Parameters:

Name	Type	Description	Default
public_key	str | None	Langfuse public key. Falls back to LANGFUSE_PUBLIC_KEY.	None
secret_key	str | None	Langfuse secret key. Falls back to LANGFUSE_SECRET_KEY.	None
host	str | None	Endpoint. Falls back to LANGFUSE_HOST, defaulting to the cloud.	None
project	str | None	Session / project name recorded on each trace.	None
tags	tuple[str, ...]	Tags applied to every trace.	()
enabled	bool	Master switch.	True
**config	Any	Forwarded to :class:~windlass.interfaces.tracer.Tracer.	{}

Raises:

Type	Description
MissingDependencyError	When langfuse is not installed.
ConfigurationError	When the key pair is incomplete.

Note

Langfuse's SDK is versioned aggressively and rewrote its span API between v2 and v3. This adapter detects which surface the installed version exposes and uses it:

• v3 / v4 — client.start_observation(as_type=...), whose observation types map almost one-to-one onto Windlass span kinds.
• v2 — client.trace() / parent.span() / parent.generation().

If neither is present the constructor raises. Silently exporting nothing while reporting healthy is a worse failure than refusing to start: you only discover it when you go looking for traces that were never sent.

Attributes:

Name	Type	Description
api		"v3" or "v2" — which surface was detected.

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

def __init__(
    self,
    *,
    public_key: str | None = None,
    secret_key: str | None = None,
    host: str | None = None,
    project: str | None = None,
    tags: tuple[str, ...] = (),
    enabled: bool = True,
    flush_timeout: float = FLUSH_TIMEOUT,
    **config: Any,
) -> None:
    super().__init__(enabled=enabled, project=project, tags=tags, **config)
    langfuse = require("langfuse", extra="observability", feature="The Langfuse tracer")
    cfg = settings()
    public = public_key or cfg.secret("langfuse_public_key")
    secret = secret_key or cfg.secret("langfuse_secret_key")
    if not public or not secret:
        raise ConfigurationError(
            "Langfuse needs both a public and a secret key.",
            hint="Set LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY, or pass them "
            "to Windlass.tracer('langfuse', ...).",
        )
    self.flush_timeout = flush_timeout
    self._client = langfuse.Langfuse(
        public_key=public, secret_key=secret, host=host or cfg.langfuse_host
    )

    if hasattr(self._client, "start_observation"):
        self.api = "v3"
    elif hasattr(self._client, "trace"):
        self.api = "v2"
    else:
        raise ConfigurationError(
            "The installed langfuse version exposes neither "
            "start_observation() (v3/v4) nor trace() (v2).",
            hint="Install a supported release: pip install 'langfuse>=2.50'",
            context={"version": getattr(langfuse, "__version__", "unknown")},
        )
    self._handles: dict[str, Any] = {}

native

native() -> Any

Return the underlying Langfuse client (Level 3 access).

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

def native(self) -> Any:
    """Return the underlying ``Langfuse`` client (Level 3 access)."""
    return self._client

start_span

start_span(span: Span) -> None

Open a Langfuse observation, nested under its parent when there is one.

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

def start_span(self, span: Span) -> None:
    """Open a Langfuse observation, nested under its parent when there is one."""
    try:
        parent = self._handles.get(span.parent_id or "")
        handle = (
            self._start_v3(span, parent) if self.api == "v3" else self._start_v2(span, parent)
        )
        if handle is not None:
            self._handles[span.id] = handle
            span.attach_native(handle)
    except Exception as exc:
        self._log.debug("Langfuse span creation failed: %s", exc)

end_span

end_span(span: Span) -> None

Record the outcome and close the Langfuse handle.

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

def end_span(self, span: Span) -> None:
    """Record the outcome and close the Langfuse handle."""
    handle = self._handles.pop(span.id, None)
    if handle is None:
        return
    try:
        payload: dict[str, Any] = {
            "output": span.outputs,
            "metadata": {**span.metadata, "duration_ms": round(span.duration_ms, 2)},
        }
        if span.error:
            payload["level"] = "ERROR"
            payload["status_message"] = span.error
        if span.usage:
            usage = {
                "input": span.usage.prompt_tokens,
                "output": span.usage.completion_tokens,
                "total": span.usage.total_tokens,
            }
            # v3 renamed the field and made it observation-level.
            payload["usage_details" if self.api == "v3" else "usage"] = usage

        if self.api == "v3":
            handle.update(**payload)
            handle.end()
        else:
            updater = getattr(handle, "end", None) or getattr(handle, "update", None)
            if callable(updater):
                updater(**payload)
    except Exception as exc:
        self._log.debug("Langfuse span close failed: %s", exc)

flush

flush() -> None

Send any buffered events, giving up after :attr:flush_timeout.

Langfuse's own flush joins an internal queue and can block forever when a worker thread has already stopped with items outstanding.

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

def flush(self) -> None:
    """Send any buffered events, giving up after :attr:`flush_timeout`.

    Langfuse's own ``flush`` joins an internal queue and can block forever
    when a worker thread has already stopped with items outstanding.
    """
    _bounded_flush(
        self._client.flush,
        label="Langfuse",
        timeout=self.flush_timeout,
        log=self._log,
    )

aclose async

aclose() -> None

Flush and shut the client down, both under a deadline.

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

async def aclose(self) -> None:
    """Flush and shut the client down, both under a deadline."""
    self.flush()
    shutdown = getattr(self._client, "shutdown", None)
    if callable(shutdown):
        _bounded_flush(
            shutdown, label="Langfuse shutdown", timeout=self.flush_timeout, log=self._log
        )