Розробка 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-моделей batching критичний: 100 запитів по 1 тексту в 100 разів повільніше за 1 запит зі 100 текстами. Динамічний batching:
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 та зворотна сумісність
Версіонування endpoint'ів (/v1/predict, /v2/predict) дозволяє міняти контракт без поломки клієнтів. /v1 підтримується мінімум 3 місяці після виходу /v2.