推荐系统（十二）—— 大语言模型与推荐系统

当 ChatGPT 横空出世，大语言模型（ LLM）的能力震惊了世界。从文本生成到代码编写，从问答对话到知识推理， LLM 展现出了前所未有的通用智能。那么，这样一个强大的工具能否应用到推荐系统中？答案是肯定的，而且正在发生。

传统的推荐系统依赖协同过滤、矩阵分解、深度学习等方法，它们擅长从用户行为数据中挖掘模式，但往往缺乏对物品语义的深度理解，也难以处理冷启动、可解释性等挑战。 LLM 的出现为推荐系统带来了新的可能性：它能够理解物品的文本描述、用户的历史偏好、甚至生成自然语言的推荐理由。

本文将深入探讨 LLM 在推荐系统中的各种应用方式：从简单的 Prompt-based 推荐，到复杂的端到端架构；从特征增强到重排序，从对话式推荐到可解释推荐。我们会看到 A-LLMRec 、 XRec 、 ChatREC 、 RA-Rec 、 ChatCRS 等前沿架构，理解它们的设计思路，并通过完整的代码实现来掌握这些技术。

LLM 在推荐系统中的角色定位

在深入具体架构之前，需要先理解 LLM 在推荐系统中可以扮演哪些角色。这决定了我们如何设计系统架构，以及如何平衡效果和效率。

传统推荐系统的局限

传统的推荐系统（协同过滤、矩阵分解、深度神经网络）主要依赖用户行为数据（点击、购买、评分等）来学习用户偏好和物品特征。这种方法虽然有效，但存在几个根本性局限：

语义理解不足：传统方法难以理解物品的文本描述、用户评论等语义信息。例如，一个电影推荐系统可能知道用户喜欢"动作片"，但无法理解"充满悬疑的动作片"和"轻松幽默的动作片"之间的区别。

冷启动问题：新用户或新物品缺乏历史行为数据，传统方法难以做出准确推荐。虽然可以用内容特征缓解，但特征工程往往需要大量人工工作。

可解释性差：深度学习模型是黑盒，难以解释为什么推荐某个物品。用户看到推荐结果时，往往不知道原因，降低了信任度。

跨域迁移困难：在一个领域训练的模型很难迁移到另一个领域，因为不同领域的特征空间差异很大。

LLM 带来的新能力

LLM 通过预训练获得了丰富的世界知识和语言理解能力，为推荐系统带来了新的可能性：

深度语义理解： LLM 能够理解物品的文本描述、用户评论、甚至隐含的语义信息。它可以将"悬疑动作片"和"轻松动作片"区分开来。

零样本推理： LLM 可以在没有训练数据的情况下进行推理。对于新物品，只需要提供文本描述， LLM 就能理解其特性并做出推荐。

自然语言生成： LLM 可以生成推荐理由，用自然语言解释为什么推荐某个物品，大大提升了可解释性。

知识迁移： LLM 的预训练知识可以迁移到不同领域，减少了对领域特定数据的需求。

LLM 在推荐系统中的角色

根据 LLM 在推荐流程中的位置和作用，可以将其分为以下几种角色：

1. 特征增强器（ Feature Enhancer）

LLM 用于提取或增强物品和用户的特征表示。例如： - 将物品的文本描述编码为向量 - 从用户评论中提取偏好特征 - 生成物品的语义标签

2. 候选生成器（ Candidate Generator）

LLM 直接用于生成推荐候选。例如： - 基于用户历史，用 LLM 生成候选物品列表 - 通过对话理解用户需求，生成推荐

3. 重排序器（ Reranker）

LLM 用于对候选物品进行精细排序。例如： - 对粗排后的候选进行语义理解和重排序 - 考虑用户意图和物品语义的匹配度

4. 可解释性生成器（ Explanation Generator）

LLM 用于生成推荐理由。例如： - 解释为什么推荐某个物品 - 生成个性化的推荐说明

5. 端到端推荐器（ End-to-End Recommender）

LLM 作为完整的推荐系统，从理解用户需求到生成推荐结果。

接下来，我们将深入探讨每种角色的具体实现方式。

Prompt-based 推荐：最简单的 LLM 应用

Prompt-based 推荐是最直观的 LLM 应用方式：将推荐任务转化为自然语言提示，让 LLM 直接生成推荐结果。虽然简单，但在某些场景下效果不错。

基本思路

Prompt-based 推荐的基本思路：将用户的历史行为和物品信息组织成自然语言提示，让 LLM 理解用户偏好并生成推荐。

示例 Prompt：

用户历史行为：
- 看过《肖申克的救赎》（评分： 5/5）
- 看过《阿甘正传》（评分： 5/5）
- 看过《当幸福来敲门》（评分： 4/5）

候选电影：
1. 《楚门的世界》- 剧情/科幻， 1998 年
2. 《美丽人生》- 剧情/战争， 1997 年
3. 《海上钢琴师》- 剧情/音乐， 1998 年
4. 《教父》- 剧情/犯罪， 1972 年
5. 《辛德勒的名单》- 剧情/历史， 1993 年

请根据用户的历史偏好，推荐 3 部最可能喜欢的电影，并说明推荐理由。

实现代码

让我们实现一个完整的 Prompt-based 推荐系统。这个实现展示了如何将推荐任务转化为自然语言提示，让 LLM 理解用户偏好并生成推荐结果。

Prompt-based 推荐的核心设计： 1. Prompt 构建：将用户历史行为和候选物品组织成结构化的自然语言提示 2. 输出格式控制：通过明确的格式要求，确保 LLM 输出可解析的结构化结果 3. 错误处理：处理 LLM 输出可能的不一致性（如添加 markdown 标记）

import json
from typing import List, Dict, Tuple
from openai import OpenAI
import os

class PromptBasedRecommender:
    """
    基于 Prompt 的推荐系统
    
    Prompt-based 推荐是最直观的 LLM 应用方式，基本思路：
    1. 将推荐任务转化为自然语言提示
    2. 让 LLM 理解用户偏好并生成推荐结果
    3. 解析 LLM 的输出得到结构化推荐
    
    优势：
    - 实现简单，无需训练模型
    - 可解释性强， LLM 会生成推荐理由
    - 零样本能力，无需领域特定数据
    
    局限：
    - 依赖 LLM 的 API，成本较高
    - 延迟较大，不适合实时推荐
    - 输出可能不稳定，需要 robust 的解析
    """
    
    def __init__(self, api_key: str = None, model: str = "gpt-3.5-turbo"):
        """
        初始化推荐器
        
        Args:
            api_key: OpenAI API 密钥
                    - 如果为 None，则从环境变量 OPENAI_API_KEY 读取
                    - 实际应用中应该安全存储，不要硬编码
            model: 使用的 LLM 模型名称
                  - "gpt-3.5-turbo": 成本较低，速度较快
                  - "gpt-4": 效果更好，但成本更高
                  - 也可以使用其他 LLM API（如 Claude 、文心一言等）
        """
        self.client = OpenAI(api_key=api_key or os.getenv("OPENAI_API_KEY"))
        self.model = model
        
    def format_user_history(self, user_history: List[Dict]) -> str:
        """
        格式化用户历史行为
        
        将用户的历史行为组织成易读的自然语言格式，帮助 LLM 理解用户偏好。
        格式化的质量直接影响 LLM 对用户偏好的理解。
        
        Args:
            user_history: 用户历史行为列表
                         - 每个元素是一个字典，包含：
                           - item_name: 物品名称（必需）
                           - rating: 评分（可选，如 4/5）
                           - category: 类别（可选，如"剧情/科幻"）
                           - timestamp: 时间戳（可选）
        
        Returns:
            格式化的历史行为字符串，例如：
            "用户历史行为：
             1. 《肖申克的救赎》 - 剧情（评分： 5/5）
             2. 《阿甘正传》 - 剧情（评分： 5/5）
             ..."
        """
        history_str = "用户历史行为：\n"
        for i, item in enumerate(user_history, 1):
            item_name = item.get('item_name', '未知')
            rating = item.get('rating', 'N/A')
            category = item.get('category', '')
            
            # 构建每个历史行为的描述
            history_str += f"{i}. 《{item_name}》"
            if category:
                history_str += f" - {category}"
            if rating != 'N/A':
                history_str += f"（评分：{rating}/5）"
            history_str += "\n"
        
        return history_str
    
    def format_candidates(self, candidates: List[Dict]) -> str:
        """
        格式化候选物品列表
        
        将候选物品组织成结构化的格式，包含足够的信息让 LLM 做出判断。
        信息越多， LLM 的判断越准确，但也会增加 token 消耗。
        
        Args:
            candidates: 候选物品列表
                        - 每个元素是一个字典，包含：
                          - item_name: 物品名称（必需）
                          - category: 类别（可选）
                          - year: 年份（可选）
                          - description: 描述（可选，但推荐包含）
        
        Returns:
            格式化的候选物品字符串，例如：
            "候选物品：
             1. 《楚门的世界》 - 剧情/科幻， 1998 年
                简介：一个关于真实与虚假的故事
             ..."
        """
        candidates_str = "候选物品：\n"
        for i, item in enumerate(candidates, 1):
            item_name = item.get('item_name', '未知')
            category = item.get('category', '')
            year = item.get('year', '')
            description = item.get('description', '')
            
            # 构建每个候选物品的描述
            candidates_str += f"{i}. 《{item_name}》"
            if category:
                candidates_str += f" - {category}"
            if year:
                candidates_str += f"，{year}年"
            if description:
                candidates_str += f"\n   简介：{description}"
            candidates_str += "\n"
        
        return candidates_str
    
    def build_prompt(self, user_history: List[Dict], 
                    candidates: List[Dict], 
                    top_k: int = 3) -> str:
        """
        构建推荐 Prompt
        
        Prompt 的质量直接影响推荐效果。好的 Prompt 应该：
        1. 清晰明确：告诉 LLM 要做什么
        2. 结构完整：包含所有必要信息（用户历史、候选物品）
        3. 格式规范：明确要求输出格式，便于解析
        
        Args:
            user_history: 用户历史行为列表
            candidates: 候选物品列表
            top_k: 推荐数量，通常设置为 3-10
        
        Returns:
            完整的 Prompt 字符串，包含：
            - 任务描述：告诉 LLM 要做什么
            - 用户历史：帮助 LLM 理解用户偏好
            - 候选物品： LLM 从中选择推荐
            - 输出格式：确保 LLM 输出可解析的结构化结果
        """
        # 格式化用户历史和候选物品
        history_str = self.format_user_history(user_history)
        candidates_str = self.format_candidates(candidates)
        
        # 构建完整的 Prompt
        # 关键设计点：
        # 1. 明确任务：告诉 LLM 要推荐 top_k 个物品
        # 2. 提供上下文：用户历史和候选物品
        # 3. 规范输出：要求 JSON 格式，便于解析
        # 4. 强调格式：只输出 JSON，避免额外文本
        prompt = f"""你是一个专业的电影推荐系统。请根据用户的历史观影偏好，从候选电影中推荐最合适的{top_k}部电影。

{history_str}

{candidates_str}

请按照以下格式输出推荐结果（ JSON 格式）：
{{ 
    "recommendations": [
        {{ 
            "item_name": "电影名称",
            "rank": 1,
            "reason": "推荐理由"
         }}
    ]
 }}

只输出 JSON，不要输出其他内容。"""
        return prompt
    
    def recommend(self, user_history: List[Dict], 
                  candidates: List[Dict], 
                  top_k: int = 3) -> List[Dict]:
        """
        生成推荐
        
        这是 Prompt-based 推荐的核心方法，流程：
        1. 构建 Prompt：将用户历史和候选物品组织成 Prompt
        2. 调用 LLM：发送 Prompt 到 LLM API
        3. 解析输出：将 LLM 的输出解析为结构化推荐结果
        4. 错误处理：处理 LLM 输出可能的不一致性
        
        Args:
            user_history: 用户历史行为列表
            candidates: 候选物品列表
            top_k: 推荐数量
        
        Returns:
            推荐结果列表，每个元素包含：
            - item_name: 推荐的物品名称
            - rank: 推荐排名（ 1 表示最推荐）
            - reason: 推荐理由（ LLM 生成的自然语言解释）
        """
        # 第一步：构建 Prompt
        prompt = self.build_prompt(user_history, candidates, top_k)
        
        try:
            # 第二步：调用 LLM API
            # 关键参数：
            # - temperature: 控制输出的随机性
            #   * 0.3: 较低随机性，输出更一致（推荐用于推荐任务）
            #   * 0.7-1.0: 较高随机性，输出更多样（适合创意任务）
            # - max_tokens: 限制输出长度，控制成本
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个专业的推荐系统助手。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3,  # 降低随机性，提高一致性
                max_tokens=1000   # 限制输出长度，控制 API 成本
            )
            
            # 第三步：提取 LLM 的输出
            content = response.choices[0].message.content.strip()
            
            # 第四步：解析 JSON 输出
            # LLM 有时会在 JSON 前后添加 markdown 标记（如```
{% endraw %}json），需要清理
            if content.startswith("```json"):
                content = content[7:]  # 移除```json 标记
            if content.startswith("```"):
                content = content[3:]   # 移除```标记
            if content.endswith("```"):
                content = content[:-3]   # 移除结尾的```标记
            content = content.strip()
            
            # 解析 JSON
            result = json.loads(content)
            return result.get("recommendations", [])
            
        except json.JSONDecodeError as e:
            # JSON 解析错误： LLM 可能没有按照要求的格式输出
            # 实际应用中应该：
            # 1. 记录错误日志
            # 2. 尝试更 robust 的解析（如正则表达式）
            # 3. 返回默认推荐或重试
            print(f"JSON 解析错误: {e}")
            print(f"LLM 输出: {content}")
            return []
        except Exception as e:
            # 其他错误： API 调用失败、网络错误等
            print(f"推荐生成错误: {e}")
            return []
    
    def recommend_with_explanation(self, user_history: List[Dict], 
                                   candidates: List[Dict], 
                                   top_k: int = 3) -> Dict:
        """
        生成带解释的推荐
        
        Returns:
            包含推荐列表和整体解释的字典
        """
        prompt = self.build_prompt(user_history, candidates, top_k)
        
        # 添加生成整体解释的要求
        prompt += "\n\n 另外，请用一段话总结用户的观影偏好特点。"
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个专业的推荐系统助手。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3,
                max_tokens=1500
            )
            
            content = response.choices[0].message.content.strip()
            
            # 解析结果
            # 这里简化处理，实际应用中需要更 robust 的解析
            if "{" in content and "}" in content:
                json_start = content.find("{")
                json_end = content.rfind("}") + 1
                json_str = content[json_start:json_end]
                
                result = json.loads(json_str)
                recommendations = result.get("recommendations", [])
                
                # 提取整体解释（ JSON 之后的内容）
                explanation = content[json_end:].strip()
                if not explanation:
                    explanation = "基于您的观影历史，为您推荐了以上电影。"
                
                return {
                    "recommendations": recommendations,
                    "user_preference_summary": explanation
                }
            
            return {"recommendations": [], "user_preference_summary": ""}
            
        except Exception as e:
            print(f"推荐生成错误: {e}")
            return {"recommendations": [], "user_preference_summary": ""}


