ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Настроить token economics для агентов

1. Цель задачи

Разработать и внедрить систему контроля затрат токенов для AI-агентов, работающих в production-среде. Решение должно включать выделение бюджета на каждого агента, динамические лимиты на вызовы и аудит расходов для предотвращения неконтролируемого роста затрат (runaway costs). Ключевой результат Работающая инфраструктура, которая автоматически ограничивает потребление токенов агентом в пределах заданного бюджета и предоставляет прозрачную отчётность по всем расходам.

2. Исходные данные

Если нет реального инструмента — симулируем:

1. Создать трёх агентов с разными задачами:
   • support_agent — отвечает на вопросы пользователей (средняя длина ответа ~200 токенов)
   • analytics_agent — генерирует SQL-запросы и объяснения (средняя длина ~500 токенов)
   • creative_agent — пишет тексты (средняя длина ~1000 токенов)
2. Написать простой скрипт, который будет циклически вызывать агентов в случайном порядке, имитируя нагрузку.
3. Использовать SQLite как simplestore для логов и бюджета (схема: agent_id, timestamp, model, prompt_tokens, completion_tokens, cost, action).
4. Если нет Prometheus/Grafana — выводить метрики в консоль и сохранять CSV-файл для последующего анализа в Jupyter.

3. Технологический стек

4. Этапы выполнения

Этап 1: Проектирование модели бюджета (30 минут)

Действия

1. Определить единицы бюджета

   • Дневной лимит токенов на агента (например, support_agent: 100000 tkn/day)
   • Месячный лимит расходов в $ (например, $5/мес на агента)
   • Дополнительно: лимит на один вызов (max_prompt + max_completion)

2. Спроектировать структуру данных для хранения бюджета:

   {
     "agent_id": "support_agent",
     "daily_token_limit": 100000,
     "monthly_cost_limit": 5.00,
     "current_day_tokens": 0,
     "current_month_cost": 0.0,
     "reset_day_at": "00:00 UTC",
     "reset_month_at": "first day of month",
     "overage_policy": "BLOCK"
   }
   

3. Решить, как сбрасывать счётчики:

   • Дневной сброс — по UTC-полуночи (можно хранить последний день сброса и сбрасывать при первом запросе нового дня)
   • Месячный — по первому числу месяца

4. Определить пороги оповещения

   • 80% от лимита — WARNING (логирование + алерт)
   • 100% — BLOCK (лимит исчерпан, агент не отвечает)

Ожидаемый результат этапа Документ с описанием модели бюджета (Markdown) и JSON-схема конфигурации.

Этап 2: Реализация счётчика и хранилища (1 час)

Действия

1. Написать модуль token_counter.py

   • Функция count_tokens(text, model="gpt-4o") — использует tiktoken для OpenAI, anthropic.get_tokenizer() для Claude.
   • Функция estimate_cost(prompt_tokens, completion_tokens, model) — по ценам из документации провайдера.

2. Создать хранилище с SQLite

   • Таблица usage_log:

     CREATE TABLE usage_log (
       id INTEGER PRIMARY KEY AUTOINCREMENT,
       agent_id TEXT NOT NULL,
       request_id TEXT UNIQUE,
       model TEXT,
       prompt_tokens INTEGER,
       completion_tokens INTEGER,
       total_tokens INTEGER,
       cost REAL,
       timestamp TEXT DEFAULT (datetime('now')),
       status TEXT -- 'success', 'blocked', 'warning'
     );
     

   • Таблица budget_config:

     CREATE TABLE budget_config (
       agent_id TEXT PRIMARY KEY,
       daily_token_limit INTEGER,
       monthly_cost_limit REAL,
       policy TEXT DEFAULT 'BLOCK',
       updated_at TEXT
     );
     

   • Таблица budget_state:

     CREATE TABLE budget_state (
       agent_id TEXT PRIMARY KEY,
       current_day_tokens INTEGER DEFAULT 0,
       current_month_cost REAL DEFAULT 0.0,
       last_reset_day TEXT,
       last_reset_month TEXT
     );
     

3. Написать функции для атомарного обновления состояния бюджета:

   • check_and_deduct(agent_id, prompt_tokens, completion_tokens) — проверяет лимиты, если можно — вычитает и записывает лог, иначе возвращает ошибку.
   • Использовать SQLite транзакции с BEGIN IMMEDIATE для избежания гонок.

Ожидаемый результат этапа Модуль budget_controller.py с функциями check_and_deduct, reset_budgets, log_usage, и get_budget_status.

Этап 3: Интеграция с агентами (1 час)

Действия

1. Обернуть вызов LLM в декоратор/функцию-посредник

   def with_budget(agent_id: str, model: str):
       def decorator(func):
           def wrapper(*args, **kwargs):
               # 1. Измерить prompt_tokens (через tiktoken)
               prompt_text = ...  # получить из args/kwargs
               prompt_tokens = count_tokens(prompt_text, model)
               
               # 2. Проверить бюджет
               budget_ok = pre_check(agent_id, prompt_tokens)
               if not budget_ok:
                   log_block(agent_id, prompt_tokens)
                   raise BudgetExceededError("Daily/monthly limit reached")
               
               # 3. Вызвать LLM
               result = func(*args, **kwargs)
               
               # 4. Измерить completion_tokens
               completion_text = result["choices"][0]["message"]["content"]
               completion_tokens = count_tokens(completion_text, model)
               
               # 5. Списать и залогировать
               check_and_deduct(agent_id, prompt_tokens, completion_tokens, model)
               
               return result
           return wrapper
       return decorator
   

