LangSmithExporter

Overview

Sends Fennec spans to LangSmith as runs. The langsmith package is an optional dependency — it's only imported the first time a LangSmithExporter is constructed, so nothing else in zerda.observability requires it to be installed.

Not re-exported from observability/__init__.py's top level; use enable_langsmith() for the zero-import-hassle path, or import it explicitly from observability.exporters (or observability.exporters.langsmith) if you need the class itself (e.g. to isinstance-check it or subclass it).

Field mapping

From the module docstring, reproduced here since it documents intended behavior directly:

Constructor

def __init__(
    self,
    api_key: Optional[str] = None,
    project_name: str = "default",
    **client_kwargs: Any,
) -> None

Raises: ImportError if the langsmith package is not installed, with the message:

Behavior detail: the source constructs the client as Client(api_key=api_key, **client_kwargs) only if api_key is truthy, and as Client(**client_kwargs) (no api_key kwarg at all) otherwise — so when api_key is None, it is not passed to Client(...) as api_key=None; it's omitted entirely, which is what allows the client's own env-var fallback to take effect.

export

def export(self, span: Span) -> None

Everything in this method is wrapped in a try/except Exception, logged via logger.exception(...) on failure and otherwise swallowed — the docstring/comment in the source is explicit that this is intentional ("never break the app"). This is in addition to, not instead of, the isolation Tracer._drain_loop already provides — belt and suspenders.

Steps performed:

1. Look up run_type via _RUN_TYPE_MAP, defaulting to "chain" for any span.type not in the map.
2. Build extra starting from a copy of span.metadata, then add extra["model"] / extra["provider"] (only if set on the span), and extra["token_usage"] (via TokenUsage.to_dict(), only if span.token_usage is set).
3. Call self._client.create_run(...) with:
   • inputs: {"input": span.input} if span.input isn't already a dict, else span.input itself (passed through unchanged).
   • outputs: {"output": span.output} if span.output is not None and not already a dict; span.output itself if it's a dict; omitted as None if span.output is None (the ternary's else branch also fires when output is None, producing outputs=None).
   • start_time / end_time: converted from epoch seconds to timezone-aware UTC datetime objects via the private _to_datetime() helper (end_time only if span.end_time is set).
   • error: span.error passed through as-is (a string or None) — note span.stack_trace is not separately forwarded despite being listed in the field-mapping table above; only error is passed to create_run.
   • extra: {"metadata": extra} — i.e. the whole built-up extra dict from step 2 is nested one level under a "metadata" key.
   • trace_id, parent_run_id (from span.parent_id), project_name (from the constructor argument), and id (from span.id).

shutdown

def shutdown(self) -> None

If the underlying langsmith.Client exposes a callable flush attribute, calls it — wrapped in its own try/except, logged on failure. If the client has no flush method, this is a no-op.

_RUN_TYPE_MAP (internal)

Module-level constant mapping every built-in SpanType value to one of the run types LangSmith's create_run accepts:

{
    "chain": "chain", "llm": "llm", "tool": "tool",
    "tool_call": "tool", "tool_result": "tool",
    "retriever": "retriever", "embedding": "embedding",
    "vector_search": "retriever", "context": "chain",
    "rerank": "chain", "agent": "chain", "planning": "chain",
    "router": "chain", "response": "chain", "generic": "chain",
}

Any span.type string not present in this dict (e.g. a custom type you invented) is mapped to "chain" by the _RUN_TYPE_MAP.get(span.type, "chain") call in export().

_to_datetime (internal)

def _to_datetime(epoch_seconds: float): ...

Converts epoch seconds to a timezone-aware UTC datetime via datetime.fromtimestamp(epoch_seconds, tz=timezone.utc). Imports datetime locally inside the function rather than at module scope.

Example

from zerda.observability import enable_langsmith, get_tracer

enable_langsmith(api_key="ls__...", project_name="Customer Support")

with get_tracer().span("Retriever", type="retriever") as span:
    span.set_input("what's the refund policy?")
    docs = retrieve(...)
    span.set_output(docs)

If langsmith is not installed:

>>> enable_langsmith()
ImportError: LangSmithExporter requires the optional 'langsmith' package. Install it with: pip install langsmith

Related: BaseExporter · Span · enable_langsmith · Installation → Optional dependency