teff.observability

teff.observability

Graph-run observability: full traces (topology, node spans, LLM payloads).

Usage::

from teff.observability import (
    GraphObserver,
    JsonlExporter,
    SQLiteExporter,
    topology_from_graph,
)

observer = GraphObserver(
    "my-flow",
    exporter=SQLiteExporter("./traces.db"),
    topology=topology_from_graph(graph),
)
await graph.run(state, tracer=observer.tracer,
                on_llm_payload=observer.on_llm_payload)
observer.export()

Modules:

Name	Description
api	FastAPI router exposing stored traces as a dashboard API.
builder	Build a :class:GraphObserver from a workflow's observability: block.
collector	Collect a full graph-run trace into a single :class:Run.
exporter	Exporters for :class:~teff.observability.model.Run traces.
model	Observability data model for a graph run.
push	Remote trace exporters: push a completed :class:Run to an HTTP endpoint.
server	Standalone trace server: ingest + dashboard (teff obs-server).
topology	Graph topology snapshot — the node/edge shape for the dashboard.

Classes:

Name	Description
CompositeExporter	Fan one run out to several exporters (SQLite + remote sinks).
GraphObserver	Assemble a :class:Run from graph events and forward it to an exporter.
GraphTopology	A node/edge snapshot of the compiled graph (for visualisation).
HttpExporter	POST each completed run as JSON to a remote ingest endpoint.
JsonlExporter	Append each run as one JSON line to a newline-delimited file.
LLMCall	One model call with the full request/response payload.
LangfuseExporter	Push runs to a Langfuse instance via the public traces API.
LangsmithExporter	Push runs to LangSmith via the /runs/batch endpoint.
NodeSpan	One node execution: timing, outcome, its LLM calls and tool calls.
Run	A single executed run, ready to export or serve over the API.
SQLiteExporter	Store runs in SQLite tables runs / nodes / llm_calls.
SpanEvent	One step of a node's execution, in chronological order.
ToolCall	One tool invocation: what the model requested and what ran.
TraceExporter	Persist a completed :class:Run to some backend.

Functions:

Name	Description
build_observability	Assemble a :class:GraphObserver from an observability: block.
build_observer_factory	Like :func:build_observability but for repeated runs.
build_remote_exporter	Construct a remote exporter from one observability.export entry.
topology_from_graph	Capture {nodes, edges} from a compiled :class:~teff.graph.Graph.

CompositeExporter

Bases: TraceExporter

Fan one run out to several exporters (SQLite + remote sinks).

A failure in one sink is logged and swallowed so the remaining sinks (and the workflow that produced the run) keep working.

Source code in teff/observability/exporter.py

class CompositeExporter(TraceExporter):
    """Fan one run out to several exporters (SQLite + remote sinks).

    A failure in one sink is logged and swallowed so the remaining sinks
    (and the workflow that produced the run) keep working.
    """

    def __init__(self, exporters: list[TraceExporter]):
        self.exporters = list(exporters)

    def export(self, run: Run) -> str | None:
        run_id: str | None = None
        for exporter in self.exporters:
            try:
                result = exporter.export(run)
            except Exception:
                logger.exception(
                    "composite exporter %s failed", type(exporter).__name__
                )
                continue
            if run_id is None and result is not None:
                run_id = result
        return run_id

    def close(self) -> None:
        for exporter in self.exporters:
            try:
                exporter.close()
            except Exception:
                logger.exception(
                    "composite exporter %s failed to close", type(exporter).__name__
                )

GraphObserver

Assemble a :class:Run from graph events and forward it to an exporter.

Methods:

Name	Description
export	Persist the collected :class:Run and return its backend run id.
on_llm_payload	Sink for the run's on_llm_payload channel.

Source code in teff/observability/collector.py

