📖 AI/ML系统设计¶

"""ML系统设计中的5大权衡"""

TRADE_OFFS = {
    "模型复杂度 vs 延迟": {
        "高复杂度": "精度高但推理慢（Transformer类）",
        "低复杂度": "延迟低但精度可能不够（LR/小模型）",
        "解法": "多阶段架构：召回(简单) → 粗排(中等) → 精排(复杂) → 重排(业务)"
    },
    "实时性 vs 准确性": {
        "实时特征": "最新但计算复杂（如实时CTR统计）",
        "离线特征": "延迟低但有信息滞后",
        "解法": "Lambda架构：离线批量 + 实时流式 双路特征"
    },
    "个性化 vs 冷启动": {
        "强个性化": "老用户体验好但新用户无数据",
        "泛化模型": "冷启动好但老用户体验差",
        "解法": "ESMM多任务学习 + 基于内容的冷启动补偿"
    },
    "探索 vs 利用": {
        "利用(Exploit)": "推荐已知高质量内容，用户满意度高",
        "探索(Explore)": "推荐新内容，可能发现长尾价值",
        "解法": "Thompson Sampling / ε-greedy / UCB"
    },
    "模型公平性 vs 收益优化": {
        "纯收益优化": "可能产生偏见和马太效应",
        "公平性约束": "降低短期收益但长期健康",
        "解法": "多目标优化 + 公平性正则化 + 反偏见采样"
    }
}

"""数据Pipeline设计模板"""

class DataPipeline:
    """
    数据Pipeline三层架构：
    1. 数据接入层（Ingestion）：多源采集
    2. 数据处理层（Processing）：清洗、转换
    3. 数据存储层（Storage）：分层存储
    """

    def __init__(self):
        self.architecture = {
            "数据接入": {
                "实时流": "Kafka/Pulsar → Flink → 实时特征",
                "批量": "MySQL/Hive → Spark → 离线特征",
                "日志": "Flume/Filebeat → ES → 行为分析"
            },
            "数据处理": {
                "清洗": ["去重", "补缺失值", "异常检测", "格式统一"],
                "转换": ["编码", "归一化", "时间窗口聚合", "特征交叉"],
                "验证": ["Schema检查", "分布偏移检测", "数据新鲜度"]
            },
            "数据存储": {
                "原始层(Raw)": "S3/HDFS — 保存原始数据，不可修改",
                "清洗层(Clean)": "Hive/Iceberg — 清洗后的结构化数据",
                "特征层(Feature)": "Redis/DynamoDB — 线上线下一致的特征",
                "标注层(Label)": "Hive — 标注数据和样本"
            }
        }

    def design_sample_generation(self):
        """样本构建是ML Pipeline中最关键的环节"""
        return {
            "正样本": "用户点击/购买/完播 → label=1",
            "负样本": {
                "曝光未点击": "最直接，但有展示位置偏差",
                "随机负采样": "缓解位置偏差，但可能引入噪声",
                "Hard Negative": "提升模型区分能力"
            },
            "样本偏差处理": {
                "位置偏差": "Position-aware模型 / IPW(逆倾向加权)",
                "选择偏差": "因果推断方法 / 无偏估计",
                "时间偏差": "使用训练时间截断，避免未来信息泄露"
            },
            "样本比例": "正负样本比通常 1:5~1:20，根据业务调整"
        }

"""分布式训练Pipeline"""

class TrainingPipeline:
    def __init__(self):
        self.config = {
            "训练框架": {
                "小模型(<1B参数)": "单机多卡 DataParallel",
                "中模型(1B-10B)": "多机多卡 DDP + ZeRO",
                "大模型(>10B)": "Megatron-LM / DeepSpeed / FSDP"
            },
            "训练策略": {
                "全量训练": "每天/每周 全量数据重训",
                "增量训练": "基于新数据微调，每小时/每天",
                "持续学习": "Online Learning，实时更新"
            },
            "超参调优": {
                "Grid Search": "小规模参数空间（<100组合）",
                "Bayesian Optimization": "中等参数空间（Optuna）",
                "Population Based Training": "大规模资源（Ray Tune）"
            }
        }

    def model_versioning(self):
        """模型版本管理——跟踪模型的全生命周期：从实验到生产到归档"""
        return {
            "模型注册表": "MLflow Model Registry / 自建",
            "版本信息": {
                "模型ID": "唯一标识",
                "训练数据": "数据版本 + 时间范围",
                "超参数": "完整的超参配置",
                "评估指标": "离线指标（AUC/F1/NDCG等）",
                "依赖环境": "Python版本 + 包版本 + CUDA版本"
            },
            "生命周期": "Staging → Canary → Production → Archived"
        }

