AI驱动的个性化推荐系统完全教程

适用读者：后端工程师、算法工程师、产品经理、技术负责人
最后更新：2025 年 5 月
阅读时长：约 25 分钟

目录

1. 推荐系统概述
2. 经典推荐架构
3. LLM 增强推荐
4. Embedding 推荐
5. 实时推荐引擎
6. A/B 测试框架
7. 冷启动解决方案
8. 多目标优化
9. 推荐系统评估指标
10. 与 RAG 结合
11. 企业级推荐系统部署
12. 总结与展望

1. 推荐系统概述

推荐系统是互联网产品的核心引擎——Netflix 80% 的观看来自推荐，Amazon 35% 的收入源于推荐。一个优秀的推荐系统需要同时解决三个问题：

• 相关性：推荐用户真正感兴趣的内容
• 多样性：避免信息茧房，提供丰富体验
• 时效性：实时捕捉用户兴趣变化

1.1 推荐系统全景架构

┌─────────────────────────────────────────────────────────┐
│                    用户交互层                              │
│         App / Web / API / 消息推送                        │
└──────────────────────┬──────────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────────┐
│                  推荐服务层                                │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐ │
│  │ 召回层    │→│ 粗排层    │→│ 精排层    │→│ 重排层    │ │
│  │ Recall   │  │ Pre-Rank │  │  Rank    │  │ Re-Rank  │ │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘ │
└──────────────────────┬──────────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────────┐
│                  数据与特征层                              │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐ │
│  │ 用户画像  │  │ 物品特征  │  │ 交互数据  │  │ 上下文   │ │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘ │
└─────────────────────────────────────────────────────────┘

1.2 核心数据模型

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

@dataclass
class User:
    user_id: str
    demographics: Dict[str, str] = field(default_factory=dict)  # 年龄、性别、地区
    interests: List[str] = field(default_factory=list)
    behavior_history: List[Dict] = field(default_factory=list)
    embedding: Optional[List[float]] = None

@dataclass
class Item:
    item_id: str
    title: str
    category: str
    tags: List[str] = field(default_factory=list)
    features: Dict[str, float] = field(default_factory=dict)
    embedding: Optional[List[float]] = None
    publish_time: datetime = None
    popularity_score: float = 0.0

@dataclass
class Interaction:
    user_id: str
    item_id: str
    action: str  # view, click, like, share, purchase, skip
    timestamp: datetime
    context: Dict[str, str] = field(default_factory=dict)  # 设备、场景、时间
    duration: float = 0.0  # 停留时长
    rating: Optional[float] = None  # 显式评分

2. 经典推荐架构

2.1 协同过滤（Collaborative Filtering）

协同过滤是最经典的推荐算法，核心思想是"相似的用户喜欢相似的物品"。

基于用户的协同过滤

import numpy as np
from scipy.sparse import csr_matrix
from sklearn.metrics.pairwise import cosine_similarity

class UserBasedCF:
    def __init__(self, n_neighbors=20):
        self.n_neighbors = n_neighbors
        self.user_item_matrix = None
        self.user_similarity = None
    
    def fit(self, interactions):
        """训练：构建用户-物品矩阵并计算相似度"""
        # 构建稀疏矩阵
        users = list(set(i.user_id for i in interactions))
        items = list(set(i.item_id for i in interactions))
        self.user_idx = {u: i for i, u in enumerate(users)}
        self.item_idx = {it: i for i, it in enumerate(items)}
        self.idx_item = {i: it for it, i in self.item_idx.items()}
        
        rows, cols, vals = [], [], []
        for inter in interactions:
            rows.append(self.user_idx[inter.user_id])
            cols.append(self.item_idx[inter.item_id])
            vals.append(1.0 if inter.action in ("click", "like") else 0.5)
        
        self.user_item_matrix = csr_matrix(
            (vals, (rows, cols)),
            shape=(len(users), len(items))
        )
        
        # 计算用户相似度
        self.user_similarity = cosine_similarity(self.user_item_matrix)
    
    def recommend(self, user_id, top_k=10):
        """为用户生成推荐"""
        user_idx = self.user_idx[user_id]
        sim_scores = self.user_similarity[user_idx]
        
        # 找到最相似的 K 个用户
        neighbor_indices = np.argsort(sim_scores)[-self.n_neighbors:]
        
        # 加权聚合邻居的喜好
        scores = np.zeros(self.user_item_matrix.shape[1])
        for ni in neighbor_indices:
            scores += sim_scores[ni] * self.user_item_matrix[ni].toarray().flatten()
        
        # 排除已交互物品
        interacted = self.user_item_matrix[user_idx].toarray().flatten() > 0
        scores[interacted] = -1
        
        # 返回 Top-K
        top_indices = np.argsort(scores)[-top_k:][::-1]
        return [(self.idx_item[i], scores[i]) for i in top_indices]

基于物品的协同过滤

