推荐系统（十六）—— 工业级架构与最佳实践

1

用户请求 → 召回 (10⁶ → 2K) → 粗排 (2K → 200) → 精排 (200 → 50) → 重排 (50 → 20) → 返回

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

class RankingService:
    """无状态排序服务——可自由复制副本。"""

    def __init__(self, model_path: str):
        self.model = load_model(model_path)
        self.feature_extractor = FeatureExtractor()

    def rank(self, user_id, candidates, context):
        features = self.feature_extractor.extract(user_id, candidates, context)
        scores = self.model.predict(features)
        return sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)

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

class FaultTolerantRecall:
    def __init__(self, channels):
        self.channels = channels
        self.fallback = PopularItemsRecall()

    def recall(self, user_id, context):
        results = []
        for channel in self.channels:
            try:
                results.extend(channel.recall(user_id, context, timeout=20))
            except Exception as exc:
                logger.warning("channel %s failed: %s", channel.name, exc)

        if not results:
            return self.fallback.recall(user_id, context)
        return deduplicate(results)

 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

import torch
import torch.nn as nn

class TwoTowerRecall(nn.Module):
    """双塔模型，物品向量离线索引。"""

    def __init__(self, user_dim: int, item_dim: int, hidden=(256, 128)):
        super().__init__()
        self.user_tower = self._tower(user_dim, hidden)
        self.item_tower = self._tower(item_dim, hidden)

    @staticmethod
    def _tower(in_dim, hidden):
        layers, prev = [], in_dim
        for h in hidden:
            layers += [nn.Linear(prev, h), nn.ReLU(), nn.BatchNorm1d(h)]
            prev = h
        return nn.Sequential(*layers)

    def recall(self, user_features, ann_index, top_k=1000):
        with torch.no_grad():
            user_emb = self.user_tower(user_features)
        # ann_index 是离线物品向量构建的 Faiss IVF-PQ 索引
        _, top_ids = ann_index.search(user_emb.cpu().numpy(), top_k)
        return top_ids

 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

from collections import defaultdict

class GraphRecall:
    """基于共同用户的物品 Jaccard 相似度做物品到物品的召回。"""

    def __init__(self, interaction_graph):
        self.graph = interaction_graph
        self.item_similarity = self._compute_similarity()

    def _compute_similarity(self):
        sim = defaultdict(dict)
        items = [n for n in self.graph.nodes() if self.graph.nodes[n]["type"] == "item"]
        for a in items:
            users_a = set(self.graph.neighbors(a))
            for b in items:
                if a == b:
                    continue
                users_b = set(self.graph.neighbors(b))
                inter, union = len(users_a & users_b), len(users_a | users_b)
                if union:
                    sim[a][b] = inter / union
        return sim

    def recall(self, user_id, top_k=1000):
        seen = [n for n in self.graph.neighbors(user_id)
                if self.graph.nodes[n]["type"] == "item"]
        scores = defaultdict(float)
        for item in seen:
            for neighbour, w in self.item_similarity.get(item, {}).items():
                scores[neighbour] += w
        return [i for i, _ in sorted(scores.items(), key=lambda x: -x[1])[:top_k]]

 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

from collections import defaultdict, deque
from datetime import datetime, timedelta

import numpy as np

class RealTimeBehaviorRecall:
    """基于最近 30 分钟行为的召回，时间衰减 + 行为加权。"""

    def __init__(self, window_minutes=30):
        self.window = timedelta(minutes=window_minutes)
        self.behaviours = defaultdict(deque)

    def add_behavior(self, user_id, item_id, action_type, ts):
        self.behaviours[user_id].append({"item": item_id, "action": action_type, "ts": ts})
        cutoff = ts - self.window
        while self.behaviours[user_id] and self.behaviours[user_id][0]["ts"] < cutoff:
            self.behaviours[user_id].popleft()

    def recall(self, user_id, top_k=500):
        weights = {"view": 1.0, "click": 2.0, "purchase": 5.0}
        scores = defaultdict(float)
        now = datetime.now()
        for b in self.behaviours.get(user_id, []):
            age_min = (now - b["ts"]).total_seconds() / 60
            recency = np.exp(-age_min / 10)
            scores[b["item"]] += recency * weights.get(b["action"], 1.0)
        return [i for i, _ in sorted(scores.items(), key=lambda x: -x[1])[:top_k]]

 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 collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed

class MultiChannelRecall:
    def __init__(self, channels, weights=None):
        self.channels = channels
        self.weights = weights or {c.name: 1.0 for c in channels}

    def recall(self, user_id, context, target=2000):
        results = {}
        with ThreadPoolExecutor(max_workers=len(self.channels)) as pool:
            futures = {pool.submit(c.recall, user_id, context): c.name for c in self.channels}
            for fut in as_completed(futures):
                name = futures[fut]
                try:
                    results[name] = fut.result(timeout=25)
                except Exception as exc:
                    logger.error("channel %s failed: %s", name, exc)

        scores = defaultdict(float)
        for name, items in results.items():
            w = self.weights.get(name, 1.0)
            for rank, item in enumerate(items):
                scores[item] += w / (rank + 1)  # 倒数排名融合
        return [i for i, _ in sorted(scores.items(), key=lambda x: -x[1])[:target]]

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

