symfonic.core.observability.metrics

metrics

ConversationMetricsCollector — aggregates token/cost per conversation.

Implements the CallbackHandler protocol so it can be registered directly with SymfonicAgent(metrics_collector=...) or added to any CallbackManager.

Sprint 4 memory safety: bounded call_records deque, LRU-capped _conversations, run-scoped map purge on on_agent_end, and observable budget-tracker fan-out errors.

ConversationMetricsCollector

ConversationMetricsCollector(metrics_sink: object = None)

Bases: MetricsExecutionMixin, MetricsQueryMixin

Aggregates token usage and cost per conversation (session_id).

Satisfies the CallbackHandler protocol via structural typing.

Usage::

collector = ConversationMetricsCollector()
agent = SymfonicAgent(..., metrics_collector=collector)
# after some runs:
record = collector.get_conversation(session_id)
all_records = collector.list_conversations()

Source code in symfonic/core/observability/metrics.py

def __init__(self, metrics_sink: object = None) -> None:
    self._conversations: dict[str, ConversationRecord] = {}
    self._metrics_sink = metrics_sink
    # run_id → most-recent TokenCompositionEvent
    self._current_composition: dict[str, TokenCompositionEvent] = {}
    # run_id → conversation_id
    self._current_conversation: dict[str, str] = {}
    # run_id → tenant_id (required for TokenBudgetTracker fan-out)
    self._current_tenant: dict[str, str] = {}
    # run_id -> (root_run_id, immediate parent_run_id)
    self._current_lineage: dict[str, tuple[str, str | None]] = {}
    self._execution_events: deque[dict[str, object]] = deque(
        maxlen=MAX_CALL_RECORDS_PER_CONVERSATION * MAX_CONVERSATIONS
    )
    # Only rows not yet handed to the sink, indexed by their run.  The
    # retained deque serves reads; scanning its global history on every
    # live event made request cost grow with the worker's lifetime.
    self._pending_execution_events: dict[str, list[dict[str, object]]] = defaultdict(list)
    self._execution_sequence: dict[str, int] = defaultdict(int)

on_agent_end async

on_agent_end(event: AgentEndEvent) -> None

Increment turn count and purge run-scoped bookkeeping.

Source code in symfonic/core/observability/metrics.py

async def on_agent_end(self, event: AgentEndEvent) -> None:
    """Increment turn count and purge run-scoped bookkeeping."""
    conv_id = self._current_conversation.get(event.run_id, event.run_id)
    if conv_id in self._conversations:
        self._conversations[conv_id].turns += 1
    self._current_composition.pop(event.run_id, None)
    self._current_conversation.pop(event.run_id, None)
    self._current_tenant.pop(event.run_id, None)
    self._current_lineage.pop(event.run_id, None)
    self._execution_sequence.pop(event.run_id, None)
    self._pending_execution_events.pop(event.run_id, None)

on_llm_end async

on_llm_end(event: LLMEndEvent) -> None

Accumulate token/cost metrics into the matching conversation record.

Also fans the same usage out to the process-wide :class:TokenBudgetTracker (if one has been registered via set_budget_tracker) so per-tenant ceilings can fire on the NEXT call. Failures in the tracker must never break the LLM flow.

Source code in symfonic/core/observability/metrics.py

