LLM与推荐系统¶

# TALLRec风格的推荐指令构造
def build_recommendation_instruction(user_history, candidate_item):
    """构造推荐指令格式的训练数据"""
    instruction = (
        "Based on the user's historical interactions, "
        "determine whether the user would like the candidate item."
    )

    # 格式化用户历史
    history_text = "User's watched movies: " + ", ".join(
        [f'"{movie}"' for movie in user_history]
    )

    # 格式化候选物品
    candidate_text = f'Candidate movie: "{candidate_item["title"]}" - {candidate_item["description"]}'

    input_text = f"{history_text}\n{candidate_text}"

    return {
        "instruction": instruction,
        "input": input_text,
        "output": "Yes" if candidate_item["label"] == 1 else "No"
    }

# 示例
user_history = ["星际穿越", "盗梦空间", "信条", "蝙蝠侠：黑暗骑士"]
candidate = {
    "title": "致命魔术",
    "description": "两位魔术师之间的竞争与复仇，充满悬疑反转",
    "label": 1
}
instruction_data = build_recommendation_instruction(user_history, candidate)
print(instruction_data)

from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, TaskType
import torch

def setup_lora_for_recommendation(model_name="meta-llama/Llama-2-7b-hf"):
    """配置LoRA微调推荐模型"""

    # 4-bit量化配置，降低显存需求
    bnb_config = BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_quant_type="nf4",
        bnb_4bit_compute_dtype=torch.float16,
        bnb_4bit_use_double_quant=True,
    )

    # 加载预训练模型
    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        quantization_config=bnb_config,
        device_map="auto",
        trust_remote_code=True,
    )

    # LoRA配置 —— 推荐任务建议的超参数
    lora_config = LoraConfig(
        task_type=TaskType.CAUSAL_LM,
        r=16,                     # LoRA秩，推荐任务通常16-64
        lora_alpha=32,            # 缩放因子
        lora_dropout=0.05,        # Dropout防止过拟合
        target_modules=[          # 对attention层做LoRA
            "q_proj", "k_proj", "v_proj", "o_proj",
            "gate_proj", "up_proj", "down_proj"
        ],
        bias="none",
    )

    # 应用LoRA
    model = get_peft_model(model, lora_config)
    model.print_trainable_parameters()
    # 输出: trainable params: ~4M || all params: ~7B || trainable%: 0.06%

    return model

# LoRA推荐模型的训练配置
from transformers import TrainingArguments

training_args = TrainingArguments(
    output_dir="./rec_lora_output",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    learning_rate=2e-4,            # LoRA推荐较高学习率
    warmup_ratio=0.1,
    logging_steps=10,
    save_strategy="epoch",
    fp16=True,
    optim="paged_adamw_32bit",
    report_to="wandb",
)

# === 推荐系统Prompt模板库 ===

class RecPromptTemplates:
    """推荐系统专用Prompt模板"""

    @staticmethod  # @staticmethod不需要实例即可调用
    def rating_prediction(user_history, target_item):
        """评分预测Prompt"""
        return f"""你是一个精准的推荐系统。请根据用户历史行为预测评分。

用户历史评分:
{chr(10).join([f'- {item["name"]}: {item["rating"]}分' for item in user_history])}

请预测用户对以下物品的评分(1-5分):
物品: {target_item["name"]}
描述: {target_item["description"]}

请直接输出评分数字:"""

    @staticmethod
    def sequential_recommendation(user_sequence, num_recs=5):
        """序列推荐Prompt"""
        return f"""你是一个序列推荐专家。根据用户的浏览历史，预测他接下来最可能感兴趣的物品。

用户浏览序列(按时间顺序):
{chr(10).join([f'{i+1}. {item}' for i, item in enumerate(user_sequence)])}  # enumerate同时获取索引和元素

请推荐{num_recs}个用户最可能感兴趣的物品，并简要说明推荐理由:"""

    @staticmethod
    def conversational_recommendation(dialog_history, user_query):
        """对话式推荐Prompt"""
        return f"""你是一个友好的推荐助手。根据对话上下文理解用户需求并推荐。

对话历史:
{chr(10).join(dialog_history)}

用户: {user_query}

请推荐并解释原因:"""

    @staticmethod
    def explainable_recommendation(user_profile, recommended_items):
        """可解释推荐Prompt"""
        return f"""你是一个推荐解释专家。请为以下推荐结果生成个性化的解释。

用户画像:
- 偏好类型: {user_profile["preferences"]}
- 近期关注: {user_profile["recent_interests"]}

推荐结果:
{chr(10).join([f'{i+1}. {item}' for i, item in enumerate(recommended_items)])}

请为每个推荐生成一句个性化的推荐理由:"""

