Saltar a contenido

API reference

Generated from the apogee-ai-observability source with mkdocstrings. Every symbol below is exported from apogee_ai_observability, so it is part of the supported public surface.

Application · Use cases

EmitSpanUseCase

Python
EmitSpanUseCase(emitter: ITelemetryEmitter)

Sends a span to the configured emitter (and optionally a CostTracker).

Kept thin on purpose: orchestration logic lives in callers (e.g., the agent runtime); this just centralises the contract.

Source code in apogee_ai_observability/application/use_cases/emit_span_use_case.py
Python
def __init__(self, emitter: ITelemetryEmitter) -> None:
    self._emitter = emitter

execute async

Python
execute(span: Span) -> None
Source code in apogee_ai_observability/application/use_cases/emit_span_use_case.py
Python
async def execute(self, span: Span) -> None:
    await self._emitter.emit_span(span)

EndTraceUseCase

Python
EndTraceUseCase(store: ITraceStore)

Mark a trace as finished and persist it to the store.

Source code in apogee_ai_observability/application/use_cases/end_trace_use_case.py
Python
def __init__(self, store: ITraceStore) -> None:
    self._store = store

execute async

Python
execute(trace: Trace) -> Trace
Source code in apogee_ai_observability/application/use_cases/end_trace_use_case.py
Python
async def execute(self, trace: Trace) -> Trace:
    trace.finished_at = datetime.now(timezone.utc)
    return await self._store.save(trace)

GetTraceUseCase

Python
GetTraceUseCase(store: ITraceStore)
Source code in apogee_ai_observability/application/use_cases/get_trace_use_case.py
Python
def __init__(self, store: ITraceStore) -> None:
    self._store = store

execute async

Python
execute(trace_id: str) -> Trace
Source code in apogee_ai_observability/application/use_cases/get_trace_use_case.py
Python
async def execute(self, trace_id: str) -> Trace:
    trace = await self._store.find(trace_id)
    if trace is None:
        raise TraceNotFoundException(trace_id)
    return trace

ListTracesUseCase

Python
ListTracesUseCase(store: ITraceStore)
Source code in apogee_ai_observability/application/use_cases/list_traces_use_case.py
Python
def __init__(self, store: ITraceStore) -> None:
    self._store = store

execute async

Python
execute(*, limit: int | None = None, tenant_id: str | None = None) -> list[Trace]
Source code in apogee_ai_observability/application/use_cases/list_traces_use_case.py
Python
async def execute(
    self,
    *,
    limit: int | None = None,
    tenant_id: str | None = None,
) -> list[Trace]:
    return await self._store.list(limit=limit, tenant_id=tenant_id)

RecordCostUseCase

Python
RecordCostUseCase(calculator: ICostCalculator, tracker: CostTracker)

Estimate cost from usage, persist into the tracker.

Source code in apogee_ai_observability/application/use_cases/record_cost_use_case.py
Python
def __init__(self, calculator: ICostCalculator, tracker: CostTracker) -> None:
    self._calculator = calculator
    self._tracker = tracker

execute async

Python
execute(usage: LLMUsage, *, trace_id: str = '', span_id: str | None = None, user_id: str | None = None, tenant_id: str | None = None) -> CostLedgerEntry
Source code in apogee_ai_observability/application/use_cases/record_cost_use_case.py
Python
async def execute(
    self,
    usage: LLMUsage,
    *,
    trace_id: str = "",
    span_id: str | None = None,
    user_id: str | None = None,
    tenant_id: str | None = None,
) -> CostLedgerEntry:
    entry = self._calculator.estimate(
        usage,
        trace_id=trace_id,
        span_id=span_id,
        user_id=user_id,
        tenant_id=tenant_id,
    )
    self._tracker.record(entry)
    return entry

ReplayTraceUseCase

Python
ReplayTraceUseCase(store: ITraceStore, emitter: ITelemetryEmitter)

Load a stored trace and re-emit through emitter.

Source code in apogee_ai_observability/application/use_cases/replay_trace_use_case.py
Python
def __init__(self, store: ITraceStore, emitter: ITelemetryEmitter) -> None:
    self._store = store
    self._replayer = TraceReplayer(emitter)

execute async

Python
execute(trace_id: str) -> int
Source code in apogee_ai_observability/application/use_cases/replay_trace_use_case.py
Python
async def execute(self, trace_id: str) -> int:
    trace = await self._store.find(trace_id)
    if trace is None:
        raise TraceNotFoundException(trace_id)
    return await self._replayer.replay(trace)

StartTraceUseCase

Construct a fresh Trace with a new id.

execute async

Python
execute(*, name: str = '', metadata: dict[str, str] | None = None) -> Trace
Source code in apogee_ai_observability/application/use_cases/start_trace_use_case.py
Python
async def execute(self, *, name: str = "", metadata: dict[str, str] | None = None) -> Trace:
    return Trace(
        trace_id=secrets.token_hex(16),
        name=name,
        metadata=dict(metadata or {}),
    )

Domain

CostLedgerEntry dataclass

Python
CostLedgerEntry(trace_id: str, span_id: str | None, provider: str, model: str, input_tokens: int, output_tokens: int, input_cost_usd: float, output_cost_usd: float, user_id: str | None = None, tenant_id: str | None = None, timestamp: datetime = (lambda: now(utc))())

Single cost charge attributable to a trace/span/user/tenant.

trace_id instance-attribute

Python
trace_id: str

span_id instance-attribute

Python
span_id: str | None

provider instance-attribute

Python
provider: str

model instance-attribute

Python
model: str

input_tokens instance-attribute

Python
input_tokens: int

output_tokens instance-attribute

Python
output_tokens: int

input_cost_usd instance-attribute

Python
input_cost_usd: float

output_cost_usd instance-attribute

Python
output_cost_usd: float

user_id class-attribute instance-attribute

Python
user_id: str | None = None

tenant_id class-attribute instance-attribute

Python
tenant_id: str | None = None

timestamp class-attribute instance-attribute

Python
timestamp: datetime = field(default_factory=lambda: now(utc))

total_cost_usd property

Python
total_cost_usd: float

total_tokens property

Python
total_tokens: int

LLMOpName

Bases: str, Enum

Standardised span names so dashboards can group across emitters.

LLM_CALL class-attribute instance-attribute

Python
LLM_CALL = 'llm.call'

LLM_STREAM class-attribute instance-attribute

Python
LLM_STREAM = 'llm.stream'

LLM_EMBEDDING class-attribute instance-attribute

Python
LLM_EMBEDDING = 'llm.embedding'

TOOL_EXEC class-attribute instance-attribute

Python
TOOL_EXEC = 'tool.exec'
Python
RETRIEVAL_SEARCH = 'retrieval.search'

AGENT_ITERATION class-attribute instance-attribute

Python
AGENT_ITERATION = 'agent.iteration'

