AI 赋能的实时推荐架构从离线模型到在线推理的性能桥接方案一、离线模型的时间差问题一个电商平台的推荐系统每天凌晨 2 点Spark 任务计算用户画像和推荐列表写入 Redis。白天用户看到的推荐基于昨天的行为。用户上午浏览了 10 款运动鞋推荐流里仍然是昨天下午看的键盘推荐——行为反馈延迟了 12-24 小时。将离线模型直接搬到在线推理不是简单的换成 Flink。实时推理面临三个挑战特征计算延迟用户最新行为必须在毫秒级完成特征转化、模型推理延迟复杂模型在 CPU 上的推理时间可能超过用户容忍度、冷启动新用户的推荐完全依赖全局热门需要实时积累行为。二、实时推荐系统的分层架构2.1 特征桥接离线特征 实时增量// 特征融合服务合并离线和实时的用户特征 type FeatureFusion struct { offlineStore *redis.Client // 离线特征每天更新 realtimeStore *redis.Client // 实时特征秒级更新 featureWeights map[string]float64 // 特征融合权重 } func (f *FeatureFusion) GetUserFeatures( ctx context.Context, userID string, ) (*UserFeatures, error) { // 1. 离线特征长期稳定的画像 offlineKey : fmt.Sprintf(user:offline:%s, userID) offline, err : f.offlineStore.HGetAll(ctx, offlineKey).Result() if err ! nil !errors.Is(err, redis.Nil) { return nil, fmt.Errorf(获取离线特征失败: %w, err) } // 2. 实时特征最近 30 分钟的行为特征 realtimeKey : fmt.Sprintf(user:realtime:%s, userID) realtime, err : f.realtimeStore.HGetAll(ctx, realtimeKey).Result() if err ! nil !errors.Is(err, redis.Nil) { return nil, fmt.Errorf(获取实时特征失败: %w, err) } // 3. 特征融合实时特征权重随时间衰减 features : make(map[string]float64) // 离线特征基础权重 for k, v : range offline { weight : f.featureWeights[k] if weight 0 { weight 0.3 // 默认离线权重 } val, _ : strconv.ParseFloat(v, 64) features[k] val * weight } // 实时特征叠加权重随时间衰减 for k, v : range realtime { weight : f.featureWeights[realtime_k] if weight 0 { weight 0.7 // 默认实时权重高 } val, _ : strconv.ParseFloat(v, 64) features[k] (features[k] * 0.3) (val * weight * 0.7) } return UserFeatures{ UserID: userID, Features: features, }, nil }三、在线推理的性能优化3.1 模型推理的批量合并// 推理批处理器1024 个独立请求合并为一个 batch type InferenceBatcher struct { queue chan InferenceRequest batchSize int // 最大 batch size maxWaitTime time.Duration // 最长等待时间 model *tf.SavedModel } type InferenceRequest struct { Features []float32 ResultCh chan *InferenceResult ErrCh chan error } func (b *InferenceBatcher) Predict(features []float32) (*InferenceResult, error) { req : InferenceRequest{ Features: features, ResultCh: make(chan *InferenceResult, 1), ErrCh: make(chan error, 1), } select { case b.queue - req: default: return nil, fmt.Errorf(推理队列已满) } select { case result : -req.ResultCh: return result, nil case err : -req.ErrCh: return nil, err } } func (b *InferenceBatcher) Run(ctx context.Context) { batch : make([]InferenceRequest, 0, b.batchSize) timer : time.NewTimer(b.maxWaitTime) defer timer.Stop() for { timer.Reset(b.maxWaitTime) select { case req : -b.queue: batch append(batch, req) if len(batch) b.batchSize { b.executeBatch(batch) batch batch[:0] } case -timer.C: if len(batch) 0 { b.executeBatch(batch) batch batch[:0] } case -ctx.Done(): return } } } func (b *InferenceBatcher) executeBatch(batch []InferenceRequest) { // 构建 batch 输入 features : make([][]float32, len(batch)) for i, req : range batch { features[i] req.Features } // 执行批量推理 results, err : b.model.Predict(features) if err ! nil { for _, req : range batch { req.ErrCh - err } return } // 分发结果 for i, req : range batch { req.ResultCh - InferenceResult{ Scores: results[i], } } }四、边界与权衡实时特征的质量衰减实时特征基于短期行为窗口如 30 分钟样本量小导致统计不稳定。用户误点一个不感兴趣的商品可能导致接下来 30 分钟的推荐全部跑偏。加入行为置信度权重——浏览权重 点击权重 加购权重 购买权重。在线推理的延迟预算推荐系统对延迟极度敏感P99 100ms。实时特征计算 模型推理 排序每一步的延迟都必须严格控制在预算内。建议分配特征获取 20ms、模型推理 50ms、排序 20ms、网络传输 10ms。批量推理的尾部延迟合并请求等待填充 batch最慢的请求延迟 等待时间 推理时间。设置合理的maxWaitTime如 50ms在吞吐和延迟之间取得平衡。五、总结离线到实时的推荐架构升级核心是特征桥接——将离线的长期画像与实时的短期行为通过加权融合。技术栈Kafka 做行为流、Flink 做实时特征、Redis 做在线特征存储、TensorFlow Serving 批量推理做模型服务。实施路径先实现用户最近 15 分钟行为特征的实时计算Feature Pipeline→ 验证实时特征对推荐效果的增量→ 再引入在线推理的批量合并优化。不要试图一步到位把所有特征都做实时代——80% 的推荐效果增益通常来自 20% 的关键实时特征。
AI 赋能的实时推荐架构从离线模型到在线推理的性能桥接方案一、离线模型的时间差问题一个电商平台的推荐系统每天凌晨 2 点Spark 任务计算用户画像和推荐列表写入 Redis。白天用户看到的推荐基于昨天的行为。用户上午浏览了 10 款运动鞋推荐流里仍然是昨天下午看的键盘推荐——行为反馈延迟了 12-24 小时。将离线模型直接搬到在线推理不是简单的换成 Flink。实时推理面临三个挑战特征计算延迟用户最新行为必须在毫秒级完成特征转化、模型推理延迟复杂模型在 CPU 上的推理时间可能超过用户容忍度、冷启动新用户的推荐完全依赖全局热门需要实时积累行为。二、实时推荐系统的分层架构2.1 特征桥接离线特征 实时增量// 特征融合服务合并离线和实时的用户特征 type FeatureFusion struct { offlineStore *redis.Client // 离线特征每天更新 realtimeStore *redis.Client // 实时特征秒级更新 featureWeights map[string]float64 // 特征融合权重 } func (f *FeatureFusion) GetUserFeatures( ctx context.Context, userID string, ) (*UserFeatures, error) { // 1. 离线特征长期稳定的画像 offlineKey : fmt.Sprintf(user:offline:%s, userID) offline, err : f.offlineStore.HGetAll(ctx, offlineKey).Result() if err ! nil !errors.Is(err, redis.Nil) { return nil, fmt.Errorf(获取离线特征失败: %w, err) } // 2. 实时特征最近 30 分钟的行为特征 realtimeKey : fmt.Sprintf(user:realtime:%s, userID) realtime, err : f.realtimeStore.HGetAll(ctx, realtimeKey).Result() if err ! nil !errors.Is(err, redis.Nil) { return nil, fmt.Errorf(获取实时特征失败: %w, err) } // 3. 特征融合实时特征权重随时间衰减 features : make(map[string]float64) // 离线特征基础权重 for k, v : range offline { weight : f.featureWeights[k] if weight 0 { weight 0.3 // 默认离线权重 } val, _ : strconv.ParseFloat(v, 64) features[k] val * weight } // 实时特征叠加权重随时间衰减 for k, v : range realtime { weight : f.featureWeights[realtime_k] if weight 0 { weight 0.7 // 默认实时权重高 } val, _ : strconv.ParseFloat(v, 64) features[k] (features[k] * 0.3) (val * weight * 0.7) } return UserFeatures{ UserID: userID, Features: features, }, nil }三、在线推理的性能优化3.1 模型推理的批量合并// 推理批处理器1024 个独立请求合并为一个 batch type InferenceBatcher struct { queue chan InferenceRequest batchSize int // 最大 batch size maxWaitTime time.Duration // 最长等待时间 model *tf.SavedModel } type InferenceRequest struct { Features []float32 ResultCh chan *InferenceResult ErrCh chan error } func (b *InferenceBatcher) Predict(features []float32) (*InferenceResult, error) { req : InferenceRequest{ Features: features, ResultCh: make(chan *InferenceResult, 1), ErrCh: make(chan error, 1), } select { case b.queue - req: default: return nil, fmt.Errorf(推理队列已满) } select { case result : -req.ResultCh: return result, nil case err : -req.ErrCh: return nil, err } } func (b *InferenceBatcher) Run(ctx context.Context) { batch : make([]InferenceRequest, 0, b.batchSize) timer : time.NewTimer(b.maxWaitTime) defer timer.Stop() for { timer.Reset(b.maxWaitTime) select { case req : -b.queue: batch append(batch, req) if len(batch) b.batchSize { b.executeBatch(batch) batch batch[:0] } case -timer.C: if len(batch) 0 { b.executeBatch(batch) batch batch[:0] } case -ctx.Done(): return } } } func (b *InferenceBatcher) executeBatch(batch []InferenceRequest) { // 构建 batch 输入 features : make([][]float32, len(batch)) for i, req : range batch { features[i] req.Features } // 执行批量推理 results, err : b.model.Predict(features) if err ! nil { for _, req : range batch { req.ErrCh - err } return } // 分发结果 for i, req : range batch { req.ResultCh - InferenceResult{ Scores: results[i], } } }四、边界与权衡实时特征的质量衰减实时特征基于短期行为窗口如 30 分钟样本量小导致统计不稳定。用户误点一个不感兴趣的商品可能导致接下来 30 分钟的推荐全部跑偏。加入行为置信度权重——浏览权重 点击权重 加购权重 购买权重。在线推理的延迟预算推荐系统对延迟极度敏感P99 100ms。实时特征计算 模型推理 排序每一步的延迟都必须严格控制在预算内。建议分配特征获取 20ms、模型推理 50ms、排序 20ms、网络传输 10ms。批量推理的尾部延迟合并请求等待填充 batch最慢的请求延迟 等待时间 推理时间。设置合理的maxWaitTime如 50ms在吞吐和延迟之间取得平衡。五、总结离线到实时的推荐架构升级核心是特征桥接——将离线的长期画像与实时的短期行为通过加权融合。技术栈Kafka 做行为流、Flink 做实时特征、Redis 做在线特征存储、TensorFlow Serving 批量推理做模型服务。实施路径先实现用户最近 15 分钟行为特征的实时计算Feature Pipeline→ 验证实时特征对推荐效果的增量→ 再引入在线推理的批量合并优化。不要试图一步到位把所有特征都做实时代——80% 的推荐效果增益通常来自 20% 的关键实时特征。