"""双塔模型（Two-Tower）向量召回实现"""
import torch
import torch.nn as nn

class TwoTowerModel(nn.Module):
    """
    双塔模型：
    - User Tower: 用户特征 → 用户向量 (128d)
    - Item Tower: 物品特征 → 物品向量 (128d)
    - 训练时：内积 + Softmax交叉熵
    - 推理时：User向量在线计算，Item向量离线存入ANN索引
    """
    def __init__(self, user_feat_dim, item_feat_dim, embed_dim=128):
        super().__init__()
        # 用户塔：将用户特征映射到embed_dim维的向量空间
        # 三层MLP + BatchNorm，逐步压缩特征维度
        self.user_tower = nn.Sequential(
            nn.Linear(user_feat_dim, 256),
            nn.ReLU(),
            nn.BatchNorm1d(256),   # 批归一化加速收敛
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, embed_dim),
            # 注意：L2归一化在 forward() 中用 F.normalize 实现
        )
        # 物品塔：结构与用户塔对称，保证向量空间一致
        self.item_tower = nn.Sequential(
            nn.Linear(item_feat_dim, 256),
            nn.ReLU(),
            nn.BatchNorm1d(256),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, embed_dim),
        )
        # 温度参数：控制softmax分布的锐度，可学习优化
        self.temperature = nn.Parameter(torch.tensor(0.07))

    def forward(self, user_feat, item_feat):
        # 用户塔前向传播 + L2归一化，得到单位向量
        user_emb = torch.nn.functional.normalize(
            self.user_tower(user_feat), dim=-1
        )
        # 物品塔前向传播 + L2归一化
        item_emb = torch.nn.functional.normalize(
            self.item_tower(item_feat), dim=-1
        )
        # 计算用户-物品相似度矩阵：[batch_u, batch_i]
        # 除以温度参数调节softmax分布的平滑程度
        logits = torch.matmul(user_emb, item_emb.T) / self.temperature.exp()
        return logits, user_emb, item_emb

# ---------- 向量检索服务 ----------
class ANNService:
    """
    近似最近邻检索服务
    离线：Item向量 → 构建ANN索引（Faiss/Milvus/Pinecone）
    在线：User向量 → ANN topK检索 → 召回候选集
    """
    def __init__(self, index_type="IVF_PQ"):
        self.index_configs = {
            "Flat": {
                "精度": "100%（精确）",
                "速度": "慢，仅适合 <100万",
                "内存": "高"
            },
            "IVF_PQ": {
                "精度": "~95%",
                "速度": "快，适合 千万~亿级",
                "内存": "低（压缩向量）",
                "参数": "nlist=1024, m=32, nprobe=64"
            },
            "HNSW": {
                "精度": "~98%",
                "速度": "极快",
                "内存": "高（需要构建图）",
                "参数": "M=32, ef_construction=200, ef_search=128"
            }
        }

"""MMOE (Multi-gate Mixture-of-Experts) 实现"""