AGENT_RUN class-attribute instance-attribute

Python
AGENT_RUN = 'agent.run'

PROMPT_RENDER class-attribute instance-attribute

Python
PROMPT_RENDER = 'prompt.render'

EVAL_CASE class-attribute instance-attribute

Python
EVAL_CASE = 'eval.case'

LLMUsage dataclass

Python
LLMUsage(provider: str, model: str, prompt_tokens: int = 0, completion_tokens: int = 0, total_tokens: int = 0)

provider instance-attribute

Python
provider: str

model instance-attribute

Python
model: str

prompt_tokens class-attribute instance-attribute

Python
prompt_tokens: int = 0

completion_tokens class-attribute instance-attribute

Python
completion_tokens: int = 0

total_tokens class-attribute instance-attribute

Python
total_tokens: int = 0

LatencyBucket dataclass

Python
LatencyBucket(p50_ms: float, p95_ms: float, p99_ms: float, max_ms: float, samples: int, window_seconds: int = 60)

Aggregated latency stats over a window.

p50_ms instance-attribute

Python
p50_ms: float

p95_ms instance-attribute

Python
p95_ms: float

p99_ms instance-attribute

Python
p99_ms: float

max_ms instance-attribute

Python
max_ms: float

samples instance-attribute

Python
samples: int

window_seconds class-attribute instance-attribute

Python
window_seconds: int = 60

MetricSample dataclass

Python
MetricSample(name: str, value: float, unit: str = '', attributes: dict[str, str] = dict(), timestamp: datetime = (lambda: now(utc))())

name instance-attribute

Python
name: str

value instance-attribute

Python
value: float

unit class-attribute instance-attribute

Python
unit: str = ''

attributes class-attribute instance-attribute

Python
attributes: dict[str, str] = field(default_factory=dict)

timestamp class-attribute instance-attribute

Python
timestamp: datetime = field(default_factory=lambda: now(utc))

Span dataclass

Python
Span(name: str, kind: SpanKind = INTERNAL, span_id: str = (lambda: token_hex(8))(), trace_id: str = (lambda: token_hex(16))(), parent_span_id: str | None = None, start_time: datetime = (lambda: now(utc))(), end_time: datetime | None = None, status: SpanStatus = UNSET, error: str | None = None, attributes: dict[str, Any] = dict(), events: list[dict[str, Any]] = list(), usage: LLMUsage | None = None)

A single operation in a trace.

Mutable because emitters often build the span before/after work runs (set end_time, status, usage post-hoc).

name instance-attribute

Python
name: str

kind class-attribute instance-attribute

Python
kind: SpanKind = INTERNAL

span_id class-attribute instance-attribute

Python
span_id: str = field(default_factory=lambda: token_hex(8))

trace_id class-attribute instance-attribute

Python
trace_id: str = field(default_factory=lambda: token_hex(16))

parent_span_id class-attribute instance-attribute

Python
parent_span_id: str | None = None

start_time class-attribute instance-attribute

Python
start_time: datetime = field(default_factory=lambda: now(utc))

end_time class-attribute instance-attribute

Python
end_time: datetime | None = None

status class-attribute instance-attribute

Python
status: SpanStatus = UNSET

error class-attribute instance-attribute

Python
error: str | None = None

attributes class-attribute instance-attribute

Python
attributes: dict[str, Any] = field(default_factory=dict)

events class-attribute instance-attribute

Python
events: list[dict[str, Any]] = field(default_factory=list)

usage class-attribute instance-attribute

Python
usage: LLMUsage | None = None

duration_ms property

Python
duration_ms: float | None

end

Python
end(*, status: SpanStatus | None = None, error: str | None = None) -> None
Source code in apogee_ai_observability/domain/entities/span.py
Python
def end(self, *, status: SpanStatus | None = None, error: str | None = None) -> None:
    self.end_time = datetime.now(timezone.utc)
    if error is not None:
        self.error = error
        self.status = status if status is not None else SpanStatus.ERROR
    else:
        self.status = status if status is not None else SpanStatus.OK

add_event

Python
add_event(name: str, **attrs: Any) -> None
Source code in apogee_ai_observability/domain/entities/span.py
Python
def add_event(self, name: str, **attrs: Any) -> None:
    self.events.append(
        {
            "name": name,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "attributes": dict(attrs),
        }
    )

set_attribute

Python
set_attribute(key: str, value: Any) -> None
Source code in apogee_ai_observability/domain/entities/span.py
Python
def set_attribute(self, key: str, value: Any) -> None:
    self.attributes[key] = value

set_usage

Python
set_usage(usage: LLMUsage) -> None
Source code in apogee_ai_observability/domain/entities/span.py
Python
def set_usage(self, usage: LLMUsage) -> None:
    self.usage = usage

SpanStatus

Bases: str, Enum

UNSET class-attribute instance-attribute

Python
UNSET = 'unset'

OK class-attribute instance-attribute

Python
OK = 'ok'

ERROR class-attribute instance-attribute

Python
ERROR = 'error'

Trace dataclass

Python
Trace(trace_id: str, name: str = '', started_at: datetime = (lambda: now(utc))(), finished_at: datetime | None = None, spans: list[Span] = list(), metadata: dict[str, str] = dict())

A collection of spans sharing the same trace_id.

trace_id instance-attribute

Python
trace_id: str

name class-attribute instance-attribute

Python
name: str = ''

started_at class-attribute instance-attribute

Python
started_at: datetime = field(default_factory=lambda: now(utc))

finished_at class-attribute instance-attribute

Python
finished_at: datetime | None = None

spans class-attribute instance-attribute

Python
spans: list[Span] = field(default_factory=list)

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

root_span property

Python
root_span: Span | None

total_duration_ms property

Python
total_duration_ms: float

error_count property

Python
error_count: int

add_span

Python
add_span(span: Span) -> None
Source code in apogee_ai_observability/domain/entities/trace.py
Python
def add_span(self, span: Span) -> None:
    if span.trace_id != self.trace_id:
        raise ValueError(
            f"span.trace_id {span.trace_id!r} != trace.trace_id {self.trace_id!r}"
        )
    self.spans.append(span)

TraceContext dataclass

Python
TraceContext(trace_id: str = (lambda: token_hex(16))(), span_id: str = (lambda: token_hex(8))(), parent_span_id: str | None = None, sampled: bool = True, baggage: dict[str, str] = dict())

Carries the active trace + span ids across async boundaries.

Modeled after the W3C trace-context: 16-byte trace id and 8-byte span id encoded as lowercase hex.

trace_id class-attribute instance-attribute

Python
trace_id: str = field(default_factory=lambda: token_hex(16))

span_id class-attribute instance-attribute

Python
span_id: str = field(default_factory=lambda: token_hex(8))

parent_span_id class-attribute instance-attribute

Python
parent_span_id: str | None = None

sampled class-attribute instance-attribute

