企业官网建设流程全解析

pydantic-ai 用量追踪与预算管控：usage与prices模块 API 详解

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai

本文围绕 pydantic-ai 的pydantic_ai.usage与pydantic_ai.prices两个 API 模块展开：前者负责把每次模型请求、每次 Agent 运行的 token 用量、工具调用次数与美元成本结构化记录并可序列化为 OpenTelemetry 属性，后者负责在后台持续刷新模型价格表使成本估算保持有效。读完本文，你将掌握RequestUsage、RunUsage、UsageLimits三个核心类的字段语义与默认值，理解请求前/响应后/工具执行前各阶段的限额校验时机，并会用usage_limits参数与update_in_background()为生产 Agent 设置成本护栏。

模块导出与定位

API 文档 将该模块的公开接口定义为两个包：pydantic_ai.usage与pydantic_ai.prices。从__init__.py看，pydantic_ai顶层直接再导出了RequestUsage、RunUsage、UsageLimits三个符号：

from pydantic_ai import RequestUsage, RunUsage, UsageLimits # 等价于 from pydantic_ai.usage import RequestUsage, RunUsage, UsageLimits

usage模块的__all__即这三个类（usage.py#L20）；prices模块只导出一个函数update_in_background（prices.py#L7）。usage模块的实现基于标准库dataclass，并通过自定义 Pydantic core schema（UsageBase.__get_pydantic_core_schema__）保证任意未知 usage 字段能在 JSON 序列化/反序列化往返中不丢失——这对把历史消息存进数据库后再次加载的场景很关键。

UsageBase：所有用量对象的公共字段

RequestUsage和RunUsage都继承自UsageBase，公共字段的完整定义与语义如下（字段说明直接来自源码 docstring）：

几个源码层面的细节值得注意：

1. 字段是“包含式桶”而非互斥桶。input_tokens的 docstring 明确说明缓存读写、音频 token 都算在其中；对 Anthropic、Bedrock 这类原始input_tokens不含缓存读写的供应商，usage 提取层会做归一化，使该约定跨供应商一致（usage.py#L94-L100）。因此input_tokens不能直接用来估算“未命中缓存的计费输入”，而cache_hit_ratio才是缓存效率指标。
2. 向后兼容别名：反序列化旧数据时，input_tokens接受别名request_tokens、output_tokens接受response_tokens（AliasChoices，usage.py#L89-L93），保证改名前存入数据库的消息仍可加载。
3. 两个常用派生属性：
   • total_tokens：input_tokens + output_tokens；
   • cache_hit_ratio：cache_read_tokens / input_tokens（无输入时返回0.0），用于比较不同供应商的 prompt 缓存命中率（usage.py#L200-L218）。

序列化为 OpenTelemetry 属性

UsageBase.opentelemetry_attributes()把用量转换为符合gen_ai.usage.*语义约定的属性字典（usage.py#L220-L255）：

• gen_ai.usage.input_tokens/gen_ai.usage.output_tokens（仅在非零时输出）；
• gen_ai.usage.cache_creation.input_tokens（写缓存）与gen_ai.usage.cache_read.input_tokens（读缓存）；
• details中的每个键以gen_ai.usage.details.前缀展开。

源码中专门维护了一个_FIRST_CLASS_TOKEN_DETAIL_KEYS集合（input_tokens、output_tokens）：当供应商适配器把同名键塞进details（如 Anthropic 流式补报、Cohere 的计费单位）时，这些键会跳过details.*的展开——否则同一物理量会在两个属性下各报一次，导致 Langfuse 等下游聚合端把 token 与成本重复累加。这是一个很具体的“防双计”设计。

RequestUsage：单次请求的用量

RequestUsage描述一次 LLM 请求的用量，它实现了genai_prices.types.AbstractUsage协议，因此可直接交给 genai-prices 库计价。关键成员：

• requests属性：恒为1，让单请求用量与RunUsage在累加接口上统一；
• incr(other)/__add__：原地或不可变地把两段 usage 相加。源码注释特别警告__add__只能用于合并同一响应不同片段的用量，不能用于累加多个请求（会破坏某些计价计算）；
• extract()类方法：从原始 API 响应数据中用 genai-prices 抽取 usage（usage.py#L305-L338）。