class ItemBasedCF:
    def __init__(self, n_neighbors=50):
        self.n_neighbors = n_neighbors
    
    def fit(self, interactions):
        """计算物品相似度矩阵"""
        # ... 矩阵构建同上
        self.item_similarity = cosine_similarity(self.user_item_matrix.T)
    
    def recommend(self, user_id, top_k=10):
        """基于用户历史物品推荐相似物品"""
        user_idx = self.user_idx[user_id]
        user_items = self.user_item_matrix[user_idx].toarray().flatten()
        
        scores = np.zeros(self.item_similarity.shape[0])
        for item_idx, rating in enumerate(user_items):
            if rating > 0:
                # 当前物品与所有物品的相似度，加权求和
                scores += self.item_similarity[item_idx] * rating
        
        scores[user_items > 0] = -1  # 排除已交互
        top_indices = np.argsort(scores)[-top_k:][::-1]
        return [(self.idx_item[i], scores[i]) for i in top_indices]

2.2 内容推荐（Content-Based）

基于物品本身的特征进行推荐，不依赖其他用户的行为数据。

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import linear_kernel

class ContentBasedRecommender:
    def __init__(self):
        self.tfidf = TfidfVectorizer(
            max_features=10000,
            stop_words="english",
            ngram_range=(1, 2)
        )
        self.item_vectors = None
        self.items = None
    
    def fit(self, items: List[Item]):
        """构建物品特征向量"""
        self.items = {item.item_id: item for item in items}
        
        # 将物品特征拼接为文本
        texts = [
            f"{item.title} {item.category} {' '.join(item.tags)}"
            for item in items
        ]
        self.item_vectors = self.tfidf.fit_transform(texts)
    
    def recommend(self, user_id, user_history: List[str], top_k=10):
        """基于用户历史偏好推荐"""
        if not user_history:
            return []
        
        # 用用户历史物品的平均向量表示用户兴趣
        history_indices = [
            i for i, item in enumerate(self.items.values())
            if item.item_id in user_history
        ]
        user_vector = self.item_vectors[history_indices].mean(axis=0)
        
        # 计算相似度
        scores = linear_kernel(user_vector, self.item_vectors).flatten()
        
        # 排除已交互
        for idx in history_indices:
            scores[idx] = -1
        
        top_indices = np.argsort(scores)[-top_k:][::-1]
        item_list = list(self.items.values())
        return [(item_list[i].item_id, scores[i]) for i in top_indices]

2.3 深度学习推荐模型

双塔模型（Two-Tower Model）

import torch
import torch.nn as nn

class TwoTowerModel(nn.Module):
    """双塔模型：用户塔和物品塔分别编码，点积计算匹配分数"""
    
    def __init__(self, user_feature_dim, item_feature_dim, embedding_dim=128):
        super().__init__()
        
        # 用户塔
        self.user_tower = nn.Sequential(
            nn.Linear(user_feature_dim, 256),
            nn.ReLU(),
            nn.BatchNorm1d(256),
            nn.Dropout(0.2),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, embedding_dim),
        )
        
        # 物品塔
        self.item_tower = nn.Sequential(
            nn.Linear(item_feature_dim, 256),
            nn.ReLU(),
            nn.BatchNorm1d(256),
            nn.Dropout(0.2),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, embedding_dim),
        )
    
    def encode_user(self, user_features):
        return self.user_tower(user_features)
    
    def encode_item(self, item_features):
        return self.item_tower(item_features)
    
    def forward(self, user_features, item_features):
        user_emb = self.encode_user(user_features)
        item_emb = self.encode_item(item_features)
        # 点积相似度
        scores = (user_emb * item_emb).sum(dim=-1)
        return scores

# 训练示例
model = TwoTowerModel(user_feature_dim=64, item_feature_dim=128)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.BCEWithLogitsLoss()

# 负采样 + 对比学习
for batch in dataloader:
    user_feat, pos_item_feat, neg_item_feat = batch
    
    pos_scores = model(user_feat, pos_item_feat)
    neg_scores = model(user_feat, neg_item_feat)
    
    # BPR Loss
    loss = -torch.log(torch.sigmoid(pos_scores - neg_scores)).mean()
    
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

3. LLM 增强推荐

大语言模型为推荐系统带来了理解能力和生成能力的质变。

3.1 LLM 作为推荐引擎

from openai import OpenAI

client = OpenAI()

def llm_recommend(user_profile, candidate_items, context=""):
    """直接使用 LLM 进行推荐"""
    
    prompt = f"""你是一个专业的个性化推荐系统。根据以下用户信息和候选商品，推荐最适合的5个商品。

## 用户画像
{user_profile}

## 当前上下文
{context}

## 候选商品
{chr(10).join(f"- {item['title']} | 类别: {item['category']} | 评分: {item['rating']}" for item in candidate_items)}

请按以下JSON格式返回推荐结果：
[{{"item_id": "xxx", "reason": "推荐理由"}}]
"""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
        response_format={"type": "json_object"}
    )
    
    return response.choices[0].message.content

3.2 LLM 增强特征工程

def extract_user_interests_llm(behavior_logs):
    """用 LLM 从用户行为日志中提取深层兴趣"""
    
    prompt = f"""分析以下用户行为日志，提取用户的深层兴趣偏好。

行为日志：
{chr(10).join(behavior_logs[:50])}

请输出：
1. 核心兴趣标签（5-10个）
2. 兴趣强度（1-10分）
3. 兴趣趋势（上升/稳定/下降）
4. 潜在兴趣（用户尚未表现但可能感兴趣的领域）

JSON格式输出。"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # 用小模型降低成本
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"}
    )
    
    return json.loads(response.choices[0].message.content)

def generate_item_description_llm(item_metadata):
    """用 LLM 为商品生成丰富描述，增强内容特征"""
    
    prompt = f"""基于以下商品信息，生成一段吸引人的推荐描述（100字以内），突出商品的核心卖点和适用场景。