# 使用示例
if __name__ == "__main__":
    # 初始化推荐器
    recommender = PromptBasedRecommender()
    
    # 用户历史
    user_history = [
        {"item_name": "肖申克的救赎", "rating": 5, "category": "剧情/犯罪"},
        {"item_name": "阿甘正传", "rating": 5, "category": "剧情/爱情"},
        {"item_name": "当幸福来敲门", "rating": 4, "category": "剧情/传记"}
    ]
    
    # 候选电影
    candidates = [
        {
            "item_name": "楚门的世界",
            "category": "剧情/科幻",
            "year": 1998,
            "description": "一个关于真实与虚假、自由与控制的深刻寓言"
        },
        {
            "item_name": "美丽人生",
            "category": "剧情/战争",
            "year": 1997,
            "description": "在集中营中用父爱和幽默保护孩子的感人故事"
        },
        {
            "item_name": "海上钢琴师",
            "category": "剧情/音乐",
            "year": 1998,
            "description": "一个天才钢琴师的一生，关于选择与坚持"
        },
        {
            "item_name": "教父",
            "category": "剧情/犯罪",
            "year": 1972,
            "description": "黑帮家族的史诗，关于权力、家族和人性"
        },
        {
            "item_name": "辛德勒的名单",
            "category": "剧情/历史",
            "year": 1993,
            "description": "二战期间拯救犹太人的真实故事"
        }
    ]
    
    # 生成推荐
    recommendations = recommender.recommend(user_history, candidates, top_k=3)
    
    print("推荐结果：")
    for rec in recommendations:
        print(f"\n 排名 {rec['rank']}: 《{rec['item_name']}》")
        print(f"推荐理由：{rec['reason']}")

Prompt 工程技巧

要让 Prompt-based 推荐效果更好，需要注意以下几点：

1. 结构化输入

将用户历史和候选物品组织成清晰的结构，使用编号、分类等让 LLM 更容易理解。

2. 明确输出格式

指定 JSON 格式输出，便于后续解析。可以使用 few-shot examples 来引导 LLM 输出正确格式。

3. 控制温度参数

推荐任务需要一致性，应该使用较低的温度（ 0.1-0.3），而不是创意生成任务的高温度（ 0.7-1.0）。

4. 添加约束条件

在 Prompt 中明确约束，例如"不要推荐用户已经看过的电影"、"优先推荐评分高的电影"等。

5. 处理长上下文

如果用户历史很长，需要截断或摘要。可以使用 LLM 先对历史进行摘要，再用于推荐。

优缺点分析

优点： - 实现简单，无需训练模型 - 可解释性强， LLM 会生成推荐理由 - 零样本能力，对新领域也能工作 - 自然语言交互，用户体验好

缺点： - 延迟高，每次推荐都需要调用 LLM API - 成本高， Token 消耗大 - 不稳定，可能生成格式错误的结果 - 难以处理大规模候选集

Prompt-based 推荐适合小规模、对延迟不敏感的场景，或者作为其他方法的补充。接下来，我们将看到更高效的架构设计。

A-LLMRec：适配器增强的 LLM 推荐架构

A-LLMRec（ Adapter-enhanced LLM for Recommendation）是一种将 LLM 与传统推荐模型结合的架构。基本思路：使用轻量级的适配器（ Adapter）来微调 LLM，使其适应推荐任务，而不是直接使用预训练的 LLM 。

架构设计

A-LLMRec 的架构包含以下几个组件：

1. LLM 编码器：使用预训练的 LLM（如 BERT 、 GPT）来编码物品文本和用户历史
2. 适配器层：在 LLM 的每一层插入轻量级的适配器，用于任务特定的微调
3. 推荐头：将 LLM 的输出映射到推荐分数

适配器机制

适配器是一种参数高效的微调方法。在 Transformer 的每一层中，适配器插入在注意力层和前馈层之后：

其中： - 是降维投影 - ，， 是瓶颈维度

这样，只需要训练 个参数（每个适配器），而不是整个 LLM 的参数。

完整实现

import torch
import torch.nn as nn
from transformers import AutoModel, AutoTokenizer
from typing import List, Dict, Optional

