GenRec:如何把推荐系统改造成 LLM 原生架构

2026-07-31 26 预计阅读时间: 1 分钟
来源: netflixtechblog.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:10 分钟

“GenRec: Towards LLM-Native Recommendation at Netflix”这个标题指向了一种重要变化:大语言模型不再只是给传统推荐结果生成一句解释,而是开始进入推荐链路的核心,用统一的语言接口理解用户意图、组织候选内容,并输出可执行的推荐决策。

由于来源摘要没有披露 Netflix 的具体架构、模型或线上指标,下面不会推断其内部实现,而是讨论一种可验证的 LLM 原生推荐设计,以及团队可以怎样搭建最小原型。

从预测分数转向生成推荐决策

传统推荐系统通常将问题拆成召回、粗排、精排和重排:模型预测点击率、观看时长或转化概率,系统再按照业务规则组合结果。LLM 原生推荐并不意味着抛弃这些模块,而是将推荐任务进一步表达为一个生成问题:

  • 输入是用户近期行为、当前意图、内容元数据和上下文约束。
  • 输出不是自由文本,而是结构化的内容 ID、推荐理由和置信度。
  • 模型需要在候选集合内决策,不能凭空创造不存在的影片或商品。
  • 最终结果仍要经过权限、地区、年龄分级、库存和多样性规则校验。

这种架构的价值在于,文本、标签、搜索词、观看历史和会话反馈可以进入同一上下文。例如,“想看一部节奏快、九十分钟左右、适合全家观看的喜剧”不必先被手工拆成多个过滤条件,模型可以直接理解组合意图。

但“能够理解”不等于“可以直接上线”。如果模型自由生成内容名称,幻觉会立即变成错误推荐。因此,LLM 原生系统更可靠的形态通常是受约束生成:检索层提供候选,模型只在候选 ID 中选择,确定性服务负责执行最后的业务规则。

一条可控的推荐链路

可以把最小链路划分为四步:

  1. 构造用户状态:压缩近期行为,保留时间、完成度、跳过行为和显式反馈。
  2. 检索候选内容:使用向量检索、协同过滤或现有召回服务得到几十到几百个候选项。
  3. 让 LLM 结构化选择:要求模型只返回候选 ID,并给出可解析的 JSON。
  4. 校验与重排:删除无效 ID,应用可用性规则,再处理重复类型、内容新鲜度和探索比例。

这里最关键的边界是:LLM 负责语义判断,普通程序负责事实约束。内容是否在某个地区可播放、是否适合儿童、是否已经下架,都应该由数据库或策略服务判断,而不是让模型回忆。

推荐理由也不应直接暴露内部画像。与其输出“因为系统判断你属于某类用户”,不如引用当前请求和真实行为,例如“符合你提出的轻松喜剧偏好”。生成前还要限制理由可以使用的证据字段,避免模型编造观看记录。

可以这样实践:运行一个受约束的 GenRec 原型

下面是一个不依赖第三方包的最小 Python 示例。为保证代码可直接运行,它使用确定性的 mock_llm 模拟结构化模型响应;接入真实模型时,只需替换这个函数,并继续保留候选 ID 校验和业务过滤。

import json
from dataclasses import dataclass


@dataclass(frozen=True)
class Item:
    item_id: str
    title: str
    genres: tuple[str, ...]
    maturity: int
    available: bool = True


CATALOG = [
    Item("m101", "City Laughs", ("comedy", "family"), 7),
    Item("m102", "Fast Track", ("action", "thriller"), 16),
    Item("m103", "Kitchen Weekend", ("comedy", "food"), 7),
    Item("m104", "Quiet Orbit", ("science-fiction", "drama"), 13, False),
]


def retrieve_candidates(query: str) -> list[Item]:
    tokens = {token.lower() for token in query.replace(",", " ").split()}
    scored = []
    for item in CATALOG:
        score = sum(genre in tokens for genre in item.genres)
        scored.append((score, item.item_id, item))
    return [item for _, _, item in sorted(scored, reverse=True)]