class MMOE(nn.Module):
    """
    多任务学习核心架构：
    - 多个Expert网络提取不同特征模式
    - 每个Task有独立的Gate网络选择Expert组合
    - 解决任务冲突问题（如点击率 vs 完播率）

    对比：
    - Shared-Bottom: 底层全共享，任务冲突严重
    - MMOE: Expert + Gate，灵活共享
    - PLE (CGC): 任务独有Expert + 共享Expert，效果最好
    """
    def __init__(self, input_dim, num_experts=8, num_tasks=3,
                 expert_dim=64):
        super().__init__()  # super()调用父类方法
        self.num_experts = num_experts
        self.num_tasks = num_tasks

        # Expert网络组：多个独立的MLP，每个学习不同的特征模式
        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)
        ])

        # 门控网络（每个任务一个）：学习如何加权组合各Expert的输出
        # Softmax保证权重和为1，实现“软选择”而非“硬切换”
        self.gates = nn.ModuleList([
            nn.Sequential(
                nn.Linear(input_dim, num_experts),
                nn.Softmax(dim=-1)
            ) for _ in range(num_tasks)
        ])

        # 任务塔（每个任务一个）：将Expert混合特征映射为最终预测值
        # Sigmoid输出[0,1]概率值，适用于点击率/点赞率等二分类任务
        self.towers = nn.ModuleList([
            nn.Sequential(
                nn.Linear(expert_dim, 32),
                nn.ReLU(),
                nn.Linear(32, 1),
                nn.Sigmoid()
            ) for _ in range(num_tasks)
        ])

    def forward(self, x):
        # 所有Expert并行计算，得到 [batch, num_experts, expert_dim]
        expert_outputs = torch.stack(
            [expert(x) for expert in self.experts], dim=1
        )

        task_outputs = []
        for i in range(self.num_tasks):
            # 第i个任务的门控权重: [batch, num_experts] → 扩展维度用于加权
            gate = self.gates[i](x).unsqueeze(-1)
            # Expert加权求和：根据门控权重融合多个Expert的输出
            mixture = (expert_outputs * gate).sum(dim=1)
            # 经过任务塔得到最终预测: [batch, 1]
            output = self.towers[i](mixture)
            task_outputs.append(output)

        # 返回各任务预测结果，如 [P(click), P(like), P(follow)]
        return task_outputs  # [P(click), P(like), P(follow)]

"""搜索排序模型设计"""

class SearchRanker:
    """
    三种LTR范式对比：

    | 范式 | 损失函数 | 特点 | 代表模型 |
    |------|---------|------|---------|
    | Pointwise | 交叉熵/MSE | 独立打分 | LR/GBDT |
    | Pairwise | BPR/Hinge | 学习偏序 | RankNet/LambdaRank |
    | Listwise | ListMLE/ApproxNDCG | 优化列表指标 | LambdaMART |
    """

    def __init__(self):
        # 特征分组：搜索排序的特征按来源分为四大类
        self.feature_groups = {
            "Query特征": [         # 描述搜索词本身的属性
                "query_length",        # Query长度（字数/词数）
                "query_intent",        # 意图类别（导航/信息/交易）
                "query_frequency",     # 搜索频次（热门词 vs 长尾词）
            ],
            "Doc特征": [          # 描述候选文档/商品的质量
                "title_quality",       # 标题质量分（可读性/规范性）
                "doc_freshness",       # 内容新鲜度（时间衰减因子）
                "authority_score",     # 权威性/PageRank（来源可信度）
                "click_rate",          # 历史点击率（曝光后的点击比例）
            ],
            "Query-Doc交叉特征": [ # 衡量搜索词与文档的匹配程度
                "bm25_score",          # BM25文本相关性（经典统计方法）
                "semantic_similarity", # 语义相似度（Embedding向量内积）
                "title_match_ratio",   # 标题匹配度（Query词在标题中的覆盖率）
                "exact_match",         # 完全匹配（布尔值，极强相关性信号）
            ],
            "用户个性化特征": [   # 利用用户历史行为进行个性化排序
                "user_category_pref",  # 类目偏好（基于历史点击统计）
                "user_brand_pref",     # 品牌偏好（购买/收藏品牌分布）
                "user_price_range",    # 价格偏好（历史消费价格区间）
                "click_history_sim",   # 历史点击相似度（当前Doc与历史点击的相似度）
            ]
        }

    def evaluation_metrics(self):
        """搜索排序评估指标"""
        return {
            "NDCG@K": "归一化折损累计增益，衡量排序质量",
            "MRR": "第一个相关结果的排名倒数",
            "MAP": "平均精度均值",
            "P@K": "TopK中的准确率",
            "离线-在线GAP": "离线指标提升不等于在线收益，"
                          "必须A/B测试验证"
        }

"""模型推理优化方案全览"""