Python
sampled: bool = True

baggage class-attribute instance-attribute

Python
baggage: dict[str, str] = field(default_factory=dict)

child

Python
child() -> TraceContext
Source code in apogee_ai_observability/domain/value_objects/trace_context.py
Python
def child(self) -> TraceContext:
    return TraceContext(
        trace_id=self.trace_id,
        span_id=secrets.token_hex(8),
        parent_span_id=self.span_id,
        sampled=self.sampled,
        baggage=dict(self.baggage),
    )

to_traceparent

Python
to_traceparent() -> str

Build a W3C traceparent header (00 version, 01 sampled bit).

Source code in apogee_ai_observability/domain/value_objects/trace_context.py
Python
def to_traceparent(self) -> str:
    """Build a W3C traceparent header (00 version, 01 sampled bit)."""
    flag = "01" if self.sampled else "00"
    return f"00-{self.trace_id}-{self.span_id}-{flag}"

semantic

LatencyBucket dataclass

Python
LatencyBucket(p50_ms: float, p95_ms: float, p99_ms: float, max_ms: float, samples: int, window_seconds: int = 60)

Aggregated latency stats over a window.

p50_ms instance-attribute
Python
p50_ms: float
p95_ms instance-attribute
Python
p95_ms: float
p99_ms instance-attribute
Python
p99_ms: float
max_ms instance-attribute
Python
max_ms: float
samples instance-attribute
Python
samples: int
window_seconds class-attribute instance-attribute
Python
window_seconds: int = 60

TraceContext dataclass

Python
TraceContext(trace_id: str = (lambda: token_hex(16))(), span_id: str = (lambda: token_hex(8))(), parent_span_id: str | None = None, sampled: bool = True, baggage: dict[str, str] = dict())

Carries the active trace + span ids across async boundaries.

Modeled after the W3C trace-context: 16-byte trace id and 8-byte span id encoded as lowercase hex.

trace_id class-attribute instance-attribute
Python
trace_id: str = field(default_factory=lambda: token_hex(16))
span_id class-attribute instance-attribute
Python
span_id: str = field(default_factory=lambda: token_hex(8))
parent_span_id class-attribute instance-attribute
Python
parent_span_id: str | None = None
sampled class-attribute instance-attribute
Python
sampled: bool = True
baggage class-attribute instance-attribute
Python
baggage: dict[str, str] = field(default_factory=dict)
child
Python
child() -> TraceContext
Source code in apogee_ai_observability/domain/value_objects/trace_context.py
Python
def child(self) -> TraceContext:
    return TraceContext(
        trace_id=self.trace_id,
        span_id=secrets.token_hex(8),
        parent_span_id=self.span_id,
        sampled=self.sampled,
        baggage=dict(self.baggage),
    )
to_traceparent
Python
to_traceparent() -> str

Build a W3C traceparent header (00 version, 01 sampled bit).

Source code in apogee_ai_observability/domain/value_objects/trace_context.py
Python
def to_traceparent(self) -> str:
    """Build a W3C traceparent header (00 version, 01 sampled bit)."""
    flag = "01" if self.sampled else "00"
    return f"00-{self.trace_id}-{self.span_id}-{flag}"

Domain · Enums

EmitterKind

Bases: str, Enum

NOOP class-attribute instance-attribute

Python
NOOP = 'noop'

IN_MEMORY class-attribute instance-attribute

Python
IN_MEMORY = 'in_memory'

CONSOLE class-attribute instance-attribute

Python
CONSOLE = 'console'

STRUCTURED_JSON class-attribute instance-attribute

Python
STRUCTURED_JSON = 'structured_json'

LANGFUSE class-attribute instance-attribute

Python
LANGFUSE = 'langfuse'

LANGSMITH class-attribute instance-attribute

Python
LANGSMITH = 'langsmith'

PHOENIX class-attribute instance-attribute

Python
PHOENIX = 'phoenix'

HELICONE class-attribute instance-attribute

Python
HELICONE = 'helicone'

OTEL class-attribute instance-attribute

Python
OTEL = 'otel'

DATADOG class-attribute instance-attribute

Python
DATADOG = 'datadog'

SpanKind

Bases: str, Enum

AGENT class-attribute instance-attribute

Python
AGENT = 'agent'

LLM class-attribute instance-attribute

Python
LLM = 'llm'

TOOL class-attribute instance-attribute

Python
TOOL = 'tool'

RETRIEVAL class-attribute instance-attribute

Python
RETRIEVAL = 'retrieval'

EMBEDDING class-attribute instance-attribute

Python
EMBEDDING = 'embedding'

PROMPT class-attribute instance-attribute

Python
PROMPT = 'prompt'

EVAL class-attribute instance-attribute

Python
EVAL = 'eval'

HTTP class-attribute instance-attribute

Python
HTTP = 'http'

DB class-attribute instance-attribute

Python
DB = 'db'

INTERNAL class-attribute instance-attribute

Python
INTERNAL = 'internal'

Domain · Exceptions

EmitterFailureException

Python
EmitterFailureException(emitter: str, message: str)

Bases: ObservabilityError

Source code in apogee_ai_observability/domain/exceptions/observability_exceptions.py
Python
def __init__(self, emitter: str, message: str) -> None:
    super().__init__(f"Emitter {emitter!r} failed: {message}")
    self.emitter = emitter

emitter instance-attribute

Python
emitter = emitter

ObservabilityError

Bases: Exception

Base exception for apogee-ai-observability.

PricingNotAvailableException

Python
PricingNotAvailableException(provider: str, model: str)

Bases: ObservabilityError

Source code in apogee_ai_observability/domain/exceptions/observability_exceptions.py
Python
def __init__(self, provider: str, model: str) -> None:
    super().__init__(f"No pricing for {provider!r}/{model!r}")
    self.provider = provider
    self.model = model

provider instance-attribute

Python
provider = provider

model instance-attribute

Python
model = model

TraceNotFoundException

Python
TraceNotFoundException(trace_id: str)

Bases: ObservabilityError

Source code in apogee_ai_observability/domain/exceptions/observability_exceptions.py
Python
def __init__(self, trace_id: str) -> None:
    super().__init__(f"Trace {trace_id!r} not found")
    self.trace_id = trace_id

trace_id instance-attribute

Python
trace_id = trace_id

Domain · Protocols (ports)

ICostCalculator

Bases: Protocol

name instance-attribute

Python
name: str

estimate

Python
estimate(usage: LLMUsage, *, trace_id: str = '', span_id: str | None = None, user_id: str | None = None, tenant_id: str | None = None) -> CostLedgerEntry
Source code in apogee_ai_observability/domain/services/i_cost_calculator.py
Python
def estimate(
    self,
    usage: LLMUsage,
    *,
    trace_id: str = "",
    span_id: str | None = None,
    user_id: str | None = None,
    tenant_id: str | None = None,
) -> CostLedgerEntry:
    ...

