推荐系统（十三）—— 公平性、去偏与可解释性

 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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

from collections import defaultdict
from typing import Dict, List

import numpy as np

class BiasMetrics:
    """每个推荐系统都该报告的六个指标，除了 NDCG/HR。"""

    def __init__(
        self,
        recommendations: Dict[int, List[int]],
        item_popularity: Dict[int, int],
        user_groups: Dict[int, str] | None = None,
        item_groups: Dict[int, str] | None = None,
    ) -> None:
        self.recs = recommendations
        self.pop = item_popularity
        self.user_groups = user_groups or {}
        self.item_groups = item_groups or {}

    def popularity_bias(self, k: int = 10) -> float:
        """推荐物品与目录之间的对数流行度差距。"""
        rec_items = [i for r in self.recs.values() for i in r[:k]]
        rec_pop = np.mean([np.log1p(self.pop.get(i, 0)) for i in rec_items])
        all_pop = np.mean([np.log1p(p) for p in self.pop.values()])
        return float(rec_pop - all_pop)

    def gini(self, k: int = 10) -> float:
        """全目录的曝光不平等程度。0 = 均匀，1 = 赢者通吃。"""
        exposure = defaultdict(int)
        for r in self.recs.values():
            for i in r[:k]:
                exposure[i] += 1
        x = np.sort(np.array(list(exposure.values()), dtype=float))
        n = len(x)
        if n == 0 or x.sum() == 0:
            return 0.0
        return float((2 * np.sum(np.arange(1, n + 1) * x)) / (n * x.sum()) - (n + 1) / n)

    def coverage(self, k: int = 10) -> float:
        """至少被推荐过一次的物品占目录的比例。"""
        seen = {i for r in self.recs.values() for i in r[:k]}
        return len(seen) / max(len(self.pop), 1)

    def demographic_disparity(self, k: int = 10) -> float:
        """不同用户群之间推荐长度均值的最大差距（服务质量代理）。"""
        if not self.user_groups:
            return 0.0
        per_group = defaultdict(list)
        for u, r in self.recs.items():
            per_group[self.user_groups.get(u, "?")].append(len(r[:k]))
        means = [np.mean(v) for v in per_group.values()]
        return float(max(means) - min(means)) if means else 0.0

    def intra_list_diversity(self, k: int = 10) -> float:
        """每个 slate 内不同类目占比的平均值。"""
        if not self.item_groups:
            return 0.0
        scores = []
        for r in self.recs.values():
            slate = r[:k]
            if len(slate) < 2:
                continue
            cats = [self.item_groups.get(i, "?") for i in slate]
            scores.append(len(set(cats)) / len(cats))
        return float(np.mean(scores)) if scores else 0.0

    def report(self, k: int = 10) -> Dict[str, float]:
        return {
            "popularity_bias": self.popularity_bias(k),
            "gini": self.gini(k),
            "coverage": self.coverage(k),
            "demographic_disparity": self.demographic_disparity(k),
            "intra_list_diversity": self.intra_list_diversity(k),
        }

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

import numpy as np
import torch
import torch.nn as nn

def ips_loss(scores: torch.Tensor, clicks: torch.Tensor,
             positions: torch.Tensor, propensities: torch.Tensor,
             clip: float = 10.0) -> torch.Tensor:
    """IPS 加权交叉熵。

    scores      : 模型 logits，形状 (B,)
    clicks      : 0/1 标签，形状 (B,)
    positions   : 物品展示的排名位置，形状 (B,)
    propensities: P(检视 | 位置 k)，形状 (max_pos+1,)
    clip        : IPS 权重上限，用于控制方差
    """
    pi = propensities[positions].clamp(min=1e-3)
    weights = (1.0 / pi).clamp(max=clip)
    per_example = nn.functional.binary_cross_entropy_with_logits(
        scores, clicks.float(), reduction="none"
    )
    return (weights * clicks * per_example).mean()

1

score(u, i) = main(u, i) - alpha * item_tower(i) - beta * user_tower(u)

 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
50
51
52
53