INFERENCE_OPTIMIZATION = {
    "模型层面": {
        "量化(Quantization)": {
            "INT8量化": "精度损失<1%，推理速度提升2-4x",
            "FP16混合精度": "几乎无损，速度提升1.5-2x",
            "INT4/GPTQ": "LLM专用，内存降4x",
            "校准方法": "Post-Training Quantization / "
                       "Quantization-Aware Training"
        },
        "剪枝(Pruning)": {
            "非结构化剪枝": "置零权重，需硬件支持稀疏计算",
            "结构化剪枝": "删除整个通道/Head，直接加速"
        },
        "蒸馏(Distillation)": {
            "思路": "大模型Teacher → 小模型Student",
            "场景": "线上用Student模型，保持90%+精度",
            "代表": "DistilBERT(BERT→6层), TinyLlama"
        },
        "模型编译": {
            "TensorRT": "NVIDIA GPU，最优的推理优化",
            "ONNX Runtime": "跨平台，CPU/GPU通用",
            "TVM/XLA": "自动算子优化和内存调度"
        }
    },
    "系统层面": {
        "Dynamic Batching": {
            "原理": "收集多个请求合并为一个batch推理",
            "参数": "max_batch_size=64, max_delay=10ms",
            "收益": "GPU利用率从20%→80%+"
        },
        "Continuous Batching": {
            "原理": "LLM场景，不同请求不同生成长度",
            "实现": "vLLM PagedAttention，减少GPU内存浪费",
            "收益": "LLM吞吐量提升2-4x"
        },
        "KV Cache优化": {
            "PagedAttention": "分页管理KV Cache（vLLM）",
            "Prefix Caching": "缓存公共前缀，减少重复计算",
            "Multi-Query Attention": "多query共享KV Head"
        },
        "模型并行": {
            "Tensor Parallelism": "模型参数切分到多GPU",
            "Pipeline Parallelism": "模型层切分到多GPU",
            "Speculative Decoding": "小模型猜测 + 大模型验证"
        }
    }
}

"""推理延迟优化实战方案"""

class LatencyOptimization:
    """
    目标：推理延迟 P99 < 100ms

    延迟分解（典型推荐系统精排）：
    ┌────────────────────────────────────────┐
    │ 总延迟 = 特征获取 + 模型推理 + 后处理    │
    │         30ms      + 40ms    + 10ms     │
    │         = 80ms (P50)                    │
    │         = 120ms (P99) ← 需要优化！       │
    └────────────────────────────────────────┘
    """

    def optimize_feature_fetch(self):
        """特征获取优化：30ms → 15ms——减少网络往返和计算开销"""
        return {
            "并行获取": "用户特征/物品特征/上下文特征 并行请求Redis",
            "本地缓存": "热门物品特征LRU缓存（命中率>80%）",
            "特征预计算": "用户画像T+1更新，不用实时计算",
            "Pipeline": "特征获取与模型预热并行"
        }

    def optimize_inference(self):
        """模型推理优化：40ms → 20ms——从计算图和GPU利用率着手"""
        return {
            "TensorRT编译": "FP16 + layer fusion → 2x加速",
            "Dynamic Batching": "合并多个请求提升GPU利用率",
            "算子融合": "多个小算子合并为一个大算子",
            "GPU内存池": "预分配显存，避免malloc延迟"
        }

    def optimize_end_to_end(self):
        """端到端优化——从流水线架构和降级方案两个维度优化"""
        return {
            "异步Pipeline": "召回/粗排/精排 流水线并行",
            "提前截断": "精排200候选足够，不需要更多",
            "降级方案": {
                "P99>200ms": "切换到轻量模型",
                "GPU故障": "CPU Fallback（精度-5%但可用）",
                "特征缺失": "用默认值填充，不阻塞推理"
            }
        }

"""工业级特征设计模板"""

FEATURE_DESIGN = {
    "用户侧特征": {
        "静态特征": ["user_id", "age", "gender", "city", "注册天数"],
        "行为统计": [
            "最近7天点击次数", "最近30天购买金额",
            "最近3个月活跃天数", "最近1天观看时长"
        ],
        "行为序列": [
            "最近50个点击Item ID序列",
            "最近20个购买Item ID序列",
            "最近搜索Query序列"
        ],
        "实时特征": [
            "当前Session点击次数",
            "距上次活跃时间",
            "当前页面停留时长"
        ]
    },
    "物品侧特征": {
        "基础属性": ["item_id", "类目", "品牌", "价格", "发布时间"],
        "统计特征": [
            "7日点击率", "30日转化率", "平均评分",
            "评价数", "收藏数"
        ],
        "内容特征": [
            "标题Embedding", "图片Embedding",
            "标签/关键词", "类目路径"
        ]
    },
    "交叉特征": {
        "User-Item交叉": [
            "用户对该类目的历史点击率",
            "用户对该品牌的偏好度",
            "用户历史价格范围 vs 物品价格"
        ],
        "Context特征": [
            "时间（工作日/周末, 上午/下午/晚上）",
            "设备（iOS/Android/PC）",
            "网络（WiFi/4G/5G）",
            "展示位置（第几位）"
        ]
    }
}

