MaaS API: разработка обёртки для AI-модели с FastAPI и мониторингом

Мы часто видим: вы обучили модель, потратили недели на настройку гиперпараметров и получили отличные метрики. Но как теперь отдать её клиентам? Просто передать веса не получится. Нужен API с аутентификацией, версионированием, мониторингом. Сырая модель — не endpoint. Наша команда разрабатывает API-обёртку, которая решает эти проблемы на уровне кода. Без грамотной обёртки модель остаётся недоступной для внешних систем, а ручная интеграция каждого клиента — путь к хаосу и утечкам данных. Такой подход превращает модель в полноценный микросервис для машинного обучения.

Архитектура MaaS API

[Client] → [API Gateway] → [Auth/Rate Limit] → [Request Validation] → [Model Router] → [Inference Service] → [Response Formatter] ↕ ↕ [Usage Logger] [Cache Layer] 

Клиент отправляет запрос, API gateway проверяет ключ, rate limiter контролирует частоту, а кеш (Redis) возвращает результат для повторяющихся запросов. Только если кеш пуст, запрос идёт к модели. Это снижает нагрузку и улучшает latency.

Реализация на FastAPI

from fastapi import FastAPI, HTTPException, Depends, Header from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field import time import hashlib app = FastAPI(title="Model-as-a-Service API", version="1.0.0") class PredictionRequest(BaseModel): inputs: list[dict] = Field(..., description="List of feature dictionaries") model_version: str = Field(default="latest") options: dict = Field(default_factory=dict) class PredictionResponse(BaseModel): predictions: list model_version: str request_id: str latency_ms: float async def verify_api_key(x_api_key: str = Header(...)): if not await api_key_store.verify(x_api_key): raise HTTPException(status_code=401, detail="Invalid API key") return await api_key_store.get_client(x_api_key) @app.post("/v1/predict", response_model=PredictionResponse) async def predict( request: PredictionRequest, client = Depends(verify_api_key) ): # Rate limiting if not await rate_limiter.check(client.id, limit=100, window=60): raise HTTPException(status_code=429, detail="Rate limit exceeded") # Cache check cache_key = hashlib.md5(str(request.inputs).encode()).hexdigest() cached = await cache.get(cache_key) if cached: return cached # Inference start = time.perf_counter() try: model = model_registry.get(request.model_version) predictions = model.predict(request.inputs) except Exception as e: await logger.error(client.id, request, str(e)) raise HTTPException(status_code=500, detail=str(e)) latency = (time.perf_counter() - start) * 1000 response = PredictionResponse( predictions=predictions, model_version=model.version, request_id=generate_request_id(), latency_ms=latency ) # Log usage await usage_logger.log(client.id, request, response, latency) await cache.set(cache_key, response, ttl=300) return response 

FastAPI использует Pydantic для валидации данных и автоматической генерации документации. В сравнении с Flask он выигрывает в производительности: P95 latency снижается в 2–3 раза при одинаковой нагрузке. Это подтверждают данные из FastAPI benchmarks.

Почему FastAPI лучше Flask для ML API?

FastAPI обеспечивает P95 latency в 2–3 раза ниже при высоких нагрузках за счёт асинхронной обработки и автоматической валидации. По данным официальных бенчмарков, он обрабатывает до 1000 RPS на одном инстансе, тогда как Flask — около 300. Это критично для продуктовых ML-сервисов, где каждая миллисекунда влияет на пользовательский опыт. REST API для ML модели должен быть отказоустойчивым и масштабируемым.

Версионирование API

# v1 — legacy формат @app.post("/v1/predict") async def predict_v1(request: PredictionRequestV1): ... # v2 — новый формат с batch поддержкой @app.post("/v2/predict") async def predict_v2(request: PredictionRequestV2): ... # Deprecation header для v1 @app.middleware("http") async def add_deprecation_header(request, call_next): response = await call_next(request) if request.url.path.startswith("/v1/"): response.headers["Deprecation"] = "true" response.headers["Sunset"] = "устанавливается при деплое" return response 

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

Как мы обеспечиваем безопасность и производительность?

Безопасность строится на трёх уровнях: аутентификация (API-ключи или JWT), rate limiting (ограничение запросов в минуту на клиента) и валидация входных данных через Pydantic. Для производительности используем кэширование в Redis с TTL 5 минут. Типичный hit rate кэша для повторяющихся запросов — 40–60%, что снижает latency на 30–50%.

Типичные проблемы при запуске ML-модели в продакшен

Часто сталкиваемся с тремя проблемами при запуске ML-модели. Первая — отсутствие контроля доступа: любой может вызвать модель, что ведёт к перегрузке и неконтролируемым расходам. Мы решаем это API-ключами и rate limiting на основе Token Bucket. Вторая — обновление модели с простоем: пока меняются веса, сервис недоступен. Помогает версионирование и blue-green deployment. Третья — отсутствие мониторинга: вы не знаете ни числа запросов, ни задержки. Мы ставим Prometheus + Grafana с автоматическими алертами.

Что входит в разработку API-обёртки?

Дополнительно: webhook callbacks для долгих предсказаний, поддержка квантованных моделей (INT4/INT8) для снижения cost per token.

Кейс: как мы сократили latency на 40%

Для клиента с LLM-моделью на базе LLaMA 3 мы внедрили батчинг запросов (batch size 8) и квантование модели до INT4. Это снизило p95 latency с 800 ms до 450 ms и увеличило пропускную способность в 2 раза. Cost per token уменьшился на 35% за счёт более эффективной утилизации GPU. Инференс выполнялся на Triton Inference Server.

Свяжитесь с нами для оценки вашего проекта. Мы разработаем API-обёртку под ключ — оценим сроки за 1–2 дня. Закажите консультацию, чтобы обсудить детали.