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

Типизированные AI эндпоинты и ошибки

Шаг 1

В предыдущем разделе мы запустили FastAPI-приложение, подключили жизненный цикл (lifespan) и добавили технический маршрут GET /health. У этого маршрута нет входных данных, а его ответ состоит из одного заранее известного поля. Для настоящих AI-функций такого простого интерфейса уже недостаточно.

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

Что такое контракт HTTP API?

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

Для анализа обращения договоримся о следующем взаимодействии:

Здесь существуют сразу три разные проверки:

  • FastAPI и Pydantic проверяют входной HTTP-запрос.
  • SupportRequestAnalyzer выполняет прикладной сценарий.
  • Внутри SupportRequestAnalyzer структурированные ответы задают форму ответа модели, а Pydantic проверяет и преобразует полученные данные.

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

Обратите внимание, мы не будем возвращать внутренний объект напрямую. В проекте уже существует SupportRequestResult и он удобен внутри Python-кода и содержит результат анализа вместе с техническими метаданными. И вроде бы можно было бы вернуть этот dataclass прямо из маршрута, но тогда внутреннее устройство сервиса фактически стало бы публичным контрактом. Добавление служебного поля или переименование атрибута могло бы неожиданно изменить ответ для клиента (например для фронтенд-приложения). Кроме того, во внутреннем результате могут появиться сведения, которые полезны для диагностики, но не предназначены для клиента.

Поэтому HTTP-слой получит собственные модели запроса и ответа. Он явно выберет только разрешённые поля и преобразует в них результат сервиса.

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

Давайте перейдем к практике, и внутри app/api создайте пакет schemas и добавьте в него пустой файл __init__.py. Название schemas здесь означает схемы HTTP-запросов и ответов и не путайте их с JSON Schema, которую мы передаём модели для структурированных ответов, хотя обе технологии используют Pydantic, но они находятся на разных границах приложения.

Создайте файл app/api/schemas/common.py и давайте напишем в него общие сведения о токенах:

from pydantic import BaseModel, Field

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)

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

Теперь создайте app/api/schemas/support_requests.py:

from pydantic import BaseModel, ConfigDict, Field

from app.api.schemas.common import TokenUsageResponse
from app.models.support_request import (
    MissingField,
    SupportCategory,
    Urgency,
)

class SupportRequestAnalyzeRequest(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
        str_strip_whitespace=True,
    )

    text: str = Field(min_length=1, max_length=10_000)

class SupportRequestAnalysisResponse(BaseModel):
    category: SupportCategory
    order_id: str | None
    product: str | None
    problem_description: str | None
    urgency: Urgency
    missing_fields: list[MissingField]
    confidence: float = Field(ge=0, le=1)

class SupportRequestAnalyzeResponse(BaseModel):
    analysis: SupportRequestAnalysisResponse
    model: str
    prompt_id: str
    prompt_version: str
    elapsed_seconds: float = Field(ge=0)
    usage: TokenUsageResponse
    repair_attempts: int = Field(ge=0)

Модель SupportRequestAnalyzeRequest описывает JSON, который приходит от клиента. Мы принимаем только поле text, а также обратите внимание на настройку str_strip_whitespace=True, она убирает пробелы по краям всех строковых полей модели. После очистки min_length=1 отклонит пустую строку и строку, состоящую только из пробелов. А максимальное ограничение max_length=10_000 задаёт понятную границу ещё до обращения к сервису и модели.

Модель SupportRequestAnalysisResponse описывает содержательную часть результата. Она похожа на внутреннюю SupportRequestAnalysis, но принадлежит уже публичному HTTP-контракту. Мы намеренно перечисляем поля, которые разрешено вернуть клиенту.

Модель SupportRequestAnalyzeResponse объединяет анализ и безопасные метаданные. Статистика токенов вынесена в usage, чтобы клиент получал связанные значения одной группой. Поле response_id провайдера в публичную модель не добавляем и позднее для диагностики HTTP-запроса будет использоваться наш собственный Request ID.

До этого момента мы описали только форму данных, а именно какой JSON клиент должен отправить и какой ответ вернёт API. Pydantic проверит входные поля и поможет сформировать ответ, но сам анализ обращения он не выполняет. Ведь после успешной проверки маршруту нужно передать текст в SupportRequestAnalyzer. Этот сервис уже создаётся при запуске приложения и хранится внутри ApplicationContainer. Поэтому теперь возникает следующий вопрос, а как получить готовый анализатор внутри эндпоинта, не создавая его заново и не обращаясь к контейнеру вручную в каждом маршруте?

Можно было бы получать контейнер прямо внутри каждого маршрута:

container = request.app.state.container
result = container.request_analyzer.analyze(payload.text)

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

Для решения этой задачи в FastAPI есть система зависимостей, или Dependency Injection. Мы отдельно описываем, как получить нужный объект, а FastAPI выполняет эту подготовку перед вызовом эндпоинта и передаёт готовое значение в его параметр.

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

Создайте файл app/api/dependencies.py:

from fastapi import Request

from app.application import ApplicationContainer

def get_container(request: Request) -> ApplicationContainer:
    container = getattr(request.app.state, "container", None)
    if not isinstance(container, ApplicationContainer):
        raise RuntimeError(
            "Контейнер приложения не инициализирован."
        )
    return container