class AdapterLayer(nn.Module):
    """适配器层"""
    
    def __init__(self, hidden_size: int, adapter_size: int = 64):
        """
        Args:
            hidden_size: LLM 的隐藏层维度
            adapter_size: 适配器的瓶颈维度
        """
        super().__init__()
        self.adapter_size = adapter_size
        
        # 降维投影
        self.down_proj = nn.Linear(hidden_size, adapter_size)
        # 上维投影
        self.up_proj = nn.Linear(adapter_size, hidden_size)
        # 激活函数
        self.activation = nn.ReLU()
        # Layer norm
        self.layer_norm = nn.LayerNorm(hidden_size)
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Args:
            x: 输入张量，形状为 (batch_size, seq_len, hidden_size)
        
        Returns:
            适配器输出
        """
        residual = x
        x = self.layer_norm(x)
        
        # 适配器前向传播
        down = self.down_proj(x)
        down = self.activation(down)
        up = self.up_proj(down)
        
        # 残差连接
        output = residual + up
        return output


class ALLMRec(nn.Module):
    """A-LLMRec 模型"""
    
    def __init__(
        self,
        model_name: str = "bert-base-chinese",
        adapter_size: int = 64,
        num_items: int = 10000,
        embedding_dim: int = 128,
        max_history_length: int = 50
    ):
        """
        Args:
            model_name: 预训练 LLM 模型名称
            adapter_size: 适配器瓶颈维度
            num_items: 物品数量
            embedding_dim: 物品嵌入维度
            max_history_length: 最大历史长度
        """
        super().__init__()
        
        # LLM 编码器
        self.llm = AutoModel.from_pretrained(model_name)
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.hidden_size = self.llm.config.hidden_size
        
        # 在每一层插入适配器
        self.adapters = nn.ModuleList()
        for _ in range(self.llm.config.num_hidden_layers):
            self.adapters.append(AdapterLayer(self.hidden_size, adapter_size))
        
        # 物品嵌入层（用于物品 ID）
        self.item_embedding = nn.Embedding(num_items, embedding_dim)
        
        # 历史编码器（将历史物品序列编码）
        self.history_encoder = nn.LSTM(
            embedding_dim, 
            self.hidden_size // 2, 
            batch_first=True,
            bidirectional=True
        )
        
        # 推荐头
        self.recommendation_head = nn.Sequential(
            nn.Linear(self.hidden_size * 2, self.hidden_size),
            nn.ReLU(),
            nn.Dropout(0.1),
            nn.Linear(self.hidden_size, 1)
        )
        
        self.max_history_length = max_history_length
        
    def encode_text(self, texts: List[str]) -> torch.Tensor:
        """
        使用 LLM + 适配器编码文本
        
        Args:
            texts: 文本列表
        
        Returns:
            文本表示，形状为 (batch_size, hidden_size)
        """
        # Tokenize
        encoded = self.tokenizer(
            texts,
            padding=True,
            truncation=True,
            max_length=512,
            return_tensors="pt"
        )
        
        # 移动到设备
        device = next(self.llm.parameters()).device
        encoded = {k: v.to(device) for k, v in encoded.items()}
        
        # LLM 编码
        outputs = self.llm(**encoded)
        hidden_states = outputs.last_hidden_state  # (batch_size, seq_len, hidden_size)
        
        # 通过适配器
        for adapter in self.adapters:
            hidden_states = adapter(hidden_states)
        
        # 取 [CLS] token 的表示
        cls_representation = hidden_states[:, 0, :]  # (batch_size, hidden_size)
        
        return cls_representation
    
    def encode_history(self, item_ids: torch.Tensor) -> torch.Tensor:
        """
        编码用户历史
        
        Args:
            item_ids: 物品 ID 序列，形状为 (batch_size, history_length)
        
        Returns:
            历史表示，形状为 (batch_size, hidden_size)
        """
        # 获取物品嵌入
        item_embeds = self.item_embedding(item_ids)  # (batch_size, seq_len, embedding_dim)
        
        # LSTM 编码
        lstm_out, (h_n, c_n) = self.history_encoder(item_embeds)
        
        # 使用最后时刻的隐藏状态
        # 双向 LSTM，拼接前向和后向的隐藏状态
        history_repr = torch.cat([h_n[0], h_n[1]], dim=1)  # (batch_size, hidden_size)
        
        return history_repr
    
    def forward(
        self,
        item_texts: List[str],
        user_history_ids: torch.Tensor,
        candidate_ids: Optional[torch.Tensor] = None
    ) -> torch.Tensor:
        """
        前向传播
        
        Args:
            item_texts: 候选物品的文本描述列表
            user_history_ids: 用户历史物品 ID，形状为 (batch_size, history_length)
            candidate_ids: 候选物品 ID（可选）
        
        Returns:
            推荐分数，形状为 (batch_size, 1)
        """
        # 编码物品文本
        item_repr = self.encode_text(item_texts)  # (batch_size, hidden_size)
        
        # 编码用户历史
        history_repr = self.encode_history(user_history_ids)  # (batch_size, hidden_size)
        
        # 拼接物品表示和历史表示
        combined = torch.cat([item_repr, history_repr], dim=1)  # (batch_size, hidden_size * 2)
        
        # 推荐头
        score = self.recommendation_head(combined)  # (batch_size, 1)
        
        return score
    
    def predict(self, item_text: str, user_history_ids: List[int]) -> float:
        """
        预测用户对物品的评分
        
        Args:
            item_text: 物品文本描述
            user_history_ids: 用户历史物品 ID 列表
        
        Returns:
            预测分数
        """
        self.eval()
        
        # 准备输入
        item_texts = [item_text]
        
        # 填充历史
        if len(user_history_ids) > self.max_history_length:
            user_history_ids = user_history_ids[-self.max_history_length:]
        else:
            user_history_ids = user_history_ids + [0] * (self.max_history_length - len(user_history_ids))
        
        history_tensor = torch.tensor([user_history_ids], dtype=torch.long)
        
        # 移动到设备
        device = next(self.parameters()).device
        history_tensor = history_tensor.to(device)
        
        with torch.no_grad():
            score = self.forward(item_texts, history_tensor)
        
        return score.item()


# 训练示例
def train_allmrec():
    """训练 A-LLMRec 模型"""
    from torch.utils.data import Dataset, DataLoader
    
    class RecommendationDataset(Dataset):
        def __init__(self, data: List[Dict]):
            self.data = data
        
        def __len__(self):
            return len(self.data)
        
        def __getitem__(self, idx):
            return self.data[idx]
    
    # 初始化模型
    model = ALLMRec(
        model_name="bert-base-chinese",
        adapter_size=64,
        num_items=10000,
        max_history_length=50
    )
    
    # 优化器（只优化适配器和推荐头的参数）
    trainable_params = []
    trainable_params.extend(model.adapters.parameters())
    trainable_params.extend(model.item_embedding.parameters())
    trainable_params.extend(model.history_encoder.parameters())
    trainable_params.extend(model.recommendation_head.parameters())
    
    optimizer = torch.optim.Adam(trainable_params, lr=1e-4)
    criterion = nn.MSELoss()
    
    # 示例数据
    train_data = [
        {
            "item_text": "一部充满悬疑的动作片",
            "user_history": [1, 2, 3, 4, 5],
            "rating": 4.5
        },
        # ... 更多数据
    ]
    
    dataset = RecommendationDataset(train_data)
    dataloader = DataLoader(dataset, batch_size=32, shuffle=True)
    
    # 训练循环
    model.train()
    for epoch in range(10):
        total_loss = 0
        for batch in dataloader:
            item_texts = batch["item_text"]
            user_history = batch["user_history"]
            ratings = batch["rating"]
            
            # 准备历史张量
            max_len = max(len(h) for h in user_history)
            history_tensor = torch.zeros(len(user_history), max_len, dtype=torch.long)
            for i, h in enumerate(user_history):
                history_tensor[i, :len(h)] = torch.tensor(h)
            
            # 前向传播
            scores = model(item_texts, history_tensor)
            loss = criterion(scores.squeeze(), ratings.float())
            
            # 反向传播
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item()
        
        print(f"Epoch {epoch+1}, Loss: {total_loss / len(dataloader):.4f}")


if __name__ == "__main__":
    # 使用示例
    model = ALLMRec()
    
    # 预测
    score = model.predict(
        item_text="一部感人的剧情片",
        user_history_ids=[1, 2, 3, 4, 5]
    )
    print(f"预测分数: {score:.2f}")

优势分析

A-LLMRec 的优势在于：

1. 参数高效：只需要训练适配器参数，而不是整个 LLM，大大降低了训练成本
2. 知识保留： LLM 的预训练知识得以保留，同时适应推荐任务
3. 可扩展性：可以轻松添加新的适配器来处理不同的推荐场景
4. 灵活性：可以针对不同层使用不同大小的适配器

XRec：可解释的 LLM 推荐系统

XRec（ Explainable Recommendation）专注于使用 LLM 生成可解释的推荐理由。用户不仅看到推荐结果，还能理解为什么推荐这些物品。

架构设计

XRec 包含两个主要组件：

1. 推荐模型：生成推荐分数（可以使用任何推荐模型）
2. 解释生成器：基于 LLM 生成推荐理由

实现代码

from typing import List, Dict, Tuple
import torch
import torch.nn as nn
from transformers import AutoModelForCausalLM, AutoTokenizer

class XRecExplainer:
    """XRec 可解释推荐系统"""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo"):
        """
        Args:
            model_name: 用于生成解释的 LLM 模型
        """
        self.model_name = model_name
        # 这里使用 OpenAI API，也可以使用本地模型
        from openai import OpenAI
        import os
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    
    def generate_explanation(
        self,
        user_history: List[Dict],
        recommended_item: Dict,
        recommendation_score: float,
        explanation_type: str = "detailed"
    ) -> str:
        """
        生成推荐解释
        
        Args:
            user_history: 用户历史行为
            recommended_item: 推荐的物品信息
            recommendation_score: 推荐分数
            explanation_type: 解释类型（"brief", "detailed", "comparative"）
        
        Returns:
            推荐解释文本
        """
        # 格式化用户历史
        history_str = self._format_history(user_history)
        
        # 构建 Prompt
        if explanation_type == "brief":
            prompt = self._build_brief_prompt(history_str, recommended_item, recommendation_score)
        elif explanation_type == "detailed":
            prompt = self._build_detailed_prompt(history_str, recommended_item, recommendation_score)
        elif explanation_type == "comparative":
            prompt = self._build_comparative_prompt(history_str, recommended_item, recommendation_score)
        else:
            prompt = self._build_detailed_prompt(history_str, recommended_item, recommendation_score)
        
        # 调用 LLM 生成解释
        try:
            response = self.client.chat.completions.create(
                model=self.model_name,
                messages=[
                    {"role": "system", "content": "你是一个专业的推荐系统解释生成器。你的任务是生成清晰、有说服力的推荐理由。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.7,
                max_tokens=500
            )
            
            explanation = response.choices[0].message.content.strip()
            return explanation
        except Exception as e:
            print(f"解释生成错误: {e}")
            return "基于您的历史偏好，我们为您推荐了此物品。"
    
    def _format_history(self, user_history: List[Dict]) -> str:
        """格式化用户历史"""
        history_str = "用户历史行为：\n"
        for i, item in enumerate(user_history[:10], 1):  # 只取最近 10 个
            item_name = item.get('item_name', '未知')
            rating = item.get('rating', 'N/A')
            category = item.get('category', '')
            history_str += f"{i}. 《{item_name}》"
            if category:
                history_str += f" ({category})"
            if rating != 'N/A':
                history_str += f" - 评分: {rating}/5"
            history_str += "\n"
        return history_str
    
    def _build_brief_prompt(self, history_str: str, item: Dict, score: float) -> str:
        """构建简短解释的 Prompt"""
        return f"""用户历史：
{history_str}

推荐物品：《{item.get('item_name', '未知')}》
推荐分数：{score:.2f}/5.0

请用一句话解释为什么推荐这个物品。解释要简洁明了，不超过 30 字。"""
    
    def _build_detailed_prompt(self, history_str: str, item: Dict, score: float) -> str:
        """构建详细解释的 Prompt"""
        item_desc = item.get('description', '')
        item_category = item.get('category', '')
        
        return f"""用户历史：
{history_str}

推荐物品：
- 名称：《{item.get('item_name', '未知')}》
- 类别：{item_category}
- 描述：{item_desc}
- 推荐分数：{score:.2f}/5.0

请详细解释为什么推荐这个物品。解释应该：
1. 说明与用户历史偏好的关联
2. 突出物品的特点和优势
3. 说明为什么用户可能会喜欢
4. 语言自然流畅，有说服力

请生成一段 100-200 字的推荐解释。"""
    
    def _build_comparative_prompt(self, history_str: str, item: Dict, score: float) -> str:
        """构建对比解释的 Prompt"""
        return f"""用户历史：
{history_str}

推荐物品：《{item.get('item_name', '未知')}》
推荐分数：{score:.2f}/5.0

请通过对比用户历史中的物品，解释为什么推荐这个新物品。说明：
1. 与历史物品的相似之处
2. 与历史物品的不同之处
3. 为什么这些特点会让用户感兴趣

