Prompt 工程化生产实践：版本管理与 A/B 测试

prompts/
├── config.yaml                 # 全局配置
├── chatbot/                    # 功能模块
│   ├── meta.yaml               # 模块元数据
│   ├── v1.0.0/                 # 语义化版本
│   │   ├── system.txt          # 系统提示
│   │   ├── few_shot.jsonl      # Few-shot 示例
│   │   └── config.yaml         # 模型参数
│   ├── v1.1.0/
│   │   ├── system.txt
│   │   ├── few_shot.jsonl
│   │   └── config.yaml
│   └── v2.0.0/                 # 大版本变更
│       ├── system.txt
│       ├── few_shot.jsonl
│       └── config.yaml
└── classifier/
    └── ...

import yaml
import json
from pathlib import Path
from dataclasses import dataclass, field
from typing import Any

@dataclass
class PromptVersion:
    """单个 Prompt 版本"""
    name: str
    version: str
    system: str
    few_shot: list[dict] = field(default_factory=list)
    model: str = "gpt-4o-mini"
    temperature: float = 0.7
    max_tokens: int = 1024
    status: str = "active"  # draft | testing | active | archived
    
    def render(self, user_input: str) -> list[dict]:
        """渲染为 API 消息格式"""
        messages = [{"role": "system", "content": self.system}]
        for example in self.few_shot:
            messages.append({"role": example["role"], "content": example["content"]})
        messages.append({"role": "user", "content": user_input})
        return messages


class PromptRegistry:
    """Prompt 版本注册中心"""
    
    def __init__(self, base_dir: str = "prompts"):
        self.base_dir = Path(base_dir)
        self._cache: dict[str, PromptVersion] = {}
    
    def load(self, name: str, version: str = "latest") -> PromptVersion:
        """加载指定版本的 Prompt"""
        if version == "latest":
            version = self._get_latest_version(name)
        
        cache_key = f"{name}@{version}"
        if cache_key in self._cache:
            return self._cache[cache_key]
        
        version_dir = self.base_dir / name / f"v{version}"
        
        # 加载系统提示
        system = (version_dir / "system.txt").read_text(encoding="utf-8")
        
        # 加载 Few-shot
        few_shot = []
        few_shot_path = version_dir / "few_shot.jsonl"
        if few_shot_path.exists():
            for line in few_shot_path.read_text(encoding="utf-8").strip().split("\n"):
                few_shot.append(json.loads(line))
        
        # 加载配置
        config_path = version_dir / "config.yaml"
        config = yaml.safe_load(config_path.read_text(encoding="utf-8"))
        
        pv = PromptVersion(
            name=name,
            version=version,
            system=system,
            few_shot=few_shot,
            model=config.get("model", "gpt-4o-mini"),
            temperature=config.get("temperature", 0.7),
            max_tokens=config.get("max_tokens", 1024),
            status=config.get("status", "active"),
        )
        
        self._cache[cache_key] = pv
        return pv
    
    def _get_latest_version(self, name: str) -> str:
        """获取最新 active 版本"""
        module_dir = self.base_dir / name
        versions = []
        for d in module_dir.iterdir():
            if d.is_dir() and d.name.startswith("v"):
                config = yaml.safe_load((d / "config.yaml").read_text(encoding="utf-8"))
                if config.get("status") == "active":
                    versions.append(d.name[1:])  # 去掉 'v' 前缀
        versions.sort(key=lambda v: [int(x) for x in v.split(".")])
        return versions[-1] if versions else "1.0.0"
    
    def diff(self, name: str, v1: str, v2: str) -> dict:
        """对比两个版本的差异"""
        p1 = self.load(name, v1)
        p2 = self.load(name, v2)
        return {
            "system_changed": p1.system != p2.system,
            "few_shot_changed": p1.few_shot != p2.few_shot,
            "model_changed": p1.model != p2.model,
            "temperature_changed": p1.temperature != p2.temperature,
        }


# 使用示例
registry = PromptRegistry("prompts")
prompt = registry.load("chatbot", "latest")
messages = prompt.render("你好，帮我查一下订单")

import random
import hashlib
from dataclasses import dataclass, field
from collections import defaultdict
import time

@dataclass
class ABTestConfig:
    """A/B 测试配置"""
    test_name: str
    variants: dict[str, PromptVersion]  # variant_name -> Prompt
    traffic_split: dict[str, float]     # variant_name -> 流量比例
    metrics: list[str] = field(default_factory=lambda: [
        "user_satisfaction", "response_length", "latency_ms", "cost"
    ])
    min_sample_size: int = 100
    
    def assign(self, user_id: str) -> str:
        """基于用户 ID 确定性分配变体（同一用户始终进入同一组）"""
        hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
        ratio = (hash_val % 10000) / 10000.0
        
        cumulative = 0.0
        for variant, weight in self.traffic_split.items():
            cumulative += weight
            if ratio < cumulative:
                return variant
        return list(self.traffic_split.keys())[-1]


