现代推荐系统架构¶

import torch
import torch.nn as nn

class ScenarioAdaptiveExpert(nn.Module):  # 继承nn.Module定义网络层
    """场景自适应的Expert网络（简化版）"""

    def __init__(self, input_dim, expert_dim, num_experts, num_scenarios):
        super().__init__()  # super()调用父类方法
        self.num_experts = num_experts

        # 共享Expert网络
        self.experts = nn.ModuleList([
            nn.Sequential(
                nn.Linear(input_dim, expert_dim),
                nn.ReLU(),
                nn.Linear(expert_dim, expert_dim)
            ) for _ in range(num_experts)
        ])

        # 场景感知的Gate网络
        self.scenario_gates = nn.ModuleList([
            nn.Sequential(
                nn.Linear(input_dim + num_scenarios, num_experts),
                nn.Softmax(dim=-1)
            ) for _ in range(num_scenarios)
        ])

        self.num_scenarios = num_scenarios

    def forward(self, x, scenario_id):
        """
        x: [batch_size, input_dim]
        scenario_id: [batch_size] 场景ID (0 ~ num_scenarios-1)
        """
        # 计算所有Expert的输出
        expert_outputs = [expert(x) for expert in self.experts]
        expert_outputs = torch.stack(expert_outputs, dim=1)  # [B, num_experts, expert_dim]  # torch.stack沿新维度拼接张量

        # 场景One-hot编码
        scenario_onehot = torch.zeros(x.size(0), self.num_scenarios, device=x.device)
        scenario_onehot.scatter_(1, scenario_id.unsqueeze(1), 1)  # unsqueeze增加一个维度

        # 场景感知Gate
        gate_input = torch.cat([x, scenario_onehot], dim=-1)  # torch.cat沿已有维度拼接张量

        # 按场景选择Gate
        outputs = torch.zeros_like(expert_outputs[:, 0])
        for s in range(self.num_scenarios):
            mask = (scenario_id == s)
            if mask.any():  # any()任一为True则返回True
                gate_weights = self.scenario_gates[s](gate_input[mask])
                outputs[mask] = torch.einsum('be,bed->bd', gate_weights, expert_outputs[mask])

        return outputs

import torch
import torch.nn as nn
import torch.nn.functional as F

class DiffRecSimplified(nn.Module):
    """DiffRec简化实现 —— 基于扩散过程的协同过滤"""

    def __init__(self, num_items, hidden_dim=256, num_steps=50):
        super().__init__()
        self.num_items = num_items
        self.num_steps = num_steps

        # 噪声调度 (线性beta schedule)
        # 使用register_buffer确保.to(device)时自动迁移到正确设备
        betas = torch.linspace(1e-4, 0.02, num_steps)
        self.register_buffer('betas', betas)
        self.register_buffer('alphas', 1.0 - betas)
        self.register_buffer('alpha_cumprod', torch.cumprod(self.alphas, dim=0))

        # 去噪网络 (U-Net简化为MLP)
        self.denoise_net = nn.Sequential(
            nn.Linear(num_items + 1, hidden_dim),  # +1 for timestep
            nn.SiLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.SiLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.SiLU(),
            nn.Linear(hidden_dim, num_items),
        )

    def forward_diffusion(self, x_0, t):
        """前向扩散：给干净数据加噪"""
        alpha_t = self.alpha_cumprod[t].unsqueeze(-1).to(x_0.device)
        noise = torch.randn_like(x_0)
        x_t = torch.sqrt(alpha_t) * x_0 + torch.sqrt(1 - alpha_t) * noise
        return x_t, noise

    def predict_noise(self, x_t, t):
        """预测噪声"""
        t_normalized = t.float() / self.num_steps
        t_embed = t_normalized.unsqueeze(-1)
        net_input = torch.cat([x_t, t_embed], dim=-1)
        return self.denoise_net(net_input)

    def training_step(self, user_interactions):
        """
        训练步骤
        user_interactions: [batch_size, num_items] 用户交互矩阵的一行
        """
        batch_size = user_interactions.size(0)

        # 随机采样时间步
        t = torch.randint(0, self.num_steps, (batch_size,), device=user_interactions.device)

        # 前向扩散
        x_t, noise = self.forward_diffusion(user_interactions, t)

        # 预测噪声
        predicted_noise = self.predict_noise(x_t, t)

        # MSE损失
        loss = F.mse_loss(predicted_noise, noise)  # F.xxx PyTorch函数式API
        return loss

    @torch.no_grad()  # 禁用梯度计算，节省内存
    def recommend(self, user_interactions, top_k=10):
        """推理：去噪生成推荐"""
        x_t = torch.randn_like(user_interactions)

        for t in reversed(range(self.num_steps)):
            t_batch = torch.full((x_t.size(0),), t, device=x_t.device, dtype=torch.long)
            predicted_noise = self.predict_noise(x_t, t_batch)

            alpha_t = self.alphas[t]
            alpha_cumprod_t = self.alpha_cumprod[t]

            x_t = (1 / torch.sqrt(alpha_t)) * (
                x_t - (self.betas[t] / torch.sqrt(1 - alpha_cumprod_t)) * predicted_noise
            )

            if t > 0:
                x_t += torch.sqrt(self.betas[t]) * torch.randn_like(x_t)

        # 屏蔽已交互物品
        x_t[user_interactions > 0] = -float('inf')

        # 返回Top-K推荐
        _, top_indices = torch.topk(x_t, top_k, dim=-1)
        return top_indices

