Разработка AI-микросервиса

Разработка AI-микросервиса

AI-микросервис — отдельный HTTP/gRPC сервис, инкапсулирующий ML-модель или AI-логику. Изолирован от основного приложения: независимый деплой, масштабирование, технологический стек. Паттерн для production AI-систем.

Структура 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"]

Batching для эффективности

Для GPU-моделей батчинг критичен: 100 запросов по 1 тексту в 100 раз медленнее 1 запроса из 100 текстов. Динамический батчинг:

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)
                # Распределяем результаты по futures
                idx = 0
                for texts, future in batch_items:
                    future.set_result(all_preds[idx:idx + len(texts)])
                    idx += len(texts)

Versioned model loading

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

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

Версионирование эндпоинтов (/v1/predict, /v2/predict) позволяет менять контракт без поломки клиентов. /v1 поддерживается минимум 3 месяца после выхода /v2.