get_container() получает текущий HTTP-запрос. Через request.app функция обращается именно к тому экземпляру FastAPI, который обрабатывает запрос, а затем читает контейнер, созданный его lifespan. А getattr() позволяет отдельно проверить, что контейнер действительно существует и имеет ожидаемый тип. При обычном запуске эта проверка всегда проходит. Она поможет быстрее понять причину ошибки, если приложение было создано без lifespan или маршрут начали вызывать до правильной инициализации.

В следующем шаге мы подключим эту функцию непосредственно к эндпоинту. FastAPI вызовет get_container() перед началом основной обработки и поместит возвращённый объект в параметр container. После этого эндпоинт сможет выбрать нужный сервис: container.request_analyzer для анализа обращения или container.chat_service для чата.

Получится короткая цепочка без дополнительных функций для каждого сервиса:

В этой схеме эндпоинт знает о составе ApplicationContainer, но не знает, где контейнер хранится и как создаются находящиеся в нём сервисы.

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

Шаг 2

В предыдущем шаге мы подготовили HTTP-контракт анализа обращения и зависимость для получения ApplicationContainer. Теперь соединим их в одном маршруте. При этом новой AI-логики в этом шаге не появится, так как именно для этого раньше мы разделили проект на отдельные модули и сервисы. Работа с промптом, структурированные ответы, Pydantic-валидация результата, семантическая проверка и исправляющие запросы уже находятся внутри SupportRequestAnalyzer. Этот сервис можно вызывать из консоли, демонстрационного файла или HTTP API, не переписывая его внутреннюю логику.

То есть теперь мы добавляем только новый способ взаимодействия с уже готовой AI-функцией. Маршрут FastAPI примет HTTP-запрос, проверит входные данные, получит SupportRequestAnalyzer из контейнера и вызовет его метод analyze(). После этого эндпоинт преобразует внутренний результат сервиса в публичную модель ответа и вернёт клиенту JSON.

Каждый уровень выполняет свою задачу: FastAPI отвечает за HTTP-взаимодействие, а SupportRequestAnalyzer — за сценарий AI-анализа обращения. Благодаря этому одну и ту же прикладную логику можно использовать через разные интерфейсы.

Давайте перейдем к практике. Внутри app/api/routes создайте файл support_requests.py:

from typing import Annotated

from fastapi import APIRouter, Depends

from app.application import ApplicationContainer
from app.api.dependencies import get_container
from app.api.schemas.common import TokenUsageResponse
from app.api.schemas.support_requests import (
    SupportRequestAnalysisResponse,
    SupportRequestAnalyzeRequest,
    SupportRequestAnalyzeResponse,
)

router = APIRouter(tags=["support requests"])

@router.post("/support-requests/analyze")
def analyze_support_request(
    payload: SupportRequestAnalyzeRequest,
    container: Annotated[
        ApplicationContainer,
        Depends(get_container),
    ],
) -> SupportRequestAnalyzeResponse:
    """Анализирует обращение клиента."""
    result = container.request_analyzer.analyze(payload.text)

    return SupportRequestAnalyzeResponse(
        analysis=SupportRequestAnalysisResponse(
            category=result.analysis.category,
            order_id=result.analysis.order_id,
            product=result.analysis.product,
            problem_description=result.analysis.problem_description,
            urgency=result.analysis.urgency,
            missing_fields=result.analysis.missing_fields,
            confidence=result.analysis.confidence,
        ),
        model=result.model,
        prompt_id=result.prompt_id,
        prompt_version=result.prompt_version,
        elapsed_seconds=result.elapsed_seconds,
        usage=TokenUsageResponse(
            prompt_tokens=result.prompt_tokens,
            completion_tokens=result.completion_tokens,
            total_tokens=result.total_tokens,
        ),
        repair_attempts=result.repair_attempts,
    )

Объект router объединяет маршруты, относящиеся к обращениям поддержки. Далее мы создаем через декоратор эндпоинт с полным адресом /support-requests/analyze, который указываем прямо в декораторе. То есть декоратор @router.post("/support-requests/analyze") регистрирует операцию POST. Этот метод подходит, потому что клиент передаёт тело запроса и запускает обработку данных.

Параметр payload имеет тип SupportRequestAnalyzeRequest. FastAPI понимает, что Pydantic-модель нужно создать из JSON-тела. Если данные не пройдут проверку, эндпоинт вообще не будет вызван.

Параметр container подключает зависимость get_container(). Увидев Depends(get_container), FastAPI вызовет эту функцию, получит ApplicationContainer и только затем начнёт выполнение analyze_support_request(). А внутри эндпоинта нам больше не нужно обращаться к request.app.state, так как параметр container уже содержит готовый контейнер. Из него мы выбираем именно тот сервис, который нужен текущей операции.

Обратите внимание, что маршрут объявлен через обычный def, а не через async def. Наш SupportRequestAnalyzer использует синхронный LLMClient. Во время обращения к ProxyAPI этот код ждёт сетевой ответ и не поддерживает await. А простое добавление async перед такой функцией не сделало бы внутренний LLM-клиент асинхронным и, наоборот, могло бы блокировать обработку других запросов.

Для нашего проекта этого вполне достаточно, поскольку LLMClient предоставляет обычные синхронные методы, эндпоинт также объявляем через def. И FastAPI выполнит такой маршрут в пуле потоков и не заблокирует основной цикл сервера. А полный переход всей цепочки AI-вызовов на async/await потребовал бы отдельного рефакторинга и выходит за рамки этого курса.

