Serving и инфраструктура

Serving и инфраструктура — как рекомендации работают на проде

Модель обучена. А дальше что? 100 миллионов пользователей, каждый запрос за 100мс, тысячи RPS. Serving — это про то, как перевести модель из Jupyter в систему, которая работает 24/7 под нагрузкой. И это не «завернул в Flask» — это архитектура из 5-6 компонентов, каждый со своим SLA.

Архитектура serving-сервиса

Типичная архитектура: API Gateway → рекомендательный сервис → Feature Store (Redis) + Model Serving (Triton) + ANN-индекс (FAISS). Всё за nginx/envoy с балансировкой и circuit breaker.

• Model serving: TorchServe, Triton (NVIDIA), vLLM. GPU-батчинг — накапливаем запросы и прогоняем батчом
• Feature serving: Redis/Memcached для онлайн-фич, <1мс на запрос. Пользовательские статистики, CTR по категориям
• ANN-индекс: FAISS, Milvus, Qdrant. Хранит item-эмбеддинги, обновляется каждый час. Поиск по 100М векторов за 5мс
• Кэширование: рекомендации для активных пользователей — в кэш на 5-15 минут. Экономит 60-80% compute

# Минимальный serving рекомендаций на FastAPI
from fastapi import FastAPI
import faiss, redis, numpy as np

app = FastAPI()
r = redis.Redis(host="redis", port=6379)
index = faiss.read_index("item_embeddings.faiss")  # ANN-индекс

@app.get("/recommend/{user_id}")
async def recommend(user_id: str, k: int = 20):
    # 1. Проверяем кэш (5мс вместо 100мс)
    cached = r.get(f"recs:{user_id}")
    if cached:
        return {"items": json.loads(cached), "source": "cache"}

    # 2. Получаем user embedding из Feature Store
    user_emb = np.frombuffer(r.get(f"emb:{user_id}"), dtype="float32").reshape(1, -1)

    # 3. ANN-поиск кандидатов (~5мс на 100М айтемов)
    scores, item_ids = index.search(user_emb, k=200)

    # 4. Ранжирование (тяжёлая модель на top-200)
    ranked = ranker.predict(user_id, item_ids[0][:200])
    top_k = ranked[:k]

    # 5. Кэшируем на 10 минут
    r.setex(f"recs:{user_id}", 600, json.dumps(top_k))
    return {"items": top_k, "source": "model"}

Задержка — бюджет по миллисекундам

Пользователь открывает ленту — и через 200мс должен увидеть рекомендации. Если дольше — уходит. Бюджет распределяется по этапам:

• Feature fetch: 5мс — Redis за user/item фичами
• Candidate generation: 20мс — ANN-поиск по FAISS/Milvus
• Ranking: 50мс — нейросеть/бустинг на top-200 кандидатов, GPU batch
• Re-ranking: 5мс — diversity, бизнес-правила, exploration slots
• Сеть + сериализация: 20мс — от сервиса до клиента
• Итого: ~100мс p50, <200мс p99. Каждый компонент оптимизируется до последней миллисекунды

Масштабирование

• Горизонтальное: добавляем реплики stateless-сервиса, балансируем через nginx/envoy
• GPU батчинг: Triton с динамическим батчингом — накапливает запросы 1-5мс, прогоняет батчом. 10x throughput
• Шардирование: ANN-индекс по 100М+ айтемов — шардим по хэшу item_id. Каждый шард на своём сервере
• Предрасчёт: для неактивных пользователей (не заходили 7+ дней) рекомендации считаем офлайн и кладём в Redis

Обновление моделей и индексов

Модель устаревает — новые товары, изменившиеся вкусы. Новые айтемы не попадают в ANN-индекс. Нужен цикл: переобучение (раз в день/неделю) → офлайн-валидация (NDCG не упал) → канареечный деплой (5% трафика) → полный выкат.

Blue-green deployment — две полные копии сервиса: Blue (текущая) и Green (новая). Переключение мгновенное через балансер. Если Green сломалась — откат за секунды. Этот подход широко используется для рекомендательных сервисов.

# Graceful degradation — fallback chain для рекомендаций
import asyncio
import logging

logger = logging.getLogger("recsys")

class FallbackChain:
    """Цепочка fallback-ов: если основная модель упала — не показываем пустоту."""
    
    def __init__(self):
        self.strategies = []
    
    def add(self, name: str, fn, timeout_ms: float = 100):
        self.strategies.append({"name": name, "fn": fn, "timeout": timeout_ms})
        return self
    
    async def execute(self, user_id: str, k: int = 10):
        for strategy in self.strategies:
            try:
                result = await asyncio.wait_for(
                    strategy["fn"](user_id, k),
                    timeout=strategy["timeout"] / 1000
                )
                if result:
                    metrics.inc("recsys.strategy", tags={"name": strategy["name"]})
                    return result
            except (asyncio.TimeoutError, Exception) as e:
                logger.warning(f"{strategy['name']} failed: {e}")
                continue
        return await get_global_popular(k)  # последний рубеж

# Использование: каждый уровень — всё проще и надёжнее
chain = FallbackChain()
chain.add("full_pipeline", full_ranking, timeout_ms=200)     # полный пайплайн
chain.add("light_ranking", light_model_only, timeout_ms=100)  # без тяжёлой модели
chain.add("cached", get_cached_recs, timeout_ms=10)           # кэш
chain.add("popular", get_popular, timeout_ms=5)                # популярное

Graceful degradation — когда что-то сломалось

Пользователь не должен видеть пустую страницу. Каждый компонент должен иметь fallback:

• Модель ранжирования упала → отдаём результаты candidate generation без ранжирования
• Redis недоступен → используем in-memory кэш (LRU, последние 10K пользователей)
• ANN-индекс не обновился → работаем со старой версией (лучше вчерашние рекомендации, чем никакие)
• Всё сломалось → popularity fallback: топ-100 айтемов за последние 24 часа