请生成一段 150-250 字的对比解释。"""


class XRecRecommender:
    """完整的 XRec 推荐系统"""
    
    def __init__(self, recommendation_model, explainer: XRecExplainer):
        """
        Args:
            recommendation_model: 推荐模型（可以是任何推荐模型）
            explainer: 解释生成器
        """
        self.recommendation_model = recommendation_model
        self.explainer = explainer
    
    def recommend_with_explanation(
        self,
        user_id: int,
        candidates: List[Dict],
        user_history: List[Dict],
        top_k: int = 5
    ) -> List[Dict]:
        """
        生成带解释的推荐
        
        Args:
            user_id: 用户 ID
            candidates: 候选物品列表
            user_history: 用户历史行为
            top_k: 推荐数量
        
        Returns:
            推荐结果列表，每个包含物品信息和解释
        """
        # 使用推荐模型生成分数
        scores = []
        for candidate in candidates:
            # 这里简化处理，实际应该调用推荐模型
            score = self.recommendation_model.predict(user_id, candidate['item_id'])
            scores.append((candidate, score))
        
        # 排序
        scores.sort(key=lambda x: x[1], reverse=True)
        
        # 生成解释
        recommendations = []
        for candidate, score in scores[:top_k]:
            explanation = self.explainer.generate_explanation(
                user_history=user_history,
                recommended_item=candidate,
                recommendation_score=score,
                explanation_type="detailed"
            )
            
            recommendations.append({
                "item": candidate,
                "score": score,
                "explanation": explanation
            })
        
        return recommendations


# 使用示例
if __name__ == "__main__":
    # 初始化解释器
    explainer = XRecExplainer()
    
    # 用户历史
    user_history = [
        {"item_name": "肖申克的救赎", "rating": 5, "category": "剧情"},
        {"item_name": "阿甘正传", "rating": 5, "category": "剧情"},
        {"item_name": "当幸福来敲门", "rating": 4, "category": "剧情"}
    ]
    
    # 推荐物品
    recommended_item = {
        "item_name": "美丽人生",
        "category": "剧情/战争",
        "description": "在集中营中用父爱和幽默保护孩子的感人故事"
    }
    
    # 生成解释
    explanation = explainer.generate_explanation(
        user_history=user_history,
        recommended_item=recommended_item,
        recommendation_score=4.8,
        explanation_type="detailed"
    )
    
    print("推荐解释：")
    print(explanation)

XRec 的核心价值在于提升推荐系统的可解释性，让用户理解推荐的原因，从而增加信任度和满意度。

LLM 作为特征增强器

LLM 可以作为特征增强器，将物品的文本信息（描述、评论、标签等）编码为高质量的向量表示，然后用于传统的推荐模型。这种方法结合了 LLM 的语义理解能力和传统推荐模型的高效性。

基本思路

使用 LLM 提取特征的基本流程：

1. 文本编码：使用 LLM 将物品文本编码为向量
2. 特征融合：将 LLM 特征与传统特征（ ID 、类别等）融合
3. 推荐预测：使用融合后的特征进行推荐

实现代码

import torch
import torch.nn as nn
from transformers import AutoModel, AutoTokenizer
from typing import List, Dict, Optional
import numpy as np

class LLMFeatureExtractor:
    """使用 LLM 提取物品特征"""
    
    def __init__(self, model_name: str = "bert-base-chinese", device: str = "cuda"):
        """
        Args:
            model_name: LLM 模型名称
            device: 设备
        """
        self.device = device
        self.model = AutoModel.from_pretrained(model_name).to(device)
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model.eval()
        
        # 冻结 LLM 参数（可选）
        for param in self.model.parameters():
            param.requires_grad = False
    
    def extract_features(self, texts: List[str], batch_size: int = 32) -> np.ndarray:
        """
        提取文本特征
        
        Args:
            texts: 文本列表
            batch_size: 批处理大小
        
        Returns:
            特征向量，形状为 (n_texts, hidden_size)
        """
        features = []
        
        for i in range(0, len(texts), batch_size):
            batch_texts = texts[i:i+batch_size]
            
            # Tokenize
            encoded = self.tokenizer(
                batch_texts,
                padding=True,
                truncation=True,
                max_length=512,
                return_tensors="pt"
            )
            
            # 移动到设备
            encoded = {k: v.to(self.device) for k, v in encoded.items()}
            
            # 编码
            with torch.no_grad():
                outputs = self.model(**encoded)
                # 使用 [CLS] token 的表示
                batch_features = outputs.last_hidden_state[:, 0, :].cpu().numpy()
            
            features.append(batch_features)
        
        return np.vstack(features)
    
    def extract_item_features(self, items: List[Dict]) -> np.ndarray:
        """
        提取物品特征
        
        Args:
            items: 物品列表，每个物品包含 text, description 等字段
        
        Returns:
            物品特征矩阵
        """
        # 组合物品的文本信息
        texts = []
        for item in items:
            text_parts = []
            
            if 'title' in item:
                text_parts.append(item['title'])
            if 'description' in item:
                text_parts.append(item['description'])
            if 'category' in item:
                text_parts.append(f"类别：{item['category']}")
            if 'tags' in item:
                tags_str = ", ".join(item['tags'])
                text_parts.append(f"标签：{tags_str}")
            
            text = " | ".join(text_parts)
            texts.append(text)
        
        return self.extract_features(texts)


class EnhancedRecommendationModel(nn.Module):
    """使用 LLM 特征增强的推荐模型"""
    
    def __init__(
        self,
        num_users: int,
        num_items: int,
        llm_feature_dim: int = 768,
        embedding_dim: int = 128,
        hidden_dims: List[int] = [256, 128, 64]
    ):
        """
        Args:
            num_users: 用户数量
            num_items: 物品数量
            llm_feature_dim: LLM 特征维度
            embedding_dim: ID 嵌入维度
            hidden_dims: 隐藏层维度列表
        """
        super().__init__()
        
        # 用户和物品 ID 嵌入
        self.user_embedding = nn.Embedding(num_users, embedding_dim)
        self.item_embedding = nn.Embedding(num_items, embedding_dim)
        
        # LLM 特征投影层（将 LLM 特征投影到与 ID 嵌入相同的维度）
        self.llm_projection = nn.Linear(llm_feature_dim, embedding_dim)
        
        # 特征融合层
        # 用户特征： ID 嵌入
        # 物品特征： ID 嵌入 + LLM 特征
        input_dim = embedding_dim * 2 + embedding_dim  # user_emb + item_emb + llm_feat
        
        # MLP 层
        layers = []
        prev_dim = input_dim
        for hidden_dim in hidden_dims:
            layers.append(nn.Linear(prev_dim, hidden_dim))
            layers.append(nn.ReLU())
            layers.append(nn.Dropout(0.2))
            prev_dim = hidden_dim
        
        # 输出层
        layers.append(nn.Linear(prev_dim, 1))
        layers.append(nn.Sigmoid())
        
        self.mlp = nn.Sequential(*layers)
    
    def forward(
        self,
        user_ids: torch.Tensor,
        item_ids: torch.Tensor,
        item_llm_features: Optional[torch.Tensor] = None
    ) -> torch.Tensor:
        """
        前向传播
        
        Args:
            user_ids: 用户 ID，形状为 (batch_size,)
            item_ids: 物品 ID，形状为 (batch_size,)
            item_llm_features: 物品的 LLM 特征，形状为 (batch_size, llm_feature_dim)
        
        Returns:
            预测分数，形状为 (batch_size, 1)
        """
        # 用户嵌入
        user_emb = self.user_embedding(user_ids)  # (batch_size, embedding_dim)
        
        # 物品 ID 嵌入
        item_emb = self.item_embedding(item_ids)  # (batch_size, embedding_dim)
        
        # LLM 特征
        if item_llm_features is not None:
            llm_emb = self.llm_projection(item_llm_features)  # (batch_size, embedding_dim)
        else:
            # 如果没有 LLM 特征，使用零向量
            llm_emb = torch.zeros_like(item_emb)
        
        # 拼接特征
        combined = torch.cat([user_emb, item_emb, llm_emb], dim=1)  # (batch_size, embedding_dim * 3)
        
        # MLP
        score = self.mlp(combined)
        
        return score


# 使用示例
if __name__ == "__main__":
    # 初始化特征提取器
    feature_extractor = LLMFeatureExtractor()
    
    # 物品数据
    items = [
        {
            "item_id": 0,
            "title": "肖申克的救赎",
            "description": "一部关于希望和友谊的经典电影",
            "category": "剧情",
            "tags": ["剧情", "犯罪", "经典"]
        },
        {
            "item_id": 1,
            "title": "阿甘正传",
            "description": "一个智障人士的传奇人生",
            "category": "剧情",
            "tags": ["剧情", "励志", "经典"]
        }
    ]
    
    # 提取 LLM 特征
    llm_features = feature_extractor.extract_item_features(items)
    print(f"LLM 特征形状: {llm_features.shape}")
    
    # 初始化推荐模型
    model = EnhancedRecommendationModel(
        num_users=1000,
        num_items=10000,
        llm_feature_dim=768
    )
    
    # 预测
    user_ids = torch.tensor([0, 1])
    item_ids = torch.tensor([0, 1])
    item_llm_feats = torch.tensor(llm_features)
    
    scores = model(user_ids, item_ids, item_llm_feats)
    print(f"预测分数: {scores}")

优势分析

使用 LLM 作为特征增强器的优势：

1. 语义理解： LLM 能够理解物品的文本描述，提取丰富的语义特征
2. 冷启动友好：对于新物品，即使没有历史行为数据，也可以通过文本描述提取特征
3. 可解释性： LLM 特征往往对应语义概念，便于理解
4. 灵活性：可以轻松添加新的文本信息（评论、标签等）来增强特征

LLM 作为重排序器

在推荐系统的多阶段架构中，重排序（ Reranking）是最后一环。 LLM 可以作为重排序器，对粗排后的候选进行精细排序，考虑用户意图、物品语义等复杂因素。

架构设计

LLM 重排序器的基本流程：

1. 候选准备：从粗排阶段获得 Top-K 候选（例如 Top-100）
2. 上下文构建：构建包含用户历史、候选物品信息的上下文
3. LLM 排序：使用 LLM 对候选进行排序
4. 结果输出：返回重排序后的 Top-N 结果

实现代码

{% raw %}

from typing import List, Dict, Tuple
import json
from openai import OpenAI
import os
import numpy as np

class LLMReranker:
    """使用 LLM 进行重排序"""
    
    def __init__(self, model: str = "gpt-3.5-turbo", max_candidates: int = 50):
        """
        Args:
            model: LLM 模型名称
            max_candidates: 最大候选数量（避免 Token 过多）
        """
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = model
        self.max_candidates = max_candidates
    
    def rerank(
        self,
        user_history: List[Dict],
        candidates: List[Dict],
        top_n: int = 10,
        strategy: str = "pairwise"
    ) -> List[Dict]:
        """
        重排序候选物品
        
        Args:
            user_history: 用户历史行为
            candidates: 候选物品列表（已按粗排分数排序）
            top_n: 最终推荐数量
            strategy: 排序策略（"pairwise", "listwise", "pointwise"）
        
        Returns:
            重排序后的物品列表
        """
        # 限制候选数量
        if len(candidates) > self.max_candidates:
            candidates = candidates[:self.max_candidates]
        
        if strategy == "pairwise":
            return self._pairwise_rerank(user_history, candidates, top_n)
        elif strategy == "listwise":
            return self._listwise_rerank(user_history, candidates, top_n)
        else:
            return self._pointwise_rerank(user_history, candidates, top_n)
    
    def _pointwise_rerank(
        self,
        user_history: List[Dict],
        candidates: List[Dict],
        top_n: int
    ) -> List[Dict]:
        """
        Pointwise 重排序：为每个候选物品单独评分
        """
        # 格式化用户历史
        history_str = self._format_history(user_history)
        
        # 为每个候选生成评分
        scored_candidates = []
        for candidate in candidates:
            score = self._score_candidate(history_str, candidate)
            scored_candidates.append({
                **candidate,
                "llm_score": score
            })
        
        # 按 LLM 分数排序
        scored_candidates.sort(key=lambda x: x["llm_score"], reverse=True)
        
        return scored_candidates[:top_n]
    
    def _pairwise_rerank(
        self,
        user_history: List[Dict],
        candidates: List[Dict],
        top_n: int
    ) -> List[Dict]:
        """
        Pairwise 重排序：通过两两比较进行排序
        """
        # 使用冒泡排序的思想，但用 LLM 进行比较
        n = len(candidates)
        ranked = candidates.copy()
        
        # 简化版：只进行有限次比较
        max_comparisons = min(n * (n - 1) // 2, 100)  # 限制比较次数
        
        comparison_count = 0
        for i in range(n - 1):
            if comparison_count >= max_comparisons:
                break
            
            swapped = False
            for j in range(n - i - 1):
                if comparison_count >= max_comparisons:
                    break
                
                # 比较两个候选
                better = self._compare_candidates(
                    user_history,
                    ranked[j],
                    ranked[j + 1]
                )
                
                comparison_count += 1
                
                if better == 1:  # ranked[j+1] 更好
                    ranked[j], ranked[j + 1] = ranked[j + 1], ranked[j]
                    swapped = True
            
            if not swapped:
                break
        
        return ranked[:top_n]
    
    def _listwise_rerank(
        self,
        user_history: List[Dict],
        candidates: List[Dict],
        top_n: int
    ) -> List[Dict]:
        """
        Listwise 重排序：一次性对所有候选进行排序
        """
        history_str = self._format_history(user_history)
        candidates_str = self._format_candidates(candidates)
        
        prompt = f"""你是一个专业的推荐系统。请根据用户的历史偏好，对以下候选物品进行排序。

用户历史：
{history_str}

候选物品：
{candidates_str}

请按照用户可能喜欢的程度，对这些物品进行排序（从最喜欢到最不喜欢）。

输出格式（ JSON）：
{{ 
    "ranking": [
        {{ "item_id": 1, "rank": 1 }},
        {{ "item_id": 2, "rank": 2 }},
        ...
    ]
 }}

只输出 JSON，不要输出其他内容。"""
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个专业的推荐系统排序器。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.1,  # 低温度保证一致性
                max_tokens=2000
            )
            
            content = response.choices[0].message.content.strip()
            
            # 解析 JSON
            if "```
{% endraw %}json" in content:
                content = content.split("```json")[1].split("```")[0].strip()
            elif "```" in content:
                content = content.split("```")[1].split("```")[0].strip()
            
            result = json.loads(content)
            ranking = result.get("ranking", [])
            
            # 构建 item_id 到 rank 的映射
            id_to_rank = {item["item_id"]: item["rank"] for item in ranking}
            
            # 按 rank 排序
            ranked_candidates = sorted(
                candidates,
                key=lambda x: id_to_rank.get(x.get("item_id", -1), 999)
            )
            
            return ranked_candidates[:top_n]
            
        except Exception as e:
            print(f"Listwise 重排序错误: {e}")
            # 降级到 pointwise
            return self._pointwise_rerank(user_history, candidates, top_n)
    
    def _score_candidate(self, history_str: str, candidate: Dict) -> float:
        """为单个候选物品评分"""
        prompt = f"""用户历史：
{history_str}

候选物品：
- 名称：《{candidate.get('item_name', '未知')}》
- 类别：{candidate.get('category', '')}
- 描述：{candidate.get('description', '')}

请评估用户对这个物品的喜欢程度，给出 0-10 分的分数。