商品信息：
- 名称：{item_metadata['title']}
- 类别：{item_metadata['category']}
- 参数：{json.dumps(item_metadata['specs'], ensure_ascii=False)}
- 评分：{item_metadata['rating']}/5

只输出描述文本。"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7
    )
    
    return response.choices[0].message.content

3.3 LLM 推荐的适用场景

4. Embedding 推荐

4.1 用户与物品 Embedding

import torch
from transformers import AutoModel, AutoTokenizer

class EmbeddingRecommender:
    """基于 Embedding 的推荐系统"""
    
    def __init__(self, model_name="sentence-transformers/all-MiniLM-L6-v2"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModel.from_pretrained(model_name)
        self.item_embeddings = {}
        self.user_embeddings = {}
    
    def encode_text(self, texts, batch_size=32):
        """文本编码为向量"""
        all_embeddings = []
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            inputs = self.tokenizer(
                batch, padding=True, truncation=True,
                max_length=512, return_tensors="pt"
            )
            with torch.no_grad():
                outputs = self.model(**inputs)
                # Mean pooling
                embeddings = outputs.last_hidden_state.mean(dim=1)
                all_embeddings.append(embeddings)
        return torch.cat(all_embeddings, dim=0)
    
    def index_items(self, items: List[Item]):
        """索引物品 Embedding"""
        texts = [
            f"{item.title} {item.category} {' '.join(item.tags)}"
            for item in items
        ]
        embeddings = self.encode_text(texts)
        
        for item, emb in zip(items, embeddings):
            self.item_embeddings[item.item_id] = emb.numpy()
    
    def recommend(self, user_id, top_k=10):
        """基于向量相似度推荐"""
        import faiss
        
        user_emb = self.user_embeddings[user_id]
        item_ids = list(self.item_embeddings.keys())
        item_matrix = np.array([self.item_embeddings[iid] for iid in item_ids])
        
        # FAISS 快速检索
        index = faiss.IndexFlatIP(item_matrix.shape[1])
        faiss.normalize_L2(item_matrix)
        index.add(item_matrix)
        
        query = user_emb.reshape(1, -1).astype('float32')
        faiss.normalize_L2(query)
        scores, indices = index.search(query, top_k)
        
        return [(item_ids[i], s) for i, s in zip(indices[0], scores[0])]

4.2 序列 Embedding 推荐

class SequentialEmbeddingModel(nn.Module):
    """基于 Transformer 的序列推荐模型"""
    
    def __init__(self, n_items, embedding_dim=128, n_heads=4, n_layers=2):
        super().__init__()
        self.item_embedding = nn.Embedding(n_items + 1, embedding_dim, padding_idx=0)
        self.position_embedding = nn.Embedding(50, embedding_dim)  # 最大序列长度50
        
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=embedding_dim, nhead=n_heads, dim_feedforward=512, dropout=0.1
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=n_layers)
        self.output_layer = nn.Linear(embedding_dim, n_items)
    
    def forward(self, item_sequence):
        """输入物品ID序列，预测下一个物品"""
        seq_len = item_sequence.size(1)
        positions = torch.arange(seq_len, device=item_sequence.device)
        
        x = self.item_embedding(item_sequence) + self.position_embedding(positions)
        
        # Causal mask（只看过去）
        mask = torch.triu(torch.ones(seq_len, seq_len), diagonal=1).bool()
        x = self.transformer(x.transpose(0, 1), mask=mask).transpose(0, 1)
        
        # 取最后一个位置的输出预测下一个物品
        logits = self.output_layer(x[:, -1, :])
        return logits

5. 实时推荐引擎

5.1 实时特征计算

import redis
import json
from datetime import datetime, timedelta

class RealTimeFeatureStore:
    """实时特征存储（基于 Redis）"""
    
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def update_user_action(self, user_id, item_id, action):
        """实时更新用户行为"""
        pipe = self.redis.pipeline()
        
        # 更新用户最近行为列表（保留最近100条）
        key = f"user:recent:{user_id}"
        pipe.lpush(key, json.dumps({
            "item_id": item_id,
            "action": action,
            "time": datetime.now().isoformat()
        }))
        pipe.ltrim(key, 0, 99)
        pipe.expire(key, 86400 * 30)  # 30天过期
        
        # 更新实时统计
        stat_key = f"user:stats:{user_id}"
        pipe.hincrby(stat_key, f"action:{action}", 1)
        pipe.hincrby(stat_key, "total_actions", 1)
        pipe.expire(stat_key, 86400 * 90)
        
        # 更新物品热度
        item_key = f"item:hot:{item_id}"
        pipe.zincrby("item:hot:global", 1, item_id)
        pipe.expire(item_key, 86400 * 7)
        
        pipe.execute()
    
    def get_user_realtime_features(self, user_id):
        """获取用户实时特征"""
        recent = self.redis.lrange(f"user:recent:{user_id}", 0, 19)
        stats = self.redis.hgetall(f"user:stats:{user_id}")
        
        return {
            "recent_items": [json.loads(r) for r in recent],
            "total_actions": int(stats.get(b"total_actions", 0)),
            "click_count": int(stats.get(b"action:click", 0)),
            "like_count": int(stats.get(b"action:like", 0)),
        }
    
    def get_trending_items(self, top_k=100):
        """获取实时热门物品"""
        return self.redis.zrevrange("item:hot:global", 0, top_k - 1, withscores=True)