2. Создать трёх агентов с разными бюджетами:

   • support_agent — daily 50k tkn, monthly $2
   • analytics_agent — daily 100k tkn, monthly $5
   • creative_agent — daily 200k tkn, monthly $10
   • Все с политикой BLOCK (блокировать при превышении).

3. Написать тестовый скрипт, имитирующий нагрузку:

   • Цикл из 50 итераций, на каждой случайный выбор агента и случайный запрос (несколько шаблонов).
   • После каждых 10 итераций вызывать print_status(agent_id) для вывода остатков бюджета.

Ожидаемый результат этапа Работающие агенты, которые при исчерпании бюджета возвращают ошибку и не тратят токены.

Этап 4: Мониторинг и алертинг (30 минут)

Действия

1. Настроить Prometheus-метрики (если есть Prometheus):

   from prometheus_client import Counter, Gauge, Histogram, generate_latest
   
   tokens_total = Counter('llm_tokens_total', 'Total tokens consumed', ['agent', 'model'])
   cost_total = Counter('llm_cost_total', 'Total cost in USD', ['agent'])
   budget_remaining = Gauge('llm_budget_remaining_tokens', 'Remaining daily tokens', ['agent'])
   blocked_requests = Counter('llm_blocked_requests_total', 'Blocked requests due to budget', ['agent'])
   

2. Или, для упрощения, логировать алерты в файл:

   • budget_warning.log — запись при превышении 80% лимита
   • budget_blocked.log — запись при блокировке

3. Создать простой дашборд (если Grafana):

   • Панель "Tokens per agent (daily)"
   • Панель "Cost per agent (monthly)"
   • Панель "Blocked requests over time"
   • Или же вывести текстовую таблицу в консоль каждые 5 вызовов.

4. Добавить алерт при блокировке

   • В реальной системе — Telegram/Slack webhook.
   • В учебной — печать ALERT: Agent {agent_id} blocked в консоль.

Ожидаемый результат этапа Скрипт мониторинга, который в реальном времени показывает состояние бюджета и оповещает о превышениях.

Этап 5: Аудит и отчёт (30 минут)

Действия

1. Написать скрипт audit_report.py

   • Выгрузить данные из SQLite (или CSV) в Pandas DataFrame.
   • Рассчитать:
     • Общее потребление токенов и стоимость по агентам.
     • Пиковые часы потребления (группировка по часам).
     • Средняя стоимость на один запрос.
   • Построить гистограмму распределения токенов на запрос.

2. Сгенерировать отчёт в формате Markdown

   # Отчёт аудита token economics
   ## Сводка
   | Agent | Total tokens | Total cost $ | Requests | Avg tokens/req |
   ...
   ## Аномалии
   - Выбросы: 3 запроса > 2x стандартного отклонения
   - Блокировки: 2 раза (support_agent превысил дневной лимит)
   ## Рекомендации
   - Увеличить лимит support_agent на 20%
   

3. Сохранить отчёт в файл audit_report_YYYY-MM-DD.md

Ожидаемый результат этапа Файл с аудит-отчётом, содержащим наглядную статистику и рекомендации.

5. Критерии приемки (Definition of Done)

• Реализован модуль бюджетирования, который хранит лимиты и текущее состояние в SQLite.
• При вызове агента с превышением дневного или месячного лимита возвращается ошибка и запись в логе типа blocked.
• Счётчик токенов корректно считает как промпт, так и completion (с использованием tiktoken/anthropic tokenizer).
• Каждый вызов агента фиксируется в таблице usage_log с полными метриками (tokens, cost, timestamp, agent).
• Мониторинг выводит текущее состояние бюджетов в консоль (или дашборд) в реальном времени.
• Сгенерирован аудит-отчёт с анализом затрат и аномалий.
• Политика BLOCK работает атомарно: не происходит списания, если лимит уже исчерпан.
• Тестовый сценарий с 50+ вызовами успешно демонстрирует блокировку хотя бы одного агента (для проверки можно искусственно занизить лимит).

6. Ожидаемый результат

Основной артефакт репозиторий с кодом, содержащий:

• budget_controller.py — логика бюджета и хранилища.
• token_counter.py — подсчёт токенов и стоимости.
• agents.py — агенты с декоратором бюджета.
• load_simulator.py — скрипт симуляции нагрузки.
• monitor.py / dashboard.py — мониторинг (консольный или Prometheus).
• audit_report.py — генерация отчёта.
• config.yaml — конфигурация лимитов.
• Сгенерированный отчёт audit_report_*.md.
• SQLite база данных (или .db файл в папке data/) с логами тестового прогона.

Дополнительные результаты (опционально):

• Docker Compose файл с Prometheus + Grafana (если включён).
• Скрипт для сброса бюджета вручную.
• Документация в README.

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я реализовал атомарную проверку и списание бюджета (без race condition).
• Я протестировал сценарий, когда лимит исчерпан — агент возвращает ошибку и не тратит деньги.
• Я убедился, что счётчик токенов считает промпт и completion раздельно и суммарно.
• Я сгенерировал аудит-отчёт, который показывает хотя бы одну аномалию.
• Я задокументировал конфигурацию бюджетов в отдельном файле (YAML/JSON) и могу легко менять лимиты.
• Я добавил логирование предупреждений при 80% лимита.
• Я проверил, что при сбросе (новый день/месяц) бюджет корректно обнуляется.