Перейти к содержанию

Надежность и ресурсные ограничения

Шаг 1

На данном этапе мы превратили учебный проект в AI-бэкенд на FastAPI. И по сути теперь несколько клиентов могут обращаться к нему по HTTP: отправлять сообщения в чат, запускать анализ обращений, передавать документы и получать потоковые ответы через SSE. А основные сценарии у нас уже проверяются автоматическими тестами.

Однако успешный ответ на один запрос ещё не означает, что приложение готово к постоянной нагрузке. Пока мы запускали сценарии вручную и по очереди, и почти невозможно было заметить две новые проблемы. Первая появляется, когда к модели одновременно обращаются несколько клиентов. А вторая проблема, это когда приложение расходует токены, но мы не понимаем, во сколько обходится отдельная AI-операция.

Представим, что четыре пользователя одновременно отправили сообщения в чат, ещё один загрузил большой документ, а другой запустил структурированный анализ обращения. FastAPI примет все запросы и передаст их соответствующим сервисам. Чат обратится к BoundedToolLoop, документ — к DocumentAIService, а обращение — к SupportRequestAnalyzer. Но в конце все эти пути сходятся в одном месте, ведь каждый сервис использует общий LLMClient для обращения к ProxyAPI.

Это хорошая архитектура, ведь работа с провайдером не продублирована в каждом сервисе, и именно поэтому LLMClient становится общей точкой, в которой удобно контролировать нагрузку. Если поставить ограничение только в маршруте чата, анализ документов сможет обойти его. Если добавить отдельный счётчик в каждый сервис, приложение всё равно не узнает общее количество активных обращений. То есть ограничение должно находиться перед единственным общим выходом к модели.

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

  • FILE_MAX_SIZE_BYTES не разрешает принять слишком большой файл.
  • CHAT_INPUT_TOKEN_BUDGET ограничивает контекст чата.
  • MAX_OUTPUT_TOKENS задаёт максимальную длину ответа модели.
  • TOOL_MAX_STEPS останавливает цикл инструментов после разрешённого количества шагов.

Отдельно проект ограничивает ожидание провайдера. Значение REQUEST_TIMEOUT_SECONDS передаётся в SDK и относится к одному сетевому обращению, а для потоковой генерации существует STREAM_TIMEOUT_SECONDS и после получения очередного фрагмента код проверяет, сколько времени прошло с начала потока.

Но очень важно, что один пользовательский запрос не всегда приводит к одному обращению к модели. Обычный анализ может закончиться одним вызовом, а исправляющий запрос добавит второй. Чат с инструментом сначала попросит модель выбрать функцию, выполнит её и затем снова обратится к модели за итоговым текстом. Большой документ может потребовать нескольких анализов фрагментов и отдельного объединения результатов.

Поэтому время всей пользовательской операции складывается из нескольких этапов. Уже существующие настройки ограничивают каждый сетевой вызов, количество шагов, размер контекста и объём документа. Но все перечисленные настройки относятся к содержимому или продолжительности отдельных операций. Но они не отвечают на вопрос, а сколько обращений к ProxyAPI может выполняться одновременно. Если двадцать клиентов одновременно отправят запросы, приложение может начать до двадцати генераций. Это увеличивает число открытых соединений, создаёт резкий скачок расходов и повышает вероятность получить ограничение от провайдера.

Именно эту проблему мы решим первой. В следующем шаге создадим ограничитель конкурентности и подключим его непосредственно к LLMClient. Там же сравним его с рейтлимитом, который ограничивает частоту запросов конкретного клиента за определённый промежуток времени.

Шаг 2

В предыдущем шаге мы увидели, что отдельные операции проекта уже ограничены по времени, размеру и количеству шагов. И установили, что у нас есть проблема, ведь FastAPI способен принять несколько запросов одновременно, и каждый из них может обратиться к общему LLMClient. А сейчас количество таких обращений ничем не ограничено внутри приложения.