import torch
import torch.nn as nn

class UniRecModel(nn.Module):  # 继承nn.Module定义网络层
    """UniRec风格的统一推荐模型（简化版）"""

    def __init__(self, num_items, id_dim=64, text_dim=768, hidden_dim=256):
        super().__init__()  # super()调用父类方法
        # ID Embedding分支
        self.id_embedding = nn.Embedding(num_items, id_dim)

        # Text Embedding分支（冻结的LLM编码器输出）
        self.text_projector = nn.Sequential(
            nn.Linear(text_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim)
        )

        # ID Embedding投影
        self.id_projector = nn.Sequential(
            nn.Linear(id_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim)
        )

        # 融合层 —— 自适应门控融合ID和Text信号
        self.fusion_gate = nn.Sequential(
            nn.Linear(hidden_dim * 2, hidden_dim),
            nn.Sigmoid()
        )

        # 预测头
        self.predictor = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, 1)
        )

    def forward(self, item_ids, text_embeddings, user_embedding):
        """
        item_ids: [batch_size] 物品ID
        text_embeddings: [batch_size, text_dim] 预计算的LLM文本向量
        user_embedding: [batch_size, hidden_dim] 用户表征
        """
        # 分别处理ID和Text特征
        id_feat = self.id_projector(self.id_embedding(item_ids))
        text_feat = self.text_projector(text_embeddings)

        # 自适应门控融合
        combined = torch.cat([id_feat, text_feat], dim=-1)  # torch.cat沿已有维度拼接张量
        gate = self.fusion_gate(combined)
        fused = gate * id_feat + (1 - gate) * text_feat

        # 用户-物品交互预测
        interaction = fused * user_embedding
        score = self.predictor(interaction)

        return score.squeeze(-1)  # squeeze压缩维度

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum

class DialogState(Enum):
    GREETING = "greeting"
    NEED_ELICITATION = "need_elicitation"
    PREFERENCE_CAPTURE = "preference_capture"
    RECOMMENDATION = "recommendation"
    FEEDBACK = "feedback"
    REFINEMENT = "refinement"

@dataclass  # @dataclass自动生成__init__等方法
class ConversationalRecState:
    """会话式推荐的状态管理"""
    current_state: DialogState = DialogState.GREETING
    user_preferences: Dict[str, list] = field(default_factory=dict)
    dialog_history: List[Dict[str, str]] = field(default_factory=list)
    recommended_items: List[str] = field(default_factory=list)
    rejected_items: List[str] = field(default_factory=list)
    satisfaction_score: float = 0.0
    turn_count: int = 0

    def update_preference(self, key: str, values: list):
        """更新用户偏好"""
        if key in self.user_preferences:
            self.user_preferences[key].extend(values)
        else:
            self.user_preferences[key] = values

    def add_dialog_turn(self, role: str, content: str):
        """添加对话轮次"""
        self.dialog_history.append({"role": role, "content": content})
        self.turn_count += 1

    def transition(self, new_state: DialogState):
        """状态转移"""
        self.current_state = new_state

    def get_context_summary(self) -> str:
        """获取当前上下文摘要"""
        prefs = ", ".join([f"{k}: {v}" for k, v in self.user_preferences.items()])
        return (
            f"对话轮次: {self.turn_count}\n"
            f"当前状态: {self.current_state.value}\n"
            f"用户偏好: {prefs}\n"
            f"已推荐: {len(self.recommended_items)}个\n"
            f"已拒绝: {len(self.rejected_items)}个"
        )

# === LLM推荐的工程优化策略 ===

