openai-agents-python 使用量追踪完全指南:基于 Usage 的成本监控与限制实施
2026/9/12 4:31:26 网站建设 项目流程

openai-agents-python 使用量追踪完全指南:基于 Usage 的成本监控与限制实施

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python

本篇技术指南聚焦 openai-agents-python(Agents SDK)的 token 使用量追踪机制,讲解 SDK 自动统计哪些指标、如何从运行结果(Runner.run)、会话(Session)、钩子(RunHooks)与运行检查点(RunState)中读取使用量数据,以及如何通过ModelSettings适配第三方模型提供商。读完本文,你将能够在自己的多智能体应用中监控成本、强制执行 token 限制、记录分析数据,并正确处理流式响应、会话恢复与嵌套智能体等场景下的使用量语义。

自动追踪的内容:Usage 数据结构全解

Agents SDK 在每次运行(run)时自动追踪 LLM 调用的 token 使用量,无需额外埋点。所有指标被标准化为Usage数据类(pydantic dataclass),在 usage.py 中定义,主要字段如下:

字段含义
requests发起的 LLM API 调用次数
input_tokens发送的输入 token 总数
output_tokens接收的输出 token 总数
total_tokens输入 + 输出
request_usage_entries每个请求的使用量明细列表(list[RequestUsage]
input_tokens_details输入明细:cached_tokens(缓存命中)、cache_write_tokens(缓存写入)
output_tokens_details输出明细:reasoning_tokens(推理 token)

其中RequestUsage是单次 API 请求的明细数据结构(usage.py),包含input_tokensoutput_tokenstotal_tokens以及对应的input_tokens_details/output_tokens_details,用于精确计算单请求成本与监控上下文窗口消耗。

从源码实现看,Usage的聚合核心是add()方法(usage.py):它将另一个Usage对象的所有字段累加到自身,同时自动维护request_usage_entries——如果被合并对象已有逐请求明细则直接深拷贝合并,否则当它代表单个请求(requests == 1total_tokens > 0)时,会按顶层字段合成一条明细。此外,__post_init__会把部分提供商缺失的可选 token 明细字段(cached_tokenscache_write_tokensreasoning_tokens)统一归一化为 0,防止后续聚合时出现TypeError

从运行结果中访问使用量

执行Runner.run(...)之后,通过result.context_wrapper.usage即可拿到本次运行的使用量:

result = await Runner.run(agent, "What's the weather in Tokyo?") usage = result.context_wrapper.usage print("Requests:", usage.requests) print("Input tokens:", usage.input_tokens) print("Output tokens:", usage.output_tokens) print("Total tokens:", usage.total_tokens)

这里的关键是RunContextWrapper:它包装了你传给Runner.run()的上下文对象,同时携带usage: Usage字段,描述"到目前为止"的智能体运行使用量。需要注意源码注释中的一条重要提示:对于流式响应,在流的最后一个数据块被处理之前,usage都是过期(stale)的,因此不要在流式运行中途依赖其精确值。

使用量会汇总运行期间的所有模型调用,包括生成工具调用(tool call)或任务转移(handoff)时触发的模型调用。仓库提供了一个完整的可运行示例 examples/basic/usage_tracking.py:它定义了一个带天气查询工具的 Agent,运行后打印聚合使用量,并遍历request_usage_entries输出每次请求的明细:

def print_usage(usage: Usage) -> None: print("\n=== Usage ===") print(f"Input tokens: {usage.input_tokens}") print(f"Output tokens: {usage.output_tokens}") print(f"Total tokens: {usage.total_tokens}") print(f"Requests: {usage.requests}") for i, request in enumerate(usage.request_usage_entries): print(f" {i + 1}: {request.input_tokens} input, {request.output_tokens} output")

使用第三方适配器时的使用量统计

不同第三方适配器和提供商后端报告使用量的方式各不相同。如果你通过第三方适配器访问模型,并且需要准确的result.context_wrapper.usage值,请参照以下规则:

  • AnyLLMModel:只要上游提供商返回使用量数据,系统就会自动传递。但从 Chat Completions 后端以流式方式获取响应时,可能需要设置ModelSettings(include_usage=True),才会发出包含使用量的数据块。
  • LitellmModel:部分提供商后端默认不报告使用量,因此通常必须设置ModelSettings(include_usage=True)

include_usageModelSettings上的布尔参数,其文档注释明确标注"仅适用于 Chat Completions API"——即是否在流式响应中包含 usage 数据块。部署前,请务必在计划使用的具体提供商后端上验证使用量报告行为。

按请求追踪使用量(request_usage_entries)

SDK 会在request_usage_entries中自动记录每一个 API 请求的使用量明细,这对详细成本核算和监控上下文窗口消耗非常有用:

result = await Runner.run(agent, "What's the weather in Tokyo?") for i, request in enumerate(result.context_wrapper.usage.request_usage_entries): print(f"Request {i + 1}: {request.input_tokens} in, {request.output_tokens} out")

request_usage_entries的典型价值在于:一次运行可能发起多次模型调用(例如 3 次 API 调用,输入 token 分别为 100K、150K、80K),聚合后的input_tokens是 330K,但request_usage_entries保留了[100K, 150K, 80K]的粒度(见 usage.py 的字段说明),便于发现单次请求的异常消耗。

保留提供商原始使用量负载(preserve_raw_usage)

Agents SDK 默认会把提供商返回的使用量归一化为统一的Usage字段,从而在不同模型提供商之间提供一致的总量。但当你需要保留提供商特有的使用量字段、或者需要区分"提供商省略了某字段"与"提供商报告该字段为 0"时,可以将ModelSettings.preserve_raw_usage设置为True

from agents import Agent, ModelSettings, Runner agent = Agent( name="Assistant", model_settings=ModelSettings(preserve_raw_usage=True), ) result = await Runner.run(agent, "What's the weather in Tokyo?") for response in result.raw_responses: print(response.raw_usage)

该机制的核心语义如下(同时见 items.py 中ModelResponse.raw_usage的定义):

  • Agents SDK 将每个ModelResponse.raw_usage值存储为该次模型调用的提供商负载的独立 JSON 兼容快照dict[str, Any] | None),快照在 SDK 归一化缺失字段之前捕获。
  • SDK 不会在整个运行过程中汇总raw_usage——它只是逐响应保留原始快照。
  • 当禁用保留功能、提供商未返回使用量负载、或上游适配器已经丢弃了原始字段存在性信息时,该值保持为None