IPricingProvider

Bases: Protocol

Returns (input_per_million, output_per_million) USD or None.

name instance-attribute

Python
name: str

lookup

Python
lookup(provider: str, model: str) -> tuple[float, float] | None
Source code in apogee_ai_observability/domain/services/i_pricing_provider.py
Python
def lookup(self, provider: str, model: str) -> tuple[float, float] | None:
    ...

ITelemetryEmitter

Bases: Protocol

Sink for spans and metrics produced by the agent runtime.

name instance-attribute

Python
name: str

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/domain/services/i_telemetry_emitter.py
Python
async def emit_span(self, span: Span) -> None:
    ...

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/domain/services/i_telemetry_emitter.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    ...

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/domain/services/i_telemetry_emitter.py
Python
async def shutdown(self) -> None:
    ...

ITraceStore

Bases: Protocol

get async

Python
get(trace_id: str) -> Trace
Source code in apogee_ai_observability/domain/services/i_trace_store.py
Python
async def get(self, trace_id: str) -> Trace:
    ...

find async

Python
find(trace_id: str) -> Trace | None
Source code in apogee_ai_observability/domain/services/i_trace_store.py
Python
async def find(self, trace_id: str) -> Trace | None:
    ...

save async

Python
save(trace: Trace) -> Trace
Source code in apogee_ai_observability/domain/services/i_trace_store.py
Python
async def save(self, trace: Trace) -> Trace:
    ...

list async

Python
list(*, limit: int | None = None, tenant_id: str | None = None) -> list[Trace]
Source code in apogee_ai_observability/domain/services/i_trace_store.py
Python
async def list(
    self,
    *,
    limit: int | None = None,
    tenant_id: str | None = None,
) -> list[Trace]:
    ...

Infrastructure

CompositeEmitter

Python
CompositeEmitter(children: Iterable[ITelemetryEmitter])

Fan-out wrapper: every event goes to every child emitter.

Failures of one child do not block the others.

Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
def __init__(self, children: Iterable[ITelemetryEmitter]) -> None:
    self._children: list[ITelemetryEmitter] = list(children)

name class-attribute instance-attribute

Python
name = 'composite'

children property

Python
children: list[ITelemetryEmitter]

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def emit_span(self, span: Span) -> None:
    for child in self._children:
        try:
            await child.emit_span(span)
        except Exception:  # noqa: BLE001 - one bad emitter must not poison the rest
            continue

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    for child in self._children:
        try:
            await child.emit_metric(sample)
        except Exception:  # noqa: BLE001
            continue

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def shutdown(self) -> None:
    for child in self._children:
        try:
            await child.shutdown()
        except Exception:  # noqa: BLE001
            continue

ConsoleEmitter

Python
ConsoleEmitter(*, stream=None)

Pretty single-line output, useful in dev.

Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
def __init__(self, *, stream=None) -> None:
    self._stream = stream or sys.stdout

name class-attribute instance-attribute

Python
name = 'console'

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def emit_span(self, span: Span) -> None:
    duration = f"{span.duration_ms:.1f}ms" if span.duration_ms is not None else "-"
    usage = ""
    if span.usage is not None:
        usage = f" tokens={span.usage.total_tokens}"
    print(
        f"[span] {span.kind.value:8s} {span.name:24s} {duration:>10s} "
        f"{span.status.value}{usage}",
        file=self._stream,
    )

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    print(
        f"[metric] {sample.name}={sample.value} {sample.unit}".rstrip(),
        file=self._stream,
    )

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def shutdown(self) -> None:
    return None

CostTracker

Python
CostTracker()

Append-only ledger with rollup helpers (by trace, user, tenant).

Source code in apogee_ai_observability/infrastructure/cost/cost_tracker.py
Python
def __init__(self) -> None:
    self._entries: list[CostLedgerEntry] = []

entries property

Python
entries: list[CostLedgerEntry]

total_usd property

Python
total_usd: float

record

Python
record(entry: CostLedgerEntry) -> None
Source code in apogee_ai_observability/infrastructure/cost/cost_tracker.py
Python
def record(self, entry: CostLedgerEntry) -> None:
    self._entries.append(entry)

extend

Python
extend(entries: Iterable[CostLedgerEntry]) -> None
Source code in apogee_ai_observability/infrastructure/cost/cost_tracker.py
Python
def extend(self, entries: Iterable[CostLedgerEntry]) -> None:
    self._entries.extend(entries)

by_trace

Python
by_trace() -> dict[str, float]
Source code in apogee_ai_observability/infrastructure/cost/cost_tracker.py
Python
def by_trace(self) -> dict[str, float]:
    out: dict[str, float] = defaultdict(float)
    for e in self._entries:
        out[e.trace_id] += e.total_cost_usd
    return {k: round(v, 6) for k, v in out.items()}

by_user

Python
by_user() -> dict[str, float]
Source code in apogee_ai_observability/infrastructure/cost/cost_tracker.py
Python
def by_user(self) -> dict[str, float]:
    out: dict[str, float] = defaultdict(float)
    for e in self._entries:
        if e.user_id:
            out[e.user_id] += e.total_cost_usd
    return {k: round(v, 6) for k, v in out.items()}

by_tenant

Python
by_tenant() -> dict[str, float]
Source code in apogee_ai_observability/infrastructure/cost/cost_tracker.py
Python
def by_tenant(self) -> dict[str, float]:
    out: dict[str, float] = defaultdict(float)
    for e in self._entries:
        if e.tenant_id:
            out[e.tenant_id] += e.total_cost_usd
    return {k: round(v, 6) for k, v in out.items()}

by_model

Python
by_model() -> dict[str, float]
Source code in apogee_ai_observability/infrastructure/cost/cost_tracker.py
Python
def by_model(self) -> dict[str, float]:
    out: dict[str, float] = defaultdict(float)
    for e in self._entries:
        out[f"{e.provider}/{e.model}"] += e.total_cost_usd
    return {k: round(v, 6) for k, v in out.items()}

DatadogLLMEmitter

Python
DatadogLLMEmitter(*, service: str = 'apogee-ai', env: str | None = None)

Adapter for Datadog LLM Observability via ddtrace.

Lazy import: install via pip install 'apogee-ai-observability[datadog]'.

Source code in apogee_ai_observability/infrastructure/emitters/datadog_emitter.py
Python
def __init__(
    self,
    *,
    service: str = "apogee-ai",
    env: str | None = None,
) -> None:
    try:
        import ddtrace  # type: ignore  # noqa: F401
    except ImportError as exc:
        raise ImportError(
            "DatadogLLMEmitter requires `ddtrace`. "
            "Install with: pip install 'apogee-ai-observability[datadog]'"
        ) from exc
    from ddtrace import tracer  # type: ignore

    self._tracer = tracer
    self._service = service
    self._env = env