Исправим именно эту проблему. Перед каждым вызовом ProxyAPI клиент будет занимать одно свободное место. Если разрешённое количество обращений уже выполняется, новый запрос немного подождёт. Если за отведённое время место не освободится, приложение остановит этот вызов с понятной ошибкой.

Для такой задачи подходит BoundedSemaphore из стандартной библиотеки Python. Семафор можно представить как счётчик свободных мест. Если его значение равно трём, одновременно пройти защищённый участок смогут три вызова. Четвёртый будет ждать, пока один из предыдущих завершится и вернёт место.

И возможно у вас сразу возникает вопрос, а почему бы просто не ограничить частоту запросов? Но это разные механизмы. Рейтлимит задаёт, сколько запросов определённый клиент может отправить за промежуток времени, например не более шестидесяти запросов в минуту. Семафор отвечает на другой вопрос: сколько обращений к модели выполняется прямо сейчас.

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

рейтлимит → ограничивает частоту запросов одного клиента
семафор → ограничивает количество одновременно выполняющихся вызовов

В реальном публичном API обычно нужны оба механизма. Но для корректного рейтлимита сначала необходимо понимать, кого именно мы ограничиваем, это пользователя, API-ключ, организацию или IP-адрес. В нашем учебном проекте пока нет аутентификации, а общий лимит для нескольких процессов потребовал бы внешнего хранилища. Поэтому сейчас реализуем только защиту общей точки обращения к модели. Аутентификацию и распределённый рейтлимит в этом учебном проекте реализовывать не будем. Мы вернёмся к ним во время финального ревью и зафиксируем этот момент как ограничения текущей версии.

Давайте перейдем к практике. Сначала добавим отдельную прикладную ошибку, для этого откройте app/llm/errors.py и после LLMRateLimitError добавьте:

class LLMCapacityError(LLMClientError):
    pass

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

Теперь откройте app/config.py и после stream_timeout_seconds добавьте две настройки:

llm_max_concurrent_requests: int = Field(
    default=4,
    gt=0,
    validation_alias="LLM_MAX_CONCURRENT_REQUESTS",
)
llm_queue_timeout_seconds: float = Field(
    default=5.0,
    gt=0.0,
    validation_alias="LLM_QUEUE_TIMEOUT_SECONDS",
)

Настройка llm_max_concurrent_requests задаёт количество вызовов модели, которые могут выполняться одновременно внутри одного процесса приложения. В реальном проекте его выбирают с учётом ограничений провайдера, времени ответа и доступного бюджета.

А настройка llm_queue_timeout_seconds определяет, сколько новый вызов готов ждать свободное место. Это не время генерации и не сетевой таймаут. После получения места продолжительность самого запроса по-прежнему ограничивается request_timeout_seconds.

При желании эти значения можно переопределить в .env:

LLM_MAX_CONCURRENT_REQUESTS=4
LLM_QUEUE_TIMEOUT_SECONDS=5

Создайте файл app/llm/limits.py. В нём будет находиться небольшой объект, который отвечает только за выдачу и освобождение мест:

from collections.abc import Iterator
from contextlib import contextmanager
from threading import BoundedSemaphore

from app.llm.errors import LLMCapacityError

class LLMConcurrencyLimiter:
    def __init__(
        self,
        max_concurrent_requests: int,
        queue_timeout_seconds: float,
    ) -> None:
        self._semaphore = BoundedSemaphore(
            max_concurrent_requests
        )
        self._queue_timeout_seconds = queue_timeout_seconds

    @contextmanager
    def slot(self) -> Iterator[None]:
        acquired = self._semaphore.acquire(
            timeout=self._queue_timeout_seconds
        )
        if not acquired:
            raise LLMCapacityError(
                "Все доступные места для обращения к модели заняты."
            )

        try:
            yield
        finally:
            self._semaphore.release()

Метод slot() является контекстным менеджером. При входе в блок with он пытается уменьшить счётчик свободных мест. Если место получено, выполнение доходит до yield и вызывающий код может обратиться к модели. При выходе из блока выполняется finally и место обязательно возвращается.

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

