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

Структурированное логирование

Шаг 1

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

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

Одного текста «анализ не выполнен» для диагностики недостаточно. Разработчику нужно понять:

  • какой эндпоинт вызывался
  • какой HTTP-статус вернуло приложение
  • сколько времени заняла обработка
  • обращалось ли приложение к модели
  • какой способ генерации использовал LLMClient
  • сколько токенов сообщил провайдер
  • какая прикладная ошибка возникла

Сейчас часть этих сведений возвращается HTTP-клиенту, но после завершения запроса у нас не остаётся собственной понятной записи. Например, заголовок X-Request-ID позволяет назвать конкретный запрос. Однако по одному идентификатору ничего нельзя найти, если сервер не добавил его в журнал событий.

В этом разделе мы начнём добавлять в проект наблюдаемость(Observability). Так называют возможность понимать состояние работающего приложения по техническим данным, которые оно создаёт во время работы. Основными источниками таких данных обычно становятся логи, метрики и трассировки.

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

Что такое лог?

Лог — это запись об отдельном событии внутри приложения. Например, сервер закончил запрос POST /support-requests/analyze, вернул статус 200 и начал отправлять ответ через 1,42 секунды. Другая запись может сообщить, что обращение к модели завершилось ошибкой LLMTimeoutError.

Несколько записей с одним Request ID помогают восстановить последовательность действий:

Если один запрос чата вызвал модель дважды, в журнале появятся два LLM-события и одно итоговое HTTP-событие. У всех записей будет одинаковый Request ID, поэтому разработчик сможет увидеть всю цепочку.

Теперь поговорим о том, а почему логи будут структурированными? Но тут все просто, обычная строка удобна человеку:

Запрос успешно обработан за 1.42 секунды

Но программе сложно надёжно извлечь из неё маршрут, статус и время. При изменении формулировки придётся менять и правила разбора. Поэтому серверные приложения часто записывают события в формате JSON:

{
  "timestamp": "2026-09-10T07:15:12.231+00:00",
  "level": "info",
  "logger": "app.api.middleware",
  "event": "http_request_completed",
  "request_id": "f875ee4d...",
  "method": "POST",
  "route": "/support-requests/analyze",
  "status_code": 200,
  "response_start_seconds": 1.42
}

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

Поле event содержит стабильное техническое имя события, а не свободное описание. Например, успешно завершённый HTTP-запрос всегда будет называться http_request_completed, а успешное обращение к модели — llm_call_completed. Поиск не будет зависеть от изменения человеческих формулировок.

Но при этом, журнал не должен становиться копией пользовательских данных. Через наш AI-бэкенд проходят сообщения, история диалога, документы, изображения, номера заказов и аргументы инструментов. В developer-инструкциях находятся внутренние правила продукта, а в конфигурации — API-ключ ProxyAPI.

Мы будем использовать список явно разрешённых технических полей. В логи попадут тип операции, шаблон маршрута, статус, длительность, модель, количество токенов, Request ID и Response ID провайдера. Но мы не станем записывать:

  • тело HTTP-запроса и его параметры
  • сообщения пользователя и историю чата
  • developer-инструкции и собранные промпты
  • содержимое документов, таблиц и изображений
  • API-ключи, заголовок Authorization и другие секреты
  • полные аргументы инструментов и тексты ответов модели

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

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

Сейчас мы начнём с HTTP-части и в следующем шаге настроим единый JSON-формат и дополним существующее middleware так, чтобы после каждого запроса в терминале оставалась безопасная структурированная запись.

Шаг 2

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

Логирование относится ко всему приложению, а не к одному конкретному сервису. События будут создавать разные части проекта: HTTP middleware запишет результат входящего запроса, LLMClient — результат обращения к модели, а позднее этот же механизм смогут использовать другие компоненты. Поэтому хранить такой код внутри api или llm было бы неправильно. Создадим для общей логики наблюдаемости отдельный пакет observability внутри app, а в нём — пустой файл __init__.py:

app/
└── observability/
    ├── __init__.py
    ├── context.py
    └── logging.py

Сначала создайте app/observability/context.py:

from contextvars import ContextVar

request_id_context: ContextVar[str | None] = ContextVar(
    "request_id",
    default=None,
)

