Prompt 版本管理与 A/B 实验

prompt_id: summarize
version: 3.2.0
model: gpt-4.1-mini-2026-04-14
params:
  temperature: 0.3
  top_p: 1.0
  max_tokens: 512
variables_schema:
  type: object
  required: [document, audience, max_sentences]
  properties:
    document:
      type: string
      minLength: 1
      maxLength: 60000
    audience:
      type: string
      enum: [general, technical, executive]
    max_sentences:
      type: integer
      minimum: 1
      maximum: 10
template: |
  你是一名专业编辑。请把下面的文档总结为不超过 {{max_sentences}} 句话，
  面向 {{audience}} 读者，保留关键数字与结论，不要编造原文没有的信息。

  文档：
  {{document}}
changelog:
  - version: 3.2.0
    note: 增加 audience 变量，支持面向高管的摘要风格
  - version: 3.1.0
    note: 显式要求保留关键数字

import re
from dataclasses import dataclass, field
from typing import Any

PLACEHOLDER = re.compile(r"\{\{\s*(\w+)\s*\}\}")

@dataclass
class PromptVersion:
    prompt_id: str
    version: str
    template: str
    model: str
    params: dict[str, Any] = field(default_factory=dict)
    schema: dict[str, Any] = field(default_factory=dict)

    def validate(self, variables: dict[str, Any]) -> None:
        spec = self.schema
        required = spec.get("required", [])
        missing = [k for k in required if k not in variables]
        if missing:
            raise ValueError(f"缺少必填变量: {missing}")
        for name, rule in spec.get("properties", {}).items():
            if name not in variables:
                continue
            value = variables[name]
            if "enum" in rule and value not in rule["enum"]:
                raise ValueError(f"变量 {name} 取值非法: {value}")
            if rule.get("type") == "integer" and not isinstance(value, int):
                raise ValueError(f"变量 {name} 必须为整数")

    def render(self, variables: dict[str, Any]) -> str:
        self.validate(variables)
        used = set(PLACEHOLDER.findall(self.template))
        unused = used - variables.keys()
        if unused:
            raise ValueError(f"模板引用了未提供的变量: {unused}")
        return PLACEHOLDER.sub(lambda m: str(variables[m.group(1)]), self.template)

    def call_metadata(self) -> dict[str, Any]:
        # 注入到网关请求的 metadata，用于线上归因
        return {"prompt_id": self.prompt_id, "prompt_version": self.version}

release:
  prompt_id: summarize
  stable: 3.1.0          # 稳定版本，默认流量
  experiment:            # 实验流量
    version: 3.2.0
    weight: 0.20         # 20% 流量
    experiment_id: summarize-v3-vs-v2
    guardrails:
      max_cost_increase: 0.25
      max_p99_increase: 0.40
  updated_at: "2026-10-04T16:00:00+08:00"
  updated_by: leeting

import json
from dataclasses import dataclass

@dataclass
class GoldenCase:
    case_id: str
    variables: dict
    must_contain: list[str]      # 必须出现的关键词
    must_not_contain: list[str]  # 不得出现的词
    max_chars: int               # 输出长度上限

def run_regression(prompt: PromptVersion, cases: list[GoldenCase],
                   call_llm, threshold: float = 0.95) -> dict:
    passed, failures = 0, []
    for case in cases:
        text = prompt.render(case.variables)
        output = call_llm(prompt.model, text, prompt.params)
        ok = True
        if not all(k in output for k in case.must_contain):
            ok = False
        if any(k in output for k in case.must_not_contain):
            ok = False
        if len(output) > case.max_chars:
            ok = False
        if ok:
            passed += 1
        else:
            failures.append(case.case_id)
    rate = passed / len(cases)
    result = {"pass_rate": rate, "failures": failures, "ok": rate >= threshold}
    if not result["ok"]:
        raise AssertionError(f"回归未通过: {json.dumps(result, ensure_ascii=False)}")
    return result

import hashlib

def bucket_of(user_id: str, experiment: str, buckets: int = 10000) -> int:
    # 关键：把实验名拼进哈希，避免不同实验的相关性
    key = f"{experiment}:{user_id}".encode("utf-8")
    digest = hashlib.sha256(key).hexdigest()
    return int(digest[:8], 16) % buckets

