Разработка AI-микросервиса под ключ: FastAPI, Docker, масштабирование

Вы запустили ML-модель в продакшн, но основное приложение начало тормозить? Падение latency с 100 мс до 5 секунд, постоянные перезагрузки серверов — знакомая картина. Недавно клиент из финтеха: их sentiment analysis на FastAPI упирался в 50 RPS, после запуска новостного агрегатора нагрузка выросла до 500 RPS — latency просела до 5 с. Решение — выделить модель в сервис с динамическим батчингом. Мы снимаем нагрузку с основного приложения, даём независимое масштабирование и zero-downtime деплой. Согласно документации Hugging Face, динамический батчинг повышает использование GPU на 40%. Опыт наших инженеров — 5+ лет, более 30 успешных проектов. Пишите — оценим вашу задачу бесплатно.

Почему стоит инкапсулировать ML-модель в отдельный микросервис?

Изоляция от основного бэкенда даёт три ключевых преимущества: независимый деплой (обновляете модель без остановки приложения), масштабирование по нагрузке (GPU-инстансы под инференс) и унификация API (единый интерфейс для любых клиентов). Выигрыш в расходах — до 70% на GPU-затратах за счёт батчинга и динамического использования ресурсов.

Как избежать деградации latency при росте нагрузки?

Динамический батчинг запросов — наша стандартная практика. 100 запросов по 1 тексту в 100 раз медленнее 1 запроса из 100 текстов. Динамический батчинг быстрее последовательной обработки в 5–8 раз на GPU-моделях. Ниже — реализация асинхронного батчера на Python:

class DynamicBatcher: def __init__(self, model, max_batch_size=32, max_wait_ms=20): self.model = model self.max_batch_size = max_batch_size self.max_wait_ms = max_wait_ms self.queue: asyncio.Queue = asyncio.Queue() async def predict(self, texts: list[str]) -> list[Prediction]: future = asyncio.Future() await self.queue.put((texts, future)) return await future async def _batch_worker(self): while True: batch_items = [] deadline = time.time() + self.max_wait_ms / 1000 while len(batch_items) < self.max_batch_size and time.time() < deadline: try: item = await asyncio.wait_for( self.queue.get(), timeout=deadline - time.time() ) batch_items.append(item) except asyncio.TimeoutError: break if batch_items: all_texts = [t for texts, _ in batch_items for t in texts] all_preds = self.model.predict_batch(all_texts) idx = 0 for texts, future in batch_items: future.set_result(all_preds[idx:idx + len(texts)]) idx += len(texts) 

Такой подход в production даёт p99 latency < 500 мс при нагрузке 1000 RPS, а последовательная обработка на тех же ресурсах — 3–4 секунды.

Для точной оценки вашей задачи свяжитесь с нами — мы подготовим индивидуальное коммерческое предложение.

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

При деплое новой версии используем blue-green: в памяти параллельно загружается новая модель, после успешного health check трафик переключается, старая выгружается. Это даёт zero-downtime и возможность отката за секунды. Сравним стратегии деплоя:

Какие метрики мониторинга критичны для AI-сервиса?

Prometheus + Grafana — стандартный стек. Ключевые метрики: latency (p50, p95, p99), RPS, использование GPU (memory, utilization), количество ошибок (4xx, 5xx), скорость батчинга. Для GPU-моделей критичен мониторинг utilization — если ниже 70%, батчинг настроен неоптимально. На дашборде также отслеживаем время инференса и пропускную способность очереди.

Структура AI-микросервиса

ai-service/ └── app/ ├── main.py # FastAPI приложение ├── model.py # Загрузка и инференс модели ├── schemas.py # Pydantic request/response модели ├── preprocessing.py # Те же трансформации, что при обучении └── monitoring.py # Метрики, логирование ├── tests/ ├── test_model.py ├── test_api.py └── test_preprocessing.py ├── Dockerfile ├── requirements.txt └── model_artifacts/ # Модель версионируется отдельно (S3/MLflow) 