ContextVar это класс из стандартного модуля Python contextvars. Он позволяет хранить отдельное значение для каждого текущего контекста выполнения. Благодаря этому одновременно обрабатываемые HTTP-запросы не смешивают свои Request ID, а сервисы и LLMClient могут получить идентификатор текущего запроса без передачи его через параметры каждого метода.

Можно было бы передавать request_id вручную через каждый сервис и метод, но тогда этот технический параметр появился бы почти во всём приложении. Контекстная переменная позволяет middleware один раз сохранить идентификатор, а форматтер журнала позднее прочитает его автоматически.

Теперь создайте app/observability/logging.py. Начнём с импортов и списка разрешённых дополнительных полей:

from datetime import UTC, datetime
import json
import logging
import sys

from app.observability.context import request_id_context

LOG_FIELDS = (
    "request_id",
    "method",
    "route",
    "status_code",
    "response_start_seconds",
    "duration_seconds",
    "operation",
    "model",
    "finish_reason",
    "prompt_tokens",
    "completion_tokens",
    "total_tokens",
    "estimated_cost",
    "cost_currency",
    "response_id",
    "error_type",
)

Этот кортеж является разрешающим списком. Форматтер возьмёт из записи только перечисленные значения. Даже если разработчик случайно передаст дополнительное поле message_text или api_key, оно не окажется в сформированном JSON.

Ниже добавьте собственный форматтер:

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, object] = {
            "timestamp": datetime.now(UTC).isoformat(),
            "level": record.levelname.lower(),
            "logger": record.name,
            "event": record.getMessage(),
        }

        request_id = getattr(record, "request_id", None)
        if request_id is None:
            request_id = request_id_context.get()

        if request_id is not None:
            payload["request_id"] = request_id

        for field_name in LOG_FIELDS:
            if field_name == "request_id":
                continue

            value = getattr(record, field_name, None)
            if value is not None:
                payload[field_name] = value

        return json.dumps(payload, ensure_ascii=False)

Когда мы вызываем logger.info() или logger.error(), стандартный модуль Python logging создаёт объект LogRecord. Это внутренняя запись, в которой хранятся сведения о событии: его уровень, название логгера, сообщение и дополнительные значения, переданные через extra.

Затем JsonFormatter получает этот объект и собирает из него словарь payload. В начало добавляются общие поля:

  • timestamp — время создания записи в UTC
  • level — уровень события, например info или error
  • logger — название Python-модуля, в котором было создано событие
  • event — техническое название события

UTC используется для того, чтобы записи из разных часовых поясов можно было правильно сопоставить по времени. Уровень info обозначает штатное событие, а error сообщает, что операция завершилась ошибкой.

Далее форматтер добавляет Request ID. Сначала он проверяет, был ли идентификатор передан непосредственно через extra. Если его там нет, значение берётся из request_id_context, куда Request ID ранее сохранило middleware.

После этого цикл переносит в результат только поля, перечисленные в LOG_FIELDS. Значения None пропускаются, поэтому в JSON не появляются пустые технические поля. Наконец json.dumps() превращает собранный словарь в одну JSON-строку, а параметр ensure_ascii=False сохраняет русский текст в читаемом виде, не заменяя буквы кодами наподобие \u041e.

После класса добавьте функцию настройки:

def configure_application_logging() -> None:
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())

    application_logger = logging.getLogger("app")
    application_logger.handlers.clear()
    application_logger.addHandler(handler)
    application_logger.setLevel(logging.INFO)
    application_logger.propagate = False

Мы настраиваем логгер с именем app, потому что весь код нашего приложения находится внутри одноимённого пакета. Например, модуль с HTTP middleware имеет полное имя app.api.middleware.

Для logging такие имена связаны между собой:

app
└── app.api
    └── app.api.middleware

Поэтому событие, созданное в app.api.middleware, автоматически попадёт в общий логгер app и будет преобразовано нашим JsonFormatter в JSON. Когда позднее в проекте появятся другие модули с логированием, для них не придётся отдельно настраивать формат.

При этом настройки затрагивают только модули внутри app. Сообщения Uvicorn, FastAPI и остальных библиотек продолжат выводиться в обычном виде. В результате в терминале мы увидим привычные служебные сообщения сервера и отдельные JSON-записи нашего приложения.

Теперь откройте app/api/middleware.py и дополните импорты:

import logging
from time import monotonic
from uuid import uuid4

from fastapi import Request, Response

from app.observability.context import request_id_context

logger = logging.getLogger(__name__)

monotonic() измеряет прошедшее время и не зависит от ручного изменения системных часов. Логгер получает имя текущего модуля app.api.middleware, поэтому его события попадут в настроенный обработчик.

Полностью замените функцию add_request_id():

async def add_request_id(
    request: Request,
    call_next,
) -> Response:
    request_id = uuid4().hex
    request.state.request_id = request_id
    context_token = request_id_context.set(request_id)
    started_at = monotonic()

    try:
        response = await call_next(request)
    except Exception as error:
        logger.error(
            "http_request_failed",
            extra={
                "method": request.method,
                "route": _route_path(request),
                "error_type": type(error).__name__,
                "response_start_seconds": (
                    monotonic() - started_at
                ),
            },
        )
        raise
    else:
        response.headers["X-Request-ID"] = request_id
        logger.info(
            "http_request_completed",
            extra={
                "method": request.method,
                "route": _route_path(request),
                "status_code": response.status_code,
                "response_start_seconds": (
                    monotonic() - started_at
                ),
            },
        )
        return response
    finally:
        request_id_context.reset(context_token)

В начале каждого запроса middleware создаёт идентификатор, как и раньше, но теперь дополнительно сохраняет его в контекстной переменной и запоминает время начала. Затем call_next(request) передаёт управление подходящему эндпоинту.

Если приложение сформировало HTTP-ответ, записывается событие http_request_completed с методом, шаблоном маршрута, статусом и длительностью. Ответ по-прежнему получает заголовок X-Request-ID. Если из приложения вышло неожиданное исключение, записывается http_request_failed, после чего ошибка передаётся FastAPI для обычной обработки.

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

В коде используется небольшая функция _route_path(). Добавьте её перед add_request_id():

def _route_path(request: Request) -> str:
    route = request.scope.get("route")
    return getattr(route, "path", "unmatched")

Она возвращает шаблон зарегистрированного маршрута. Например для разных чат-сессий в журнале останется одно значение:

/chat/sessions/{session_id}

Мы не записываем полный request.url, потому что он может содержать конкретный session_id и пользовательские данные из параметров запроса. Но вместо него сохраняем безопасный шаблон маршрута, например /chat/sessions/{session_id}. По нему понятно, какой эндпоинт был вызван, а одинаковые запросы удобно группировать.

У поля response_start_seconds есть важное название. Для обычного JSON-ответа оно почти совпадает с полной продолжительностью запроса. Но у SSE-потока middleware получает StreamingResponse до передачи всех событий клиенту. Поэтому поле показывает время до начала HTTP-ответа, а не длительность всей потоковой генерации. Полное время обращения к модели измерим непосредственно в LLMClient.

Осталось включить созданную настройку. Откройте app/api/main.py и добавьте импорт:

from app.observability.logging import configure_application_logging

Перед строкой app = create_api() вызовите функцию:

configure_application_logging()
app = create_api()

Настройка выполняется один раз при импорте точки входа FastAPI. Консольный main.py пока менять не нужно: созданные события относятся к прохождению HTTP-запросов.

Запустите приложение и откройте http://127.0.0.1:8000/health. Помимо служебных строк Uvicorn в терминале появится одна JSON-строка примерно такого вида:

{
  "timestamp": "2026-09-10T07:30:15.442+00:00",
  "level": "info",
  "logger": "app.api.middleware",
  "event": "http_request_completed",
  "request_id": "1064d534b98a47f9919a57acb1216373",
  "method": "GET",
  "route": "/health",
  "status_code": 200,
  "response_start_seconds": 0.00470487505663186
}

В примере запись разбита на строки только для удобства чтения. Сравните request_id в журнале с заголовком X-Request-ID ответа, это должно быть одно значение:

 content-length: 15
 content-type: application/json
 date: Fri,11 Sep 2026 07:10:29 GMT
 server: uvicorn
 x-request-id: 1064d534b98a47f9919a57acb1216373