5.2 流式推荐管道

from kafka import KafkaConsumer, KafkaProducer
import json

class StreamingRecommendationPipeline:
    """基于 Kafka 的实时推荐管道"""
    
    def __init__(self, feature_store, model_service):
        self.feature_store = feature_store
        self.model_service = model_service
        self.consumer = KafkaConsumer(
            'user_actions',
            bootstrap_servers=['localhost:9092'],
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            group_id='recommendation'
        )
        self.producer = KafkaProducer(
            bootstrap_servers=['localhost:9092'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
    
    def process_events(self):
        """消费用户行为事件，实时更新推荐"""
        for message in self.consumer:
            event = message.value
            user_id = event['user_id']
            item_id = event['item_id']
            action = event['action']
            
            # 1. 更新实时特征
            self.feature_store.update_user_action(user_id, item_id, action)
            
            # 2. 触发增量推荐更新
            if action in ('purchase', 'like', 'share'):
                # 高价值行为触发即时推荐更新
                new_recs = self.model_service.refresh_recommendations(user_id)
                self.producer.send('recommendation_updates', {
                    'user_id': user_id,
                    'recommendations': new_recs,
                    'trigger': action,
                    'timestamp': datetime.now().isoformat()
                })
            
            # 3. 更新物品热度
            if action == 'view':
                self.feature_store.update_item_popularity(item_id)

6. A/B 测试框架

6.1 实验分流系统

import hashlib
from dataclasses import dataclass
from typing import Dict, List, Optional

@dataclass
class Experiment:
    experiment_id: str
    name: str
    variants: Dict[str, float]  # variant_name -> traffic_ratio
    status: str = "running"  # running, paused, completed
    
class ABTestFramework:
    """A/B 测试分流框架"""
    
    def __init__(self):
        self.experiments: Dict[str, Experiment] = {}
    
    def create_experiment(self, experiment: Experiment):
        """创建实验"""
        assert abs(sum(experiment.variants.values()) - 1.0) < 1e-6, \
            "流量比例之和必须为 1"
        self.experiments[experiment.experiment_id] = experiment
    
    def assign_variant(self, experiment_id: str, user_id: str) -> str:
        """为用户分配实验组（确定性分流）"""
        exp = self.experiments[experiment_id]
        
        # 确定性哈希分流
        hash_input = f"{experiment_id}:{user_id}"
        hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
        bucket = (hash_value % 10000) / 10000.0
        
        cumulative = 0.0
        for variant_name, ratio in exp.variants.items():
            cumulative += ratio
            if bucket < cumulative:
                return variant_name
        
        return list(exp.variants.keys())[-1]
    
    def get_recommendation_strategy(self, user_id: str) -> dict:
        """根据用户所在的实验组返回推荐策略"""
        strategy = {}
        
        for exp_id, exp in self.experiments.items():
            if exp.status != "running":
                continue
            variant = self.assign_variant(exp_id, user_id)
            strategy[exp_id] = variant
        
        return strategy

# 使用示例
framework = ABTestFramework()

# 创建实验
framework.create_experiment(Experiment(
    experiment_id="exp_ranking_model",
    name="精排模型对比",
    variants={
        "control": 0.5,       # 对照组：现有模型
        "new_model": 0.3,     # 实验组A：新模型
        "ensemble": 0.2,      # 实验组B：集成模型
    }
))

framework.create_experiment(Experiment(
    experiment_id="exp_rerank_strategy",
    name="重排策略对比",
    variants={
        "control": 0.5,
        "diversity_boost": 0.5,
    }
))

# 用户分流
user_id = "user_12345"
strategy = framework.get_recommendation_strategy(user_id)
# {'exp_ranking_model': 'control', 'exp_rerank_strategy': 'diversity_boost'}

6.2 实验指标收集与分析

import numpy as np
from scipy import stats

class ExperimentAnalyzer:
    """实验结果分析器"""
    
    def __init__(self):
        self.metrics = {}  # experiment_id -> {variant -> [values]}
    
    def log_metric(self, experiment_id, variant, metric_name, value):
        """记录实验指标"""
        key = (experiment_id, variant, metric_name)
        if key not in self.metrics:
            self.metrics[key] = []
        self.metrics[key].append(value)
    
    def analyze(self, experiment_id, metric_name):
        """分析实验结果（两样本 t 检验）"""
        exp = self.experiments[experiment_id]
        results = {}
        
        # 获取对照组数据
        control_key = (experiment_id, "control", metric_name)
        control_data = self.metrics.get(control_key, [])
        
        for variant in exp.variants:
            if variant == "control":
                continue
            
            variant_key = (experiment_id, variant, metric_name)
            variant_data = self.metrics.get(variant_key, [])
            
            if not control_data or not variant_data:
                continue
            
            # t 检验
            t_stat, p_value = stats.ttest_ind(control_data, variant_data)
            
            control_mean = np.mean(control_data)
            variant_mean = np.mean(variant_data)
            lift = (variant_mean - control_mean) / control_mean * 100
            
            results[variant] = {
                "control_mean": control_mean,
                "variant_mean": variant_mean,
                "lift_pct": lift,
                "p_value": p_value,
                "significant": p_value < 0.05,
                "sample_size": len(variant_data),
            }
        
        return results

# 分析示例
analyzer = ExperimentAnalyzer()

# 假设已收集了足够数据
result = analyzer.analyze("exp_ranking_model", "ctr")
# {
#   "new_model": {
#     "control_mean": 0.032,
#     "variant_mean": 0.038,
#     "lift_pct": 18.75,
#     "p_value": 0.003,
#     "significant": True,
#     "sample_size": 50000
#   }
# }

7. 冷启动解决方案

冷启动是推荐系统最经典的难题之一。以下是分场景的解决方案。

7.1 新用户冷启动

class NewUserColdStart:
    """新用户冷启动策略"""
    
    def __init__(self, popularity_recommender, content_recommender, llm_client):
        self.popularity = popularity_recommender
        self.content = content_recommender
        self.llm = llm_client
    
    def recommend(self, user_context, stage="brand_new"):
        """
        stage:
        - brand_new: 完全新用户（0次交互）
        - early: 早期用户（1-5次交互）
        - warm: 暖用户（5-20次交互）
        """
        
        if stage == "brand_new":
            return self._brand_new_strategy(user_context)
        elif stage == "early":
            return self._early_user_strategy(user_context)
        else:
            return self._warm_user_strategy(user_context)
    
    def _brand_new_strategy(self, context):
        """全新用户：基于人口统计和热门内容"""
        # 1. 基于地区/年龄/性别的热门内容
        demographic_recs = self.popularity.recommend_by_demographic(
            region=context.get("region"),
            age_group=context.get("age_group"),
            gender=context.get("gender"),
            top_k=20
        )
        
        # 2. 全局热门（保底）
        trending_recs = self.popularity.get_trending(top_k=10)
        
        # 3. 多样性注入：确保覆盖多个类别
        diversified = self._diversify_results(
            demographic_recs + trending_recs,
            category_limit=3
        )
        
        return diversified[:10]
    
    def _early_user_strategy(self, context):
        """早期用户：结合少量行为 + 兴趣探索"""
        # 用少量行为推断兴趣
        user_interests = self._infer_interests_from_sparse(context["interactions"])
        
        # 70% 利用（exploit）+ 30% 探索（explore）
        exploit_recs = self.content.recommend_by_interests(user_interests, top_k=7)
        explore_recs = self._explore_new_categories(
            user_interests, 
            exclude=context["interactions"],
            top_k=3
        )
        
        return exploit_recs + explore_recs
    
    def _infer_interests_from_sparse(self, interactions):
        """从稀疏行为中推断兴趣"""
        if len(interactions) <= 3:
            # 行为太少，直接用类别统计
            categories = [i["category"] for i in interactions]
            return {"categories": list(set(categories))}
        
        # 用 LLM 理解少量行为的深层含义
        prompt = f"""基于以下{len(interactions)}条用户行为，推断用户兴趣：
{json.dumps(interactions, ensure_ascii=False)}
返回JSON格式的兴趣标签列表。"""
        
        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)

7.2 新物品冷启动

class NewItemColdStart:
    """新物品冷启动策略"""
    
    def __init__(self, content_model, embedding_model):
        self.content_model = content_model
        self.embedding_model = embedding_model
    
    def get_initial_exposure(self, new_item: Item, top_k_users=1000):
        """为新物品获取初始曝光"""
        
        # 1. 基于内容找到相似的已热门物品
        similar_items = self.content_model.find_similar(new_item, top_k=10)
        
        # 2. 找到喜欢这些相似物品的用户
        candidate_users = set()
        for sim_item in similar_items:
            users = self.get_item_fans(sim_item.item_id)
            candidate_users.update(users)
        
        # 3. 筛选高活跃度、高探索意愿的用户
        exploration_users = [
            u for u in candidate_users
            if self.get_user_exploration_score(u) > 0.7
        ]
        
        # 4. 分配初始流量
        return exploration_users[:top_k_users]
    
    def estimate_item_quality(self, new_item: Item, early_signals: dict):
        """基于早期信号预估物品质量"""
        
        # 早期信号权重
        weights = {
            "click_through_rate": 0.3,
            "completion_rate": 0.25,
            "like_rate": 0.2,
            "share_rate": 0.15,
            "comment_sentiment": 0.1,
        }
        
        quality_score = sum(
            early_signals.get(metric, 0) * weight
            for metric, weight in weights.items()
        )
        
        return quality_score

8. 多目标优化

现代推荐系统需要同时优化多个目标：点击率、停留时长、分享率、付费转化等。

8.1 多任务学习模型

class MultiTaskRecommender(nn.Module):
    """多任务推荐模型（MMoE 架构）"""
    
    def __init__(self, input_dim, n_experts=8, expert_dim=128, n_tasks=4):
        super().__init__()
        self.n_experts = n_experts
        self.n_tasks = n_tasks
        
        # 专家网络
        self.experts = nn.ModuleList([
            nn.Sequential(
                nn.Linear(input_dim, expert_dim),
                nn.ReLU(),
                nn.Linear(expert_dim, expert_dim),
            )
            for _ in range(n_experts)
        ])
        
        # 门控网络（每个任务一个）
        self.gates = nn.ModuleList([
            nn.Sequential(
                nn.Linear(input_dim, n_experts),
                nn.Softmax(dim=-1)
            )
            for _ in range(n_tasks)
        ])
        
        # 任务塔
        self.task_towers = nn.ModuleList([
            nn.Sequential(
                nn.Linear(expert_dim, 64),
                nn.ReLU(),
                nn.Linear(64, 1),
                nn.Sigmoid()
            )
            for _ in range(n_tasks)
        ])
        
        self.task_names = ["ctr", "duration", "share", "purchase"]
    
    def forward(self, x):
        # 计算所有专家输出
        expert_outputs = [expert(x) for expert in self.experts]
        expert_outputs = torch.stack(expert_outputs, dim=1)  # (batch, n_experts, dim)
        
        task_outputs = {}
        for i, task_name in enumerate(self.task_names):
            # 门控权重
            gate_weights = self.gates[i](x).unsqueeze(-1)  # (batch, n_experts, 1)
            
            # 加权聚合专家输出
            task_input = (expert_outputs * gate_weights).sum(dim=1)  # (batch, dim)
            
            # 任务塔预测
            task_outputs[task_name] = self.task_towers[i](task_input)
        
        return task_outputs

# 训练
model = MultiTaskRecommender(input_dim=256)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)

