Skip to content

lpgcast.llm.usage

Token and call accounting.

Tokens a provider billed are kept apart from tokens replayed out of the response cache, so a cached rerun never reads as spend and a cost column still says what a fresh run would cost.

Classes

TokenUsage

Bases: BaseModel

Token counts for one call, as far as the provider reports them.

None means not reported, which is not zero.

Source code in lpgcast/llm/usage.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class TokenUsage(BaseModel):
    """Token counts for one call, as far as the provider reports them.

    ``None`` means *not reported*, which is not zero.
    """

    input_tokens: int | None = Field(default=None, description="Prompt tokens.")
    output_tokens: int | None = Field(default=None, description="Completion tokens.")
    reasoning_tokens: int | None = Field(
        default=None, description="Thinking tokens, counted inside output_tokens."
    )
    cache_read_input_tokens: int | None = Field(
        default=None,
        description=(
            "Prompt tokens served from the provider's prompt cache, counted "
            "inside input_tokens. Unrelated to LPGCast's response cache."
        ),
    )

    def is_empty(self) -> bool:
        """True when the provider reported nothing."""
        return all(value is None for value in self.model_dump().values())

    def __add__(self, other: TokenUsage) -> TokenUsage:
        def _sum(a: int | None, b: int | None) -> int | None:
            return None if a is None and b is None else (a or 0) + (b or 0)

        return TokenUsage(
            **{
                name: _sum(getattr(self, name), getattr(other, name))
                for name in TokenUsage.model_fields
            }
        )

Attributes

cache_read_input_tokens = Field(default=None, description="Prompt tokens served from the provider's prompt cache, counted inside input_tokens. Unrelated to LPGCast's response cache.") class-attribute instance-attribute
input_tokens = Field(default=None, description='Prompt tokens.') class-attribute instance-attribute
output_tokens = Field(default=None, description='Completion tokens.') class-attribute instance-attribute
reasoning_tokens = Field(default=None, description='Thinking tokens, counted inside output_tokens.') class-attribute instance-attribute

Methods:

__add__(other)
Source code in lpgcast/llm/usage.py 
41
42
43
44
45
46
47
48
49
50
def __add__(self, other: TokenUsage) -> TokenUsage:
    def _sum(a: int | None, b: int | None) -> int | None:
        return None if a is None and b is None else (a or 0) + (b or 0)

    return TokenUsage(
        **{
            name: _sum(getattr(self, name), getattr(other, name))
            for name in TokenUsage.model_fields
        }
    )
is_empty()

True when the provider reported nothing.

Source code in lpgcast/llm/usage.py 
37
38
39
def is_empty(self) -> bool:
    """True when the provider reported nothing."""
    return all(value is None for value in self.model_dump().values())

UsageTracker

Bases: BaseModel

Calls, events and tokens for a run or a scope within it.

Source code in lpgcast/llm/usage.py 
 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
122
123
124
125
126
127
128
129
130
class UsageTracker(BaseModel):
    """Calls, events and tokens for a run or a scope within it."""

    calls: int = Field(default=0, description="Provider calls that returned.")
    cache_hits: int = Field(default=0, description="Responses served from cache.")
    calls_failed: int = Field(default=0, description="Provider calls that raised.")
    timeouts: int = Field(default=0, description="Subset of calls_failed.")
    rate_limited: int = Field(default=0, description="Subset of calls_failed.")
    rejected: int = Field(
        default=0, description="Subset of calls_failed: configuration faults."
    )
    truncated: int = Field(
        default=0, description="Replies cut off at the output token limit."
    )
    refused: int = Field(default=0, description="Replies the model refused.")
    parse_retries: int = Field(default=0, description="Re-asks after bad JSON.")
    parse_abandoned: int = Field(default=0, description="Requests given up on.")
    json_repairs: int = Field(default=0, description="Bracket-kind repairs.")
    json_schema_fallbacks: int = Field(
        default=0, description="Switches from json_schema to json_object."
    )
    chars_sent: int = Field(default=0, description="Prompt characters sent.")
    chars_received: int = Field(default=0, description="Response characters.")
    billed: TokenUsage = Field(
        default_factory=TokenUsage, description="Tokens from live provider calls."
    )
    replayed: TokenUsage = Field(
        default_factory=TokenUsage, description="Tokens of responses served from cache."
    )

    def incr(self, counter: str, n: int = 1) -> None:
        """Add *n* to an integer counter."""
        setattr(self, counter, getattr(self, counter) + n)

    def add_call(self, chars_sent: int, chars_received: int, usage: TokenUsage) -> None:
        """Charge one live provider call."""
        self.calls += 1
        self.chars_sent += chars_sent
        self.chars_received += chars_received
        self.billed = self.billed + usage

    def add_hit(self, usage: TokenUsage | None) -> None:
        """Charge one cache hit."""
        self.cache_hits += 1
        if usage is not None:
            self.replayed = self.replayed + usage