Теперь подключим ограничитель в одной общей точке. Откройте app/llm/client.py и добавьте импорт:

from app.llm.limits import LLMConcurrencyLimiter

В конструкторе LLMClient, перед созданием OpenAI, создайте ограничитель:

self._limiter = LLMConcurrencyLimiter(
    max_concurrent_requests=(
        settings.llm_max_concurrent_requests
    ),
    queue_timeout_seconds=settings.llm_queue_timeout_seconds,
)

После этого оберните публичный метод generate() в полученный контекстный менеджер:

def generate(self, messages: list[Message]) -> LLMResult:
    with self._limiter.slot():
        try:
            return self._generate_once(messages)
        except openai.APIError as error:
            raise self._convert_error(error) from error

Таким же способом добавьте with self._limiter.slot(): в generate_structured(), generate_with_tools() и generate_vision_structured(). Внутренний код этих методов не меняется, он только выполняется после получения свободного места.

Потоковый метод нужно изменить отдельно. Его место нельзя освобождать сразу после создания объекта потока, ведь у нас соединение продолжает работать, пока генератор передаёт фрагменты. Поэтому блок with должен охватывать весь yield from:

def stream(
    self,
    messages: list[Message],
    tools: list[ToolDefinition] | None = None,
    tool_choice: Literal["auto", "none"] = "auto",
) -> Generator[LLMStreamEvent, None, None]:
    with self._limiter.slot():
        try:
            yield from self._stream_once(
                messages=messages,
                tools=tools,
                tool_choice=tool_choice,
            )
        except openai.APIError as error:
            raise self._convert_error(error) from error

Теперь место остаётся занятым с начала потокового обращения и до успешного завершения, ошибки или закрытия генератора. Это важно: если освободить его после получения самого объекта stream, приложение снова сможет открыть неограниченное количество активных SSE-соединений к модели.

Один экземпляр ограничителя находится внутри единственного LLMClient, который создаёт build_application(). Этим клиентом пользуются чат, анализатор обращений, анализ документов и анализ изображений. Поэтому ограничение автоматически действует для всех AI-функций, а нам не пришлось добавлять отдельные счётчики в каждый сервис или эндпоинт.

Так мы ограничили количество одновременно выполняющихся обращений, но ещё не знаем, во сколько обходятся успешно завершённые запросы. В следующем шаге мы создадим независимый калькулятор, который переведёт входные и выходные токены в примерную стоимость по тарифам выбранной модели.

Шаг 3

До сих пор мы смотрели на количество токенов в основном как на техническую характеристику запроса. Во время ручной проверки одного сообщения этого было достаточно. Но после появления HTTP API ситуация изменилась, и приложением могут одновременно пользоваться несколько клиентов, а одна пользовательская операция иногда вызывает модель несколько раз.

Например чат сначала может попросить модель выбрать инструмент, а после выполнения функции обратиться к ней ещё раз за итоговым ответом. Анализ большого документа обрабатывает несколько фрагментов, а затем объединяет результаты. Даже если каждый отдельный вызов кажется недорогим, их общая стоимость постепенно складывается. Без отдельного расчёта разработчик замечает проблему только тогда, когда баланс провайдера уже уменьшился значительно сильнее ожидаемого.

При этом количество токенов само по себе не отвечает на вопрос о расходах. Две модели могут обработать одинаковый объём текста, но иметь разные тарифы. Кроме того, входные и выходные токены обычно оцениваются по-разному. Поэтому значение total_tokens=1000 ещё не говорит, это дорого это или дёшево.

То есть простыми словами мы можем сказать, что оценка стоимости нужна не только для показа красивого числа в HTTP-ответе. А она помогает сравнивать модели и версии промптов, находить неожиданно дорогие сценарии, понимать стоимость одной AI-функции и заранее планировать бюджет. Например если после добавления длинной инструкции обычный ответ чата стал обходиться вдвое дороже, разработчик сможет увидеть это изменение и решить, оправдано ли оно.

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