# 多任务损失（加权）
task_weights = {"ctr": 1.0, "duration": 0.5, "share": 0.3, "purchase": 0.8}

for batch in dataloader:
    features, labels = batch
    predictions = model(features)
    
    total_loss = 0
    for task_name, weight in task_weights.items():
        task_loss = nn.BCELoss()(predictions[task_name], labels[task_name])
        total_loss += weight * task_loss
    
    optimizer.zero_grad()
    total_loss.backward()
    optimizer.step()

8.2 多目标融合策略

class MultiObjectiveRanker:
    """多目标融合排序"""
    
    def __init__(self, model: MultiTaskRecommender):
        self.model = model
    
    def rank(self, user_features, candidate_items, business_goals):
        """
        business_goals: {
            "ctr_weight": 0.4,
            "duration_weight": 0.2,
            "share_weight": 0.1,
            "purchase_weight": 0.3,
        }
        """
        scores = []
        
        for item in candidate_items:
            # 模型预测各目标分数
            item_features = self._extract_features(user_features, item)
            predictions = self.model(item_features)
            
            # 加权融合
            final_score = sum(
                predictions[task].item() * business_goals.get(f"{task}_weight", 0)
                for task in self.model.task_names
            )
            
            # 业务规则调整
            final_score = self._apply_business_rules(final_score, item, user_features)
            
            scores.append((item, final_score))
        
        # 按分数排序
        scores.sort(key=lambda x: x[1], reverse=True)
        return scores
    
    def _apply_business_rules(self, score, item, user_context):
        """业务规则调整"""
        # 新品加权
        if item.is_new:
            score *= 1.2
        
        # 时间衰减
        hours_since_publish = (datetime.now() - item.publish_time).total_seconds() / 3600
        time_boost = max(0, 1 - hours_since_publish / 168)  # 7天内衰减
        score *= (1 + 0.1 * time_boost)
        
        # 多样性惩罚（同类别连续出现）
        if self._is_same_category_as_recent(item, user_context):
            score *= 0.8
        
        return score

