Skip to content

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 Jarvio-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 only on :mod:symfonic.core.providers (for the Anthropic cache-TTL wire-shape owner) and the standard library, so all three call sites can import it without an import cycle.

extract_usage 

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

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 Jarvio 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, int]

Dict with input_tokens and output_tokens (always present,
dict[str, int]

defaulting to 0). When the provider surfaces Anthropic cache
dict[str, int]

tokens, additionally includes cache_read_input_tokens and
dict[str, int]

cache_creation_input_tokens so TokenUsage.from_dict can
dict[str, int]

materialise the typed breakdown and the pricing layer can compute
dict[str, int]

the discounted cost. When the Anthropic 1h cache TTL is in play,
dict[str, int]

a cache_ttl string is included. Extended-thinking reasoning
dict[str, int]

tokens (Opus-4-x and forward-compat for Opus-5+) are surfaced as a
dict[str, int]

flat reasoning_tokens key.
Source code in src/symfonic/core/usage.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def extract_usage(message: Any) -> dict[str, int]:
    """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 Jarvio 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:
        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, int] = {
            "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 Jarvio-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.
            # ``AnthropicProvider.extract_cache_ttl`` is the wire-shape
            # owner; we stay provider-agnostic by delegating.  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).
            from .providers import AnthropicProvider

            cache_ttl = AnthropicProvider.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),
    }