class RecommendationAgent:
    """推荐Agent核心框架"""

    def __init__(self, llm, tools, memory):
        self.llm = llm          # LLM作为推理引擎
        self.tools = tools      # 工具集合
        self.memory = memory    # 记忆模块

    def recommend(self, user_query: str, user_id: str) -> dict:
        """完整推荐流程"""

        # Step 1: 意图理解
        intent = self._understand_intent(user_query)
        # {"type": "探索式", "domain": "电影", "constraints": ["科幻", "2024年"]}

        # Step 2: 策略规划
        plan = self._plan_strategy(intent, user_id)
        # ["查询用户画像", "召回候选集", "语义重排", "生成解释"]

        # Step 3: 工具调用执行
        results = self._execute_plan(plan, user_id)

        # Step 4: 结果生成
        recommendations = self._generate_results(results, intent)

        # Step 5: 记忆更新
        self.memory.update(user_id, intent, recommendations)

        return recommendations

    def _understand_intent(self, query):
        """用LLM理解用户意图"""
        prompt = f"分析用户推荐请求的意图：\n{query}\n提取：类型、领域、约束条件"
        return self.llm.analyze(prompt)

    def _plan_strategy(self, intent, user_id):
        """基于意图和用户历史规划推荐策略"""
        user_history = self.memory.get_long_term(user_id)
        prompt = f"根据意图{intent}和用户历史，规划推荐步骤"
        return self.llm.plan(prompt)

    def _execute_plan(self, plan, user_id):
        """依次执行计划中的工具调用"""
        results = {}
        for step in plan:
            tool = self.tools.get(step)
            if tool:
                results[step] = tool.execute(user_id=user_id)
        return results

    def _generate_results(self, results, intent):
        """整合工具结果，生成最终推荐"""
        prompt = f"根据以下信息生成推荐：\n工具结果：{results}\n用户意图：{intent}"
        return self.llm.generate(prompt)

# Feast Feature Store 示例
from feast import FeatureStore

# 初始化Feature Store
store = FeatureStore(repo_path="feature_repo/")

# 在线获取用户特征（低延迟）
user_features = store.get_online_features(
    features=[
        "user_profile:age",
        "user_profile:gender",
        "user_behavior:click_count_7d",
        "user_behavior:purchase_count_30d",
        "user_embedding:embedding_vector",
    ],
    entity_rows=[{"user_id": "12345"}]
).to_dict()

# 离线获取训练数据（批量）
training_df = store.get_historical_features(
    entity_df=entity_df,  # 包含user_id和timestamp的DataFrame
    features=[
        "user_profile:age",
        "item_features:category",
        "item_features:price",
        "user_item_cross:ctr_history",
    ]
).to_df()

# 使用Milvus进行大规模向量召回
from pymilvus import MilvusClient