源码层面,usage.py 的_raw_usage_snapshot()负责生成快照:它接受 Mapping 或带model_dump的对象,通过TypeAdapter(dict[str, JsonValue])校验后序列化为 JSON 兼容字典;任何无法表示为 JSON 的适配器特定值都会被安全丢弃,绝不会让一次成功的模型调用因为负载保留失败而报错(该机制被定位为诊断性元数据)。

另外两点边界务必注意:

  1. preserve_raw_usage只保留"到达模型适配器"的使用量负载,该设置本身不会向提供商请求使用量数据。当流式 Chat Completions 提供商要求显式请求使用量时,还应同时设置ModelSettings(include_usage=True)
  2. LitellmModel目前无论是流式还是非流式运行,都不会填充ModelResponse.raw_usage,因此preserve_raw_usage=True对该适配器不生效。使用LitellmModel时请继续使用标准化的Usage字段;若确实需要提供商特定字段的存在性信息,请选择支持原始使用量保留的适配器。

会话(Session)场景下的使用量

使用Session(例如SQLiteSession)时,每次调用Runner.run(...)返回的使用量仅代表该次特定运行。会话会保留对话历史作为上下文,但各次运行的使用量彼此独立:

session = SQLiteSession("my_conversation") first = await Runner.run(agent, "Hi!", session=session) print(first.context_wrapper.usage.total_tokens) # Usage for first run second = await Runner.run(agent, "Can you elaborate?", session=session) print(second.context_wrapper.usage.total_tokens) # Usage for second run

