Prompt版本控制与A/B测试：数据驱动的Prompt优化

from dataclasses import dataclass
from typing import Optional
import hashlib

@dataclass
class PromptVersion:
    """Prompt版本"""
    major: int      # 不兼容的修改（如输出格式变更）
    minor: int      # 向后兼容的功能新增
    patch: int      # Bug修复和微调
    hash: str       # 内容哈希
    
    @property
    def version_string(self) -> str:
        return f"v{self.major}.{self.minor}.{self.patch}"
    
    @staticmethod
    def compute_hash(content: str) -> str:
        return hashlib.sha256(content.encode()).hexdigest()[:12]

class VersionedPrompt:
    """带版本控制的Prompt"""
    
    def __init__(self):
        self.versions: list[dict] = []
        self.active_version: Optional[str] = None
    
    def commit(self, prompt_content: str, 
               change_type: str = "patch",
               changelog: str = "") -> str:
        """
        提交新版本
        change_type: major / minor / patch
        """
        # 确定版本号
        if not self.versions:
            version = PromptVersion(1, 0, 0, "")
        else:
            latest = self.versions[-1]["version"]
            if change_type == "major":
                version = PromptVersion(latest.major + 1, 0, 0, "")
            elif change_type == "minor":
                version = PromptVersion(latest.major, latest.minor + 1, 0, "")
            else:
                version = PromptVersion(latest.major, latest.minor, 
                                       latest.patch + 1, "")
        
        version.hash = PromptVersion.compute_hash(prompt_content)
        
        version_id = f"{version.version_string}-{version.hash}"
        
        self.versions.append({
            "version_id": version_id,
            "version": version,
            "content": prompt_content,
            "changelog": changelog,
            "timestamp": datetime.now(),
            "is_active": False
        })
        
        return version_id
    
    def rollback(self, version_id: str):
        """回滚到指定版本"""
        for v in self.versions:
            v["is_active"] = (v["version_id"] == version_id)
        self.active_version = version_id
    
    def diff(self, version_a: str, version_b: str) -> str:
        """比较两个版本的差异"""
        import difflib
        
        content_a = next(v["content"] for v in self.versions 
                        if v["version_id"] == version_a)
        content_b = next(v["content"] for v in self.versions 
                        if v["version_id"] == version_b)
        
        diff = difflib.unified_diff(
            content_a.splitlines(keepends=True),
            content_b.splitlines(keepends=True),
            fromfile=version_a,
            tofile=version_b
        )
        
        return ''.join(diff)

from dataclasses import dataclass
from enum import Enum
import numpy as np
from scipy import stats

class MetricType(Enum):
    ACCURACY = "accuracy"
    LATENCY = "latency"
    COST = "cost"
    USER_SATISFACTION = "satisfaction"
    TASK_COMPLETION = "completion"

@dataclass
class ABTestConfig:
    """A/B测试配置"""
    name: str
    description: str
    
    # 变体
    control_version: str        # 基线版本（A）
    treatment_version: str      # 实验版本（B）
    
    # 流量分配
    traffic_split: float        # treatment流量比例 (0-1)
    
    # 指标
    primary_metric: MetricType  # 主要指标
    secondary_metrics: list[MetricType]
    
    # 统计参数
    significance_level: float = 0.05  # α
    statistical_power: float = 0.8     # 1-β
    minimum_detectable_effect: float = 0.05  # MDE
    
    # 持续时间
    min_samples_per_variant: int = 1000
    max_duration_days: int = 14