class LLMRecServingOptimizer:
    """LLM推荐系统的Serving优化"""

    def __init__(self):
        self.strategies = {
            "离线预计算": {
                "description": "离线批量计算LLM Embedding，存入向量数据库",
                "延迟降低": "99%",
                "适用场景": "物品侧特征编码"
            },
            "模型蒸馏": {
                "description": "用大模型（GPT-4）蒸馏小模型（BERT-base）",
                "延迟降低": "90%",
                "适用场景": "在线排序"
            },
            "量化部署": {
                "description": "INT4/INT8量化 + vLLM/TensorRT-LLM",
                "延迟降低": "60-80%",
                "适用场景": "在线生成式推荐"
            },
            "语义缓存": {
                "description": "相似Query命中缓存，避免重复推理",
                "延迟降低": "70-90%",
                "适用场景": "对话式推荐"
            },
            "异步流水线": {
                "description": "LLM推理与传统推荐并行，取先返回结果",
                "延迟降低": "50%",
                "适用场景": "混合推荐系统"
            }
        }

    def select_strategy(self, latency_budget_ms, qps_requirement):
        """根据延迟预算和QPS要求选择优化策略"""
        if latency_budget_ms < 20:
            return "离线预计算"
        elif latency_budget_ms < 50:
            return "模型蒸馏"
        elif latency_budget_ms < 200:
            return "量化部署"
        elif qps_requirement < 100:
            return "语义缓存"
        else:
            return "异步流水线"

"""
Text-based Recommendation with Sentence Transformers
使用语义相似度做物品推荐
"""
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Dict

class TextBasedRecommender:
    """基于文本语义的推荐系统"""

    def __init__(self, model_name='sentence-transformers/all-MiniLM-L6-v2'):
        self.model = SentenceTransformer(model_name)
        self.item_catalog = []
        self.item_embeddings = None

    def build_item_index(self, items: List[Dict]):
        """
        构建物品索引
        items: [{"id": "1", "title": "...", "description": "..."}]
        """
        self.item_catalog = items
        texts = [f"{item['title']}. {item['description']}" for item in items]
        self.item_embeddings = self.model.encode(texts, normalize_embeddings=True)
        print(f"已索引 {len(items)} 个物品, Embedding维度: {self.item_embeddings.shape[1]}")

    def recommend_by_text(self, query: str, top_k: int = 5) -> List[Dict]:
        """根据文本查询推荐"""
        query_embedding = self.model.encode([query], normalize_embeddings=True)
        similarities = np.dot(self.item_embeddings, query_embedding.T).flatten()  # np.dot矩阵/向量点乘

        top_indices = np.argsort(similarities)[::-1][:top_k]

        results = []
        for idx in top_indices:
            results.append({
                "item": self.item_catalog[idx],
                "score": float(similarities[idx])
            })
        return results

    def recommend_by_history(self, user_history: List[Dict], top_k: int = 5) -> List[Dict]:
        """根据用户历史推荐（基于历史物品的聚合语义向量）"""
        history_texts = [f"{item['title']}. {item['description']}" for item in user_history]
        history_embeddings = self.model.encode(history_texts, normalize_embeddings=True)

        # 用户兴趣向量 = 历史物品Embedding的加权平均
        user_vector = np.mean(history_embeddings, axis=0, keepdims=True)
        user_vector = user_vector / np.linalg.norm(user_vector)  # np.linalg线性代数运算

        similarities = np.dot(self.item_embeddings, user_vector.T).flatten()

        # 排除已交互物品
        history_ids = {item.get("id") for item in user_history}
        for i, item in enumerate(self.item_catalog):
            if item["id"] in history_ids:
                similarities[i] = -1

        top_indices = np.argsort(similarities)[::-1][:top_k]

        results = []
        for idx in top_indices:
            results.append({
                "item": self.item_catalog[idx],
                "score": float(similarities[idx])
            })
        return results

# === 使用示例 ===
if __name__ == "__main__":
    recommender = TextBasedRecommender()

    # 构建物品库
    items = [
        {"id": "1", "title": "深度学习入门", "description": "从零开始学习神经网络和深度学习"},
        {"id": "2", "title": "推荐系统实践", "description": "工业级推荐系统的设计与实现"},
        {"id": "3", "title": "自然语言处理", "description": "NLP基础到大语言模型的完整教程"},
        {"id": "4", "title": "计算机视觉", "description": "图像识别、目标检测和图像分割"},
        {"id": "5", "title": "强化学习导论", "description": "马尔可夫决策过程和策略梯度方法"},
        {"id": "6", "title": "大模型微调", "description": "LoRA、QLoRA等高效微调技术实战"},
        {"id": "7", "title": "Transformer架构", "description": "注意力机制和Transformer模型详解"},
        {"id": "8", "title": "GNN图神经网络", "description": "图卷积网络在推荐和社交中的应用"},
    ]
    recommender.build_item_index(items)

    # 文本查询推荐
    print("=== 文本查询推荐 ===")
    results = recommender.recommend_by_text("我想学习大模型相关技术")
    for r in results:
        print(f"  {r['item']['title']}: {r['score']:.4f}")

    # 基于历史推荐
    print("\n=== 基于历史推荐 ===")
    user_history = [items[0], items[2], items[6]]  # 深度学习、NLP、Transformer
    results = recommender.recommend_by_history(user_history)
    for r in results:
        print(f"  {r['item']['title']}: {r['score']:.4f}")