需要特别留意的行为是:虽然会话会在多次运行之间保留对话上下文,但先前的消息会作为输入被重新传入每一次运行,因此后续轮次的输入 token 数量会包含历史消息,反映真实的多轮对话成本。

RunState 检查点中的使用量隔离

RunResult.to_state()会捕获截至当前已累计使用量的独立快照。从该检查点恢复的运行以捕获的总量为起点,并在此基础上累加自身模型调用的使用量;恢复后的运行不会把新增总量写回原始的RunResult,也不会写入根据该结果创建的其他检查点:

first = await Runner.run(agent, "First request") checkpoint_a = first.to_state() checkpoint_b = first.to_state() resumed_a = await Runner.run(agent, checkpoint_a) resumed_b = await Runner.run(agent, checkpoint_b) assert resumed_a.context_wrapper.usage is not first.context_wrapper.usage assert resumed_b.context_wrapper.usage is not resumed_a.context_wrapper.usage

这种隔离的实现依据在 run_context.py 的_copy_for_run_state():它通过copy.deepcopy(self.usage)生成独立的Usage副本,避免多个检查点共享同一个可变的Usage实例(Usage.add()是原地聚合,还会扩展request_usage_entries,共享实例会让一个检查点的 token 污染其他检查点及来源结果)。

这种隔离同样适用于Usage中的request_usage_entries列表。唯一的例外是恢复后的嵌套Agent.as_tool()运行:嵌套运行在恢复后的模型使用量会被有意汇总进当前外层运行的使用量,与嵌套运行此前的模型调用处理方式保持一致。

在 RunHooks 中记录使用量

如果你使用RunHooks,传递给每个钩子的context对象(即RunContextWrapper)都包含usage,可以借此在生命周期的关键时刻记录使用量:

class MyHooks(RunHooks): async def on_agent_end(self, context: RunContextWrapper, agent: Agent, output: Any) -> None: u = context.usage print(f"{agent.name} → {u.requests} requests, {u.total_tokens} total tokens")

这在多智能体场景中尤其有用:每个 Agent 结束时都可以输出自己累计的请求数与 token 总量,形成逐智能体的成本日志。注意context.usage同样是"到目前为止"的累计值,流式场景下需等待流结束才准确。

压缩会话(Compaction)与使用量的关系

OpenAIResponsesCompactionSession在运行结束前自动压缩历史记录时,该responses.compact请求报告的使用量也会被累加到同一次运行的总量中。对应的测试 tests/memory/test_openai_responses_compaction_session.py 验证了压缩后requests == 2input_tokens == 150_000total_tokens == 192_000request_usage_entries仅含 1 条记录的行为。

与自动压缩不同,在运行之外手动调用run_compaction()时,由于没有包含该调用的运行上下文,因此不会更新先前运行返回的使用量对象。相关会话机制的完整说明请参阅 OpenAI Responses 压缩会话。

API 参考

符号说明
Usage使用量追踪数据结构(聚合总量 + 逐请求明细)
RequestUsage每个请求的使用量详情
RunContextWrapper从运行上下文中访问usage
RunHooks接入使用量追踪的生命周期钩子

配套的序列化支持同样值得关注:usage.py 的serialize_usage/deserialize_usage提供了Usage与 JSON 字典之间的双向转换(input_tokens_detailsoutput_tokens_details在序列化时包装为列表,反序列化时兼容新旧格式),这为将使用量持久化到数据库、跨进程传递或接入追踪系统(tracing span 数据)提供了基础设施;tests/model_settings/test_serialization.py 则验证了preserve_raw_usageModelSettings序列化中的往返一致性。综合运用上述 API,即可构建从"单次请求明细"到"整次运行总量"的完整成本观测链路。

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询