Двигаемся дальше. Файл support_requests.py уже содержит готовый маршрут, но основной экземпляр FastAPI пока о нём не знает. Объект router существует отдельно, поэтому его нужно подключить во время сборки HTTP-приложения.

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

from app.api.routes.support_requests import router as support_requests_router

Затем после health_router подключите маршрутизатор анализа обращений:

application.include_router(health_router)
application.include_router(support_requests_router)

Полностью функция сборки HTTP-приложения теперь выглядит так:

def create_api() -> FastAPI:
    application = FastAPI(
        title="AI Support Desk API",
        version="1.0.0",
        lifespan=lifespan,
    )
    application.include_router(health_router)
    application.include_router(support_requests_router)
    return application

Метод include_router() переносит в FastAPI-приложение все операции, зарегистрированные в переданном маршрутизаторе. После этой строки сервер знает и о GET /health, и о POST /support-requests/analyze. Когда мы создадим новые группы маршрутов, подключим их здесь таким же способом.

Давайте запустим сервер:

fastapi dev app/api/main.py

Откройте http://127.0.0.1:8000/docs. В Swagger UI должна появиться группа support requests и операция POST /support-requests/analyze. Нажмите Try it out и отправьте:

{
  "text": "Заказ 8041 с кофемашиной пришёл вчера, но устройство не включается."
}

И после отправки запроса, FastAPI сначала создаст SupportRequestAnalyzeRequest. Затем зависимости получат контейнер и анализатор. Сервис обратится к ProxyAPI и проверит структурированный результат, после чего эндпоинт создаст публичный SupportRequestAnalyzeResponse. FastAPI сериализует его в JSON и вернёт статус 200 OK.

Ответ будет похож на следующий:

Теперь отправьте заведомо неправильное тело:

{
  "message": "Кофемашина не включается"
}

Поле text отсутствует, а неизвестное поле message запрещено. Поэтому Pydantic остановит запрос до маршрута, SupportRequestAnalyzer не запустится и токены ProxyAPI не будут потрачены. FastAPI вернёт стандартный статус 422 Unprocessable Content с описанием ошибок валидации.

А значит наш маршрут уже работает при успешном сценарии, но обработка ошибок пока неоднородна. Ошибка входного JSON возвращается в стандартном формате FastAPI, а сбой ProxyAPI или непроверенный результат модели может превратиться в обычный внутренний ответ сервера. А клиенту нужен единый и безопасный контракт ошибки и в следующем шаге мы добавим такой контракт, и научимся преобразовывать прикладные ошибки в подходящие HTTP-статусы и присвоим каждому запросу собственный Request ID.

Шаг 3

Наш первый AI эндпоинт уже принимает типизированный запрос и возвращает успешный ответ. Теперь нужно привести к такому же порядку ошибки, ведь для HTTP-клиента недостаточно получить длинный трейсбек (traceback) или непостоянный текст исключения, а ему нужны статус, короткий программный код и безопасное объяснение.

При этом нельзя просто вернуть клиенту str(error), так как внутренняя ошибка может содержать сведения о провайдере, параметрах запроса или устройстве приложения. Такие технические подробности нужны разработчику в логах, а внешний ответ должен раскрывать только то, что клиенту действительно разрешено знать.

Какие HTTP-статусы нам понадобятся?

HTTP-статус кратко сообщает общий результат запроса. В этом разделе будем использовать три основных значения:

  • 200 OK — запрос успешно обработан
  • 422 Unprocessable Content — JSON прочитан, но не соответствует Pydantic-контракту
  • 502 Bad Gateway — наш API получил запрос, но не смог завершить AI-операцию из-за внешнего провайдера или некорректного результата модели

А статус 500 Internal Server Error будет означать, что внутри приложения произошла неожиданная ошибка, которую мы заранее не предусмотрели. Например если в коде возник AttributeError, была допущена ошибка при преобразовании результата или нарушилось внутреннее состояние приложения. Клиент не может исправить такую проблему изменением своего запроса, так как её должен найти и устранить разработчик.

И вот именно поэтому не следует превращать любую проблему в 422 Unprocessable Content. Этот статус означает, что клиент передал данные, которые не соответствуют контракту, и может исправить запрос. А если же проблема возникла при обращении к провайдеру, мы лучше вернём 502 Bad Gateway, а непредвиденную ошибку нашего кода оставим со статусом 500.

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

Давайте перейдем к практике и первым делом создайте файл app/api/schemas/errors.py:

from pydantic import BaseModel

class ValidationIssueResponse(BaseModel):
    field: str
    message: str

class ErrorContentResponse(BaseModel):
    code: str
    message: str
    request_id: str
    details: list[ValidationIssueResponse] | None = None

class ErrorResponse(BaseModel):
    error: ErrorContentResponse

Далее все контролируемые ошибки будут иметь верхнее поле error. А уже внутри находятся машинный код, понятное сообщение и ID запроса. Поле details нужно только там, где клиент может исправить конкретные поля. Для сбоя модели оно останется None и вообще не попадёт в JSON.

