Token 与成本核算

聊天消息总 token 计数封装

stage 上下文: 该单元位于 Token & Cost Accounting 主题下，承担对单个 Chat 当前消息列表做总 token 计数的最小封装。源码仅体现一次库函数调用：读取 self._model_name 与 chat.messages，将计数工作委托给 LiteLLM；本函数本身不维护累计寄存器，也不落盘或写回统计状态。

这段代码在干什么

函数 _count_total_tokens(self, chat: Chat) -> int 接收一个 Chat 对象，读取其中的 chat.messages，并结合实例上的 self._model_name 计算整段聊天消息的总 token 数。它直接返回一个整数结果，不修改 self、chat 或任何 register。函数体内也没有本地异常处理；底层导入或计数调用若失败，将直接向外传播。

接口 · 参数 / IO

(self, chat: Chat) -> int

• 参数: chat: Chat — 提供待统计消息列表；实际读取的是 chat.messages
• 读状态: self._model_name, chat.messages
• 返回: 返回 token_counter(model=self._model_name, messages=chat.messages) 的整数结果，即该聊天全部消息的总 token 数。
• 副作用: 函数体内执行 from litellm.utils import token_counter，依赖外部库导入与调用; 无对象状态写入；不修改 self、chat 或任何 register

执行流

1. 在函数体内执行 from litellm.utils import token_counter，取得外部库的 token 计数函数。
2. 读取 self._model_name 作为 model 参数，读取 chat.messages 作为 messages 参数。
3. 调用 token_counter(model=self._model_name, messages=chat.messages)。
4. 直接返回该调用结果；函数内无额外加工、缓存或状态更新。

源码

    def _count_total_tokens(self, chat: Chat) -> int:
        """Count total tokens across all messages in the chat."""
        from litellm.utils import token_counter

        return token_counter(model=self._model_name, messages=chat.messages)

Non-obvious 设计决策

• 计数逻辑没有在本地展开，而是完整委托给 litellm.utils.token_counter；源码可见本函数只负责参数转交与返回结果。
• 模型选择不是从 chat 推导，而是显式读取实例状态 self._model_name 传给库函数；因此计数规则由该字段与外部库共同决定。
• 导入语句写在函数体内部而非模块顶层，这是源码可见的组织方式；本函数每次调用都会先执行该局部导入语句。
• 函数没有 try/except、没有默认值分支、也没有 None 检查；导入失败、chat.messages 不可用或 token_counter 抛错时，异常都会原样向外传播。
• 函数不接触任何 self._... 累计寄存器，也不写回 chat；其产出仅为返回值。

上下游关系

• 调用方: unknown
• 核心被调用: litellm.utils.token_counter
• 配置/状态来源: self._model_name; chat.messages
• 结果去向: 直接作为函数返回值交还调用方（具体去向在本源码片段中不可见）

使用量对象归一化提取器

stage 上下文: 该函数是一个局部的数据整形单元：把 usage_info 转成固定四元组 (prompt_tokens, completion_tokens, cached_tokens, cost_usd)。从函数体可证的信息仅包括一次 falsy 判定和一次直接元组返回；其外部触发时机与 stage 内位置无法由此片段单独证明。

这段代码在干什么

函数 _extract_usage_metrics(self, usage_info) -> tuple[int, int, int, float] 接收一个 usage_info 对象或其他可参与真值判断的值。若 usage_info 为 falsy，则直接返回 0, 0, 0, 0；否则读取其 prompt_tokens、completion_tokens、cache_tokens 与 cost_usd 属性，返回一个四元组，其中 cost_usd 会在不大于 0 时被归零。函数不读取任何 self 上的状态，也不产生副作用。

接口 · 参数 / IO

(self, usage_info) -> tuple[int, int, int, float]

• 参数: self: ? — 实例方法接收者；本函数体内未使用其实例状态; usage_info: ? — 待提取的使用量对象；先参与 if not usage_info 真值判断，truthy 路径上必须提供 prompt_tokens、completion_tokens、cache_tokens、cost_usd 属性
• 返回: 返回四元组 (prompt_tokens, completion_tokens, cached_tokens, cost_usd)；若 usage_info 为 falsy，则返回 (0, 0, 0, 0)