"""
使用LoRA微调LLaMA-2做推荐任务
简化的训练Pipeline
"""
import torch
from datasets import Dataset
from transformers import (
    AutoModelForCausalLM, AutoTokenizer,
    TrainingArguments, BitsAndBytesConfig
)
from peft import LoraConfig, get_peft_model
from trl import SFTTrainer

def prepare_recommendation_dataset():
    """准备推荐数据集（MovieLens风格）"""
    examples = [
        {
            "text": (
                "### 指令: 根据用户观影历史，判断用户是否会喜欢候选电影。\n"
                "### 用户历史: 肖申克的救赎(5分), 教父(5分), 辛德勒的名单(4分)\n"
                "### 候选电影: 美丽人生 - 二战背景的感人故事\n"
                "### 回答: 是。该用户偏好高评分经典剧情片，《美丽人生》同为高口碑经典剧情片，非常匹配。"
            )
        },
        {
            "text": (
                "### 指令: 根据用户观影历史，判断用户是否会喜欢候选电影。\n"
                "### 用户历史: 复仇者联盟(4分), 蜘蛛侠(4分), 钢铁侠(5分)\n"
                "### 候选电影: 傲慢与偏见 - 简·奥斯汀经典文学改编\n"
                "### 回答: 否。该用户明显偏好漫威超级英雄类电影，文学改编剧情片与其偏好不匹配。"
            )
        },
        # ... 实际需要数千条训练数据
    ]
    return Dataset.from_list(examples)

def train_rec_llama():
    """微调LLaMA做推荐"""
    model_name = "meta-llama/Llama-2-7b-hf"

    # 量化配置
    bnb_config = BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_quant_type="nf4",
        bnb_4bit_compute_dtype=torch.float16,
    )

    # 加载模型和分词器
    model = AutoModelForCausalLM.from_pretrained(
        model_name, quantization_config=bnb_config, device_map="auto"
    )
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    tokenizer.pad_token = tokenizer.eos_token

    # LoRA配置
    lora_config = LoraConfig(
        r=16, lora_alpha=32, lora_dropout=0.05,
        target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
        task_type="CAUSAL_LM"
    )
    model = get_peft_model(model, lora_config)

    # 训练配置
    training_args = TrainingArguments(
        output_dir="./rec-llama-lora",
        num_train_epochs=3,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        learning_rate=2e-4,
        fp16=True,
        logging_steps=10,
        save_strategy="epoch",
        warmup_ratio=0.1,
    )

    # 准备数据
    dataset = prepare_recommendation_dataset()

    # 使用SFTTrainer训练
    trainer = SFTTrainer(
        model=model,
        args=training_args,
        train_dataset=dataset,
        tokenizer=tokenizer,
        max_seq_length=512,
    )

    trainer.train()
    trainer.save_model("./rec-llama-lora-final")
    print("推荐模型微调完成！")

def inference_rec_llama(model_path, user_history, candidate):
    """推荐推理"""
    from peft import PeftModel

    base_model = AutoModelForCausalLM.from_pretrained(
        "meta-llama/Llama-2-7b-hf", device_map="auto", torch_dtype=torch.float16
    )
    model = PeftModel.from_pretrained(base_model, model_path)
    tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-2-7b-hf")

    prompt = (
        f"### 指令: 根据用户观影历史，判断用户是否会喜欢候选电影。\n"
        f"### 用户历史: {user_history}\n"
        f"### 候选电影: {candidate}\n"
        f"### 回答: "
    )

    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
    with torch.no_grad():  # 禁用梯度计算，节省内存
        outputs = model.generate(**inputs, max_new_tokens=100, temperature=0.7)

    response = tokenizer.decode(outputs[0], skip_special_tokens=True)
    return response.split("### 回答: ")[-1]