Теперь пойдем дальше, представим, что несколько пользователей одновременно обращаются к нашему API, а один из них получает ошибку и сообщает: «Анализ обращения не работает». В это время сервер может обрабатывать десятки похожих запросов, поэтому разработчику будет сложно определить, какой именно запрос завершился ошибкой.

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

При этом Request ID не нужно путать с другими идентификаторами проекта:

  • session_id объединяет несколько сообщений одного диалога
  • response_id обозначает отдельный ответ, полученный от ProxyAPI
  • А Request ID относится к одному обращению клиента к нашему HTTP API

Например пользователь отправляет одно сообщение в чат и сервер создаёт для этого HTTP-запроса один Request ID. Во время обработки ChatService может несколько раз обратиться к ProxyAPI, поэтому внутри одного HTTP-запроса может появиться несколько response_id. При следующем сообщении сохранится прежний session_id, но будет создан новый Request ID.

Чтобы Request ID получал каждый HTTP-запрос, добавим Middleware — функцию, которая выполняется при обращении к любому маршруту приложения. Сначала middleware перехватывает входящий запрос и создаёт для него идентификатор. Затем передаёт управление нужному эндпоинт и дожидается результата. Когда эндпоинт сформирует ответ, middleware добавит в него HTTP-заголовок с тем же Request ID и только после этого отправит ответ клиенту.

HTTP-запрос
→ middleware создаёт Request ID
→ выполняется нужный эндпоинт
→ middleware получает готовый ответ
→ добавляет заголовок X-Request-ID
→ ответ отправляется клиенту

Давайте перейдем к практике, создайте app/api/middleware.py:

from uuid import uuid4

from fastapi import Request, Response

async def add_request_id(
    request: Request,
    call_next,
) -> Response:
    request_id = uuid4().hex
    request.state.request_id = request_id

    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

uuid4().hex создаёт случайную строку из 32 шестнадцатеричных символов. Она достаточно уникальна для идентификации наших HTTP-запросов и дальше он сохраняется в request.state. Это состояние принадлежит одному запросу и отличается от request.app.state, где lifespan хранит общий контейнер всего приложения.

call_next(request) передаёт запрос дальше, уже зависимостям, маршруту и обработчикам. После получения ответа middleware добавляет заголовок X-Request-ID. Поэтому идентификатор будет доступен и в успешном ответе, и в контролируемой ошибке.

Теперь создайте app/api/errors.py:

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

from app.api.schemas.errors import (
    ErrorContentResponse,
    ErrorResponse,
    ValidationIssueResponse,
)

class ApiError(Exception):
    def __init__(
        self,
        status_code: int,
        code: str,
        message: str,
    ) -> None:
        super().__init__(message)
        self.status_code = status_code
        self.code = code
        self.message = message

def _get_request_id(request: Request) -> str:
    return getattr(request.state, "request_id", "unknown")

def _error_response(
    request: Request,
    status_code: int,
    code: str,
    message: str,
    details: list[ValidationIssueResponse] | None = None,
) -> JSONResponse:
    payload = ErrorResponse(
        error=ErrorContentResponse(
            code=code,
            message=message,
            request_id=_get_request_id(request),
            details=details,
        )
    )
    return JSONResponse(
        status_code=status_code,
        content=payload.model_dump(exclude_none=True),
    )

async def api_error_handler(
    request: Request,
    error: ApiError,
) -> JSONResponse:
    return _error_response(
        request=request,
        status_code=error.status_code,
        code=error.code,
        message=error.message,
    )

async def validation_error_handler(
    request: Request,
    error: RequestValidationError,
) -> JSONResponse:
    details = [
        ValidationIssueResponse(
            field=".".join(str(part) for part in issue["loc"]),
            message=issue["msg"],
        )
        for issue in error.errors()
    ]
    return _error_response(
        request=request,
        status_code=422,
        code="validation_error",
        message="Запрос не соответствует контракту.",
        details=details,
    )

def register_error_handlers(application: FastAPI) -> None:
    application.add_exception_handler(ApiError, api_error_handler)
    application.add_exception_handler(
        RequestValidationError,
        validation_error_handler,
    )

Давайте подробно разберем этот код. ApiError описывает контролируемую ошибку HTTP-слоя. Мы создаём её в тех ситуациях, когда заранее понимаем, какой ответ должен получить клиент. Например если анализатор не смог получить корректный результат модели, эндпоинт может выбросить:

raise ApiError(
    status_code=502,
    code="support_analysis_failed",
    message="Не удалось проанализировать обращение.",
)

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

После raise ApiError(...) выполнение эндпоинта останавливается, а FastAPI перехватывает исключение и ищет обработчик, зарегистрированный именно для класса ApiError. Таким обработчиком является api_error_handler():

эндпоинт
→ raise ApiError(...)
→ FastAPI перехватывает исключение
→ api_error_handler()
→ _error_response()
→ JSONResponse

Функция api_error_handler() получает текущий Request и выброшенный объект ApiError. Из ошибки она берёт статус, код и сообщение, а затем передаёт их в общую функцию _error_response().

_error_response() отвечает за окончательное формирование ответа. Сначала она получает Request ID из request.state, а затем создаёт Pydantic-модель ErrorResponse. Благодаря этому все контролируемые ошибки имеют одинаковую структуру:

{
  "error": {
    "code": "support_analysis_failed",
    "message": "Не удалось проанализировать обращение.",
    "request_id": "a1b2c3d4"
  }
}

