Перейти к содержимому

Агент проедает бюджет за один run

Дневной лимит на LLM API ловит «за сутки перерасходовали» — но один автономный агент способен сжечь $40 за один run, если ушёл в дорогую петлю: длинный контекст, повтор tool-calls, retry за retry-ем. К утру дневной алёрт уже бесполезен — деньги потрачены.

Нужен guard внутри одного run-а: он считает накопленную стоимость по ходу дела и реагирует до того, как счёт вырастет в разы. Здесь — не про дневной расход, а именно про runaway в пределах одного прогона.

Оборачиваем каждый вызов модели: считаем токены → доллары, копим сумму, и на двух порогах (warn / hard) шлём push. На hard — поднимаем исключение и останавливаем агента.

import os, requests
# цены за 1M токенов (пример — подставьте свои для вашей модели)
PRICE_IN = 3.00 / 1_000_000
PRICE_OUT = 15.00 / 1_000_000
class BudgetGuard:
def __init__(self, run_id, warn_usd=5.0, hard_usd=15.0):
self.run_id = run_id
self.warn = warn_usd
self.hard = hard_usd
self.spent = 0.0
self._warned = False
def add(self, usage):
# usage — объект ответа модели с input_tokens / output_tokens
self.spent += usage.input_tokens * PRICE_IN
self.spent += usage.output_tokens * PRICE_OUT
if self.spent >= self.hard:
notify("💸🛑 Агент пробил hard-лимит",
f"run {self.run_id}: потрачено ${self.spent:.2f} "
f"(потолок ${self.hard:.2f}). Останавливаю run.",
priority=10)
raise RuntimeError(f"budget hard limit: ${self.spent:.2f}")
if self.spent >= self.warn and not self._warned:
self._warned = True
notify("💸⚠️ Агент близок к лимиту бюджета",
f"run {self.run_id}: уже ${self.spent:.2f} "
f"из ${self.hard:.2f}. Присматривайте.",
priority=6)
def snapshot(self):
return round(self.spent, 4)
def notify(title, message, priority):
requests.post(f"{os.environ['NOTIFLY_URL']}/message",
params={"token": os.environ["NOTIFLY_TOKEN"]},
json={"title": title, "message": message, "priority": priority},
timeout=5)

Встраивание в agent-loop:

guard = BudgetGuard(run_id="refactor-auth-2026-07-01")
def call_model(messages):
resp = client.messages.create(model="claude-sonnet-4-5",
max_tokens=2000, messages=messages)
guard.add(resp.usage) # пробьёт лимит — поднимет исключение
return resp
# ... агент крутит цикл; при hard-лимите run падает с алёртом

Ключевое: warn шлёт push один раз (флаг _warned), чтобы не спамить на каждой итерации. hard останавливает — лучше упавший run, чем счёт на $40.

Скорость сжигания: не только абсолютная сумма

Заголовок раздела «Скорость сжигания: не только абсолютная сумма»

Абсолютный потолок ловит долгий run, но не ловит внезапный всплеск. Считаем $/минуту — если агент вдруг начал жечь втрое быстрее обычного, это ранний сигнал петли:

import time
class BurnRateGuard:
def __init__(self, run_id, max_usd_per_min=2.0):
self.run_id = run_id
self.max = max_usd_per_min
self.t0 = time.time()
self.spent = 0.0
self._fired = False
def add(self, delta_usd):
self.spent += delta_usd
mins = max((time.time() - self.t0) / 60, 0.1)
rate = self.spent / mins
if rate > self.max and not self._fired:
self._fired = True
notify("🔥 Скорость сжигания бюджета выросла",
f"run {self.run_id}: ${rate:.2f}/мин "
f"(норма ${self.max:.2f}). Возможно, петля.",
priority=8)

Если агент — чёрный ящик (сторонний CLI, чужой SDK) и обернуть вызовы модели изнутри нельзя, ловите порог снаружи. У OpenAI/Anthropic есть usage-webhooks и budget-алёрты — направьте их в Webhook Router, а он превратит событие в push. Для провайдеров без вебхуков подойдёт Email Inbox: budget-alert приходит письмом → Notifly шлёт push.

Точность ниже (задержка биллинга), зато работает для любого агента без правки его кода.

  • run_id — какой именно прогон, чтобы найти его в логах;
  • потрачено / потолок / скорость $/мин;
  • топ-1 «дорогой» tool или самый длинный запрос за run;
  • намёк на причину: retry-петля, контекст раздулся до 180k токенов.