只输出一个数字（ 0-10），不要输出其他内容。"""
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个专业的推荐评分器。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.1,
                max_tokens=10
            )
            
            score_str = response.choices[0].message.content.strip()
            score = float(score_str)
            return max(0, min(10, score))  # 限制在 0-10 范围
            
        except Exception as e:
            print(f"评分错误: {e}")
            return 5.0  # 默认分数
    
    def _compare_candidates(
        self,
        user_history: List[Dict],
        candidate1: Dict,
        candidate2: Dict
    ) -> int:
        """
        比较两个候选物品
        
        Returns:
            0: candidate1 更好
            1: candidate2 更好
        """
        history_str = self._format_history(user_history)
        
        prompt = f"""用户历史：
{history_str}

请比较以下两个物品，判断用户更可能喜欢哪一个：

物品 A：
- 名称：《{candidate1.get('item_name', '未知')}》
- 类别：{candidate1.get('category', '')}
- 描述：{candidate1.get('description', '')}

物品 B：
- 名称：《{candidate2.get('item_name', '未知')}》
- 类别：{candidate2.get('category', '')}
- 描述：{candidate2.get('description', '')}

只输出 "A" 或 "B"，表示用户更可能喜欢哪一个。"""
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个专业的推荐比较器。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.1,
                max_tokens=5
            )
            
            result = response.choices[0].message.content.strip().upper()
            if "A" in result:
                return 0
            elif "B" in result:
                return 1
            else:
                return 0  # 默认返回第一个
                
        except Exception as e:
            print(f"比较错误: {e}")
            return 0
    
    def _format_history(self, user_history: List[Dict]) -> str:
        """格式化用户历史"""
        history_str = "用户历史行为：\n"
        for i, item in enumerate(user_history[:10], 1):
            item_name = item.get('item_name', '未知')
            rating = item.get('rating', 'N/A')
            history_str += f"{i}. 《{item_name}》"
            if rating != 'N/A':
                history_str += f"（评分：{rating}/5）"
            history_str += "\n"
        return history_str
    
    def _format_candidates(self, candidates: List[Dict]) -> str:
        """格式化候选物品"""
        candidates_str = ""
        for i, item in enumerate(candidates, 1):
            item_name = item.get('item_name', '未知')
            category = item.get('category', '')
            description = item.get('description', '')
            item_id = item.get('item_id', i)
            
            candidates_str += f"{i}. [ID: {item_id}] 《{item_name}》"
            if category:
                candidates_str += f" - {category}"
            if description:
                candidates_str += f"\n   描述：{description}"
            candidates_str += "\n"
        return candidates_str


# 使用示例
if __name__ == "__main__":
    reranker = LLMReranker()
    
    # 用户历史
    user_history = [
        {"item_name": "肖申克的救赎", "rating": 5},
        {"item_name": "阿甘正传", "rating": 5},
        {"item_name": "当幸福来敲门", "rating": 4}
    ]
    
    # 候选物品（已粗排）
    candidates = [
        {"item_id": 1, "item_name": "美丽人生", "category": "剧情", "description": "感人的战争片"},
        {"item_id": 2, "item_name": "教父", "category": "犯罪", "description": "黑帮经典"},
        {"item_id": 3, "item_name": "海上钢琴师", "category": "剧情", "description": "音乐传记"},
        {"item_id": 4, "item_name": "楚门的世界", "category": "科幻", "description": "哲学思考"},
        {"item_id": 5, "item_name": "辛德勒的名单", "category": "历史", "description": "二战题材"}
    ]
    
    # 重排序
    reranked = reranker.rerank(
        user_history=user_history,
        candidates=candidates,
        top_n=3,
        strategy="listwise"
    )
    
    print("重排序结果：")
    for i, item in enumerate(reranked, 1):
        print(f"{i}. 《{item['item_name']}》")

LLM 作为重排序器的优势在于能够考虑复杂的语义匹配和用户意图，但需要注意 Token 消耗和延迟问题。

对话式推荐： ChatREC

对话式推荐（ Conversational Recommendation）允许用户通过自然语言对话与推荐系统交互，系统可以理解用户意图、询问澄清问题、提供推荐并解释理由。 ChatREC 是这一领域的代表性架构。

架构设计

ChatREC 的核心组件：

1. 对话管理器：管理多轮对话状态
2. 意图理解器：理解用户意图（搜索、浏览、澄清等）
3. 推荐引擎：基于对话历史生成推荐
4. 响应生成器：生成自然语言响应

实现代码

from typing import List, Dict, Optional, Tuple
from enum import Enum
import json
from openai import OpenAI
import os

class UserIntent(Enum):
    """用户意图类型"""
    SEARCH = "search"  # 搜索特定物品
    BROWSE = "browse"  # 浏览推荐
    CLARIFY = "clarify"  # 澄清需求
    COMPARE = "compare"  # 比较物品
    EXPLAIN = "explain"  # 询问推荐理由
    FEEDBACK = "feedback"  # 提供反馈


class ChatREC:
    """对话式推荐系统"""
    
    def __init__(self, model: str = "gpt-3.5-turbo"):
        """
        Args:
            model: LLM 模型名称
        """
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = model
        
        # 对话历史
        self.conversation_history: Dict[str, List[Dict]] = {}
        
        # 用户状态
        self.user_states: Dict[str, Dict] = {}
    
    def chat(self, user_id: str, message: str) -> str:
        """
        处理用户消息并生成响应
        
        Args:
            user_id: 用户 ID
            message: 用户消息
        
        Returns:
            系统响应
        """
        # 初始化对话历史
        if user_id not in self.conversation_history:
            self.conversation_history[user_id] = []
            self.user_states[user_id] = {
                "preferences": [],
                "current_recommendations": [],
                "context": {}
            }
        
        # 添加用户消息
        self.conversation_history[user_id].append({
            "role": "user",
            "content": message
        })
        
        # 理解用户意图
        intent = self._detect_intent(user_id, message)
        
        # 根据意图处理
        if intent == UserIntent.SEARCH:
            response = self._handle_search(user_id, message)
        elif intent == UserIntent.BROWSE:
            response = self._handle_browse(user_id)
        elif intent == UserIntent.CLARIFY:
            response = self._handle_clarify(user_id, message)
        elif intent == UserIntent.COMPARE:
            response = self._handle_compare(user_id, message)
        elif intent == UserIntent.EXPLAIN:
            response = self._handle_explain(user_id, message)
        elif intent == UserIntent.FEEDBACK:
            response = self._handle_feedback(user_id, message)
        else:
            response = self._handle_general(user_id, message)
        
        # 添加系统响应
        self.conversation_history[user_id].append({
            "role": "assistant",
            "content": response
        })
        
        return response
    
    def _detect_intent(self, user_id: str, message: str) -> UserIntent:
        """检测用户意图"""
        prompt = f"""分析以下用户消息的意图：

用户消息："{message}"

对话历史：
{self._format_history(self.conversation_history.get(user_id, []))}

可能的意图类型：
1. search - 搜索特定物品（如"我想看动作片"）
2. browse - 浏览推荐（如"给我推荐一些电影"）
3. clarify - 澄清需求（如"不要太暴力的"）
4. compare - 比较物品（如"这两个电影哪个更好"）
5. explain - 询问推荐理由（如"为什么推荐这个"）
6. feedback - 提供反馈（如"我不喜欢这个"）

只输出意图类型（ search/browse/clarify/compare/explain/feedback），不要输出其他内容。"""
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个意图识别系统。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.1,
                max_tokens=20
            )
            
            intent_str = response.choices[0].message.content.strip().lower()
            
            # 映射到枚举
            intent_map = {
                "search": UserIntent.SEARCH,
                "browse": UserIntent.BROWSE,
                "clarify": UserIntent.CLARIFY,
                "compare": UserIntent.COMPARE,
                "explain": UserIntent.EXPLAIN,
                "feedback": UserIntent.FEEDBACK
            }
            
            return intent_map.get(intent_str, UserIntent.BROWSE)
            
        except Exception as e:
            print(f"意图检测错误: {e}")
            return UserIntent.BROWSE
    
    def _handle_search(self, user_id: str, message: str) -> str:
        """处理搜索意图"""
        # 提取搜索关键词
        prompt = f"""从以下用户消息中提取搜索需求：

用户消息："{message}"

请提取：
1. 物品类型（如"电影"、"书籍"）
2. 关键词或特征（如"动作片"、"悬疑"）
3. 其他约束（如"不要太长"、"评分高的"）

输出 JSON 格式：
{{ 
    "item_type": "...",
    "keywords": ["..."],
    "constraints": ["..."]
}} """
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个需求提取系统。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.1,
                max_tokens=200
            )
            
            content = response.choices[0].message.content.strip()
            if "```json" in content:
                content = content.split("```json")[1].split("```")[0].strip()
            
            search_info = json.loads(content)
            
            # 更新用户状态
            self.user_states[user_id]["context"].update(search_info)
            
            # 生成推荐
            recommendations = self._generate_recommendations(user_id, search_info)
            
            # 格式化响应
            response_text = f"根据您的需求，我为您推荐以下内容：\n\n"
            for i, rec in enumerate(recommendations[:5], 1):
                response_text += f"{i}. 《{rec['item_name']}》\n"
                if rec.get('description'):
                    response_text += f"   {rec['description']}\n"
            
            return response_text
            
        except Exception as e:
            print(f"搜索处理错误: {e}")
            return "抱歉，我理解您的需求时遇到了问题。请再试一次。"
    
    def _handle_browse(self, user_id: str) -> str:
        """处理浏览意图"""
        # 基于用户历史生成推荐
        user_state = self.user_states[user_id]
        preferences = user_state.get("preferences", [])
        
        recommendations = self._generate_recommendations(user_id, {})
        
        response_text = "为您推荐以下内容：\n\n"
        for i, rec in enumerate(recommendations[:5], 1):
            response_text += f"{i}. 《{rec['item_name']}》\n"
            if rec.get('description'):
                response_text += f"   {rec['description']}\n"
        
        # 保存当前推荐
        self.user_states[user_id]["current_recommendations"] = recommendations[:5]
        
        return response_text
    
    def _handle_clarify(self, user_id: str, message: str) -> str:
        """处理澄清意图"""
        # 更新用户偏好
        user_state = self.user_states[user_id]
        
        # 提取澄清信息
        prompt = f"""从以下消息中提取用户的偏好或约束：

用户消息："{message}"

输出 JSON 格式：
{{ 
    "preferences": ["..."],
    "constraints": ["..."]
}} """
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个偏好提取系统。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.1,
                max_tokens=200
            )
            
            content = response.choices[0].message.content.strip()
            if "```json" in content:
                content = content.split("```json")[1].split("```")[0].strip()
            
            clarify_info = json.loads(content)
            
            # 更新偏好
            user_state["preferences"].extend(clarify_info.get("preferences", []))
            user_state["context"].update(clarify_info)
            
            # 重新生成推荐
            recommendations = self._generate_recommendations(user_id, user_state["context"])
            
            response_text = "好的，我理解了您的偏好。根据更新后的需求，为您推荐：\n\n"
            for i, rec in enumerate(recommendations[:5], 1):
                response_text += f"{i}. 《{rec['item_name']}》\n"
            
            self.user_states[user_id]["current_recommendations"] = recommendations[:5]
            
            return response_text
            
        except Exception as e:
            print(f"澄清处理错误: {e}")
            return "好的，我已经记下了您的偏好。"
    
    def _handle_explain(self, user_id: str, message: str) -> str:
        """处理解释意图"""
        current_recs = self.user_states[user_id].get("current_recommendations", [])
        preferences = self.user_states[user_id].get("preferences", [])
        
        if not current_recs:
            return "目前没有正在推荐的物品。请先让我为您推荐一些内容。"
        
        # 生成解释
        prompt = f"""用户偏好：
{json.dumps(preferences, ensure_ascii=False)}

当前推荐：
{json.dumps(current_recs[:3], ensure_ascii=False, indent=2)}

请解释为什么推荐这些物品。说明：
1. 与用户偏好的关联
2. 每个推荐的特点
3. 为什么用户可能会喜欢