async def on_llm_end(self, event: LLMEndEvent) -> None:
    """Accumulate token/cost metrics into the matching conversation record.

    Also fans the same usage out to the process-wide
    :class:`TokenBudgetTracker` (if one has been registered via
    ``set_budget_tracker``) so per-tenant ceilings can fire on the
    NEXT call.  Failures in the tracker must never break the LLM flow.
    """
    usage = TokenUsage.from_dict(event.usage, event.model)
    conv_id = self._current_conversation.get(event.run_id, event.run_id)
    tenant_id = self._current_tenant.get(event.run_id, "")

    conv = self._conversations.setdefault(
        conv_id,
        ConversationRecord(conversation_id=conv_id, tenant_id=tenant_id),
    )
    self._evict_conversations_if_over_cap()
    # Back-fill tenant_id if the record pre-dates set_tenant() being called.
    if tenant_id and not conv.tenant_id:
        conv.tenant_id = tenant_id
    conv.llm_calls += 1
    conv.input_tokens += usage.input_tokens
    conv.output_tokens += usage.output_tokens
    conv.cached_tokens += usage.cached_tokens
    conv.cost_usd = round(conv.cost_usd + usage.cost_usd, 6)
    conv.last_seen = datetime.now(UTC)

    # Fan usage into the budget tracker (optional / best-effort).
    # Budget tracking must never break the LLM flow, but failures
    # MUST be observable -- silent suppression previously hid real
    # tracker regressions for weeks.
    from symfonic.core.observability.budget import get_budget_tracker
    tracker = get_budget_tracker()
    if tracker is not None and conv.tenant_id:
        try:
            await tracker.record(
                conv.tenant_id,
                event.model,
                {
                    "input_tokens": usage.input_tokens,
                    "output_tokens": usage.output_tokens,
                    "cached_tokens": usage.cached_tokens,
                    "cache_write_tokens": getattr(
                        usage, "cache_creation_tokens", 0,
                    ),
                },
            )
        except Exception as exc:
            logger.warning(
                "budget tracker fan-out failed: %s", exc, exc_info=True,
            )

    comp = self._current_composition.get(event.run_id)
    root_run_id, parent_run_id = self._current_lineage.get(
        event.run_id, (event.run_id, None)
    )
    conv.call_records.append(
        LLMCallRecord(
            call_id=event.run_id,
            conversation_id=conv_id,
            model=event.model,
            # Real call-site discriminator emitted since v7.4.3.  Fall
            # back to "react" only when the emitter left it empty (pre-
            # v7.4.3 payloads) so the historical default is preserved.
            node=event.node_name or "react",
            input_tokens=usage.input_tokens,
            cached_tokens=usage.cached_tokens,
            output_tokens=usage.output_tokens,
            cost_usd=usage.cost_usd,
            # iteration 0 is the turn's opening call; later React hops
            # are tool follow-ups.  The event carries no explicit
            # call_type, so derive it from the iteration index.
            call_type="initial" if event.iteration_index == 0 else "tool_followup",
            timestamp=datetime.now(UTC),
            composition={
                "system_prompt": comp.system_prompt_tokens if comp else 0,
                "tool_definitions": comp.tool_definitions_tokens if comp else 0,
                "conversation_history": comp.conversation_history_tokens if comp else 0,
                "memory_context": comp.memory_context_tokens if comp else 0,
            },
            cache_creation_tokens=usage.cache_creation_tokens,
            reasoning_tokens=usage.reasoning_tokens,
            duration_ms=event.duration_ms,
            iteration_index=event.iteration_index,
            turn_index=event.turn_index,
            root_run_id=root_run_id,
            parent_run_id=parent_run_id,
        )
    )
    self.record_model_round(event, conv_id, tenant_id)
    self._persist_execution_events(event.run_id)

    # v8.8.0 — fan the same per-call detail into the durable metrics
    # store, if one is registered, so the admin dashboards survive a
    # worker restart.  Best-effort + synchronous O(1) enqueue (the
    # BufferedMetricsSink batches the actual DB I/O off the hot path);
    # a sink failure must NEVER break the LLM flow.
    sink = self._metrics_sink
    if sink is None:
        from symfonic.core.observability.metrics_store import get_metrics_store

        sink = get_metrics_store()
    if sink is not None:
        try:
            sink.record(_record_to_snapshot(conv.call_records[-1], conv.tenant_id))
        except Exception as exc:  # noqa: BLE001 - telemetry never propagates
            logger.warning("metrics sink fan-out failed: %s", exc, exc_info=True)

on_token_composition async