9. 推荐系统评估指标

9.1 离线评估指标

import numpy as np
from sklearn.metrics import roc_auc_score, average_precision_score

class RecommenderEvaluator:
    """推荐系统评估器"""
    
    @staticmethod
    def precision_at_k(recommended, relevant, k):
        """Precision@K"""
        rec_k = recommended[:k]
        hits = len(set(rec_k) & set(relevant))
        return hits / k
    
    @staticmethod
    def recall_at_k(recommended, relevant, k):
        """Recall@K"""
        rec_k = recommended[:k]
        hits = len(set(rec_k) & set(relevant))
        return hits / len(relevant) if relevant else 0
    
    @staticmethod
    def ndcg_at_k(recommended, relevant, k):
        """NDCG@K（归一化折损累积增益）"""
        rec_k = recommended[:k]
        dcg = sum(
            1 / np.log2(i + 2) for i, item in enumerate(rec_k) if item in relevant
        )
        ideal_dcg = sum(1 / np.log2(i + 2) for i in range(min(len(relevant), k)))
        return dcg / ideal_dcg if ideal_dcg > 0 else 0
    
    @staticmethod
    def map_score(recommended, relevant):
        """MAP（平均精度均值）"""
        hits = 0
        sum_precision = 0
        for i, item in enumerate(recommended):
            if item in relevant:
                hits += 1
                sum_precision += hits / (i + 1)
        return sum_precision / len(relevant) if relevant else 0
    
    @staticmethod
    def coverage(all_recommendations, total_items):
        """覆盖率：推荐了多少比例的物品"""
        unique_items = set()
        for recs in all_recommendations:
            unique_items.update(recs)
        return len(unique_items) / total_items
    
    @staticmethod
    def diversity(recommended_items, item_embeddings):
        """多样性：推荐列表中物品之间的平均距离"""
        if len(recommended_items) < 2:
            return 0
        
        embeddings = [item_embeddings[iid] for iid in recommended_items]
        total_dist = 0
        count = 0
        for i in range(len(embeddings)):
            for j in range(i + 1, len(embeddings)):
                total_dist += np.linalg.norm(embeddings[i] - embeddings[j])
                count += 1
        
        return total_dist / count if count > 0 else 0
    
    @staticmethod
    def novelty(recommended_items, item_popularity):
        """新颖性：推荐了多少冷门物品"""
        scores = []
        for item_id in recommended_items:
            pop = item_popularity.get(item_id, 0)
            # 越冷门，新颖性越高
            scores.append(-np.log2(pop + 1e-10))
        return np.mean(scores)

