Build a Catalog-Safe LLM Reranker

Build a catalog-safe LLM reranker and compare offline ranking metrics.

Introduction

30 Second Summary

A streaming app can suggest the right kind of show yet fail to identify the exact title in its own library. Every downstream screen needs a stable item to display.

In this project, you will build a catalog-constrained LLM reranker using Python in Google Colab. The finished script turns viewing context into valid candidate rankings with measurable offline results.

What You'll Build

Run your finished script to watch a free-text suggestion give way to a complete catalog ranking plus an offline metrics comparison.

By the end of this project, you'll have:

  • A visible integration test that shows why an appealing free-text title still fails when it has no stable catalog key.
  • A catalog-safe ranking path that returns every supplied candidate ID exactly once. Deterministic repair fills omissions while removing duplicates.
  • An offline comparison across popularity, short-context, and rich-context rankings. The report measures each path with MRR plus Hit Rate at 3 on held-out synthetic labels.
  • Secret Mission: add context compaction that keeps only the two highest-signal history events before reranking.

Are there any prerequisites?

You need experience with recommender systems plus Python. You also need a Google account with access to Google Colab.

The project uses Google Colab's free runtime. Accelerator availability is not guaranteed.

Before We Start

Before any hands-on work, commit to an honest GenRec-inspired prototype with popularity retrieval, LLM reranking, and deterministic catalog repair. Its synthetic offline MRR and Hit Rate at 3 measure this small pipeline's behavior only, so they cannot prove Netflix production lift, proprietary training results, or label efficiency.

Set Up the Reproducible Colab Runtime

The mini-ranker depends on repeatable package behavior. A pinned notebook keeps environment drift from muddying later ranking results.

You'll use Google Colab as the hosted notebook environment. You'll pin its runtime before installing Hugging Face Transformers.

In this step, get ready to:
  • Prepare a new Google Colab notebook on runtime version 2026.07.
  • Install Transformers version 5.19.0 before importing it.
  • Verify the Python runtime plus the package versions plus the selected compute device.
Pin a new notebook

The notebook lives in Google Drive. Its code survives after Colab removes the temporary virtual machine.

  • Go to Google Colab in your browser.
  • Complete the Google account sign-in prompt if Colab displays one.
  • Create a new notebook from the Colab start screen.
  • Select Runtime from the top menu.
  • Select Change runtime type.
  • Open the Runtime Version dropdown.

Free Runtime Capacity Varies

Google Colab is free to use. Its managed resources remain dynamic.

An accelerator may be unavailable during your session. The project selects CPU whenever CUDA is unavailable.

  • Match the runtime dialog to the scenario below.

✔️ I see the pinned runtime

The Runtime Version field already shows 2026.07.

  • Leave Hardware accelerator at its available setting.
  • Apply the runtime configuration with the dialog's confirmation button.

ⓧ I see an older runtime

An older runtime can provide different package versions. Select the pinned runtime before installing anything.

  • Open the Runtime Version dropdown.
  • Select 2026.07.
  • Leave Hardware accelerator at its available setting.
  • Apply the runtime configuration with the dialog's confirmation button.

ⓧ The pinned runtime is missing

The project relies on the pinned runtime for its expected versions. A different runtime would weaken the reproducibility check.

  • Reload the notebook page.
  • Return to Change runtime type from the Runtime menu.
  • Check the Runtime Version dropdown for 2026.07 again.
  • Pause the project if the pinned runtime remains unavailable.
  • Wait for the notebook to show that its latest changes are saved in Google Drive.

Good, your notebook now has a fixed runtime base. Later metric changes can be traced to ranking logic with more confidence.

Prepare the package layer

The notebook needs the pinned Transformers package before any Transformers import runs. This order prevents the preinstalled package from defining the session's behavior.

  • Install the pinned package in the first notebook cell by running this command:
!pip install transformers==5.19.0

What Does This Command Do?

  • The notebook passes the installation request to pip.
  • The transformers==5.19.0 pin gives later imports the package version expected by this project.
  • Wait for the installation cell to finish.
  • Confirm the cell stops running without a traceback.
  • Wait for the notebook to finish saving the completed cell.

The package installation is complete. Transformers is ready for the version check.

Package Installation Did Not Finish?

  • Confirm the cell starts with the exact installation command shown above.
  • Rerun the cell once if a temporary network or runtime issue interrupted the download.
  • Return to the runtime selector if Colab disconnected the notebook.
  • Help me diagnose my Colab package installation.
Verify the runtime

The final cell checks Python plus PyTorch plus Transformers. It also uses CUDA availability to select the compute device.

  • Predict whether this free session selects cpu or cuda.
  • Check the pinned versions plus the available device by running this cell:
import sys

import torch
import transformers

# Select CUDA when Colab exposes it, with CPU as the fallback
device = "cuda" if torch.cuda.is_available() else "cpu"

# Print the runtime facts that anchor later ranking comparisons
print(f"Python: {sys.version}")
print(f"PyTorch: {torch.__version__}")
print(f"Transformers: {transformers.__version__}")
print(f"torch.cuda.is_available(): {torch.cuda.is_available()}")
print(f"Selected device: {device}")

What Does This Code Check?

  • The sys.version value identifies the active Python runtime.
  • The torch.__version__ value identifies the active PyTorch package.
  • The transformers.__version__ value confirms that the pinned installation is active.
  • The torch.cuda.is_available() call reports whether CUDA is available to this session.
  • The device value gives the rest of the project one valid compute target.
  • Confirm the Python: line begins with 3.12.13.
  • Confirm the PyTorch: line begins with 2.11.0.
  • Confirm the Transformers: line shows 5.19.0.
  • Confirm the torch.cuda.is_available(): line shows a Boolean result.
  • Confirm the Selected device: line ends with cpu or cuda.

That's the environment locked down. Your notebook now exposes the exact runtime facts behind every later ranking run.

Seeing Different Versions?

  • Return to Change runtime type from the Runtime menu.
  • Confirm the Runtime Version field shows 2026.07.
  • Rerun the Transformers installation cell before rerunning the version-check cell.
  • Help me fix my Colab runtime version check.

✔️ Awesome, I've got everything!

Your notebook is saved with the pinned package cell plus the successful runtime check.

  • Confirm the latest notebook changes have finished saving to Google Drive.

ⓧ I'd like to double check the full code

Your notebook should contain these two cells in this order.

  • Compare the first cell with this installation reference:
!pip install transformers==5.19.0

First Cell Purpose

This cell installs the pinned Transformers package before any Transformers import runs.

  • Compare the second cell with this runtime-check reference:
import sys

import torch
import transformers

# Select CUDA when Colab exposes it, with CPU as the fallback
device = "cuda" if torch.cuda.is_available() else "cpu"

# Print the runtime facts that anchor later ranking comparisons
print(f"Python: {sys.version}")
print(f"PyTorch: {torch.__version__}")
print(f"Transformers: {transformers.__version__}")
print(f"torch.cuda.is_available(): {torch.cuda.is_available()}")
print(f"Selected device: {device}")

Second Cell Purpose

This cell confirms the runtime versions plus the compute device used by the mini-ranker.

Your reproducible Colab runtime is ready. Next, you'll generate a recommendation that exposes the missing catalog-key problem.

Generate a Naive Recommendation

Your Google Colab runtime is pinned. Now you can test a recommendation idea without package drift changing the result.

This step asks Qwen for one title from viewing history alone. The result will show whether a readable answer is enough for a downstream catalog pipeline.

In this step, get ready to:
  • Create a synthetic catalog with three viewer histories.
  • Load the instruction-tuned model with a reusable generation helper.
  • Generate one free-text recommendation for U01.