А после проверки Pydantic-модель преобразуется в обычный Python-словарь методом:

payload.model_dump(exclude_none=True)

Второй обработчик, validation_error_handler(), работает с ошибками входных данных. Такие ошибки создаёт сам FastAPI, когда JSON не соответствует Pydantic-модели запроса. Например, клиент не передал обязательное поле text или указал строку, превышающую разрешённую длину.

В этом случае эндпоинт вообще не запускается:

HTTP-запрос
→ FastAPI проверяет JSON через Pydantic
→ данные не проходят проверку
→ RequestValidationError
→ validation_error_handler()
→ _error_response()
→ ответ со статусом 422

Метод error.errors() возвращает подробный список найденных проблем. В нём могут находиться расположение поля, сообщение, тип ошибки и часть исходных данных. Клиенту нужны не все технические сведения, поэтому мы оставляем только название проблемного поля и понятное сообщение:

{
  "field": "body.text",
  "message": "Field required"
}

Полное тело запроса в ответ не возвращается. Оно может содержать персональные или другие чувствительные данные.

Обе функции начнут работать только после регистрации в FastAPI. Для этого используется register_error_handlers():

def register_error_handlers(application: FastAPI) -> None:
    application.add_exception_handler(
        ApiError,
        api_error_handler,
    )
    application.add_exception_handler(
        RequestValidationError,
        validation_error_handler,
    )

Каждый вызов add_exception_handler() связывает тип исключения с функцией, которая должна его обработать.

Теперь откройте app/api/main.py и добавьте импорты:

from app.api.errors import register_error_handlers
from app.api.middleware import add_request_id

Внутри create_api(), сразу после создания application, подключите middleware и обработчики:

application.middleware("http")(add_request_id)
register_error_handlers(application)

После всех изменений функция будет выглядеть так:

def create_api() -> FastAPI:
    application = FastAPI(
        title="AI Support Desk API",
        version="1.0.0",
        lifespan=lifespan,
    )
    application.middleware("http")(add_request_id)
    register_error_handlers(application)
    application.include_router(health_router)
    application.include_router(support_requests_router)
    return application

Запись application.middleware("http")(add_request_id) программно применяет тот же механизм, который в простом примере обычно записывают через декоратор @app.middleware("http"). Мы используем такую форму, потому что экземпляр FastAPI создаётся внутри фабрики, а сама функция middleware находится в отдельном модуле.

И теперь нам осталось связать внутреннюю ошибку анализатора с HTTP-контрактом. Откройте app/api/routes/support_requests.py и дополните импорты:

from fastapi import APIRouter, Depends, status

from app.api.errors import ApiError
from app.api.schemas.errors import ErrorResponse
from app.services.support_request_analyzer import SupportRequestAnalyzerError

А теперь замените прежний декоратор @router.post("/support-requests/analyze"). Функцию под ним пока не удаляйте:

@router.post(
    "/support-requests/analyze",
    responses={
        422: {
            "model": ErrorResponse,
            "description": "Запрос не прошёл проверку.",
        },
        502: {
            "model": ErrorResponse,
            "description": "AI-анализ не выполнен.",
        },
    },
)

Этот параметр responses не выполняет обработку ошибок. Он дополняет OpenAPI-документацию и сообщает клиентам, что кроме успешной модели маршрут может вернуть ErrorResponse со статусом 422 или 502. Сам ответ формируют зарегистрированные обработчики.

После этого внутри analyze_support_request() найдите строку:

result = container.request_analyzer.analyze(payload.text)

Замените только её следующим блоком:

try:
    result = container.request_analyzer.analyze(payload.text)
except SupportRequestAnalyzerError as error:
    raise ApiError(
        status_code=status.HTTP_502_BAD_GATEWAY,
        code="support_analysis_failed",
        message="Не удалось проанализировать обращение.",
    ) from error

Весь код ниже, где создаются SupportRequestAnalyzeResponse, SupportRequestAnalysisResponse и TokenUsageResponse, оставьте без изменений. И мы добавили обработку вокруг вызова сервиса, но не меняли формирование успешного ответа.

И теперь маршрут перехватывает только ожидаемую прикладную ошибку SupportRequestAnalyzerError, при этом клиент не получает её исходный текст. Вместо него создаётся безопасная ApiError со статусом 502 и стабильным кодом support_analysis_failed. А конструкция from error сохраняет исходную причину для Python и будущих логов, но не добавляет её в JSON-ответ.

Перезапустите сервер и снова отправьте неправильный JSON. Теперь ответ будет иметь единый вид:

При этом в заголовках ответа находится тот же идентификатор:

x-request-id: bcd5ad027e5e4dd4881f6f3de0ce7a9a

Теперь у HTTP-слоя появилась общая основа для ошибок. Pydantic-валидация и сбой AI-операции возвращают один контракт, а Request ID помогает связать ответ клиента с обработкой конкретного запроса. В следующем шаге мы используем эту основу для второго AI эндпоинта, в котором подключим обычный ответ чата, проверим сохранение истории по session_id и разберём итоговую OpenAPI-документацию раздела.

Шаг 4

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

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

Создайте файл app/api/schemas/chat.py:

from pydantic import BaseModel, ConfigDict, Field

from app.api.schemas.common import TokenUsageResponse