RequestUsage.extract( data, # 供应商原始响应 provider='anthropic', # 实际 provider ID provider_url='https://api.anthropic.com/v1/', provider_fallback='anthropic', # 快照中查不到 provider 时的回退 ID api_flavor='chat', # 如 OpenAI 的 'chat' 或 'responses' )

extract会遍历iter_provider_references给出的候选 (provider, url) 组合，逐个尝试get_snapshot().find_provider(...).extract_usage(...)，全部失败时返回一个仅含details的空用量——即“抽取失败不抛错，只是没有 token 数”。

RunUsage：一次 Agent 运行的聚合用量

RunUsage额外携带运行维度的两个计数字段，职责定位是“模型负责计算单请求用量，Pydantic AI 只负责跨请求求和”：

• requests：本次运行发出的 LLM API 请求数；
• tool_calls：成功执行的工具调用数。

聚合接口与RequestUsage对应，但更丰富：

• incr()接受RunUsage | RequestUsage（跨请求累加会同时加上requests与tool_calls）；
• __add__同理，便于把多次运行的用量求和；
• __sub__(other)：返回从other时刻以来逐字段的增量（usage.py#L396-L418）。这服务于嵌套操作共享同一个可变RunUsage对象、却只想报告“本操作新增了多少”的场景；cost的差值在任一侧为None时保持None。

成本累计行为有专门的回归测试 tests/test_cost.py 锁定：每次模型响应追加进运行时，其尽力而为的 USD 成本会累加到RunUsage.cost；流式响应必须在流消费完毕后再计价（测试注释明确了这一历史缺陷）；无法计价的模型（如TestModel/FunctionModel）不贡献成本也不告警；意外的计价失败则以CostCalculationFailedWarning呈现而非让运行崩溃。

UsageLimits：运行预算与强制时机

UsageLimits是一个关键字参数 dataclass，用于在运行时限制 token、请求数、工具调用数与成本。完整字段与默认值（取自源码定义）：

所有限额置None即关闭该限制。注意默认UsageLimits()本身就带request_limit=50的兜底：从agent/__init__.py看，run/iter等入口在未显式传入时执行usage_limits = usage_limits or _usage.UsageLimits()，所以一次普通运行默认最多 50 个模型请求；而会话（session）类运行则不同——未传限时会构造UsageLimits(request_limit=None)，即限额对会话是显式开启的（agent/init.py#L3364-L3369）。

校验时机：请求前、响应后、工具执行前

UsageLimits提供五个check_*方法，全部通过抛出UsageLimitExceeded触发限额。从_agent_graph.py的调用点看，执行图把它们插在三个位置：

1. 发请求前（_agent_graph.py中check_before_request两处调用点，L1706-L1708、L1820）：检查request_limit（“下一次请求将超过上限”）、累计input_tokens_limit/total_tokens_limit与cost_limit的当前值；
2. 收到响应后（L1885-L1892）：check_tokens校验output_tokens_limit等累计 token 上限，check_cost校验成本；若未设任何 token 限额，has_token_limits()返回False，流式迭代器可整体跳过这段处理以省开销；
3. 执行工具前（_tool_execution.py）：check_before_tool_call用“投影用量”（当前tool_calls+ 本批待执行调用数）校验tool_calls_limit。若模型并行返回的工具调用整体越限，则一个都不执行。

这种“后置 token 校验、前置请求/工具校验”的组合，源于两类信息的可得性：token 数只能来自供应商响应（或显式预计），而请求数与工具调用数由框架自己掌握。

实战示例

以下示例继承自官方文档 docs/agent.md 的 Usage Limits 章节，均通过run/run_sync/run_stream的usage_limits参数传入。

限制输出 token：

from pydantic_ai import Agent, UsageLimitExceeded, UsageLimits agent = Agent('anthropic:claude-sonnet-4-6') result_sync = agent.run_sync( 'What is the capital of Italy? Answer with just the city.', usage_limits=UsageLimits(output_tokens_limit=10), ) print(result_sync.output) #> Rome print(result_sync.usage) #> RunUsage(cost=Decimal('0.000201'), input_tokens=62, output_tokens=1, requests=1) try: result_sync = agent.run_sync( 'What is the capital of Italy? Answer with a paragraph.', usage_limits=UsageLimits(output_tokens_limit=10), ) except UsageLimitExceeded as e: print(e) #> Exceeded the output_tokens_limit of 10 (output_tokens=32). ...

tests/test_usage_limits.py 用TestModel对每一类限额都做了精确断言，例如输入超限时报Exceeded the input_tokens_limit of 5 (input_tokens=59)，请求数超限时报The next request would exceed the request_limit of 1——与上面异常消息格式一致。

用request_limit防无限循环：

from typing_extensions import TypedDict from pydantic_ai import Agent, ModelRetry, UsageLimitExceeded, UsageLimits class NeverOutputType(TypedDict): """Never ever coerce data to this type.""" never_use_this: str agent = Agent( 'anthropic:claude-sonnet-4-6', retries={'tools': 3}, output_type=NeverOutputType, system_prompt='Any time you get a response, call the `infinite_retry_tool` to produce another response.', ) @agent.tool_plain(retries=5) # (1) 该工具可重试 5 次，模拟可能卡住的重试循环 def infinite_retry_tool() -> int: raise ModelRetry('Please try again.') try: result_sync = agent.run_sync('Begin infinite retry loop!', usage_limits=UsageLimits(request_limit=3)) # (2) except UsageLimitExceeded as e: print(e) #> The next request would exceed the request_limit of 3. ...

运行在第 3 次请求前被拦截，从而阻止无限工具调用。（2）处request_limit检查发生在发请求之前，所以越限的那次请求根本不会发出、也不会计费。

用tool_calls_limit封顶工具调用：

from pydantic_ai import Agent from pydantic_ai.exceptions import UsageLimitExceeded from pydantic_ai.usage import UsageLimits agent = Agent('anthropic:claude-sonnet-4-6') @agent.tool_plain def do_work() -> str: return 'ok' try: # 允许本次运行最多执行 1 次工具调用 agent.run_sync('Please call the tool twice', usage_limits=UsageLimits(tool_calls_limit=1)) except UsageLimitExceeded as e: print(e) #> The next tool call(s) would exceed the tool_calls_limit of 1 (tool_calls=2). ...

工具与 capability 还可以从RunContext读取当前运行的限额：ctx.usage_limits（运行中始终为UsageLimits实例，仅在运行外才可能为None），配合ctx.usage（已发生的用量）可写出“预算感知”的工具——披露或适配剩余预算，而不必再单独配置一份限额副本。按约定它是只读的：它就是运行正在对照执行的那个活对象，改字段会改变后续请求的强制行为。

用per_request_input_tokens_limit限制单请求上下文：

from pydantic_ai import Agent, UsageLimitExceeded, UsageLimits agent = Agent('anthropic:claude-sonnet-4-6') try: agent.run_sync( 'What is the capital of Italy? Answer with just the city.', usage_limits=UsageLimits(per_request_input_tokens_limit=10), ) except UsageLimitExceeded as e: print(e) #> Exceeded the per_request_input_tokens_limit of 10 (request_input_tokens=62). ...

前面几类 token 限额都是整次运行累计的；此项则针对单个请求的输入规模（实际发给模型的上下文窗口大小）。在 prompt 缓存使“累计输入”不再好地代理成本的场景下，限制单请求上下文更对症：过大的上下文既拖累模型表现，又是 cache-miss 的主要花费来源。注意该限额统计的是归一化后的input_tokens（含缓存前缀），所以它限制的是上下文大小而非 cache-miss 成本。默认在响应后按供应商报告的input_tokens校验，因此越大的请求已经发出并计费（与input_tokens_limit语义一致）；设为count_tokens_before_request=True后可提前拦截：

usage_limits=UsageLimits( per_request_input_tokens_limit=20_000, count_tokens_before_request=True, # 发请求前先 count_tokens，超限则不发送 )

count_tokens_before_request会调用模型的count_tokensAPI，带来额外开销，故默认关闭。源码注释声明当前支持：Anthropic、Google、Bedrock Converse、OpenAI Responses（usage.py#L494-L500）。开启后，cost_limit也会把预计的输入 token 先计价，若仅输入下限就超成本上限则直接拒发请求。

用cost_limit封顶美元支出：

from decimal import Decimal from pydantic_ai import Agent, UsageLimitExceeded, UsageLimits agent = Agent('anthropic:claude-sonnet-4-6') try: agent.run_sync( 'What is the capital of Italy? Answer with just the city.', usage_limits=UsageLimits(cost_limit=Decimal('0.0001')), ) except UsageLimitExceeded as e: print(e) #> Exceeded the `cost_limit` of 0.0001 (`usage.cost`=Decimal('0.000201')). ...

与output_tokens_limit相同，cost_limit在每次响应后校验，因为输出成本要等响应到达才知道。成本是尽力而为的：genai-prices 无定价数据的模型/provider 得到cost=None；此时若配置了cost_limit，运行会发出CostNotFoundWarning而不是被静默放行；意外的计价失败发CostCalculationFailedWarning。文档同时提醒不要把cost_limit当作硬性的账单保证——应与request_limit或供应商自身的消费控制搭配使用。

pydantic_ai.prices：保持价格表新鲜

成本估算依赖 genai-prices 的价格快照。Pydantic AI 在发布时打包一份模型价格表（tests/test_usage_limits.py#L44-L46 中可直接用calc_price(usage, model_ref='gpt-4o')得到快照价），但发布后新上线的模型无法计价——此时RequestUsage.cost的 docstring 指向update_in_background()作为解决方案：

"""Keep model prices up to date by downloading the latest price list in the background.""" from genai_prices import UpdatePrices def update_in_background() -> UpdatePrices: """...""" updater = UpdatePrices() updater.start() return updater

（源码见 prices.py#L10-L27）

语义要点：

• 立即下载一次，之后每小时刷新，全程在后台线程执行，下载从不阻塞业务代码；
• 失败不降级可用性：某次下载失败时，最近一次成功下载的价格表继续使用，失败记录到genai-priceslogger；
• 返回UpdatePrices更新器实例，应用退出时应调用其stop()；
• 需要自定义下载地址或调度周期时，直接使用genai_prices.UpdatePrices，它与本函数共享同一个后台下载任务，不会产生双份下载。

典型接入方式是在应用启动时调用一次：

from pydantic_ai.prices import update_in_background updater = update_in_background() # ... 应用生命周期 ... updater.stop()

测试与验证入口

本模块的行为主要由两组测试锁定，便于改动时回归验证：

• tests/test_usage_limits.py：覆盖各类UsageLimits的触发路径（输入/输出/总 token、请求数、工具调用数）、流式运行中的限额校验、UsageLimits作为RunContext字段的序列化，以及 genai-prices 计价快照（如gpt-4o下 100 输入 + 50 输出 =Decimal('0.00075')）；
• tests/test_cost.py：锁定RunUsage.cost的跨响应累加，包括流式响应“消费完才计价”的回归、不可计价模型零贡献且无告警、计价失败降级为CostCalculationFailedWarning。

小结

pydantic_ai.usage提供“包含式”token 桶模型（input_tokens恒含缓存与音频分量）、跨供应商归一化的 usage 提取（RequestUsage.extract）、运行级聚合与差值（RunUsage的incr/__sub__）以及面向 OTel 的防双计属性导出；UsageLimits以默认request_limit=50兜底，并在请求前、响应后、工具执行前三个点位强制 token/请求/工具/成本四维限额。pydantic_ai.prices则以一个后台小时级价格刷新器，保证cost字段与cost_limit在模型持续更新的前提下依然有效。三者配合，即可在 pydantic-ai 中实现从“可观测的用量”到“可执行的预算护栏”的完整闭环。

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai