symfonic.core.usage

usage

Canonical token-usage extraction from AIMessage responses (v8.6.8).

Single source of truth for reading LangChain's usage_metadata off an AIMessage-like response. Consolidates three previously-duplicated readouts (DRY):

• callbacks/emit.py:_extract_usage -- the adopter-VALIDATED cost path, carrying real production cost numbers (v7.14.2).
• nodes/react.py:_extract_usage -- the original (emit was hoisted from it); differed only on the object-style fallback branch.
• streaming/transpiler.py -- an inline input/output-only readout with no cache handling (the v8.6.7 getattr-on-dict bug lived here, in the divergence this module closes).

This function reproduces emit's validated logic byte-for-byte: the dict path real providers (Anthropic, OpenAI) exercise is unchanged, and the object-style fallback adopts emit's attribute read (the canonical/validated behaviour) rather than react's zero-return.

The module depends on the standard library and nothing else, so all three call sites can import it without an import cycle.

TA2.3 (LAY-ADR): this module is ruled kernel-contracts, because the memory capability's three LLM call sites need the same validated readout and capability -> facade-compiler is no with no port escape. The purity that ruling requires is why :func:extract_cache_ttl now lives here instead of on AnthropicProvider -- see its docstring for what that costs.

extract_cache_ttl

extract_cache_ttl(input_token_details: Any) -> str | None

Mine the Anthropic 1h-cache TTL from a LangChain usage_metadata.input_token_details dict.

Anthropic's API surfaces the prompt-cache write rate via the cache_creation: {ephemeral_5m_input_tokens, ephemeral_1h_input_tokens} sub-object on the response usage block. langchain-anthropic (>= 0.3.20) flattens that into input_token_details with the same key names -- see langchain_anthropic.chat_models:2600-2610.

When ephemeral_1h_input_tokens > 0 (the 1h cache was used on this turn), this returns "1h" so the cost layer picks the cache_write_1h rate from the pricing row (T-7.20.0.5). Otherwise -- pure cache_read, only 5m cache, or no cache_creation at all -- returns None so the cost layer falls through to cache_write (the 5-min rate, which IS the documented default in the registry).

Where this lives, and why it moved (TA2.3). The 2026-06-05 architect verdict SS5.c placed this read on AnthropicProvider -- "wire-shape translation lives on the provider that owns the wire contract" -- so that framework-level extractors stayed free of Anthropic key names. Under LAY-ADR that placement is unreachable: symfonic.memory is ruled capability, its three LLM call sites need :func:extract_usage, and capability -> facade-compiler is no with no port cell. The read therefore moved to this kernel-contracts module, which is the same resolution TA2.2 applied to the provider-specific pricing table (core/observability/pricing.py).

The cost is stated rather than hidden: an Anthropic wire key now appears in a kernel-contracts module, so this extractor is no longer provider-agnostic in the sense the verdict intended. What survives is the verdict's interface: AnthropicProvider.extract_cache_ttl is unchanged in name, signature and behaviour and remains the declared provider-side entry point -- it delegates here, so there is exactly one implementation.

Parameters:

Name	Type	Description	Default
input_token_details	Any	The input_token_details sub-dict of a LangChain usage_metadata payload. May be None (no cache info), an empty dict, or any non-dict value (defensive -- never raises).	required

Returns:

Type	Description
str | None	"1h" when the 1h cache was used this turn. None otherwise
str | None	(5m cache only, no cache, or non-dict input).

Source code in symfonic/core/usage.py

