为什么 LLM 应用需要特殊的 CI/CD

传统软件 CI/CD 关注代码编译、单元测试、部署。LLM 应用的 CI/CD 还需要处理：Prompt 变更的不确定性、模型版本漂移、输出质量回归、A/B 测试的统计显著性。一行 Prompt 改动可能让整个系统的回答质量崩塌，而你很难用传统测试覆盖。

Prompt 版本控制

目录结构

prompts/
├── v1/
│   ├── system.txt
│   ├── fewshot.json
│   └── config.yaml
├── v2/
│   ├── system.txt
│   ├── fewshot.json
│   └── config.yaml
└── current -> v2/    # 符号链接指向当前版本

Prompt 配置文件

# prompts/v2/config.yaml
version: "2.1.0"
model: gpt-4o
temperature: 0.3
max_tokens: 2000
system_prompt_file: system.txt
fewshot_file: fewshot.json
variables:
  - name: user_query
    required: true
  - name: context
    required: false
    default: ""
tests:
  - name: "basic_qa"
    dataset: "datasets/qa_test_100.jsonl"
    min_score: 0.85
  - name: "safety_check"
    dataset: "datasets/safety_test_50.jsonl"
    min_score: 0.98

Prompt 加载与版本注入

import yaml
from pathlib import Path

class PromptManager:
    def __init__(self, prompts_dir="prompts"):
        self.prompts_dir = Path(prompts_dir)

    def load(self, version="current"):
        path = self.prompts_dir / version
        config = yaml.safe_load((path / "config.yaml").read_text())
        system = (path / config["system_prompt_file"]).read_text()
        fewshot = json.loads(
            (path / config["fewshot_file"]).read_text()
        )
        return Prompt(
            version=config["version"],
            system=system,
            fewshot=fewshot,
            model=config["model"],
            temperature=config["temperature"],
            max_tokens=config["max_tokens"],
        )

自动评估门禁

CI 流水线中必须在部署前运行自动评估，不达标的版本被拦截。

评估流程

PR 提交 → 代码检查 → 单元测试 → 评估测试集 → 质量门禁 → 合并
                                                      ↓
                                               [未达标 → 拒绝]

评估脚本

import json
from dataclasses import dataclass

@dataclass
class EvalResult:
    total: int
    passed: int
    avg_score: float
    failures: list

async def run_eval(test_file, prompt_version, judge_model="gpt-4o"):
    cases = [json.loads(line) for line in open(test_file)]
    results = []

    for case in cases:
        # 1. 用待测 prompt 生成回答
        response = await llm.complete(
            prompt_version, case["input"]
        )
        # 2. 用 judge model 评分（LLM-as-a-Judge）
        score = await judge(
            judge_model,
            question=case["input"],
            answer=response,
            reference=case.get("expected"),
            rubric=case.get("rubric", "correctness, completeness, clarity"),
        )
        results.append({
            "case_id": case["id"],
            "score": score,
            "passed": score >= case.get("threshold", 0.8),
        })

    passed = sum(1 for r in results if r["passed"])
    avg = sum(r["score"] for r in results) / len(results)
    failures = [r for r in results if not r["passed"]]

    return EvalResult(len(results), passed, avg, failures)

async def judge(model, question, answer, reference, rubric):
    prompt = f"""请按以下标准评分（0-1）：
问题：{question}
回答：{answer}
参考答案：{reference or '无'}
评分标准：{rubric}
只输出 JSON：{{"score": 0.0, "reason": "..."}}"""
    result = await llm.complete(model, prompt)
    return json.loads(result)["score"]

门禁规则

class QualityGate:
    def __init__(self, config):
        self.rules = config["tests"]

    def check(self, results: dict[str, EvalResult]) -> bool:
        for name, result in results.items():
            rule = next(r for r in self.rules if r["name"] == name)
            if result.avg_score < rule["min_score"]:
                print(f"GATE FAILED: {name} "
                      f"avg={result.avg_score:.2f} "
                      f"required={rule['min_score']}")
                return False
            if result.passed / result.total < 0.95:
                print(f"GATE FAILED: {name} "
                      f"pass_rate={result.passed/result.total:.1%}")
                return False
        return True

GitHub Actions 流水线

# .github/workflows/llm-ci.yml
name: LLM CI/CD

on:
  pull_request:
    paths: ["prompts/**", "src/**"]
  push:
    branches: [main]

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -r requirements.txt
      - run: ruff check src/
      - run: pytest tests/unit/

  eval-gate:
    needs: lint-and-test
    runs-on: ubuntu-latest
    if: github.event_name == 'pull_request'
    steps:
      - uses: actions/checkout@v4
      - name: Run evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python scripts/run_eval.py \
            --prompt-version pr-${{ github.event.pull_request.number }} \
            --datasets datasets/qa_test_100.jsonl datasets/safety_50.jsonl \
            --output eval_results.json
      - name: Quality gate check
        run: python scripts/check_gate.py eval_results.json
      - name: Upload eval results
        uses: actions/upload-artifact@v4
        with:
          name: eval-results
          path: eval_results.json

  deploy-staging:
    needs: eval-gate
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Deploy to staging
        run: |
          ./scripts/deploy.sh staging
      - name: Run smoke tests
        run: python scripts/smoke_test.py --env staging

