12.5 Monitoring Usage with Metrics and Logs

12.6 Example: Optimizing an Expensive Build Pipeline

Even a well-structured CI pipeline can become slow and costly once container builds, dependency installs, long test suites, and Claude-powered documentation/generation steps pile up. This example shows how to turn a sluggish, expensive pipeline into a fast, budget-aware workflow. You will implement deterministic caching, change-based execution, token-cost gating for Claude tasks, and parallelized tests. The result is a pipeline that ships the same quality artifacts with less wall-clock time and lower AI spend.

Concept Development

Performance and cost in CI hinge on four levers. First, avoid redundant work by reusing caches and skipping jobs when nothing relevant changed. Second, make expensive steps (container builds, dependency installs) incremental and deterministic so caches actually hit. Third, bound AI costs: estimate tokens before calling Claude and short-circuit to a cheaper model or a local fallback when the budget would be exceeded. Fourth, run what remains in parallel with minimal coordination, then merge artifacts.

We will apply those levers using a small FastAPI project and a GitHub Actions workflow. The build will cache Python wheels and Docker layers, only run tests affected by recent changes, preflight Claude calls using a token estimator, and push artifacts when all gates pass.

Hands-On Example

Create this minimal project structure:

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