def extract_cache_ttl(input_token_details: Any) -> str | None:
    """Mine the Anthropic 1h-cache TTL from a LangChain
    ``usage_metadata.input_token_details`` dict.

    Anthropic's API surfaces the prompt-cache write rate via the
    ``cache_creation: {ephemeral_5m_input_tokens, ephemeral_1h_input_tokens}``
    sub-object on the response usage block.  langchain-anthropic
    (>= 0.3.20) flattens that into ``input_token_details`` with the same key
    names -- see ``langchain_anthropic.chat_models:2600-2610``.

    When ``ephemeral_1h_input_tokens > 0`` (the 1h cache was used on this
    turn), this returns ``"1h"`` so the cost layer picks the
    ``cache_write_1h`` rate from the pricing row (T-7.20.0.5).  Otherwise --
    pure cache_read, only 5m cache, or no cache_creation at all -- returns
    ``None`` so the cost layer falls through to ``cache_write`` (the 5-min
    rate, which IS the documented default in the registry).

    **Where this lives, and why it moved (TA2.3).**  The 2026-06-05 architect
    verdict SS5.c placed this read on ``AnthropicProvider`` -- "wire-shape
    translation lives on the provider that owns the wire contract" -- so that
    framework-level extractors stayed free of Anthropic key names.  Under
    LAY-ADR that placement is unreachable: ``symfonic.memory`` is ruled
    ``capability``, its three LLM call sites need :func:`extract_usage`, and
    ``capability -> facade-compiler`` is ``no`` with no port cell.  The read
    therefore moved to this ``kernel-contracts`` module, which is the same
    resolution TA2.2 applied to the provider-specific pricing table
    (``core/observability/pricing.py``).

    The cost is stated rather than hidden: an Anthropic wire key now appears
    in a kernel-contracts module, so this extractor is no longer
    provider-agnostic in the sense the verdict intended.  What survives is the
    verdict's *interface*: ``AnthropicProvider.extract_cache_ttl`` is
    unchanged in name, signature and behaviour and remains the declared
    provider-side entry point -- it delegates here, so there is exactly one
    implementation.

    Args:
        input_token_details: The ``input_token_details`` sub-dict of a
            LangChain ``usage_metadata`` payload.  May be ``None`` (no cache
            info), an empty dict, or any non-dict value (defensive -- never
            raises).

    Returns:
        ``"1h"`` when the 1h cache was used this turn.  ``None`` otherwise
        (5m cache only, no cache, or non-dict input).
    """
    if not isinstance(input_token_details, dict):
        return None
    # ``or 0`` guards against the field being present but ``None``
    # (older langchain-anthropic surfaced a None when 1h cache was
    # absent rather than omitting the key).
    ephemeral_1h = input_token_details.get("ephemeral_1h_input_tokens") or 0
    if ephemeral_1h > 0:
        return "1h"
    return None

extract_usage

extract_usage(message: Any) -> dict[str, Any]

Pull token usage from an AIMessage-like response.

Canonical extractor for ALL LLM call sites (react node, the seven non-react bypass sites, and the streaming transpiler). Validated against real adopter production cost numbers (hoisted from the v7.14.2 callbacks/emit.py implementation).

LangChain stores usage as usage_metadata on AIMessage -- a UsageMetadata TypedDict that is a plain dict at runtime. Anthropic cache tokens are NESTED under input_token_details.{cache_read, cache_creation} -- a flat {k: int(v) for k, v in raw.items()} walk drops them (the nested dict fails the isinstance(v, int) type filter). The dict shape returned here matches what TokenUsage.from_dict consumes (cache_read_input_tokens / cache_creation_input_tokens, see contracts/callbacks.py).

Parameters:

Name	Type	Description	Default
message	Any	An AIMessage-like object. Read non-destructively via getattr(message, "usage_metadata", None) -- a missing attribute degrades to the zero-default shape rather than raising.	required

Returns:

Type	Description
dict[str, Any]	A mixed-value mapping, hence dict[str, Any]: every token count
dict[str, Any]	is an int, but cache_ttl is a str ("1h"). The earlier
dict[str, Any]	dict[str, int] annotation described the majority of the keys
dict[str, Any]	rather than the return type, so every caller threading cache_ttl
dict[str, Any]	onwards inherited an int where a str was required — a type
dict[str, Any]	error at the caller for a value this function had always produced
dict[str, Any]	correctly.
dict[str, Any]	Dict with input_tokens and output_tokens (always present,
dict[str, Any]	defaulting to 0). When the provider surfaces Anthropic cache
dict[str, Any]	tokens, additionally includes cache_read_input_tokens and
dict[str, Any]	cache_creation_input_tokens so TokenUsage.from_dict can
dict[str, Any]	materialise the typed breakdown and the pricing layer can compute
dict[str, Any]	the discounted cost. When the Anthropic 1h cache TTL is in play,
dict[str, Any]	a cache_ttl string is included. Extended-thinking reasoning
dict[str, Any]	tokens (Opus-4-x and forward-compat for Opus-5+) are surfaced as a
dict[str, Any]	flat reasoning_tokens key.

Source code in symfonic/core/usage.py

