TensorFlow-v2.9实战案例：推荐系统上线全流程详解
·
TensorFlow-v2.9实战案例：推荐系统上线全流程详解
推荐系统，这个听起来高大上的技术，其实离我们很近。当你打开购物网站，首页上那些“猜你喜欢”的商品；当你刷短视频，源源不断推送你感兴趣的内容；甚至当你点外卖，平台推荐的餐厅和套餐——背后都有推荐系统的影子。
对于很多开发者来说，搭建一个推荐系统原型并不难，但如何把它从实验室的Jupyter Notebook变成一个稳定、高效、能服务真实用户的生产系统，却是一个充满挑战的过程。今天，我就以TensorFlow 2.9为例，带你走一遍推荐系统从开发到上线的完整流程。
1. 项目准备与环境搭建
在开始任何机器学习项目之前，明确目标和准备好环境是第一步。
1.1 明确业务目标与数据
推荐系统不是凭空产生的，它需要解决具体的业务问题。在动手之前，我们需要问自己几个关键问题：
- 我们要推荐什么？ 是商品、视频、新闻还是音乐？
- 推荐的目标是什么？ 是提高点击率、增加购买转化，还是提升用户停留时间？
- 我们有什么数据？ 用户行为数据、物品属性数据、用户画像数据是否齐全？
以电商场景为例，我们可能有以下数据：
- 用户行为数据：用户ID、商品ID、行为类型（点击、购买、收藏）、时间戳
- 商品数据：商品ID、类别、价格、品牌、上架时间
- 用户数据：用户ID、年龄、性别、地域、注册时间
1.2 TensorFlow 2.9环境快速部署
TensorFlow 2.9提供了一个稳定且功能完整的深度学习平台。我们可以通过CSDN星图镜像快速搭建开发环境。
一键启动Jupyter开发环境：
- 在镜像广场选择TensorFlow-v2.9镜像
- 点击“立即创建”启动实例
- 等待实例状态变为“运行中”
- 点击“访问地址”中的Jupyter链接
进入Jupyter界面后，你会看到一个完整的Python开发环境，TensorFlow 2.9及其常用依赖都已经预装好了。我们可以通过以下代码验证环境：
import tensorflow as tf
import numpy as np
import pandas as pd
print(f"TensorFlow版本: {tf.__version__}")
print(f"GPU是否可用: {tf.config.list_physical_devices('GPU')}")
如果一切正常，你会看到TensorFlow 2.9.x的版本信息。对于推荐系统这种计算密集型任务，如果有GPU可用，训练速度会大大提升。
2. 推荐模型开发实战
有了环境和数据，我们就可以开始构建推荐模型了。这里我以一个简化的电商推荐场景为例，展示从数据处理到模型训练的全过程。
2.1 数据预处理与特征工程
推荐系统的质量很大程度上取决于数据的质量。我们需要将原始数据转换成模型能够理解的格式。
import pandas as pd
from sklearn.preprocessing import LabelEncoder, MinMaxScaler
# 加载示例数据
def load_sample_data():
"""生成模拟的电商用户行为数据"""
np.random.seed(42)
# 模拟1000个用户，5000个商品
n_users = 1000
n_items = 5000
n_interactions = 50000
# 生成交互数据
user_ids = np.random.randint(0, n_users, n_interactions)
item_ids = np.random.randint(0, n_items, n_interactions)
# 模拟不同的行为类型：1-浏览，2-收藏，3-加购，4-购买
behaviors = np.random.choice([1, 2, 3, 4], n_interactions, p=[0.6, 0.1, 0.2, 0.1])
# 模拟时间戳（最近30天）
timestamps = np.random.randint(1609459200, 1612051200, n_interactions)
# 创建DataFrame
df = pd.DataFrame({
'user_id': user_ids,
'item_id': item_ids,
'behavior': behaviors,
'timestamp': timestamps
})
# 添加用户特征
user_features = pd.DataFrame({
'user_id': range(n_users),
'age': np.random.randint(18, 60, n_users),
'gender': np.random.choice([0, 1], n_users), # 0-女，1-男
'vip_level': np.random.choice([0, 1, 2, 3], n_users, p=[0.5, 0.3, 0.15, 0.05])
})
# 添加商品特征
item_features = pd.DataFrame({
'item_id': range(n_items),
'category': np.random.choice(['电子产品', '服装', '家居', '食品', '图书'], n_items),
'price': np.random.uniform(10, 1000, n_items),
'brand': np.random.choice(['品牌A', '品牌B', '品牌C', '品牌D', '品牌E'], n_items)
})
return df, user_features, item_features
# 加载数据
interactions_df, user_features_df, item_features_df = load_sample_data()
print(f"交互数据: {interactions_df.shape}")
print(f"用户特征: {user_features_df.shape}")
print(f"商品特征: {item_features_df.shape}")
# 数据预处理
def preprocess_data(interactions_df, user_features_df, item_features_df):
"""数据预处理函数"""
# 1. 编码分类特征
user_encoder = LabelEncoder()
item_encoder = LabelEncoder()
interactions_df['user_idx'] = user_encoder.fit_transform(interactions_df['user_id'])
interactions_df['item_idx'] = item_encoder.fit_transform(interactions_df['item_id'])
# 2. 归一化数值特征
scaler = MinMaxScaler()
user_features_df['age_norm'] = scaler.fit_transform(user_features_df[['age']])
user_features_df['vip_level_norm'] = scaler.fit_transform(user_features_df[['vip_level']])
item_features_df['price_norm'] = scaler.fit_transform(item_features_df[['price']])
# 3. 编码分类特征
user_features_df['gender_encoded'] = user_features_df['gender'] # 已经是数值
item_features_df['category_encoded'] = LabelEncoder().fit_transform(item_features_df['category'])
item_features_df['brand_encoded'] = LabelEncoder().fit_transform(item_features_df['brand'])
return interactions_df, user_features_df, item_features_df
# 执行预处理
interactions_df, user_features_df, item_features_df = preprocess_data(
interactions_df, user_features_df, item_features_df
)
2.2 构建深度推荐模型
TensorFlow 2.9的Keras API让模型构建变得非常简单。这里我们构建一个结合了用户特征和物品特征的深度神经网络推荐模型。
import tensorflow as tf
from tensorflow.keras import layers, Model
class DeepRecommendationModel(Model):
"""深度推荐模型"""
def __init__(self, n_users, n_items, user_feature_dim, item_feature_dim, embedding_dim=64):
super(DeepRecommendationModel, self).__init__()
# 用户ID嵌入层
self.user_embedding = layers.Embedding(
input_dim=n_users,
output_dim=embedding_dim,
name='user_embedding'
)
# 物品ID嵌入层
self.item_embedding = layers.Embedding(
input_dim=n_items,
output_dim=embedding_dim,
name='item_embedding'
)
# 用户特征处理网络
self.user_feature_net = tf.keras.Sequential([
layers.Dense(128, activation='relu'),
layers.Dropout(0.2),
layers.Dense(64, activation='relu'),
layers.Dropout(0.2),
layers.Dense(embedding_dim, activation='relu')
], name='user_feature_net')
# 物品特征处理网络
self.item_feature_net = tf.keras.Sequential([
layers.Dense(128, activation='relu'),
layers.Dropout(0.2),
layers.Dense(64, activation='relu'),
layers.Dropout(0.2),
layers.Dense(embedding_dim, activation='relu')
], name='item_feature_net')
# 融合层
self.fusion_net = tf.keras.Sequential([
layers.Dense(256, activation='relu'),
layers.Dropout(0.3),
layers.Dense(128, activation='relu'),
layers.Dropout(0.3),
layers.Dense(64, activation='relu'),
layers.Dense(1, activation='sigmoid') # 输出点击概率
], name='fusion_net')
def call(self, inputs):
# 解包输入
user_id, item_id, user_features, item_features = inputs
# 获取嵌入向量
user_emb = self.user_embedding(user_id)
item_emb = self.item_embedding(item_id)
# 处理特征
user_feat_emb = self.user_feature_net(user_features)
item_feat_emb = self.item_feature_net(item_features)
# 合并所有特征
user_combined = tf.concat([user_emb, user_feat_emb], axis=-1)
item_combined = tf.concat([item_emb, item_feat_emb], axis=-1)
# 计算交互特征
interaction = user_combined * item_combined
# 合并所有信息
combined = tf.concat([user_combined, item_combined, interaction], axis=-1)
# 预测
prediction = self.fusion_net(combined)
return prediction
# 准备训练数据
def prepare_training_data(interactions_df, user_features_df, item_features_df):
"""准备训练数据"""
# 创建标签（这里简化处理，购买行为作为正样本）
interactions_df['label'] = (interactions_df['behavior'] == 4).astype(int)
# 获取特征
user_features = user_features_df.set_index('user_id')[
['age_norm', 'vip_level_norm', 'gender_encoded']
].values
item_features = item_features_df.set_index('item_id')[
['price_norm', 'category_encoded', 'brand_encoded']
].values
# 准备输入数据
user_ids = interactions_df['user_idx'].values
item_ids = interactions_df['item_idx'].values
labels = interactions_df['label'].values
# 获取对应的特征
user_feat_input = user_features[user_ids]
item_feat_input = item_features[item_ids]
return (user_ids, item_ids, user_feat_input, item_feat_input), labels
# 准备数据
X, y = prepare_training_data(interactions_df, user_features_df, item_features_df)
user_ids, item_ids, user_feat_input, item_feat_input = X
# 划分训练集和测试集
from sklearn.model_selection import train_test_split
train_idx, test_idx = train_test_split(
range(len(y)), test_size=0.2, random_state=42, stratify=y
)
X_train = [arr[train_idx] for arr in [user_ids, item_ids, user_feat_input, item_feat_input]]
y_train = y[train_idx]
X_test = [arr[test_idx] for arr in [user_ids, item_ids, user_feat_input, item_feat_input]]
y_test = y[test_idx]
print(f"训练集大小: {len(y_train)}")
print(f"测试集大小: {len(y_test)}")
print(f"正样本比例: {y_train.mean():.2%}")
# 创建模型
n_users = len(user_features_df)
n_items = len(item_features_df)
user_feature_dim = user_feat_input.shape[1]
item_feature_dim = item_feat_input.shape[1]
model = DeepRecommendationModel(
n_users=n_users,
n_items=n_items,
user_feature_dim=user_feature_dim,
item_feature_dim=item_feature_dim,
embedding_dim=64
)
# 编译模型
model.compile(
optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),
loss='binary_crossentropy',
metrics=['accuracy', tf.keras.metrics.AUC(name='auc')]
)
# 查看模型结构
model.build([(None,), (None,), (None, user_feature_dim), (None, item_feature_dim)])
model.summary()
2.3 模型训练与评估
有了模型和数据，我们就可以开始训练了。在推荐系统中，我们通常需要关注多个评估指标。
# 设置回调函数
callbacks = [
tf.keras.callbacks.EarlyStopping(
monitor='val_auc',
patience=10,
mode='max',
restore_best_weights=True
),
tf.keras.callbacks.ReduceLROnPlateau(
monitor='val_loss',
factor=0.5,
patience=5,
min_lr=1e-6
),
tf.keras.callbacks.ModelCheckpoint(
'best_model.h5',
monitor='val_auc',
mode='max',
save_best_only=True
)
]
# 训练模型
history = model.fit(
X_train,
y_train,
batch_size=512,
epochs=50,
validation_split=0.1,
callbacks=callbacks,
verbose=1
)
# 评估模型
test_results = model.evaluate(X_test, y_test, verbose=0)
print("\n测试集评估结果:")
print(f"损失: {test_results[0]:.4f}")
print(f"准确率: {test_results[1]:.4f}")
print(f"AUC: {test_results[2]:.4f}")
# 绘制训练曲线
import matplotlib.pyplot as plt
def plot_training_history(history):
"""绘制训练历史"""
fig, axes = plt.subplots(1, 3, figsize=(15, 4))
# 损失曲线
axes[0].plot(history.history['loss'], label='训练损失')
axes[0].plot(history.history['val_loss'], label='验证损失')
axes[0].set_title('损失曲线')
axes[0].set_xlabel('Epoch')
axes[0].set_ylabel('Loss')
axes[0].legend()
axes[0].grid(True)
# 准确率曲线
axes[1].plot(history.history['accuracy'], label='训练准确率')
axes[1].plot(history.history['val_accuracy'], label='验证准确率')
axes[1].set_title('准确率曲线')
axes[1].set_xlabel('Epoch')
axes[1].set_ylabel('Accuracy')
axes[1].legend()
axes[1].grid(True)
# AUC曲线
axes[2].plot(history.history['auc'], label='训练AUC')
axes[2].plot(history.history['val_auc'], label='验证AUC')
axes[2].set_title('AUC曲线')
axes[2].set_xlabel('Epoch')
axes[2].set_ylabel('AUC')
axes[2].legend()
axes[2].grid(True)
plt.tight_layout()
plt.show()
plot_training_history(history)
3. 模型部署与服务化
模型训练好了，但这只是开始。要让推荐系统真正发挥作用，我们需要把它部署成可以服务线上请求的API。
3.1 模型保存与转换
TensorFlow提供了多种模型保存格式，我们需要根据部署环境选择合适的格式。
# 保存为SavedModel格式（推荐用于生产环境）
model.save('recommendation_model', save_format='tf')
# 也可以保存为H5格式
model.save('recommendation_model.h5')
# 验证保存的模型
loaded_model = tf.keras.models.load_model('recommendation_model')
# 测试加载的模型
test_predictions = loaded_model.predict(X_test[:10])
print(f"原始模型预测: {model.predict(X_test[:10])}")
print(f"加载模型预测: {test_predictions}")
print(f"预测一致性: {np.allclose(model.predict(X_test[:10]), test_predictions)}")
# 如果需要部署到移动端或边缘设备，可以转换为TFLite格式
converter = tf.lite.TFLiteConverter.from_saved_model('recommendation_model')
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()
# 保存TFLite模型
with open('recommendation_model.tflite', 'wb') as f:
f.write(tflite_model)
print("TFLite模型转换完成")
3.2 构建推荐服务API
在实际生产环境中，我们通常使用Web服务框架来提供推荐接口。这里使用Flask构建一个简单的推荐API。
# app.py - 推荐服务API
from flask import Flask, request, jsonify
import tensorflow as tf
import numpy as np
import pandas as pd
import pickle
from typing import Dict, List, Any
app = Flask(__name__)
class RecommendationService:
"""推荐服务类"""
def __init__(self, model_path: str, user_features_path: str, item_features_path: str):
"""初始化推荐服务"""
# 加载模型
self.model = tf.keras.models.load_model(model_path)
# 加载特征数据
self.user_features = pd.read_pickle(user_features_path)
self.item_features = pd.read_pickle(item_features_path)
# 加载编码器
with open('encoders.pkl', 'rb') as f:
self.encoders = pickle.load(f)
print("推荐服务初始化完成")
def preprocess_request(self, user_id: int, item_ids: List[int]) -> Dict[str, np.ndarray]:
"""预处理请求数据"""
# 获取用户特征
if user_id in self.user_features.index:
user_feat = self.user_features.loc[user_id].values.reshape(1, -1)
else:
# 新用户使用平均特征
user_feat = self.user_features.mean().values.reshape(1, -1)
# 获取物品特征
item_feats = []
valid_item_ids = []
for item_id in item_ids:
if item_id in self.item_features.index:
item_feats.append(self.item_features.loc[item_id].values)
valid_item_ids.append(item_id)
if not item_feats:
raise ValueError("没有有效的物品ID")
item_feats = np.array(item_feats)
# 编码用户ID和物品ID
user_encoder = self.encoders['user_encoder']
item_encoder = self.encoders['item_encoder']
try:
user_idx = user_encoder.transform([user_id])[0]
except:
# 新用户使用-1表示
user_idx = -1
item_idxs = []
for item_id in valid_item_ids:
try:
idx = item_encoder.transform([item_id])[0]
item_idxs.append(idx)
except:
# 新物品跳过
continue
if not item_idxs:
raise ValueError("没有可编码的物品ID")
# 重复用户特征以匹配物品数量
user_feat_repeated = np.repeat(user_feat, len(item_idxs), axis=0)
user_idxs_repeated = np.full(len(item_idxs), user_idx)
return {
'user_ids': np.array(user_idxs_repeated),
'item_ids': np.array(item_idxs),
'user_features': user_feat_repeated,
'item_features': item_feats
}
def get_recommendations(self, user_id: int, item_ids: List[int], top_k: int = 10) -> List[Dict]:
"""获取推荐结果"""
try:
# 预处理数据
processed_data = self.preprocess_request(user_id, item_ids)
# 批量预测
predictions = self.model.predict([
processed_data['user_ids'],
processed_data['item_ids'],
processed_data['user_features'],
processed_data['item_features']
])
# 获取top-k推荐
top_indices = np.argsort(predictions.flatten())[-top_k:][::-1]
results = []
for idx in top_indices:
item_id = item_ids[idx]
score = float(predictions[idx][0])
results.append({
'item_id': int(item_id),
'score': score,
'rank': len(results) + 1
})
return results
except Exception as e:
print(f"推荐失败: {str(e)}")
return []
# 初始化服务
recommendation_service = RecommendationService(
model_path='recommendation_model',
user_features_path='user_features.pkl',
item_features_path='item_features.pkl'
)
@app.route('/health', methods=['GET'])
def health_check():
"""健康检查接口"""
return jsonify({'status': 'healthy', 'service': 'recommendation'})
@app.route('/recommend', methods=['POST'])
def recommend():
"""推荐接口"""
try:
data = request.get_json()
# 验证输入
if 'user_id' not in data:
return jsonify({'error': '缺少user_id参数'}), 400
if 'item_ids' not in data or not data['item_ids']:
return jsonify({'error': '缺少item_ids参数或列表为空'}), 400
user_id = int(data['user_id'])
item_ids = [int(i) for i in data['item_ids']]
top_k = data.get('top_k', 10)
# 获取推荐结果
recommendations = recommendation_service.get_recommendations(
user_id, item_ids, top_k
)
return jsonify({
'user_id': user_id,
'recommendations': recommendations,
'count': len(recommendations)
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/batch_recommend', methods=['POST'])
def batch_recommend():
"""批量推荐接口"""
try:
data = request.get_json()
if 'requests' not in data:
return jsonify({'error': '缺少requests参数'}), 400
results = []
for req in data['requests']:
user_id = int(req['user_id'])
item_ids = [int(i) for i in req['item_ids']]
top_k = req.get('top_k', 10)
recommendations = recommendation_service.get_recommendations(
user_id, item_ids, top_k
)
results.append({
'user_id': user_id,
'recommendations': recommendations
})
return jsonify({'results': results})
except Exception as e:
return jsonify({'error': str(e)}), 500
if __name__ == '__main__':
# 保存特征数据（在实际应用中，这些数据应该来自数据库）
user_features_df.to_pickle('user_features.pkl')
item_features_df.to_pickle('item_features.pkl')
# 保存编码器
encoders = {
'user_encoder': user_encoder,
'item_encoder': item_encoder
}
with open('encoders.pkl', 'wb') as f:
pickle.dump(encoders, f)
print("特征数据和编码器已保存")
print("启动推荐服务...")
app.run(host='0.0.0.0', port=5000, debug=False)
3.3 使用Docker容器化部署
为了确保服务在不同环境中的一致性，我们使用Docker进行容器化部署。
# Dockerfile
FROM tensorflow/tensorflow:2.9.0
# 设置工作目录
WORKDIR /app
# 复制依赖文件
COPY requirements.txt .
# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 暴露端口
EXPOSE 5000
# 启动命令
CMD ["python", "app.py"]
# requirements.txt
flask==2.3.2
tensorflow==2.9.0
pandas==1.5.3
numpy==1.24.3
scikit-learn==1.3.0
gunicorn==20.1.0
构建和运行Docker容器：
# 构建Docker镜像
docker build -t recommendation-service .
# 运行容器
docker run -d \
-p 5000:5000 \
--name recommendation-service \
--restart always \
-v $(pwd)/models:/app/models \
recommendation-service
# 查看日志
docker logs -f recommendation-service
4. 生产环境优化与监控
模型上线后，我们的工作还没有结束。生产环境中的推荐系统需要持续的优化和监控。
4.1 性能优化策略
推荐系统在生产环境中面临的主要挑战是性能和稳定性。以下是一些优化策略：
# performance_optimizer.py - 性能优化工具
import time
from functools import wraps
from concurrent.futures import ThreadPoolExecutor
import redis
import json
class RecommendationCache:
"""推荐结果缓存"""
def __init__(self, host='localhost', port=6379, db=0, ttl=300):
"""初始化缓存"""
self.redis_client = redis.Redis(host=host, port=port, db=db)
self.ttl = ttl # 缓存过期时间（秒）
def get_cache_key(self, user_id: int, item_ids: List[int]) -> str:
"""生成缓存键"""
sorted_items = sorted(item_ids)
return f"rec:{user_id}:{hash(tuple(sorted_items))}"
def get(self, user_id: int, item_ids: List[int]) -> Optional[List[Dict]]:
"""获取缓存结果"""
cache_key = self.get_cache_key(user_id, item_ids)
cached_data = self.redis_client.get(cache_key)
if cached_data:
return json.loads(cached_data)
return None
def set(self, user_id: int, item_ids: List[int], recommendations: List[Dict]):
"""设置缓存结果"""
cache_key = self.get_cache_key(user_id, item_ids)
self.redis_client.setex(
cache_key,
self.ttl,
json.dumps(recommendations)
)
def invalidate_user(self, user_id: int):
"""使用户缓存失效"""
pattern = f"rec:{user_id}:*"
keys = self.redis_client.keys(pattern)
if keys:
self.redis_client.delete(*keys)
class BatchProcessor:
"""批量处理器"""
def __init__(self, max_workers=4, batch_size=100):
self.executor = ThreadPoolExecutor(max_workers=max_workers)
self.batch_size = batch_size
def process_batch(self, requests: List[Dict]) -> List[Dict]:
"""批量处理推荐请求"""
results = []
# 分批处理
for i in range(0, len(requests), self.batch_size):
batch = requests[i:i + self.batch_size]
batch_results = list(self.executor.map(
self._process_single,
batch
))
results.extend(batch_results)
return results
def _process_single(self, request: Dict) -> Dict:
"""处理单个请求"""
# 这里调用实际的推荐逻辑
time.sleep(0.01) # 模拟处理时间
return {'result': 'processed'}
def timing_decorator(func):
"""计时装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
# 记录执行时间（在实际应用中应该记录到监控系统）
print(f"{func.__name__} 执行时间: {end_time - start_time:.4f}秒")
return result
return wrapper
# 模型预测优化
class OptimizedModelPredictor:
"""优化模型预测"""
def __init__(self, model_path: str):
self.model = tf.keras.models.load_model(model_path)
# 使用TF Serving风格的批处理
self.batch_size = 32
@timing_decorator
def batch_predict(self, inputs: List[np.ndarray]) -> np.ndarray:
"""批量预测"""
# 将输入数据分批
n_samples = len(inputs[0])
predictions = []
for i in range(0, n_samples, self.batch_size):
batch_inputs = [
arr[i:i + self.batch_size] for arr in inputs
]
batch_pred = self.model.predict(batch_inputs, verbose=0)
predictions.append(batch_pred)
return np.vstack(predictions)
4.2 监控与日志系统
生产环境的推荐系统需要有完善的监控和日志系统。
# monitoring.py - 监控系统
import logging
from datetime import datetime
from typing import Dict, Any
import psutil
import requests
class RecommendationMonitor:
"""推荐系统监控"""
def __init__(self, service_name: str):
self.service_name = service_name
self.logger = self._setup_logger()
# 监控指标
self.metrics = {
'request_count': 0,
'success_count': 0,
'error_count': 0,
'avg_response_time': 0.0,
'cache_hit_rate': 0.0
}
def _setup_logger(self) -> logging.Logger:
"""设置日志记录器"""
logger = logging.getLogger(self.service_name)
logger.setLevel(logging.INFO)
# 文件处理器
file_handler = logging.FileHandler(f'{self.service_name}.log')
file_handler.setLevel(logging.INFO)
# 控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.WARNING)
# 格式化器
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
return logger
def log_request(self, user_id: int, item_count: int):
"""记录请求日志"""
self.metrics['request_count'] += 1
self.logger.info(f"推荐请求 - 用户: {user_id}, 物品数: {item_count}")
def log_success(self, response_time: float):
"""记录成功日志"""
self.metrics['success_count'] += 1
# 更新平均响应时间
total_time = self.metrics['avg_response_time'] * (self.metrics['success_count'] - 1)
self.metrics['avg_response_time'] = (total_time + response_time) / self.metrics['success_count']
self.logger.info(f"推荐成功 - 响应时间: {response_time:.3f}秒")
def log_error(self, error_type: str, error_msg: str):
"""记录错误日志"""
self.metrics['error_count'] += 1
self.logger.error(f"推荐错误 - 类型: {error_type}, 信息: {error_msg}")
def log_cache_hit(self, hit: bool):
"""记录缓存命中"""
total = self.metrics.get('cache_total', 0) + 1
hits = self.metrics.get('cache_hits', 0) + (1 if hit else 0)
self.metrics['cache_total'] = total
self.metrics['cache_hits'] = hits
self.metrics['cache_hit_rate'] = hits / total if total > 0 else 0.0
if hit:
self.logger.debug("缓存命中")
else:
self.logger.debug("缓存未命中")
def get_system_metrics(self) -> Dict[str, Any]:
"""获取系统指标"""
cpu_percent = psutil.cpu_percent(interval=1)
memory_info = psutil.virtual_memory()
return {
'timestamp': datetime.now().isoformat(),
'service': self.service_name,
'cpu_percent': cpu_percent,
'memory_percent': memory_info.percent,
'memory_used_gb': memory_info.used / (1024**3),
'business_metrics': self.metrics.copy()
}
def report_metrics(self, monitoring_url: str):
"""上报监控指标"""
try:
metrics = self.get_system_metrics()
response = requests.post(
monitoring_url,
json=metrics,
timeout=5
)
if response.status_code == 200:
self.logger.debug("监控指标上报成功")
else:
self.logger.warning(f"监控指标上报失败: {response.status_code}")
except Exception as e:
self.logger.error(f"监控指标上报异常: {str(e)}")
# 使用示例
monitor = RecommendationMonitor('recommendation-service')
# 在推荐服务中使用监控
@app.route('/recommend', methods=['POST'])
def recommend_with_monitoring():
"""带监控的推荐接口"""
start_time = time.time()
try:
data = request.get_json()
user_id = int(data['user_id'])
item_ids = [int(i) for i in data['item_ids']]
# 记录请求
monitor.log_request(user_id, len(item_ids))
# 检查缓存
cache = RecommendationCache()
cached_result = cache.get(user_id, item_ids)
if cached_result:
monitor.log_cache_hit(True)
response_time = time.time() - start_time
monitor.log_success(response_time)
return jsonify(cached_result)
monitor.log_cache_hit(False)
# 执行推荐
recommendations = recommendation_service.get_recommendations(
user_id, item_ids
)
# 缓存结果
cache.set(user_id, item_ids, recommendations)
response_time = time.time() - start_time
monitor.log_success(response_time)
return jsonify({
'user_id': user_id,
'recommendations': recommendations
})
except Exception as e:
monitor.log_error(type(e).__name__, str(e))
return jsonify({'error': str(e)}), 500
4.3 A/B测试与模型迭代
推荐系统需要持续优化，A/B测试是评估新模型效果的重要手段。
# ab_testing.py - A/B测试框架
import hashlib
from typing import Dict, List, Any
import pandas as pd
from datetime import datetime, timedelta
class ABTestManager:
"""A/B测试管理器"""
def __init__(self, experiments: Dict[str, Dict]):
"""
初始化A/B测试管理器
Args:
experiments: 实验配置
{
'experiment_name': {
'variants': ['A', 'B'],
'traffic_split': [0.5, 0.5],
'start_date': '2024-01-01',
'end_date': '2024-01-31'
}
}
"""
self.experiments = experiments
self.results = {}
def assign_variant(self, user_id: int, experiment_name: str) -> str:
"""分配实验变体"""
if experiment_name not in self.experiments:
return 'control' # 默认对照组
experiment = self.experiments[experiment_name]
# 检查实验是否在运行期间
current_date = datetime.now().date()
start_date = datetime.strptime(experiment['start_date'], '%Y-%m-%d').date()
end_date = datetime.strptime(experiment['end_date'], '%Y-%m-%d').date()
if current_date < start_date or current_date > end_date:
return 'control'
# 使用一致性哈希分配变体
hash_input = f"{user_id}_{experiment_name}".encode()
hash_value = int(hashlib.md5(hash_input).hexdigest(), 16)
# 根据流量分配比例分配变体
traffic_split = experiment['traffic_split']
cumulative_prob = 0
for i, variant in enumerate(experiment['variants']):
cumulative_prob += traffic_split[i]
if (hash_value % 10000) / 10000 < cumulative_prob:
return variant
return experiment['variants'][-1] # 默认返回最后一个变体
def track_event(self, user_id: int, experiment_name: str, variant: str,
event_type: str, event_data: Dict[str, Any]):
"""跟踪事件"""
if experiment_name not in self.results:
self.results[experiment_name] = {}
if variant not in self.results[experiment_name]:
self.results[experiment_name][variant] = {
'users': set(),
'events': []
}
# 记录用户
self.results[experiment_name][variant]['users'].add(user_id)
# 记录事件
event_record = {
'timestamp': datetime.now(),
'user_id': user_id,
'event_type': event_type,
'event_data': event_data
}
self.results[experiment_name][variant]['events'].append(event_record)
def analyze_results(self, experiment_name: str) -> Dict[str, Any]:
"""分析实验结果"""
if experiment_name not in self.results:
return {'error': '实验不存在'}
experiment_results = self.results[experiment_name]
analysis = {}
for variant, data in experiment_results.items():
users = data['users']
events = data['events']
# 计算基本指标
user_count = len(users)
event_count = len(events)
# 按事件类型统计
event_types = {}
for event in events:
event_type = event['event_type']
if event_type not in event_types:
event_types[event_type] = 0
event_types[event_type] += 1
# 计算转化率（示例：点击事件）
click_events = event_types.get('click', 0)
conversion_rate = click_events / user_count if user_count > 0 else 0
analysis[variant] = {
'user_count': user_count,
'event_count': event_count,
'event_types': event_types,
'conversion_rate': conversion_rate
}
return analysis
def get_recommendation_strategy(self, user_id: int) -> Dict[str, Any]:
"""获取推荐策略（根据A/B测试分配）"""
strategies = {}
for exp_name in self.experiments:
variant = self.assign_variant(user_id, exp_name)
strategies[exp_name] = {
'variant': variant,
'experiment': exp_name
}
return strategies
# 使用示例
# 定义A/B测试实验
experiments = {
'model_version': {
'variants': ['v1', 'v2'],
'traffic_split': [0.5, 0.5],
'start_date': '2024-01-01',
'end_date': '2024-12-31'
},
'ranking_strategy': {
'variants': ['popularity', 'personalized'],
'traffic_split': [0.3, 0.7],
'start_date': '2024-01-01',
'end_date': '2024-12-31'
}
}
# 初始化A/B测试管理器
ab_test_manager = ABTestManager(experiments)
# 在推荐服务中使用A/B测试
class ABTestRecommendationService:
"""支持A/B测试的推荐服务"""
def __init__(self, base_service, ab_test_manager):
self.base_service = base_service
self.ab_test_manager = ab_test_manager
def get_recommendations(self, user_id: int, item_ids: List[int]) -> List[Dict]:
"""获取推荐结果（支持A/B测试）"""
# 获取用户的实验分配
strategies = self.ab_test_manager.get_recommendation_strategy(user_id)
# 根据实验变体选择不同的推荐策略
model_variant = strategies['model_version']['variant']
ranking_variant = strategies['ranking_strategy']['variant']
# 获取基础推荐结果
recommendations = self.base_service.get_recommendations(user_id, item_ids)
# 根据实验变体调整推荐结果
if ranking_variant == 'popularity':
# 热门度排序策略
recommendations.sort(key=lambda x: x.get('popularity_score', 0), reverse=True)
else:
# 个性化排序策略（默认）
recommendations.sort(key=lambda x: x['score'], reverse=True)
# 跟踪推荐事件
self.ab_test_manager.track_event(
user_id=user_id,
experiment_name='model_version',
variant=model_variant,
event_type='recommendation',
event_data={'item_count': len(recommendations)}
)
return recommendations
def track_user_action(self, user_id: int, item_id: int, action: str):
"""跟踪用户行为"""
strategies = self.ab_test_manager.get_recommendation_strategy(user_id)
for exp_name, strategy in strategies.items():
self.ab_test_manager.track_event(
user_id=user_id,
experiment_name=exp_name,
variant=strategy['variant'],
event_type=action,
event_data={'item_id': item_id}
)
5. 总结
通过这个完整的实战案例，我们走过了推荐系统从开发到上线的全流程。从环境搭建、模型开发，到服务部署、性能优化，再到监控和A/B测试，每个环节都有其重要性。
5.1 关键要点回顾
-
环境准备是基础：TensorFlow 2.9提供了稳定高效的开发环境，配合Jupyter Notebook可以快速进行模型实验和迭代。
-
数据质量决定上限：推荐系统的效果很大程度上取决于数据的质量和特征工程的质量。在实际项目中，需要花费大量时间在数据清洗和特征构建上。
-
模型服务化是关键：训练好的模型需要能够服务线上请求。通过Flask等Web框架，我们可以将模型封装成RESTful API，方便其他系统调用。
-
性能优化不可少：生产环境的推荐系统需要处理高并发请求，缓存、批量处理、异步计算等优化手段是必不可少的。
-
监控和迭代是常态：推荐系统上线后，需要通过完善的监控系统跟踪性能指标，通过A/B测试验证新模型的效果，持续迭代优化。
5.2 实践经验分享
在实际项目中，还有一些经验值得分享：
关于模型选择：
- 对于中小型数据集，深度神经网络可能不是最佳选择，传统的矩阵分解或逻辑回归可能效果更好且更稳定
- 模型复杂度要与数据量匹配，避免过拟合
- 在线学习（Online Learning）对于数据分布变化快的场景很有价值
关于工程实践：
- 版本控制很重要：模型版本、数据版本、代码版本都要管理好
- 自动化测试：推荐系统的测试包括单元测试、集成测试、压力测试
- 灰度发布：新模型上线要逐步放量，观察效果后再全量
关于业务理解：
- 推荐系统最终要服务于业务目标，理解业务场景比技术选型更重要
- 多与产品、运营沟通，了解他们的需求和痛点
- 建立合理的评估体系，既要看技术指标（AUC、准确率），也要看业务指标（点击率、转化率）
5.3 下一步学习建议
如果你对这个领域感兴趣，可以从以下几个方面深入：
- 学习更先进的模型：如Transformer在推荐系统中的应用、图神经网络推荐模型等
- 探索多目标优化：现实中的推荐系统往往需要平衡多个目标（点击率、转化率、多样性等）
- 研究冷启动问题：如何处理新用户和新物品的推荐
- 了解实时推荐：如何利用实时用户行为数据更新推荐结果
- 学习大规模系统架构：如何设计支持亿级用户、千万级物品的推荐系统
推荐系统是一个既有深度又有广度的领域，需要机器学习、系统工程、产品思维等多方面的能力。希望这个实战案例能为你提供一个良好的起点，在实际项目中不断积累经验，构建出真正有价值的推荐系统。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
更多推荐
4
0- 0
已为社区贡献6条内容
所有评论(0)