core/agent: char-based token estimate fallback for usage

When API usage is missing, estimate prompt/completion tokens at ~4
chars per token and mark source as estimate (or mixed when combined).
This commit is contained in:
2026-07-15 10:52:37 +08:00
parent d81e168f3b
commit 0a3127454d
2 changed files with 156 additions and 12 deletions
+88 -4
View File
@@ -1,34 +1,74 @@
from __future__ import annotations
from typing import cast
from typing import TYPE_CHECKING, cast
from msgspec import UNSET, Struct
from plyngent.agent.budget import estimate_message_chars, estimate_messages_chars
if TYPE_CHECKING:
from collections.abc import Sequence
from plyngent.lmproto.openai_compatible.model import AnyChatMessage, AssistantChatMessage
# Rough OpenAI-style heuristic: ~4 characters per token (not model-accurate).
DEFAULT_CHARS_PER_TOKEN = 4.0
class TokenUsage(Struct, omit_defaults=True):
"""Token counts from a provider ``usage`` object (OpenAI-compatible)."""
"""Token counts from API ``usage`` or a char-based estimate."""
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
# "api" | "estimate" | "mixed" (session totals combining both)
source: str = "api"
def add(self, other: TokenUsage) -> TokenUsage:
if self.is_zero():
return other
if other.is_zero():
return self
source = self.source if self.source == other.source else "mixed"
return TokenUsage(
prompt_tokens=self.prompt_tokens + other.prompt_tokens,
completion_tokens=self.completion_tokens + other.completion_tokens,
total_tokens=self.total_tokens + other.total_tokens,
source=source,
)
def is_zero(self) -> bool:
return self.prompt_tokens == 0 and self.completion_tokens == 0 and self.total_tokens == 0
def format_line(self) -> str:
tag = ""
if self.source == "estimate":
tag = " (est)"
elif self.source == "mixed":
tag = " (api+est)"
return (
f"tokens prompt={self.prompt_tokens} completion={self.completion_tokens} "
f"total={self.total_tokens}"
f"total={self.total_tokens}{tag}"
)
def chars_to_tokens(chars: int, *, chars_per_token: float = DEFAULT_CHARS_PER_TOKEN) -> int:
"""Convert character count to a rough token estimate (ceiling, min 0)."""
if chars <= 0 or chars_per_token <= 0:
return 0
# Ceiling division without float drift for large values
return max(0, int((chars + chars_per_token - 1e-9) // chars_per_token))
def estimate_tokens_from_chars(
chars: int,
*,
chars_per_token: float = DEFAULT_CHARS_PER_TOKEN,
) -> int:
"""Alias for :func:`chars_to_tokens` (public name for fallback counters)."""
return chars_to_tokens(chars, chars_per_token=chars_per_token)
def _as_nonneg_int(value: object) -> int:
if isinstance(value, bool):
return 0
@@ -53,4 +93,48 @@ def token_usage_from_api(usage: object) -> TokenUsage | None:
total = prompt + completion
if prompt == 0 and completion == 0 and total == 0:
return None
return TokenUsage(prompt_tokens=prompt, completion_tokens=completion, total_tokens=total)
return TokenUsage(
prompt_tokens=prompt,
completion_tokens=completion,
total_tokens=total,
source="api",
)
def estimate_token_usage(
prompt_messages: Sequence[AnyChatMessage],
assistant: AssistantChatMessage | None = None,
*,
chars_per_token: float = DEFAULT_CHARS_PER_TOKEN,
) -> TokenUsage:
"""Char-based fallback when the provider does not report ``usage``."""
prompt_chars = estimate_messages_chars(prompt_messages)
completion_chars = 0
if assistant is not None:
completion_chars = estimate_message_chars(assistant)
prompt_tokens = chars_to_tokens(prompt_chars, chars_per_token=chars_per_token)
completion_tokens = chars_to_tokens(completion_chars, chars_per_token=chars_per_token)
return TokenUsage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
source="estimate",
)
def resolve_round_usage(
api_usage: object,
prompt_messages: Sequence[AnyChatMessage],
assistant: AssistantChatMessage,
*,
chars_per_token: float = DEFAULT_CHARS_PER_TOKEN,
) -> TokenUsage:
"""Prefer API usage; otherwise estimate from message characters."""
parsed = token_usage_from_api(api_usage)
if parsed is not None:
return parsed
return estimate_token_usage(
prompt_messages,
assistant,
chars_per_token=chars_per_token,
)