用自然语言回答， 100-200 字。"""
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个推荐解释生成器。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.7,
                max_tokens=500
            )
            
            explanation = response.choices[0].message.content.strip()
            return explanation
            
        except Exception as e:
            print(f"解释生成错误: {e}")
            return "这些推荐是基于您的历史偏好和当前需求生成的。"
    
    def _handle_feedback(self, user_id: str, message: str) -> str:
        """处理反馈"""
        # 提取反馈信息
        prompt = f"""从以下消息中提取用户的反馈：

用户消息："{message}"

判断：
1. 是正面反馈（喜欢）还是负面反馈（不喜欢）
2. 针对哪个物品（如果有）
3. 具体原因（如果有）

输出 JSON：
{{ 
    "sentiment": "positive/negative",
    "item_name": "...",
    "reason": "..."
}} """
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个反馈分析系统。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.1,
                max_tokens=200
            )
            
            content = response.choices[0].message.content.strip()
            if "```json" in content:
                content = content.split("```json")[1].split("```")[0].strip()
            
            feedback = json.loads(content)
            
            # 更新用户偏好
            if feedback.get("sentiment") == "positive":
                if feedback.get("item_name"):
                    self.user_states[user_id]["preferences"].append({
                        "item": feedback["item_name"],
                        "type": "liked"
                    })
            else:
                if feedback.get("item_name"):
                    self.user_states[user_id]["preferences"].append({
                        "item": feedback["item_name"],
                        "type": "disliked"
                    })
            
            return "谢谢您的反馈！我会根据您的意见调整推荐。"
            
        except Exception as e:
            print(f"反馈处理错误: {e}")
            return "谢谢您的反馈！"
    
    def _handle_compare(self, user_id: str, message: str) -> str:
        """处理比较意图"""
        # 提取要比较的物品
        current_recs = self.user_states[user_id].get("current_recommendations", [])
        
        prompt = f"""从以下消息中提取用户要比较的物品：

用户消息："{message}"

当前推荐：
{json.dumps(current_recs, ensure_ascii=False, indent=2)}

输出要比较的物品名称列表（ JSON 数组）。"""
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个物品提取系统。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.1,
                max_tokens=100
            )
            
            content = response.choices[0].message.content.strip()
            if "[" in content:
                items = json.loads(content)
            else:
                items = [content]
            
            # 生成比较
            comparison_prompt = f"""请比较以下物品：

{json.dumps(items, ensure_ascii=False)}

从以下角度比较：
1. 类型和风格
2. 适合的观众
3. 优缺点

用自然语言回答， 150-250 字。"""
            
            comp_response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个物品比较系统。"},
                    {"role": "user", "content": comparison_prompt}
                ],
                temperature=0.7,
                max_tokens=500
            )
            
            comparison = comp_response.choices[0].message.content.strip()
            return comparison
            
        except Exception as e:
            print(f"比较处理错误: {e}")
            return "抱歉，我无法比较这些物品。"
    
    def _handle_general(self, user_id: str, message: str) -> str:
        """处理一般消息"""
        return "我是您的推荐助手。您可以：\n1. 搜索特定内容\n2. 浏览推荐\n3. 询问推荐理由\n4. 提供反馈\n\n 请告诉我您需要什么帮助？"
    
    def _generate_recommendations(self, user_id: str, context: Dict) -> List[Dict]:
        """生成推荐（简化版，实际应该调用推荐模型）"""
        # 这里是示例，实际应该调用真实的推荐模型
        return [
            {"item_name": "示例电影 1", "description": "示例描述 1"},
            {"item_name": "示例电影 2", "description": "示例描述 2"},
            {"item_name": "示例电影 3", "description": "示例描述 3"}
        ]
    
    def _format_history(self, history: List[Dict], max_turns: int = 5) -> str:
        """格式化对话历史"""
        recent_history = history[-max_turns * 2:]  # 最近几轮
        history_str = ""
        for msg in recent_history:
            role = msg["role"]
            content = msg["content"]
            history_str += f"{role}: {content}\n"
        return history_str


# 使用示例
if __name__ == "__main__":
    chatrec = ChatREC()
    
    user_id = "user123"
    
    # 第一轮：浏览推荐
    response1 = chatrec.chat(user_id, "给我推荐一些电影")
    print(f"系统: {response1}\n")
    
    # 第二轮：澄清需求
    response2 = chatrec.chat(user_id, "我想要感人的剧情片")
    print(f"系统: {response2}\n")
    
    # 第三轮：询问理由
    response3 = chatrec.chat(user_id, "为什么推荐这些？")
    print(f"系统: {response3}\n")

ChatREC 实现了自然语言交互的推荐系统，大大提升了用户体验和系统的灵活性。

RA-Rec：检索增强的推荐架构

RA-Rec（ Retrieval-Augmented Recommendation）结合了检索和生成的优势。它使用检索模块快速找到相关候选，然后使用 LLM 进行精细排序和解释生成。

架构设计

RA-Rec 包含以下组件：

1. 检索模块：使用向量检索快速找到相关候选
2. LLM 排序器：对检索结果进行精细排序
3. 解释生成器：生成推荐理由

实现代码

import numpy as np
from typing import List, Dict, Tuple
from sentence_transformers import SentenceTransformer
import faiss
from openai import OpenAI
import os

class RARec:
    """检索增强的推荐系统"""
    
    def __init__(
        self,
        embedding_model: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
        llm_model: str = "gpt-3.5-turbo"
    ):
        """
        Args:
            embedding_model: 用于检索的嵌入模型
            llm_model: 用于排序和解释的 LLM 模型
        """
        # 初始化嵌入模型
        self.embedding_model = SentenceTransformer(embedding_model)
        
        # 初始化 LLM
        self.llm_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.llm_model = llm_model
        
        # 物品索引
        self.item_index = None
        self.items = []
        self.item_embeddings = None
    
    def build_index(self, items: List[Dict]):
        """
        构建物品索引
        
        Args:
            items: 物品列表，每个物品包含 text, item_id 等字段
        """
        self.items = items
        
        # 提取物品文本
        texts = []
        for item in items:
            text_parts = []
            if 'title' in item:
                text_parts.append(item['title'])
            if 'description' in item:
                text_parts.append(item['description'])
            if 'category' in item:
                text_parts.append(item['category'])
            text = " | ".join(text_parts)
            texts.append(text)
        
        # 生成嵌入
        print("生成物品嵌入...")
        self.item_embeddings = self.embedding_model.encode(texts, show_progress_bar=True)
        
        # 构建 FAISS 索引
        dimension = self.item_embeddings.shape[1]
        self.item_index = faiss.IndexFlatIP(dimension)  # 内积索引（用于余弦相似度）
        
        # 归一化嵌入（用于余弦相似度）
        faiss.normalize_L2(self.item_embeddings)
        
        # 添加向量到索引
        self.item_index.add(self.item_embeddings.astype('float32'))
        
        print(f"索引构建完成，包含 {len(items)} 个物品")
    
    def retrieve(
        self,
        query: str,
        top_k: int = 50
    ) -> List[Dict]:
        """
        检索相关物品
        
        Args:
            query: 查询文本（用户需求或历史偏好）
            top_k: 检索数量
        
        Returns:
            检索到的物品列表
        """
        if self.item_index is None:
            raise ValueError("请先调用 build_index() 构建索引")
        
        # 生成查询嵌入
        query_embedding = self.embedding_model.encode([query])
        faiss.normalize_L2(query_embedding)
        
        # 检索
        distances, indices = self.item_index.search(
            query_embedding.astype('float32'),
            top_k
        )
        
        # 构建结果
        results = []
        for idx, dist in zip(indices[0], distances[0]):
            if idx < len(self.items):
                item = self.items[idx].copy()
                item['retrieval_score'] = float(dist)
                results.append(item)
        
        return results
    
    def rerank_with_llm(
        self,
        user_context: str,
        candidates: List[Dict],
        top_n: int = 10
    ) -> List[Dict]:
        """
        使用 LLM 对检索结果进行重排序
        
        Args:
            user_context: 用户上下文（历史偏好、当前需求等）
            candidates: 候选物品列表
            top_n: 最终推荐数量
        
        Returns:
            重排序后的物品列表
        """
        # 格式化候选
        candidates_str = ""
        for i, item in enumerate(candidates[:30], 1):  # 限制候选数量
            candidates_str += f"{i}. 《{item.get('title', '未知')}》"
            if item.get('description'):
                candidates_str += f" - {item['description'][:100]}"
            candidates_str += f" (检索分数: {item.get('retrieval_score', 0):.3f})\n"
        
        # 构建 Prompt
        prompt = f"""用户上下文：
{user_context}

检索到的候选物品：
{candidates_str}

请根据用户上下文，对这些候选物品进行排序。考虑：
1. 与用户需求的匹配度
2. 物品的质量和相关性
3. 多样性（避免推荐过于相似的内容）

输出排序后的物品编号列表（ JSON 格式）：
{{ 
    "ranking": [1, 5, 3, ...]
}} 

只输出 JSON，不要输出其他内容。"""
        
        try:
            response = self.llm_client.chat.completions.create(
                model=self.llm_model,
                messages=[
                    {"role": "system", "content": "你是一个专业的推荐排序器。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.1,
                max_tokens=500
            )
            
            content = response.choices[0].message.content.strip()
            if "```json" in content:
                content = content.split("```json")[1].split("```")[0].strip()
            
            result = json.loads(content)
            ranking = result.get("ranking", [])
            
            # 按排序结果重新组织
            reranked = []
            for rank in ranking[:top_n]:
                if 1 <= rank <= len(candidates):
                    reranked.append(candidates[rank - 1])
            
            # 如果 LLM 排序结果不足，补充剩余的候选
            if len(reranked) < top_n:
                remaining = [c for c in candidates if c not in reranked]
                reranked.extend(remaining[:top_n - len(reranked)])
            
            return reranked
            
        except Exception as e:
            print(f"LLM 重排序错误: {e}")
            # 降级到按检索分数排序
            return sorted(candidates, key=lambda x: x.get('retrieval_score', 0), reverse=True)[:top_n]
    
    def recommend(
        self,
        user_query: str,
        user_history: List[Dict] = None,
        top_n: int = 10
    ) -> List[Dict]:
        """
        完整的推荐流程：检索 + LLM 重排序
        
        Args:
            user_query: 用户查询
            user_history: 用户历史行为（可选）
            top_n: 推荐数量
        
        Returns:
            推荐结果列表
        """
        # 构建用户上下文
        context_parts = [f"当前需求：{user_query}"]
        if user_history:
            history_str = "历史偏好：\n"
            for item in user_history[:10]:
                history_str += f"- 《{item.get('item_name', '未知')}》\n"
            context_parts.append(history_str)
        user_context = "\n".join(context_parts)
        
        # 检索
        retrieved = self.retrieve(user_query, top_k=50)
        
        # LLM 重排序
        reranked = self.rerank_with_llm(user_context, retrieved, top_n)
        
        return reranked
    
    def explain_recommendation(
        self,
        user_context: str,
        recommended_item: Dict
    ) -> str:
        """生成推荐解释"""
        prompt = f"""用户上下文：
{user_context}

推荐物品：
- 名称：《{recommended_item.get('title', '未知')}》
- 描述：{recommended_item.get('description', '')}
- 类别：{recommended_item.get('category', '')}