class GraphObserver:
    """Assemble a :class:`Run` from graph events and forward it to an exporter."""

    def __init__(
        self,
        name: str,
        *,
        exporter: TraceExporter | None = None,
        topology: GraphTopology | None = None,
        owner: str | None = None,
        checkpoint_id: str | None = None,
        redact: bool = True,
        redact_fn: Callable[[Any], Any] = _default_redact,
    ):
        self.name = name
        self.exporter = exporter
        self.topology = topology or GraphTopology()
        self.owner = owner
        self.checkpoint_id = checkpoint_id
        self._redact_fn = redact_fn if redact else (lambda value: value)

        self.tracer = RunTracer()
        self._start = time.monotonic()
        self._wall_start = time.time()
        self._spans: dict[str, NodeSpan] = {}
        self._active: list[NodeSpan] = []
        self._tool_seen: dict[str, dict[str, ToolCall]] = {}
        self._status = "ok"
        self._error: str | None = None
        self._total_ms = 0.0
        self._wire_tracer()

    async def on_llm_payload(
        self,
        provider: str,
        model: str,
        messages: list[dict[str, Any]],
        response: str,
        usage: dict[str, Any],
        latency_ms: float,
        cached: bool,
    ) -> None:
        """Sink for the run's ``on_llm_payload`` channel."""
        redact = self._redact_fn
        call = LLMCall(
            node_id=self._active[-1].node_id if self._active else None,
            provider=provider,
            model=model,
            messages=redact(messages),
            response=str(redact(response) or ""),
            prompt_tokens=int(usage.get("prompt", 0) or 0),
            completion_tokens=int(usage.get("completion", 0) or 0),
            latency_ms=latency_ms,
            cached=cached,
        )
        if self._active:
            span = self._active[-1]
            # Tool calls are discovered in the *next* call's request payload
            # (they ran after the previous reply), so capture them first to
            # keep the event list in real chronological order.
            self._capture_tool_calls(span, messages)
            span.llm_calls.append(call)
            span.events.append(SpanEvent(kind="llm", index=len(span.llm_calls) - 1))

    def _capture_tool_calls(
        self, span: NodeSpan, messages: list[dict[str, Any]]
    ) -> None:
        """Extract tool calls from an LLM payload into *span*.

        An assistant ``tool_calls`` block and its matching ``role: tool``
        result can arrive in *different* payloads (the result is appended
        before the next model call), so already-seen calls are backfilled
        with their result instead of duplicated.
        """
        results: dict[str, str] = {}
        for msg in messages:
            if msg.get("role") == "tool":
                results[str(msg.get("tool_call_id") or "")] = str(
                    msg.get("content") or ""
                )

        seen = self._tool_seen.setdefault(span.node_id, {})
        for msg in messages:
            if msg.get("role") != "assistant":
                continue
            for tc in msg.get("tool_calls") or []:
                name, raw_args, call_id = _tool_call_parts(tc)
                result = results.get(call_id)

                existing = seen.get(call_id)
                if existing is not None:
                    if result is not None and not existing.result:
                        existing.result = self._redact_fn(result)
                        existing.ok = not _is_tool_error(result)
                    continue

                call = ToolCall(
                    name=name,
                    args=self._redact_fn(raw_args),
                    result=self._redact_fn(result) if result else "",
                    ok=not (result and _is_tool_error(result)),
                )
                seen[call_id] = call
                span.tool_calls.append(call)
                span.events.append(
                    SpanEvent(kind="tool", index=len(span.tool_calls) - 1)
                )

    def _start_node(self, node_id: str, node_type: str) -> None:
        span = self._spans.get(node_id)
        if span is None:
            span = NodeSpan(
                node_id=node_id,
                node_type=node_type,
                start_ms=(time.monotonic() - self._start) * 1000.0,
            )
            self._spans[node_id] = span
        # A node can be visited many times in one run (react loops, retries);
        # reuse the span so its LLM calls, tool calls and events accumulate
        # in chronological order instead of keeping only the last visit.
        self._active.append(span)

    def _end_node(
        self, node_id: str, status: str = "ok", error: str | None = None
    ) -> None:
        span = self._spans.get(node_id)
        if span is None:
            return
        span.end_ms = (time.monotonic() - self._start) * 1000.0
        span.status = status
        span.error = error
        if self._active and self._active[-1] is span:
            self._active.pop()

    def _wire_tracer(self) -> None:
        tracer = self.tracer
        node_start = tracer.node_start
        node_end = tracer.node_end
        node_error = tracer.node_error
        run_end = tracer.run_end

        def _node_start(node_id, node_type):
            node_start(node_id, node_type)
            self._start_node(node_id, node_type)

        def _node_end(node_id, node_type, duration_ms):
            node_end(node_id, node_type, duration_ms)
            self._end_node(node_id)

        def _node_error(node_id, node_type, duration_ms, error):
            node_error(node_id, node_type, duration_ms, error)
            self._end_node(node_id, status="error", error=str(error))

        def _run_end(status, total_ms, error=None):
            run_end(status, total_ms, error)
            self._status = status
            self._total_ms = total_ms
            if error is not None:
                self._error = str(error)
            for span in list(self._active):
                self._end_node(span.node_id)

        tracer.node_start = _node_start  # type: ignore[method-assign]
        tracer.node_end = _node_end  # type: ignore[method-assign]
        tracer.node_error = _node_error  # type: ignore[method-assign]
        tracer.run_end = _run_end  # type: ignore[method-assign]

    def build(self) -> Run:
        return Run(
            name=self.name,
            status=self._status,
            total_ms=self._total_ms,
            owner=self.owner,
            checkpoint_id=self.checkpoint_id,
            created_at=self._wall_start,
            topology=self.topology,
            nodes=list(self._spans.values()),
        )

    def export(self) -> str | None:
        """Persist the collected :class:`Run` and return its backend run id.

        Returns ``None`` when no exporter is attached (tracing disabled) or
        the backend does not expose ids.
        """
        if self.exporter is None:
            return None
        return self.exporter.export(self.build())

    def close(self) -> None:
        if self.exporter is not None:
            self.exporter.close()

export

export()

Persist the collected :class:Run and return its backend run id.

Returns None when no exporter is attached (tracing disabled) or the backend does not expose ids.

Source code in teff/observability/collector.py

def export(self) -> str | None:
    """Persist the collected :class:`Run` and return its backend run id.

    Returns ``None`` when no exporter is attached (tracing disabled) or
    the backend does not expose ids.
    """
    if self.exporter is None:
        return None
    return self.exporter.export(self.build())

on_llm_payload async

on_llm_payload(provider, model, messages, response, usage, latency_ms, cached)

Sink for the run's on_llm_payload channel.

Source code in teff/observability/collector.py

async def on_llm_payload(
    self,
    provider: str,
    model: str,
    messages: list[dict[str, Any]],
    response: str,
    usage: dict[str, Any],
    latency_ms: float,
    cached: bool,
) -> None:
    """Sink for the run's ``on_llm_payload`` channel."""
    redact = self._redact_fn
    call = LLMCall(
        node_id=self._active[-1].node_id if self._active else None,
        provider=provider,
        model=model,
        messages=redact(messages),
        response=str(redact(response) or ""),
        prompt_tokens=int(usage.get("prompt", 0) or 0),
        completion_tokens=int(usage.get("completion", 0) or 0),
        latency_ms=latency_ms,
        cached=cached,
    )
    if self._active:
        span = self._active[-1]
        # Tool calls are discovered in the *next* call's request payload
        # (they ran after the previous reply), so capture them first to
        # keep the event list in real chronological order.
        self._capture_tool_calls(span, messages)
        span.llm_calls.append(call)
        span.events.append(SpanEvent(kind="llm", index=len(span.llm_calls) - 1))

GraphTopology dataclass

A node/edge snapshot of the compiled graph (for visualisation).

Source code in teff/observability/model.py

@dataclass
class GraphTopology:
    """A node/edge snapshot of the compiled graph (for visualisation)."""

    nodes: list[dict[str, Any]] = field(default_factory=list)
    edges: list[dict[str, Any]] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        return {"nodes": self.nodes, "edges": self.edges}

    @staticmethod
    def from_dict(data: dict[str, Any]) -> "GraphTopology":
        return GraphTopology(
            nodes=list(data.get("nodes") or []),
            edges=list(data.get("edges") or []),
        )

HttpExporter

Bases: TraceExporter

POST each completed run as JSON to a remote ingest endpoint.

The body is :meth:Run.to_dict plus created_at (seconds since the epoch) so the receiving side can order runs. Sends are asynchronous: :meth:export returns immediately, :meth:close drains the queue.

Source code in teff/observability/push.py

