taolib.symphony.observability.metrics#

令牌核算和运行时统计。

跟踪 Codex 会话的令牌使用(input/output/total)和运行时间。 优先使用绝对线程总计(thread/tokenUsage/updated),跟踪增量避免重复计数。

Classes#

TokenUsage

令牌用量汇总。

MetricsCollector

令牌和运行时聚合器。

Module Contents#

class taolib.symphony.observability.metrics.TokenUsage#

令牌用量汇总。

input_tokens: int = 0#
output_tokens: int = 0#
total_tokens: int = 0#
to_dict() dict[str, int]#

转换为字典。

class taolib.symphony.observability.metrics.MetricsCollector#

令牌和运行时聚合器。

累计跟踪所有 Codex 会话的令牌消耗和总运行时间。 支持从 Codex 事件流中提取绝对令牌值或增量累加。

_totals: TokenUsage#
_last_reported_input: int = 0#
_last_reported_output: int = 0#
_last_reported_total: int = 0#
_start_time: datetime.datetime#
update_from_event(event: dict[str, Any]) None#

从 Codex 事件更新令牌计数。

支持两种事件格式: 1. 绝对值格式(thread 级 tokenUsage.updated):直接覆盖总计。 2. 增量格式(单次 turn 的 input/output):累加到当前总计。

参数:

event -- Codex 事件字典,可能包含 "tokenUsage" 键。

update_from_thread(thread_data: dict[str, Any]) None#

从线程级别的 tokenUsage 数据更新总计(绝对值)。

适用于 Codex thread 对象中的 tokenUsage 字段。

参数:

thread_data -- 包含 tokenUsage 的线程数据。

property totals: TokenUsage#

当前令牌用量汇总。

property seconds_running: float#

从启动到现在的运行秒数。

to_dict() dict[str, Any]#

导出为字典,包含令牌用量和运行时间。

reset() None#

重置所有计数器和计时器。