@dataclass
class ExperimentResult:
    variant: str
    user_id: str
    metric: str
    value: float
    timestamp: float = field(default_factory=time.time)


class ABTestRunner:
    """A/B 测试运行器"""
    
    def __init__(self):
        self.results: list[ExperimentResult] = []
    
    def run(
        self,
        test_config: ABTestConfig,
        user_id: str,
        user_input: str,
        execute_fn,  # callable: (PromptVersion, str) -> dict
    ) -> dict:
        """执行一次 A/B 测试请求"""
        variant_name = test_config.assign(user_id)
        prompt = test_config.variants[variant_name]
        
        # 执行并收集指标
        result = execute_fn(prompt, user_input)
        
        # 记录指标
        for metric in test_config.metrics:
            if metric in result:
                self.results.append(ExperimentResult(
                    variant=variant_name,
                    user_id=user_id,
                    metric=metric,
                    value=result[metric],
                ))
        
        return {"variant": variant_name, "result": result}
    
    def analyze(self, test_name: str) -> dict:
        """分析实验结果"""
        stats = defaultdict(lambda: defaultdict(list))
        
        for r in self.results:
            stats[r.variant][r.metric].append(r.value)
        
        report = {}
        for variant, metrics in stats.items():
            report[variant] = {}
            for metric, values in metrics.items():
                vals = sorted(values)
                report[variant][metric] = {
                    "count": len(vals),
                    "mean": sum(vals) / len(vals),
                    "median": vals[len(vals) // 2],
                    "p95": vals[int(len(vals) * 0.95)] if len(vals) > 20 else None,
                }
        
        return report
    
    def is_significant(self, test_name: str, metric: str, alpha: float = 0.05) -> bool:
        """简单的统计显著性检验（Z 检验）"""
        import math
        
        variants = [r.variant for r in self.results if r.metric == metric]
        if len(set(variants)) < 2:
            return False
        
        # 按 variant 分组
        groups = defaultdict(list)
        for r in self.results:
            if r.metric == metric:
                groups[r.variant].append(r.value)
        
        if len(groups) < 2:
            return False
        
        v1, v2 = list(groups.keys())[:2]
        s1, s2 = groups[v1], groups[v2]
        
        if len(s1) < 30 or len(s2) < 30:
            return False  # 样本不足
        
        m1, m2 = sum(s1) / len(s1), sum(s2) / len(s2)
        var1 = sum((x - m1) ** 2 for x in s1) / len(s1)
        var2 = sum((x - m2) ** 2 for x in s2) / len(s2)
        
        se = math.sqrt(var1 / len(s1) + var2 / len(s2))
        if se == 0:
            return False
        
        z = abs(m1 - m2) / se
        return z > 1.96  # 95% 置信度


# 使用示例
runner = ABTestRunner()

test_config = ABTestConfig(
    test_name="chatbot_tone_v2",
    variants={
        "control": registry.load("chatbot", "1.0.0"),
        "treatment": registry.load("chatbot", "1.1.0"),
    },
    traffic_split={"control": 0.5, "treatment": 0.5},
)

def execute_fn(prompt: PromptVersion, user_input: str) -> dict:
    # 实际调用 LLM
    return {
        "user_satisfaction": 4.5,  # 用户评分
        "response_length": 320,
        "latency_ms": 850,
        "cost": 0.003,
    }

# 模拟 200 次请求
for i in range(200):
    runner.run(test_config, f"user-{i}", "帮我查订单", execute_fn)

# 分析结果
report = runner.analyze("chatbot_tone_v2")
for variant, metrics in report.items():
    print(f"\n=== {variant} ===")
    for metric, stats in metrics.items():
        print(f"  {metric}: mean={stats['mean']:.2f}, p95={stats['p95']}")

print(f"\n统计显著: {runner.is_significant('chatbot_tone_v2', 'user_satisfaction')}")

@dataclass
class TestCase:
    """评测用例"""
    id: str
    input: str
    expected_keywords: list[str]      # 期望出现的关键词
    expected_format: str | None       # 期望格式: json, markdown, etc.
    max_latency_ms: int = 5000
    category: str = "general"


class PromptEvaluator:
    """Prompt 回归评测器"""
    
    def __init__(self, test_cases: list[TestCase]):
        self.test_cases = test_cases
    
    def evaluate(
        self,
        prompt: PromptVersion,
        execute_fn,
    ) -> dict:
        """执行完整评测"""
        results = []
        
        for tc in self.test_cases:
            messages = prompt.render(tc.input)
            start = time.time()
            response = execute_fn(messages, prompt.model)
            latency = (time.time() - start) * 1000
            
            # 评估
            score = self._score(response, tc, latency)
            results.append(score)
        
        # 汇总
        pass_rate = sum(1 for r in results if r["passed"]) / len(results)
        avg_latency = sum(r["latency_ms"] for r in results) / len(results)
        
        return {
            "prompt_version": f"{prompt.name}@{prompt.version}",
            "total_cases": len(results),
            "pass_rate": pass_rate,
            "avg_latency_ms": avg_latency,
            "failed_cases": [r for r in results if not r["passed"]],
        }
    
    def _score(self, response: str, tc: TestCase, latency: float) -> dict:
        """评分逻辑"""
        keyword_hit = all(kw in response for kw in tc.expected_keywords)
        format_ok = True
        if tc.expected_format == "json":
            import json
            try:
                json.loads(response)
            except json.JSONDecodeError:
                format_ok = False
        latency_ok = latency < tc.max_latency_ms
        
        return {
            "case_id": tc.id,
            "passed": keyword_hit and format_ok and latency_ok,
            "keyword_hit": keyword_hit,
            "format_ok": format_ok,
            "latency_ms": latency,
            "latency_ok": latency_ok,
            "response_preview": response[:200],
        }


# 评测用例示例
test_cases = [
    TestCase(
        id="tc_001",
        input="订单 #12345 的状态",
        expected_keywords=["已发货", "物流"],
        expected_format=None,
        category="order_query",
    ),
    TestCase(
        id="tc_002",
        input="退货流程是什么",
        expected_keywords=["退货", "申请", "审核"],
        expected_format=None,
        category="policy",
    ),
]

evaluator = PromptEvaluator(test_cases)

class GradualRollout:
    """Prompt 灰度发布"""
    
    stages = [
        {"name": "internal", "traffic": 0.0, "duration_hours": 2},
        {"name": "canary_1", "traffic": 0.05, "duration_hours": 4},
        {"name": "canary_5", "traffic": 0.20, "duration_hours": 8},
        {"name": "canary_50", "traffic": 0.50, "duration_hours": 24},
        {"name": "full", "traffic": 1.0, "duration_hours": 0},
    ]
    
    def check_stage_health(self, stage: str, metrics: dict) -> bool:
        """检查当前阶段是否健康"""
        thresholds = {
            "error_rate": 0.05,      # 错误率 < 5%
            "latency_p95": 3000,      # P95 延迟 < 3s
            "satisfaction": 3.5,      # 满意度 > 3.5/5
            "format_compliance": 0.95,  # 格式合规率 > 95%
        }
        
        for key, threshold in thresholds.items():
            if key in metrics:
                if key in ("error_rate", "latency_p95"):
                    if metrics[key] > threshold:
                        return False
                else:
                    if metrics[key] < threshold:
                        return False
        return True
    
    def should_advance(self, current_stage: str, metrics: dict) -> bool:
        """是否应该推进到下一阶段"""
        return self.check_stage_health(current_stage, metrics)

新 Prompt 创建
    │
    ▼
内部测试 (0% 流量, 2h)
    │ ✅ 通过
    ▼
金丝雀 5% (4h)
    │ ✅ 指标正常
    ▼
灰度 20% (8h)
    │ ✅ 指标正常
    ▼
灰度 50% (24h)
    │ ✅ 指标正常
    ▼
全量发布 (100%)
    │
    └── ❌ 任何阶段失败 → 自动回滚到上一版本

# .github/workflows/prompt-ci.yml
name: Prompt CI
on:
  pull_request:
    paths: ["prompts/**"]

jobs:
  prompt-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Validate Prompt Structure
        run: python scripts/validate_prompts.py prompts/
      
      - name: Run Regression Tests
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python scripts/evaluate_prompt.py \
            --prompt chatbot@pr-${{ github.event.pull_request.number }} \
            --test-suite regression_v3 \
            --min-pass-rate 0.90
      
      - name: Token Count Check
        run: |
          python scripts/check_tokens.py \
            --prompt prompts/chatbot/ \
            --max-system-tokens 500 \
            --max-fewshot-tokens 1000
      
      - name: Cost Estimation
        run: |
          python scripts/estimate_cost.py \
            --prompt chatbot@latest \
            --daily-volume 100000

## Prompt 变更申请

### 变更内容
- 模块: chatbot
- 版本: 1.0.0 → 1.1.0
- 类型: Minor

### 变更原因
用户反馈当前回复过于冗长，需要精简输出格式。

### 回归测试结果
- 通过率: 94/100 (94%)
- 新增失败: tc_078 (边界场景)
- 已知问题: 无影响

### A/B 测试数据
- 实验周期: 7 天
- 样本量: 5,000
- 满意度提升: +8.2% (p < 0.01)
- 响应长度减少: -23%
- 成本节省: -15%

### 回滚方案
将 active 标记切回 v1.0.0，预计 < 1 分钟。