class CFairER(nn.Module):
    """推荐系统嵌入的对抗去偏方法。

    两个网络进行 minimax 博弈：
    - 判别器：从用户嵌入预测受保护属性
    - 编码器：既要生成高质量推荐分数，又要迷惑判别器
    """

    def __init__(self, n_users: int, n_items: int, dim: int = 64) -> None:
        super().__init__()
        self.user_emb = nn.Embedding(n_users, dim)
        self.item_emb = nn.Embedding(n_items, dim)
        self.predictor = nn.Sequential(
            nn.Linear(2 * dim, 128), nn.ReLU(),
            nn.Linear(128, 1),
        )
        self.discriminator = nn.Sequential(
            nn.Linear(dim, 64), nn.ReLU(),
            nn.Linear(64, 1),
        )

    def forward(self, u: torch.Tensor, i: torch.Tensor):
        ue = self.user_emb(u)
        ie = self.item_emb(i)
        score = self.predictor(torch.cat([ue, ie], dim=-1)).squeeze(-1)
        attr_logit = self.discriminator(ue).squeeze(-1)
        return score, attr_logit, ue

def train_step(model: CFairER, batch, opt_main, opt_disc,
               lam_fair: float = 1.0) -> dict:
    u, i, y, a = batch  # 用户、物品、评分、受保护属性
    score, attr_logit, ue = model(u, i)

    # 1) 更新判别器（基于 detach 后的嵌入）
    opt_disc.zero_grad()
    a_logit = model.discriminator(ue.detach()).squeeze(-1)
    d_loss = nn.functional.binary_cross_entropy_with_logits(a_logit, a.float())
    d_loss.backward()
    opt_disc.step()

    # 2) 更新编码器和预测器（既要推荐准确，又要迷惑判别器）
    opt_main.zero_grad()
    score, attr_logit, _ = model(u, i)
    pred_loss = nn.functional.mse_loss(score, y.float())
    # 让判别器变得不确定（概率空间目标为 0.5）
    fair_loss = nn.functional.binary_cross_entropy_with_logits(
        attr_logit, torch.full_like(attr_logit, 0.5)
    )
    total = pred_loss + lam_fair * fair_loss
    total.backward()
    opt_main.step()

    return {"pred": pred_loss.item(), "disc": d_loss.item(), "fair": fair_loss.item()}

 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 sklearn.linear_model import Ridge

def lime_explain(predict_fn, x: np.ndarray, n_samples: int = 1000,
                 sigma: float = 0.1, n_top: int = 8):
    """返回对 predict_fn(x) 影响最大的前 k 个特征。

    predict_fn : callable(np.ndarray of shape (N, d)) -> (N,)
    x          : 要解释的实例, shape (d,)
    sigma      : 扰动噪声尺度
    """
    d = x.shape[0]
    # 在 x 周围生成扰动样本
    samples = x + np.random.normal(0, sigma, size=(n_samples, d))
    preds = predict_fn(samples)

    # 按距离加权（RBF 核）
    dist = np.linalg.norm(samples - x, axis=1)
    weights = np.exp(-(dist ** 2) / (2 * sigma ** 2))

    # 拟合稀疏线性代理模型
    surrogate = Ridge(alpha=1.0)
    surrogate.fit(samples, preds, sample_weight=weights)

    # 按系数绝对值取前 k 个特征
    coefs = surrogate.coef_
    top = np.argsort(-np.abs(coefs))[:n_top]
    return [(int(j), float(coefs[j])) for j in top]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def render_explanation(user_id: int, item_id: int,
                       shap_values: dict, history: list) -> dict:
    top_pos = sorted(shap_values.items(), key=lambda x: -x[1])[:2]
    top_neg = sorted(shap_values.items(), key=lambda x: x[1])[:1]
    return {
        "headline": f"推荐理由：你之前喜欢 {history[0]} 和 {history[1]}",
        "confidence": min(99, int(50 + 100 * sum(v for _, v in top_pos))),
        "positive_factors": [name for name, _ in top_pos],
        "negative_factors": [name for name, _ in top_neg],
        "feedback_options": ["多推类似内容", "少推类似内容", "完全不感兴趣"],
    }