Create the synthetic catalog

A repeatable recommendation test needs fixed items with fixed viewer histories. The synthetic data gives every later ranking path the same starting point.

How do these notebook cells create a file?

The %%writefile cell magic writes the rest of a notebook cell into a file. The -a option appends later chunks without replacing earlier content.

  • Create genrec_mini_ranker.py with the imports and first four catalog records by running this cell:
%%writefile genrec_mini_ranker.py
import re

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

MODEL_ID = "Qwen/Qwen2.5-0.5B-Instruct"

CATALOG = {
    "T01": {
        "title": "City of Embers",
        "genres": "science fiction, mystery",
        "description": "A transit engineer uncovers a conspiracy beneath a neon megacity.",
    },
    "T02": {
        "title": "Second Serve",
        "genres": "sports, comedy",
        "description": "A retired tennis player coaches an unlikely neighborhood team.",
    },
    "T03": {
        "title": "Quiet Orbit",
        "genres": "science fiction, drama",
        "description": "Two astronauts repair a silent research station while confronting an old loss.",
    },
    "T04": {
        "title": "Kitchen Rivals",
        "genres": "reality, food",
        "description": "Home cooks race through playful regional cooking challenges.",
    },

What does this code do?

  • The imports provide regular-expression support plus the model classes used later in the script.
  • The MODEL_ID value identifies the public instruction-tuned model that the loader retrieves.
  • Each CATALOG entry has a stable title ID with metadata that can be passed into a ranking prompt.
  • Check the cell output for confirmation that genrec_mini_ranker.py was written.

File not created?

  • Make sure %%writefile genrec_mini_ranker.py is the first line in the cell.
  • Check that the cell ran after your pinned runtime connected.
  • Ask for help with creating the Python file from the notebook cell.
  • Append the next five catalog records by running this cell:
%%writefile -a genrec_mini_ranker.py
    "T05": {
        "title": "Shadow Protocol",
        "genres": "spy, thriller",
        "description": "An analyst follows a digital trail through a tense international operation.",
    },
    "T06": {
        "title": "Wild Coasts",
        "genres": "nature, documentary",
        "description": "A calm journey through wildlife communities along remote coastlines.",
    },
    "T07": {
        "title": "Maple Street",
        "genres": "family, comedy",
        "description": "Three neighboring families turn ordinary weekends into comic disasters.",
    },
    "T08": {
        "title": "Ancient Machines",
        "genres": "history, documentary, mystery",
        "description": "Engineers reconstruct ingenious devices from lost civilizations.",
    },
    "T09": {
        "title": "Weekend Detectives",
        "genres": "comedy, mystery",
        "description": "Two siblings investigate harmless mysteries around their small town.",
    },

What does this catalog chunk add?

These records expand the available genres without changing the structure of the catalog. Stable IDs such as T05 and T09 remain separate from their display titles.

  • Check the cell output for confirmation that the catalog records were appended.

Earlier catalog records disappeared?

  • Confirm the first line includes -a so the cell appends to the existing file.
  • Rerun the first catalog cell if this cell replaced the file.
  • Ask for help with restoring the appended catalog chunks.
  • Finish the catalog with T10 by running this cell:
%%writefile -a genrec_mini_ranker.py
    "T10": {
        "title": "Cosmic Frontiers",
        "genres": "space, documentary",
        "description": "Scientists explain new discoveries through vivid journeys across the solar system.",
    },
}

USERS = [
    {
        "user_id": "U01",
        "history": [
            {"title_id": "T01", "completion": 98, "liked": True, "days_ago": 2},
            {"title_id": "T03", "completion": 94, "liked": True, "days_ago": 7},
            {"title_id": "T05", "completion": 31, "liked": False, "days_ago": 1},
        ],
        "request_context": "evening, television, wants something thoughtful",
        "candidates": ["T02", "T08", "T05", "T06", "T09"],
        "target": "T08",
    },

What does the first viewer contain?

  • The closing brace completes the ten-item CATALOG dictionary.
  • The U01 history records completion percentage with explicit feedback.
  • The candidate list represents the popularity-ordered items available to a later reranking stage.
  • Check the cell output for confirmation that the first viewer was appended.

First viewer missing from the file?

  • Check that this cell ran after both earlier catalog cells.
  • Confirm the cell starts with %%writefile -a genrec_mini_ranker.py.
  • Ask for help with rebuilding the catalog and first viewer in order.
  • Append the remaining viewer records by running this cell:
%%writefile -a genrec_mini_ranker.py
    {
        "user_id": "U02",
        "history": [
            {"title_id": "T02", "completion": 100, "liked": True, "days_ago": 3},
            {"title_id": "T07", "completion": 91, "liked": True, "days_ago": 6},
            {"title_id": "T04", "completion": 84, "liked": False, "days_ago": 4},
        ],
        "request_context": "weekend afternoon, tablet, wants something light",
        "candidates": ["T05", "T01", "T09", "T08", "T03"],
        "target": "T09",
    },
    {
        "user_id": "U03",
        "history": [
            {"title_id": "T06", "completion": 97, "liked": True, "days_ago": 1},
            {"title_id": "T08", "completion": 88, "liked": True, "days_ago": 9},
            {"title_id": "T03", "completion": 52, "liked": False, "days_ago": 12},
        ],
        "request_context": "late evening, laptop, prefers calm factual stories",
        "candidates": ["T02", "T07", "T05", "T10", "T09"],
        "target": "T10",
    },
]

Why include three viewers?

The three histories cover different engagement patterns with different next-watch targets. This small set supports repeatable pipeline checks later in the project.

  • Check the cell output for confirmation that the remaining viewers were appended.

User list looks incomplete?

  • Confirm this cell ends with the closing bracket for USERS.
  • Check that U02 appears before U03.
  • Ask for help with finding a missing bracket in the synthetic user data.
Add model generation and the naive prompt

The script needs one boundary for loading the model. It also needs one helper that turns a chat prompt into generated text.

  • Append the system message with the model loader by running this cell:
%%writefile -a genrec_mini_ranker.py

SYSTEM_MESSAGE = (
    "You are a recommendation ranking service. Follow the requested output format exactly."
)


def load_model():
    device = "cuda" if torch.cuda.is_available() else "cpu"
    tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
    model = AutoModelForCausalLM.from_pretrained(MODEL_ID).to(device)
    return tokenizer, model, device

What does the loader do?

  • The SYSTEM_MESSAGE gives the model a consistent recommendation-service role.
  • The load_model() function selects cuda when acceleration is available.
  • The tokenizer converts prompt text into model inputs.
  • The model moves onto the selected PyTorch device.
  • Check the cell output for confirmation that the loader was appended.

Loader chunk not appended?

  • Confirm the cell uses the same genrec_mini_ranker.py filename as the catalog cells.
  • Check that the installed Transformers package remains available in the connected runtime.
  • Ask for help with adding the model loader to the existing file.
  • Append the reusable text-generation helper by running this cell:
%%writefile -a genrec_mini_ranker.py


def generate_text(prompt, tokenizer, model, device, max_new_tokens=64):
    messages = [
        {"role": "system", "content": SYSTEM_MESSAGE},
        {"role": "user", "content": prompt},
    ]
    text = tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True,
    )
    model_inputs = tokenizer([text], return_tensors="pt").to(device)
    generated_ids = model.generate(
        **model_inputs,
        max_new_tokens=max_new_tokens,
        do_sample=False,
    )
    generated_only = [
        output_ids[len(input_ids) :]
        for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
    ]
    return tokenizer.batch_decode(generated_only, skip_special_tokens=True)[0].strip()