name class-attribute instance-attribute

Python
name = 'datadog'

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/infrastructure/emitters/datadog_emitter.py
Python
async def emit_span(self, span: Span) -> None:
    try:
        with self._tracer.trace(
            span.name,
            service=self._service,
            resource=span.kind.value,
            span_type="llm",
        ) as dd_span:
            for key, value in span.attributes.items():
                dd_span.set_tag(key, value)
            if span.usage is not None:
                dd_span.set_metric("llm.prompt_tokens", span.usage.prompt_tokens)
                dd_span.set_metric("llm.completion_tokens", span.usage.completion_tokens)
                dd_span.set_metric("llm.total_tokens", span.usage.total_tokens)
                dd_span.set_tag("llm.model", span.usage.model)
                dd_span.set_tag("llm.provider", span.usage.provider)
            if self._env:
                dd_span.set_tag("env", self._env)
            if span.error:
                dd_span.set_traceback()
                dd_span.set_tag("error", True)
                dd_span.set_tag("error.message", span.error)
    except Exception as exc:  # noqa: BLE001
        raise EmitterFailureException(self.name, str(exc)) from exc

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/infrastructure/emitters/datadog_emitter.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    return None

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/infrastructure/emitters/datadog_emitter.py
Python
async def shutdown(self) -> None:
    return None

HeliconeEmitter

Python
HeliconeEmitter(*, api_key: str, base_url: str = 'https://api.helicone.ai')

Posts manual log events to Helicone's custom-events endpoint.

Helicone's primary mode is header-based proxying — this adapter targets the supplemental /v1/log style endpoint for spans we observed locally. Lazy import via [helicone] extra (httpx).

Source code in apogee_ai_observability/infrastructure/emitters/helicone_emitter.py
Python
def __init__(
    self,
    *,
    api_key: str,
    base_url: str = "https://api.helicone.ai",
) -> None:
    try:
        import httpx  # type: ignore  # noqa: F401
    except ImportError as exc:
        raise ImportError(
            "HeliconeEmitter requires `httpx`. "
            "Install with: pip install 'apogee-ai-observability[helicone]'"
        ) from exc
    self._api_key = api_key
    self._base_url = base_url.rstrip("/")
    self._client = None

name class-attribute instance-attribute

Python
name = 'helicone'

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/infrastructure/emitters/helicone_emitter.py
Python
async def emit_span(self, span: Span) -> None:
    try:
        payload = {
            "trace_id": span.trace_id,
            "span_id": span.span_id,
            "name": span.name,
            "kind": span.kind.value,
            "status": span.status.value,
            "start": span.start_time.isoformat(),
            "end": span.end_time.isoformat() if span.end_time else None,
            "duration_ms": span.duration_ms,
            "attributes": dict(span.attributes),
            "usage": (
                {
                    "model": span.usage.model,
                    "prompt_tokens": span.usage.prompt_tokens,
                    "completion_tokens": span.usage.completion_tokens,
                }
                if span.usage
                else None
            ),
        }
        response = await self._get_client().post("/v1/log", json=payload)
        response.raise_for_status()
    except Exception as exc:  # noqa: BLE001
        raise EmitterFailureException(self.name, str(exc)) from exc

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/infrastructure/emitters/helicone_emitter.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    return None

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/infrastructure/emitters/helicone_emitter.py
Python
async def shutdown(self) -> None:
    if self._client is not None:
        await self._client.aclose()
        self._client = None

InMemoryEmitter

Python
InMemoryEmitter()

Captures everything in memory — perfect for tests and CI replay.

Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
def __init__(self) -> None:
    self.spans: list[Span] = []
    self.metrics: list[MetricSample] = []

name class-attribute instance-attribute

Python
name = 'in_memory'

spans instance-attribute

Python
spans: list[Span] = []

metrics instance-attribute

Python
metrics: list[MetricSample] = []

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def emit_span(self, span: Span) -> None:
    self.spans.append(span)

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    self.metrics.append(sample)

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def shutdown(self) -> None:
    return None

clear

Python
clear() -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
def clear(self) -> None:
    self.spans.clear()
    self.metrics.clear()

InMemoryTraceStore

Python
InMemoryTraceStore()
Source code in apogee_ai_observability/infrastructure/replay/in_memory_trace_store.py
Python
def __init__(self) -> None:
    self._store: dict[str, Trace] = {}

name class-attribute instance-attribute

Python
name = 'memory'

get async

Python
get(trace_id: str) -> Trace
Source code in apogee_ai_observability/infrastructure/replay/in_memory_trace_store.py
Python
async def get(self, trace_id: str) -> Trace:
    if trace_id not in self._store:
        raise TraceNotFoundException(trace_id)
    return self._store[trace_id]

find async

Python
find(trace_id: str) -> Trace | None
Source code in apogee_ai_observability/infrastructure/replay/in_memory_trace_store.py
Python
async def find(self, trace_id: str) -> Trace | None:
    return self._store.get(trace_id)

save async

Python
save(trace: Trace) -> Trace
Source code in apogee_ai_observability/infrastructure/replay/in_memory_trace_store.py
Python
async def save(self, trace: Trace) -> Trace:
    self._store[trace.trace_id] = deepcopy(trace)
    return self._store[trace.trace_id]

list async

Python
list(*, limit: int | None = None, tenant_id: str | None = None) -> list[Trace]
Source code in apogee_ai_observability/infrastructure/replay/in_memory_trace_store.py
Python
async def list(
    self,
    *,
    limit: int | None = None,
    tenant_id: str | None = None,
) -> list[Trace]:
    items = sorted(self._store.values(), key=lambda t: t.started_at, reverse=True)
    if tenant_id is not None:
        items = [
            t for t in items if t.metadata.get("tenant.id") == tenant_id
        ]
    if limit is not None:
        items = items[:limit]
    return items

JsonTraceStore

Python
JsonTraceStore(root: str | Path)

One <root>/<trace_id>.json per trace + _index.json.

Source code in apogee_ai_observability/infrastructure/replay/json_trace_store.py
Python
def __init__(self, root: str | Path) -> None:
    self._root = Path(root)

name class-attribute instance-attribute

Python
name = 'json'

get async

Python
get(trace_id: str) -> Trace
Source code in apogee_ai_observability/infrastructure/replay/json_trace_store.py
Python
async def get(self, trace_id: str) -> Trace:
    trace = await self.find(trace_id)
    if trace is None:
        raise TraceNotFoundException(trace_id)
    return trace

find async

Python
find(trace_id: str) -> Trace | None
Source code in apogee_ai_observability/infrastructure/replay/json_trace_store.py
Python
async def find(self, trace_id: str) -> Trace | None:
    return await asyncio.to_thread(self._read_one, trace_id)

save async