class PromptABTest:
    """Prompt A/B测试执行器"""
    
    def __init__(self, config: ABTestConfig, prompt_registry):
        self.config = config
        self.registry = prompt_registry
        self.results: dict[str, list] = {
            "control": [],
            "treatment": []
        }
    
    def assign_variant(self, user_id: str) -> str:
        """分配用户到变体"""
        # 使用用户ID哈希确保同一用户始终在同一组
        hash_value = hash(user_id) % 100 / 100
        
        if hash_value < self.config.traffic_split:
            return "treatment"
        else:
            return "control"
    
    def get_prompt(self, variant: str) -> str:
        """获取对应变体的Prompt"""
        version = (self.config.control_version if variant == "control" 
                   else self.config.treatment_version)
        return self.registry.get(version)
    
    def record_result(self, variant: str, result: dict):
        """记录实验结果"""
        self.results[variant].append(result)
    
    def analyze(self) -> dict:
        """分析实验结果"""
        control_data = [r[self.config.primary_metric.value] 
                       for r in self.results["control"]]
        treatment_data = [r[self.config.primary_metric.value] 
                         for r in self.results["treatment"]]
        
        # 描述性统计
        control_mean = np.mean(control_data)
        treatment_mean = np.mean(treatment_data)
        
        # 统计检验
        if self._is_continuous(self.config.primary_metric):
            # 连续指标：t检验
            t_stat, p_value = stats.ttest_ind(
                treatment_data, control_data
            )
            effect_size = (treatment_mean - control_mean) / np.std(control_data)
        else:
            # 二分类指标：卡方检验
            control_success = sum(control_data)
            treatment_success = sum(treatment_data)
            chi2, p_value = stats.chi2_contingency([
                [control_success, len(control_data) - control_success],
                [treatment_success, len(treatment_data) - treatment_success]
            ])[:2]
            effect_size = (treatment_mean - control_mean) / control_mean
        
        # 置信区间
        ci = self._confidence_interval(
            treatment_data, control_data, 
            self.config.significance_level
        )
        
        # 结论
        significant = p_value < self.config.significance_level
        winner = "treatment" if significant and treatment_mean > control_mean else \
                 "control" if significant else "inconclusive"
        
        return {
            "control_mean": control_mean,
            "treatment_mean": treatment_mean,
            "relative_improvement": (treatment_mean - control_mean) / control_mean,
            "p_value": p_value,
            "effect_size": effect_size,
            "confidence_interval": ci,
            "significant": significant,
            "winner": winner,
            "sample_sizes": {
                "control": len(control_data),
                "treatment": len(treatment_data)
            }
        }
    
    def _confidence_interval(self, treatment, control, alpha):
        """计算置信区间"""
        diff = np.mean(treatment) - np.mean(control)
        se = np.sqrt(np.var(treatment)/len(treatment) + 
                     np.var(control)/len(control))
        z = stats.norm.ppf(1 - alpha/2)
        return (diff - z * se, diff + z * se)