How does generation work?

  • The helper combines the system instruction with the viewer-specific prompt.
  • The chat template formats both messages for the selected model.
  • The tokenizer returns tensors on the selected device.
  • The do_sample=False setting uses greedy decoding to reduce output variability.
  • The final line decodes only the newly generated tokens into text.
  • Check the cell output for confirmation that generate_text() was appended.

Generation helper looks truncated?

  • Confirm the function ends with the batch_decode return line.
  • Check that each continuation line keeps the indentation shown in the cell.
  • Ask for help with repairing the generation helper indentation.
  • Append the title lookup with the naive recommendation function by running this cell:
%%writefile -a genrec_mini_ranker.py


def title_name(title_id):
    return CATALOG[title_id]["title"]


def naive_recommend(user, tokenizer, model, device):
    watched = ", ".join(title_name(event["title_id"]) for event in user["history"])
    prompt = (
        f"The viewer watched: {watched}. Recommend one title they may enjoy next. "
        "Respond with a title name only. Do not use catalog IDs."
    )
    return generate_text(prompt, tokenizer, model, device, max_new_tokens=24)

What makes this recommendation naive?

  • The title_name() helper converts each watched ID into its display title.
  • The naive_recommend() function joins those display titles into readable history text.
  • The prompt requests one title name without supplying candidate IDs.
  • The shorter token limit keeps the answer focused on a single recommendation.
  • Check the cell output for confirmation that the naive function was appended.

Naive prompt missing text?

  • Confirm both prompt strings remain inside the same parentheses.
  • Check that watched appears inside the formatted first string.
  • Ask for help with fixing the naive recommendation prompt.
  • Make the partial script runnable by appending its entry point with this cell:
%%writefile -a genrec_mini_ranker.py


def main():
    tokenizer, model, device = load_model()
    print(f"Model ready on: {device}")

    naive_output = naive_recommend(USERS[0], tokenizer, model, device)
    print(f"\nNaive recommendation: {naive_output}")
    print("Stable catalog key: None")


if __name__ == "__main__":
    main()

What does the entry point do?

  • The main() function loads the tokenizer with the model.
  • The function passes USERS[0] into the naive path.
  • The final print statements expose the generated title with the catalog-key field used by this experiment.
  • The name guard runs main() when you execute the file as a script.
  • Check the cell output for confirmation that the runnable entry point was appended.

Entry point missing?

  • Confirm main() appears below naive_recommend().
  • Check that the final line calls main() with four spaces of indentation.
  • Ask for help with checking the completed partial script.

✔️ Awesome, I've got everything!

Great. Your temporary-runtime file now contains the complete naive recommendation path.

ⓧ I'd like to double check the full code

  • Compare your genrec_mini_ranker.py file with this cumulative version:
import re

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

MODEL_ID = "Qwen/Qwen2.5-0.5B-Instruct"

CATALOG = {
    "T01": {
        "title": "City of Embers",
        "genres": "science fiction, mystery",
        "description": "A transit engineer uncovers a conspiracy beneath a neon megacity.",
    },
    "T02": {
        "title": "Second Serve",
        "genres": "sports, comedy",
        "description": "A retired tennis player coaches an unlikely neighborhood team.",
    },
    "T03": {
        "title": "Quiet Orbit",
        "genres": "science fiction, drama",
        "description": "Two astronauts repair a silent research station while confronting an old loss.",
    },
    "T04": {
        "title": "Kitchen Rivals",
        "genres": "reality, food",
        "description": "Home cooks race through playful regional cooking challenges.",
    },
    "T05": {
        "title": "Shadow Protocol",
        "genres": "spy, thriller",
        "description": "An analyst follows a digital trail through a tense international operation.",
    },
    "T06": {
        "title": "Wild Coasts",
        "genres": "nature, documentary",
        "description": "A calm journey through wildlife communities along remote coastlines.",
    },
    "T07": {
        "title": "Maple Street",
        "genres": "family, comedy",
        "description": "Three neighboring families turn ordinary weekends into comic disasters.",
    },
    "T08": {
        "title": "Ancient Machines",
        "genres": "history, documentary, mystery",
        "description": "Engineers reconstruct ingenious devices from lost civilizations.",
    },
    "T09": {
        "title": "Weekend Detectives",
        "genres": "comedy, mystery",
        "description": "Two siblings investigate harmless mysteries around their small town.",
    },
    "T10": {
        "title": "Cosmic Frontiers",
        "genres": "space, documentary",
        "description": "Scientists explain new discoveries through vivid journeys across the solar system.",
    },
}

USERS = [
    {
        "user_id": "U01",
        "history": [
            {"title_id": "T01", "completion": 98, "liked": True, "days_ago": 2},
            {"title_id": "T03", "completion": 94, "liked": True, "days_ago": 7},
            {"title_id": "T05", "completion": 31, "liked": False, "days_ago": 1},
        ],
        "request_context": "evening, television, wants something thoughtful",
        "candidates": ["T02", "T08", "T05", "T06", "T09"],
        "target": "T08",
    },
    {
        "user_id": "U02",
        "history": [
            {"title_id": "T02", "completion": 100, "liked": True, "days_ago": 3},
            {"title_id": "T07", "completion": 91, "liked": True, "days_ago": 6},
            {"title_id": "T04", "completion": 84, "liked": False, "days_ago": 4},
        ],
        "request_context": "weekend afternoon, tablet, wants something light",
        "candidates": ["T05", "T01", "T09", "T08", "T03"],
        "target": "T09",
    },
    {
        "user_id": "U03",
        "history": [
            {"title_id": "T06", "completion": 97, "liked": True, "days_ago": 1},
            {"title_id": "T08", "completion": 88, "liked": True, "days_ago": 9},
            {"title_id": "T03", "completion": 52, "liked": False, "days_ago": 12},
        ],
        "request_context": "late evening, laptop, prefers calm factual stories",
        "candidates": ["T02", "T07", "T05", "T10", "T09"],
        "target": "T10",
    },
]

SYSTEM_MESSAGE = (
    "You are a recommendation ranking service. Follow the requested output format exactly."
)


def load_model():
    device = "cuda" if torch.cuda.is_available() else "cpu"
    tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
    model = AutoModelForCausalLM.from_pretrained(MODEL_ID).to(device)
    return tokenizer, model, device


def generate_text(prompt, tokenizer, model, device, max_new_tokens=64):
    messages = [
        {"role": "system", "content": SYSTEM_MESSAGE},
        {"role": "user", "content": prompt},
    ]
    text = tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True,
    )
    model_inputs = tokenizer([text], return_tensors="pt").to(device)
    generated_ids = model.generate(
        **model_inputs,
        max_new_tokens=max_new_tokens,
        do_sample=False,
    )
    generated_only = [
        output_ids[len(input_ids) :]
        for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
    ]
    return tokenizer.batch_decode(generated_only, skip_special_tokens=True)[0].strip()


def title_name(title_id):
    return CATALOG[title_id]["title"]


def naive_recommend(user, tokenizer, model, device):
    watched = ", ".join(title_name(event["title_id"]) for event in user["history"])
    prompt = (
        f"The viewer watched: {watched}. Recommend one title they may enjoy next. "
        "Respond with a title name only. Do not use catalog IDs."
    )
    return generate_text(prompt, tokenizer, model, device, max_new_tokens=24)


def main():
    tokenizer, model, device = load_model()
    print(f"Model ready on: {device}")

    naive_output = naive_recommend(USERS[0], tokenizer, model, device)
    print(f"\nNaive recommendation: {naive_output}")
    print("Stable catalog key: None")


