智能体在推荐系统中的应用实践

# 推荐系统智能体示例 import numpy as np import torch import torch.nn as nn import torch.optim as optim # from torch.utils.data import Dataset import DataLoader from sklearn.metrics import roc_auc_score import precision_recall_curve import pandas as pd class RecommendationAgent: def __init__(self, num_users, num_items, embedding_dim=64, hidden_dims=[128, 64]): self.num_users = num_users self.num_items = num_items self.embedding_dim = embedding_dim self.hidden_dims = hidden_dims # 构建推荐模型 self.model = # DeepRecommendationModel( num_users, num_items, embedding_dim, hidden_dims ) self.optimizer = optim.Adam(self.model.parameters(), lr=0.001) self.criterion = nn.BCELoss() # 用户和物品嵌入 # self.user_embeddings = nn.Embedding(num_users, embedding_dim) self.item_embeddings = nn.Embedding(num_items, embedding_dim) # 推荐历史 self.recommendation_history = {} self.user_feedback = # {} def train(self, train_data, epochs=10, batch_size=256): """训练推荐模型""" train_loader = DataLoader(train_data, batch_size=batch_size, shuffle=True) for epoch in range(epochs): total_loss = 0 for batch in train_loader: user_ids, item_ids, ratings = batch # 前向传播 predictions = # self.model(user_ids, item_ids) loss = self.criterion(predictions, ratings.float()) # 反向传播 # self.optimizer.zero_grad() loss.backward() self.optimizer.step() total_loss += loss.item() print(f"Epoch {epoch+1}, Loss: {total_loss/len(train_loader):.4f}") def recommend(self, user_id, top_k=10, candidate_items=None): """为用户生成推荐""" if candidate_items is None: candidate_items = list(range(self.num_items)) user_tensor = torch.tensor([user_id] * len(candidate_items)) item_tensor = torch.tensor(candidate_items) with torch.no_grad(): scores = self.model(user_tensor, item_tensor) scores = scores.squeeze().numpy() # 获取top-k推荐 top_indices = np.argsort(scores)[-top_k:][::-1] # recommendations = [(candidate_items[i], scores[i]) for i in top_indices] # 记录推荐历史 # self.recommendation_history[user_id] = recommendations return recommendations def update_with_feedback(self, user_id, item_id, feedback): """根据用户反馈更新模型""" if user_id not in self.user_feedback: self.user_feedback[user_id] = [] self.user_feedback[user_id].append((item_id, feedback)) # 在线学习更新 self.online_update(user_id, item_id, feedback) def online_update(self, # user_id, item_id, feedback): """在线学习更新""" user_tensor = torch.tensor([user_id]) item_tensor = torch.tensor([item_id]) target = torch.tensor([feedback]) # 前向传播 prediction = # self.model(user_tensor, item_tensor) loss = self.criterion(prediction, target.float()) # 反向传播 # self.optimizer.zero_grad() loss.backward() self.optimizer.step() def evaluate(self, test_data): """评估推荐效果""" test_loader = DataLoader(test_data, batch_size=256, shuffle=False) all_predictions = [] all_targets = [] with torch.no_grad(): for batch in test_loader: user_ids, item_ids, ratings = batch predictions = self.model(user_ids, item_ids) all_predictions.extend(predictions.numpy()) all_targets.extend(ratings.numpy()) # 计算AUC auc = roc_auc_score(all_targets, all_predictions) return { # 'auc': auc, 'predictions': all_predictions, 'targets': all_targets } class DeepRecommendationModel(nn.Module): def __init__(self, num_users, num_items, embedding_dim, hidden_dims): super().__init__() self.num_users = num_users self.num_items = num_items self.embedding_dim = embedding_dim # 嵌入层 self.user_embedding = nn.Embedding(num_users, embedding_dim) # self.item_embedding = nn.Embedding(num_items, embedding_dim) # 深度网络 layers = [] input_dim = # embedding_dim * 2 for hidden_dim in hidden_dims: layers.append(nn.Linear(input_dim, hidden_dim)) layers.append(nn.ReLU()) layers.append(nn.Dropout(0.2)) input_dim = hidden_dim layers.append(nn.Linear(input_dim, 1)) layers.append(nn.Sigmoid()) self.network = nn.Sequential(*layers) def forward(self, user_ids, item_ids): user_emb = self.user_embedding(user_ids) item_emb = self.item_embedding(item_ids) # 拼接用户和物品嵌入 combined = torch.cat([user_emb, item_emb], dim=1)# 通过深度网络 output = self.network(combined) return output class MultiArmedBanditAgent: def __init__(self, num_arms, epsilon=0.1, alpha=0.1): self.num_arms = num_arms self.epsilon = epsilon# 探索率 # self.alpha = alpha # 学习率# 每个臂的统计信息 self.counts = np.zeros(num_arms) self.values = # np.zeros(num_arms) self.total_rewards = np.zeros(num_arms) def select_arm(self): """选择臂（推荐物品）""" if np.random.random() < self.epsilon: # 探索：随机选择 return np.random.randint(self.num_arms) else:# 利用：选择价值最高的臂 return np.argmax(self.values) def update(self, arm, reward): """更新臂的统计信息""" # self.counts[arm] += 1 self.total_rewards[arm] += reward # 更新价值估计 self.values[arm] = # self.total_rewards[arm] / self.counts[arm] def get_confidence_interval(self, arm, confidence=0.95): """获取置信区间""" if self.counts[arm] == 0: return (0, 1) mean = self.values[arm] std = np.sqrt(1.0 / self.counts[arm]) # 简化的置信区间计算 margin = 1.96 * std# 95%置信区间 return (max(0, mean - margin), # min(1, mean + margin)) class ContextualBanditAgent: def __init__(self, context_dim, num_arms, learning_rate=0.01): self.context_dim = context_dim self.num_arms = num_arms self.learning_rate = learning_rate # 每个臂的线性模型 self.weights = np.random.normal(0, 0.1, (num_arms, context_dim)) # self.counts = np.zeros(num_arms) def select_arm(self, context): """基于上下文选择臂""" scores = np.dot(self.weights, context) return np.argmax(scores) def update(self, context, arm, reward): """更新模型参数""" self.counts[arm] += 1 # 线性回归更新 prediction = np.dot(self.weights[arm], context) # error = reward - prediction # 梯度更新 self.weights[arm] += self.learning_rate * error * context class RecommendationDataset(Dataset): def __init__(self, user_ids, item_ids, ratings): self.user_ids = # torch.tensor(user_ids, dtype=torch.long) self.item_ids = torch.tensor(item_ids, dtype=torch.long) self.ratings = torch.tensor(ratings, dtype=torch.float) def __len__(self): return len(self.user_ids) def __getitem__(self, idx): return self.user_ids[idx], self.item_ids[idx], self.ratings[idx] # 使用示例 def main():# 模拟数据 num_users = 1000 num_items = 500 num_interactions = 10000# 生成随机数据 user_ids = # np.random.randint(0, num_users, num_interactions) item_ids = np.random.randint(0, num_items, num_interactions) ratings = np.random.randint(0, 2, num_interactions) # 0或1# 创建数据集 dataset = # RecommendationDataset(user_ids, item_ids, ratings) # 创建推荐智能体 agent = # RecommendationAgent(num_users, num_items) # 训练模型 agent.train(dataset, epochs=5)# 生成推荐 user_id = # 0 recommendations = agent.recommend(user_id, top_k=5) print(f"用户 {user_id} 的推荐:") for item_id, score in recommendations: print(f" 物品 {item_id}: {score:.3f}") # 多臂老虎机示例 bandit_agent = # MultiArmedBanditAgent(num_items) # 模拟推荐和反馈 for _ in range(100): arm = bandit_agent.select_arm() # reward = np.random.random() # 模拟奖励 bandit_agent.update(arm, reward) print(f"\n多臂老虎机结果:") # print(f"最佳臂: {np.argmax(bandit_agent.values)}") print(f"各臂价值: {bandit_agent.values[:10]}") if __name__ == "__main__": main()