Исходные данные для такого расчёта у нас уже есть. Сервисы сохраняют статистику использования модели, поэтому в ответах чата, анализа обращения и документов можно увидеть prompt_tokens, completion_tokens и total_tokens. Заново измерять длину текста не придётся: калькулятор получит уже готовое количество входных и выходных токенов.

Провайдеры обычно указывают отдельную цену для входных и выходных токенов. Входными считаются инструкция, история и данные пользователя, которые мы передали модели. А выходные токены это созданный моделью ответ. Их нельзя просто сложить и умножить на одну цену, потому что тарифы для входа и выхода обычно отличаются.

стоимость входа = входные токены × цена входа / 1 000 000
стоимость выхода = выходные токены × цена выхода / 1 000 000

общая стоимость = стоимость входа + стоимость выхода

И считать стоимость мы будем за миллион токенов. Именно такой формат используется в тарифах моделей и позволяет не хранить очень маленькую цену одного токена. Если выбранный сервис показывает цену за тысячу токенов, её нужно сначала умножить на тысячу.

Почему стоимость будет примерной?

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

Поэтому цену нельзя навсегда записывать непосредственно в Python-коде. Мы вынесем её в настройки. Ведь можно посмотреть актуальный тариф выбранной модели в личном кабинете или документации провайдера и изменить значения в .env, не переписывая приложение.

А для вычислений мы будем использовать Decimal, а не float. Двоичные числа с плавающей точкой удобны для обычных измерений, но некоторые десятичные дроби представляются в них неточно, а Decimal хранит десятичное значение предсказуемо и лучше подходит для денежных расчётов.

Откройте app/config.py. В начало файла добавьте импорт:

from decimal import Decimal

В существующий импорт Pydantic добавьте model_validator:

from pydantic import Field, SecretStr, model_validator

После настроек ограничения конкурентности добавьте:

llm_input_price_per_million: Decimal | None = Field(
    default=None,
    ge=0,
    validation_alias="LLM_INPUT_PRICE_PER_MILLION",
)
llm_output_price_per_million: Decimal | None = Field(
    default=None,
    ge=0,
    validation_alias="LLM_OUTPUT_PRICE_PER_MILLION",
)
llm_cost_currency: str = Field(
    default="RUB",
    min_length=3,
    max_length=3,
    validation_alias="LLM_COST_CURRENCY",
)

Обе цены допускают None. Благодаря этому проект продолжит запускаться, даже если мы не хотим рассчитывать стоимость. В таком случае API просто вернёт null вместо выдуманного числа. А поле валюты не участвует в формуле расчета и только объясняет, в каких единицах записан результат.

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

@model_validator(mode="after")
def validate_llm_pricing(self) -> "Settings":
    input_price_is_set = (
        self.llm_input_price_per_million is not None
    )
    output_price_is_set = (
        self.llm_output_price_per_million is not None
    )

    if input_price_is_set != output_price_is_set:
        raise ValueError(
            "Нужно указать обе цены модели или не указывать ни одной."
        )

    return self

В .env цены указываются без символа валюты. И перед тем, как их указывать, давайте посмотрим на цены модели, которую мы используем с самого начала:

И теперь добавьте в .env переменные и значения стоимости:

LLM_INPUT_PRICE_PER_MILLION=516
LLM_OUTPUT_PRICE_PER_MILLION=2062
LLM_COST_CURRENCY=RUB

Но не забывайте, если модель или провайдер изменятся, настройки стоимости также нужно обновить.

Теперь создайте файл app/llm/cost.py. В нём будет находиться расчёт, который не зависит от FastAPI, чата или конкретного AI-сервиса:

from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP

TOKENS_PER_MILLION = Decimal("1000000")
COST_PRECISION = Decimal("0.000001")

@dataclass(frozen=True)
class LLMCostEstimate:
    amount: Decimal
    currency: str