if __name__ == "__main__":
    main()
Run the naive path

The first execution downloads public model artifacts of approximately 1,000 MB. Expect it to take longer than later runs while those files enter the temporary runtime cache.

Before you run the script, do you think a generated title alone gives the pipeline a stable catalog key?

  • Run genrec_mini_ranker.py from a new notebook code cell with this command:
!python genrec_mini_ranker.py

You should see the selected cpu or cuda device. The console then prints a free-text value after Naive recommendation:.

The next line reads Stable catalog key: None. This is the intended shortfall because the generated title string has no dependable identifier for a catalog join.

The missing key is the lesson

A plausible title can look useful to a person while remaining unsafe for a downstream system. The pipeline needs an identifier from the supplied catalog before it can retrieve metadata or record an impression.

Script did not reach the recommendation?

  • Confirm every file-building cell ran from top to bottom.
  • Check that the runtime still contains the installed transformers==5.19.0 package.
  • Rerun the script after a temporary model-download interruption.
  • Ask for help with debugging the naive recommendation run.

You have a visible recommendation with a concrete integration gap. Next, you will make the ranking safe for the catalog.

Constrain the Model to the Catalog

Your naive path produces a readable recommendation. However, its title string has no stable catalog key that downstream systems can join to inventory.

In this step, you will add a catalog constraint around the model. The model will perform listwise reranking over a supplied candidate set. Deterministic repair will turn every response into a complete catalog-safe ordering.

In this step, get ready to:
  • Convert viewing behavior plus request conditions into readable model context.
  • Extract valid candidate IDs from the model response with deterministic repair.
  • Run the constrained path to produce a complete duplicate-free ranking.
Encode viewer and candidate context

Context engineering turns interaction data into text the model can interpret. The viewer block captures engagement signals. The candidate block exposes only the items that the reranker may return.

  • Return to genrec_mini_ranker.py in the Google Colab file editor.
  • Find the end of naive_recommend().
  • Add build_rich_context() below naive_recommend() by pasting this code:
def build_rich_context(user):
    lines = [f"Request context: {user['request_context']}", "Viewing history:"]
    for event in user["history"]:
        liked = "yes" if event["liked"] else "no"
        lines.append(
            f"- {event['title_id']} {title_name(event['title_id'])} | "
            f"completion {event['completion']}% | liked {liked} | "
            f"{event['days_ago']} days ago"
        )
    return "\n".join(lines)

What does this code do?

  • The first line records the viewing situation stored in request_context.
  • The loop converts each history event into one readable line with its catalog ID.
  • Completion percentage captures depth of engagement. The liked value captures explicit feedback. The recency value records when the interaction happened.
  • The final join returns one multiline context string for the ranking prompt.
  • Save genrec_mini_ranker.py.
  • Search genrec_mini_ranker.py for def build_rich_context(user):.

You should see the new function directly below naive_recommend(). Its last line should return the joined context.

Cannot find the context builder?

  • Check that you pasted the function outside naive_recommend().
  • Confirm that def build_rich_context(user): begins at the left edge of the file.
  • Make sure the two continued formatted strings remain inside lines.append().
  • Help me check the placement of build_rich_context().
  • Add build_candidate_block() below build_rich_context() by pasting this code:
def build_candidate_block(candidate_ids):
    lines = []
    for title_id in candidate_ids:
        item = CATALOG[title_id]
        lines.append(
            f"- {title_id} | {item['title']} | {item['genres']} | {item['description']}"
        )
    return "\n".join(lines)

How does the candidate block work?

  • The function accepts the candidate IDs chosen by the earlier retrieval stage.
  • Each ID looks up its title metadata in CATALOG.
  • Each output line pairs a stable ID with a title. It also includes genres plus a description for ranking context.
  • The model receives a closed candidate set that downstream code already knows.
  • Save genrec_mini_ranker.py.
  • Search genrec_mini_ranker.py for def build_candidate_block(candidate_ids):.

You should see build_candidate_block() below the rich context builder. Its returned text includes only IDs supplied through candidate_ids.

Candidate metadata looks incomplete?

  • Check that item = CATALOG[title_id] remains inside the loop.
  • Confirm that the formatted string includes title_id plus the three catalog fields.
  • Make sure the final return stays outside the loop.
  • Help me debug my candidate block builder.
Validate and repair model output

A strict prompt guides the model toward valid IDs. The parser still treats generated text as untrusted output. It preserves valid IDs in model order. It removes repeats before the fallback fills every gap.

  • Add extract_catalog_ids() below build_candidate_block() by pasting this code:
def extract_catalog_ids(raw_output, candidate_ids):
    matches = re.findall(r"\bT\d{2}\b", raw_output.upper())
    raw_valid = [title_id for title_id in matches if title_id in candidate_ids]
    unique_valid = []
    for title_id in raw_valid:
        if title_id not in unique_valid:
            unique_valid.append(title_id)
    return raw_valid, unique_valid

How does output validation work?

  • The regular expression finds uppercase ID-shaped tokens such as T02.
  • The raw_valid list keeps only IDs that belong to the supplied candidate set.
  • The loop builds unique_valid in the order produced by the model.
  • Returning both lists lets the next function detect omissions plus duplicates.
  • Save genrec_mini_ranker.py.
  • Search genrec_mini_ranker.py for return raw_valid, unique_valid.

You should see the returned pair at the end of extract_catalog_ids(). The deduplication loop should appear immediately above it.

Parser keeps duplicate IDs?

  • Confirm that unique_valid = [] appears before the loop.
  • Check that unique_valid.append(title_id) remains inside the duplicate guard.
  • Make sure the membership filter compares matches against candidate_ids.
  • Help me fix duplicate handling in extract_catalog_ids().
  • Add rank_candidates() below extract_catalog_ids() by pasting this code:
def rank_candidates(user, context, tokenizer, model, device):
    candidate_ids = user["candidates"]
    prompt = f"""Rank the candidate catalog IDs for this viewer.

Viewer context:
{context}

Candidates:
{build_candidate_block(candidate_ids)}

Rules:
1. Use only the supplied candidate IDs.
2. Output every candidate ID exactly once.
3. Rank most likely next watch first.
4. Return one line in this exact format: RANKING: T01,T02,T03
"""
    raw_output = generate_text(prompt, tokenizer, model, device)
    raw_valid, unique_valid = extract_catalog_ids(raw_output, candidate_ids)
    missing = [title_id for title_id in candidate_ids if title_id not in unique_valid]
    ranking = unique_valid + missing
    repaired = len(unique_valid) != len(candidate_ids) or len(raw_valid) != len(unique_valid)
    return {
        "user_id": user["user_id"],
        "target": user["target"],
        "raw_output": raw_output,
        "ranking": ranking,
        "repaired": repaired,
    }

How does deterministic repair work?

  • The prompt contains only the user's supplied candidates. Four rules request a complete one-line ranking.
  • The parser separates all valid occurrences from the deduplicated model order.
  • The missing list follows the original candidate order. That order acts as the popularity fallback.
  • The final ranking keeps every valid model choice first. It appends any missing candidates afterward.
  • The repaired flag becomes true when an ID is omitted. It also becomes true when a valid ID is repeated.
  • Save genrec_mini_ranker.py.
  • Search genrec_mini_ranker.py for ranking = unique_valid + missing.

You should see the repaired ranking assembled before the returned result dictionary. That placement guarantees the caller receives a complete list.

Ranking can still omit an ID?

  • Check that missing iterates over candidate_ids.
  • Confirm that the final list uses unique_valid + missing in that order.
  • Make sure extract_catalog_ids() receives the same candidate_ids used to build the prompt.
  • Help me trace a missing candidate through rank_candidates().