def assign(user_id: str, experiment: str, weights: dict[str, float]) -> str:
    # weights 形如 {"A": 0.5, "B": 0.5}，按累计权重切分
    assert abs(sum(weights.values()) - 1.0) < 1e-9, "权重必须归一"
    point = bucket_of(user_id, experiment) / 10000.0
    acc = 0.0
    for variant, w in weights.items():
        acc += w
        if point < acc:
            return variant
    return list(weights)[-1]

if __name__ == "__main__":
    counts = {"A": 0, "B": 0}
    for i in range(100000):
        counts[assign(f"user-{i}", "summarize-v3-vs-v2", {"A": 0.5, "B": 0.5})] += 1
    print(counts)   # 期望接近 {"A": 50000, "B": 50000}

import math

def two_proportion_test(n_a: int, c_a: int, n_b: int, c_b: int) -> dict:
    p_a, p_b = c_a / n_a, c_b / n_b
    p_pool = (c_a + c_b) / (n_a + n_b)
    se_pool = math.sqrt(p_pool * (1 - p_pool) * (1 / n_a + 1 / n_b))
    if se_pool == 0:
        return {"lift": 0.0, "z": 0.0, "p_value": 1.0, "ci95": (0.0, 0.0)}
    z = (p_b - p_a) / se_pool
    # 双侧 p 值
    p_value = 2 * (1 - 0.5 * (1 + math.erf(abs(z) / math.sqrt(2))))
    se_diff = math.sqrt(p_a * (1 - p_a) / n_a + p_b * (1 - p_b) / n_b)
    ci = ((p_b - p_a) - 1.96 * se_diff, (p_b - p_a) + 1.96 * se_diff)
    return {
        "p_a": p_a, "p_b": p_b,
        "lift": (p_b - p_a) / p_a if p_a else float("inf"),
        "z": z, "p_value": p_value, "ci95": ci,
    }

def required_sample_size(p0: float, mde: float, alpha: float = 0.05,
                         power: float = 0.8) -> int:
    # 两比例检验每组所需样本量（近似）
    z_alpha = 1.96 if abs(alpha - 0.05) < 1e-9 else 2.576
    z_beta = 0.84 if abs(power - 0.8) < 1e-9 else 1.28
    p1 = p0 + mde
    p_bar = (p0 + p1) / 2
    num = (z_alpha * math.sqrt(2 * p_bar * (1 - p_bar))
           + z_beta * math.sqrt(p0 * (1 - p0) + p1 * (1 - p1))) ** 2
    return math.ceil(num / (mde ** 2))

if __name__ == "__main__":
    # A 组 5000 次，成功 3400；B 组 5000 次，成功 3600
    r = two_proportion_test(5000, 3400, 5000, 3600)
    print(f"pA={r['p_a']:.3f} pB={r['p_b']:.3f} lift={r['lift']:+.1%}")
    print(f"p_value={r['p_value']:.5f} ci95=({r['ci95'][0]:+.3f}, {r['ci95'][1]:+.3f})")
    print("每组所需样本:", required_sample_size(0.68, 0.02))

import time
from dataclasses import dataclass

@dataclass
class Guardrail:
    name: str
    limit: float          # 相对基线的最大劣化比例
    window_min: int       # 需连续满足的分钟数

GUARDRAILS = [
    Guardrail("cost_per_request", 0.25, 15),
    Guardrail("p99_latency_ms", 0.40, 15),
    Guardrail("error_rate", 0.50, 5),
]

def check_and_rollback(metrics_fn, registry, experiment_id: str,
                       stable_version: str) -> bool:
    breached = {}
    for g in GUARDRAILS:
        series = metrics_fn(experiment_id, g.name, g.window_min)
        if len(series) < g.window_min:
            continue
        baseline = series["baseline"]
        worst = max(series["current"])
        if baseline > 0 and (worst - baseline) / baseline > g.limit:
            breached[g.name] = worst / baseline - 1
    if breached:
        registry.set_weight(experiment_id, 0.0)
        registry.set_pointer(experiment_id, stable_version)
        print(f"[rollback] {experiment_id} 触发回滚: {breached}")
        return True
    return False

if __name__ == "__main__":
    while True:
        check_and_rollback(metrics_fn=lambda *a: {"baseline": 0.02, "current": [0.03]},
                           registry=None, experiment_id="summarize-v3-vs-v2",
                           stable_version="3.1.0")
        time.sleep(60)