灰度发布

金丝雀部署策略

class CanaryDeployer:
    def __init__(self, config_store, traffic_mgr):
        self.config = config_store
        self.traffic = traffic_mgr

    async def deploy_canary(self, new_version, percentage=5):
        """将 percentage% 流量切到新版本"""
        # 1. 部署新版本
        await self.config.set("canary_version", new_version)
        await self.config.set("canary_percentage", percentage)

        # 2. 设置分流规则
        await self.traffic.set_rules([
            {"match": {"header": "x-canary"}, "version": new_version},
            {"weight": percentage, "version": new_version},
            {"weight": 100 - percentage, "version": "stable"},
        ])

    async def check_canary_health(self):
        """检查金丝雀版本健康度"""
        stable_metrics = await self.get_metrics("stable")
        canary_metrics = await self.get_metrics("canary")

        checks = {
            "error_rate": canary_metrics.error_rate < stable_metrics.error_rate * 1.5,
            "latency_p99": canary_metrics.p99 < stable_metrics.p99 * 1.3,
            "quality_score": canary_metrics.quality >= stable_metrics.quality * 0.95,
        }
        return all(checks.values()), checks

    async def promote_or_rollback(self):
        healthy, detail = await self.check_canary_health()
        if healthy:
            # 逐步增加流量：5% → 25% → 50% → 100%
            current = await self.config.get("canary_percentage")
            next_pct = min(current * 5, 100)
            if next_pct >= 100:
                await self.config.set("stable_version",
                                      await self.config.get("canary_version"))
                await self.config.set("canary_percentage", 0)
                return "promoted"
            await self.config.set("canary_percentage", next_pct)
            return f"ramped to {next_pct}%"
        else:
            await self.config.set("canary_percentage", 0)
            return f"rolled back: {detail}"

A/B 测试

import scipy.stats as stats

class ABTest:
    def __init__(self, name, variants):
        self.name = name
        self.variants = variants  # ["control", "treatment"]
        self.data = {v: [] for v in variants}

    def record(self, variant, score):
        self.data[variant].append(score)

    def analyze(self, min_samples=100):
        if len(self.data["control"]) < min_samples:
            return {"status": "insufficient_data"}

        ctrl = self.data["control"]
        treat = self.data["treatment"]

        # Welch's t-test
        t_stat, p_value = stats.ttest_ind(treat, ctrl, equal_var=False)

        effect_size = (np.mean(treat) - np.mean(ctrl)) / np.std(ctrl)

        return {
            "status": "conclusive" if p_value < 0.05 else "inconclusive",
            "control_mean": np.mean(ctrl),
            "treatment_mean": np.mean(treat),
            "p_value": p_value,
            "effect_size": effect_size,
            "winner": "treatment" if (p_value < 0.05 and
                                      np.mean(treat) > np.mean(ctrl))
                      else "control",
        }

回滚机制

class RollbackManager:
    def __init__(self, version_store):
        self.versions = version_store

    async def deploy(self, version):
        await self.versions.set_active(version)
        await self.record_baseline(version)

    async def auto_rollback(self, current_version, prev_version,
                            watch_minutes=15):
        """部署后自动监控，异常则回滚"""
        import asyncio
        await asyncio.sleep(watch_minutes * 60)

        metrics = await self.get_current_metrics()
        baseline = await self.get_baseline(prev_version)

        anomalies = []
        if metrics.error_rate > baseline.error_rate * 2:
            anomalies.append("error_rate_spike")
        if metrics.p99_latency > baseline.p99 * 1.5:
            anomalies.append("latency_spike")
        if metrics.quality_score < baseline.quality * 0.9:
            anomalies.append("quality_degradation")

        if anomalies:
            await self.deploy(prev_version)
            await self.notify(f"Auto-rollback: {anomalies}")
            return True
        return False

总结

LLM 应用的 CI/CD 比传统软件多了两层：Prompt 版本管理和自动评估门禁。核心实践包括：Prompt 配置文件化并纳入 Git 管理；CI 中用 LLM-as-a-Judge 跑评估测试集做质量门禁；灰度发布从小流量开始逐步放量；A/B 测试用统计方法确认改进显著性；回滚机制自动化，异常时秒级回退。工具链上 GitHub Actions/GitLab CI 足以胜任，关键是将评估自动化做扎实。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。