Run the catalog-constrained path

The helpers now form a guarded boundary around generation. The runnable section needs one constrained call so you can inspect the raw response beside the repaired result.

  • Find print("Stable catalog key: None") inside main().
  • Add the constrained ranking block directly below that line by pasting this code:
    constrained = rank_candidates(
        USERS[0],
        build_rich_context(USERS[0]),
        tokenizer,
        model,
        device,
    )
    print(f"\nRaw constrained output: {constrained['raw_output']}")
    print(f"Constrained ranking: {constrained['ranking']}")
    print(f"Output repaired: {constrained['repaired']}")

What does this runnable block show?

  • The call reranks the candidate set belonging to USERS[0] with its rich context.
  • The raw output reveals what the model generated before validation.
  • The constrained output reveals the complete ordering after deterministic repair.
  • The final boolean makes any repair visible during testing.
  • Save genrec_mini_ranker.py.

✔️ Awesome, I've got everything!

Your saved file now contains the context builders, output parser, catalog-safe ranker, and constrained runnable block.

ⓧ I'd like to double check the full code

Compare your cumulative genrec_mini_ranker.py with this version:

import re

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

MODEL_ID = "Qwen/Qwen2.5-0.5B-Instruct"

CATALOG = {
    "T01": {
        "title": "City of Embers",
        "genres": "science fiction, mystery",
        "description": "A transit engineer uncovers a conspiracy beneath a neon megacity.",
    },
    "T02": {
        "title": "Second Serve",
        "genres": "sports, comedy",
        "description": "A retired tennis player coaches an unlikely neighborhood team.",
    },
    "T03": {
        "title": "Quiet Orbit",
        "genres": "science fiction, drama",
        "description": "Two astronauts repair a silent research station while confronting an old loss.",
    },
    "T04": {
        "title": "Kitchen Rivals",
        "genres": "reality, food",
        "description": "Home cooks race through playful regional cooking challenges.",
    },
    "T05": {
        "title": "Shadow Protocol",
        "genres": "spy, thriller",
        "description": "An analyst follows a digital trail through a tense international operation.",
    },
    "T06": {
        "title": "Wild Coasts",
        "genres": "nature, documentary",
        "description": "A calm journey through wildlife communities along remote coastlines.",
    },
    "T07": {
        "title": "Maple Street",
        "genres": "family, comedy",
        "description": "Three neighboring families turn ordinary weekends into comic disasters.",
    },
    "T08": {
        "title": "Ancient Machines",
        "genres": "history, documentary, mystery",
        "description": "Engineers reconstruct ingenious devices from lost civilizations.",
    },
    "T09": {
        "title": "Weekend Detectives",
        "genres": "comedy, mystery",
        "description": "Two siblings investigate harmless mysteries around their small town.",
    },
    "T10": {
        "title": "Cosmic Frontiers",
        "genres": "space, documentary",
        "description": "Scientists explain new discoveries through vivid journeys across the solar system.",
    },
}

USERS = [
    {
        "user_id": "U01",
        "history": [
            {"title_id": "T01", "completion": 98, "liked": True, "days_ago": 2},
            {"title_id": "T03", "completion": 94, "liked": True, "days_ago": 7},
            {"title_id": "T05", "completion": 31, "liked": False, "days_ago": 1},
        ],
        "request_context": "evening, television, wants something thoughtful",
        "candidates": ["T02", "T08", "T05", "T06", "T09"],
        "target": "T08",
    },
    {
        "user_id": "U02",
        "history": [
            {"title_id": "T02", "completion": 100, "liked": True, "days_ago": 3},
            {"title_id": "T07", "completion": 91, "liked": True, "days_ago": 6},
            {"title_id": "T04", "completion": 84, "liked": False, "days_ago": 4},
        ],
        "request_context": "weekend afternoon, tablet, wants something light",
        "candidates": ["T05", "T01", "T09", "T08", "T03"],
        "target": "T09",
    },
    {
        "user_id": "U03",
        "history": [
            {"title_id": "T06", "completion": 97, "liked": True, "days_ago": 1},
            {"title_id": "T08", "completion": 88, "liked": True, "days_ago": 9},
            {"title_id": "T03", "completion": 52, "liked": False, "days_ago": 12},
        ],
        "request_context": "late evening, laptop, prefers calm factual stories",
        "candidates": ["T02", "T07", "T05", "T10", "T09"],
        "target": "T10",
    },
]

SYSTEM_MESSAGE = (
    "You are a recommendation ranking service. Follow the requested output format exactly."
)


def load_model():
    device = "cuda" if torch.cuda.is_available() else "cpu"
    tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
    model = AutoModelForCausalLM.from_pretrained(MODEL_ID).to(device)
    return tokenizer, model, device


def generate_text(prompt, tokenizer, model, device, max_new_tokens=64):
    messages = [
        {"role": "system", "content": SYSTEM_MESSAGE},
        {"role": "user", "content": prompt},
    ]
    text = tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True,
    )
    model_inputs = tokenizer([text], return_tensors="pt").to(device)
    generated_ids = model.generate(
        **model_inputs,
        max_new_tokens=max_new_tokens,
        do_sample=False,
    )
    generated_only = [
        output_ids[len(input_ids) :]
        for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
    ]
    return tokenizer.batch_decode(generated_only, skip_special_tokens=True)[0].strip()


def title_name(title_id):
    return CATALOG[title_id]["title"]


def naive_recommend(user, tokenizer, model, device):
    watched = ", ".join(title_name(event["title_id"]) for event in user["history"])
    prompt = (
        f"The viewer watched: {watched}. Recommend one title they may enjoy next. "
        "Respond with a title name only. Do not use catalog IDs."
    )
    return generate_text(prompt, tokenizer, model, device, max_new_tokens=24)


def build_rich_context(user):
    lines = [f"Request context: {user['request_context']}", "Viewing history:"]
    for event in user["history"]:
        liked = "yes" if event["liked"] else "no"
        lines.append(
            f"- {event['title_id']} {title_name(event['title_id'])} | "
            f"completion {event['completion']}% | liked {liked} | "
            f"{event['days_ago']} days ago"
        )
    return "\n".join(lines)


def build_candidate_block(candidate_ids):
    lines = []
    for title_id in candidate_ids:
        item = CATALOG[title_id]
        lines.append(
            f"- {title_id} | {item['title']} | {item['genres']} | {item['description']}"
        )
    return "\n".join(lines)


def extract_catalog_ids(raw_output, candidate_ids):
    matches = re.findall(r"\bT\d{2}\b", raw_output.upper())
    raw_valid = [title_id for title_id in matches if title_id in candidate_ids]
    unique_valid = []
    for title_id in raw_valid:
        if title_id not in unique_valid:
            unique_valid.append(title_id)
    return raw_valid, unique_valid


def rank_candidates(user, context, tokenizer, model, device):
    candidate_ids = user["candidates"]
    prompt = f"""Rank the candidate catalog IDs for this viewer.

Viewer context:
{context}

Candidates:
{build_candidate_block(candidate_ids)}

Rules:
1. Use only the supplied candidate IDs.
2. Output every candidate ID exactly once.
3. Rank most likely next watch first.
4. Return one line in this exact format: RANKING: T01,T02,T03
"""
    raw_output = generate_text(prompt, tokenizer, model, device)
    raw_valid, unique_valid = extract_catalog_ids(raw_output, candidate_ids)
    missing = [title_id for title_id in candidate_ids if title_id not in unique_valid]
    ranking = unique_valid + missing
    repaired = len(unique_valid) != len(candidate_ids) or len(raw_valid) != len(unique_valid)
    return {
        "user_id": user["user_id"],
        "target": user["target"],
        "raw_output": raw_output,
        "ranking": ranking,
        "repaired": repaired,
    }


