Guardrails: Budgets, Timeouts, and Human-in-the-Loop

Автономность без лимитов — это неограниченная доверенность на ваш кошелёк и ваши системы. Потолок шагов из урока 2 — необходимый минимум, но его легко обойти: агент может уложиться в десять шагов, раздув каждый до сотен тысяч токенов. Поэтому бюджет должен быть многомерным: шаги, токены, деньги, wall-clock время — и проверяться перед каждым дорогим действием, а не после.

from dataclasses import dataclass, field
import time

@dataclass
class Budget:
    max_steps: int = 15
    max_tokens: int = 200_000
    max_usd: float = 0.50
    max_seconds: float = 120.0
    steps: int = 0
    tokens: int = 0
    usd: float = 0.0
    started: float = field(default_factory=time.monotonic)

    def charge(self, usage, price_in, price_out):
        self.steps += 1
        self.tokens += usage.total_tokens
        self.usd += (usage.prompt_tokens * price_in
                     + usage.completion_tokens * price_out) / 1e6

    def exceeded(self) -> str | None:
        if self.steps >= self.max_steps: return "steps"
        if self.tokens >= self.max_tokens: return "tokens"
        if self.usd >= self.max_usd: return "usd"
        if time.monotonic() - self.started >= self.max_seconds: return "time"
        return None

Поле usage приходит в каждом ответе API — используйте его, а не оценки на глаз. Ключевая деталь UX: при исчерпании бюджета не обрывайте прогон молча. Дайте модели один финальный вызов без инструментов с просьбой подвести итог: что успели, что осталось, что рекомендуется дальше. Частичный отчёт ценнее пустого таймаута.

Порядок проверок имеет значение: exceeded() вызывается перед обращением к модели и перед каждым пишущим инструментом — дешёвая проверка до дорогого действия, а не после. Значения лимитов подбирайте от задачи: интерактивному помощнику хватит пятнадцати шагов и пары минут, фоновому batch-прогону — сотни шагов, но жёсткий долларовый потолок. И заведите отдельный, более щедрый пресет для отладки: ловить ложные срабатывания бюджета на этапе разработки так же вредно, как не ловить настоящие в проде.

Бюджеты ограничивают объём работы, таймауты и песочница — её радиус поражения. Здесь работает принцип вложенных таймаутов: у каждого сетевого вызова инструмента свой таймаут (10-30 секунд), у вызова модели — свой (60-120 секунд, стриминг помогает заметить зависание раньше), у прогона целиком — общий потолок из бюджета. Наружный таймаут без внутренних — плохая конструкция: он сработает, но вы не узнаете, кто именно завис.

Песочница — это ответ на вопрос «что худшее может сделать этот инструмент, если модель ошибётся или её обманут». Минимальный набор для агента, который трогает файлы и сеть: файловые операции заперты в рабочем каталоге (нормализуйте путь и проверяйте, что он внутри корня, — иначе агент однажды прочитает ~/.ssh/id_rsa по вполне невинной причине); сетевые запросы идут только на allowlist доменов, а не «куда угодно»; секреты не лежат в промпте — инструмент сам подставляет токены из окружения, модель их не видит и не может утечь.

Отдельный режим, который стоит встроить с первого дня, — dry-run: все пишущие инструменты не исполняются, а возвращают «[dry-run] создал бы комментарий: ...». Это дёшево в реализации (один флаг в реестре инструментов) и бесценно в отладке: вы смотрите полный план действий агента на реальных данных, ничем не рискуя. Капстоун этого курса начнётся именно с dry-run-режима, и это общий паттерн запуска любого агента в новое окружение: сначала неделя dry-run и чтение логов, потом включение записи.

В уроке 3 мы пометили инструменты как readonly, write и irreversible. Теперь эта метаданная становится политикой: readonly исполняется молча, write — по правилам (лимит на количество, dry-run в новых окружениях), irreversible — только через явное подтверждение человека.

def execute(call, policy):
    tool = REGISTRY[call.name]
    if tool.danger == "readonly":
        return tool.fn(**call.args)
    if tool.danger == "irreversible" or policy.confirm_writes:
        answer = ask_human(f"Агент хочет: {call.name}({call.args}). Разрешить?")
        if answer != "yes":
            return f"отклонено человеком: {answer or 'без причины'}"
    return tool.fn(**call.args)

Обратите внимание: отказ человека возвращается модели как результат инструмента, с причиной. Это не сбой, а информация — агент скорректирует план («хорошо, тогда только предложу метки в комментарии, без изменения issue»). В консольном скрипте ask_human — это input(); в сервисе — запись «ожидает подтверждения» в состояние прогона и уведомление, к этому вернёмся в следующем уроке.

Три правила зрелого HITL. Первое: подтверждение показывает полные аргументы, а не только имя инструмента — «send_email(to=all-staff@...)» и «send_email(to=мне)» — разные решения. Второе: агрегируйте, где можно, — двадцать однотипных подтверждений подряд приучают человека жать «да» не глядя, и весь шлюз обесценивается; лучше одно подтверждение плана из двадцати пунктов. Третье: каждое решение человека пишется в аудит-лог вместе с шагами агента — JSONL-файл, где на строку приходится одно событие: вызов модели, вызов инструмента, результат, вердикт человека, счётчики бюджета. Когда агент сделает что-то странное — а он сделает, — этот лог будет единственным способом восстановить картину и починить первопричину, а не симптом.