Wrapping the pipeline in FastAPI

The orchestrator we built is a class. The service exposes that class behind an HTTP endpoint. We use FastAPI lifespan to wire up shared state once, lazy-load the models on first call so the container boots fast, and return a typed Pydantic response.

recsys/serving/app.py
python
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pipeline = None
    app.state.store = PostgresStore()
    app.state.vectors = VectorStore()
    app.state.registry = ModelRegistry()
    app.state.llm = LLMRerankerService()
    yield

app = FastAPI(title="Personalized Recommender API", lifespan=lifespan)


def _get_pipeline() -> RecommenderPipeline:
    if app.state.pipeline is None:
        app.state.pipeline = RecommenderPipeline(
            store=app.state.store,
            vectors=app.state.vectors,
            registry=app.state.registry,
            llm=app.state.llm,
        )
    return app.state.pipeline


@app.post("/recommend", response_model=RecommendResponse)
def recommend(req: RecommendRequest) -> RecommendResponse:
    pipeline = _get_pipeline()
    result = pipeline.recommend(
        customer_id=req.customer_id,
        top_k=req.top_k,
        transaction_date=req.transaction_date,
        use_llm=req.use_llm,
    )
    return RecommendResponse(**result)

The FastAPI app. Lifespan creates the shared state. _get_pipeline lazy-loads on first /recommend call.

recsys/serving/schemas.py
python
class RecommendRequest(BaseModel):
    customer_id: str = Field(..., description="H&M customer identifier")
    top_k: int = Field(12, ge=1, le=100, description="Number of items to return")
    transaction_date: Optional[str] = None
    use_llm: bool = False


class RecommendedItem(BaseModel):
    article_id: str
    score: float
    llm_score: Optional[float] = None
    prod_name: Optional[str] = None
    product_type_name: Optional[str] = None
    product_group_name: Optional[str] = None
    image_url: Optional[str] = None


class RecommendResponse(BaseModel):
    customer_id: str
    top_k: int
    use_llm: bool
    items: list[RecommendedItem]

Pydantic schemas keep the contract tight and give us free OpenAPI docs.

terminal
bash
make serve
curl -X POST http://localhost:8000/recommend \
  -H "Content-Type: application/json" \
  -d '{"customer_id":"c001000000","top_k":12}' | jq .

Start the API. The OpenAPI docs page lives at /docs.

Quiz: Quiz

Loading practice…