"""生产级RAG系统架构设计"""

class ProductionRAG:
    """
    RAG系统设计要点：
    1. 索引构建Pipeline
    2. 查询理解与路由
    3. 检索与重排
    4. 生成与后处理
    5. 评估与迭代
    """

    def indexing_pipeline(self):
        """索引构建Pipeline——将原始文档转化为可检索的向量索引"""
        return {
            "文档处理": {
                "解析": "PDF/Word/HTML → 纯文本（Unstructured）",
                "分块": {
                    "固定大小": "chunk_size=512, overlap=50",
                    "语义分块": "基于句子边界 + 主题切换",
                    "递归分块": "先段落→再句子→再固定长度",
                    "最佳实践": "chunk_size=500-1000 tokens, overlap=10-20%"
                },
                "增强": "标题+摘要 作为元数据一同嵌入"
            },
            "向量化": {
                "Embedding模型": "bge-large-zh / text-embedding-3-large",
                "多路索引": "稠密向量 + 稀疏向量(BM25) + 知识图谱",
                "存储": "Milvus/Qdrant/Weaviate"
            },
            "增量更新": {
                "策略": "CDC(Change Data Capture) → 增量索引",
                "去重": "基于文档hash去重 + 版本管理",
                "刷新": "定时全量重建 + 实时增量更新"
            }
        }

    def query_pipeline(self):
        """查询Pipeline——从用户Query到最终回答的全链路处理"""
        return {
            "查询理解": {
                "Query改写": "LLM将用户问题改写为更适合检索的query",
                "HyDE": "用LLM先生成假设答案，用答案做检索",
                "多Query": "将复杂问题拆分为子查询，分别检索"
            },
            "检索策略": {
                "混合检索": "Dense(0.7) + Sparse BM25(0.3)",
                "重排序": "Cross-encoder reranker (bge-reranker-v2)",
                "TopK选择": "检索Top20 → 重排后取Top5"
            },
            "上下文构建": {
                "窗口扩展": "召回chunk的前后各1个chunk",
                "压缩": "LongContext → 压缩器提取关键信息",
                "排序": "最相关的chunk离prompt最近"
            },
            "生成": {
                "Prompt": "基于以下参考信息回答问题：\n{context}\n问题：{query}",
                "Citation": "要求模型标注引用来源[1][2]...",
                "后处理": "检测幻觉 + 一致性验证"
            }
        }

    def evaluation(self):
        """多层次RAG评估体系：检索质量 + 生成质量 + 端到端效果"""
        return {
            "检索评估": {
                "Recall@K": "召回相关文档的比例",
                "MRR": "第一个相关文档的排名",
                "Context Relevance": "检索结果与问题的相关性"
            },
            "生成评估": {
                "Faithfulness": "回答是否忠于检索到的信息",
                "Answer Relevance": "回答是否切题",
                "Hallucination Rate": "幻觉率"
            },
            "端到端评估": {
                "EM/F1": "与标准答案的匹配度",
                "LLM-as-Judge": "用GPT-4评分",
                "人工评估": "相关性/准确性/完整性 三维评分"
            },
            "框架": "RAGAS / TruLens / DeepEval"
        }

"""LLM Gateway核心设计"""

