ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Агент с cost tracking

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

Разработать LLM-агента (на базе LangChain / LlamaIndex), который в реальном времени отслеживает стоимость каждого вызова модели, агрегирует затраты за день и отправляет алерт при превышении дневного бюджета в $10. Система должна быть интегрирована с прометеевским стеком мониторинга (Prometheus + Grafana + Alertmanager) или, как минимально жизнеспособное решение, с локальными логами и простым скриптом оповещения.

Ключевой результат Работающий пайплайн сбора, хранения и визуализации cost‑метрик LLM-агента, а также настроенный алерт, срабатывающий при превышении порога $10/день.

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

Перед началом работы необходимо иметь:

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

1. Используем litellm с флагом mock_request=True для эмуляции затрат (лимиты токенов генерируются случайно).
2. Для Prometheus — запускаем локально в Docker (docker run -p 9090:9090 prom/prometheus).
3. Если нет Grafana — алерт можно реализовать через отправку e-mail (smtplib) или Telegram-бота из Python.
4. Бюджет $10/день настраивается константой в коде (реальная оплата не требуется, все цены берутся из публичного прайс-листа провайдера).

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

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

Этап 1: Создание агента с подсчётом токенов (1 – 1.5 часа)

Действия

1. Создать виртуальное окружение и установить зависимости:

   pip install langchain langchain-openai tiktoken litellm prometheus-client
   

2. Написать класс CostTracker, который оборачивает вызов LLM и логирует количество входных/выходных токенов, а также вычисляет стоимость по актуальным тарифам. Пример структуры:

   import tiktoken
   from datetime import datetime, date
   import sqlite3
   
   class CostTracker:
       PRICING = {
           "gpt-4o": {"input": 5.0 / 1e6, "output": 15.0 / 1e6}
       }
       
       def __init__(self, db_path="costs.db"):
           self.conn = sqlite3.connect(db_path)
           self.conn.execute("CREATE TABLE IF NOT EXISTS usage (ts TEXT, model TEXT, tokens_input INT, tokens_output INT, cost REAL)")
       
       def log_call(self, model, tokens_input, tokens_output):
           cost = tokens_input * self.PRICING[model]["input"] + tokens_output * self.PRICING[model]["output"]
           self.conn.execute("INSERT INTO usage VALUES (?, ?, ?, ?, ?)",
                             (datetime.utcnow().isoformat(), model, tokens_input, tokens_output, cost))
           self.conn.commit()
           return cost
   

3. Интегрировать CostTracker в агента через callback LangChain или декоратор. Использовать langchain.callbacks.BaseCallbackHandler:

   class CostCallback(BaseCallbackHandler):
       def __init__(self, tracker: CostTracker):
           self.tracker = tracker
       
       def on_llm_end(self, response, **kwargs):
           if hasattr(response.llm_output, 'token_usage'):
               tu = response.llm_output.token_usage
               self.tracker.log_call(
                   model=response.llm_output.model_name,
                   tokens_input=tu.prompt_tokens,
                   tokens_output=tu.completion_tokens
               )
   

4. Проверить корректность на 2–3 простых запросах: сумма цен из трекера должна совпадать с расчётной по формуле.

Ожидаемый результат этапа Класс CostTracker и callback, корректно логирующий каждый вызов LLM в SQLite.

Этап 2: Агрегация дневных затрат и экспорт метрик (1 час)

Действия

1. Реализовать функцию daily_spend(tracker) -> float, которая суммирует стоимость всех записей за текущие сутки:

   def daily_spend(tracker):
       today = date.today().isoformat()
       cursor = tracker.conn.execute(
           "SELECT SUM(cost) FROM usage WHERE date(ts) = ?", (today,)
       )
       return cursor.fetchone()[0] or 0.0
   

2. Подготовить метрику Prometheus (Gauge) для ежедневных затрат:

   from prometheus_client import Gauge, push_to_gateway
   DAILY_SPEND = Gauge('llm_daily_spend_usd', 'Total LLM spend today')
   
   def push_metric(gateway="localhost:9091"):
       DAILY_SPEND.set(daily_spend(tracker))
       push_to_gateway(gateway, job='llm_agent', registry=registry)
   

3. Запустить Pushgateway (Docker):

   docker run -d -p 9091:9091 prom/pushgateway
   

4. Обернуть вызовы агента таким образом, чтобы после каждого LLM-запроса (или раз в минуту) вызывался push_metric. Удобно использовать фоновый поток, отправляющий метрику по таймеру.

Ожидаемый результат этапа В Pushgateway появляется метрика llm_daily_spend_usd, обновляемая после каждого запроса.

Этап 3: Настройка Prometheus и Grafana (1 час)

Действия

1. Создать конфиг Prometheus (prometheus.yml):

   scrape_configs:
     - job_name: 'pushgateway'
       honor_labels: true
       static_configs:
         - targets: ['localhost:9091']
   