Теперь мы можем найти итог HTTP-запроса, но пока не видим отдельные обращения к модели. Один запуск Document AI способен вызвать LLM несколько раз, а один ход чата — обратиться к инструменту и снова вызвать модель. Все обращения к модели в нашем проекте проходят через LLMClient, независимо от того, какой сервис их выполняет. Поэтому в следующем шаге мы добавим логирование именно в этот класс. Так мы сможем фиксировать каждый запрос к модели в одном месте и не повторять одинаковый код в разных сервисах.

Шаг 3

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

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

Создайте app/observability/llm.py и добавьте импорты:

import logging

from app.llm.cost import LLMCostEstimator

logger = logging.getLogger(__name__)

В этом файле создадим отдельный объект, который будет получать технический результат вызова и записывать его. Назовём его LLMObserver. Слово observer можно перевести как «наблюдатель»: он не формирует запрос и не меняет ответ модели, а только получает сведения о завершённой операции.

class LLMObserver:
    def __init__(
        self,
        cost_estimator: LLMCostEstimator,
    ) -> None:
        self._cost_estimator = cost_estimator

Наблюдателю передаётся уже созданный LLMCostEstimator. Благодаря этому в журнал попадёт оценка стоимости каждого отдельного вызова, а не только общая сумма, которую HTTP-ответ показывает клиенту.

Продолжим писать в этом же классе. После конструктора добавьте метод для успешно полученного ответа:

def record_success(
    self,
    operation: str,
    model: str,
    finish_reason: str | None,
    prompt_tokens: int | None,
    completion_tokens: int | None,
    total_tokens: int | None,
    response_id: str,
    duration_seconds: float,
) -> None:
    estimate = self._cost_estimator.estimate(
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
    )

    estimated_cost: float | None = None
    cost_currency: str | None = None

    if estimate is not None:
        estimated_cost = float(estimate.amount)
        cost_currency = estimate.currency

    logger.info(
        "llm_call_completed",
        extra={
            "operation": operation,
            "model": model,
            "finish_reason": finish_reason,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": total_tokens,
            "estimated_cost": estimated_cost,
            "cost_currency": cost_currency,
            "response_id": response_id,
            "duration_seconds": duration_seconds,
        },
    )

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

На основе количества токенов калькулятор определяет приблизительную стоимость вызова. Если тарифы не настроены или провайдер не вернул статистику, стоимость останется None и не попадёт в JSON-лог.

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

После record_success() добавьте обработку неудачной попытки:

def record_failure(
    self,
    operation: str,
    error_type: str,
    duration_seconds: float,
) -> None:
    logger.error(
        "llm_call_failed",
        extra={
            "operation": operation,
            "error_type": error_type,
            "duration_seconds": duration_seconds,
        },
    )

Метод record_failure() вызывается, когда обращение к модели завершается ошибкой. Он не исправляет ошибку и не заменяет её обработку, а только создаёт техническую запись llm_call_failed, по которой разработчик позднее сможет понять, какая операция не выполнилась и сколько времени она заняла.

При неудачном запросе провайдер может не вернуть статистику токенов и Response ID. Поэтому метод записывает только название операции, продолжительность и безопасный тип ошибки.

Наблюдатель готов, но пока ни один вызов его не использует. Откройте app/llm/client.py. Импорт Generator из collections.abc дополните типом Callable:

from collections.abc import Callable, Generator

Добавьте импорт нового класса:

from app.observability.llm import LLMObserver

Измените начало конструктора LLMClient и сохраните наблюдатель в поле:

    def __init__(self, settings: Settings, observer: LLMObserver | None = None) -> None:
        self._settings = settings
        self._observer = observer
        ...

Остальной код конструктора с ограничителем и объектом OpenAI оставьте без изменений. Наблюдатель необязателен, поэтому даже демонстрационные файлы и старые unit-тесты смогут по-прежнему создать LLMClient(settings). Основная сборка приложения передаст настоящий объект.

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

получить место в ограничителе
→ отправить запрос
→ обработать возможную ошибку
→ записать результат и время
→ вернуть LLMResult

Если добавить эту последовательность отдельно в каждый метод, одинаковый код придётся повторить несколько раз. Поэтому вынесем общую логику в метод _run_observed(). Он будет выполнять переданный ему запрос, измерять время и вызывать нужный метод LLMObserver.

Добавьте _run_observed() перед методом _generate_vision_structured_once():