# 综合评估
evaluator = RecommenderEvaluator()
results = {
    "precision@5": evaluator.precision_at_k(recs, relevant, 5),
    "recall@10": evaluator.recall_at_k(recs, relevant, 10),
    "ndcg@10": evaluator.ndcg_at_k(recs, relevant, 10),
    "coverage": evaluator.coverage(all_recs, total_items),
    "diversity": evaluator.diversity(recs, embeddings),
    "novelty": evaluator.novelty(recs, popularity),
}

9.2 在线评估指标

10. 与 RAG 结合

推荐系统与 RAG（检索增强生成）的结合是当前的热门方向。

10.1 RAG 增强推荐解释

class RAGEnhancedRecommender:
    """RAG 增强推荐：为推荐结果生成解释"""
    
    def __init__(self, retriever, llm, base_recommender):
        self.retriever = retriever  # 向量检索器
        self.llm = llm
        self.base_recommender = base_recommender
    
    def recommend_with_explanation(self, user_id, top_k=5):
        """推荐并生成个性化解释"""
        
        # 1. 基础推荐
        recommendations = self.base_recommender.recommend(user_id, top_k=top_k)
        
        # 2. 为每个推荐结果检索相关知识
        enhanced_recs = []
        for item in recommendations:
            # 检索与该物品和用户相关的内容
            context_docs = self.retriever.search(
                query=f"{item.title} {item.category}",
                filters={"type": "review"},
                top_k=3
            )
            
            # 3. 用 LLM 生成个性化推荐理由
            explanation = self._generate_explanation(
                user_id=user_id,
                item=item,
                context_docs=context_docs
            )
            
            enhanced_recs.append({
                "item": item,
                "explanation": explanation,
                "context": [doc.content for doc in context_docs]
            })
        
        return enhanced_recs
    
    def _generate_explanation(self, user_id, item, context_docs):
        """生成个性化推荐理由"""
        
        prompt = f"""基于以下信息，为用户生成一段简洁的推荐理由（50字以内）。

用户ID：{user_id}
推荐商品：{item.title}（{item.category}）
用户评价参考：
{chr(10).join(f"- {doc.content[:100]}" for doc in context_docs)}

要求：
1. 语气亲切自然
2. 突出商品与用户兴趣的匹配点
3. 可以引用真实用户评价
只输出推荐理由。"""
        
        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=100
        )
        
        return response.choices[0].message.content

10.2 知识增强推荐

class KnowledgeEnhancedRecommender:
    """知识图谱 + RAG 增强推荐"""
    
    def __init__(self, knowledge_graph, vector_store, llm):
        self.kg = knowledge_graph
        self.vector_store = vector_store
        self.llm = llm
    
    def recommend_with_knowledge(self, user_profile, candidates):
        """利用知识图谱增强推荐"""
        
        enriched_candidates = []
        for item in candidates:
            # 从知识图谱获取物品关系
            kg_info = self.kg.get_entity_relations(item.item_id)
            # 例如：商品 -> 品牌 -> 品类 -> 竞品
            # "iPhone 15" -> Apple -> 手机 -> [Samsung S24, Pixel 8]
            
            # 从向量库检索相关评价
            reviews = self.vector_store.search(
                query=f"{item.title} 评价",
                top_k=3
            )
            
            # LLM 综合分析
            analysis = self._analyze_item_fit(
                user_profile, item, kg_info, reviews
            )
            
            enriched_candidates.append({
                "item": item,
                "kg_relations": kg_info,
                "reviews": reviews,
                "fit_score": analysis["score"],
                "fit_reason": analysis["reason"]
            })
        
        # 按匹配度排序
        enriched_candidates.sort(key=lambda x: x["fit_score"], reverse=True)
        return enriched_candidates
    
    def _analyze_item_fit(self, user_profile, item, kg_info, reviews):
        prompt = f"""评估以下商品与用户的匹配度。

用户画像：{json.dumps(user_profile, ensure_ascii=False)}
商品：{item.title}（{item.category}）
商品关系：{json.dumps(kg_info, ensure_ascii=False)[:200]}
用户评价摘要：{chr(10).join(r.content[:80] for r in reviews[:3])}

请返回JSON：{{"score": 0-100的匹配分数, "reason": "简短理由"}}"""
        
        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)

11. 企业级推荐系统部署

11.1 微服务架构