class LLMGateway:
    """
    LLM Gateway是LLM应用的统一入口，
    负责路由、缓存、安全、监控。

    面试关键点：
    1. 为什么需要Gateway？（统一管理多Provider）
    2. 语义缓存如何实现？（embedding相似度）
    3. 如何做成本优化？（模型路由策略）
    """

    def routing_strategy(self):
        """智能模型路由——根据任务复杂度、成本、可用性选择最优模型"""
        return {
            "基于任务复杂度": {
                "简单任务": "GPT-3.5 / Claude Haiku（便宜快速）",
                "复杂推理": "GPT-4o / Claude Opus（高质量）",
                "代码生成": "Claude Sonnet / GPT-4o",
                "判断方法": "基于Query分类器 或 级联调用"
            },
            "基于成本": {
                "策略": "先用小模型，不满意再升级大模型",
                "实现": "小模型输出 → 质量评估器 → "
                       "通过则返回 / 不通过则调用大模型"
            },
            "基于可用性": {
                "主备切换": "OpenAI故障 → 自动切换Anthropic",
                "负载均衡": "多API Key轮询避免限流",
                "降级": "大模型过载 → 降级到小模型"
            }
        }

    def semantic_cache(self):
        """语义缓存——基于向量相似度复用历史回答，降低LLM调用成本"""
        return {
            "原理": "相似的Query返回缓存的响应",
            "实现步骤": [
                "1. Query → Embedding",
                "2. 在向量数据库中搜索最相似的历史Query",
                "3. 相似度 > 阈值(0.95) → 返回缓存",
                "4. 否则 → 调用LLM → 结果写入缓存"
            ],
            "效果": "缓存命中率20-40%，节省对应成本",
            "注意": "时效性内容不适合缓存，需设置TTL"
        }

    def safety(self):
        """安全防护——输入过滤、输出审核、全链路审计三层防线"""
        return {
            "输入安全": {
                "Prompt注入检测": "规则 + 分类模型双重检测",
                "PII检测": "检测并脱敏个人信息",
                "内容过滤": "敏感词 + 内容分类"
            },
            "输出安全": {
                "输出过滤": "检测有害内容/虚假信息",
                "格式验证": "JSON Schema验证结构化输出",
                "引用验证": "检查引用来源的真实性"
            },
            "审计日志": "所有请求/响应完整记录，支持回溯"
        }

"""ML系统监控全景"""

ML_MONITORING = {
    "数据质量监控": {
        "Schema检查": "字段类型/非空约束/值域范围",
        "分布偏移": {
            "PSI": "Population Stability Index > 0.2 告警",
            "KS检验": "分布差异检验",
            "监控频率": "每小时检查特征分布"
        },
        "新鲜度": "数据延迟超过SLA → 告警",
        "覆盖率": "特征缺失率 > 5% → 告警"
    },
    "模型性能监控": {
        "精度指标": {
            "实时AUC": "每小时计算，下降>1%告警",
            "转化率": "日环比，下降>5%告警",
            "NDCG@10": "搜索质量，每天评估"
        },
        "系统指标": {
            "推理延迟": "P50/P95/P99",
            "吞吐量": "QPS",
            "错误率": "推理失败比例",
            "GPU利用率": "GPU Memory / GPU Compute"
        },
        "业务指标": {
            "CTR": "点击率",
            "CVR": "转化率",
            "ARPU": "用户平均收入",
            "DAU/留存": "用户活跃度"
        }
    },
    "模型漂移检测": {
        "概念漂移": "P(Y|X)变化 — 用户偏好变了",
        "数据漂移": "P(X)变化 — 输入分布变了",
        "检测方法": [
            "ADWIN: 自适应窗口检测",
            "DDM: 基于学习器错误率",
            "Page-Hinkley: 均值变化检测"
        ],
        "应对策略": {
            "轻微漂移": "增量训练/微调",
            "显著漂移": "全量重训",
            "突变": "立即回滚到上一版本 + 排查原因"
        }
    }
}

"""A/B测试系统设计"""

class ABTestPlatform:
    """
    ML模型上线必须经过严格的A/B测试

    面试关键问题：
    1. 如何计算需要的实验样本量？
    2. 如何避免辛普森悖论？
    3. 如何处理多个实验的干扰？
    """

    def experiment_design(self):
        """实验设计——覆盖样本量计算、分流方案和统计检验方法"""
        return {
            "样本量计算": {
                "公式": "n = (Z_α/2 + Z_β)² × 2σ² / δ²",
                "参数": {
                    "α": "显著性水平，通常0.05",
                    "β": "1-Power，通常0.2（Power=80%）",
                    "δ": "最小可检测效应(MDE)",
                    "σ": "指标标准差"
                },
                "经验值": "CTR类实验通常需要 每组 10万+ 样本"
            },
            "分流方案": {
                "用户级分流": "hash(user_id) % 100",
                "层级设计": "多个实验层互不干扰",
                "AA测试": "先跑AA验证分流无偏"
            },
            "统计检验": {
                "T检验": "连续指标（如观看时长）",
                "Z检验": "比例指标（如CTR）",
                "Bootstrap": "复杂指标（如人均GMV）",
                "多重检验校正": "Bonferroni / FDR"
            }
        }

    def common_pitfalls(self):
        """避坑指南——A/B测试中的常见统计陷阱和偏差问题"""
        return {
            "过早停止": "看到显著就停，增加假阳性",
            "辛普森悖论": "分层分析 vs 总体分析结论相反",
            "新奇效应": "用户对新功能的短期兴奋",
            "长期效应": "短期正向但长期负面（如更多推送）",
            "网络效应": "社交场景中实验组和对照组互相影响",
            "SRM": "Sample Ratio Mismatch，分流比例异常"
        }