import xgboost as xgb

class CoarseRanker:
    def __init__(self):
        self.model = xgb.XGBRanker(
            objective="rank:pairwise",
            tree_method="hist",
            max_depth=4,
            n_estimators=50,
        )

    def fit(self, X, y, group):
        self.model.fit(X, y, group=group)

    def predict(self, X):
        return self.model.predict(X)

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

class WideDeepRanking(nn.Module):
    """Google Wide & Deep：线性记忆 + 深度泛化。"""

    def __init__(self, wide_dim, embed_dims, deep_hidden):
        super().__init__()
        self.wide = nn.Linear(wide_dim, 1)
        self.embeddings = nn.ModuleDict(
            {name: nn.Embedding(vocab, dim) for name, (vocab, dim) in embed_dims.items()}
        )
        deep_in = sum(dim for _, dim in embed_dims.values())
        layers, prev = [], deep_in
        for h in deep_hidden:
            layers += [nn.Linear(prev, h), nn.ReLU(), nn.BatchNorm1d(h), nn.Dropout(0.2)]
            prev = h
        layers.append(nn.Linear(prev, 1))
        self.deep = nn.Sequential(*layers)

    def forward(self, wide_feats, sparse_ids, dense_feats):
        wide_out = self.wide(wide_feats)
        emb = [self.embeddings[name](ids) for name, ids in sparse_ids.items()]
        deep_out = self.deep(torch.cat(emb + [dense_feats], dim=1))
        return wide_out + deep_out

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

class DIN(nn.Module):
    """Deep Interest Network：在用户行为序列上做注意力。"""

    def __init__(self, item_dim, user_dim, hidden=(128, 64)):
        super().__init__()
        self.attention = nn.Sequential(
            nn.Linear(item_dim * 4, 36), nn.ReLU(), nn.Linear(36, 1)
        )
        in_dim = item_dim + user_dim + item_dim
        layers, prev = [], in_dim
        for h in hidden:
            layers += [nn.Linear(prev, h), nn.ReLU(), nn.Dropout(0.2)]
            prev = h
        layers.append(nn.Linear(prev, 1))
        self.mlp = nn.Sequential(*layers)

    def forward(self, candidate, history, user_profile):
        cand_exp = candidate.unsqueeze(1).expand_as(history)
        attn_in = torch.cat(
            [history, cand_exp, history - cand_exp, history * cand_exp], dim=2
        )
        weights = torch.softmax(self.attention(attn_in).squeeze(-1), dim=1)
        weighted = (history * weights.unsqueeze(-1)).sum(dim=1)
        return self.mlp(torch.cat([candidate, weighted, user_profile], dim=1))

 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

import numpy as np

class DiversityReranker:
    def __init__(self, lambda_div=0.3):
        self.lam = lambda_div

    def rerank(self, items, scores, features, top_k=20):
        selected, remaining = [], list(zip(items, scores, features))
        while len(selected) < top_k and remaining:
            best_idx, best_obj = None, -np.inf
            for idx, (it, sc, ft) in enumerate(remaining):
                if selected:
                    diversity = min(self._dist(ft, s_ft) for _, _, s_ft in selected)
                else:
                    diversity = 1.0
                obj = (1 - self.lam) * sc + self.lam * diversity
                if obj > best_obj:
                    best_obj, best_idx = obj, idx
            selected.append(remaining.pop(best_idx))
        return [it for it, _, _ in selected]

    @staticmethod
    def _dist(a, b):
        va, vb = np.array(list(a.values())), np.array(list(b.values()))
        return 1 - va @ vb / (np.linalg.norm(va) * np.linalg.norm(vb) + 1e-8)

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

class BusinessRulesReranker:
    def __init__(self, rules):
        self.rules = rules

    def rerank(self, items, scores, metadata):
        out = []
        for item, score in zip(items, scores):
            meta, adj, ok = metadata.get(item, {}), score, True
            for rule in self.rules:
                if not rule.check(item, meta):
                    ok = False
                    break
                adj += rule.score_adjustment(item, meta)
            if ok:
                out.append((item, adj))
        out.sort(key=lambda x: -x[1])
        return [i for i, _ in out]

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

from datetime import datetime
import numpy as np

class FreshnessReranker:
    def __init__(self, decay_hours=24, max_boost=0.3):
        self.decay, self.max_boost = decay_hours, max_boost

    def rerank(self, items, scores, timestamps):
        now, out = datetime.now(), []
        for item, score in zip(items, scores):
            ts = timestamps.get(item)
            if ts:
                age_h = (now - ts).total_seconds() / 3600
                boost = self.max_boost * np.exp(-age_h / self.decay)
                out.append((item, score * (1 + boost)))
            else:
                out.append((item, score))
        out.sort(key=lambda x: -x[1])
        return [i for i, _ in out]

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

import json