def build_prompt(query: str, candidates: list[Item]) -> str:
    payload = [
        {"id": item.item_id, "title": item.title, "genres": item.genres}
        for item in candidates
    ]
    return f"""You are a recommendation ranker.
User request: {query}
Candidates: {json.dumps(payload)}
Return JSON only: {{"items": [{{"id": "candidate-id", "reason": "short reason"}}]}}
Use only candidate IDs. Return at most 3 items.
"""


def mock_llm(prompt: str, candidates: list[Item]) -> str:
    # Replace this function with an LLM API call that requests JSON output.
    selected = [
        {"id": item.item_id, "reason": f"Matches: {', '.join(item.genres)}"}
        for item in candidates[:3]
    ]
    return json.dumps({"items": selected})


def validate(response: str, candidates: list[Item], max_maturity: int) -> list[dict]:
    data = json.loads(response)
    allowed = {item.item_id: item for item in candidates}
    result = []
    seen = set()

    for entry in data.get("items", []):
        item_id = entry.get("id")
        item = allowed.get(item_id)
        if not item or item_id in seen:
            continue
        if not item.available or item.maturity > max_maturity:
            continue
        seen.add(item_id)
        result.append({
            "id": item_id,
            "title": item.title,
            "reason": str(entry.get("reason", ""))[:120],
        })
    return result[:3]


def recommend(query: str, max_maturity: int = 13) -> list[dict]:
    candidates = retrieve_candidates(query)
    prompt = build_prompt(query, candidates)
    response = mock_llm(prompt, candidates)
    return validate(response, candidates, max_maturity)


if __name__ == "__main__":
    recommendations = recommend("comedy family fast", max_maturity=13)
    print(json.dumps(recommendations, indent=2))

将代码保存为 genrec_demo.py 后可以直接执行:

python genrec_demo.py

接入真实 LLM 时,应把候选列表控制在模型上下文可承受的范围内,并使用服务端支持的 JSON Schema 或结构化输出能力。即便模型声称遵守 Schema,应用层仍然必须检查 ID、数量、字段长度和内容可用性。

评估不能只看离线相关性

LLM 引入了传统排序模型不明显的新变量:提示词版本、模型版本、采样参数和上下文长度都可能改变结果。评估数据因此至少要记录 prompt_versionmodel_version、候选集合、原始结构化响应和校验后的结果。

离线评估可以覆盖:

  • 候选 ID 合法率与 JSON 解析成功率。
  • 已观看内容重复率和类型多样性。
  • 对年龄、地区与可用性规则的违反率。
  • 同一输入多次调用时的排序稳定性。
  • 相比现有排序器的 NDCG、Recall@K 或人工偏好胜率。
  • 延迟、输入输出 token 数量和单次请求成本。

线上实验则要同时观察参与度与负向信号。点击率上升并不一定代表体验改善;快速退出、频繁跳过、隐藏内容和取消播放都可能说明模型生成了“看起来合理但实际不合适”的结果。

采用时保留确定性的护栏

GenRec 更适合作为渐进式改造,而不是一次替换整个推荐栈。团队可以先让 LLM 重排一个小候选集,保留原有召回和策略服务;随后再测试会话式偏好理解、推荐理由以及跨场景用户状态。

上线前应确认这些事项:

  • 模型只能选择真实、可用的候选 ID。
  • 权限、年龄和地区策略在模型之外强制执行。
  • 超时、解析失败和限流时能回退到稳定排序器。
  • 日志中不保存未经处理的敏感用户文本。
  • 提示词和模型升级必须经过可复现的离线评估与灰度实验。
  • 成本预算按峰值流量计算,而不是只看单次演示。

LLM 原生推荐真正值得关注的地方,不是把推荐系统包装成聊天机器人,而是让复杂意图成为排序过程的一等输入。工程上的成败仍取决于候选质量、约束执行、评估体系和回退能力;生成模型只是其中更灵活、也更需要约束的一层。


相关推荐