Attributes

billed = Field(default_factory=TokenUsage, description='Tokens from live provider calls.') class-attribute instance-attribute
cache_hits = Field(default=0, description='Responses served from cache.') class-attribute instance-attribute
calls = Field(default=0, description='Provider calls that returned.') class-attribute instance-attribute
calls_failed = Field(default=0, description='Provider calls that raised.') class-attribute instance-attribute
chars_received = Field(default=0, description='Response characters.') class-attribute instance-attribute
chars_sent = Field(default=0, description='Prompt characters sent.') class-attribute instance-attribute
json_repairs = Field(default=0, description='Bracket-kind repairs.') class-attribute instance-attribute
json_schema_fallbacks = Field(default=0, description='Switches from json_schema to json_object.') class-attribute instance-attribute
parse_abandoned = Field(default=0, description='Requests given up on.') class-attribute instance-attribute
parse_retries = Field(default=0, description='Re-asks after bad JSON.') class-attribute instance-attribute
rate_limited = Field(default=0, description='Subset of calls_failed.') class-attribute instance-attribute
refused = Field(default=0, description='Replies the model refused.') class-attribute instance-attribute
rejected = Field(default=0, description='Subset of calls_failed: configuration faults.') class-attribute instance-attribute
replayed = Field(default_factory=TokenUsage, description='Tokens of responses served from cache.') class-attribute instance-attribute
timeouts = Field(default=0, description='Subset of calls_failed.') class-attribute instance-attribute
truncated = Field(default=0, description='Replies cut off at the output token limit.') class-attribute instance-attribute

Methods:

add_call(chars_sent, chars_received, usage)

Charge one live provider call.

Source code in lpgcast/llm/usage.py 
119
120
121
122
123
124
def add_call(self, chars_sent: int, chars_received: int, usage: TokenUsage) -> None:
    """Charge one live provider call."""
    self.calls += 1
    self.chars_sent += chars_sent
    self.chars_received += chars_received
    self.billed = self.billed + usage
add_hit(usage)

Charge one cache hit.

Source code in lpgcast/llm/usage.py 
126
127
128
129
130
def add_hit(self, usage: TokenUsage | None) -> None:
    """Charge one cache hit."""
    self.cache_hits += 1
    if usage is not None:
        self.replayed = self.replayed + usage
incr(counter, n=1)

Add n to an integer counter.

Source code in lpgcast/llm/usage.py 
115
116
117
def incr(self, counter: str, n: int = 1) -> None:
    """Add *n* to an integer counter."""
    setattr(self, counter, getattr(self, counter) + n)

Functions:

active_tracker()

The tracker bound by the innermost :func:track_usage, if any.

Source code in lpgcast/llm/usage.py 
152
153
154
def active_tracker() -> UsageTracker | None:
    """The tracker bound by the innermost :func:`track_usage`, if any."""
    return _active_tracker.get()

token_usage_from_openai(usage)

Parse an OpenAI-shaped usage object, an SDK model or a dict.

Source code in lpgcast/llm/usage.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def token_usage_from_openai(usage: Any) -> TokenUsage:
    """Parse an OpenAI-shaped ``usage`` object, an SDK model or a dict."""
    prompt_tokens = _opt_int(usage, "prompt_tokens")
    completion_tokens = _opt_int(usage, "completion_tokens")
    if prompt_tokens is None or completion_tokens is None:
        return TokenUsage()
    return TokenUsage(
        input_tokens=prompt_tokens,
        output_tokens=completion_tokens,
        reasoning_tokens=_opt_int(
            _get(usage, "completion_tokens_details"), "reasoning_tokens"
        ),
        cache_read_input_tokens=_opt_int(
            _get(usage, "prompt_tokens_details"), "cached_tokens"
        ),
    )

track_usage(tracker=None)

Charge calls made in this block (and tasks it spawns) to tracker too.

Task-local: concurrent episodes each charge their own tracker.

Source code in lpgcast/llm/usage.py 
138
139
140
141
142
143
144
145
146
147
148
149
@contextmanager
def track_usage(tracker: UsageTracker | None = None) -> Generator[UsageTracker]:
    """Charge calls made in this block (and tasks it spawns) to *tracker* too.

    Task-local: concurrent episodes each charge their own tracker.
    """
    tracker = tracker if tracker is not None else UsageTracker()
    token = _active_tracker.set(tracker)
    try:
        yield tracker
    finally:
        _active_tracker.reset(token)