class LLMCostEstimator:
    def __init__(
        self,
        input_price_per_million: Decimal | None,
        output_price_per_million: Decimal | None,
        currency: str,
    ) -> None:
        self._input_price = input_price_per_million
        self._output_price = output_price_per_million
        self._currency = currency.upper()

    def estimate(
        self,
        prompt_tokens: int | None,
        completion_tokens: int | None,
    ) -> LLMCostEstimate | None:
        if (
            prompt_tokens is None
            or completion_tokens is None
            or self._input_price is None
            or self._output_price is None
        ):
            return None

        if prompt_tokens < 0 or completion_tokens < 0:
            raise ValueError("Количество токенов не может быть отрицательным.")

        input_cost = (
            Decimal(prompt_tokens) * self._input_price
            / TOKENS_PER_MILLION
        )
        output_cost = (
            Decimal(completion_tokens) * self._output_price
            / TOKENS_PER_MILLION
        )
        amount = (input_cost + output_cost).quantize(
            COST_PRECISION,
            rounding=ROUND_HALF_UP,
        )
        return LLMCostEstimate(
            amount=amount,
            currency=self._currency,
        )

Класс LLMCostEstimate хранит сумму и её валюту как одно согласованное значение. Метод estimate() сначала проверяет, можно ли выполнить расчёт. Если SDK не вернул статистику либо тарифы не настроены, метод возвращает None. Это честнее, чем подставлять ноль, ведь нулевая стоимость и неизвестная стоимость означают разные ситуации.

Затем входные и выходные токены рассчитываются отдельно. В конце сумма округляется до шести знаков после запятой. Нам не нужна точность до бесконечного количества символов, но для небольших LLM-запросов двух знаков после запятой было бы недостаточно: полезное значение часто меньше одной денежной единицы.

Чтобы один объект калькулятора использовался всеми HTTP-эндпоинтами, добавим его в общую сборку приложения. Откройте app/application.py и добавьте импорт:

from app.llm.cost import LLMCostEstimator

В ApplicationContainer после llm_client добавьте поле:

cost_estimator: LLMCostEstimator

В функции build_application(), сразу после создания llm_client, создайте калькулятор:

cost_estimator = LLMCostEstimator(
    input_price_per_million=(
        settings.llm_input_price_per_million
    ),
    output_price_per_million=(
        settings.llm_output_price_per_million
    ),
    currency=settings.llm_cost_currency,
)

И передайте его в возвращаемый ApplicationContainer:

cost_estimator=cost_estimator,

На этом калькулятор создан и связан с приложением, но публичные ответы пока содержат только токены. В следующем шаге мы добавим оценку стоимости в общий TokenUsageResponse. И благодаря общей модели изменение сразу станет доступно чату, SSE, анализу обращений и документов.

Шаг 4

В двух предыдущих шагах мы создали два независимых механизма. Ограничитель конкурентности уже подключён к LLMClient и автоматически контролирует все обращения к ProxyAPI. И калькулятор стоимости тоже создан и добавлен в ApplicationContainer, поэтому любой эндпоинт может получить к нему доступ.

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

{
  "prompt_tokens": 420,
  "completion_tokens": 80,
  "total_tokens": 500
}

Теперь нам нужно связать уже готовый калькулятор с формированием HTTP-ответов.

Эту проблему решим через уже существующую общую модель TokenUsageResponse. В пятом модуле мы специально вынесли статистику токенов в отдельный HTTP-контракт, которым пользуются разные AI-эндпоинты. Теперь расширим его полями стоимости и добавим одну общую функцию сборки. Она получит токены, обратится к LLMCostEstimator и вернёт готовый блок usage.

После подключения запустим автоматические тесты и настоящий HTTP-запрос. Так мы проверим обе части раздела, а именно что ограничитель не разрешает открыть лишнее одновременное обращение, а калькулятор правильно добавляет приблизительную стоимость в публичный ответ.

Перейдём к коду. Откройте app/api/schemas/common.py и добавьте импорт калькулятора:

from app.llm.cost import LLMCostEstimator

Дополните TokenUsageResponse двумя полями:

class TokenUsageResponse(BaseModel):
    prompt_tokens: int | None = Field(default=None, ge=0)
    completion_tokens: int | None = Field(default=None, ge=0)
    total_tokens: int | None = Field(default=None, ge=0)
    estimated_cost: float | None = Field(default=None, ge=0)
    cost_currency: str | None = None

Поле estimated_cost содержит приблизительную сумму, а cost_currency объясняет её единицу. Оба поля допускают None, потому что SDK иногда не возвращает статистику, а настройки тарифа могут быть не заполнены.

Одной модели недостаточно, ведь в разных эндпоинтах нам нужно одинаково рассчитать стоимость и создать TokenUsageResponse. Поэтому в этом же файле, сразу после класса TokenUsageResponse, добавьте общую функцию build_token_usage_response(). Она будет отвечать за создание этого блока HTTP-ответа и избавит нас от повторения расчёта в нескольких маршрутах.

def build_token_usage_response(
    prompt_tokens: int | None,
    completion_tokens: int | None,
    total_tokens: int | None,
    cost_estimator: LLMCostEstimator,
) -> TokenUsageResponse:
    estimate = cost_estimator.estimate(
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
    )
    return TokenUsageResponse(
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        total_tokens=total_tokens,
        estimated_cost=(
            float(estimate.amount)
            if estimate is not None
            else None
        ),
        cost_currency=(
            estimate.currency
            if estimate is not None
            else None
        ),
    )

Эта функция отвечает только за статистику использования модели. Она вызывает уже готовый калькулятор и собирает одинаковый TokenUsageResponse для всех AI-эндпоинтов.

Теперь откройте app/api/routes/support_requests.py. В импорте из app.api.schemas.common укажите новую функцию:

from app.api.schemas.common import build_token_usage_response

В возвращаемом SupportRequestAnalyzeResponse найдите создание TokenUsageResponse(...)

usage=build_token_usage_response(
    prompt_tokens=result.prompt_tokens,
    completion_tokens=result.completion_tokens,
    total_tokens=result.total_tokens,
    cost_estimator=container.cost_estimator,
),

В этом маршруте прямой импорт TokenUsageResponse больше не понадобится, если он не используется в другом месте. Теперь расчёт получает уже собранную статистику SupportRequestAnalyzer. В обычном сценарии анализатор обращается к модели один раз, поэтому эта статистика описывает весь анализ. Если выполнялся исправляющий запрос, текущий SupportRequestAnalyzer возвращает токены только последнего вызова. Это одно из ограничений такой оценки, но уже в разделе о логировании мы будем учитывать каждый вызов LLMClient отдельно.

Такую же замену выполните в app/api/routes/documents.py. Добавьте импорт build_token_usage_response и замените создание TokenUsageResponse тем же вызовом с полями результата:

usage=build_token_usage_response(
    prompt_tokens=result.prompt_tokens,
    completion_tokens=result.completion_tokens,
    total_tokens=result.total_tokens,
    cost_estimator=container.cost_estimator,
),

Для большого документа DocumentAIService уже возвращает сумму токенов нескольких обращений к модели. Поэтому калькулятор оценивает стоимость всей операции, а не только последнего фрагмента.

Наконец, откройте app/api/routes/chat.py и измените импорт:

from app.api.schemas.common import build_token_usage_response

В этом файле создание TokenUsageResponse встречается дважды: в обычном reply_to_chat() и внутри события completed потокового ответа. Замените оба блока:

usage=build_token_usage_response(
    prompt_tokens=result.prompt_tokens,
    completion_tokens=result.completion_tokens,
    total_tokens=result.total_tokens,
    cost_estimator=container.cost_estimator,
),

В обоих случаях используются одни и те же итоговые значения ChatReply. Если модель вызывала инструмент, BoundedToolLoop уже сложил статистику всех шагов. Поэтому стоимость относится ко всему ответу чата.

Перед созданием новых тестов нужно обновить уже существующую тестовую сборку. В пятом модуле фикстура fake_container возвращала пустой SimpleNamespace, а каждый тест добавлял в него только необходимый сервис. Раньше этого было достаточно. Теперь все AI-эндпоинты дополнительно обращаются к container.cost_estimator, поэтому без этого объекта прежний успешный HTTP-тест завершится ошибкой AttributeError.