def extract_usage(message: Any) -> dict[str, Any]:
    """Pull token usage from an AIMessage-like response.

    Canonical extractor for ALL LLM call sites (react node, the seven
    non-react bypass sites, and the streaming transpiler).  Validated
    against real adopter production cost numbers (hoisted from the v7.14.2
    ``callbacks/emit.py`` implementation).

    LangChain stores usage as ``usage_metadata`` on AIMessage -- a
    ``UsageMetadata`` ``TypedDict`` that is a plain ``dict`` at runtime.
    Anthropic cache tokens are NESTED under
    ``input_token_details.{cache_read, cache_creation}`` -- a flat
    ``{k: int(v) for k, v in raw.items()}`` walk drops them (the nested
    dict fails the ``isinstance(v, int)`` type filter).  The dict shape
    returned here matches what ``TokenUsage.from_dict`` consumes
    (``cache_read_input_tokens`` / ``cache_creation_input_tokens``,
    see ``contracts/callbacks.py``).

    Args:
        message: An ``AIMessage``-like object.  Read non-destructively via
            ``getattr(message, "usage_metadata", None)`` -- a missing
            attribute degrades to the zero-default shape rather than
            raising.

    Returns:
        A *mixed-value* mapping, hence ``dict[str, Any]``: every token count
        is an ``int``, but ``cache_ttl`` is a ``str`` (``"1h"``). The earlier
        ``dict[str, int]`` annotation described the majority of the keys
        rather than the return type, so every caller threading ``cache_ttl``
        onwards inherited an ``int`` where a ``str`` was required — a type
        error at the caller for a value this function had always produced
        correctly.

        Dict with ``input_tokens`` and ``output_tokens`` (always present,
        defaulting to 0).  When the provider surfaces Anthropic cache
        tokens, additionally includes ``cache_read_input_tokens`` and
        ``cache_creation_input_tokens`` so ``TokenUsage.from_dict`` can
        materialise the typed breakdown and the pricing layer can compute
        the discounted cost.  When the Anthropic 1h cache TTL is in play,
        a ``cache_ttl`` string is included.  Extended-thinking reasoning
        tokens (Opus-4-x and forward-compat for Opus-5+) are surfaced as a
        flat ``reasoning_tokens`` key.
    """
    meta = getattr(message, "usage_metadata", None)
    if isinstance(meta, dict):
        usage: dict[str, Any] = {
            "input_tokens": int(meta.get("input_tokens") or 0),
            "output_tokens": int(meta.get("output_tokens") or 0),
        }
        # Anthropic cache tokens live in a nested input_token_details
        # dict.  A flat-dict walk silently drops them (which was the bug
        # in the original adopter-proposed patch and the pre-v7.14.2
        # behaviour at the emit site).
        details = meta.get("input_token_details")
        if isinstance(details, dict):
            cache_read = details.get("cache_read", 0)
            cache_creation = details.get("cache_creation", 0)
            if cache_read:
                usage["cache_read_input_tokens"] = int(cache_read)
            if cache_creation:
                usage["cache_creation_input_tokens"] = int(cache_creation)
            # v7.20.0 T-7.20.0.6: surface the Anthropic 1h-cache TTL.
            # Returns ``"1h"`` only when ``ephemeral_1h_input_tokens > 0``
            # on this turn; otherwise None and the field is omitted (the
            # cost layer's ``cache_ttl is None`` branch picks the 5-min
            # ``cache_write`` rate, the documented default).
            cache_ttl = extract_cache_ttl(details)
            if cache_ttl is not None:
                usage["cache_ttl"] = cache_ttl
        # v7.15.0: extended-thinking reasoning tokens live under
        # ``output_token_details.reasoning`` on Opus-4-x.  Top-level
        # ``reasoning_tokens`` is checked as forward-compat for Opus-5+
        # separated billing.  Surfaced as a flat ``reasoning_tokens``
        # key in the returned dict so ``TokenUsage.from_dict`` and
        # adopter bridges have a uniform read path regardless of
        # provider shape.
        reasoning = meta.get("reasoning_tokens", 0)
        if not reasoning:
            out_details = meta.get("output_token_details")
            if isinstance(out_details, dict):
                reasoning = out_details.get("reasoning", 0)
        if reasoning:
            usage["reasoning_tokens"] = int(reasoning)
        return usage

    # Object-style fallback: some adapters return a UsageMetadata-like
    # object instead of a dict.  The pre-v7.14.2 emit helper already
    # handled this branch; preserve back-compat.  Cache fields aren't
    # surfaced here because no known object-style adapter exposes them
    # -- if one ever does we add the getattr branch in a follow-up.
    return {
        "input_tokens": int(getattr(meta, "input_tokens", 0) or 0),
        "output_tokens": int(getattr(meta, "output_tokens", 0) or 0),
    }