def _run_observed(
    self,
    operation: str,
    call: Callable[[], LLMResult],
) -> LLMResult:
    started_at = monotonic()

    try:
        with self._limiter.slot():
            result = call()
    except openai.APIError as error:
        converted_error = self._convert_error(error)
        self._record_failure(
            operation,
            type(converted_error).__name__,
            started_at,
        )
        raise converted_error from error
    except Exception as error:
        self._record_failure(
            operation,
            type(error).__name__,
            started_at,
        )
        raise

    self._record_success(operation, result, started_at)
    return result

Параметр call содержит функцию без аргументов, которая выполнит конкретный запрос. _run_observed() не знает подробностей генерации. Он запускает таймер, получает место в ограничителе, вызывает переданную функцию и фиксирует итог.

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

Методы _record_success() и _record_failure() пока не существуют. Добавьте их сразу после _run_observed():

def _record_success(
    self,
    operation: str,
    result: LLMResult | LLMStreamCompleted,
    started_at: float,
) -> None:
    if self._observer is None:
        return

    self._observer.record_success(
        operation=operation,
        model=result.model,
        finish_reason=result.finish_reason,
        prompt_tokens=result.prompt_tokens,
        completion_tokens=result.completion_tokens,
        total_tokens=result.total_tokens,
        response_id=result.response_id,
        duration_seconds=monotonic() - started_at,
    )

def _record_failure(
    self,
    operation: str,
    error_type: str,
    started_at: float,
) -> None:
    if self._observer is None:
        return

    self._observer.record_failure(
        operation=operation,
        error_type=error_type,
        duration_seconds=monotonic() - started_at,
    )

Оба помощника сначала проверяют, передан ли наблюдатель. Тип LLMResult | LLMStreamCompleted позволяет использовать одну функцию для обычного и потокового результата: необходимые технические поля у них совпадают.

Теперь полностью замените generate():

def generate(self, messages: list[Message]) -> LLMResult:
    return self._run_observed(
        "generate",
        lambda: self._generate_once(messages),
    )

lambda не отправляет запрос в момент создания. Она сохраняет вызов вместе с аргументами, чтобы _run_observed() запустил его внутри таймера и ограничителя.

Таким же способом замените generate_structured():

def generate_structured(
    self,
    messages: list[Message],
    schema_name: str,
    schema: JsonSchema,
) -> LLMResult:
    return self._run_observed(
        "generate_structured",
        lambda: self._generate_structured_once(
            messages=messages,
            schema_name=schema_name,
            schema=schema,
        ),
    )

Замените generate_with_tools():

def generate_with_tools(
    self,
    messages: list[Message],
    tools: list[ToolDefinition],
    tool_choice: Literal["auto", "none"] = "auto",
) -> LLMResult:
    return self._run_observed(
        "generate_with_tools",
        lambda: self._generate_with_tools_once(
            messages=messages,
            tools=tools,
            tool_choice=tool_choice,
        ),
    )

И в нижней части класса замените generate_vision_structured():

def generate_vision_structured(
    self,
    instructions: str,
    content: list[VisionContentPart],
    schema_name: str,
    schema: JsonSchema,
) -> LLMResult:
    return self._run_observed(
        "generate_vision_structured",
        lambda: self._generate_vision_structured_once(
            instructions=instructions,
            content=content,
            schema_name=schema_name,
            schema=schema,
        ),
    )

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

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

def stream(
    self,
    messages: list[Message],
    tools: list[ToolDefinition] | None = None,
    tool_choice: Literal["auto", "none"] = "auto",
) -> Generator[LLMStreamEvent, None, None]:
    started_at = monotonic()
    completed: LLMStreamCompleted | None = None

    try:
        with self._limiter.slot():
            for event in self._stream_once(
                messages=messages,
                tools=tools,
                tool_choice=tool_choice,
            ):
                if isinstance(event, LLMStreamCompleted):
                    completed = event
                yield event
    except openai.APIError as error:
        converted_error = self._convert_error(error)
        self._record_failure(
            "stream",
            type(converted_error).__name__,
            started_at,
        )
        raise converted_error from error
    except GeneratorExit:
        if completed is None:
            self._record_failure(
                "stream",
                "stream_closed",
                started_at,
            )
        else:
            self._record_success(
                "stream",
                completed,
                started_at,
            )
        raise
    except Exception as error:
        self._record_failure(
            "stream",
            type(error).__name__,
            started_at,
        )
        raise
    else:
        if completed is None:
            self._record_failure(
                "stream",
                "incomplete_stream",
                started_at,
            )
        else:
            self._record_success(
                "stream",
                completed,
                started_at,
            )

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