def main():
    tokenizer, model, device = load_model()
    print(f"Model ready on: {device}")

    naive_output = naive_recommend(USERS[0], tokenizer, model, device)
    print(f"\nNaive recommendation: {naive_output}")
    print("Stable catalog key: None")

    constrained = rank_candidates(
        USERS[0],
        build_rich_context(USERS[0]),
        tokenizer,
        model,
        device,
    )
    print(f"\nRaw constrained output: {constrained['raw_output']}")
    print(f"Constrained ranking: {constrained['ranking']}")
    print(f"Output repaired: {constrained['repaired']}")


if __name__ == "__main__":
    main()

Before you run the script, do you expect the raw model response to be enough on its own every time?

  • Return to the command cell from the previous step.
  • Run the command cell again.

You should first see Raw constrained output. Its exact text can vary.

You should then see Constrained ranking containing T02, T08, T05, T06, and T09 exactly once. The order can vary.

The final line shows Output repaired with a boolean value. That is the key guardrail working: your reranker now always hands downstream code a complete catalog-safe ordering.

Constrained output is missing?

  • Check that the constrained block is inside main().
  • Confirm that the constrained block appears after the naive output lines.
  • Compare each function name with the full-code reference. Python identifiers are case-sensitive.
  • Help me debug the constrained ranking path.

Your reranker now converts generated text into a catalog-safe result. Next, you will compare popularity, short-context, and rich-context ranking with offline metrics.

Enrich Context and Evaluate Rankings

Your guarded reranker now produces a complete catalog-safe ordering for one user. A single plausible list gives you too little evidence to judge the pipeline.

You will use held-out next-watch targets to compare three ranking paths. MRR plus Hit Rate at 3 will reveal where each target appears.

In this step, get ready to:
  • Encode watched titles as a short context.
  • Evaluate popularity plus two LLM context paths.
  • Print per-user rankings plus aggregate metrics.
Add context scoring helpers

A controlled comparison keeps the synthetic users fixed. The short context contains watched titles without engagement signals or request conditions.

  • Return to the Google Colab file editor from earlier.
  • Select genrec_mini_ranker.py.
  • Find the build_rich_context() function.
  • Add the short-context builder directly above it by pasting this code:
def build_short_context(user):
    watched = "; ".join(title_name(event["title_id"]) for event in user["history"])
    return f"Watched titles: {watched}."

What does this code do?

  • The watched value converts each history ID into its catalog title.
  • The returned sentence gives the model one compact record of prior viewing.
  • This creates the limited context path used in the comparison.
  • Find the closing brace of rank_candidates().
  • Add the two ranking metric functions directly below it by pasting this code:
def reciprocal_rank(ranking, target):
    return 1.0 / (ranking.index(target) + 1) if target in ranking else 0.0


def hit_at_k(ranking, target, k=3):
    return 1.0 if target in ranking[:k] else 0.0

How do the metrics work?

  • The reciprocal_rank() function gives earlier target positions a larger score.
  • A target in the first position receives 1.0.
  • The hit_at_k() function returns 1.0 when the target appears within the first three positions.
  • A target outside those positions receives 0.0.
  • Save genrec_mini_ranker.py.
  • Confirm that the new helpers load by running this command:
python genrec_mini_ranker.py

What should I see?

You should still see the naive catalog-key gap. You should also see the constrained ranking complete without a Python error.

Seeing a syntax error?

  • Check that build_short_context() starts at the leftmost indentation level.
  • Check that both metric functions start after the closing brace of rank_candidates().
  • Use the traceback with this prompt: Help me fix my context or metric helpers.
Build evaluation reporting

The popularity baseline preserves each supplied candidate order. The reusable evaluator sends every user through the same guarded reranker with one chosen context builder.

  • Find the end of hit_at_k().
  • Add the popularity baseline evaluator directly below it by pasting this code:
def evaluate_baseline():
    return [
        {
            "user_id": user["user_id"],
            "target": user["target"],
            "ranking": list(user["candidates"]),
            "repaired": False,
        }
        for user in USERS
    ]

What does the baseline preserve?

  • The ranking field copies the original candidate order for each user.
  • The target field stores the held-out next watch used for scoring.
  • The repaired field remains False because the baseline does not parse model output.
  • Add the reusable model evaluator directly below evaluate_baseline() by pasting this code:
def evaluate_llm(label, context_builder, tokenizer, model, device):
    results = []
    for user in USERS:
        result = rank_candidates(
            user,
            context_builder(user),
            tokenizer,
            model,
            device,
        )
        results.append(result)
    return label, results

Why pass in a context builder?

  • The context_builder parameter selects the short or rich representation.
  • Every user still passes through rank_candidates().
  • The candidate constraint plus repair path therefore remains identical across both model evaluations.
  • Save genrec_mini_ranker.py.
  • Confirm that both evaluators load by running this command:
python genrec_mini_ranker.py

What should I see?

You should see the existing naive output complete successfully. The constrained ranking should still contain all five supplied IDs exactly once.

Does the script stop early?

  • Check that the list returned by evaluate_baseline() closes with a square bracket.
  • Check that results.append(result) remains inside the loop in evaluate_llm().
  • Use the traceback with this prompt: Help me debug my evaluation helpers.

Aggregate metrics are easier to trust when you can inspect the underlying target ranks. The next two helpers print both levels of evidence.

  • Find the end of evaluate_llm().
  • Add the aggregate metric printer directly below it by pasting this code:
def print_metrics(label, results):
    mrr = sum(reciprocal_rank(row["ranking"], row["target"]) for row in results) / len(results)
    hit_rate = sum(hit_at_k(row["ranking"], row["target"], 3) for row in results) / len(results)
    print(f"{label:<18} MRR={mrr:.3f} HitRate@3={hit_rate:.3f}")

What does the metric printer calculate?

  • The mrr value averages reciprocal rank across the three synthetic users.
  • The hit_rate value averages the top-three hit result across those users.
  • The formatted row displays both values with three decimal places.
  • Add the per-user rank printer directly below print_metrics() by pasting this code:
def print_user_ranks(label, results):
    for row in results:
        rank = row["ranking"].index(row["target"]) + 1
        print(
            f"{label} {row['user_id']} target={row['target']} "
            f"rank={rank} repaired={row['repaired']} ranking={row['ranking']}"
        )

What does the rank printer expose?

  • The rank value converts Python's zero-based position into a position starting at one.
  • Each row includes the user ID plus the held-out target.
  • Each row also displays the repair status plus the final catalog-safe ranking.
  • Save genrec_mini_ranker.py.
  • Confirm that both reporting helpers load by running this command:
python genrec_mini_ranker.py

What should I see?

You should see the existing constrained-ranking output finish successfully. This confirms that every evaluation helper is available to the completed main flow.

Seeing an indentation error?

  • Keep print_metrics() at the leftmost indentation level.
  • Keep print_user_ranks() at the leftmost indentation level.
  • Use the traceback with this prompt: Help me correct my reporting helper indentation.
Run the three-way comparison

The completed evaluation compares untouched popularity order with two context encodings. Every model-produced list passes through catalog validation before scoring.

  • Find the line in main() that prints Output repaired.
  • Add the evaluation calls plus per-user report directly below it by pasting this code:
    baseline_results = evaluate_baseline()
    short_label, short_results = evaluate_llm(
        "short-context",
        build_short_context,
        tokenizer,
        model,
        device,
    )
    rich_label, rich_results = evaluate_llm(
        "rich-context",
        build_rich_context,
        tokenizer,
        model,
        device,
    )

    print("\nPer-user ranks:")
    print_user_ranks("popularity", baseline_results)
    print_user_ranks(short_label, short_results)
    print_user_ranks(rich_label, rich_results)

