aaa.observe

Observer

Observer(
    engine: Engine,
    sinks: Sequence[TraceSink] | TraceSink,
    *,
    capture_content: bool = False,
    content_limit: int = 4000,
    sample: float = 1.0,
    tenant: str | None = None,
    settle_s: float = 0.2,
    flush_interval_s: float | None = None,
    sink_timeout_s: float = 30.0,
    max_tracked: int = 10000,
    group: str = "aaa.observe",
)

drain async

drain(timeout_s: float = 10.0) -> bool

Wait until every trace for runs that have completed so far has been exported. (Events reach the observer through the bus, so the bus is flushed first; a distributed bus delivers on its own schedule and has nothing to flush.)

export_trace async

export_trace(
    root_run_id: str, *, final: bool = True
) -> int

Export the trace rooted at root_run_id. final=False sends only closed spans not yet sent, plus a provisional root.

flush_open async

flush_open() -> int

Export what is new in every root run that is still open (sessions, or runs hosted by a crashed process).

backfill async

backfill(
    *,
    since: datetime | None = None,
    limit: int | None = None,
) -> int

Export traces for finished root runs already in the log (to point a new backend at history).

TraceSink

Bases: Protocol

export async

export(spans: Sequence[Span]) -> None

Deliver spans (ordered parents first). Raise ExportError on failure; the observer logs it and carries on, so a broken sink never affects a run. Must be idempotent: ids are deterministic and the same span may be sent again after a restart.

InMemorySink

InMemorySink()

Keeps everything it is given. The reference sink (and the one tests assert against).

ConsoleSink

ConsoleSink(out: TextIO | None = None)

Prints each trace as an indented tree: the quickest way to see what a run did.

CallableSink

CallableSink(
    export: Callable[
        [Sequence[Span]], Any | Awaitable[Any]
    ],
)

Wrap a function f(spans) (sync or async) as a sink: the shortest way to ship traces to anything.

OTelSink

OTelSink(
    exporter: Any = None,
    *,
    dialect: Dialect | None = None,
    service_name: str = "aaa",
    resource_attributes: dict[str, Any] | None = None,
    batch_size: int = 512,
)

Export spans through any OpenTelemetry SpanExporter (OTLP/HTTP, console, in-memory, a vendor's).

score async

score(
    trace_id: str,
    scores: Sequence[Score],
    *,
    span_id: str | None = None,
) -> None

Attach scores as evaluate <name> evaluator spans under span_id (the generic OpenTelemetry way: a backend that understands evaluators shows them, one that does not still keeps them as spans). Ids are fixed per (trace, span, name).

LangfuseSink

LangfuseSink(
    public_key: str,
    secret_key: str,
    host: str = "https://cloud.langfuse.com",
    *,
    environment: str | None = None,
    tags: Sequence[str] = (),
    http: AsyncClient | None = None,
    **kw: Any,
)

Bases: _HttpScores, OTelSink

Langfuse over OTLP/HTTP (/api/public/otel; Langfuse accepts HTTP only, not gRPC).

score async

score(
    trace_id: str,
    scores: Sequence[Score],
    *,
    span_id: str | None = None,
) -> None

Send scores to POST /api/public/scores (attached to the trace, or to the observation span_id). Idempotent.

LangSmithSink

LangSmithSink(
    api_key: str,
    project: str | None = None,
    endpoint: str = "https://api.smith.langchain.com/otel",
    *,
    provider: str | None = None,
    api_url: str = "https://api.smith.langchain.com",
    http: AsyncClient | None = None,
    **kw: Any,
)

Bases: _HttpScores, OTelSink

LangSmith over OTLP/HTTP. Note LangSmith silently drops a span whose parent never arrives; the observer exports whole traces parents-first for this reason.

score async

score(
    trace_id: str,
    scores: Sequence[Score],
    *,
    span_id: str | None = None,
) -> None

Send scores as LangSmith feedback on the run that span_id became (span_id is required: LangSmith run ids come from span ids, not trace ids; an OnlineEvaluator passes it).

PhoenixSink

PhoenixSink(
    endpoint: str = "http://localhost:6006/v1/traces",
    project: str | None = None,
    api_key: str | None = None,
    http: AsyncClient | None = None,
    **kw: Any,
)

Bases: _HttpScores, OTelSink

Arize Phoenix over OTLP/HTTP, using OpenInference attributes.

score async

score(
    trace_id: str,
    scores: Sequence[Score],
    *,
    span_id: str | None = None,
) -> None

Send scores as Phoenix annotations (on the span span_id, else on the trace). Repeats update in place; if the span has not been ingested yet the call is retried briefly.

Span dataclass

Span(
    trace_id: str,
    span_id: str,
    parent_id: str | None,
    name: str,
    kind: SpanKind,
    start: datetime,
    end: datetime | None,
    status: Status = "ok",
    status_message: str | None = None,
    attributes: dict[str, Any] = dict(),
    events: list[SpanEvent] = list(),
    run_id: str = "",
    tenant: str | None = None,
    user: str | None = None,
    input: Any = None,
    output: Any = None,
)

spans_from_events

spans_from_events(
    events: Sequence[Event],
    *,
    parent_events: Sequence[Event] | None = None,
    capture_content: bool = False,
    content_limit: int = 4000,
) -> list[Span]

The spans of ONE run, from that run's events (in log order).

parent_events (the parent run's events) lets a child run's root span be parented to the exact tool call or session turn that started it; without it the child hangs under the parent's run span. Spans still open (no end event) have end=None. With capture_content, prompts, replies, tool arguments and results are attached (clipped); events are already redacted if the engine has a redactor, but content leaves your process, so it is opt-in.

Score

Bases: BaseModel

numeric property

numeric: float | None

The score as a number if it is one (bool -> 0/1), else None (a label).

passed property

passed: bool | None

True/False if the score is pass/fail-like (a bool, or a number against the 0.5 midpoint), else None.