class HttpExporter(TraceExporter):
    """POST each completed run as JSON to a remote ingest endpoint.

    The body is :meth:`Run.to_dict` plus ``created_at`` (seconds since the
    epoch) so the receiving side can order runs.  Sends are asynchronous:
    :meth:`export` returns immediately, :meth:`close` drains the queue.
    """

    def __init__(
        self,
        url: str,
        *,
        headers: dict[str, str] | None = None,
        timeout: float = 10.0,
        retries: int = 3,
        backoff: float = 1.0,
    ):
        self.url = url
        self.headers = dict(headers or {})
        self.timeout = timeout
        self.retries = retries
        self.backoff = backoff
        self._pool = ThreadPoolExecutor(max_workers=1)

    def export(self, run: Run) -> None:
        payload = run.to_dict()
        self._pool.submit(
            _post_json,
            self.url,
            payload,
            headers=self.headers,
            timeout=self.timeout,
            retries=self.retries,
            backoff=self.backoff,
        )

    def close(self) -> None:
        self._pool.shutdown(wait=True)

JsonlExporter

Bases: TraceExporter

Append each run as one JSON line to a newline-delimited file.

Source code in teff/observability/exporter.py

class JsonlExporter(TraceExporter):
    """Append each run as one JSON line to a newline-delimited file."""

    def __init__(self, path: str | Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._handle = self.path.open("a", encoding="utf-8")

    def export(self, run: Run) -> None:
        self._handle.write(json.dumps(run.to_dict(), ensure_ascii=False) + "\n")
        self._handle.flush()

    def close(self) -> None:
        self._handle.close()

LLMCall dataclass

One model call with the full request/response payload.

Source code in teff/observability/model.py

@dataclass
class LLMCall:
    """One model call with the full request/response payload."""

    node_id: str | None
    provider: str
    model: str
    messages: list[dict[str, Any]]
    response: str
    prompt_tokens: int
    completion_tokens: int
    latency_ms: float
    cached: bool = False

    def to_dict(self) -> dict[str, Any]:
        return {
            "node_id": self.node_id,
            "provider": self.provider,
            "model": self.model,
            "messages": self.messages,
            "response": self.response,
            "prompt_tokens": self.prompt_tokens,
            "completion_tokens": self.completion_tokens,
            "latency_ms": round(self.latency_ms, 3),
            "cached": self.cached,
        }

    @staticmethod
    def from_dict(data: dict[str, Any]) -> "LLMCall":
        return LLMCall(
            node_id=data.get("node_id"),
            provider=str(data["provider"]),
            model=str(data["model"]),
            messages=list(data.get("messages") or []),
            response=str(data.get("response") or ""),
            prompt_tokens=int(data.get("prompt_tokens") or 0),
            completion_tokens=int(data.get("completion_tokens") or 0),
            latency_ms=float(data.get("latency_ms") or 0.0),
            cached=bool(data.get("cached")),
        )

LangfuseExporter

Bases: TraceExporter

Push runs to a Langfuse instance via the public traces API.

Requires host plus a public_key/secret_key pair (Basic auth). Each node becomes a span observation and every LLM call a generation observation attached to its node.

Source code in teff/observability/push.py

class LangfuseExporter(TraceExporter):
    """Push runs to a Langfuse instance via the public traces API.

    Requires ``host`` plus a ``public_key``/``secret_key`` pair (Basic
    auth).  Each node becomes a ``span`` observation and every LLM call a
    ``generation`` observation attached to its node.
    """

    def __init__(
        self,
        host: str,
        public_key: str,
        secret_key: str,
        *,
        timeout: float = 10.0,
        retries: int = 3,
        backoff: float = 1.0,
    ):
        self.url = host.rstrip("/") + "/api/public/traces"
        token = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
        self.headers = {"Authorization": f"Basic {token}"}
        self.timeout = timeout
        self.retries = retries
        self.backoff = backoff
        self._pool = ThreadPoolExecutor(max_workers=1)

    def export(self, run: Run) -> None:
        base = _run_base(run)
        observations: list[dict[str, Any]] = []
        for node in run.nodes:
            observations.append(
                {
                    "id": node.node_id,
                    "type": "span",
                    "name": node.node_id,
                    "startTime": _iso_timestamp(base, node.start_ms),
                    "endTime": _iso_timestamp(base, node.end_ms or node.start_ms),
                    "level": "ERROR" if node.status == "error" else "DEFAULT",
                    "metadata": {"node_type": node.node_type, "error": node.error},
                }
            )
            for i, call in enumerate(node.llm_calls):
                observations.append(
                    {
                        "id": f"{node.node_id}:llm:{i}",
                        "type": "generation",
                        "parentObservationId": node.node_id,
                        "name": f"{node.node_id}.llm",
                        "model": call.model,
                        "input": call.messages,
                        "output": call.response,
                        "usage": {
                            "input": call.prompt_tokens,
                            "output": call.completion_tokens,
                        },
                        "startTime": _iso_timestamp(base, node.start_ms),
                        "endTime": _iso_timestamp(base, node.end_ms or node.start_ms),
                        "metadata": {"provider": call.provider, "cached": call.cached},
                    }
                )
        payload = {
            "name": run.name,
            "timestamp": _iso_timestamp(base, 0.0),
            "userId": run.owner,
            "sessionId": run.checkpoint_id,
            "metadata": {
                "status": run.status,
                "total_ms": round(run.total_ms, 3),
                "tags": run.tags,
                "notes": run.notes,
                "topology": run.topology.to_dict(),
            },
            "observations": observations,
        }
        self._pool.submit(
            _post_json,
            self.url,
            payload,
            headers=self.headers,
            timeout=self.timeout,
            retries=self.retries,
            backoff=self.backoff,
        )

    def close(self) -> None:
        self._pool.shutdown(wait=True)

LangsmithExporter

Bases: TraceExporter

Push runs to LangSmith via the /runs/batch endpoint.

Requires an API key (x-api-key header). The run becomes a chain run, each node a child chain run, and each LLM call an llm run. project is passed as extra metadata so the UI can group by project.

Source code in teff/observability/push.py

class LangsmithExporter(TraceExporter):
    """Push runs to LangSmith via the ``/runs/batch`` endpoint.

    Requires an API key (``x-api-key`` header).  The run becomes a ``chain``
    run, each node a child ``chain`` run, and each LLM call an ``llm`` run.
    ``project`` is passed as extra metadata so the UI can group by project.
    """

    def __init__(
        self,
        api_url: str,
        api_key: str,
        *,
        project: str | None = None,
        timeout: float = 10.0,
        retries: int = 3,
        backoff: float = 1.0,
    ):
        self.url = api_url.rstrip("/") + "/runs/batch"
        headers = {"x-api-key": api_key}
        if project:
            headers["x-langchain-project"] = project
        self.headers = headers
        self.timeout = timeout
        self.retries = retries
        self.backoff = backoff
        self._pool = ThreadPoolExecutor(max_workers=1)

    def export(self, run: Run) -> None:
        base = _run_base(run)
        run_id = f"teff-{run.name}-{int(base * 1000)}"
        metadata = {
            "teff": True,
            "status": run.status,
            "owner": run.owner,
            "checkpoint_id": run.checkpoint_id,
            "tags": run.tags,
            "notes": run.notes,
            "topology": run.topology.to_dict(),
        }
        runs: list[dict[str, Any]] = [
            {
                "id": run_id,
                "name": run.name,
                "run_type": "chain",
                "inputs": {},
                "outputs": {},
                "start_time": _iso_timestamp(base, 0.0),
                "end_time": _iso_timestamp(base, run.total_ms),
                "extra": {"metadata": metadata},
                "error": None if run.status != "error" else run.notes or "error",
            }
        ]
        for node in run.nodes:
            node_id = f"{run_id}:{node.node_id}"
            runs.append(
                {
                    "id": node_id,
                    "name": node.node_id,
                    "run_type": "chain",
                    "parent_run_id": run_id,
                    "inputs": {},
                    "outputs": {},
                    "start_time": _iso_timestamp(base, node.start_ms),
                    "end_time": _iso_timestamp(base, node.end_ms or node.start_ms),
                    "extra": {
                        "metadata": {
                            "node_type": node.node_type,
                            "error": node.error,
                        }
                    },
                    "error": None if node.status != "error" else node.error,
                }
            )
            for i, call in enumerate(node.llm_calls):
                runs.append(
                    {
                        "id": f"{node_id}:llm:{i}",
                        "name": f"{node.node_id}.llm",
                        "run_type": "llm",
                        "parent_run_id": node_id,
                        "inputs": {"messages": call.messages},
                        "outputs": {"response": call.response},
                        "start_time": _iso_timestamp(base, node.start_ms),
                        "end_time": _iso_timestamp(base, node.end_ms or node.start_ms),
                        "extra": {
                            "metadata": {
                                "provider": call.provider,
                                "model": call.model,
                                "prompt_tokens": call.prompt_tokens,
                                "completion_tokens": call.completion_tokens,
                                "cached": call.cached,
                            }
                        },
                    }
                )
        self._pool.submit(
            _post_json,
            self.url,
            runs,
            headers=self.headers,
            timeout=self.timeout,
            retries=self.retries,
            backoff=self.backoff,
        )

    def close(self) -> None:
        self._pool.shutdown(wait=True)

NodeSpan dataclass

One node execution: timing, outcome, its LLM calls and tool calls.

Source code in teff/observability/model.py

@dataclass
class NodeSpan:
    """One node execution: timing, outcome, its LLM calls and tool calls."""

    node_id: str
    node_type: str
    start_ms: float
    end_ms: float | None = None
    status: str = "ok"
    error: str | None = None
    llm_calls: list[LLMCall] = field(default_factory=list)
    tool_calls: list[ToolCall] = field(default_factory=list)
    events: list[SpanEvent] = field(default_factory=list)

    @property
    def duration_ms(self) -> float:
        return (self.end_ms or self.start_ms) - self.start_ms

    def to_dict(self) -> dict[str, Any]:
        return {
            "node_id": self.node_id,
            "node_type": self.node_type,
            "start_ms": round(self.start_ms, 3),
            "end_ms": None if self.end_ms is None else round(self.end_ms, 3),
            "duration_ms": round(self.duration_ms, 3),
            "status": self.status,
            "error": self.error,
            "llm_calls": [call.to_dict() for call in self.llm_calls],
            "tool_calls": [call.to_dict() for call in self.tool_calls],
            "events": [event.to_dict() for event in self.events],
        }

    @staticmethod
    def from_dict(data: dict[str, Any]) -> "NodeSpan":
        end = data.get("end_ms")
        return NodeSpan(
            node_id=str(data["node_id"]),
            node_type=str(data.get("node_type") or ""),
            start_ms=float(data.get("start_ms") or 0.0),
            end_ms=None if end is None else float(end),
            status=str(data.get("status") or "ok"),
            error=data.get("error"),
            llm_calls=[
                LLMCall.from_dict(call) for call in (data.get("llm_calls") or [])
            ],
            tool_calls=[
                ToolCall.from_dict(call) for call in (data.get("tool_calls") or [])
            ],
            events=[SpanEvent.from_dict(event) for event in (data.get("events") or [])],
        )

Run dataclass

A single executed run, ready to export or serve over the API.

Source code in teff/observability/model.py

@dataclass
class Run:
    """A single executed run, ready to export or serve over the API."""

    name: str
    status: str
    total_ms: float
    run_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    owner: str | None = None
    checkpoint_id: str | None = None
    tags: list[str] = field(default_factory=list)
    notes: str = ""
    created_at: float | None = None
    topology: GraphTopology = field(default_factory=GraphTopology)
    nodes: list[NodeSpan] = field(default_factory=list)

    @property
    def llm_calls(self) -> list[LLMCall]:
        calls: list[LLMCall] = []
        for node in self.nodes:
            calls.extend(node.llm_calls)
        return calls

    @property
    def prompt_tokens(self) -> int:
        return sum(c.prompt_tokens for c in self.llm_calls)

    @property
    def completion_tokens(self) -> int:
        return sum(c.completion_tokens for c in self.llm_calls)

    def to_dict(self) -> dict[str, Any]:
        return {
            "run_id": self.run_id,
            "name": self.name,
            "status": self.status,
            "total_ms": round(self.total_ms, 3),
            "owner": self.owner,
            "checkpoint_id": self.checkpoint_id,
            "tags": self.tags,
            "notes": self.notes,
            "created_at": self.created_at,
            "topology": self.topology.to_dict(),
            "nodes": [node.to_dict() for node in self.nodes],
            "llm_calls": [call.to_dict() for call in self.llm_calls],
            "prompt_tokens": self.prompt_tokens,
            "completion_tokens": self.completion_tokens,
        }

    @staticmethod
    def from_dict(data: dict[str, Any]) -> "Run":
        created_at = data.get("created_at")
        return Run(
            name=str(data["name"]),
            status=str(data.get("status") or "ok"),
            total_ms=float(data.get("total_ms") or 0.0),
            run_id=str(data.get("run_id") or uuid.uuid4()),
            owner=data.get("owner"),
            checkpoint_id=data.get("checkpoint_id"),
            tags=list(data.get("tags") or []),
            notes=str(data.get("notes") or ""),
            created_at=None if created_at is None else float(created_at),
            topology=GraphTopology.from_dict(data.get("topology") or {}),
            nodes=[NodeSpan.from_dict(node) for node in (data.get("nodes") or [])],
        )

SQLiteExporter

Bases: TraceExporter

Store runs in SQLite tables runs / nodes / llm_calls.

The runs table also carries the graph topology as JSON, and the metadata needed by a dashboard (owner, checkpoint id, status, tokens).

Methods:

Name	Description
get_run	Full run payload: metadata, topology, node spans, LLM calls.
list_runs	Dashboard rows (no payloads) with filtering and pagination.
update_run	Patch a run's tags / notes. Returns False if the run doesn't exist.

Source code in teff/observability/exporter.py

class SQLiteExporter(TraceExporter):
    """Store runs in SQLite tables ``runs`` / ``nodes`` / ``llm_calls``.

    The ``runs`` table also carries the graph topology as JSON, and the
    metadata needed by a dashboard (owner, checkpoint id, status, tokens).
    """

    def __init__(self, path: str | Path = "./traces.db"):
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        self.path = str(path)
        self._conn = sqlite3.connect(self.path, check_same_thread=False)
        self._create_schema()
        self._migrate()

    def _create_schema(self) -> None:
        self._conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS runs (
                run_id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                status TEXT NOT NULL,
                total_ms REAL NOT NULL,
                owner TEXT,
                checkpoint_id TEXT,
                prompt_tokens INTEGER NOT NULL DEFAULT 0,
                completion_tokens INTEGER NOT NULL DEFAULT 0,
                topology TEXT NOT NULL,
                tags TEXT NOT NULL DEFAULT '[]',
                notes TEXT NOT NULL DEFAULT '',
                created_at REAL NOT NULL
            );
            CREATE TABLE IF NOT EXISTS nodes (
                run_id TEXT NOT NULL REFERENCES runs(run_id),
                node_id TEXT NOT NULL,
                node_type TEXT NOT NULL,
                start_ms REAL NOT NULL,
                end_ms REAL,
                status TEXT NOT NULL,
                error TEXT,
                tool_calls TEXT NOT NULL DEFAULT '[]',
                events TEXT NOT NULL DEFAULT '[]',
                UNIQUE(run_id, node_id)
            );
            CREATE TABLE IF NOT EXISTS llm_calls (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                run_id TEXT NOT NULL REFERENCES runs(run_id),
                node_id TEXT,
                provider TEXT NOT NULL,
                model TEXT NOT NULL,
                messages TEXT NOT NULL,
                response TEXT NOT NULL,
                prompt_tokens INTEGER NOT NULL,
                completion_tokens INTEGER NOT NULL,
                latency_ms REAL NOT NULL,
                cached INTEGER NOT NULL DEFAULT 0
            );
            """
        )
        self._conn.commit()

    def _migrate(self) -> None:
        """Upgrade databases created by older versions.

        Runs from the pre-uuid era used an ``INTEGER PRIMARY KEY
        AUTOINCREMENT``; the id column is now a uuid text key, which cannot
        be altered in place, so the trace tables are rebuilt empty (a
        ``runs`` table of the wrong type means the whole layout is old).
        """
        run_cols = {
            row[1]: row[2]
            for row in self._conn.execute("PRAGMA table_info(runs)").fetchall()
        }
        if run_cols.get("run_id", "").upper() != "TEXT":
            self._conn.executescript("DROP TABLE IF EXISTS llm_calls;")
            self._conn.executescript("DROP TABLE IF EXISTS nodes;")
            self._conn.executescript("DROP TABLE IF EXISTS runs;")
            self._create_schema()
            run_cols = {
                row[1]: row[2]
                for row in self._conn.execute("PRAGMA table_info(runs)").fetchall()
            }
        if "tags" not in run_cols:
            self._conn.execute(
                "ALTER TABLE runs ADD COLUMN tags TEXT NOT NULL DEFAULT '[]'"
            )
        if "notes" not in run_cols:
            self._conn.execute(
                "ALTER TABLE runs ADD COLUMN notes TEXT NOT NULL DEFAULT ''"
            )
        node_cols = {
            row[1] for row in self._conn.execute("PRAGMA table_info(nodes)").fetchall()
        }
        if "tool_calls" not in node_cols:
            self._conn.execute(
                "ALTER TABLE nodes ADD COLUMN tool_calls TEXT NOT NULL DEFAULT '[]'"
            )
        if "events" not in node_cols:
            self._conn.execute(
                "ALTER TABLE nodes ADD COLUMN events TEXT NOT NULL DEFAULT '[]'"
            )
        self._conn.commit()

    def export(self, run: Run) -> str:
        self._conn.execute(
            "INSERT INTO runs (run_id, name, status, total_ms, owner, checkpoint_id, "
            "prompt_tokens, completion_tokens, topology, tags, notes, created_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (
                run.run_id,
                run.name,
                run.status,
                run.total_ms,
                run.owner,
                run.checkpoint_id,
                run.prompt_tokens,
                run.completion_tokens,
                json.dumps(run.topology.to_dict()),
                json.dumps(list(run.tags)),
                run.notes,
                run.created_at or time.time(),
            ),
        )
        run_id = run.run_id
        for node in run.nodes:
            self._conn.execute(
                "INSERT INTO nodes (run_id, node_id, node_type, start_ms, end_ms, "
                "status, error, tool_calls, events) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
                (
                    run_id,
                    node.node_id,
                    node.node_type,
                    node.start_ms,
                    node.end_ms,
                    node.status,
                    node.error,
                    json.dumps([t.to_dict() for t in node.tool_calls]),
                    json.dumps([e.to_dict() for e in node.events]),
                ),
            )
        for call in run.llm_calls:
            self._conn.execute(
                "INSERT INTO llm_calls (run_id, node_id, provider, model, messages, "
                "response, prompt_tokens, completion_tokens, latency_ms, cached) "
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                (
                    run_id,
                    call.node_id,
                    call.provider,
                    call.model,
                    json.dumps(call.messages),
                    call.response,
                    call.prompt_tokens,
                    call.completion_tokens,
                    call.latency_ms,
                    1 if call.cached else 0,
                ),
            )
        self._conn.commit()
        return run_id

    def list_runs(
        self,
        limit: int = 100,
        offset: int = 0,
        status: str | None = None,
        name: str | None = None,
        owner: str | None = None,
        tag: str | None = None,
    ) -> dict[str, Any]:
        """Dashboard rows (no payloads) with filtering and pagination.

        Returns ``{"items": [...], "total": n}`` where *total* is the count
        before pagination, so the UI can render a page size / total.
        *name* / *owner* are case-insensitive substrings, *tag* matches an
        exact tag, *status* matches the run status exactly.
        """
        where: list[str] = []
        args: list[Any] = []
        if status:
            where.append("status = ?")
            args.append(status)
        if name:
            where.append("LOWER(name) LIKE ?")
            args.append(f"%{name.lower()}%")
        if owner:
            where.append("LOWER(COALESCE(owner, '')) LIKE ?")
            args.append(f"%{owner.lower()}%")
        if tag:
            where.append(
                "EXISTS (SELECT 1 FROM json_each(tags) WHERE json_each.value = ?)"
            )
            args.append(tag)
        where_sql = (" WHERE " + " AND ".join(where)) if where else ""

        total = self._conn.execute(
            f"SELECT COUNT(*) FROM runs{where_sql}", tuple(args)
        ).fetchone()[0]

        rows = self._conn.execute(
            f"SELECT run_id, name, status, total_ms, owner, checkpoint_id, "
            f"prompt_tokens, completion_tokens, tags, notes, created_at FROM runs"
            f"{where_sql} ORDER BY created_at DESC LIMIT ? OFFSET ?",
            (*args, limit, offset),
        ).fetchall()
        cols = [
            "run_id",
            "name",
            "status",
            "total_ms",
            "owner",
            "checkpoint_id",
            "prompt_tokens",
            "completion_tokens",
            "tags",
            "notes",
            "created_at",
        ]
        items = []
        for row in rows:
            item = dict(zip(cols, row))
            item["tags"] = json.loads(item["tags"])
            items.append(item)
        return {"items": items, "total": total}

    def get_run(self, run_id: str) -> dict[str, Any] | None:
        """Full run payload: metadata, topology, node spans, LLM calls."""
        row = self._conn.execute(
            "SELECT name, status, total_ms, owner, checkpoint_id, prompt_tokens, "
            "completion_tokens, topology, tags, notes, created_at FROM runs "
            "WHERE run_id = ?",
            (run_id,),
        ).fetchone()
        if row is None:
            return None
        run = {
            "run_id": run_id,
            "name": row[0],
            "status": row[1],
            "total_ms": row[2],
            "owner": row[3],
            "checkpoint_id": row[4],
            "prompt_tokens": row[5],
            "completion_tokens": row[6],
            "tags": json.loads(row[8]),
            "notes": row[9],
            "created_at": row[10],
        }
        topology = json.loads(row[7])
        run["topology"] = topology

        node_rows = self._conn.execute(
            "SELECT node_id, node_type, start_ms, end_ms, status, error, "
            "tool_calls, events "
            "FROM nodes WHERE run_id = ? ORDER BY start_ms",
            (run_id,),
        ).fetchall()
        calls: dict[str, list[dict]] = {}
        for call in self._conn.execute(
            "SELECT node_id, provider, model, messages, response, prompt_tokens, "
            "completion_tokens, latency_ms, cached FROM llm_calls WHERE run_id = ? "
            "ORDER BY id",
            (run_id,),
        ).fetchall():
            payload = {
                "node_id": call[0],
                "provider": call[1],
                "model": call[2],
                "messages": json.loads(call[3]),
                "response": call[4],
                "prompt_tokens": call[5],
                "completion_tokens": call[6],
                "latency_ms": call[7],
                "cached": bool(call[8]),
            }
            calls.setdefault(call[0], []).append(payload)

        nodes = []
        for nid, ntype, start, end, status, error, tool_calls, events in node_rows:
            nodes.append(
                {
                    "node_id": nid,
                    "node_type": ntype,
                    "start_ms": start,
                    "end_ms": end,
                    "duration_ms": None if end is None else round(end - start, 3),
                    "status": status,
                    "error": error,
                    "llm_calls": calls.get(nid, []),
                    "tool_calls": json.loads(tool_calls),
                    "events": json.loads(events),
                }
            )
        run["nodes"] = nodes
        run["llm_calls"] = [call for node in nodes for call in node["llm_calls"]]
        return run

    def update_run(
        self,
        run_id: str,
        *,
        tags: list[str] | None = None,
        notes: str | None = None,
    ) -> bool:
        """Patch a run's tags / notes. Returns False if the run doesn't exist."""
        fields: list[str] = []
        args: list[Any] = []
        if tags is not None:
            fields.append("tags = ?")
            args.append(json.dumps(list(tags)))
        if notes is not None:
            fields.append("notes = ?")
            args.append(notes)
        if not fields:
            return False
        args.append(run_id)
        cur = self._conn.execute(
            f"UPDATE runs SET {', '.join(fields)} WHERE run_id = ?", tuple(args)
        )
        self._conn.commit()
        return cur.rowcount > 0

    def close(self) -> None:
        self._conn.close()

get_run

get_run(run_id)

Full run payload: metadata, topology, node spans, LLM calls.

Source code in teff/observability/exporter.py

def get_run(self, run_id: str) -> dict[str, Any] | None:
    """Full run payload: metadata, topology, node spans, LLM calls."""
    row = self._conn.execute(
        "SELECT name, status, total_ms, owner, checkpoint_id, prompt_tokens, "
        "completion_tokens, topology, tags, notes, created_at FROM runs "
        "WHERE run_id = ?",
        (run_id,),
    ).fetchone()
    if row is None:
        return None
    run = {
        "run_id": run_id,
        "name": row[0],
        "status": row[1],
        "total_ms": row[2],
        "owner": row[3],
        "checkpoint_id": row[4],
        "prompt_tokens": row[5],
        "completion_tokens": row[6],
        "tags": json.loads(row[8]),
        "notes": row[9],
        "created_at": row[10],
    }
    topology = json.loads(row[7])
    run["topology"] = topology

    node_rows = self._conn.execute(
        "SELECT node_id, node_type, start_ms, end_ms, status, error, "
        "tool_calls, events "
        "FROM nodes WHERE run_id = ? ORDER BY start_ms",
        (run_id,),
    ).fetchall()
    calls: dict[str, list[dict]] = {}
    for call in self._conn.execute(
        "SELECT node_id, provider, model, messages, response, prompt_tokens, "
        "completion_tokens, latency_ms, cached FROM llm_calls WHERE run_id = ? "
        "ORDER BY id",
        (run_id,),
    ).fetchall():
        payload = {
            "node_id": call[0],
            "provider": call[1],
            "model": call[2],
            "messages": json.loads(call[3]),
            "response": call[4],
            "prompt_tokens": call[5],
            "completion_tokens": call[6],
            "latency_ms": call[7],
            "cached": bool(call[8]),
        }
        calls.setdefault(call[0], []).append(payload)

    nodes = []
    for nid, ntype, start, end, status, error, tool_calls, events in node_rows:
        nodes.append(
            {
                "node_id": nid,
                "node_type": ntype,
                "start_ms": start,
                "end_ms": end,
                "duration_ms": None if end is None else round(end - start, 3),
                "status": status,
                "error": error,
                "llm_calls": calls.get(nid, []),
                "tool_calls": json.loads(tool_calls),
                "events": json.loads(events),
            }
        )
    run["nodes"] = nodes
    run["llm_calls"] = [call for node in nodes for call in node["llm_calls"]]
    return run

list_runs

list_runs(limit=100, offset=0, status=None, name=None, owner=None, tag=None)

Dashboard rows (no payloads) with filtering and pagination.

Returns {"items": [...], "total": n} where total is the count before pagination, so the UI can render a page size / total. name / owner are case-insensitive substrings, tag matches an exact tag, status matches the run status exactly.

Source code in teff/observability/exporter.py

def list_runs(
    self,
    limit: int = 100,
    offset: int = 0,
    status: str | None = None,
    name: str | None = None,
    owner: str | None = None,
    tag: str | None = None,
) -> dict[str, Any]:
    """Dashboard rows (no payloads) with filtering and pagination.

    Returns ``{"items": [...], "total": n}`` where *total* is the count
    before pagination, so the UI can render a page size / total.
    *name* / *owner* are case-insensitive substrings, *tag* matches an
    exact tag, *status* matches the run status exactly.
    """
    where: list[str] = []
    args: list[Any] = []
    if status:
        where.append("status = ?")
        args.append(status)
    if name:
        where.append("LOWER(name) LIKE ?")
        args.append(f"%{name.lower()}%")
    if owner:
        where.append("LOWER(COALESCE(owner, '')) LIKE ?")
        args.append(f"%{owner.lower()}%")
    if tag:
        where.append(
            "EXISTS (SELECT 1 FROM json_each(tags) WHERE json_each.value = ?)"
        )
        args.append(tag)
    where_sql = (" WHERE " + " AND ".join(where)) if where else ""

    total = self._conn.execute(
        f"SELECT COUNT(*) FROM runs{where_sql}", tuple(args)
    ).fetchone()[0]

    rows = self._conn.execute(
        f"SELECT run_id, name, status, total_ms, owner, checkpoint_id, "
        f"prompt_tokens, completion_tokens, tags, notes, created_at FROM runs"
        f"{where_sql} ORDER BY created_at DESC LIMIT ? OFFSET ?",
        (*args, limit, offset),
    ).fetchall()
    cols = [
        "run_id",
        "name",
        "status",
        "total_ms",
        "owner",
        "checkpoint_id",
        "prompt_tokens",
        "completion_tokens",
        "tags",
        "notes",
        "created_at",
    ]
    items = []
    for row in rows:
        item = dict(zip(cols, row))
        item["tags"] = json.loads(item["tags"])
        items.append(item)
    return {"items": items, "total": total}

update_run

update_run(run_id, *, tags=None, notes=None)

Patch a run's tags / notes. Returns False if the run doesn't exist.

Source code in teff/observability/exporter.py

def update_run(
    self,
    run_id: str,
    *,
    tags: list[str] | None = None,
    notes: str | None = None,
) -> bool:
    """Patch a run's tags / notes. Returns False if the run doesn't exist."""
    fields: list[str] = []
    args: list[Any] = []
    if tags is not None:
        fields.append("tags = ?")
        args.append(json.dumps(list(tags)))
    if notes is not None:
        fields.append("notes = ?")
        args.append(notes)
    if not fields:
        return False
    args.append(run_id)
    cur = self._conn.execute(
        f"UPDATE runs SET {', '.join(fields)} WHERE run_id = ?", tuple(args)
    )
    self._conn.commit()
    return cur.rowcount > 0

SpanEvent dataclass

One step of a node's execution, in chronological order.

kind is "llm" or "tool" and index points into the span's llm_calls / tool_calls lists. Together they let a UI render the exact sequence a node followed — LLM call, tool call and its result, next LLM call, and so on — instead of two separate piles.

Source code in teff/observability/model.py

@dataclass
class SpanEvent:
    """One step of a node's execution, in chronological order.

    ``kind`` is ``"llm"`` or ``"tool"`` and ``index`` points into the
    span's ``llm_calls`` / ``tool_calls`` lists.  Together they let a UI
    render the exact sequence a node followed — LLM call, tool call and
    its result, next LLM call, and so on — instead of two separate piles.
    """

    kind: str
    index: int

    def to_dict(self) -> dict[str, Any]:
        return {"kind": self.kind, "index": self.index}

    @staticmethod
    def from_dict(data: dict[str, Any]) -> "SpanEvent":
        return SpanEvent(
            kind=str(data.get("kind") or ""),
            index=int(data.get("index") or 0),
        )

ToolCall dataclass

One tool invocation: what the model requested and what ran.

Tool calls are parsed out of the LLM message payloads (assistant tool_calls blocks matched to the following role: tool results), so a node's tool usage is a first-class citizen, not buried in the raw messages. ok is False when the tool returned an "Error: ..." result.

Source code in teff/observability/model.py

@dataclass
class ToolCall:
    """One tool invocation: what the model requested and what ran.

    Tool calls are parsed out of the LLM message payloads (assistant
    ``tool_calls`` blocks matched to the following ``role: tool`` results),
    so a node's tool usage is a first-class citizen, not buried in the raw
    messages.  ``ok`` is ``False`` when the tool returned an ``"Error: ..."``
    result.
    """

    name: str
    args: str = "{}"
    result: str = ""
    ok: bool = True

    def to_dict(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "args": self.args,
            "result": self.result,
            "ok": self.ok,
        }

    @staticmethod
    def from_dict(data: dict[str, Any]) -> "ToolCall":
        return ToolCall(
            name=str(data.get("name") or ""),
            args=str(data.get("args") or "{}"),
            result=str(data.get("result") or ""),
            ok=bool(data.get("ok", True)),
        )

TraceExporter

Bases: ABC

Persist a completed :class:Run to some backend.

Methods:

Name	Description
close	Release any resources held by the exporter.
export	Persist run and return its backend run id (None if unknown).

Source code in teff/observability/exporter.py

class TraceExporter(ABC):
    """Persist a completed :class:`Run` to some backend."""

    @abstractmethod
    def export(self, run: Run) -> str | None:
        """Persist *run* and return its backend run id (``None`` if unknown).

        Idempotent when the store uses run ids.
        """

    @abstractmethod
    def close(self) -> None:
        """Release any resources held by the exporter."""

close abstractmethod

close()

Release any resources held by the exporter.

Source code in teff/observability/exporter.py

@abstractmethod
def close(self) -> None:
    """Release any resources held by the exporter."""

export abstractmethod

export(run)

Persist run and return its backend run id (None if unknown).

Idempotent when the store uses run ids.

Source code in teff/observability/exporter.py

@abstractmethod
def export(self, run: Run) -> str | None:
    """Persist *run* and return its backend run id (``None`` if unknown).

    Idempotent when the store uses run ids.
    """

build_observability

build_observability(config, *, base_dir='.', graph=None, name='workflow')

Assemble a :class:GraphObserver from an observability: block.

Returns None when the block is missing or declares no sinks. The observer is wired with the graph topology so remote exporters and the dashboard can render the flow.

Source code in teff/observability/builder.py

def build_observability(
    config: dict[str, Any] | None,
    *,
    base_dir: str = ".",
    graph=None,
    name: str = "workflow",
) -> GraphObserver | None:
    """Assemble a :class:`GraphObserver` from an ``observability:`` block.

    Returns ``None`` when the block is missing or declares no sinks.  The
    observer is wired with the graph topology so remote exporters and the
    dashboard can render the flow.
    """
    if not config:
        return None
    exporters = _build_exporters(config, base_dir=base_dir)
    if not exporters:
        return None
    return GraphObserver(
        name,
        exporter=CompositeExporter(exporters),
        topology=topology_from_graph(graph) if graph is not None else None,
    )

build_observer_factory

build_observer_factory(config, *, base_dir='.', graph=None, name='workflow')

Like :func:build_observability but for repeated runs.

Returns a zero-arg callable that yields a fresh observer per run while sharing one set of exporters — the right shape for a daemon that traces every tick. None when observability is not configured.

Source code in teff/observability/builder.py

def build_observer_factory(
    config: dict[str, Any] | None,
    *,
    base_dir: str = ".",
    graph=None,
    name: str = "workflow",
) -> "Callable[[], GraphObserver] | None":
    """Like :func:`build_observability` but for repeated runs.

    Returns a zero-arg callable that yields a *fresh* observer per run while
    sharing one set of exporters — the right shape for a daemon that traces
    every tick.  ``None`` when observability is not configured.
    """
    if not config:
        return None
    exporters = _build_exporters(config, base_dir=base_dir)
    if not exporters:
        return None
    composite = CompositeExporter(exporters)
    topology = topology_from_graph(graph) if graph is not None else None

    def factory() -> GraphObserver:
        return GraphObserver(name, exporter=composite, topology=topology)

    return factory

build_remote_exporter

build_remote_exporter(spec, *, base_dir='.')

Construct a remote exporter from one observability.export entry.

Source code in teff/observability/builder.py

def build_remote_exporter(
    spec: dict[str, Any], *, base_dir: str = "."
) -> TraceExporter:
    """Construct a remote exporter from one ``observability.export`` entry."""
    kind = spec.get("type")
    timeout = float(spec.get("timeout", 10.0))
    retries = int(spec.get("retries", 3))
    backoff = float(spec.get("backoff", 1.0))
    if kind == "webhook":
        url = spec.get("url")
        if not url and spec.get("url_env"):
            url = os.environ.get(spec["url_env"])
        if not url:
            raise ConfigError(
                "observability.export.webhook: 'url' or 'url_env' is required"
            )
        return HttpExporter(
            url,
            headers=dict(spec.get("headers") or {}),
            timeout=timeout,
            retries=retries,
            backoff=backoff,
        )
    if kind == "langfuse":
        host = spec.get("host")
        if not host:
            raise ConfigError("observability.export.langfuse: 'host' is required")
        public_key = _require_env(
            spec, "public_key_env", "LANGFUSE_PUBLIC_KEY", "langfuse"
        )
        secret_key = _require_env(
            spec, "secret_key_env", "LANGFUSE_SECRET_KEY", "langfuse"
        )
        return LangfuseExporter(
            host,
            public_key,
            secret_key,
            timeout=timeout,
            retries=retries,
            backoff=backoff,
        )
    if kind == "langsmith":
        api_url = (
            spec.get("api_url")
            or os.environ.get("LANGCHAIN_ENDPOINT")
            or "https://api.smith.langchain.com"
        )
        api_key = _require_env(spec, "api_key_env", "LANGCHAIN_API_KEY", "langsmith")
        project = spec.get("project") or os.environ.get("LANGCHAIN_PROJECT")
        return LangsmithExporter(
            api_url,
            api_key,
            project=project,
            timeout=timeout,
            retries=retries,
            backoff=backoff,
        )
    raise ConfigError(f"observability.export: unknown exporter type {kind!r}")

topology_from_graph

topology_from_graph(graph)

Capture {nodes, edges} from a compiled :class:~teff.graph.Graph.

Each node is {"id", "type"}; each edge is {"source", "target", "condition"} (condition omitted when unconditional).

Source code in teff/observability/topology.py

def topology_from_graph(graph) -> GraphTopology:
    """Capture ``{nodes, edges}`` from a compiled :class:`~teff.graph.Graph`.

    Each node is ``{"id", "type"}``; each edge is ``{"source", "target",
    "condition"}`` (``condition`` omitted when unconditional).
    """
    nodes = [
        {"id": node_id, "type": node.type} for node_id, node in graph.nodes.items()
    ]
    edges: list[dict[str, Any]] = []
    for edge in graph.edges:
        item: dict[str, Any] = {"source": edge.source_id, "target": edge.target_id}
        if getattr(edge, "condition", None):
            item["condition"] = edge.condition
        edges.append(item)
    return GraphTopology(nodes=nodes, edges=edges)