class ChatReplyRequest(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
        str_strip_whitespace=True,
    )

    session_id: str = Field(min_length=1, max_length=128)
    message: str = Field(min_length=1, max_length=10_000)

class ChatReplyResponse(BaseModel):
    text: str
    session_id: str
    history_messages: int = Field(ge=0)
    model: str
    prompt_id: str
    prompt_version: str
    elapsed_seconds: float = Field(ge=0)
    usage: TokenUsageResponse
    tool_calls_count: int = Field(ge=0)
    tool_names: list[str]
    summary_updated: bool
    has_summary: bool

Запрос содержит два обязательных поля. session_id сообщает сервису, к какому диалогу относится сообщение, а message содержит новую реплику пользователя. Ограничения отклоняют пустые и слишком большие строки ещё до вызова ChatService.

Пока session_id является обычной строкой, потому что именно такой контракт уже поддерживает сервис. Клиент может использовать, например такую строку demo-session. В следующем разделе мы сделаем так, что идентификаторы будет создавать сервер, и клиент перестанет придумывать их самостоятельно.

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

Обратите внимание, tool_names представлен как JSON-массив, поэтому во внешней модели используем list[str], хотя внутренний ChatReply хранит неизменяемый кортеж. Это ещё один пример того, что HTTP-модель не обязана полностью повторять внутренний Python-тип.

Для маршрута чата не будем создавать ещё одну функцию зависимости. Мы повторно используем get_container(), а затем прямо внутри эндпоинта обратимся к container.chat_service. Так все маршруты получают общий контейнер одинаковым способом, но самостоятельно выбирают нужный сервис.

Теперь давайте подключим ChatService к HTTP API, для этого создайте файл app/api/routes/chat.py:

from typing import Annotated

from fastapi import APIRouter, Depends, status

from app.application import ApplicationContainer
from app.api.dependencies import get_container
from app.api.errors import ApiError
from app.api.schemas.chat import ChatReplyRequest, ChatReplyResponse
from app.api.schemas.common import TokenUsageResponse
from app.api.schemas.errors import ErrorResponse
from app.services.chat_service import ChatServiceError

router = APIRouter(tags=["chat"])

@router.post(
    "/chat/reply",
    responses={
        422: {
            "model": ErrorResponse,
            "description": "Запрос не прошёл проверку.",
        },
        502: {
            "model": ErrorResponse,
            "description": "Ответ чата не получен.",
        },
    },
)
def reply_to_chat(
    payload: ChatReplyRequest,
    container: Annotated[
        ApplicationContainer,
        Depends(get_container),
    ],
) -> ChatReplyResponse:
    """Возвращает полный ответ чата без потоковой передачи."""
    try:
        result = container.chat_service.reply(
            session_id=payload.session_id,
            user_text=payload.message,
        )
    except ChatServiceError as error:
        raise ApiError(
            status_code=status.HTTP_502_BAD_GATEWAY,
            code="chat_reply_failed",
            message="Не удалось получить ответ чата.",
        ) from error

    return ChatReplyResponse(
        text=result.text,
        session_id=result.session_id,
        history_messages=result.history_messages,
        model=result.model,
        prompt_id=result.prompt_id,
        prompt_version=result.prompt_version,
        elapsed_seconds=result.elapsed_seconds,
        usage=TokenUsageResponse(
            prompt_tokens=result.prompt_tokens,
            completion_tokens=result.completion_tokens,
            total_tokens=result.total_tokens,
        ),
        tool_calls_count=result.tool_calls_count,
        tool_names=list(result.tool_names),
        summary_updated=result.summary_updated,
        has_summary=result.has_summary,
    )

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

Маршрут вызывает обычный метод reply(), поэтому клиент получает только завершённый ответ целиком. Внутри сервиса этот метод по-прежнему использует потоковые события, но собирает их до ChatStreamCompleted.

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

Маршрут чата пока находится только в файле chat.py. Чтобы он появился в приложении, откройте app/api/main.py и добавьте импорт:

from app.api.routes.chat import router as chat_router

Затем внутри create_api(), после уже подключённых маршрутизаторов, добавьте ещё одну строку:

application.include_router(health_router)
application.include_router(support_requests_router)
application.include_router(chat_router)

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

Перейдем к проверке работы, для этого запустите сервер:

fastapi dev app/api/main.py

Откройте http://127.0.0.1:8000/docs. Теперь документация содержит три группы операций:

  • system — технический маршрут GET /health
  • support requests — структурированный анализ обращения
  • chat — обычный ответ чата

Откройте POST /chat/reply, нажмите Try it out и отправьте:

{
  "session_id": "demo-session",
  "message": "Монитор пришёл с треснутой подставкой."
}

И мы получим следующий ответ:

После ответа повторно вызовите тот же эндпоинт с тем же session_id:

{
  "session_id": "demo-session",
  "message": "Номер заказа 7718."
}

ChatService найдёт существующую сессию и добавит новую пару сообщений в её историю. В поле history_messages станет больше элементов, а модель сможет связать номер заказа с проблемой из предыдущей реплики.

Если заменить session_id на "another-session", сервис создаст отдельную историю и новая сессия не узнает о треснутой подставке.

Также посмотрите ниже Swagger UI. FastAPI построил примеры ответов из Pydantic-моделей. В них видны обязательные поля, ограничения длины, перечисления категорий и срочности, успешные ответы и добавленные нами ошибки.

