Google 全球服务舰队上的大规模 A/B 实验系统：如何让分布式实验不再打架

# experiments.yaml — 分层实验定义
layers:
  ui_layer:
    description: "界面相关参数，层内实验互斥"
    experiments:
      button_color_test:
        salt: "btn_color_2024q3"
        params:
          button_color:
            control: "gray"
            treatment: "blue"
        traffic_split: { control: 50, treatment: 50 }
      font_size_test:
        salt: "font_sz_2024q3"
        params:
          font_size_px:
            control: 14
            treatment: 18
        traffic_split: { control: 50, treatment: 50 }
  algo_layer:
    description: "算法相关参数，与 ui_layer 互不影响"
    experiments:
      ranking_algo_test:
        salt: "rank_algo_v2"
        params:
          ranking_weight:
            control: 1.0
            treatment: 1.3
        traffic_split: { control: 70, treatment: 30 }

# experiment_framework.py — 可直接运行
import yaml, hashlib, json, time, logging

logging.basicConfig(level=logging.INFO, format="%(message)s")

def load_config(path="experiments.yaml"):
    with open(path) as f:
        return yaml.safe_load(f)

def assign_group(user_id: str, experiment: dict) -> str:
    """确定性分组：同一用户同一实验永远同一组"""
    salt = experiment["salt"]
    hash_val = int(hashlib.sha256(f"{user_id}:{salt}".encode()).hexdigest(), 16)
    bucket = hash_val % 100
    splits = experiment["traffic_split"]
    cumulative = 0
    for group, pct in splits.items():
        cumulative += pct
        if bucket < cumulative:
            return group
    return list(splits.keys())[-1]  # fallback

def resolve_params(user_id: str, config: dict) -> dict:
    """遍历所有层和实验，返回该用户的完整参数集"""
    params = {}
    for layer_name, layer in config["layers"].items():
        for exp_name, exp in layer["experiments"].items():
            group = assign_group(user_id, exp)
            for param_name, values in exp["params"].items():
                params[param_name] = values[group]
                # 曝光日志：在真正读取参数时记录
                log_exposure(user_id, exp_name, param_name, group, values[group])
    return params

def log_exposure(user_id, exp_name, param_name, group, value):
    """写曝光日志（生产环境换成 Kafka/BigQuery 管道）"""
    record = {
        "timestamp": time.time(),
        "user_id": user_id,
        "experiment_id": exp_name,
        "param_name": param_name,
        "group": group,
        "value": value,
    }
    logging.info(json.dumps(record, ensure_ascii=False))

# ---- 运行示例 ----
if __name__ == "__main__":
    config = load_config()
    # 模拟三个用户
    for uid in ["user_001", "user_002", "user_003"]:
        print(f"\n--- {uid} 的实验参数 ---")
        params = resolve_params(uid, config)
        print(params)

# 依赖：pip install pyyaml
python experiment_framework.py

# watch_config.sh — 每 30 秒检查配置版本，有更新就触发热加载
CONFIG_URL="http://config-center.internal/experiments.yaml"
LOCAL_PATH="/etc/app/experiments.yaml"
CURRENT_HASH=""

while true; do
    NEW_HASH=$(curl -s "$CONFIG_URL" | sha256sum | cut -d' ' -f1)
    if [ "$NEW_HASH" != "$CURRENT_HASH" ]; then
        echo "配置变更 detected，拉取新版本..."
        curl -s "$CONFIG_URL" -o "$LOCAL_PATH"
        CURRENT_HASH="$NEW_HASH"
        # 通知应用热加载（比如发信号或调 admin API）
        kill -USR1 $(pgrep -f experiment_framework) 2>/dev/null || true
        echo "已切换到配置版本 $NEW_HASH"
    fi
    sleep 30
done