What do these calls compare?

  • The popularity path uses each candidate set in its original order.
  • The short-context path gives the model watched titles only.
  • The rich-context path includes engagement signals plus request conditions.
  • The per-user rows reveal the target position produced by each path.
  • Save genrec_mini_ranker.py.

Before you run this check, which context path do you expect to place the held-out targets earliest?

  • Run the per-user comparison with this command:
python genrec_mini_ranker.py

What should I see?

You should see nine per-user rows covering popularity plus short-context plus rich-context.

Each row shows a target rank plus repair status plus catalog-safe ordering. The model-produced rankings can vary across environments.

Does evaluation stop on a user?

  • Check that each evaluate_llm() call receives the tokenizer plus model plus device.
  • Check that build_short_context is passed without parentheses.
  • Use the traceback with this prompt: Help me trace the failed evaluation call.
  • Add the offline metric report directly below the per-user printing calls by pasting this code:
    print("\nOffline comparison:")
    print_metrics("popularity", baseline_results)
    print_metrics(short_label, short_results)
    print_metrics(rich_label, rich_results)
    print(
        "\nInterpretation: these synthetic offline metrics test pipeline behavior; "
        "they do not prove production lift or label efficiency."
    )

Why limit the interpretation?

  • The metric rows summarize target positions across three synthetic cases.
  • Those cases exercise the ranking plus scoring pipeline.
  • They do not establish production lift or label efficiency.
  • Save genrec_mini_ranker.py.

Before you run the final check, do you expect all three metric rows to have the same values?

  • Run the completed offline evaluation with this command:
python genrec_mini_ranker.py

What should I see?

You should see metric rows named popularity plus short-context plus rich-context.

Every row should include MRR plus HitRate@3 values. The concluding interpretation should limit the result to pipeline behavior.

You now have a measurable recommendation pipeline. Every aggregate result can be traced back to a catalog-safe ranking for one synthetic user.

Missing a metric row?

  • Check that all three print_metrics() calls remain inside main().
  • Check that short_label receives the label returned by the short-context evaluation.
  • Check that rich_label receives the label returned by the rich-context evaluation.
  • Use the console output with this prompt: Help me find why a metric row is missing.

✔️ Awesome, I've got everything!

Great work. Your completed genrec_mini_ranker.py now compares three catalog-safe ranking paths with traceable offline metrics.

ⓧ I'd like to double check the full code

Compare your saved genrec_mini_ranker.py with this complete reference:

import re

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

MODEL_ID = "Qwen/Qwen2.5-0.5B-Instruct"

CATALOG = {
    "T01": {
        "title": "City of Embers",
        "genres": "science fiction, mystery",
        "description": "A transit engineer uncovers a conspiracy beneath a neon megacity.",
    },
    "T02": {
        "title": "Second Serve",
        "genres": "sports, comedy",
        "description": "A retired tennis player coaches an unlikely neighborhood team.",
    },
    "T03": {
        "title": "Quiet Orbit",
        "genres": "science fiction, drama",
        "description": "Two astronauts repair a silent research station while confronting an old loss.",
    },
    "T04": {
        "title": "Kitchen Rivals",
        "genres": "reality, food",
        "description": "Home cooks race through playful regional cooking challenges.",
    },
    "T05": {
        "title": "Shadow Protocol",
        "genres": "spy, thriller",
        "description": "An analyst follows a digital trail through a tense international operation.",
    },
    "T06": {
        "title": "Wild Coasts",
        "genres": "nature, documentary",
        "description": "A calm journey through wildlife communities along remote coastlines.",
    },
    "T07": {
        "title": "Maple Street",
        "genres": "family, comedy",
        "description": "Three neighboring families turn ordinary weekends into comic disasters.",
    },
    "T08": {
        "title": "Ancient Machines",
        "genres": "history, documentary, mystery",
        "description": "Engineers reconstruct ingenious devices from lost civilizations.",
    },
    "T09": {
        "title": "Weekend Detectives",
        "genres": "comedy, mystery",
        "description": "Two siblings investigate harmless mysteries around their small town.",
    },
    "T10": {
        "title": "Cosmic Frontiers",
        "genres": "space, documentary",
        "description": "Scientists explain new discoveries through vivid journeys across the solar system.",
    },
}

USERS = [
    {
        "user_id": "U01",
        "history": [
            {"title_id": "T01", "completion": 98, "liked": True, "days_ago": 2},
            {"title_id": "T03", "completion": 94, "liked": True, "days_ago": 7},
            {"title_id": "T05", "completion": 31, "liked": False, "days_ago": 1},
        ],
        "request_context": "evening, television, wants something thoughtful",
        "candidates": ["T02", "T08", "T05", "T06", "T09"],
        "target": "T08",
    },
    {
        "user_id": "U02",
        "history": [
            {"title_id": "T02", "completion": 100, "liked": True, "days_ago": 3},
            {"title_id": "T07", "completion": 91, "liked": True, "days_ago": 6},
            {"title_id": "T04", "completion": 84, "liked": False, "days_ago": 4},
        ],
        "request_context": "weekend afternoon, tablet, wants something light",
        "candidates": ["T05", "T01", "T09", "T08", "T03"],
        "target": "T09",
    },
    {
        "user_id": "U03",
        "history": [
            {"title_id": "T06", "completion": 97, "liked": True, "days_ago": 1},
            {"title_id": "T08", "completion": 88, "liked": True, "days_ago": 9},
            {"title_id": "T03", "completion": 52, "liked": False, "days_ago": 12},
        ],
        "request_context": "late evening, laptop, prefers calm factual stories",
        "candidates": ["T02", "T07", "T05", "T10", "T09"],
        "target": "T10",
    },
]

SYSTEM_MESSAGE = (
    "You are a recommendation ranking service. Follow the requested output format exactly."
)


def load_model():
    device = "cuda" if torch.cuda.is_available() else "cpu"
    tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
    model = AutoModelForCausalLM.from_pretrained(MODEL_ID).to(device)
    return tokenizer, model, device


def generate_text(prompt, tokenizer, model, device, max_new_tokens=64):
    messages = [
        {"role": "system", "content": SYSTEM_MESSAGE},
        {"role": "user", "content": prompt},
    ]
    text = tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True,
    )
    model_inputs = tokenizer([text], return_tensors="pt").to(device)
    generated_ids = model.generate(
        **model_inputs,
        max_new_tokens=max_new_tokens,
        do_sample=False,
    )
    generated_only = [
        output_ids[len(input_ids) :]
        for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
    ]
    return tokenizer.batch_decode(generated_only, skip_special_tokens=True)[0].strip()


def title_name(title_id):
    return CATALOG[title_id]["title"]


def naive_recommend(user, tokenizer, model, device):
    watched = ", ".join(title_name(event["title_id"]) for event in user["history"])
    prompt = (
        f"The viewer watched: {watched}. Recommend one title they may enjoy next. "
        "Respond with a title name only. Do not use catalog IDs."
    )
    return generate_text(prompt, tokenizer, model, device, max_new_tokens=24)


def build_short_context(user):
    watched = "; ".join(title_name(event["title_id"]) for event in user["history"])
    return f"Watched titles: {watched}."


