Skip to content

symfonic.agent.message_usage

message_usage

Per-message usage: what the provider reported for one round (CORE-21).

AgentResult.usage is the run's aggregate. This is the per-round figure that sits on the assistant message the round produced, so a caller can see which round spent what -- and so the legacy projection can hand adopters the same per-message usage_metadata the legacy body returns.

Three rules keep it truthful:

  • Absent is None, not 0. A provider that reported nothing leaves Message.usage None; a dimension it did not report stays None here. A reported 0 stays 0. UsageDelta and TokenUsage cannot tell the two apart, which is why this type does not reuse them.
  • The provider's dict survives verbatim. reported is a deep copy of LangChain's usage_metadata, so provider detail this type has no typed field for (audio, the 5-minute ephemeral split, whatever comes next) is carried rather than dropped.
  • Only a round's own usage. Usage spent without closing a message -- an empty terminal synthesis -- reaches the aggregate and no message. The per-message sum may therefore be less than AgentResult.usage; it is never more, and no round is credited with another's tokens.

MessageUsage

Bases: BaseModel

One round's provider-reported usage; None fields were not reported.

The cache counts are inside input_tokens exactly as the provider reports them -- see UsageDelta -- so a consumer must not add them on top.

from_reported classmethod 

from_reported(reported: Any) -> MessageUsage | None

Project LangChain usage_metadata; None when nothing was reported.

Source code in src/symfonic/agent/message_usage.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@classmethod
def from_reported(cls, reported: Any) -> MessageUsage | None:
    """Project LangChain ``usage_metadata``; ``None`` when nothing was reported."""
    if not isinstance(reported, Mapping) or not reported:
        return None
    verbatim = deepcopy(dict(reported))
    inputs = _details(verbatim, "input_token_details")
    outputs = _details(verbatim, "output_token_details")
    reasoning = _count(verbatim, "reasoning_tokens")
    return cls(
        input_tokens=_count(verbatim, "input_tokens"),
        output_tokens=_count(verbatim, "output_tokens"),
        total_tokens=_count(verbatim, "total_tokens"),
        cache_read_tokens=_count(inputs, "cache_read"),
        cache_creation_tokens=_count(inputs, "cache_creation"),
        # Top-level first, as ``extract_usage`` reads it: a provider that
        # bills reasoning separately reports it there.
        reasoning_tokens=reasoning if reasoning is not None else _count(outputs, "reasoning"),
        cache_ttl=extract_cache_ttl(inputs),
        reported=verbatim,
    )

of classmethod 

of(response: Any) -> MessageUsage | None

The usage a provider response (AIMessage or chunk) carries.

Source code in src/symfonic/agent/message_usage.py 
81
82
83
84
@classmethod
def of(cls, response: Any) -> MessageUsage | None:
    """The usage a provider response (``AIMessage`` or chunk) carries."""
    return cls.from_reported(getattr(response, "usage_metadata", None))