Agent成本控制配置:给模型调用加上预算与限流
Agent 的成本控制,核心是把“花多少钱”变成代码里可判断的条件,而不是上线后看账单才发现超支。
比较稳妥的做法是三层约束:单次请求限制输出长度,单个任务限制累计 token 或费用,整体调用限制频率和并发。
下面按可直接落地的配置方式说明。
先明确要限制的计量单位
不同模型供应商的计费口径不完全一样,但通常按输入 token 和输出 token 分别计价。
配置前先确认两件事:当前使用的模型名称,以及它在控制台或文档中标注的计价单位。
不要凭印象写死单价,价格会调整,建议把单价放在独立配置项里,方便以后修改。
一个可维护的配置结构大致如下,用 YAML 保存,代码启动时读取:
agent_budget:
model: "your-model-name"
price_input_per_1k: 0.0 # 按实际控制台价格填写
price_output_per_1k: 0.0
max_output_tokens: 1024
task_token_limit: 20000
task_cost_limit: 0.50
rate_limit_per_minute: 30
max_concurrency: 3
这里的 task_token_limit 和 task_cost_limit 是同一个任务所有模型调用的累计值,不是单次值。
只有区分这两个层级,才能挡住“单次不贵、循环几十次”的常见浪费。
在调用链路上做预算判断
配置写好只是第一步,关键是每次调用大模型前先检查预算。
以 Python 伪代码为例,维护一个任务级的累加器:
class Budget:
def __init__(self, cfg):
self.cfg = cfg
self.used_tokens = 0
self.used_cost = 0.0
def allow(self, est_tokens):
if self.used_tokens + est_tokens > self.cfg["task_token_limit"]:
return False
return self.used_cost < self.cfg["task_cost_limit"]
def record(self, in_tokens, out_tokens):
self.used_tokens += in_tokens + out_tokens
self.used_cost += in_tokens / 1000 * self.cfg["price_input_per_1k"]
self.used_cost += out_tokens / 1000 * self.cfg["price_output_per_1k"]
调用前用 allow() 判断,调用后用接口返回的 usage 字段记录实际消耗。不要用自己估算的 token 数代替接口返回值,估算偏差会让预算形同虚设。
超限时的处理也要提前定义:是直接返回错误,还是降级到更便宜的模型,或者把任务截断后交给人工。
限流与并发要一起配置
预算控制的是总量,限流控制的是速度。
两者结合才能避免短时间打满配额。
速率限制可以在进程内用令牌桶实现,多实例部署时建议放到 Redis 等共享存储里,否则每个实例各自计数,实际请求量会翻倍。
并发上限同样重要:一个 Agent 如果同时发起多个子任务,瞬时消耗可能远超预期。
把 max_concurrency 设成较小的值,配合队列排队,比事后限流更容易排查问题。
验证配置是否真的生效
配置完成后,至少做三次检查:
- 构造一个会触发
task_token_limit的长任务,确认调用被拦截并返回预期错误。 - 在日志中打印每次调用的输入输出 token 和累计费用,核对与接口返回是否一致。
- 短时间内连续发起超过
rate_limit_per_minute的请求,确认多余请求被拒绝或排队而不是全部放行。
如果发现预算没有生效,优先检查两处:调用前是否真的走了 allow() 判断,以及多实例场景下计数器是否共享。
容易踩的坑
把单价写死在代码里,模型切换后忘记更新,算出来的费用没有参考价值。
只限制单次输出长度,不限制任务总次数,循环型 Agent 仍会持续消耗。
限流只做在入口网关,内部子任务绕过网关直接调用模型,导致限制失效。
另外,超限后直接抛异常而不做记录,会让后续排查缺少依据,建议在日志中保留任务 ID 和已消耗量。
成本控制的目标不是把 Agent 限制到不能用,而是让超支在发生前被拦截、发生后能追溯。
先按最小配置跑通这三层约束,再根据真实用量逐步收紧阈值,比一开始就设置过严的限制更稳妥。