执行流

1. 先对参数 usage_info 执行 falsy 检查：命中 if not usage_info 时，直接返回四个 0 组成的结果，不再读取任何属性。
2. 若 usage_info 为 truthy，则直接构造并返回一个元组字面量：前三项分别取自 usage_info.prompt_tokens、usage_info.completion_tokens、usage_info.cache_tokens。
3. 同一 truthy 路径中，第四项读取 usage_info.cost_usd，仅当其大于 0 时原样返回；否则返回 0。

源码

    def _extract_usage_metrics(self, usage_info) -> tuple[int, int, int, float]:
        """Extract and normalize metrics from usage info.

        Args:
            usage_info: Usage information from LLM response

        Returns:
            Tuple of (prompt_tokens, completion_tokens, cached_tokens, cost_usd)
        """
        if not usage_info:
            return 0, 0, 0, 0

        return (
            usage_info.prompt_tokens,
            usage_info.completion_tokens,
            usage_info.cache_tokens,
            usage_info.cost_usd if usage_info.cost_usd > 0 else 0,
        )

Non-obvious 设计决策

• 函数把“无可用 usage 信息”的判定实现为宽松的 falsy 检查（if not usage_info），而不是仅检查 None。这样任何假值输入都会被统一折叠为全 0 结果。
• token 相关三个字段在 truthy 路径上不做额外归一化、类型转换或下界裁剪，而是直接透传对象属性值；因此该函数的归一化策略只覆盖 cost_usd。
• 对 cost_usd 采用 > 0 的门槛，意味着非正值都会被压成 0，避免把负值或 0 以原样形式传播到返回结果中。

上下游关系

• 调用方: 外部调用方未由此代码片段直接给出；只能确定任何需要把 usage_info 变为四元组的代码都可调用它
• 核心被调用: 无显式函数调用; 依赖 usage_info 的真值判定协议; 依赖 usage_info.prompt_tokens 属性访问; 依赖 usage_info.completion_tokens / cache_tokens / cost_usd 属性访问
• 配置/状态来源: 无；函数体不读取任何 self._* 配置或 register
• 结果去向: 返回给直接调用者的四元组结果; 其中第一项对应 prompt_tokens; 第二项对应 completion_tokens; 第三项对应 cached_tokens; 第四项对应已归一化的 cost_usd

子代理用量累计更新器

stage 上下文: 该函数位于 Token & Cost Accounting 这一横切 stage 中，职责是把一次子代理 LLM 调用返回的 usage 信息并入独立的子代理累计器。它通常在 _run_subagent 获得响应 usage 元数据后立即触发。与兄弟函数相比，_extract_usage_metrics 负责把 usage 对象拆成数值元组，而本函数直接把 usage_info 累加到 self._subagent_metrics；二者共同服务于主对话指标与子代理指标分账的设计。

这段代码在干什么

函数 _update_subagent_metrics(self, usage_info) -> None 接收一次子代理 LLM 响应的 usage_info。当该参数为 falsy 时，函数直接返回，不产生任何副作用；否则把其中的 prompt_tokens、completion_tokens、cache_tokens 和 cost_usd 累加到 self._subagent_metrics 的对应总量字段。函数本身不返回业务结果，真正产出是对实例内子代理指标累计器的原地更新。

接口 · 参数 / IO

(self, usage_info) -> None

• 参数: self: ? — Terminus2 实例，提供子代理指标累计状态 self._subagent_metrics; usage_info: ? — 一次子代理 LLM 响应附带的用量信息，包含 token 数与成本字段
• 读状态: self._subagent_metrics
• 返回: 返回 None；实际结果是把本次子代理调用的 token 与成本增量写入累计状态。
• 副作用: 修改 self._subagent_metrics.total_prompt_tokens; 修改 self._subagent_metrics.total_completion_tokens; 修改 self._subagent_metrics.total_cached_tokens; 修改 self._subagent_metrics.total_cost_usd

执行流

1. 先检查 usage_info 的真值；若其为 falsy，则立即返回 None，本次不做任何累计。
2. 若存在 usage 数据，则读取 usage_info.prompt_tokens，并把该值加到 self._subagent_metrics.total_prompt_tokens。
3. 继续读取 usage_info.completion_tokens 与 usage_info.cache_tokens，分别累计到 total_completion_tokens 和 total_cached_tokens。
4. 最后读取 usage_info.cost_usd，并把金额增量加到 self._subagent_metrics.total_cost_usd，完成一次子代理用量折叠。

源码

    def _update_subagent_metrics(self, usage_info) -> None:
        """Update subagent metrics with usage information from an LLM response.

        Args:
            usage_info: Usage information from LLM response containing token counts and cost
        """
        if not usage_info:
            return

        self._subagent_metrics.total_prompt_tokens += usage_info.prompt_tokens
        self._subagent_metrics.total_completion_tokens += usage_info.completion_tokens
        self._subagent_metrics.total_cached_tokens += usage_info.cache_tokens
        self._subagent_metrics.total_cost_usd += usage_info.cost_usd

Non-obvious 设计决策

• 函数对 usage_info 采用显式 falsy 短路返回，而不是在字段访问时依赖异常暴露问题；这表明“无 usage 元数据”被当作可接受情况处理，目标是让缺失统计时退化为 no-op，而不打断子代理执行路径。
• 子代理指标被写入 self._subagent_metrics 这一独立累计器，而不是混入主 chat 的计数状态；这与本 stage 对主对话指标和子代理指标分开记账、在 stage-5 再统一汇总的设计一致。
• 这里直接累加 usage_info 的属性值，没有像兄弟函数 _extract_usage_metrics 那样对 cost_usd 做非正值归零处理；这说明该辅助函数的职责被刻意收窄为“原样并入子代理 usage 记录”，避免在多个入口重复定义清洗规则。

上下游关系

• 调用方: Terminus2._run_subagent（在子代理 LLM 响应带回 usage 元数据后调用）
• 核心被调用: 无显式函数调用; 读取 usage_info.prompt_tokens; 读取 usage_info.completion_tokens; 读取 usage_info.cache_tokens; 读取 usage_info.cost_usd
• 配置/状态来源: usage_info 参数提供本次增量数据; self._subagent_metrics 提供当前累计基线
• 结果去向: self._subagent_metrics.total_prompt_tokens; self._subagent_metrics.total_completion_tokens; self._subagent_metrics.total_cached_tokens; self._subagent_metrics.total_cost_usd; stage-5 的总量汇总逻辑（主对话指标与子代理指标合并）
• 同类 sibling: Terminus2._extract_usage_metrics：也是 usage 读取辅助函数，但返回数值元组且对 falsy 输入给出全零结果；本函数则直接更新子代理累计器。; Terminus2._count_total_tokens：计算 chat 消息 token 总数且无副作用；本函数不计算消息 token，而是累加响应提供的 usage 指标。

把子代理响应写入轨迹步骤的封装点

stage 上下文: 该函数位于 Token & Cost Accounting 相关阶段中的一个落点：它在记录子代理响应时，按是否拿到 usage_info 决定是否把 token 与 cost 指标一并挂到 Step.metrics。触发前提是调用方已经拿到一个 LLMResponse，并准备把这次子代理输出追加到某个 steps 列表。它本身不维护累计寄存器，只负责构造并追加单条轨迹记录。

这段代码在干什么

函数 _append_subagent_response_step 接收目标步骤列表 steps、新步骤编号 step_id、模型响应 response、可选的 usage_info 与子代理名 subagent_name，向 steps 追加一条 source="agent" 的 Step。当 usage_info 为 truthy 时，追加的步骤会带有 Metrics，其中写入 prompt_tokens、completion_tokens、cached_tokens、cost_usd 以及来自 response 的 token-id / logprobs 字段；当其缺失时，只记录基础响应内容并发出一条 warning 日志。函数返回 None，真正产出是对传入列表的原地追加，以及在缺失用量信息时的日志副作用。

接口 · 参数 / IO

(self, steps: list[Step], step_id: int, response: LLMResponse, usage_info, subagent_name: str) -> None

• 参数: steps: list[Step] — 被原地追加的目标轨迹步骤列表; step_id: int — 新建 Step 的编号; response: LLMResponse — 响应内容来源；函数读取其 model_name、content、reasoning_content、prompt_token_ids、completion_token_ids、logprobs; usage_info: ? — 可选用量对象；若为 truthy，代码要求其暴露 prompt_tokens、completion_tokens、cache_tokens、cost_usd 属性; subagent_name: str — 当 usage_info 缺失时，用于 warning 文案中的子代理名称
• 读状态: self._model_name, self.logger
• 返回: 返回 None；真实产出是向 steps 追加一条带有新鲜 UTC ISO 时间戳的 Step，并在缺失 usage_info 时写出 warning 日志。
• 副作用: 原地修改传入的 steps 列表; 调用 self.logger.warning(...) 记录缺失用量信息; 为每条追加的 Step 生成 datetime.now(timezone.utc).isoformat() 时间戳

执行流

1. 先对 usage_info 做真值分支：若其为 truthy，进入“带指标”追加路径；否则进入“仅响应内容”追加路径。
2. 在带指标路径中，函数构造一个 Step 并追加到 steps：固定写入 step_id、当前 UTC 时间的 ISO 字符串、source="agent"、response.model_name or self._model_name、response.content 与 response.reasoning_content。
3. 同一带指标路径还会构造 Metrics：把 usage_info.prompt_tokens、usage_info.completion_tokens、usage_info.cache_tokens 写入对应字段；cost_usd 仅在 usage_info.cost_usd > 0 时保留原值，否则写成 None。
4. 带指标路径中，response.prompt_token_ids、response.completion_token_ids 与 response.logprobs 也被写进 Metrics；因此这些字段只会在 usage_info 为 truthy、且 Step 含有 metrics 时出现。
5. 若 usage_info 为 falsy，函数先调用 self.logger.warning(f"Failed to get token usage for {subagent_name}") 发出告警，再追加一个不含 metrics 的基础 Step。
6. 在无指标路径中，追加的 Step 仍包含新鲜 UTC ISO 时间戳、source="agent"、模型名回退逻辑、response.content 与 response.reasoning_content，但不会写入 token 数、cost、token ids 或 logprobs。

源码

    def _append_subagent_response_step(
        self,
        steps: list[Step],
        step_id: int,
        response: LLMResponse,
        usage_info,
        subagent_name: str,
    ) -> None:
        """Append a response step with conditional metrics to trajectory steps.

        Args:
            steps: List of steps to append to
            step_id: ID for the new step
            response: LLM response
            usage_info: Usage info (may be None)
            subagent_name: Name of subagent for warning message
        """
        if usage_info:
            steps.append(
                Step(
                    step_id=step_id,
                    timestamp=datetime.now(timezone.utc).isoformat(),
                    source="agent",
                    model_name=response.model_name or self._model_name,
                    message=response.content,
                    reasoning_content=response.reasoning_content,
                    metrics=Metrics(
                        prompt_tokens=usage_info.prompt_tokens,
                        completion_tokens=usage_info.completion_tokens,
                        cached_tokens=usage_info.cache_tokens,
                        cost_usd=usage_info.cost_usd
                        if usage_info.cost_usd > 0
                        else None,
                        prompt_token_ids=response.prompt_token_ids,
                        completion_token_ids=response.completion_token_ids,
                        logprobs=response.logprobs,
                    ),
                )
            )
        else:
            self.logger.warning(f"Failed to get token usage for {subagent_name}")
            steps.append(
                Step(
                    step_id=step_id,
                    timestamp=datetime.now(timezone.utc).isoformat(),
                    source="agent",
                    model_name=response.model_name or self._model_name,
                    message=response.content,
                    reasoning_content=response.reasoning_content,
                )
            )

Non-obvious 设计决策

