Prompt CI/CD：版本、评测、发布门禁与回滚

{
  "releaseId": "support-rag/sha256:4e89c6...",
  "sourceCommit": "9f5cb7...",
  "prompt": {
    "artifact": "prompts/support-answer.json",
    "sha256": "c8f4a2..."
  },
  "model": {
    "provider": "provider-a",
    "revision": "immutable-model-revision",
    "parameters": {
      "temperature": 0.2,
      "maxOutputTokens": 900
    }
  },
  "tools": {
    "schemaSha256": "2a70b1...",
    "policySha256": "79945d..."
  },
  "retrieval": {
    "corpusSnapshot": "support-docs-184",
    "indexBuild": "index-2026-08-09-03"
  },
  "output": {
    "schemaSha256": "b5cb3a...",
    "parserRevision": "parser/7d34fe..."
  },
  "evaluation": {
    "datasetRevision": "support-eval/31",
    "runnerRevision": "eval-runner/8c18a0...",
    "judgeRevision": "judge-rubric/12"
  }
}

task: refund_explanation
locale: zh-CN
risk: financial
tenant_policy: standard
input_source: production_sample
expected_contract: cited_answer_without_mutation

from __future__ import annotations

import hashlib
import json
import random
from pathlib import Path
from statistics import fmean
from typing import Any


REQUIRED_MANIFEST_PATHS = (
    ("prompt", "sha256"),
    ("model", "revision"),
    ("tools", "schemaSha256"),
    ("tools", "policySha256"),
    ("retrieval", "indexBuild"),
    ("output", "schemaSha256"),
    ("evaluation", "datasetRevision"),
    ("evaluation", "runnerRevision"),
)


def nested_value(document: dict[str, Any], path: tuple[str, ...]) -> Any:
    value: Any = document
    for key in path:
        if not isinstance(value, dict) or key not in value:
            raise ValueError(f"missing manifest field: {'.'.join(path)}")
        value = value[key]
    return value


def release_digest(manifest: dict[str, Any]) -> str:
    for path in REQUIRED_MANIFEST_PATHS:
        nested_value(manifest, path)
    payload = json.dumps(
        manifest, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    ).encode("utf-8")
    return "sha256:" + hashlib.sha256(payload).hexdigest()


def paired_differences(
    baseline: list[dict[str, Any]],
    candidate: list[dict[str, Any]],
    metric: str,
    slice_name: str,
) -> list[float]:
    def select(rows: list[dict[str, Any]]) -> dict[tuple[str, int], float]:
        selected = {}
        for row in rows:
            if row["slice"] == slice_name:
                selected[(row["caseId"], row["repeat"])] = float(row[metric])
        return selected

    left, right = select(baseline), select(candidate)
    if left.keys() != right.keys() or not left:
        raise ValueError(f"unpaired or empty results for {metric}/{slice_name}")
    return [right[key] - left[key] for key in sorted(left)]


def bootstrap_interval(
    values: list[float], confidence: float, samples: int, seed: int
) -> tuple[float, float]:
    rng = random.Random(seed)
    estimates = sorted(
        fmean(rng.choice(values) for _ in values) for _ in range(samples)
    )
    tail = (1.0 - confidence) / 2.0
    low = estimates[int(tail * (samples - 1))]
    high = estimates[int((1.0 - tail) * (samples - 1))]
    return low, high


def evaluate_gate(
    baseline: list[dict[str, Any]],
    candidate: list[dict[str, Any]],
    policy: dict[str, Any],
) -> dict[str, Any]:
    checks = []
    for rule in policy["rules"]:
        diffs = paired_differences(
            baseline, candidate, rule["metric"], rule["slice"]
        )
        low, high = bootstrap_interval(
            diffs,
            confidence=policy["confidence"],
            samples=policy["bootstrapSamples"],
            seed=policy["seed"],
        )
        passed = (
            low >= rule["minimumDelta"]
            if rule["direction"] == "higher"
            else high <= rule["maximumDelta"]
        )
        checks.append(
            {
                "metric": rule["metric"],
                "slice": rule["slice"],
                "pairs": len(diffs),
                "meanDelta": fmean(diffs),
                "interval": [low, high],
                "passed": passed,
            }
        )
    return {"passed": all(item["passed"] for item in checks), "checks": checks}


def main() -> None:
    manifest = json.loads(Path("release-manifest.json").read_text())
    baseline = json.loads(Path("baseline-results.json").read_text())
    candidate = json.loads(Path("candidate-results.json").read_text())
    policy = json.loads(Path("release-policy.json").read_text())
    report = {
        "releaseDigest": release_digest(manifest),
        **evaluate_gate(baseline, candidate, policy),
    }
    Path("gate-report.json").write_text(
        json.dumps(report, indent=2, ensure_ascii=False) + "\n"
    )
    if not report["passed"]:
        raise SystemExit("release gate failed; inspect gate-report.json")


if __name__ == "__main__":
    main()

{
  "confidence": 0.95,
  "bootstrapSamples": 10000,
  "seed": 20260809,
  "rules": [
    {
      "metric": "taskScore",
      "slice": "all",
      "direction": "higher",
      "minimumDelta": -0.01
    },
    {
      "metric": "unsafeActionRate",
      "slice": "financial",
      "direction": "lower",
      "maximumDelta": 0.0
    }
  ]
}

name: prompt-contracts
on:
  pull_request:
    paths:
      - "prompts/**"
      - "evals/**"
      - "release-policy.json"

permissions:
  contents: read

jobs:
  deterministic-contracts:
    runs-on: ubuntu-latest
    timeout-minutes: 10
    steps:
      - name: Check trusted repository workspace
        run: |
          test -f scripts/validate_release.py
          python scripts/validate_release.py --no-network