推荐系统（十五）—— 实时推荐与在线学习

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from kafka import KafkaProducer
import json, time

producer = KafkaProducer(
    bootstrap_servers=["kafka-1:9092", "kafka-2:9092", "kafka-3:9092"],
    value_serializer=lambda v: json.dumps(v).encode(),
    key_serializer=lambda k: k.encode(),
    acks="all",        # 等待所有 ISR 副本确认
    enable_idempotence=True,  # 生产者侧 exactly-once
    compression_type="lz4",
    linger_ms=5,       # 5 毫秒微批—— 5ms 延迟换 10x 吞吐
)

def emit_click(user_id: str, item_id: str, position: int) -> None:
    """发送一次点击事件。同一用户落同一分区。"""
    producer.send(
        topic="clicks",
        key=user_id,                # 分区键
        value={
            "user_id": user_id,
            "item_id": item_id,
            "position": position,
            "ts": int(time.time() * 1000),
        },
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

# Flink SQL —— 每用户最近 10 分钟点击数，写入特征存储
from pyflink.table import EnvironmentSettings, StreamTableEnvironment

t_env = StreamTableEnvironment.create(
    environment_settings=EnvironmentSettings.in_streaming_mode()
)

t_env.execute_sql("""
CREATE TABLE clicks (
    user_id STRING,
    item_id STRING,
    ts BIGINT,
    event_time AS TO_TIMESTAMP_LTZ(ts, 3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'clicks',
    'properties.bootstrap.servers' = 'kafka-1:9092',
    'format' = 'json',
    'scan.startup.mode' = 'latest-offset'
)
""")

t_env.execute_sql("""
CREATE TABLE user_features (
    user_id STRING,
    window_end TIMESTAMP(3),
    clicks_10m BIGINT,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
    'connector' = 'redis',
    'host' = 'redis.prod',
    'ttl-sec' = '900'
)
""")

# 10 分钟跳跃窗口，步长 1 分钟 —— 每分钟吐一次新鲜的 CTR
t_env.execute_sql("""
INSERT INTO user_features
SELECT
    user_id,
    HOP_END(event_time, INTERVAL '1' MINUTE, INTERVAL '10' MINUTE) AS window_end,
    COUNT(*) AS clicks_10m
FROM clicks
GROUP BY
    user_id,
    HOP(event_time, INTERVAL '1' MINUTE, INTERVAL '10' MINUTE)
""")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

import numpy as np

class FTRLProximal:
    """FTRL-Proximal 在线逻辑回归。

    参考文献：McMahan 等，"Ad Click Prediction: a View from the
    Trenches"，KDD 2013。Google 当时广告系统真正运行的算法。
    """
    def __init__(self, n_features: int, alpha=0.1, beta=1.0,
                 l1=1.0, l2=1.0):
        self.n = n_features
        self.alpha, self.beta, self.l1, self.l2 = alpha, beta, l1, l2
        # 只持久化 z 和 n，权重 w 按需懒计算
        self.z = np.zeros(n_features)
        self.n_sq = np.zeros(n_features)

    def _weight(self, i: int) -> float:
        """从 (z_i, n_sq_i) 懒计算 w_i。"""
        z, n_sq = self.z[i], self.n_sq[i]
        if abs(z) <= self.l1:
            return 0.0
        sign = 1.0 if z > 0 else -1.0
        return -(z - sign * self.l1) / (
            (self.beta + np.sqrt(n_sq)) / self.alpha + self.l2
        )

    def predict_proba(self, x_indices: np.ndarray, x_values: np.ndarray) -> float:
        """稀疏预测，只在非零特征上累加。"""
        s = sum(self._weight(i) * v for i, v in zip(x_indices, x_values))
        return 1.0 / (1.0 + np.exp(-max(min(s, 35.0), -35.0)))

    def update(self, x_indices: np.ndarray, x_values: np.ndarray, y: int) -> float:
        """单样本更新，返回更新前的预测概率。"""
        p = self.predict_proba(x_indices, x_values)
        for i, v in zip(x_indices, x_values):
            g = (p - y) * v                       # 该特征上的梯度
            sigma = (np.sqrt(self.n_sq[i] + g * g) - np.sqrt(self.n_sq[i])) / self.alpha
            self.z[i]   += g - sigma * self._weight(i)
            self.n_sq[i] += g * g
        return p

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

class ThompsonSampling:
    """伯努利 Thompson Sampling。生产实现就是这样——
    唯一要加的是衰减（每天 alpha, beta *= 0.99），
    用来处理非稳态的物品池。"""

    def __init__(self, n_arms: int):
        self.alpha = np.ones(n_arms)   # 成功次数 + 1
        self.beta  = np.ones(n_arms)   # 失败次数 + 1

    def select(self) -> int:
        return int(np.argmax(np.random.beta(self.alpha, self.beta)))

    def update(self, arm: int, reward: int) -> None:
        if reward:
            self.alpha[arm] += 1
        else:
            self.beta[arm] += 1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

class LinUCB:
    """LinUCB —— disjoint 模型版本。
    参考文献：Li, Chu, Langford, Schapire，WWW 2010。"""

    def __init__(self, n_arms: int, n_features: int, alpha: float = 1.0):
        self.alpha = alpha
        self.A = [np.eye(n_features) for _ in range(n_arms)]
        self.b = [np.zeros(n_features) for _ in range(n_arms)]

    def select(self, x: np.ndarray) -> int:
        ucbs = []
        for A_a, b_a in zip(self.A, self.b):
            A_inv = np.linalg.inv(A_a)
            theta = A_inv @ b_a
            ucbs.append(x @ theta + self.alpha * np.sqrt(x @ A_inv @ x))
        return int(np.argmax(ucbs))

    def update(self, arm: int, x: np.ndarray, reward: float) -> None:
        self.A[arm] += np.outer(x, x)
        self.b[arm] += reward * x

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

class DriftAdaptiveLearner:
    def __init__(self, base_learner, base_lr=0.1, window=200, z_thresh=3.0):
        self.learner = base_learner
        self.base_lr = base_lr
        self.window = window
        self.z_thresh = z_thresh
        self.recent = []        # 滚动误差
        self.ref_mean = None
        self.ref_std = None

    def update(self, x, y):
        pred = self.learner.predict(x)
        err = abs(pred - y)
        self.recent.append(err)
        if len(self.recent) > self.window:
            self.recent.pop(0)

        # 等第一个窗口稳定后建立参考
        if self.ref_mean is None and len(self.recent) == self.window:
            self.ref_mean = np.mean(self.recent)
            self.ref_std = np.std(self.recent) + 1e-6

        # 检测漂移
        if self.ref_mean is not None:
            z = (np.mean(self.recent[-50:]) - self.ref_mean) / self.ref_std
            if abs(z) > self.z_thresh:
                self.learner.learning_rate = self.base_lr * 4   # 拉高
                # 处理后重置参考——避免永久过激状态
                self.recent = []
                self.ref_mean = None

        self.learner.update(x, y)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

class RealtimeRanker:
    """端到端实时排序器。整合：
      - 特征存储（Flink 产出的新鲜特征）
      - FTRL-Proximal 在线学习器（按事件更新权重）
      - 在学习器输出之上加一层 LinUCB（探索）
      - 漂移检测器（在 regime change 时拉高学习率）
      - 快照回滚（生产安全）
    """

    def __init__(self, n_features: int, n_strategies: int = 4):
        self.scorer = FTRLProximal(n_features=n_features, alpha=0.1,
                                    beta=1.0, l1=1.0, l2=1.0)
        # Bandit 在排序策略上选择，而非物品上
        self.bandit = LinUCB(n_arms=n_strategies, n_features=n_features, alpha=0.5)
        self.drift  = DriftDetector(z_thresh=3.0)
        self.snapshots = []  # (ts, scorer) 的环形缓冲

    def rank(self, ctx: np.ndarray, candidates: list[dict]) -> list[str]:
        # 1. Bandit 选策略
        strategy = self.bandit.select(ctx)
        # 2. FTRL 给每个候选打分——稀疏向量
        scored = [
            (c["id"], self.scorer.predict_proba(c["idx"], c["val"]), c)
            for c in candidates
        ]
        # 3. 策略修正打分（如新品助推、创作者助推）
        scored = apply_strategy(strategy, scored)
        scored.sort(key=lambda r: -r[1])
        return [r[0] for r in scored[:20]]

    def observe(self, ctx, x_idx, x_val, y, strategy):
        """一次反馈循环。"""
        # 更新在线学习器
        p = self.scorer.update(x_idx, x_val, y)
        # 给实际播出去的策略更新 bandit
        self.bandit.update(strategy, ctx, reward=float(y))
        # 把校准误差喂给漂移检测器
        if self.drift.update(abs(p - y)):
            self.scorer.alpha *= 4              # 拉高学习率
        # 周期性安全快照
        if len(self.snapshots) == 0 or time.time() - self.snapshots[-1][0] > 600:
            self.snapshots.append((time.time(), copy.deepcopy(self.scorer)))
            self.snapshots = self.snapshots[-12:]  # 保留 2 小时