Building an AI Recommendation System with Python

On a health and nutrition product I worked on, the first "recommendation" endpoint was a spreadsheet in code. Mifflin-St Jeor for BMR, an activity multiplier for TDEE, a fixed calorie offset for the goal, then a hand-tuned macro split. Users got a number. They did not get a ranked meal list that respected allergies, cuisine preference, or what they had actually eaten last week.

That rules layer never went away. It became the floor every smarter ranker had to respect. This post is how I grew that FastAPI service from targets-only into a real recommendation path: content-based ranking first, a light collaborative signal second, a tiny feature store for the inputs, and an evaluation loop that caught regressions before users did.

The product API stayed in Node. Python owned scoring. I wrote about that split in Node.js + Python architecture for AI apps. Here the focus is the recommender itself.

Start with rules, then rank inside the budget

Rules answer "how many calories and grams of protein does this person need today?" Ranking answers "which of these candidate meals should we show first?" Mixing those two jobs in one function is how you get a beautiful ranking that quietly violates a medical constraint.

I kept the pipeline as three steps with named versions:

1. Targets (deterministic): BMR → TDEE → calories and macros.
2. Candidates: a short list from the catalog (Postgres on the Node side, or a cached snapshot Python already holds).
3. Score: content-based features first; collaborative boost only when we have enough history.

from pydantic import BaseModel, Field
from typing import Literal

MODEL_VERSION = "recs-content-2026.10"


class Profile(BaseModel):
    user_id: str
    calories: int = Field(gt=800, lt=6000)
    protein_g: int
    allergies: list[str] = []
    cuisine_prefs: list[str] = []


class Meal(BaseModel):
    meal_id: str
    calories: int
    protein_g: int
    tags: list[str]
    cuisine: str


class ScoredMeal(BaseModel):
    meal_id: str
    score: float
    reasons: list[str]
    model_version: str


def hard_filter(profile: Profile, meals: list[Meal]) -> list[Meal]:
    banned = {a.lower() for a in profile.allergies}
    out = []
    for m in meals:
        tags = {t.lower() for t in m.tags}
        if banned & tags:
            continue
        if abs(m.calories - profile.calories / 3) > 250:
            continue
        out.append(m)
    return out


def content_score(profile: Profile, meal: Meal) -> tuple[float, list[str]]:
    reasons: list[str] = []
    score = 0.0

    cal_err = abs(meal.calories - profile.calories / 3) / max(profile.calories / 3, 1)
    score += max(0.0, 1.0 - cal_err)
    reasons.append("calorie_fit")

    if meal.protein_g >= profile.protein_g / 3 * 0.8:
        score += 0.4
        reasons.append("protein_floor")

    if meal.cuisine.lower() in {c.lower() for c in profile.cuisine_prefs}:
        score += 0.3
        reasons.append("cuisine_pref")

    return score, reasons

Hard filters run before any soft score. An allergy match is not a negative weight; it is a drop. Soft scores are for trade-offs. That distinction matters when someone files a support ticket about peanuts.

Content-based first (it works with one user)

Collaborative filtering needs overlap: users who ate meal A also liked meal B. On day one of a nutrition app, almost nobody has that graph. Content-based ranking only needs the meal's own attributes and the user's profile, so it is the right first model.

I treated each meal as a small feature vector: normalized calories distance to the per-meal budget, protein ratio, cuisine one-hots, and a few dietary tags. Scores stayed explainable (reasons in the response) so the product team could debug "why did we show jollof for a low-carb day?" without opening a notebook.

A useful gotcha: do not let the catalog grow unbounded inside the request. Cap candidates (for example top 200 by coarse calorie bucket) before scoring. Ranking 40,000 meals on every open of the home screen will look fine in a load test with warm caches and fail under real traffic.

Collaborative signal as a boost, not the brain

Once we had enough meal-complete events, I added a simple item-item boost: for meals the user finished last week, look up co-completion counts and nudge related candidates. The collaborative term never overrode the hard filters or the calorie budget. It was a multiplier on an already-valid shortlist.

def collab_boost(user_history: list[str], meal_id: str, pair_counts: dict[tuple[str, str], int]) -> float:
    if not user_history:
        return 0.0
    total = 0
    for past in user_history[-10:]:
        a, b = sorted((past, meal_id))
        total += pair_counts.get((a, b), 0)
    # Log-scale so a viral meal does not dominate every list.
    return min(0.5, 0.1 * (total ** 0.5))

Cold start for a new user is still content-based only. Cold start for a new meal is the same: until it appears in completions, it competes on attributes alone. I did not pretend matrix factorization was required on day thirty; pairwise counts in Redis or a small SQL table were enough to learn whether the boost helped.

Feature store-lite (enough to be honest)

You do not need Feast on day one. You do need a single place that defines "what we scored on" and a version string on every response.

What I kept durable:

• Profile features at request time: targets, allergies, prefs (from Node).
• Meal features refreshed on a schedule: calories, macros, tags, cuisine (snapshot JSON or a read replica table Python can query).
• History features: last N completed meal_ids, stored per user with a short TTL cache in Redis.
• model_version on every scored list, persisted next to the impression in Node.

That is a feature store in spirit: named inputs, reproducible scoring, no silent drift when someone edits a meal's tags in the CMS. When tags change mid-day, either version the meal row or accept that yesterday's evaluation set no longer matches today's catalog.

Evaluate like a product, not a Kaggle notebook

Offline metrics that mattered for us:

I kept a JSONL regression set: fixed profiles, fixed candidate snapshots, expected "must include / must exclude" meal ids. A CI job scored the set and failed the build on constraint violations. Ranking quality could move; safety could not.

Online, I logged impressions and completions with model_version. Without that, A/B tests lie: you compare two UIs while three model deploys land in the same week.

Failure modes I hit in practice:

• Feedback loops. Popular meals get more impressions, then more "collaborative" weight. Cap the boost and keep a exploration slot (even one random eligible meal in the list) so new dishes can surface.
• Stale snapshots. Python scored against yesterday's catalog while Node had already unpublished a meal. Prefer candidate ids from Node in the request body; let Python only score what it is given.
• Leaky async. CPU-bound scoring inside async def blocks the event loop. Use plain def handlers in FastAPI so work runs in the thread pool, same as the targets endpoint.

What I would not put in the model service

Auth, rate limits, and "should this user see recommendations today?" stay in Node. Python should not open the product wallet database "just to be helpful." Keep the contract small: profile + candidates in, scored list out, version stamped. That boundary is what let us swap the ranker without shipping a new mobile build.

Takeaways

• Ship rules for budgets, then rank inside the safe set. Hard filters before soft scores.
• Prefer content-based until you have real overlap; add collaborative signal as a capped boost.
• A feature store-lite (named inputs + model_version + snapshot discipline) beats a platform you do not operate yet.
• Evaluate constraints in CI; treat hit rate and diversity as product metrics with versioned logs.
• Keep Python stateless and internal. Node owns the product; Python owns the score. See Node.js + Python architecture for AI apps for the service boundary.