# 连接Milvus
client = MilvusClient(uri="http://localhost:19530")

# 创建推荐物品集合
client.create_collection(
    collection_name="item_embeddings",
    dimension=768,  # LLM Embedding维度
    metric_type="COSINE",
)

# 插入物品向量
client.insert(
    collection_name="item_embeddings",
    data=[
        {"id": item_id, "vector": embedding.tolist(), "category": cat}
        for item_id, embedding, cat in zip(item_ids, embeddings, categories)  # zip按位置配对
    ]
)

# 向量检索召回
results = client.search(
    collection_name="item_embeddings",
    data=[user_embedding.tolist()],
    limit=100,                    # 召回Top 100
    output_fields=["category"],   # 返回额外字段
    filter='category == "电影"',  # 标量过滤
)

class RecommendationABTest:
    """推荐系统AB测试框架"""

    def __init__(self):
        self.experiments = {}

    def create_experiment(self, name, control_model, treatment_model,
                         traffic_split=0.5, metrics=None):
        """创建AB实验"""
        self.experiments[name] = {
            "control": control_model,
            "treatment": treatment_model,
            "traffic_split": traffic_split,
            "metrics": metrics or ["ctr", "cvr", "revenue", "dwell_time"],
            "status": "running",
            "results": {"control": [], "treatment": []}
        }

    def assign_group(self, user_id, experiment_name):
        """用户分组（基于hash确保一致性）"""
        hash_val = hash(f"{user_id}_{experiment_name}") % 100
        split = int(self.experiments[experiment_name]["traffic_split"] * 100)
        return "treatment" if hash_val < split else "control"

    def analyze_results(self, experiment_name):
        """分析实验结果（t检验）"""
        from scipy import stats
        import numpy as np

        exp = self.experiments[experiment_name]
        results = {}

        for metric in exp["metrics"]:
            control_data = [r[metric] for r in exp["results"]["control"]]
            treatment_data = [r[metric] for r in exp["results"]["treatment"]]

            t_stat, p_value = stats.ttest_ind(control_data, treatment_data)

            control_mean = np.mean(control_data)
            treatment_mean = np.mean(treatment_data)
            lift = (treatment_mean - control_mean) / control_mean * 100

            results[metric] = {
                "control_mean": control_mean,
                "treatment_mean": treatment_mean,
                "lift": f"{lift:.2f}%",
                "p_value": p_value,
                "significant": p_value < 0.05
            }

        return results

import numpy as np

class FairRecommender:
    """公平性约束的推荐系统"""

    def __init__(self, base_recommender, fairness_type="demographic_parity"):
        self.base_recommender = base_recommender
        self.fairness_type = fairness_type

    def recommend_with_fairness(self, user_id, candidate_items,
                                 sensitive_attr, top_k=10):
        """
        带公平性约束的推荐
        sensitive_attr: 物品的敏感属性（如供应商、类别、地区）
        """
        # 获取基础推荐分数
        scores = self.base_recommender.predict(user_id, candidate_items)

        if self.fairness_type == "demographic_parity":
            return self._demographic_parity_rerank(
                candidate_items, scores, sensitive_attr, top_k
            )
        elif self.fairness_type == "equal_opportunity":
            return self._equal_opportunity_rerank(
                candidate_items, scores, sensitive_attr, top_k
            )

    def _demographic_parity_rerank(self, items, scores, attr, top_k):
        """
        Demographic Parity: 各组别物品在推荐列表中的比例应与候选集中相同
        """
        # 统计各组别在候选集中的比例
        groups = set(attr)
        target_ratio = {g: sum(1 for a in attr if a == g) / len(attr) for g in groups}

        # 贪心选择：交替从各组中选择高分物品
        selected = []
        group_counts = {g: 0 for g in groups}
        sorted_indices = np.argsort(scores)[::-1]

        for idx in sorted_indices:
            if len(selected) >= top_k:
                break
            group = attr[idx]
            current_ratio = group_counts[group] / max(len(selected), 1)

            if current_ratio <= target_ratio[group] * 1.2:  # 允许20%偏差
                selected.append(idx)
                group_counts[group] += 1

        return [items[i] for i in selected]