Python
save(trace: Trace) -> Trace
Source code in apogee_ai_observability/infrastructure/replay/json_trace_store.py
Python
async def save(self, trace: Trace) -> Trace:
    await asyncio.to_thread(self._write_one, trace)
    return trace

list async

Python
list(*, limit: int | None = None, tenant_id: str | None = None) -> list[Trace]
Source code in apogee_ai_observability/infrastructure/replay/json_trace_store.py
Python
async def list(
    self,
    *,
    limit: int | None = None,
    tenant_id: str | None = None,
) -> list[Trace]:
    return await asyncio.to_thread(self._list, limit, tenant_id)

LangSmithEmitter

Python
LangSmithEmitter(*, api_key: str | None = None, project: str | None = None)

Adapter for LangChain LangSmith.

Lazy import: install via pip install 'apogee-ai-observability[langsmith]'.

Source code in apogee_ai_observability/infrastructure/emitters/langsmith_emitter.py
Python
def __init__(
    self,
    *,
    api_key: str | None = None,
    project: str | None = None,
) -> None:
    try:
        import langsmith  # type: ignore  # noqa: F401
    except ImportError as exc:
        raise ImportError(
            "LangSmithEmitter requires `langsmith`. "
            "Install with: pip install 'apogee-ai-observability[langsmith]'"
        ) from exc
    from langsmith import Client  # type: ignore

    self._client = Client(api_key=api_key)
    self._project = project

name class-attribute instance-attribute

Python
name = 'langsmith'

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/infrastructure/emitters/langsmith_emitter.py
Python
async def emit_span(self, span: Span) -> None:
    try:
        self._client.create_run(
            name=span.name,
            run_type=_map_kind(span.kind.value),
            inputs=span.attributes,
            outputs={"events": span.events} if span.events else None,
            start_time=span.start_time,
            end_time=span.end_time,
            error=span.error,
            project_name=self._project,
            trace_id=span.trace_id,
            parent_run_id=span.parent_span_id,
        )
    except Exception as exc:  # noqa: BLE001
        raise EmitterFailureException(self.name, str(exc)) from exc

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/infrastructure/emitters/langsmith_emitter.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    # LangSmith only accepts metrics tied to a run; fall back to attribute log.
    return None

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/infrastructure/emitters/langsmith_emitter.py
Python
async def shutdown(self) -> None:
    return None

LangfuseEmitter

Python
LangfuseEmitter(*, public_key: str | None = None, secret_key: str | None = None, host: str | None = None)

Adapter for Langfuse cloud or self-hosted.

Lazy import: install via pip install 'apogee-ai-observability[langfuse]'.

Source code in apogee_ai_observability/infrastructure/emitters/langfuse_emitter.py
Python
def __init__(
    self,
    *,
    public_key: str | None = None,
    secret_key: str | None = None,
    host: str | None = None,
) -> None:
    try:
        import langfuse  # type: ignore  # noqa: F401
    except ImportError as exc:
        raise ImportError(
            "LangfuseEmitter requires `langfuse`. "
            "Install with: pip install 'apogee-ai-observability[langfuse]'"
        ) from exc
    from langfuse import Langfuse  # type: ignore

    self._client = Langfuse(
        public_key=public_key,
        secret_key=secret_key,
        host=host,
    )

name class-attribute instance-attribute

Python
name = 'langfuse'

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/infrastructure/emitters/langfuse_emitter.py
Python
async def emit_span(self, span: Span) -> None:
    try:
        attributes = dict(span.attributes)
        usage = None
        if span.usage is not None:
            usage = {
                "input": span.usage.prompt_tokens,
                "output": span.usage.completion_tokens,
                "total": span.usage.total_tokens,
                "unit": "TOKENS",
            }
        kwargs = {
            "trace_id": span.trace_id,
            "id": span.span_id,
            "name": span.name,
            "start_time": span.start_time,
            "end_time": span.end_time,
            "metadata": attributes,
            "level": "ERROR" if span.error else "DEFAULT",
            "status_message": span.error,
        }
        if usage:
            self._client.generation(
                **kwargs,
                model=span.usage.model if span.usage else None,
                usage=usage,
            )
        else:
            self._client.span(**kwargs)
    except Exception as exc:  # noqa: BLE001
        raise EmitterFailureException(self.name, str(exc)) from exc

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/infrastructure/emitters/langfuse_emitter.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    try:
        self._client.score(
            name=sample.name,
            value=sample.value,
            comment=str(dict(sample.attributes)),
        )
    except Exception as exc:  # noqa: BLE001
        raise EmitterFailureException(self.name, str(exc)) from exc

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/infrastructure/emitters/langfuse_emitter.py
Python
async def shutdown(self) -> None:
    try:
        self._client.flush()
    except Exception:  # noqa: BLE001 - best effort
        return

NoopEmitter

name class-attribute instance-attribute

Python
name = 'noop'

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def emit_span(self, span: Span) -> None:
    return None

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    return None

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def shutdown(self) -> None:
    return None

OTelEmitter

Python
OTelEmitter(*, endpoint: str | None = None, service_name: str = 'apogee-ai', insecure: bool = True)

OpenTelemetry adapter — works with any OTLP backend (Jaeger, Tempo, Honeycomb, Datadog APM, Lightstep, etc).

Lazy import: install via pip install 'apogee-ai-observability[otel]'.

Source code in apogee_ai_observability/infrastructure/emitters/otel_emitter.py
Python
def __init__(
    self,
    *,
    endpoint: str | None = None,
    service_name: str = "apogee-ai",
    insecure: bool = True,
) -> None:
    try:
        from opentelemetry import metrics, trace  # type: ignore
        from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (  # type: ignore
            OTLPMetricExporter,
        )
        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (  # type: ignore
            OTLPSpanExporter,
        )
        from opentelemetry.sdk.metrics import MeterProvider  # type: ignore
        from opentelemetry.sdk.metrics.export import (  # type: ignore
            PeriodicExportingMetricReader,
        )
        from opentelemetry.sdk.resources import Resource  # type: ignore
        from opentelemetry.sdk.trace import TracerProvider  # type: ignore
        from opentelemetry.sdk.trace.export import (  # type: ignore
            BatchSpanProcessor,
        )
    except ImportError as exc:
        raise ImportError(
            "OTelEmitter requires opentelemetry-* packages. "
            "Install with: pip install 'apogee-ai-observability[otel]'"
        ) from exc

    resource = Resource.create({"service.name": service_name})

    tracer_provider = TracerProvider(resource=resource)
    span_exporter = OTLPSpanExporter(endpoint=endpoint, insecure=insecure)
    tracer_provider.add_span_processor(BatchSpanProcessor(span_exporter))
    trace.set_tracer_provider(tracer_provider)
    self._tracer = trace.get_tracer("apogee-ai-observability")

    metric_exporter = OTLPMetricExporter(endpoint=endpoint, insecure=insecure)
    meter_provider = MeterProvider(
        resource=resource,
        metric_readers=[PeriodicExportingMetricReader(metric_exporter)],
    )
    metrics.set_meter_provider(meter_provider)
    self._meter = metrics.get_meter("apogee-ai-observability")
    self._counters: dict[str, object] = {}

    self._tracer_provider = tracer_provider
    self._meter_provider = meter_provider