"""
基于LangChain的对话式推荐Agent
支持工具调用、多轮对话、个性化推荐
"""
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage
from typing import List
import json

# === 定义推荐工具 ===

@tool
def search_items(query: str, category: str = "all") -> str:
    """搜索物品库，返回匹配的物品列表"""
    # 模拟物品搜索（实际项目中对接真实搜索引擎或向量数据库）
    mock_results = {
        "科幻电影": ["沙丘2 (2024)", "奥本海默 (2023)", "流浪地球2 (2023)"],
        "动作电影": ["碟中谍7 (2023)", "疾速追杀4 (2023)", "速度与激情10 (2023)"],
        "书籍推荐": ["《大模型时代》", "《芯片战争》", "《人类简史》"],
    }
    for key, items in mock_results.items():
        if any(k in query for k in key):  # any()任一为True则返回True
            return json.dumps({"results": items, "total": len(items)}, ensure_ascii=False)  # json.dumps将Python对象序列化为JSON字符串
    return json.dumps({"results": ["暂无匹配结果"], "total": 0}, ensure_ascii=False)

@tool
def get_user_profile(user_id: str) -> str:
    """获取用户画像信息"""
    profiles = {
        "user_001": {
            "偏好类型": ["科幻", "悬疑", "纪录片"],
            "年龄段": "25-30",
            "活跃度": "高",
            "平均评分": 4.2,
            "最近观看": ["三体(剧集)", "星际穿越", "黑暗荣耀"]
        }
    }
    profile = profiles.get(user_id, {"error": "用户不存在"})
    return json.dumps(profile, ensure_ascii=False)

@tool
def get_item_details(item_name: str) -> str:
    """获取物品详细信息"""
    details = {
        "沙丘2": {
            "评分": 8.5, "类型": "科幻/冒险",
            "导演": "丹尼斯·维伦纽瓦", "年份": 2024,
            "简介": "保罗·厄崔迪与弗里曼人联合对抗哈克南家族"
        }
    }
    detail = details.get(item_name, {"error": "物品信息未收录"})
    return json.dumps(detail, ensure_ascii=False)

@tool
def collaborative_filtering_recommend(user_id: str, top_k: int = 5) -> str:
    """基于协同过滤的推荐（调用传统推荐模型）"""
    # 模拟CF推荐结果
    cf_results = [
        {"item": "银翼杀手2049", "score": 0.95},
        {"item": "降临", "score": 0.91},
        {"item": "头号玩家", "score": 0.88},
        {"item": "攻壳机动队", "score": 0.85},
        {"item": "黑客帝国", "score": 0.82},
    ]
    return json.dumps(cf_results[:top_k], ensure_ascii=False)

# === 构建推荐Agent ===

def create_rec_agent():
    """创建对话式推荐Agent"""

    llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

    tools = [search_items, get_user_profile, get_item_details,
             collaborative_filtering_recommend]

    system_prompt = """你是一个专业的个性化推荐助手。你的职责是：
1. 通过对话理解用户的真实需求和偏好
2. 利用工具获取用户画像和物品信息
3. 结合协同过滤推荐和语义理解给出个性化推荐
4. 解释推荐理由，帮助用户做出决策
5. 根据用户反馈动态调整推荐策略

注意事项：
- 推荐时要说明推荐理由
- 如果用户不满意，主动询问偏好并调整
- 控制推荐数量（一般3-5个）
- 优先推荐与用户历史偏好匹配的物品"""

    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ])

    agent = create_openai_tools_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    return executor

# === 运行对话式推荐 ===

def run_conversational_recommendation():
    """运行对话式推荐Demo"""
    agent = create_rec_agent()
    chat_history = []

    print("=== 对话式推荐系统 ===")
    print("输入 'quit' 退出\n")

    while True:
        user_input = input("你: ")
        if user_input.lower() == 'quit':
            break

        response = agent.invoke({
            "input": user_input,
            "chat_history": chat_history,
        })

        ai_response = response["output"]
        print(f"\n推荐助手: {ai_response}\n")

        # 更新对话历史
        chat_history.append(HumanMessage(content=user_input))
        chat_history.append(AIMessage(content=ai_response))

if __name__ == "__main__":
    run_conversational_recommendation()