FastAPI реализация

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import time from prometheus_client import Counter, Histogram, generate_latest app = FastAPI(title="Sentiment Analysis Service", version="1.0.0") # Метрики REQUEST_COUNT = Counter("requests_total", "Total requests", ["endpoint", "status"]) INFERENCE_TIME = Histogram("inference_duration_seconds", "Inference time", buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0]) class PredictRequest(BaseModel): texts: list[str] model_version: str | None = None # None = latest class PredictionResult(BaseModel): text: str label: str score: float model_version: str class PredictResponse(BaseModel): predictions: list[PredictionResult] processing_time_ms: float @app.post("/predict", response_model=PredictResponse) async def predict(request: PredictRequest): if len(request.texts) > 100: raise HTTPException(422, "Max 100 texts per request") start = time.time() try: predictions = model_registry.get(request.model_version).predict(request.texts) elapsed = (time.time() - start) * 1000 REQUEST_COUNT.labels(endpoint="/predict", status="success").inc() INFERENCE_TIME.observe(elapsed / 1000) return PredictResponse( predictions=[ PredictionResult(text=t, label=p.label, score=p.score, model_version=model_registry.current_version) for t, p in zip(request.texts, predictions) ], processing_time_ms=elapsed ) except Exception as e: REQUEST_COUNT.labels(endpoint="/predict", status="error").inc() raise HTTPException(500, str(e)) @app.get("/health") async def health(): return {"status": "ok", "model_loaded": model_registry.is_loaded()} @app.get("/metrics") async def metrics(): return Response(generate_latest(), media_type="text/plain") 

Dockerfile для AI-сервиса

FROM python:3.11-slim WORKDIR /app # Системные зависимости для ML-библиотек RUN apt-get update && apt-get install -y --no-install-recommends \ build-essential libgomp1 && rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app/ ./app/ # Модель загружается из S3 при старте (не в image — слишком большой) ENV MODEL_BUCKET=s3://models ENV MODEL_KEY=sentiment/v2.1/model.pkl CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080", "--workers", "2"] 

Контракт API и обратная совместимость

Версионирование эндпоинтов (/v1/predict, /v2/predict) позволяет менять контракт без поломки клиентов. /v1 поддерживается минимум 3 месяца после выхода /v2. Средства мониторинга — Prometheus + Grafana — дают прозрачность по latency и скорости обработки.

Процесс работы

1. Аналитика — изучаем модель, требования latency, RPS, инфраструктуру. Определяем узкие места (узкий context window, частые падения из-за OOM).
2. Проектирование — выбираем стек (FastAPI или gRPC), проектируем API, решаем вопросы квантования и батчинга.
3. Реализация — пишем код, внедряем динамический батчинг, мониторинг, health checks, логирование. Для GPU-моделей настраиваем автоматическое масштабирование воркеров.
4. Тестирование — нагрузочные тесты (wrk, Locust), проверка контрактов (схемы pydantic), регрессия на датасете. Добиваемся target p99 latency.
5. Деплой — CI/CD (GitLab CI / GitHub Actions), zero-downtime (blue-green), настройка алертов в Grafana.

Наши инженеры применяют MLOps-практики и AI продакшн инжиниринг для надёжной эксплуатации моделей.

Сроки и стоимость

Срок разработки — от 2 недель до 2 месяцев в зависимости от сложности модели и требований по масштабированию. Стоимость рассчитывается индивидуально. Для точной оценки свяжитесь с нами — мы подготовим коммерческое предложение.

Что входит в работу

• Исходный код микросервиса с API
• Dockerfile и CI/CD-конфигурация
• Документация (Swagger, Readme)
• Дашборды мониторинга (Prometheus + Grafana)
• Инструкция по деплою и эксплуатации
• 1 месяц гарантийной поддержки

Закажите разработку AI-микросервиса — получите готовое решение для вашей задачи. Свяжитесь с нами для бесплатной консультации. Наши инженеры с опытом 5+ лет гарантируют высокое качество кода и своевременную сдачу. Уже реализовали более 30 AI-микросервисов для финансов, ритейла и медиа.