Откройте tests/conftest.py и добавьте импорты:

from decimal import Decimal

from app.llm.cost import LLMCostEstimator

Затем замените фикстуру fake_container():

@pytest.fixture
def fake_container() -> SimpleNamespace:
    return SimpleNamespace(
        cost_estimator=LLMCostEstimator(
            input_price_per_million=Decimal("2"),
            output_price_per_million=Decimal("8"),
            currency="RUB",
        )
    )

Мы добавляем калькулятор в общую фикстуру, потому что он нужен сразу нескольким эндпоинтам. Теперь любой HTTP-тест, использующий fake_container, получит готовый cost_estimator. Учебные тарифы выбраны специально для предсказуемого расчёта и не связаны с актуальной ценой настоящей модели.

После этого откройте tests/test_support_requests_api.py. В функции test_analyze_support_request_returns_contract() после проверки total_tokens добавьте:

assert body["usage"]["estimated_cost"] == 0.00006
assert body["usage"]["cost_currency"] == "RUB"

Fake-анализатор возвращает 10 входных и 5 выходных токенов. При заданных тестовых тарифах стоимость входа равна 0.00002, стоимость выхода — 0.00004, а общая оценка — 0.00006 RUB. Так прежний тест не только продолжит работать с обновлённым контейнером, но и проверит два новых поля HTTP-ответа.

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

Теперь добавим две отдельные автоматические проверки новых компонентов. Создайте tests/test_llm_resources.py:

from decimal import Decimal

import pytest

from app.llm.cost import LLMCostEstimator
from app.llm.errors import LLMCapacityError
from app.llm.limits import LLMConcurrencyLimiter

def test_cost_estimator_uses_separate_prices() -> None:
    estimator = LLMCostEstimator(
        input_price_per_million=Decimal("2"),
        output_price_per_million=Decimal("8"),
        currency="RUB",
    )

    result = estimator.estimate(
        prompt_tokens=1_000_000,
        completion_tokens=500_000,
    )

    assert result is not None
    assert result.amount == Decimal("6.000000")
    assert result.currency == "RUB"

def test_limiter_rejects_call_when_slot_is_busy() -> None:
    limiter = LLMConcurrencyLimiter(
        max_concurrent_requests=1,
        queue_timeout_seconds=0.01,
    )

    with limiter.slot():
        with pytest.raises(LLMCapacityError):
            with limiter.slot():
                pass

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

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

Запустите этот тест:

pytest tests/test_llm_resources.py

Он должен пройти успешно, а значит теперь можно запустить FastAPI:

fastapi dev app/api/main.py

Откройте Swagger UI и выполните любой AI-запрос, например анализ обращения или обычное сообщение чата. Если обе цены заполнены в .env и провайдер вернул статистику, внутри usage появятся дополнительные поля:

  "usage": {
    "prompt_tokens": 1077,
    "completion_tokens": 52,
    "total_tokens": 1129,
    "estimated_cost": 0.662956,
    "cost_currency": "RUB"
  },

Числа в этом примере показывают только форму ответа. Фактическое значение зависит от тарифа, модели и количества токенов. Если цены не настроены, последние два поля будут равны null, но остальные функции приложения продолжат работать.

Проверять ограничение конкурентности через множество настоящих запросов мы не будем, это потратило бы баланс и зависело бы от скорости сети. Детерминированный тест уже подтвердил основное правило семафора. А позднее в логах мы сможем видеть случаи, когда приложение отклонило вызов из-за занятых мест.

Теперь давайте подведём итоги раздела. Мы не переписывали уже существующие таймауты, ограничения контекста, шагов и размера файлов. Вместо этого закрыли две оставшиеся задачи, а именно ограничили количество одновременных вызовов общего LLMClient и связали статистику токенов с настраиваемой оценкой стоимости.