• 函数把 usage_info 缺失视为可容忍情形，而不是抛异常中断；从 if usage_info: ... else: self.logger.warning(...) 可见，其选择是在保留响应轨迹记录的同时，用日志暴露计量缺口。
• 模型名采用 response.model_name or self._model_name 的回退策略，说明记录层优先保留响应自带模型标识，但允许在响应未提供时退回实例配置值，避免 Step.model_name 留空。
• 成本字段显式做了 usage_info.cost_usd > 0 判断，非正值统一写成 None 而非原样记录；这体现了一个本地数据整形决策：把零或负成本视为“无可用成本值”，避免与有效正成本混淆。
• prompt_token_ids、completion_token_ids 与 logprobs 被放在 Metrics(...) 内，而不是 Step 顶层字段；因此一旦缺失 usage_info，这些与细粒度计量绑定的响应附加信息也会整体缺席。

上下游关系

• 调用方: 某个在拿到子代理 LLMResponse 后、希望把结果写入其轨迹列表的上层流程
• 核心被调用: Step; Metrics; datetime.now; timezone.utc; self.logger.warning
• 配置/状态来源: self._model_name; self.logger
• 结果去向: 传入的 steps 列表; 追加生成的 Step.metrics（仅当 usage_info 为 truthy）; 日志输出系统

记录单次 API 请求耗时

stage 上下文: 该函数属于 Token & Cost Accounting 阶段中的通用度量辅助逻辑，职责非常单一：把一次调用的耗时样本写入实例上的请求时间列表。与同 stage 的 _count_total_tokens、_extract_usage_metrics、_update_subagent_metrics 类似，它服务于指标采集；但不同于前两者的纯读取/返回模式，这里唯一产出是对 self._api_request_times 的原地追加。

这段代码在干什么

函数 _track_api_request_time(self, start_time: float) -> None 接收一个先前保存的起始时间戳 start_time，随后获取当前时间并计算本次经过的毫秒数。它将精确按表达式 (end_time - start_time) * 1000 得到的浮点值追加到 self._api_request_times。函数本身不返回业务值，唯一副作用是向该列表新增一个耗时样本。

接口 · 参数 / IO

(self, start_time: float) -> None

• 参数: start_time: float — 先前用 time.time() 记录的请求起始时间戳
• 返回: 返回 None；真正产出是向 self._api_request_times 追加一个以毫秒计的浮点耗时样本。
• 副作用: 调用 time.time() 读取当前墙钟时间; 通过 .append(...) 原地修改 self._api_request_times

执行流

1. 调用 time.time() 取得当前时间，并保存到局部变量 end_time。
2. 用源码中的精确公式 (end_time - start_time) * 1000 计算请求耗时，结果保存为 request_time_ms。
3. 将 request_time_ms 追加到 self._api_request_times 列表末尾。

源码

    def _track_api_request_time(self, start_time: float) -> None:
        """Track API request time from start timestamp.

        Args:
            start_time: Start time from time.time()
        """
        end_time = time.time()
        request_time_ms = (end_time - start_time) * 1000
        self._api_request_times.append(request_time_ms)

Non-obvious 设计决策

• 函数接受外部传入的 start_time，而不是在函数内自行决定起点；这使它只负责“结束计时并记账”这一件事，边界清晰。
• 代码把结果以单个样本的形式写入 self._api_request_times，而不是在此处做求和、平均值或聚合；因此该函数只承担原始度量记录，不承担统计汇总。
• 耗时单位在写入前直接乘以 1000 转为毫秒，说明该列表保存的是毫秒尺度样本，而非秒值。

上下游关系

• 调用方: 未在所给源码片段中展示；可确定其调用方需先提供 start_time
• 核心被调用: time.time; self._api_request_times.append
• 配置/状态来源: 参数 start_time; 实例属性 self._api_request_times
• 结果去向: 写入 self._api_request_times 供后续统计或对外汇总使用; 对应 state register reg-api-request-times 的实例内承载列表
• 同类 sibling: 与 _count_total_tokens 相同点是都服务于指标采集；不同点是本函数不返回计量结果，而是直接写入实例列表。; 与 _extract_usage_metrics 相比，本函数不做字段拆解或归零规则处理，只记录一个简单的耗时样本。; 与 _update_subagent_metrics 相似之处在于都以副作用更新累计度量容器；不同点是这里追加的是单次耗时样本，而非把 token/cost 加总进聚合字段。