И в конце, давайте подведем итоги наших доработок. За этот раздел мы превратили два существующих Python-сервиса в типизированные HTTP-операции. Pydantic проверяет тело до запуска AI-функции, зависимости извлекают готовые сервисы из контейнера, а маршруты только связывают HTTP-контракт с прикладными методами.

Мы также добавили безопасное преобразование прикладных ошибок и присвоили каждому запросу диагностический идентификатор. При этом ни SupportRequestAnalyzer, ни ChatService не получили зависимости от FastAPI.

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

Шаг 5

Что произойдёт, если тело запроса не соответствует Pydantic-модели параметра маршрута?

Правильный ответ
  • ✅ FastAPI вернёт ошибку 422 до выполнения функции маршрута

Шаг 6

Зачем HTTP-слою нужны отдельные модели запросов и ответов?

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

Правильный ответ
  • ✅ Они дают FastAPI данные для проверки и OpenAPI-документации
  • ✅ Они позволяют не раскрывать внутренние служебные поля
  • ✅ Они явно задают публичный контракт API

Шаг 7

Что делает выражение Depends(get_container) в параметре эндпоинта?

Правильный ответ
  • ✅ Просит FastAPI вызвать зависимость и передать её результат в эндпоинт

Шаг 8

Модель тела запроса анализа

Эндпоинт POST /support-requests/analyze получает текст обращения в JSON-теле запроса. Вам нужно создать Pydantic-модель, которая проверит поле text до передачи данных в функцию-эндпоинт.

Задача

Создайте Pydantic-модель SupportRequestAnalyzeRequest для тела запроса POST /support-requests/analyze. В модели должно быть одно обязательное строковое поле text с указанными ограничениями.

Требования

  • Не изменяйте имя класса SupportRequestAnalyzeRequest и поля text.
  • Класс должен наследоваться от BaseModel.
  • Задайте model_config = ConfigDict(extra="forbid", str_strip_whitespace=True).
  • Объявите обязательное поле text: str без значения по умолчанию.
  • Используйте Field(min_length=1, max_length=10_000).
  • Пробелы по краям строки должны удаляться до передачи значения в маршрут.
  • Отсутствующее поле, пустая строка, строка из пробелов, значение длиннее 10 000 символов и любой дополнительный ключ должны приводить к HTTP-статусу 422.

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

Клиент отправляет запрос:

POST /support-requests/analyze
Content-Type: application/json

{"text":"  Кофемашина не включается.  "}

Тестовый маршрут возвращает очищенное значение:

200 OK
Content-Type: application/json

{"text":"Кофемашина не включается."}
Правильный ответ 
from pydantic import BaseModel, ConfigDict, Field

class SupportRequestAnalyzeRequest(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
        str_strip_whitespace=True,
    )

    text: str = Field(
        min_length=1,
        max_length=10_000,
    )

Шаг 9

Зависимость для получения контейнера

Общий контейнер с сервисами уже сохранён в request.app.state.container при запуске FastAPI-приложения. Вам нужно написать функцию-зависимость, которая получит этот контейнер, проверит его тип и передаст его функции-эндпоинту.

Задача

Реализуйте функцию-зависимость get_container(request). Она должна вернуть корректный ApplicationContainer из состояния приложения или выбросить заданную ошибку, если контейнер отсутствует либо имеет неверный тип.

Требования

  • Не изменяйте имя функции и аргумент request.
  • Получите значение из request.app.state.container через getattr(request.app.state, "container", None).
  • Проверьте объект выражением isinstance(container, ApplicationContainer).
  • Если контейнер отсутствует или имеет другой тип, выбросьте RuntimeError с текстом "Контейнер приложения не инициализирован.".
  • Если проверка прошла, верните тот же объект контейнера.
  • Не вызывайте ApplicationContainer(), Settings() или build_application() внутри зависимости.
  • Функция вызывается для каждого запроса, но всегда читает контейнер, который заранее сохранил lifespan.

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

Когда контейнер находится в app.state, клиент отправляет запрос:

GET /container

Тестовый маршрут получает объект через Depends и отвечает:

200 OK
Content-Type: application/json

{"ready":true}

Импорт ApplicationContainer из app.application уже добавлен в редактор.

Правильный ответ 
from fastapi import Request

from app.application import ApplicationContainer

def get_container(request: Request):
    container = getattr(request.app.state, "container", None)

    if not isinstance(container, ApplicationContainer):
        raise RuntimeError("Контейнер приложения не инициализирован.")

    return container

Шаг 10

Middleware с Request ID

В заготовке уже объявлена middleware-функция add_request_id, которая выполняется для каждого HTTP-запроса. Вам нужно создать Request ID, сохранить его в объекте запроса и добавить это же значение в заголовок ответа.

Задача

Реализуйте асинхронную middleware-функцию add_request_id(request, call_next). Сохраните новый идентификатор в состоянии запроса, вызовите следующий обработчик и добавьте идентификатор в заголовок полученного HTTP-ответа.

Требования

  • Не изменяйте имя функции и её аргументы.
  • До вызова следующего обработчика создайте идентификатор единственным вызовом uuid4().hex.
  • Сохраните идентификатор в request.state.request_id.
  • Вызовите response = await call_next(request) ровно один раз.
  • После получения ответа добавьте тот же идентификатор в заголовок X-Request-ID.
  • Верните полученный объект response, не создавая новый HTTP-ответ.
  • Для разных запросов должны создаваться разные идентификаторы.
  • Не перехватывайте исключения, возникшие внутри следующего обработчика.

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