Метод сохраняет LLMStreamCompleted в переменной completed, чтобы после окончания потока передать его данные в LLMObserver и создать полноценную запись об успешном вызове.

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

Осталось передать наблюдатель в клиент из общей сборки. Откройте app/application.py и добавьте импорт:

from app.observability.llm import LLMObserver

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

llm_observer: LLMObserver

В начале build_application() калькулятор должен создаваться до клиента, потому что он требуется наблюдателю. До этого момента начало build_application() выглядит следующим образом:

def build_application(settings: Settings) -> ApplicationContainer:
    llm_client = LLMClient(settings)

    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,
    )

Такой порядок работал, пока LLMClient не использовал наблюдатель. Теперь LLMObserver должен получить готовый калькулятор стоимости, а LLMClient — готовый наблюдатель. Поэтому нам нужно изменить порядок создания объектов.

Полностью замените показанный выше блок:

def build_application(settings: Settings) -> ApplicationContainer:
    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,
    )
    llm_observer = LLMObserver(cost_estimator)
    llm_client = LLMClient(
        settings,
        observer=llm_observer,
    )

Сначала создаётся LLMCostEstimator. Затем он передаётся в LLMObserver, а готовый наблюдатель — в LLMClient. Старую строку llm_client = LLMClient(settings) при этом нужно удалить, а не оставлять выше нового блока. В результате в приложении по-прежнему существует только один экземпляр LLMClient.

А в возвращаемый ApplicationContainer рядом с cost_estimator передайте:

llm_observer=llm_observer,

Теперь вся цепочка связана при старте приложения. Сервис вызывает обычный метод LLMClient, клиент выполняет запрос, а наблюдатель создаёт безопасную JSON-запись. Сервисы ничего не знают о журнале и не повторяют его код.

Запустите FastAPI и через Swagger UI выполните анализ обращения или отправьте сообщение в чат:

fastapi dev app/api/main.py

В терминале сначала появится событие отдельного вызова модели:

{
  "timestamp": "2026-09-10T08:05:21.318+00:00",
  "level": "info",
  "logger": "app.observability.llm",
  "event": "llm_call_completed",
  "request_id": "ac76296d95794a5797aa988a617eb952",
  "operation": "generate_structured",
  "model": "gpt-4.1-2025-04-14",
  "finish_reason": "stop",
  "prompt_tokens": 546,
  "completion_tokens": 30,
  "total_tokens": 576,
  "estimated_cost": 0.343596,
  "cost_currency": "RUB",
  "response_id": "chatcmpl-EMqPAnlG0HSTonUQpJP9xPAkdfZx",
  "duration_seconds": 1.73
}

После него middleware запишет http_request_completed с тем же Request ID:

{
  "timestamp": "2026-09-11T07:56:08.848023+00:00",
  "level": "info",
  "logger": "app.api.middleware",
  "event": "http_request_completed",
  "request_id": "ac76296d95794a5797aa988a617eb952",
  "method": "POST",
  "route": "/chat/sessions/{session_id}/messages",
  "status_code": 200,
  "response_start_seconds": 4.580702665960416
}

А если один HTTP-запрос обращался к модели несколько раз, вы увидите несколько LLM-событий, связанных одним идентификатором.

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

Шаг 4

Почему для журналов AI-бэкенда используется одна JSON-запись на одно событие?

Правильный ответ
  • ✅ Стабильные поля можно автоматически искать, фильтровать и агрегировать

Шаг 5

Какие поля уместно записать в событие успешного LLM-вызова?

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

Правильный ответ
  • ✅ Название операции
  • ✅ Название модели
  • ✅ Количество входных и выходных токенов
  • ✅ Продолжительность вызова

Шаг 6

Почему в HTTP-логе лучше сохранять шаблон маршрута /chat/sessions/{session_id}, а не фактический URL каждой сессии?

Правильный ответ
  • ✅ Шаблон не создаёт отдельное значение для каждого идентификатора и удобнее для группировки