2. Запустить Prometheus (Docker):

   docker run -d -p 9090:9090 -v $(pwd)/prometheus.yml:/etc/prometheus/prometheus.yml prom/prometheus
   

3. Запустить Grafana (Docker):

   docker run -d -p 3000:3000 grafana/grafana
   

4. Создать дашборд «LLM Daily Spend» с метрикой llm_daily_spend_usd:

   • Тип графика: Gauge (числовое отображение) или Time Series (часы).
   • Добавить линию тренда за последние 24 часа.
   • Выгрузить JSON дашборда для сохранения в репозиторий.

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

Этап 4: Алертинг при превышении бюджета (1 час)

Действия

1. Настроить Alertmanager (Docker):

   • alertmanager.yml с получателем (например, webhook или email):

     route:
       receiver: 'admin'
     receivers:
       - name: 'admin'
         email_configs:
           - to: 'admin@example.com'
             from: 'alertmanager@example.com'
             smarthost: 'smtp.example.com:587'
     

2. Добавить правило алерта в Prometheus (alerts.yml):

   groups:
     - name: llm_cost
       rules:
         - alert: DailyLLMBudgetExceeded
           expr: llm_daily_spend_usd > 10
           for: 1m
           labels:
             severity: warning
           annotations:
             summary: "Дневной лимит $10 превышен! Текущий расход: {{ $value }} USD"
   

3. Подключить файл правил в prometheus.yml:

   rule_files: ['alerts.yml']
   

4. Протестировать алерт: выполнить серию LLM-запросов, пока симуляция не даст daily_spend > 10. Убедиться, что Prometheus переходит в состояние Pending -> Firing, а уведомление приходит на почту / Telegram / Slack.

5. Если нет Alertmanager — написать простой Python-скрипт, который раз в минуту читает SQLite и отправляет e-mail при превышении лимита:

   while True:
       if daily_spend(tracker) > 10:
           send_email("Budget exceeded!")
       time.sleep(60)
   

Ожидаемый результат этапа Алерт срабатывает в течение 1 минуты после превышения дневного бюджета $10.

Этап 5: Тестирование и документирование (30 минут)

Действия

1. Написать unit-тесты для CostTracker (проверить корректность суммирования, работу с разными моделями).
2. Провести интеграционный тест — запустить агента на 10 запросах, убедиться, что метрика в Pushgateway меняется.
3. Проверить алерт: симулировать превышение лимита (временно уменьшить бюджет до $0.01).
4. Зафиксировать код, конфиги Prometheus, Alertmanager, JSON дашборда Grafana в Git-репозитории.
5. Написать README с инструкцией по запуску.

Ожидаемый результат этапа Репозиторий с полным кодом и конфигами, все тесты проходят.

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

• Агент выполняет LLM-запросы, а CostTracker корректно считает количество токенов и стоимость.
• Затраты сохраняются в SQLite (или CSV) с временной меткой.
• Метрика llm_daily_spend_usd отправляется в Pushgateway и видна в Prometheus.
• Дашборд Grafana отображает текущий расход за день.
• При превышении порога $10 (или тестового лимита $0.01) алерт срабатывает и уведомление доходит до получателя.
• Все компоненты (агент, Pushgateway, Prometheus, Grafana, Alertmanager) запускаются одной командой docker-compose up или через Makefile.
• Написаны минимум 2 unit-теста для трекера и 1 интеграционный тест для метрики.
• README содержит полное описание, команды запуска и пример алерта.

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

Артефакты

• agent_cost_tracker.py — модуль с классами CostTracker и CostCallback.
• agent_runner.py — скрипт, запускающий агента с фоновой отправкой метрик.
• docker-compose.yml — сервисы Prometheus, Pushgateway, Grafana, Alertmanager.
• prometheus/ — конфиг Prometheus и файл правил алерта.
• grafana_dashboard.json — готовый дашборд «LLM Daily Spend».
• alertmanager.yml — конфиг Alertmanager (или скрипт budget_alerter.py).
• tests/ — тесты для CostTracker.
• README.md — документация.

Содержание README краткое описание архитектуры, команды для поднятия стека, пример вызова агента, скриншот дашборда и скриншот алерта (текст сообщения).

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

• Интеграция с Telegram-ботом для алертов.
• Автоматический сброс дневных затрат в полночь.
• Multi-model поддержка (несколько провайдеров с разными ценами).

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

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

Примечание Если вы впервые работаете с Prometheus / Grafana, заложите дополнительно 1 час на знакомство с интерфейсами.

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

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

• Я проверил, что стоимость считается по актуальным ценам провайдера (или симулируется).
• Я убедился, что метрика llm_daily_spend_usd действительно обновляется в Prometheus после каждого запроса агента.
• Я протестировал алерт: принудительно уменьшил лимит до $0.01 и выполнил 1–2 вызова — алерт сработал.
• Я зафиксировал все конфиги (Prometheus, Alertmanager, дашборд Grafana) в репозитории.
• Я написал инструкцию в README, чтобы любой коллега мог запустить проект за 5 минут одной командой docker-compose up.