Теперь приложение лучше контролирует собственную нагрузку, а клиент API видит не только техническое количество токенов, но и понятную приблизительную сумму. В следующем разделе добавим структурированные логи. Они позволят связать HTTP-запрос с обращениями к модели и увидеть продолжительность, количество токенов, приблизительную стоимость и результат каждого вызова.

Шаг 5

В каком месте приложения следует ограничивать общее количество одновременных обращений к модели?

Правильный ответ
  • ✅ В общем LLM-клиенте, через который проходят все обращения к провайдеру

Шаг 6

Что должен гарантировать контекстный менеджер slot()?

with limiter.slot():
    return provider_call()
Правильный ответ
  • ✅ Освободить занятое место и после результата, и после исключения

Шаг 7

В каких случаях LLMCostEstimator.estimate() должен вернуть None?

Выберите все правильные варианты.

Правильный ответ
  • ✅ В настройках отсутствует цена выходных токенов
  • ✅ Провайдер не сообщил количество входных токенов
  • ✅ В настройках отсутствует цена входных токенов
  • ✅ Провайдер не сообщил количество выходных токенов

Шаг 8

Цена миллиона входных токенов равна 2 RUB, а миллиона выходных — 8 RUB. Сколько будет стоить вызов с 1 000 000 входных и 500 000 выходных токенов?

Правильный ответ
  • ✅ 6 RUB

Шаг 9

Оценка стоимости LLM-вызова

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

Задача

Реализуйте конструктор и метод estimate() класса LLMCostEstimator. Результат должен возвращаться в неизменяемом объекте LLMCostEstimate.

Требования

  • В конструкторе сохраните входную цену, выходную цену и код валюты. Код валюты сохраните в верхнем регистре.
  • Если любое количество токенов или любая цена равны None, метод estimate() должен вернуть None.
  • Если количество входных или выходных токенов отрицательное, выбросьте ValueError с текстом "Количество токенов не может быть отрицательным.".
  • Рассчитайте входную и выходную части отдельно: количество токенов умножается на соответствующую цену и делится на TOKENS_PER_MILLION.
  • Сложите обе части и округлите результат до COST_PRECISION с режимом ROUND_HALF_UP.
  • Верните LLMCostEstimate(amount=..., currency=...).
  • Во всех вычислениях используйте Decimal, а не float.

Пример результата

estimator = LLMCostEstimator(
    Decimal("2"),
    Decimal("8"),
    "rub",
)
result = estimator.estimate(1_000_000, 500_000)

result.amount    # Decimal("6.000000")
result.currency  # "RUB"

Импорты, константы и класс LLMCostEstimate уже находятся в заготовке.

Правильный ответ 
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP

TOKENS_PER_MILLION = Decimal("1000000")
COST_PRECISION = Decimal("0.000001")

@dataclass(frozen=True)
class LLMCostEstimate:
    amount: Decimal
    currency: str

class LLMCostEstimator:
    def __init__(
        self,
        input_price_per_million: Decimal | None,
        output_price_per_million: Decimal | None,
        currency: str,
    ) -> None:
        self.input_price_per_million = input_price_per_million
        self.output_price_per_million = output_price_per_million
        self.currency = currency.upper()

    def estimate(
        self,
        prompt_tokens: int | None,
        completion_tokens: int | None,
    ) -> LLMCostEstimate | None:
        if (
            prompt_tokens is None
            or completion_tokens is None
            or self.input_price_per_million is None
            or self.output_price_per_million is None
        ):
            return None

        if prompt_tokens < 0 or completion_tokens < 0:
            raise ValueError("Количество токенов не может быть отрицательным.")

        input_cost = (
            Decimal(prompt_tokens)
            * self.input_price_per_million
            / TOKENS_PER_MILLION
        )
        output_cost = (
            Decimal(completion_tokens)
            * self.output_price_per_million
            / TOKENS_PER_MILLION
        )

        total_cost = (input_cost + output_cost).quantize(
            COST_PRECISION,
            rounding=ROUND_HALF_UP,
        )

        return LLMCostEstimate(
            amount=total_cost,
            currency=self.currency,
        )