┌──────────────────────────────────────────────────────────┐
│                      API Gateway                          │
│                    (Nginx / Kong)                         │
└──────────────┬───────────┬───────────┬───────────────────┘
               │           │           │
    ┌──────────▼──┐ ┌──────▼──────┐ ┌─▼──────────────┐
    │  推荐服务    │ │  用户服务    │ │  物品服务       │
    │  (gRPC)     │ │  (REST)     │ │  (REST)        │
    │  3 实例     │ │  2 实例     │ │  2 实例        │
    └──────┬──────┘ └──────┬──────┘ └───────┬────────┘
           │               │                │
    ┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼────────┐
    │  模型服务    │ │  特征服务    │ │  向量检索      │
    │  (Triton)   │ │  (Redis)    │ │  (Qdrant)     │
    └─────────────┘ └─────────────┘ └───────────────┘

11.2 容量规划

def estimate_recommendation_capacity(
    daily_active_users: int,
    avg_recs_per_user: int = 50,
    peak_factor: float = 3.0,
    latency_target_ms: int = 100
):
    """估算推荐系统容量需求"""
    
    # 日请求量
    daily_requests = daily_active_users * avg_recs_per_user
    
    # 峰值 QPS
    avg_qps = daily_requests / 86400
    peak_qps = avg_qps * peak_factor
    
    # 实例数（假设单实例 500 QPS）
    instance_qps = 500
    min_instances = max(3, int(peak_qps / instance_qps) + 1)
    
    # 内存需求（假设每用户特征 1KB）
    user_cache_gb = daily_active_users * 1024 / (1024**3)
    
    return {
        "daily_requests": f"{daily_requests:,}",
        "avg_qps": f"{avg_qps:.0f}",
        "peak_qps": f"{peak_qps:.0f}",
        "min_instances": min_instances,
        "user_cache_gb": f"{user_cache_gb:.1f}",
        "estimated_monthly_cost_usd": min_instances * 200 + user_cache_gb * 50,
    }

# 示例：百万 DAU 的推荐系统
capacity = estimate_recommendation_capacity(
    daily_active_users=1_000_000,
    avg_recs_per_user=50,
    peak_factor=3.0
)
# {
#   "daily_requests": "50,000,000",
#   "avg_qps": "579",
#   "peak_qps": "1736",
#   "min_instances": 4,
#   "user_cache_gb": "1.0",
#   "estimated_monthly_cost_usd": 850
# }

11.3 监控与告警

class RecommendationMonitor:
    """推荐系统监控"""
    
    METRICS = {
        "latency_p99": {"threshold": 200, "unit": "ms"},
        "error_rate": {"threshold": 0.01, "unit": "%"},
        "ctr": {"threshold": 0.02, "direction": "below"},
        "cache_hit_rate": {"threshold": 0.8, "direction": "below"},
        "qps": {"threshold": 5000, "direction": "above"},
    }
    
    def check_health(self, current_metrics):
        """健康检查"""
        alerts = []
        
        for metric, config in self.METRICS.items():
            value = current_metrics.get(metric)
            if value is None:
                continue
            
            if config.get("direction") == "below":
                if value < config["threshold"]:
                    alerts.append(f"⚠️ {metric} = {value}，低于阈值 {config['threshold']}")
            else:
                if value > config["threshold"]:
                    alerts.append(f"🔴 {metric} = {value}，超过阈值 {config['threshold']}")
        
        return {
            "status": "healthy" if not alerts else "degraded",
            "alerts": alerts,
            "metrics": current_metrics
        }

11.4 最佳实践清单

1. 分层召回：多路召回（协同过滤 + 内容 + 热门 + 向量）保证覆盖率
2. 特征一致性：训练和推理使用完全相同的特征工程管道
3. 实时特征：用 Redis/Flink 计算实时特征，避免特征穿越
4. 模型版本化：用 MLflow 管理模型版本，支持快速回滚
5. 降级策略：模型服务不可用时自动降级到规则推荐
6. 缓存策略：用户级缓存（30分钟）+ 物品级缓存（24小时）
7. 日志完备：记录每次推荐的完整上下文，支持离线分析
8. A/B 测试常态化：每个模型变更都经过 A/B 验证
9. 数据质量监控：实时检测数据异常（空值、分布漂移）
10. 成本控制：LLM 调用分级（高频用小模型，低频用大模型）

12. 总结与展望

核心要点回顾

未来趋势

1. LLM-Native 推荐：大模型从辅助角色变为核心引擎，直接理解用户意图并生成推荐
2. 多模态推荐：结合文本、图像、视频、音频的统一推荐
3. 联邦推荐：在保护隐私的前提下跨平台协作推荐
4. 实时个性化：从"用户画像驱动"转向"实时上下文驱动"
5. 可解释性：用户越来越需要理解"为什么推荐这个给我"

起步建议

如果你正在从零构建推荐系统，建议按以下路径推进：

1. 第 1 周：基于热门 + 规则推荐，快速上线
2. 第 2-4 周：接入协同过滤（Item-CF），建立 A/B 测试框架
3. 第 2-3 月：引入 Embedding 推荐，构建特征服务
4. 第 3-6 月：上线深度学习模型，支持多目标优化
5. 持续迭代：引入 LLM 增强，优化实时推荐能力

记住：推荐系统是数据驱动的系统，数据质量 > 模型复杂度。先把数据管道和评估体系建好，再逐步升级模型。

参考资源

• RecBole — 统一推荐算法框架
• TorchRec — PyTorch 推荐系统库
• Feature Store — 特征工程最佳实践
• 论文：Deep Learning based Recommender System: A Survey and New Perspectives (ACM Computing Surveys, 2019)