12.5 使用指标和日志监控使用情况

12.6 示例：优化昂贵的构建流水线

即便一个结构良好的 CI 流水线,在容器构建、依赖安装、冗长的测试套件以及由 Claude 驱动的文档/生成步骤层层叠加后,也会变得缓慢且成本高昂。本示例展示如何将一条迟缓且昂贵的流水线转变为快速、预算可控的工作流。你将实现确定性缓存、基于变更的执行、Claude 任务的 token 成本门控,以及并行化测试。最终得到的流水线将以更短的挂钟时间和更低的 AI 支出交付同等质量的制品。

概念阐述

CI 中的性能与成本取决于四个杠杆。第一,避免冗余工作,方法是复用缓存并在没有相关变更时跳过作业。第二,使昂贵的步骤(容器构建、依赖安装)具备增量和确定性,这样缓存才能真正命中。第三,限制 AI 成本:在调用 Claude 之前估算 token,并在预算将被超支时短路切换到更便宜的模型或本地回退方案。第四,并行运行剩余工作并尽量减少协调开销,然后合并制品。

我们将通过一个小型 FastAPI 项目和一个 GitHub Actions 工作流来应用这些杠杆。构建过程将缓存 Python wheels 和 Docker 层,仅运行受最近变更影响的测试,在调用 Claude 前使用 token 估算器进行预检,并在所有门控通过后推送制品。

动手实践

创建以下最小化的项目结构:

app/

  app.py

  requirements.txt

tests/

  test_app.py

scripts/

  estimate_tokens.py

  changed_paths.py

  selective_tests.py

  build_image.sh

.github/

  workflows/

    ci.yml

app/requirements.txt

fastapi==0.114.0
uvicorn==0.30.5

app/app.py

from fastapi import FastAPI

app = FastAPI(title="Optimized Pipeline Demo")

@app.get("/health")
def health():
    return {"status": "ok"}

tests/test_app.py

from fastapi.testclient import TestClient
from app.app import app

def test_health():
    c = TestClient(app)
    r = c.get("/health")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"

scripts/estimate_tokens.py

import math
import json
import os
import sys

PRICING = {
    "claude-3.5-haiku": {"in": 0.0008, "out": 0.004},
    "claude-3.5-sonnet": {"in": 0.003, "out": 0.015},
    "claude-3-opus": {"in": 0.010, "out": 0.050},
}

def approx_tokens(text: str) -> int:
    # ~4 chars per token heuristic
    return math.ceil(len(text) / 4)

def main():
    model = os.getenv("CLAUDE_MODEL", "claude-3.5-sonnet")
    budget = float(os.getenv("CLAUDE_BUDGET_USD", "0.05"))
    expected_output_tokens = int(os.getenv("EXPECTED_OUTPUT_TOKENS", "1500"))
    prompt_path = sys.argv[1] if len(sys.argv) > 1 else "-"
    prompt = sys.stdin.read() if prompt_path == "-" else open(prompt_path, "r", encoding="utf-8").read()

    tin = approx_tokens(prompt)
    tout = expected_output_tokens
    rate_in = PRICING[model]["in"]
    rate_out = PRICING[model]["out"]
    cost = (tin/1000)*rate_in + (tout/1000)*rate_out
    decision = "allow" if cost <= budget else "switch"

    print(json.dumps({
        "model": model,
        "input_tokens": tin,
        "output_tokens": tout,
        "estimated_cost": round(cost, 4),
        "budget": budget,
        "decision": decision
    }, indent=2))

    # Exit code communicates gating to CI
    if decision == "allow":
        sys.exit(0)
    else:
        sys.exit(42)

if __name__ == "__main__":
    main()

scripts/changed_paths.py

import subprocess
import sys
import json

# Determine changed files against the base ref (default: origin/main)
base = sys.argv[1] if len(sys.argv) > 1 else "origin/main"
cmd = ["git", "diff", "--name-only", f"{base}...HEAD"]
out = subprocess.check_output(cmd, text=True)
files = [f.strip() for f in out.splitlines() if f.strip()]
print(json.dumps({"changed": files}, indent=2))

scripts/selective_tests.py

import json
import sys
from pathlib import Path