请生成一段 100-150 字的推荐解释，说明为什么推荐这个物品。"""
        
        try:
            response = self.llm_client.chat.completions.create(
                model=self.llm_model,
                messages=[
                    {"role": "system", "content": "你是一个推荐解释生成器。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.7,
                max_tokens=300
            )
            
            explanation = response.choices[0].message.content.strip()
            return explanation
            
        except Exception as e:
            print(f"解释生成错误: {e}")
            return "基于您的偏好，我们为您推荐了这个物品。"


# 使用示例
if __name__ == "__main__":
    # 初始化系统
    rarec = RARec()
    
    # 构建索引
    items = [
        {"item_id": 1, "title": "肖申克的救赎", "description": "关于希望和友谊的经典", "category": "剧情"},
        {"item_id": 2, "title": "阿甘正传", "description": "智障人士的传奇人生", "category": "剧情"},
        # ... 更多物品
    ]
    rarec.build_index(items)
    
    # 推荐
    recommendations = rarec.recommend(
        user_query="我想看感人的剧情片",
        user_history=[
            {"item_name": "当幸福来敲门"}
        ],
        top_n=5
    )
    
    print("推荐结果：")
    for i, rec in enumerate(recommendations, 1):
        print(f"{i}. 《{rec['title']}》")
        explanation = rarec.explain_recommendation(
            "我想看感人的剧情片",
            rec
        )
        print(f"   理由：{explanation}\n")

RA-Rec 结合了检索的高效性和 LLM 的语义理解能力，在效果和效率之间取得了平衡。

ChatCRS：对话式推荐系统框架

ChatCRS（ Chat-based Conversational Recommendation System）是一个完整的对话式推荐系统框架，集成了意图理解、推荐生成、解释生成等功能。

架构特点

ChatCRS 的特点：

1. 多轮对话管理：维护对话状态和历史
2. 混合推荐策略：结合多种推荐方法
3. 自然语言生成：生成流畅的对话响应
4. 个性化适应：根据用户反馈调整推荐

核心实现

from typing import List, Dict, Optional
from dataclasses import dataclass
from datetime import datetime
import json

@dataclass
class ConversationTurn:
    """对话轮次"""
    user_message: str
    system_response: str
    intent: str
    timestamp: datetime
    recommendations: List[Dict] = None


class ChatCRS:
    """对话式推荐系统框架"""
    
    def __init__(self):
        self.conversations: Dict[str, List[ConversationTurn]] = {}
        self.user_profiles: Dict[str, Dict] = {}
    
    def process_message(
        self,
        user_id: str,
        message: str
    ) -> Dict:
        """
        处理用户消息
        
        Returns:
            包含响应、推荐、解释等的字典
        """
        # 初始化用户会话
        if user_id not in self.conversations:
            self.conversations[user_id] = []
            self.user_profiles[user_id] = {
                "preferences": [],
                "history": [],
                "feedback": []
            }
        
        # 理解意图
        intent = self._understand_intent(user_id, message)
        
        # 更新用户状态
        self._update_user_state(user_id, message, intent)
        
        # 生成推荐
        recommendations = self._generate_recommendations(user_id, intent)
        
        # 生成响应
        response = self._generate_response(user_id, message, intent, recommendations)
        
        # 生成解释
        explanations = self._generate_explanations(user_id, recommendations)
        
        # 记录对话
        turn = ConversationTurn(
            user_message=message,
            system_response=response,
            intent=intent,
            timestamp=datetime.now(),
            recommendations=recommendations
        )
        self.conversations[user_id].append(turn)
        
        return {
            "response": response,
            "recommendations": recommendations,
            "explanations": explanations,
            "intent": intent
        }
    
    def _understand_intent(self, user_id: str, message: str) -> str:
        """理解用户意图（简化版）"""
        # 实际应该使用 LLM 或分类器
        message_lower = message.lower()
        
        if any(word in message_lower for word in ["推荐", "给我", "想看"]):
            return "browse"
        elif any(word in message_lower for word in ["搜索", "找", "找找"]):
            return "search"
        elif any(word in message_lower for word in ["为什么", "理由", "原因"]):
            return "explain"
        elif any(word in message_lower for word in ["不喜欢", "讨厌", "不感兴趣"]):
            return "negative_feedback"
        elif any(word in message_lower for word in ["喜欢", "不错", "很好"]):
            return "positive_feedback"
        else:
            return "general"
    
    def _update_user_state(self, user_id: str, message: str, intent: str):
        """更新用户状态"""
        profile = self.user_profiles[user_id]
        
        if intent == "positive_feedback":
            # 提取喜欢的物品
            profile["preferences"].append({
                "type": "liked",
                "message": message,
                "timestamp": datetime.now()
            })
        elif intent == "negative_feedback":
            profile["preferences"].append({
                "type": "disliked",
                "message": message,
                "timestamp": datetime.now()
            })
    
    def _generate_recommendations(self, user_id: str, intent: str) -> List[Dict]:
        """生成推荐（简化版）"""
        # 实际应该调用推荐模型
        return [
            {"item_id": 1, "item_name": "示例 1", "score": 0.9},
            {"item_id": 2, "item_name": "示例 2", "score": 0.85},
            {"item_id": 3, "item_name": "示例 3", "score": 0.8}
        ]
    
    def _generate_response(
        self,
        user_id: str,
        message: str,
        intent: str,
        recommendations: List[Dict]
    ) -> str:
        """生成系统响应"""
        if intent == "browse":
            response = "为您推荐以下内容：\n\n"
            for i, rec in enumerate(recommendations[:5], 1):
                response += f"{i}. 《{rec['item_name']}》\n"
            return response
        elif intent == "search":
            return f"根据您的需求，我找到了以下相关内容：\n\n" + \
                   "\n".join([f"{i}. 《{r['item_name']}》" for i, r in enumerate(recommendations[:5], 1)])
        else:
            return "我理解您的需求，正在为您处理..."
    
    def _generate_explanations(
        self,
        user_id: str,
        recommendations: List[Dict]
    ) -> List[str]:
        """生成推荐解释"""
        # 实际应该使用 LLM 生成
        return [f"推荐《{r['item_name']}》是因为它符合您的偏好。" for r in recommendations]

ChatCRS 提供了一个完整的对话式推荐框架，可以根据具体需求进行扩展和定制。

Token 效率优化

LLM 在推荐系统中的应用面临一个关键挑战： Token 消耗。每次调用 LLM 都需要消耗大量 Token，成本高昂。需要优化策略来减少 Token 使用。

优化策略

1. 文本摘要

对于长文本（如物品描述、用户评论），先进行摘要再输入 LLM：

def summarize_text(text: str, max_length: int = 100) -> str:
    """摘要文本"""
    if len(text) <= max_length:
        return text
    
    # 使用 LLM 摘要（或简单的截断）
    # 这里简化处理
    sentences = text.split('。')
    summary = ""
    for sent in sentences:
        if len(summary) + len(sent) <= max_length:
            summary += sent + "。"
        else:
            break
    
    return summary if summary else text[:max_length] + "..."

2. 批量处理

将多个请求合并为批量请求，减少 API 调用次数：

def batch_recommend(queries: List[str], batch_size: int = 10) -> List[List[Dict]]:
    """批量推荐"""
    results = []
    
    for i in range(0, len(queries), batch_size):
        batch_queries = queries[i:i+batch_size]
        
        # 构建批量 Prompt
        batch_prompt = "请为以下用户生成推荐：\n\n"
        for j, query in enumerate(batch_queries):
            batch_prompt += f"用户{j+1}: {query}\n"
        
        # 调用 LLM（需要支持批量输出）
        # ...
    
    return results

3. 缓存机制

缓存常见查询的结果，避免重复调用 LLM：

from functools import lru_cache
import hashlib

class CachedLLMRecommender:
    """带缓存的 LLM 推荐器"""
    
    def __init__(self):
        self.cache = {}
    
    def _get_cache_key(self, query: str, context: Dict) -> str:
        """生成缓存键"""
        key_str = json.dumps({"query": query, "context": context}, sort_keys=True)
        return hashlib.md5(key_str.encode()).hexdigest()
    
    def recommend(self, query: str, context: Dict) -> List[Dict]:
        """带缓存的推荐"""
        cache_key = self._get_cache_key(query, context)
        
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        # 调用 LLM
        result = self._call_llm(query, context)
        
        # 缓存结果
        self.cache[cache_key] = result
        
        return result

4. 使用更小的模型

对于某些任务，可以使用更小的模型（如 GPT-3.5-turbo 而不是 GPT-4），在效果和成本之间平衡。

5. 结构化 Prompt

使用结构化的 Prompt 格式，减少冗余信息：

def build_efficient_prompt(user_history: List[Dict], candidates: List[Dict]) -> str:
    """构建高效的 Prompt"""
    # 只包含关键信息
    history_summary = "; ".join([item['name'] for item in user_history[:5]])
    candidates_list = "\n".join([f"{i}. {c['name']}" for i, c in enumerate(candidates[:20], 1)])
    
    prompt = f"""H: {history_summary}
C: {candidates_list}
R: Top 5"""
    
    return prompt

6. 两阶段策略

先用小模型/快速方法筛选，再用大模型精细处理：

def two_stage_recommend(user_query: str, all_items: List[Dict]) -> List[Dict]:
    """两阶段推荐"""
    # 第一阶段：快速筛选（使用向量检索）
    stage1_results = vector_search(user_query, all_items, top_k=50)
    
    # 第二阶段： LLM 精细排序（只处理少量候选）
    stage2_results = llm_rerank(user_query, stage1_results, top_k=10)
    
    return stage2_results

通过这些优化策略，可以显著降低 Token 消耗和成本，同时保持推荐效果。

完整代码示例：端到端 LLM 推荐系统

下面是一个完整的端到端 LLM 推荐系统实现，整合了前面提到的各种技术：

import torch
import torch.nn as nn
from transformers import AutoModel, AutoTokenizer
from typing import List, Dict, Optional, Tuple
import numpy as np
from dataclasses import dataclass
from datetime import datetime
import json
from openai import OpenAI
import os

@dataclass
class RecommendationResult:
    """推荐结果"""
    item_id: int
    item_name: str
    score: float
    explanation: str
    metadata: Dict = None


class EndToEndLLMRecommender:
    """端到端 LLM 推荐系统"""
    
    def __init__(
        self,
        embedding_model: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
        llm_model: str = "gpt-3.5-turbo",
        use_cache: bool = True
    ):
        """
        Args:
            embedding_model: 嵌入模型（用于检索）
            llm_model: LLM 模型（用于排序和解释）
            use_cache: 是否使用缓存
        """
        # 初始化组件
        from sentence_transformers import SentenceTransformer
        self.embedding_model = SentenceTransformer(embedding_model)
        
        self.llm_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.llm_model = llm_model
        
        # 数据存储
        self.items: List[Dict] = []
        self.item_embeddings: Optional[np.ndarray] = None
        self.item_index = None
        
        # 缓存
        self.cache = {} if use_cache else None
    
    def load_items(self, items: List[Dict]):
        """加载物品数据"""
        self.items = items
        
        # 生成嵌入
        texts = []
        for item in items:
            text = self._item_to_text(item)
            texts.append(text)
        
        print("生成物品嵌入...")
        self.item_embeddings = self.embedding_model.encode(texts, show_progress_bar=True)
        
        # 构建索引
        import faiss
        dimension = self.item_embeddings.shape[1]
        self.item_index = faiss.IndexFlatIP(dimension)
        faiss.normalize_L2(self.item_embeddings)
        self.item_index.add(self.item_embeddings.astype('float32'))
        
        print(f"加载了 {len(items)} 个物品")
    
    def _item_to_text(self, item: Dict) -> str:
        """将物品转换为文本"""
        parts = []
        if 'title' in item:
            parts.append(item['title'])
        if 'description' in item:
            parts.append(item['description'])
        if 'category' in item:
            parts.append(f"类别：{item['category']}")
        return " | ".join(parts)
    
    def recommend(
        self,
        user_query: str,
        user_history: Optional[List[Dict]] = None,
        top_k: int = 10,
        generate_explanation: bool = True
    ) -> List[RecommendationResult]:
        """
        生成推荐
        
        Args:
            user_query: 用户查询
            user_history: 用户历史行为
            top_k: 推荐数量
            generate_explanation: 是否生成解释
        
        Returns:
            推荐结果列表
        """
        # 检查缓存
        cache_key = self._get_cache_key(user_query, user_history)
        if self.cache and cache_key in self.cache:
            return self.cache[cache_key]
        
        # 第一阶段：检索
        retrieved = self._retrieve(user_query, top_k=50)
        
        # 第二阶段： LLM 重排序
        reranked = self._llm_rerank(user_query, user_history, retrieved, top_k)
        
        # 第三阶段：生成解释
        results = []
        for item in reranked:
            explanation = ""
            if generate_explanation:
                explanation = self._generate_explanation(user_query, user_history, item)
            
            result = RecommendationResult(
                item_id=item.get('item_id', 0),
                item_name=item.get('title', '未知'),
                score=item.get('llm_score', 0.0),
                explanation=explanation,
                metadata=item
            )
            results.append(result)
        
        # 缓存结果
        if self.cache:
            self.cache[cache_key] = results
        
        return results
    
    def _retrieve(self, query: str, top_k: int = 50) -> List[Dict]:
        """检索相关物品"""
        if self.item_index is None:
            raise ValueError("请先调用 load_items() 加载物品")
        
        # 生成查询嵌入
        query_embedding = self.embedding_model.encode([query])
        faiss.normalize_L2(query_embedding)
        
        # 检索
        distances, indices = self.item_index.search(
            query_embedding.astype('float32'),
            top_k
        )
        
        # 构建结果
        results = []
        for idx, dist in zip(indices[0], distances[0]):
            if idx < len(self.items):
                item = self.items[idx].copy()
                item['retrieval_score'] = float(dist)
                results.append(item)
        
        return results
    
    def _llm_rerank(
        self,
        query: str,
        user_history: Optional[List[Dict]],
        candidates: List[Dict],
        top_k: int
    ) -> List[Dict]:
        """使用 LLM 重排序"""
        # 构建上下文
        context = self._build_context(query, user_history)
        
        # 格式化候选
        candidates_str = self._format_candidates(candidates[:30])
        
        # 构建 Prompt
        prompt = f"""用户需求：{query}