on_token_composition(event: TokenCompositionEvent) -> None

Cache the composition breakdown keyed by run_id.

Source code in symfonic/core/observability/metrics.py

async def on_token_composition(self, event: TokenCompositionEvent) -> None:
    """Cache the composition breakdown keyed by run_id."""
    self._current_composition[event.run_id] = event

set_conversation

set_conversation(run_id: str, conversation_id: str) -> None

Associate a run_id with a conversation (session) id.

Call this before the agent run so metrics land under the right conversation bucket (e.g. use the session_id as conversation_id).

Source code in symfonic/core/observability/metrics.py

def set_conversation(self, run_id: str, conversation_id: str) -> None:
    """Associate a run_id with a conversation (session) id.

    Call this before the agent run so metrics land under the right
    conversation bucket (e.g. use the session_id as conversation_id).
    """
    self._current_conversation[run_id] = conversation_id
    for row in self._pending_execution_events.get(run_id, ()):
        if not row["conversation_id"]:
            row["conversation_id"] = conversation_id
    self._persist_execution_events(run_id)

set_lineage

set_lineage(run_id: str, *, root_run_id: str, parent_run_id: str | None = None) -> None

Associate one execution with its delegation root and parent.

Source code in symfonic/core/observability/metrics.py

def set_lineage(
    self,
    run_id: str,
    *,
    root_run_id: str,
    parent_run_id: str | None = None,
) -> None:
    """Associate one execution with its delegation root and parent."""
    self._current_lineage[run_id] = (root_run_id or run_id, parent_run_id)

set_tenant

set_tenant(run_id: str, tenant_id: str) -> None

Associate a run_id with a tenant_id.

Call this before the agent run so on_llm_end can fan usage into the registered TokenBudgetTracker. Without this the tracker never fires because the ConversationRecord's tenant_id stays empty.

Source code in symfonic/core/observability/metrics.py

def set_tenant(self, run_id: str, tenant_id: str) -> None:
    """Associate a run_id with a tenant_id.

    Call this before the agent run so ``on_llm_end`` can fan usage
    into the registered ``TokenBudgetTracker``.  Without this the
    tracker never fires because the ConversationRecord's tenant_id
    stays empty.
    """
    if tenant_id:
        self._current_tenant[run_id] = tenant_id
        for row in self._pending_execution_events.get(run_id, ()):
            if not row["tenant_id"]:
                row["tenant_id"] = tenant_id
        self._persist_execution_events(run_id)

ConversationRecord dataclass

ConversationRecord(conversation_id: str, tenant_id: str, turns: int = 0, llm_calls: int = 0, input_tokens: int = 0, output_tokens: int = 0, cached_tokens: int = 0, cost_usd: float = 0.0, first_seen: datetime = (lambda: datetime.now(UTC))(), last_seen: datetime = (lambda: datetime.now(UTC))(), call_records: deque[LLMCallRecord] = (lambda: deque(maxlen=MAX_CALL_RECORDS_PER_CONVERSATION))())

Aggregated per-conversation metrics.

LLMCallRecord dataclass

LLMCallRecord(call_id: str, conversation_id: str, model: str, node: str, input_tokens: int, cached_tokens: int, output_tokens: int, cost_usd: float, call_type: str, timestamp: datetime, composition: dict[str, int] = dict(), cache_creation_tokens: int = 0, reasoning_tokens: int = 0, duration_ms: float = 0.0, iteration_index: int = 0, turn_index: int = 0, root_run_id: str = '', parent_run_id: str | None = None)

Single LLM call record for dashboard display.

v8.7.11 — the record now carries the full per-call telemetry the framework already emits on :class:LLMEndEvent (real node name, call_type, iteration_index, turn_index, duration_ms, and the cache_creation/reasoning token splits) instead of hardcoding node="react" / call_type="initial" and dropping latency. All new fields default to zero/empty so pre-v8.7.11 callers that construct the record positionally with the original arg set keep working unchanged.