"""广告CTR预估系统设计要点"""

CTR_SYSTEM = {
    "问题定义": {
        "目标": "预估P(click|user, ad, context)",
        "收入公式": "Revenue = Σ(bid × P(click)) — eCPM排序",
        "约束": "推理延迟<10ms（广告对延迟要求极高）"
    },
    "模型演进": {
        "V1 LR": "手工特征 + LR，简单但需大量特征工程",
        "V2 GBDT+LR": "GBDT自动交叉特征 → LR",
        "V3 Wide&Deep": "Wide(记忆) + Deep(泛化)",
        "V4 DIN": "用户行为序列 + Target Attention",
        "V5 DIEN": "兴趣进化网络，GRU建模兴趣演变",
        "V6 多任务": "MMOE预测 click/convert/uninstall",
        "当前": "多任务 + 行为序列 + 实时特征 + 因果去偏"
    },
    "样本工程": {
        "正样本": "点击",
        "负样本": "曝光未点击（注意位置偏差）",
        "校准": "Platt Scaling校准预测概率",
        "延迟转化": "CVR使用延迟反馈模型（FSIW/DFM）"
    },
    "部署考虑": {
        "延迟要求": "<10ms，必须极致优化",
        "量化": "INT8量化 + TensorRT编译",
        "特征缓存": "用户特征预加载，物品特征本地缓存",
        "模型更新": "实时增量更新（Online Learning）"
    }
}

"""常见深度追问及回答思路"""

DEEP_DIVE_QUESTIONS = {
    "如果数据有偏差怎么办？": {
        "回答思路": [
            "1. 识别偏差类型：位置偏差/选择偏差/曝光偏差",
            "2. 数据层面：IPW逆倾向加权/因果推断",
            "3. 模型层面：Position-aware模型/去偏正则",
            "4. 评估层面：离线评估使用无偏估计方法"
        ]
    },
    "冷启动怎么处理？": {
        "回答思路": [
            "新用户：基于人口属性相似用户的推荐 + 热门补充",
            "新物品：基于内容特征（标题/标签Embedding）召回",
            "跨域迁移：从其他产品线迁移用户偏好",
            "主动探索：Bandit方法（Thompson Sampling）"
        ]
    },
    "模型更新频率怎么确定？": {
        "回答思路": [
            "取决于数据分布变化速度",
            "广告系统：实时/小时级（用户兴趣快速变化）",
            "推荐系统：天级增量 + 周级全量",
            "搜索排序：周级更新（查询模式相对稳定）",
            "监控漂移指标自动触发重训"
        ]
    },
    "如何处理模型和业务指标不一致？": {
        "回答思路": [
            "离线AUC提升但线上CTR没变 → 可能是离线评估方法有问题",
            "CTR提升但留存下降 → 模型过度优化短期指标",
            "解法：多目标优化 + 设置护栏指标(Guardrail Metrics)",
            "长期指标纳入评估（如留存/LTV不能降低）"
        ]
    },
    "系统延迟怎么优化到<10ms？": {
        "回答思路": [
            "1. 减少候选：更激进的截断（精排只看100个）",
            "2. 模型压缩：量化 + 蒸馏 + 剪枝",
            "3. 系统优化：Pipeline并行 + 本地缓存 + GPU优化",
            "4. 架构优化：预计算（用户特征预加载到GPU）",
            "5. 降级方案：超时用简单模型兜底"
        ]
    },
    "如何保证训练和推理一致性？": {
        "回答思路": [
            "Feature Store统一管理：训练和推理用同一份特征定义",
            "Point-in-time join：训练时用历史快照避免信息泄露",
            "Feature validation：自动对比训练特征和线上特征分布",
            "End-to-end测试：用线上数据回放验证推理结果"
        ]
    }
}