{context}

候选物品：
{candidates_str}

请根据用户需求，对这些候选物品进行排序。输出排序后的编号列表（ JSON 格式）：
{{ "ranking": [1, 5, 3, ...]}} 

只输出 JSON 。"""
        
        try:
            response = self.llm_client.chat.completions.create(
                model=self.llm_model,
                messages=[
                    {"role": "system", "content": "你是一个专业的推荐排序器。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.1,
                max_tokens=500
            )
            
            content = response.choices[0].message.content.strip()
            if "```json" in content:
                content = content.split("```json")[1].split("```")[0].strip()
            
            result = json.loads(content)
            ranking = result.get("ranking", [])
            
            # 重新组织
            reranked = []
            for rank in ranking[:top_k]:
                if 1 <= rank <= len(candidates):
                    item = candidates[rank - 1].copy()
                    item['llm_score'] = 1.0 - (ranking.index(rank) / len(ranking))
                    reranked.append(item)
            
            # 补充不足的
            if len(reranked) < top_k:
                remaining = [c for c in candidates if c not in reranked]
                reranked.extend(remaining[:top_k - len(reranked)])
            
            return reranked
            
        except Exception as e:
            print(f"LLM 重排序错误: {e}")
            return candidates[:top_k]
    
    def _generate_explanation(
        self,
        query: str,
        user_history: Optional[List[Dict]],
        item: Dict
    ) -> str:
        """生成推荐解释"""
        context = self._build_context(query, user_history)
        
        prompt = f"""用户需求：{query}

{context}

推荐物品：《{item.get('title', '未知')}》
描述：{item.get('description', '')}

请生成一段 100-150 字的推荐解释。"""
        
        try:
            response = self.llm_client.chat.completions.create(
                model=self.llm_model,
                messages=[
                    {"role": "system", "content": "你是一个推荐解释生成器。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.7,
                max_tokens=300
            )
            
            return response.choices[0].message.content.strip()
            
        except Exception as e:
            print(f"解释生成错误: {e}")
            return "基于您的需求，我们为您推荐了这个物品。"
    
    def _build_context(self, query: str, user_history: Optional[List[Dict]]) -> str:
        """构建用户上下文"""
        context_parts = []
        
        if user_history:
            history_str = "用户历史：\n"
            for item in user_history[:10]:
                history_str += f"- 《{item.get('item_name', '未知')}》\n"
            context_parts.append(history_str)
        
        return "\n".join(context_parts) if context_parts else ""
    
    def _format_candidates(self, candidates: List[Dict]) -> str:
        """格式化候选物品"""
        candidates_str = ""
        for i, item in enumerate(candidates, 1):
            candidates_str += f"{i}. 《{item.get('title', '未知')}》"
            if item.get('description'):
                candidates_str += f" - {item['description'][:80]}"
            candidates_str += "\n"
        return candidates_str
    
    def _get_cache_key(self, query: str, user_history: Optional[List[Dict]]) -> str:
        """生成缓存键"""
        import hashlib
        key_data = {"query": query, "history": user_history}
        key_str = json.dumps(key_data, sort_keys=True, ensure_ascii=False)
        return hashlib.md5(key_str.encode()).hexdigest()


# 使用示例
if __name__ == "__main__":
    # 初始化推荐器
    recommender = EndToEndLLMRecommender()
    
    # 加载物品数据
    items = [
        {
            "item_id": 1,
            "title": "肖申克的救赎",
            "description": "关于希望和友谊的经典电影",
            "category": "剧情"
        },
        {
            "item_id": 2,
            "title": "阿甘正传",
            "description": "智障人士的传奇人生",
            "category": "剧情"
        },
        # ... 更多物品
    ]
    recommender.load_items(items)
    
    # 生成推荐
    results = recommender.recommend(
        user_query="我想看感人的剧情片",
        user_history=[
            {"item_name": "当幸福来敲门"}
        ],
        top_k=5,
        generate_explanation=True
    )
    
    # 输出结果
    print("推荐结果：\n")
    for i, result in enumerate(results, 1):
        print(f"{i}. 《{result.item_name}》")
        print(f"   分数：{result.score:.3f}")
        print(f"   理由：{result.explanation}\n")

常见问题解答（ Q&A）

Q1: LLM 推荐系统相比传统推荐系统有什么优势？

A: LLM 推荐系统的主要优势包括：

1. 语义理解能力： LLM 能够理解物品的文本描述、用户评论等语义信息，而传统方法主要依赖数值特征
2. 零样本能力：对于新物品或新用户， LLM 可以在没有训练数据的情况下进行推理
3. 可解释性： LLM 可以生成自然语言的推荐理由，提升用户体验和信任度
4. 跨域迁移： LLM 的预训练知识可以迁移到不同领域
5. 自然语言交互：支持对话式推荐，用户体验更好

但也要注意 LLM 的劣势：延迟高、成本高、需要大量 Token 。

Q2: 如何平衡 LLM 推荐的效果和效率？

A: 可以采用以下策略：

1. 混合架构：使用传统方法进行粗排， LLM 只用于重排序和解释生成
2. 检索增强：先用向量检索快速筛选候选，再用 LLM 精细处理
3. 缓存机制：缓存常见查询的结果
4. 批量处理：将多个请求合并处理
5. 模型选择：根据任务复杂度选择合适大小的模型（小任务用小模型）

Q3: LLM 推荐系统如何处理冷启动问题？

A: LLM 推荐系统在冷启动方面有明显优势：

1. 新物品冷启动：只需要提供物品的文本描述， LLM 就能理解其特性并做出推荐
2. 新用户冷启动：可以通过对话了解用户需求，或者使用 LLM 理解用户的自然语言描述
3. 零样本推荐： LLM 的预训练知识使其能够在没有领域特定数据的情况下工作

Q4: Prompt 工程在 LLM 推荐中有多重要？

A: Prompt 工程非常关键，直接影响推荐效果：

1. 结构化输入：清晰的组织用户历史和候选物品信息
2. 明确输出格式：指定 JSON 等格式，便于解析
3. Few-shot 示例：提供示例引导 LLM 输出正确格式
4. 约束条件：明确约束（如"不要推荐已看过的"）
5. 上下文管理：合理控制上下文长度，避免 Token 浪费

Q5: 如何评估 LLM 推荐系统的效果？

A: 评估可以从多个维度进行：

1. 离线指标：
   • 准确率（ Precision）、召回率（ Recall）
   • NDCG 、 MAP 等排序指标
   • 多样性、新颖性等指标
2. 在线指标：
   • 点击率（ CTR）
   • 转化率
   • 用户满意度
3. LLM 特定指标：
   • 解释质量（人工评估或自动评估）
   • 对话流畅度
   • Token 效率
4. A/B 测试：与基线系统对比，评估实际业务指标

Q6: LLM 推荐系统的成本如何控制？

A: 成本控制策略：

1. Token 优化：
   • 文本摘要，减少输入长度
   • 结构化 Prompt，避免冗余
   • 批量处理，提高效率
2. 缓存策略：
   • 缓存常见查询
   • 缓存物品特征（避免重复编码）
3. 模型选择：
   • 简单任务用小模型
   • 复杂任务用大模型
   • 考虑使用开源模型（如 LLaMA）
4. 架构优化：
   • 只在关键环节使用 LLM
   • 其他环节用传统方法

Q7: LLM 推荐系统如何处理用户隐私？

A: 隐私保护措施：

1. 数据脱敏：移除敏感信息（如真实姓名、地址）
2. 本地部署：使用开源模型本地部署，避免数据上传
3. 差分隐私：在训练或推理时添加噪声
4. 访问控制：限制对用户数据的访问
5. 数据加密：传输和存储时加密

Q8: 如何将 LLM 推荐系统集成到现有系统中？

A: 集成策略：

1. 渐进式集成：
   • 先作为补充模块（如解释生成）
   • 逐步扩展到更多环节
   • A/B 测试验证效果
2. API 封装：
   • 将 LLM 推荐封装为独立服务
   • 通过 API 调用，降低耦合
3. 降级策略：
   • LLM 服务失败时，降级到传统方法
   • 设置超时和重试机制
4. 监控和日志：
   • 监控延迟、错误率、 Token 消耗
   • 记录推荐结果，便于分析和优化

Q9: LLM 推荐系统在哪些场景下效果最好？

A: LLM 推荐系统在以下场景效果较好：

1. 文本丰富的领域：如书籍、电影、新闻等，有丰富的文本描述
2. 冷启动场景：新用户或新物品，缺乏历史数据
3. 需要解释的场景：用户希望理解推荐理由
4. 对话式交互：用户通过自然语言表达需求
5. 跨域推荐：需要在不同领域间迁移知识

但在以下场景可能不如传统方法： - 大规模实时推荐（延迟要求高） - 纯数值特征（如价格、评分） - 成本敏感的场景

Q10: 未来 LLM 推荐系统的发展方向是什么？

A: 未来发展方向：

1. 多模态融合：结合文本、图像、音频等多种模态
2. 个性化微调：为每个用户微调模型
3. 强化学习：使用 RL 优化长期用户满意度
4. 知识图谱集成：结合知识图谱提供更丰富的语义信息
5. 效率优化：模型压缩、量化、蒸馏等技术
6. 可解释性增强：更自然、更准确的解释生成
7. 隐私保护：联邦学习、差分隐私等技术

总结

LLM 为推荐系统带来了新的可能性，从简单的 Prompt-based 推荐到复杂的端到端架构，从特征增强到对话式交互， LLM 正在改变推荐系统的面貌。

关键要点：

1. LLM 的角色多样：可以是特征增强器、重排序器、解释生成器或端到端推荐器
2. 架构设计重要：需要平衡效果和效率，合理使用 LLM
3. Prompt 工程关键：好的 Prompt 能显著提升效果
4. 成本需要控制：通过缓存、批量处理、模型选择等策略降低成本
5. 评估要全面：不仅要看准确率，还要看解释质量、用户体验等

实践建议：

1. 从小规模开始，逐步扩展
2. 结合传统方法，发挥各自优势
3. 重视 Prompt 工程和上下文管理
4. 建立完善的监控和评估体系
5. 关注成本控制，确保可持续性

LLM 推荐系统仍处于快速发展阶段，新的架构和方法不断涌现。作为推荐系统工程师，需要持续学习，在实践中不断优化和改进。

希望这篇文章能帮助你理解 LLM 在推荐系统中的应用，并为你的实践提供参考。如果你有任何问题或想法，欢迎交流讨论！