Клиент отправляет запрос:

GET /request-id

Маршрут читает идентификатор из request.state, а middleware добавляет его в заголовок:

200 OK
X-Request-ID: a1b2c3d4...
Content-Type: application/json

{"request_id":"a1b2c3d4..."}
Правильный ответ 
from uuid import uuid4

async def add_request_id(request, call_next):
    request_id = uuid4().hex
    request.state.request_id = request_id

    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id

    return response

Шаг 11

Единый HTTP-ответ об ошибке

Клиент должен получать все ошибки API в одинаковом JSON-формате с кодом, сообщением и Request ID. Вам нужно написать вспомогательную функцию, которая соберёт такое тело и вернёт JSONResponse с нужным HTTP-статусом.

Задача

Реализуйте вспомогательную функцию build_error_response(request, status_code, code, message, details=None). Она должна сформировать тело ошибки, взять Request ID из состояния запроса и вернуть JSONResponse с переданным HTTP-статусом.

Требования

  • Не изменяйте имя функции, аргументы и значение details по умолчанию.
  • Получите Request ID через getattr(request.state, "request_id", "unknown").
  • Соберите тело вида {"error": {...}}.
  • Внутри error обязательно должны находиться ключи code, message и request_id.
  • Добавляйте details только когда аргумент не равен None. Пустой список необходимо сохранить.
  • Создайте JSONResponse(status_code=status_code, content=payload).
  • Верните созданный JSONResponse.
  • Не добавляйте traceback, исходное исключение или другие служебные данные.

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

Тестовый маршрут вызывает функцию для ошибки проверки данных:

GET /demo-error

Ожидаемый HTTP-ответ:

422 Unprocessable Entity
Content-Type: application/json

{
  "error": {
    "code": "validation_error",
    "message": "Запрос не соответствует контракту.",
    "request_id": "req-1"
  }
}
Правильный ответ 
from fastapi.responses import JSONResponse

def build_error_response(request, status_code: int, code: str, message: str, details=None) -> JSONResponse:
    request_id = getattr(request.state, "request_id", "unknown")

    payload = {
        "error": {
            "code": code,
            "message": message,
            "request_id": request_id,
        }
    }

    if details is not None:
        payload["error"]["details"] = details

    return JSONResponse(
        status_code=status_code,
        content=payload,
    )

Шаг 12

Типизированный эндпоинт анализа обращения

В заготовке уже находятся Pydantic-модель AnalyzeRequest и функция-эндпоинт analyze_support_request для запроса POST /support-requests/analyze. Вам нужно настроить проверку текста, вызвать анализатор и вернуть клиенту результат либо безопасную ошибку со статусом 502.

Задача

Дополните модель AnalyzeRequest и функцию-эндпоинт analyze_support_request, зарегистрированную для POST /support-requests/analyze. Настройте проверку поля text, вызовите полученный через зависимость анализатор и сформируйте успешный либо ошибочный ответ.

Требования

  • В AnalyzeRequest задайте ConfigDict(extra="forbid", str_strip_whitespace=True).
  • Объявите обязательное поле text: str = Field(min_length=1, max_length=10_000).
  • Сохраните декоратор @router.post("/support-requests/analyze").
  • Получайте анализатор через параметр analyzer=Depends(get_analyzer). Не вызывайте зависимость вручную.
  • Вызовите analyzer.analyze(payload.text) ровно один раз.
  • При успехе верните словарь ровно с ключами category и order_id, взяв значения из результата.
  • Если анализатор выбросил AnalyzerError, создайте ApiError со статусом 502, кодом "support_analysis_failed" и сообщением "Не удалось проанализировать обращение.".
  • Сохраните исходное исключение конструкцией raise ... from error, но не включайте его текст в HTTP-ответ.
  • Пустой текст и дополнительные поля должны отклоняться FastAPI со статусом 422 до вызова анализатора.

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

Клиент отправляет корректный запрос:

POST /support-requests/analyze
Content-Type: application/json

{"text":"Заказ 8041, кофемашина не включается."}

Фейковый анализатор возвращает результат, а маршрут формирует HTTP-ответ:

200 OK
Content-Type: application/json

{
  "category": "technical_issue",
  "order_id": "8041"
}

Если анализатор завершится ожидаемой ошибкой, API должен вернуть статус 502 и публичный код support_analysis_failed.

Правильный ответ 
from fastapi import Depends
from pydantic import BaseModel, ConfigDict, Field

from task_environment import AnalyzerError, ApiError, get_analyzer, router

class AnalyzeRequest(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
        str_strip_whitespace=True,
    )

    text: str = Field(
        min_length=1,
        max_length=10_000,
    )

@router.post("/support-requests/analyze")
def analyze_support_request(payload: AnalyzeRequest, analyzer=Depends(get_analyzer)):
    try:
        result = analyzer.analyze(payload.text)
    except AnalyzerError as error:
        raise ApiError(
            status_code=502,
            code="support_analysis_failed",
            message="Не удалось проанализировать обращение.",
        ) from error

    return {
        "category": result.category,
        "order_id": result.order_id,
    }