Token Counting и бюджетирование LLM-запросов: управление расходами

Как не потерять бюджет на LLM: token counting и бюджетирование

Представьте: команда разработчиков интегрировала GPT-4 для обработки запросов клиентов, а спустя месяц счёт за API вырос до $5000 вместо ожидаемых $500. Причина — каждый запрос уходил с системным промптом на 2000 токенов, а дублирующиеся запросы не кэшировались. Token counting и бюджетирование — это не опция, а необходимость для любого продакшена, работающего с LLM. Без них вы рискуете получить счёт, в 5–10 раз превышающий ожидания.

Мы внедрили систему управления расходами для SaaS-платформы с 10 000 ежедневных запросов к GPT-4 и Claude. Результат: снижение затрат на 35% за первый месяц. Как мы это сделали и как повторить успех — разберём ниже.

Почему важно бюджетировать запросы LLM?

LLM-запросы непредсказуемы по стоимости: длина выходного токена варьируется, а системные промпты часто раздуваются без контроля. Без бюджетирования легко получить счёт, в 5–10 раз превышающий ожидания. Бюджетирование решает три задачи:

• Предотвращение перерасхода: алерты при превышении дневного или месячного лимита.
• Оптимизация стоимости: выявление неэффективных промптов и моделей.
• Прозрачность для команд: дашборды с детализацией по пользователям, функциям и моделям.

Принципы работы token counting в продакшне

Token counting — это оценка количества токенов запроса до отправки. Для OpenAI используем библиотеку tiktoken (OpenAI Tokenizer documentation), для Anthropic — встроенный счётчик. Пример подсчёта:

import tiktoken def count_tokens_openai(text: str, model: str = "gpt-4") -> int: enc = tiktoken.encoding_for_model(model) return len(enc.encode(text)) def estimate_request_cost(prompt: str, max_completion: int = 1000, model: str = "gpt-4-turbo") -> dict: input_tokens = count_tokens_openai(prompt, model) total_tokens = input_tokens + max_completion prices = { "gpt-4-turbo": {"input": 10.0, "output": 30.0}, "gpt-4o": {"input": 5.0, "output": 15.0}, "gpt-4o-mini": {"input": 0.15, "output": 0.60}, "claude-3-5-sonnet": {"input": 3.0, "output": 15.0}, } price = prices.get(model, {"input": 10.0, "output": 30.0}) estimated_cost = (input_tokens / 1_000_000 * price["input"] + max_completion / 1_000_000 * price["output"]) return {"input_tokens": input_tokens, "max_output_tokens": max_completion, "estimated_cost_usd": estimated_cost} 

Оценка стоимости позволяет отклонять запросы, превышающие лимит пользователя или команды.

Методы снижения расходов на LLM

Один из эффективных методов — автоматический роутинг моделей. Например, для простых задач (извлечение данных, базовая классификация) используем GPT-4o-mini, который в 10 раз дешевле GPT-4-turbo, но даёт сравнимые результаты. Для сложных генераций оставляем топовые модели. В нашем кейсе это снизило среднюю стоимость запроса с $0.03 до $0.008.

Также внедряем кэширование: дублирующиеся запросы (например, повторные вопросы от пользователей) обслуживаем из Redis-кэша, что экономит до 40% бюджета. Используем LRU-кэш с TTL 24 часа.

Система бюджетирования с Redis и Middleware

Для хранения лимитов используем Redis — он быстр и поддерживает атомарные операции. Структура бюджета:

from dataclasses import dataclass, field import threading @dataclass class TokenBudget: daily_limit_usd: float monthly_limit_usd: float per_user_daily_limit_usd: float = 1.0 spent_today: float = field(default=0.0) spent_month: float = field(default=0.0) _lock: threading.Lock = field(default_factory=threading.Lock) class LLMBudgetManager: def __init__(self, redis_client, budget: TokenBudget): self.redis = redis_client self.budget = budget def check_and_reserve(self, user_id: str, estimated_cost: float) -> bool: """Проверка бюджета перед запросом""" daily_spent = float(self.redis.get(f"budget:daily") or 0) if daily_spent + estimated_cost > self.budget.daily_limit_usd: raise BudgetExceededError(f"Daily budget ${self.budget.daily_limit_usd} exceeded") user_spent = float(self.redis.get(f"budget:user:{user_id}:daily") or 0) if user_spent + estimated_cost > self.budget.per_user_daily_limit_usd: raise BudgetExceededError(f"User daily budget ${self.budget.per_user_daily_limit_usd} exceeded") pipe = self.redis.pipeline() pipe.incrbyfloat(f"budget:daily", estimated_cost) pipe.expire(f"budget:daily", 86400) pipe.incrbyfloat(f"budget:user:{user_id}:daily", estimated_cost) pipe.expire(f"budget:user:{user_id}:daily", 86400) pipe.execute() return True def record_actual_cost(self, user_id: str, actual_cost: float, estimated_cost: float): correction = actual_cost - estimated_cost if abs(correction) > 0.001: self.redis.incrbyfloat("budget:daily", correction) 

import functools def track_llm_cost(model: str = "gpt-4o"): def decorator(func): @functools.wraps(func) async def wrapper(*args, **kwargs): result = await func(*args, **kwargs) if hasattr(result, 'usage'): cost = compute_cost(result.usage.prompt_tokens, result.usage.completion_tokens, model) analytics.record(function=func.__name__, model=model, cost=cost, input_tokens=result.usage.prompt_tokens, output_tokens=result.usage.completion_tokens) return result return wrapper return decorator