def build_rich_context(user):
    lines = [f"Request context: {user['request_context']}", "Viewing history:"]
    for event in user["history"]:
        liked = "yes" if event["liked"] else "no"
        lines.append(
            f"- {event['title_id']} {title_name(event['title_id'])} | "
            f"completion {event['completion']}% | liked {liked} | "
            f"{event['days_ago']} days ago"
        )
    return "\n".join(lines)


def build_candidate_block(candidate_ids):
    lines = []
    for title_id in candidate_ids:
        item = CATALOG[title_id]
        lines.append(
            f"- {title_id} | {item['title']} | {item['genres']} | {item['description']}"
        )
    return "\n".join(lines)


def extract_catalog_ids(raw_output, candidate_ids):
    matches = re.findall(r"\bT\d{2}\b", raw_output.upper())
    raw_valid = [title_id for title_id in matches if title_id in candidate_ids]
    unique_valid = []
    for title_id in raw_valid:
        if title_id not in unique_valid:
            unique_valid.append(title_id)
    return raw_valid, unique_valid


def rank_candidates(user, context, tokenizer, model, device):
    candidate_ids = user["candidates"]
    prompt = f"""Rank the candidate catalog IDs for this viewer.

Viewer context:
{context}

Candidates:
{build_candidate_block(candidate_ids)}

Rules:
1. Use only the supplied candidate IDs.
2. Output every candidate ID exactly once.
3. Rank most likely next watch first.
4. Return one line in this exact format: RANKING: T01,T02,T03
"""
    raw_output = generate_text(prompt, tokenizer, model, device)
    raw_valid, unique_valid = extract_catalog_ids(raw_output, candidate_ids)
    missing = [title_id for title_id in candidate_ids if title_id not in unique_valid]
    ranking = unique_valid + missing
    repaired = len(unique_valid) != len(candidate_ids) or len(raw_valid) != len(unique_valid)
    return {
        "user_id": user["user_id"],
        "target": user["target"],
        "raw_output": raw_output,
        "ranking": ranking,
        "repaired": repaired,
    }


def reciprocal_rank(ranking, target):
    return 1.0 / (ranking.index(target) + 1) if target in ranking else 0.0


def hit_at_k(ranking, target, k=3):
    return 1.0 if target in ranking[:k] else 0.0


def evaluate_baseline():
    return [
        {
            "user_id": user["user_id"],
            "target": user["target"],
            "ranking": list(user["candidates"]),
            "repaired": False,
        }
        for user in USERS
    ]


def evaluate_llm(label, context_builder, tokenizer, model, device):
    results = []
    for user in USERS:
        result = rank_candidates(
            user,
            context_builder(user),
            tokenizer,
            model,
            device,
        )
        results.append(result)
    return label, results


def print_metrics(label, results):
    mrr = sum(reciprocal_rank(row["ranking"], row["target"]) for row in results) / len(results)
    hit_rate = sum(hit_at_k(row["ranking"], row["target"], 3) for row in results) / len(results)
    print(f"{label:<18} MRR={mrr:.3f} HitRate@3={hit_rate:.3f}")


def print_user_ranks(label, results):
    for row in results:
        rank = row["ranking"].index(row["target"]) + 1
        print(
            f"{label} {row['user_id']} target={row['target']} "
            f"rank={rank} repaired={row['repaired']} ranking={row['ranking']}"
        )


def main():
    tokenizer, model, device = load_model()
    print(f"Model ready on: {device}")

    naive_output = naive_recommend(USERS[0], tokenizer, model, device)
    print(f"\nNaive recommendation: {naive_output}")
    print("Stable catalog key: None")

    constrained = rank_candidates(
        USERS[0],
        build_rich_context(USERS[0]),
        tokenizer,
        model,
        device,
    )
    print(f"\nRaw constrained output: {constrained['raw_output']}")
    print(f"Constrained ranking: {constrained['ranking']}")
    print(f"Output repaired: {constrained['repaired']}")

    baseline_results = evaluate_baseline()
    short_label, short_results = evaluate_llm(
        "short-context",
        build_short_context,
        tokenizer,
        model,
        device,
    )
    rich_label, rich_results = evaluate_llm(
        "rich-context",
        build_rich_context,
        tokenizer,
        model,
        device,
    )

    print("\nPer-user ranks:")
    print_user_ranks("popularity", baseline_results)
    print_user_ranks(short_label, short_results)
    print_user_ranks(rich_label, rich_results)

    print("\nOffline comparison:")
    print_metrics("popularity", baseline_results)
    print_metrics(short_label, short_results)
    print_metrics(rich_label, rich_results)
    print(
        "\nInterpretation: these synthetic offline metrics test pipeline behavior; "
        "they do not prove production lift or label efficiency."
    )


if __name__ == "__main__":
    main()

How is the full file organized?

The file moves from model setup to context construction. It then applies guarded ranking plus offline evaluation.

The main() function runs every ranking path against the same users plus candidate sets.

Secret mission

Add a Context Budget

Rich viewer histories can outgrow the space available to a reranker. Build a deterministic compaction policy that keeps two high-signal events per user before evaluating the compacted context with the existing catalog-safe pipeline.

Clean Up Your Resources

Clean Up Your Resources

Your Google Colab notebook uses a free managed runtime, so this project leaves no paid cloud resources or ongoing compute charges. Decide whether to keep the notebook, release the temporary runtime, or delete the project entirely.

Resources you used:

  • The Google Colab notebook stored in Google Drive.
  • The temporary Colab managed runtime containing genrec_mini_ranker.py plus secret_mission_context_compaction.py.
  • The public model cache stored inside the temporary runtime.

Keep everything running

No action is needed while you are actively using the notebook or planning another ranking experiment.

  • Keep the notebook in Google Drive if you plan to rerun the metrics.
  • Download genrec_mini_ranker.py from the temporary runtime to preserve a separate copy.
  • Download secret_mission_context_compaction.py from the temporary runtime to preserve the context-budget experiment.

The notebook persists in Google Drive. Colab may reclaim the managed runtime after idleness or its maximum lifetime.

Pause - I'll come back to this later

Releasing the managed runtime removes the temporary files plus the model cache. Your notebook remains stored in Google Drive.

  • Switch back to the Google Colab notebook.
  • Select the Runtime menu.
  • Select Disconnect and delete runtime.

Colab releases the temporary virtual machine. You can return to the notebook later and create a fresh runtime.

Delete - I don't want to use this again

Deleting the project removes the notebook from its Google Drive location. It also clears both project files plus the cached model from the temporary runtime.

  • Switch back to the Google Colab notebook.
  • Select the Runtime menu.
  • Select Disconnect and delete runtime.

Colab releases the managed virtual machine. The temporary project files plus the model cache are removed with it.

  • Return to Google Drive.
  • Delete the Google Colab notebook from Google Drive.
  • Confirm the notebook no longer appears in its previous Google Drive location.

Cleanup complete. Your temporary runtime contents are gone.

Nice Work!

Nice Work!

You did it! Your Google Colab prototype now turns viewing behavior into catalog-safe recommendations. Your guarded reranker also reports offline evidence without presenting three synthetic cases as production lift.

You've learned how to:

  • Build a synthetic multi-stage recommendation pipeline that starts with popularity retrieval. Place an LLM reranker behind that candidate boundary.
  • Expose the missing catalog key in the naive recommendation path. Produce a catalog-safe ranking through ID validation plus deterministic missing-ID repair.
  • Compare popularity, short-context, and rich-context using MRR plus Hit Rate at 3. Keep the resulting synthetic evidence separate from production claims.
  • Complete the optional Secret Mission by adding a two-event context budget. Evaluate the resulting compacted-rich ranking with the existing metrics.

Ready to quiz yourself?