class SampleSizeCalculator:
    """计算所需样本量"""
    
    @staticmethod
    def for_proportion(baseline_rate: float, mde: float,
                       alpha: float = 0.05, power: float = 0.8) -> int:
        """
        比率指标（如准确率）的样本量计算
        
        baseline_rate: 基线比率
        mde: 最小可检测效应
        alpha: 显著性水平
        power: 统计功效
        """
        from scipy.stats import norm
        
        z_alpha = norm.ppf(1 - alpha/2)
        z_beta = norm.ppf(power)
        
        p1 = baseline_rate
        p2 = baseline_rate + mde
        p_avg = (p1 + p2) / 2
        
        n = ((z_alpha * np.sqrt(2 * p_avg * (1 - p_avg)) + 
              z_beta * np.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2) / (p1 - p2) ** 2
        
        return int(np.ceil(n))
    
    @staticmethod
    def for_continuous(baseline_std: float, mde: float,
                       alpha: float = 0.05, power: float = 0.8) -> int:
        """
        连续指标（如延迟）的样本量计算
        """
        from scipy.stats import norm
        
        z_alpha = norm.ppf(1 - alpha/2)
        z_beta = norm.ppf(power)
        
        n = 2 * ((z_alpha + z_beta) * baseline_std / mde) ** 2
        
        return int(np.ceil(n))

class MultivariatePromptTest:
    """
    多变量Prompt测试
    同时测试多个Prompt组件的变化
    """
    
    def __init__(self):
        self.factors = {}  # 因子及其变体
    
    def add_factor(self, name: str, variants: list[str]):
        """添加测试因子"""
        self.factors[name] = variants
    
    def generate_combinations(self) -> list[dict]:
        """生成所有组合"""
        from itertools import product
        
        factor_names = list(self.factors.keys())
        factor_values = list(self.factors.values())
        
        combinations = []
        for values in product(*factor_values):
            combinations.append(dict(zip(factor_names, values)))
        
        return combinations
    
    def design(self) -> dict:
        """实验设计"""
        combos = self.generate_combinations()
        
        return {
            "total_combinations": len(combos),
            "combinations": combos,
            "traffic_per_combo": 1.0 / len(combos),
            "min_samples_per_combo": self._calc_min_samples(),
            "estimated_duration_days": self._estimate_duration(),
        }

# 案例：优化电商客服Prompt

# 基线Prompt (v1.0.0)
control_prompt = """
你是一个电商客服助手。请回答用户的问题。
"""

# 实验Prompt (v1.1.0) - 添加了情绪感知和解决方案导向
treatment_prompt = """
你是一个专业的电商客服助手。

回答原则：
1. 先理解用户的情绪和核心诉求
2. 提供具体的解决方案，而非泛泛而谈
3. 如果需要转人工，明确说明原因
4. 保持友好但专业的语调

回答结构：
- 确认问题：简要复述用户的问题
- 解决方案：给出具体步骤
- 后续支持：提供额外帮助选项
"""

# 配置A/B测试
config = ABTestConfig(
    name="客服Prompt优化-情绪感知版",
    description="测试添加情绪感知和结构化回答是否提升客户满意度",
    control_version="v1.0.0",
    treatment_version="v1.1.0",
    traffic_split=0.5,
    primary_metric=MetricType.USER_SATISFACTION,
    secondary_metrics=[MetricType.TASK_COMPLETION, MetricType.LATENCY],
    minimum_detectable_effect=0.03,  # 3%提升
    min_samples_per_variant=2000,
)

# 运行测试
ab_test = PromptABTest(config, prompt_registry)

# 分析结果
results = ab_test.analyze()
"""
预期输出：
{
    "control_mean": 0.78,        # 基线满意度 78%
    "treatment_mean": 0.83,      # 实验组满意度 83%
    "relative_improvement": 0.064,  # 6.4%提升
    "p_value": 0.002,            # p < 0.05，显著
    "effect_size": 0.12,         # 小到中等效应
    "confidence_interval": (0.02, 0.08),
    "significant": True,
    "winner": "treatment"
}
"""

class ProgressiveRollout:
    """
    渐进式发布
    获胜的变体逐步增加流量
    """
    
    ROLLOUT_STAGES = [
        {"traffic": 0.05, "duration_hours": 24, "min_success_rate": 0.85},
        {"traffic": 0.20, "duration_hours": 48, "min_success_rate": 0.87},
        {"traffic": 0.50, "duration_hours": 72, "min_success_rate": 0.88},
        {"traffic": 1.00, "duration_hours": None, "min_success_rate": 0.88},
    ]
    
    async def rollout(self, prompt_version: str):
        """渐进式发布"""
        for stage in self.ROLLOUT_STAGES:
            # 设置流量
            await self.traffic_manager.set_traffic(
                prompt_version, stage["traffic"]
            )
            
            # 等待观察期
            if stage["duration_hours"]:
                await asyncio.sleep(stage["duration_hours"] * 3600)
                
                # 检查指标
                success_rate = await self.metrics.get_success_rate(
                    prompt_version,
                    window_hours=stage["duration_hours"]
                )
                
                if success_rate < stage["min_success_rate"]:
                    # 回滚
                    await self.traffic_manager.rollback(prompt_version)
                    await self.alerting.notify(
                        f"发布失败：成功率 {success_rate:.1%} < "
                        f"阈值 {stage['min_success_rate']:.1%}"
                    )
                    return False
        
        return True