|
1 | | -"""Validated recommendation routes with bounded inference and fallback behavior.""" |
2 | | - |
3 | | -from __future__ import annotations |
4 | | - |
5 | | -import hashlib |
6 | | -import json |
7 | | -import logging |
8 | | -import time |
9 | | -from dataclasses import dataclass |
10 | | -from typing import Literal |
11 | | - |
12 | | -from fastapi import APIRouter, Header, HTTPException |
13 | | -from pydantic import BaseModel, Field |
14 | | - |
15 | | -from app.core.config import settings |
16 | | -from app.core.data_processor import SequenceProcessor |
17 | | -from app.core.metrics import ( |
18 | | - active_requests, |
19 | | - cache_hits_total, |
20 | | - cache_misses_total, |
21 | | - feedback_events_total, |
22 | | - model_inference_latency, |
23 | | - recommendation_latency, |
24 | | - recommendations_total, |
25 | | -) |
26 | | -from app.core.model import DeepSequenceModel |
27 | | -from app.core.security import api_key_is_valid |
28 | | -from app.core.serving import AdmissionController, RateLimiter, RecommendationCache |
29 | | - |
30 | | -router = APIRouter(prefix="/recommendations", tags=["recommendations"]) |
31 | | - |
32 | | - |
33 | | -@dataclass |
34 | | -class ModelRuntime: |
35 | | - processor: SequenceProcessor |
36 | | - model: DeepSequenceModel |
37 | | - model_version: str |
38 | | - trained: bool |
39 | | - popular_items: list[str] |
40 | | - |
41 | | - |
42 | | -_runtime: ModelRuntime | None = None |
43 | | -_admission = AdmissionController(settings.max_concurrent_inferences) |
44 | | -_cache = RecommendationCache(settings.cache_ttl_seconds) |
45 | | -_rate_limiter = RateLimiter(settings.requests_per_minute) |
46 | | -logger = logging.getLogger(__name__) |
47 | | - |
48 | | - |
49 | | -def init_model( |
50 | | - processor: SequenceProcessor, |
51 | | - model: DeepSequenceModel, |
52 | | - *, |
53 | | - model_version: str, |
54 | | - trained: bool, |
55 | | - popular_items: list[str] | None = None, |
56 | | -) -> None: |
57 | | - global _runtime |
58 | | - _runtime = ModelRuntime( |
59 | | - processor=processor, |
60 | | - model=model, |
61 | | - model_version=model_version, |
62 | | - trained=trained, |
63 | | - popular_items=popular_items or list(processor.export_vocabulary())[: settings.max_top_k], |
64 | | - ) |
65 | | - |
66 | | - |
67 | | -class RecommendRequest(BaseModel): |
68 | | - user_id: str = Field(min_length=1, max_length=128) |
69 | | - item_sequence: list[str] = Field(min_length=1, max_length=500) |
70 | | - top_k: int = Field(default=settings.top_k, ge=1, le=settings.max_top_k) |
71 | | - |
72 | | - |
73 | | -class RecommendResponse(BaseModel): |
74 | | - user_id: str |
75 | | - recommendations: list[str] |
76 | | - latency_ms: float |
77 | | - model_version: str |
78 | | - fallback: bool = False |
79 | | - cache_hit: bool = False |
80 | | - |
81 | | - |
82 | | -class FeedbackRequest(BaseModel): |
83 | | - impression_id: str = Field(min_length=1, max_length=128) |
84 | | - user_id: str = Field(min_length=1, max_length=128) |
85 | | - item_id: str = Field(min_length=1, max_length=256) |
86 | | - event_type: Literal["impression", "click", "skip", "cart", "purchase", "dislike"] |
87 | | - position: int | None = Field(default=None, ge=0, le=10_000) |
88 | | - model_version: str = Field(min_length=1, max_length=128) |
89 | | - |
90 | | - |
91 | | -def _authorize(api_key: str | None) -> None: |
92 | | - if not api_key_is_valid(api_key, settings.api_key): |
93 | | - raise HTTPException(status_code=401, detail="Invalid API key") |
94 | | - |
95 | | - |
96 | | -def _response( |
97 | | - request: RecommendRequest, |
98 | | - recommendations: list[str], |
99 | | - started: float, |
100 | | - *, |
101 | | - fallback: bool = False, |
102 | | - cache_hit: bool = False, |
103 | | -) -> RecommendResponse: |
104 | | - assert _runtime is not None |
105 | | - return RecommendResponse( |
106 | | - user_id=request.user_id, |
107 | | - recommendations=recommendations, |
108 | | - latency_ms=(time.perf_counter() - started) * 1_000, |
109 | | - model_version=_runtime.model_version, |
110 | | - fallback=fallback, |
111 | | - cache_hit=cache_hit, |
112 | | - ) |
113 | | - |
114 | | - |
115 | | -@router.post("/", response_model=RecommendResponse, summary="Generate recommendations") |
116 | | -def recommend( |
117 | | - req: RecommendRequest, x_api_key: str | None = Header(default=None) |
118 | | -) -> RecommendResponse: |
119 | | - _authorize(x_api_key) |
120 | | - if not _rate_limiter.allow(req.user_id): |
121 | | - raise HTTPException(status_code=429, detail="Recommendation rate limit exceeded") |
122 | | - if _runtime is None: |
123 | | - raise HTTPException(status_code=503, detail="Model not initialised") |
124 | | - if req.top_k > _runtime.processor.vocab_size: |
125 | | - raise HTTPException(status_code=422, detail="top_k exceeds catalogue size") |
126 | | - |
127 | | - known_items = [ |
128 | | - item for item in req.item_sequence if _runtime.processor.item_to_idx(item) != 0 |
129 | | - ] |
130 | | - if not known_items: |
131 | | - raise HTTPException(status_code=422, detail="Sequence contains no known catalogue items") |
132 | | - remaining_items = _runtime.processor.vocab_size - len(set(known_items)) |
133 | | - if req.top_k > remaining_items: |
134 | | - raise HTTPException(status_code=422, detail="top_k exceeds remaining eligible items") |
135 | | - |
136 | | - started = time.perf_counter() |
137 | | - cache_key = _cache.key(_runtime.model_version, known_items, req.top_k) |
138 | | - cached = _cache.get(cache_key) |
139 | | - if cached is not None: |
140 | | - cache_hits_total.inc() |
141 | | - recommendations_total.labels(status="cache_hit").inc() |
142 | | - return _response(req, cached, started, cache_hit=True) |
143 | | - cache_misses_total.inc() |
144 | | - |
145 | | - if not _admission.acquire(): |
146 | | - recommendations_total.labels(status="fallback_overload").inc() |
147 | | - return _response(req, _runtime.popular_items[: req.top_k], started, fallback=True) |
148 | | - |
149 | | - active_requests.inc() |
150 | | - try: |
151 | | - tensor = _runtime.processor.to_tensor(known_items) |
152 | | - infer_started = time.perf_counter() |
153 | | - indices = _runtime.model.recommend( |
154 | | - tensor, |
155 | | - top_k=req.top_k, |
156 | | - exclude_ids=[_runtime.processor.item_to_idx(item) for item in known_items], |
157 | | - ) |
158 | | - inference_ms = (time.perf_counter() - infer_started) * 1_000 |
159 | | - model_inference_latency.observe(inference_ms / 1_000) |
160 | | - if inference_ms > settings.max_inference_ms: |
161 | | - recommendations_total.labels(status="fallback_latency").inc() |
162 | | - return _response(req, _runtime.popular_items[: req.top_k], started, fallback=True) |
163 | | - |
164 | | - decoded = _runtime.processor.decode_recommendations(indices) |
165 | | - recommendations = [item for item in decoded if item is not None] |
166 | | - _cache.put(cache_key, recommendations) |
167 | | - recommendations_total.labels(status="success").inc() |
168 | | - return _response(req, recommendations, started) |
169 | | - except (ValueError, RuntimeError, KeyError) as exc: |
170 | | - recommendations_total.labels(status="error").inc() |
171 | | - raise HTTPException(status_code=500, detail="Recommendation inference failed") from exc |
172 | | - finally: |
173 | | - recommendation_latency.observe(time.perf_counter() - started) |
174 | | - active_requests.dec() |
175 | | - _admission.release() |
176 | | - |
177 | | - |
178 | | -@router.post("/feedback", status_code=202, summary="Capture recommendation feedback") |
179 | | -def feedback(req: FeedbackRequest, x_api_key: str | None = Header(default=None)) -> dict: |
180 | | - """Emit a privacy-minimized event for collection by the platform log pipeline.""" |
181 | | - _authorize(x_api_key) |
182 | | - anonymized_user = hashlib.sha256(req.user_id.encode()).hexdigest()[:16] |
183 | | - event = req.model_dump(exclude={"user_id"}) | {"anonymous_user_id": anonymized_user} |
184 | | - logger.info("recommendation_feedback=%s", json.dumps(event, sort_keys=True)) |
185 | | - feedback_events_total.labels(event_type=req.event_type).inc() |
186 | | - return {"accepted": True, "impression_id": req.impression_id} |
187 | | - |
188 | | - |
189 | | -@router.get("/health", summary="Model readiness check") |
190 | | -def health() -> dict: |
191 | | - return { |
192 | | - "status": "ready" if _runtime is not None else "not_ready", |
193 | | - "model_loaded": _runtime is not None, |
194 | | - "trained_model": _runtime.trained if _runtime else False, |
195 | | - "model_version": _runtime.model_version if _runtime else None, |
196 | | - "vocab_size": _runtime.processor.vocab_size if _runtime else 0, |
197 | | - } |
| 1 | +"""Validated recommendation routes with bounded inference and fallback behavior.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +import hashlib |
| 6 | +import json |
| 7 | +import logging |
| 8 | +import time |
| 9 | +from dataclasses import dataclass |
| 10 | +from typing import Literal |
| 11 | + |
| 12 | +from fastapi import APIRouter, Header, HTTPException |
| 13 | +from pydantic import BaseModel, Field |
| 14 | + |
| 15 | +from app.core.config import settings |
| 16 | +from app.core.data_processor import SequenceProcessor |
| 17 | +from app.core.metrics import ( |
| 18 | + active_requests, |
| 19 | + cache_hits_total, |
| 20 | + cache_misses_total, |
| 21 | + feedback_events_total, |
| 22 | + model_inference_latency, |
| 23 | + recommendation_latency, |
| 24 | + recommendations_total, |
| 25 | +) |
| 26 | +from app.core.model import DeepSequenceModel |
| 27 | +from app.core.security import api_key_is_valid |
| 28 | +from app.core.serving import AdmissionController, RateLimiter, RecommendationCache |
| 29 | + |
| 30 | +router = APIRouter(prefix="/recommendations", tags=["recommendations"]) |
| 31 | + |
| 32 | + |
| 33 | +@dataclass |
| 34 | +class ModelRuntime: |
| 35 | + processor: SequenceProcessor |
| 36 | + model: DeepSequenceModel |
| 37 | + model_version: str |
| 38 | + trained: bool |
| 39 | + popular_items: list[str] |
| 40 | + |
| 41 | + |
| 42 | +_runtime: ModelRuntime | None = None |
| 43 | +_admission = AdmissionController(settings.max_concurrent_inferences) |
| 44 | +_cache = RecommendationCache(settings.cache_ttl_seconds) |
| 45 | +_rate_limiter = RateLimiter(settings.requests_per_minute) |
| 46 | +logger = logging.getLogger(__name__) |
| 47 | + |
| 48 | + |
| 49 | +def init_model( |
| 50 | + processor: SequenceProcessor, |
| 51 | + model: DeepSequenceModel, |
| 52 | + *, |
| 53 | + model_version: str, |
| 54 | + trained: bool, |
| 55 | + popular_items: list[str] | None = None, |
| 56 | +) -> None: |
| 57 | + global _runtime |
| 58 | + _runtime = ModelRuntime( |
| 59 | + processor=processor, |
| 60 | + model=model, |
| 61 | + model_version=model_version, |
| 62 | + trained=trained, |
| 63 | + popular_items=popular_items or list(processor.export_vocabulary())[: settings.max_top_k], |
| 64 | + ) |
| 65 | + |
| 66 | + |
| 67 | +class RecommendRequest(BaseModel): |
| 68 | + user_id: str = Field(min_length=1, max_length=128) |
| 69 | + item_sequence: list[str] = Field(min_length=1, max_length=500) |
| 70 | + top_k: int = Field(default=settings.top_k, ge=1, le=settings.max_top_k) |
| 71 | + |
| 72 | + |
| 73 | +class RecommendResponse(BaseModel): |
| 74 | + user_id: str |
| 75 | + recommendations: list[str] |
| 76 | + latency_ms: float |
| 77 | + model_version: str |
| 78 | + fallback: bool = False |
| 79 | + cache_hit: bool = False |
| 80 | + |
| 81 | + |
| 82 | +class FeedbackRequest(BaseModel): |
| 83 | + impression_id: str = Field(min_length=1, max_length=128) |
| 84 | + user_id: str = Field(min_length=1, max_length=128) |
| 85 | + item_id: str = Field(min_length=1, max_length=256) |
| 86 | + event_type: Literal["impression", "click", "skip", "cart", "purchase", "dislike"] |
| 87 | + position: int | None = Field(default=None, ge=0, le=10_000) |
| 88 | + model_version: str = Field(min_length=1, max_length=128) |
| 89 | + |
| 90 | + |
| 91 | +def _authorize(api_key: str | None) -> None: |
| 92 | + if not api_key_is_valid(api_key, settings.api_key): |
| 93 | + raise HTTPException(status_code=401, detail="Invalid API key") |
| 94 | + |
| 95 | + |
| 96 | +def _response( |
| 97 | + request: RecommendRequest, |
| 98 | + recommendations: list[str], |
| 99 | + started: float, |
| 100 | + *, |
| 101 | + fallback: bool = False, |
| 102 | + cache_hit: bool = False, |
| 103 | +) -> RecommendResponse: |
| 104 | + assert _runtime is not None |
| 105 | + return RecommendResponse( |
| 106 | + user_id=request.user_id, |
| 107 | + recommendations=recommendations, |
| 108 | + latency_ms=(time.perf_counter() - started) * 1_000, |
| 109 | + model_version=_runtime.model_version, |
| 110 | + fallback=fallback, |
| 111 | + cache_hit=cache_hit, |
| 112 | + ) |
| 113 | + |
| 114 | + |
| 115 | +@router.post("/", response_model=RecommendResponse, summary="Generate recommendations") |
| 116 | +def recommend( |
| 117 | + req: RecommendRequest, x_api_key: str | None = Header(default=None) |
| 118 | +) -> RecommendResponse: |
| 119 | + _authorize(x_api_key) |
| 120 | + if not _rate_limiter.allow(req.user_id): |
| 121 | + raise HTTPException(status_code=429, detail="Recommendation rate limit exceeded") |
| 122 | + if _runtime is None: |
| 123 | + raise HTTPException(status_code=503, detail="Model not initialised") |
| 124 | + if req.top_k > _runtime.processor.vocab_size: |
| 125 | + raise HTTPException(status_code=422, detail="top_k exceeds catalogue size") |
| 126 | + |
| 127 | + known_items = [item for item in req.item_sequence if _runtime.processor.item_to_idx(item) != 0] |
| 128 | + if not known_items: |
| 129 | + raise HTTPException(status_code=422, detail="Sequence contains no known catalogue items") |
| 130 | + remaining_items = _runtime.processor.vocab_size - len(set(known_items)) |
| 131 | + if req.top_k > remaining_items: |
| 132 | + raise HTTPException(status_code=422, detail="top_k exceeds remaining eligible items") |
| 133 | + |
| 134 | + started = time.perf_counter() |
| 135 | + cache_key = _cache.key(_runtime.model_version, known_items, req.top_k) |
| 136 | + cached = _cache.get(cache_key) |
| 137 | + if cached is not None: |
| 138 | + cache_hits_total.inc() |
| 139 | + recommendations_total.labels(status="cache_hit").inc() |
| 140 | + return _response(req, cached, started, cache_hit=True) |
| 141 | + cache_misses_total.inc() |
| 142 | + |
| 143 | + if not _admission.acquire(): |
| 144 | + recommendations_total.labels(status="fallback_overload").inc() |
| 145 | + return _response(req, _runtime.popular_items[: req.top_k], started, fallback=True) |
| 146 | + |
| 147 | + active_requests.inc() |
| 148 | + try: |
| 149 | + tensor = _runtime.processor.to_tensor(known_items) |
| 150 | + infer_started = time.perf_counter() |
| 151 | + indices = _runtime.model.recommend( |
| 152 | + tensor, |
| 153 | + top_k=req.top_k, |
| 154 | + exclude_ids=[_runtime.processor.item_to_idx(item) for item in known_items], |
| 155 | + ) |
| 156 | + inference_ms = (time.perf_counter() - infer_started) * 1_000 |
| 157 | + model_inference_latency.observe(inference_ms / 1_000) |
| 158 | + if inference_ms > settings.max_inference_ms: |
| 159 | + recommendations_total.labels(status="fallback_latency").inc() |
| 160 | + return _response(req, _runtime.popular_items[: req.top_k], started, fallback=True) |
| 161 | + |
| 162 | + decoded = _runtime.processor.decode_recommendations(indices) |
| 163 | + recommendations = [item for item in decoded if item is not None] |
| 164 | + _cache.put(cache_key, recommendations) |
| 165 | + recommendations_total.labels(status="success").inc() |
| 166 | + return _response(req, recommendations, started) |
| 167 | + except (ValueError, RuntimeError, KeyError) as exc: |
| 168 | + recommendations_total.labels(status="error").inc() |
| 169 | + raise HTTPException(status_code=500, detail="Recommendation inference failed") from exc |
| 170 | + finally: |
| 171 | + recommendation_latency.observe(time.perf_counter() - started) |
| 172 | + active_requests.dec() |
| 173 | + _admission.release() |
| 174 | + |
| 175 | + |
| 176 | +@router.post("/feedback", status_code=202, summary="Capture recommendation feedback") |
| 177 | +def feedback(req: FeedbackRequest, x_api_key: str | None = Header(default=None)) -> dict: |
| 178 | + """Emit a privacy-minimized event for collection by the platform log pipeline.""" |
| 179 | + _authorize(x_api_key) |
| 180 | + anonymized_user = hashlib.sha256(req.user_id.encode()).hexdigest()[:16] |
| 181 | + event = req.model_dump(exclude={"user_id"}) | {"anonymous_user_id": anonymized_user} |
| 182 | + logger.info("recommendation_feedback=%s", json.dumps(event, sort_keys=True)) |
| 183 | + feedback_events_total.labels(event_type=req.event_type).inc() |
| 184 | + return {"accepted": True, "impression_id": req.impression_id} |
| 185 | + |
| 186 | + |
| 187 | +@router.get("/health", summary="Model readiness check") |
| 188 | +def health() -> dict: |
| 189 | + return { |
| 190 | + "status": "ready" if _runtime is not None else "not_ready", |
| 191 | + "model_loaded": _runtime is not None, |
| 192 | + "trained_model": _runtime.trained if _runtime else False, |
| 193 | + "model_version": _runtime.model_version if _runtime else None, |
| 194 | + "vocab_size": _runtime.processor.vocab_size if _runtime else 0, |
| 195 | + } |
0 commit comments