haive.core.schema.prebuilt.messages.token_usage¶

Token usage tracking schema for LLM interactions.

This module provides schemas for tracking token usage, costs, and capacity across different LLM providers and models. It supports comprehensive metrics including cached tokens, audio tokens, and reasoning tokens.

Classes¶

TokenUsage

Comprehensive token usage tracking with cost calculation.

Functions¶

`aggregate_token_usage`(messages)	Aggregate token usage across multiple messages.
`calculate_token_cost`(usage, input_cost_per_1k, ...[, ...])	Calculate costs based on token usage and pricing.
`extract_token_usage_from_message`(message[, provider])	Extract token usage information from a message.

Module Contents¶

class haive.core.schema.prebuilt.messages.token_usage.TokenUsage(/, **data)[source]¶

Bases: pydantic.BaseModel

Comprehensive token usage tracking with cost calculation.

This class tracks all aspects of token usage including: - Input/output/total tokens - Cached tokens (for providers that support caching) - Audio tokens (for multimodal models) - Reasoning tokens (for models with explicit reasoning steps) - Cost calculation based on provider pricing - Capacity percentage for context window management

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

add(other)[source]¶

Add two TokenUsage instances together.

Parameters:: other (TokenUsage)
Return type:: TokenUsage

validate_totals()[source]¶

Ensure total_tokens and total_cost are calculated if not set.

Return type:: Self

haive.core.schema.prebuilt.messages.token_usage.aggregate_token_usage(messages)[source]¶

Aggregate token usage across multiple messages.

Parameters:: messages (list[langchain_core.messages.BaseMessage]) – List of messages to aggregate usage from
Returns:: Combined TokenUsage instance
Return type:: TokenUsage

haive.core.schema.prebuilt.messages.token_usage.calculate_token_cost(usage, input_cost_per_1k, output_cost_per_1k, cached_input_cost_per_1k=None)[source]¶

Calculate costs based on token usage and pricing.

Parameters:

usage (TokenUsage) – TokenUsage instance to calculate costs for
input_cost_per_1k (float) – Cost per 1000 input tokens
output_cost_per_1k (float) – Cost per 1000 output tokens
cached_input_cost_per_1k (float | None) – Optional cost per 1000 cached input tokens

Returns:

New TokenUsage instance with calculated costs

Return type:

TokenUsage

haive.core.schema.prebuilt.messages.token_usage.extract_token_usage_from_message(message, provider=None)[source]¶

Extract token usage information from a message.

Parameters:

message (langchain_core.messages.BaseMessage) – The message to extract usage from
provider (str | None) – Optional provider name for provider-specific handling

Returns:

TokenUsage instance if usage info found, None otherwise

Return type:

TokenUsage | None