name class-attribute instance-attribute

Python
name = 'otel'

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/infrastructure/emitters/otel_emitter.py
Python
async def emit_span(self, span: Span) -> None:
    try:
        from opentelemetry.trace import StatusCode  # type: ignore

        with self._tracer.start_as_current_span(span.name) as ot_span:
            for key, value in span.attributes.items():
                ot_span.set_attribute(key, _coerce(value))
            if span.usage is not None:
                ot_span.set_attribute("llm.prompt_tokens", span.usage.prompt_tokens)
                ot_span.set_attribute("llm.completion_tokens", span.usage.completion_tokens)
                ot_span.set_attribute("llm.total_tokens", span.usage.total_tokens)
                ot_span.set_attribute("llm.model", span.usage.model)
                ot_span.set_attribute("llm.provider", span.usage.provider)
            if span.error:
                ot_span.record_exception(Exception(span.error))
                ot_span.set_status(StatusCode.ERROR, span.error)
            elif span.status == SpanStatus.OK:
                ot_span.set_status(StatusCode.OK)
    except Exception as exc:  # noqa: BLE001
        raise EmitterFailureException(self.name, str(exc)) from exc

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/infrastructure/emitters/otel_emitter.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    try:
        counter = self._counters.get(sample.name)
        if counter is None:
            counter = self._meter.create_counter(sample.name, unit=sample.unit or "1")
            self._counters[sample.name] = counter
        counter.add(sample.value, attributes=dict(sample.attributes))  # type: ignore[attr-defined]
    except Exception as exc:  # noqa: BLE001
        raise EmitterFailureException(self.name, str(exc)) from exc

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/infrastructure/emitters/otel_emitter.py
Python
async def shutdown(self) -> None:
    try:
        self._tracer_provider.shutdown()
        self._meter_provider.shutdown()
    except Exception:  # noqa: BLE001
        return

PRICING_TABLE module-attribute

Python
PRICING_TABLE: dict[str, dict[str, tuple[float, float]]] = {'openai': {'gpt-4o': (5.0, 15.0), 'gpt-4o-mini': (0.15, 0.6), 'gpt-4-turbo': (10.0, 30.0), 'gpt-4': (30.0, 60.0), 'gpt-3.5-turbo': (0.5, 1.5), 'o1': (15.0, 60.0), 'o1-mini': (3.0, 12.0), 'o3-mini': (1.1, 4.4), 'text-embedding-3-large': (0.13, 0.0), 'text-embedding-3-small': (0.02, 0.0)}, 'anthropic': {'claude-3-5-sonnet': (3.0, 15.0), 'claude-3-5-haiku': (0.8, 4.0), 'claude-3-opus': (15.0, 75.0), 'claude-3-sonnet': (3.0, 15.0), 'claude-3-haiku': (0.25, 1.25), 'claude-opus-4': (15.0, 75.0), 'claude-sonnet-4': (3.0, 15.0), 'claude-haiku-4': (0.8, 4.0)}, 'google': {'gemini-1.5-pro': (1.25, 5.0), 'gemini-1.5-flash': (0.075, 0.3), 'gemini-2.0-flash': (0.1, 0.4), 'gemini-2.5-pro': (1.25, 5.0), 'gemini-2.5-flash': (0.1, 0.4)}, 'openrouter': {'default': (1.0, 3.0)}, 'bedrock': {'anthropic.claude-3-5-sonnet': (3.0, 15.0), 'anthropic.claude-3-haiku': (0.25, 1.25), 'amazon.titan-text-express': (0.2, 0.6), 'meta.llama3-70b-instruct': (2.65, 3.5)}}

PhoenixEmitter

Python
PhoenixEmitter(*, endpoint: str | None = None, project: str | None = None)

Adapter for Arize Phoenix (open source LLM observability).

Phoenix consumes OpenTelemetry spans, so this emitter wraps an OTel tracer and tags spans with OpenInference semantic conventions.

Lazy import: install via pip install 'apogee-ai-observability[phoenix]'.

Source code in apogee_ai_observability/infrastructure/emitters/phoenix_emitter.py
Python
def __init__(self, *, endpoint: str | None = None, project: str | None = None) -> None:
    try:
        import opentelemetry  # type: ignore  # noqa: F401
    except ImportError as exc:
        raise ImportError(
            "PhoenixEmitter requires opentelemetry-api/sdk and arize-phoenix. "
            "Install with: pip install 'apogee-ai-observability[phoenix]'"
        ) from exc
    try:
        from phoenix.otel import register  # type: ignore
    except ImportError as exc:  # pragma: no cover
        raise ImportError(
            "phoenix-otel is required: pip install 'apogee-ai-observability[phoenix]'"
        ) from exc
    self._tracer_provider = register(
        project_name=project or "apogee-ai",
        endpoint=endpoint,
    )
    self._tracer = self._tracer_provider.get_tracer("apogee-ai-observability")

name class-attribute instance-attribute

Python
name = 'phoenix'

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/infrastructure/emitters/phoenix_emitter.py
Python
async def emit_span(self, span: Span) -> None:
    try:
        with self._tracer.start_as_current_span(span.name) as ot_span:
            for key, value in span.attributes.items():
                ot_span.set_attribute(key, _coerce(value))
            if span.usage is not None:
                ot_span.set_attribute("llm.token_count.prompt", span.usage.prompt_tokens)
                ot_span.set_attribute("llm.token_count.completion", span.usage.completion_tokens)
                ot_span.set_attribute("llm.token_count.total", span.usage.total_tokens)
            if span.error:
                ot_span.record_exception(Exception(span.error))
    except Exception as exc:  # noqa: BLE001
        raise EmitterFailureException(self.name, str(exc)) from exc

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/infrastructure/emitters/phoenix_emitter.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    return None

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/infrastructure/emitters/phoenix_emitter.py
Python
async def shutdown(self) -> None:
    try:
        self._tracer_provider.shutdown()
    except Exception:  # noqa: BLE001
        return

SpanNode dataclass

Python
SpanNode(span: Span, children: list[SpanNode] = list())

In-memory tree node — useful for inspecting causality.

span instance-attribute

Python
span: Span

children class-attribute instance-attribute

Python
children: list[SpanNode] = field(default_factory=list)

StructuredJsonEmitter

Python
StructuredJsonEmitter(*, stream=None)

NDJSON emitter for log aggregators (Loki, OpenSearch, Datadog logs).

Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
def __init__(self, *, stream=None) -> None:
    self._stream = stream or sys.stdout

name class-attribute instance-attribute