class FeatureStore:
    """基于 Redis 的在线特征仓库，支持批量读取。"""

    def __init__(self, redis_client, ttl=3600):
        self.redis, self.ttl = redis_client, ttl

    def set(self, entity, eid, name, value):
        self.redis.setex(f"{entity}:{eid}:{name}", self.ttl, json.dumps(value))

    def batch_get(self, entity, eids, names):
        keys = [f"{entity}:{e}:{n}" for e in eids for n in names]
        values = self.redis.mget(keys)
        n = len(names)
        return [
            [json.loads(v) if v else None for v in values[i:i + n]]
            for i in range(0, len(values), n)
        ]

1
2
3
4
5
6
7
8
9

def temporal_features(ts):
    dt = datetime.fromtimestamp(ts)
    return {
        "hour_sin": np.sin(2 * np.pi * dt.hour / 24),
        "hour_cos": np.cos(2 * np.pi * dt.hour / 24),
        "dow_sin":  np.sin(2 * np.pi * dt.weekday() / 7),
        "dow_cos":  np.cos(2 * np.pi * dt.weekday() / 7),
        "is_weekend": int(dt.weekday() >= 5),
    }

 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
49

import numpy as np
from collections import defaultdict
from scipy.stats import norm

class ABTestFramework:
    """生产级 A/B 框架：一致性分流 + z 检验。"""

    def __init__(self):
        self.experiments, self.events = {}, defaultdict(list)

    def create(self, exp_id, variants_split):  # 例如 {"control": 50, "v1": 50}
        self.experiments[exp_id] = variants_split

    def assign(self, user_id, exp_id):
        h = hash(f"{user_id}:{exp_id}") % 100
        cum = 0
        for variant, split in self.experiments[exp_id].items():
            cum += split
            if h < cum:
                return variant
        return "control"

    def log(self, user_id, exp_id, variant, event):
        self.events[exp_id].append({"user": user_id, "variant": variant, "event": event})

    def analyse(self, exp_id):
        stats = defaultdict(lambda: {"impr": 0, "click": 0})
        for ev in self.events[exp_id]:
            if ev["event"] == "impression":
                stats[ev["variant"]]["impr"] += 1
            elif ev["event"] == "click":
                stats[ev["variant"]]["click"] += 1

        out, ctrl = {}, stats.get("control")
        for v, s in stats.items():
            ctr = s["click"] / max(s["impr"], 1)
            out[v] = {"ctr": ctr, **s}

        if ctrl:
            for v, s in out.items():
                if v == "control":
                    continue
                p_pool = (ctrl["click"] + s["click"]) / (ctrl["impr"] + s["impr"])
                se = np.sqrt(p_pool * (1 - p_pool) * (1 / ctrl["impr"] + 1 / s["impr"]))
                z = (s["ctr"] - out["control"]["ctr"]) / (se + 1e-8)
                s["lift"] = (s["ctr"] - out["control"]["ctr"]) / (out["control"]["ctr"] + 1e-8)
                s["p_value"] = 2 * (1 - norm.cdf(abs(z)))
                s["significant"] = s["p_value"] < 0.05
        return out

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

class CanaryDeployment:
    def deploy(self, version, initial_pct=10):
        if not self.validate_offline(version):
            raise ValueError("offline validation failed")
        canary = self.deploy_canary(version, traffic=initial_pct)
        metrics = self.monitor(canary, minutes=60)
        if self.healthy(metrics):
            self.rollout(canary, target=100)
        else:
            self.rollback(canary)
            self.alert("canary failed", metrics)

    @staticmethod
    def healthy(m):
        return (m["p95_latency"] < 100
                and m["error_rate"] < 0.01
                and m.get("ctr_delta", 0) > -0.05)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

import torch.quantization as quant

def quantize(model, calibration_loader):
    model.eval()
    model.qconfig = quant.get_default_qconfig("fbgemm")
    quant.prepare(model, inplace=True)
    with torch.no_grad():
        for batch in calibration_loader:
            model(batch)
    return quant.convert(model, inplace=False)

1
2
3
4
5
6
7
8

import torch.nn.functional as F

def distill_loss(student_logits, teacher_logits, labels, T=3.0, alpha=0.7):
    s_soft = F.log_softmax(student_logits / T, dim=1)
    t_soft = F.softmax(teacher_logits / T, dim=1)
    soft = F.kl_div(s_soft, t_soft, reduction="batchmean") * (T ** 2)
    hard = F.cross_entropy(student_logits, labels)
    return alpha * soft + (1 - alpha) * hard

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

class RecommendationMonitor:
    def __init__(self, metrics):
        self.metrics = metrics

    def log(self, user_id, scores, latency_ms):
        self.metrics.histogram("pred.latency", latency_ms)
        self.metrics.histogram("pred.score_mean", float(np.mean(scores)))

    def check(self):
        recent = self.metrics.recent("pred.latency", minutes=5)
        if recent["p95"] > 150:
            self.alert("high_latency", recent)
        if self.metrics.error_rate(minutes=5) > 0.05:
            self.alert("high_error_rate")
        scores = self.metrics.recent("pred.score_mean", minutes=30)
        baseline = self.metrics.baseline("pred.score_mean")
        if abs(scores["mean"] - baseline["mean"]) > 2 * baseline["std"]:
            self.alert("distribution_shift", scores)