Розробка AI-мікросервісу: FastAPI, Docker, масштабування

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

Чому варто інкапсулювати 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-мікросервісу — отримайте готове рішення для вашого завдання. Зв'яжіться з нами для безкоштовної консультації. Наші інженери з досвідом 10+ років гарантують високу якість коду та своєчасне виконання. Вже реалізували понад 40 AI-мікросервісів для фінансів, рітейлу та медіа.