Python
name = 'structured_json'

emit_span async

Python
emit_span(span: Span) -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def emit_span(self, span: Span) -> None:
    record = {
        "kind": "span",
        "trace_id": span.trace_id,
        "span_id": span.span_id,
        "parent_span_id": span.parent_span_id,
        "name": span.name,
        "type": span.kind.value,
        "status": span.status.value,
        "start_time": span.start_time.isoformat(),
        "end_time": span.end_time.isoformat() if span.end_time else None,
        "duration_ms": span.duration_ms,
        "attributes": dict(span.attributes),
        "events": list(span.events),
        "error": span.error,
    }
    if span.usage is not None:
        record["usage"] = {
            "provider": span.usage.provider,
            "model": span.usage.model,
            "prompt_tokens": span.usage.prompt_tokens,
            "completion_tokens": span.usage.completion_tokens,
            "total_tokens": span.usage.total_tokens,
        }
    self._stream.write(json.dumps(record, default=str) + "\n")
    self._stream.flush()

emit_metric async

Python
emit_metric(sample: MetricSample) -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def emit_metric(self, sample: MetricSample) -> None:
    record = {
        "kind": "metric",
        "name": sample.name,
        "value": sample.value,
        "unit": sample.unit,
        "timestamp": sample.timestamp.isoformat(),
        "attributes": dict(sample.attributes),
    }
    self._stream.write(json.dumps(record, default=str) + "\n")
    self._stream.flush()

shutdown async

Python
shutdown() -> None
Source code in apogee_ai_observability/infrastructure/emitters/builtin_emitters.py
Python
async def shutdown(self) -> None:
    return None

TableCostCalculator

Python
TableCostCalculator(pricing: IPricingProvider | None = None)
Source code in apogee_ai_observability/infrastructure/cost/table_cost_calculator.py
Python
def __init__(self, pricing: IPricingProvider | None = None) -> None:
    self._pricing = pricing if pricing is not None else TablePricingProvider()

name class-attribute instance-attribute

Python
name = 'table'

estimate

Python
estimate(usage: LLMUsage, *, trace_id: str = '', span_id: str | None = None, user_id: str | None = None, tenant_id: str | None = None) -> CostLedgerEntry
Source code in apogee_ai_observability/infrastructure/cost/table_cost_calculator.py
Python
def estimate(
    self,
    usage: LLMUsage,
    *,
    trace_id: str = "",
    span_id: str | None = None,
    user_id: str | None = None,
    tenant_id: str | None = None,
) -> CostLedgerEntry:
    price = self._pricing.lookup(usage.provider, usage.model)
    input_per_m, output_per_m = price if price is not None else (0.0, 0.0)
    input_cost = (usage.prompt_tokens / 1_000_000.0) * input_per_m
    output_cost = (usage.completion_tokens / 1_000_000.0) * output_per_m
    return CostLedgerEntry(
        trace_id=trace_id,
        span_id=span_id,
        provider=usage.provider,
        model=usage.model,
        input_tokens=usage.prompt_tokens,
        output_tokens=usage.completion_tokens,
        input_cost_usd=round(input_cost, 6),
        output_cost_usd=round(output_cost, 6),
        user_id=user_id,
        tenant_id=tenant_id,
    )

TablePricingProvider

Python
TablePricingProvider(pricing: dict[str, dict[str, tuple[float, float]]] | None = None)
Source code in apogee_ai_observability/infrastructure/cost/pricing_table.py
Python
def __init__(
    self,
    pricing: dict[str, dict[str, tuple[float, float]]] | None = None,
) -> None:
    self._pricing = pricing if pricing is not None else PRICING_TABLE

name class-attribute instance-attribute

Python
name = 'table'

lookup

Python
lookup(provider: str, model: str) -> tuple[float, float] | None
Source code in apogee_ai_observability/infrastructure/cost/pricing_table.py
Python
def lookup(self, provider: str, model: str) -> tuple[float, float] | None:
    p = provider.lower()
    if p not in self._pricing:
        return None
    catalog = self._pricing[p]
    m = model.lower()
    if m in catalog:
        return catalog[m]
    for key in catalog:
        if key in m:
            return catalog[key]
    if "default" in catalog:
        return catalog["default"]
    return None

TraceContextScope

Python
TraceContextScope(ctx: TraceContext)

async with TraceContextScope(ctx): — restores previous context on exit.

Source code in apogee_ai_observability/infrastructure/context/active_context.py
Python
def __init__(self, ctx: TraceContext) -> None:
    self._ctx = ctx
    self._token = None

TraceReplayer

Python
TraceReplayer(emitter: ITelemetryEmitter)

Re-emits a previously stored trace through any emitter.

Useful for debugging: load a production trace into an in-memory or Phoenix instance for visual inspection without re-running the agent.

Source code in apogee_ai_observability/infrastructure/replay/trace_replayer.py
Python
def __init__(self, emitter: ITelemetryEmitter) -> None:
    self._emitter = emitter

replay async

Python
replay(trace: Trace) -> int

Re-emit every span. Returns the number of spans emitted.

Source code in apogee_ai_observability/infrastructure/replay/trace_replayer.py
Python
async def replay(self, trace: Trace) -> int:
    """Re-emit every span. Returns the number of spans emitted."""
    spans_in_order = self._topological_order(trace)
    for span in spans_in_order:
        await self._emitter.emit_span(deepcopy(span))
    return len(spans_in_order)

build_tree

Python
build_tree(trace: Trace) -> SpanNode | None

Reconstruct the parent/child tree from flat span list.

Source code in apogee_ai_observability/infrastructure/replay/trace_replayer.py
Python
def build_tree(self, trace: Trace) -> SpanNode | None:
    """Reconstruct the parent/child tree from flat span list."""
    nodes: dict[str, SpanNode] = {s.span_id: SpanNode(span=s) for s in trace.spans}
    root: SpanNode | None = None
    for s in trace.spans:
        node = nodes[s.span_id]
        if s.parent_span_id is None:
            root = node
            continue
        parent = nodes.get(s.parent_span_id)
        if parent is None:
            # Orphan; treat as root
            root = root or node
            continue
        parent.children.append(node)
    if root is None and nodes:
        root = next(iter(nodes.values()))
    return root

get_active

Python
get_active() -> TraceContext | None
Source code in apogee_ai_observability/infrastructure/context/active_context.py
Python
def get_active() -> TraceContext | None:
    return _active.get()

reset_active

Python
reset_active(token) -> None
Source code in apogee_ai_observability/infrastructure/context/active_context.py
Python
def reset_active(token) -> None:  # noqa: ANN001 - opaque ContextVar token
    _active.reset(token)

set_active

Python
set_active(ctx: TraceContext | None)
Source code in apogee_ai_observability/infrastructure/context/active_context.py
Python
def set_active(ctx: TraceContext | None):
    return _active.set(ctx)