5. Cost Monitoring in the Pipeline

Overview

AI systems have a variable cost that traditional software doesn't have: every request to the LLM costs money. A prompt change that looks innocuous can triple the cost per request. A new feature that calls gpt-4o instead of gpt-4o-mini can multiply the spend by 10. Without cost monitoring in your pipeline, you discover these increases when the invoice arrives — weeks later.

Cost monitoring in the pipeline gives you real-time visibility. Every deployment generates a cost estimate based on the prompts your app uses. That estimate gets compared against the previous deployment. If the cost rises more than a threshold (e.g. +20%), the pipeline generates an alert. The result is an artifact with the cost history that you can analyze to understand the trend.

Connection with the final pipeline: In the capstone pipeline, cost monitoring runs as part of the ai-checks job, blocking the merge if a prompt change raises the estimated cost above the defined threshold.

The problem: Invisible costs

The scenario without cost monitoring

Week 1: Deploy with gpt-4o-mini → $0.02 per request
Week 2: A developer switches to gpt-4o for "better quality" → $0.20 per request
Week 3: 10,000 requests → $2,000 instead of $200
Week 4: Finance asks "why did the OpenAI invoice multiply by 10?"

The scenario with cost monitoring

Week 1: Deploy → cost estimate: $0.02/request → the baseline
Week 2: A PR changes the model → cost estimate: $0.20/request
          Pipeline alert: "⚠️ Cost increase: +900% ($0.02 → $0.20)"
          Developer: "Oh, I don't need gpt-4o for this. Revert."
          Total overspend: $0 (detected before the deploy)

The cost estimation script

The estimator

# scripts/cost_estimation.py
"""Estimates the cost of prompts per request based on token count and pricing."""

import json
import sys
import argparse
from pathlib import Path


MODEL_PRICING = {
    "gpt-4o-mini": {
        "input_per_1m": 0.15,
        "output_per_1m": 0.60,
    },
    "gpt-4o": {
        "input_per_1m": 2.50,
        "output_per_1m": 10.00,
    },
    "gpt-4-turbo": {
        "input_per_1m": 10.00,
        "output_per_1m": 30.00,
    },
}

CHARS_PER_TOKEN = 4


def estimate_tokens(text: str) -> int:
    return max(1, len(text) // CHARS_PER_TOKEN)


def load_prompts(prompts_file: str) -> list[dict]:
    path = Path(prompts_file)
    if not path.exists():
        print(f"Warning: {prompts_file} not found, using defaults")
        return [
            {
                "name": "system_prompt",
                "model": "gpt-4o-mini",
                "system": "You are a helpful AI assistant.",
                "user_template": "Answer this question: {question}",
                "avg_output_tokens": 150,
            }
        ]

    with open(path) as f:
        return json.load(f)


def estimate_cost_per_request(prompt: dict) -> dict:
    model = prompt.get("model", "gpt-4o-mini")
    pricing = MODEL_PRICING.get(model, MODEL_PRICING["gpt-4o-mini"])

    system_tokens = estimate_tokens(prompt.get("system", ""))
    user_tokens = estimate_tokens(prompt.get("user_template", ""))
    input_tokens = system_tokens + user_tokens
    output_tokens = prompt.get("avg_output_tokens", 150)

    input_cost = (input_tokens / 1_000_000) * pricing["input_per_1m"]
    output_cost = (output_tokens / 1_000_000) * pricing["output_per_1m"]
    total_cost = input_cost + output_cost

    return {
        "name": prompt.get("name", "unknown"),
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "input_cost": round(input_cost, 6),
        "output_cost": round(output_cost, 6),
        "total_cost_per_request": round(total_cost, 6),
    }


def generate_report(
    prompts_file: str = "prompts.json",
    previous_report: str | None = None,
    threshold: float = 0.20,
) -> dict:
    prompts = load_prompts(prompts_file)

    estimates = [estimate_cost_per_request(p) for p in prompts]
    total_per_request = sum(e["total_cost_per_request"] for e in estimates)

    projections = {
        "per_request": round(total_per_request, 6),
        "per_1000_requests": round(total_per_request * 1000, 4),
        "per_10000_requests": round(total_per_request * 10000, 2),
    }

    comparison = None
    alert = False

    if previous_report:
        prev_path = Path(previous_report)
        if prev_path.exists():
            with open(prev_path) as f:
                prev = json.load(f)
            prev_cost = prev.get("projections", {}).get("per_request", 0)
            if prev_cost > 0:
                change_pct = (total_per_request - prev_cost) / prev_cost
                comparison = {
                    "previous_per_request": prev_cost,
                    "current_per_request": round(total_per_request, 6),
                    "change_percentage": round(change_pct * 100, 1),
                    "threshold_percentage": threshold * 100,
                }
                alert = change_pct > threshold

    return {
        "estimates": estimates,
        "projections": projections,
        "comparison": comparison,
        "alert": alert,
        "threshold": threshold,
    }


def print_report(report: dict) -> None:
    print(f"\n{'='*55}")
    print(f"  Cost Estimation Report")
    print(f"{'='*55}")

    for est in report["estimates"]:
        print(f"\n  {est['name']}:")
        print(f"    Model:  {est['model']}")
        print(f"    Input:  {est['input_tokens']} tokens (${est['input_cost']:.6f})")
        print(f"    Output: {est['output_tokens']} tokens (${est['output_cost']:.6f})")
        print(f"    Total:  ${est['total_cost_per_request']:.6f}/request")

    proj = report["projections"]
    print(f"\n  Projections:")
    print(f"    Per request:       ${proj['per_request']:.6f}")
    print(f"    Per 1,000 req:     ${proj['per_1000_requests']:.4f}")
    print(f"    Per 10,000 req:    ${proj['per_10000_requests']:.2f}")

    if report["comparison"]:
        comp = report["comparison"]
        emoji = "🚨" if report["alert"] else "✅"
        print(f"\n  {emoji} Comparison with previous:")
        print(f"    Previous: ${comp['previous_per_request']:.6f}/req")
        print(f"    Current:  ${comp['current_per_request']:.6f}/req")
        print(f"    Change:   {comp['change_percentage']:+.1f}%")
        print(f"    Threshold: {comp['threshold_percentage']}%")
        if report["alert"]:
            print(f"    ⚠️ ALERT: Cost increase exceeds threshold!")

    print(f"{'='*55}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="AI cost estimation")
    parser.add_argument("--prompts", default="prompts.json")
    parser.add_argument("--previous", default=None, help="Previous report for comparison")
    parser.add_argument("--threshold", type=float, default=0.20, help="Alert threshold (0.20 = 20%)")
    parser.add_argument("--output", default=None, help="Output file")
    parser.add_argument("--json-output", action="store_true")
    args = parser.parse_args()

    report = generate_report(args.prompts, args.previous, args.threshold)

    if args.output:
        with open(args.output, "w") as f:
            json.dump(report, f, indent=2)

    if args.json_output:
        print(json.dumps(report, indent=2))
    else:
        print_report(report)

    if report["alert"]:
        sys.exit(1)

The prompts file

[
  {
    "name": "chat_system_prompt",
    "model": "gpt-4o-mini",
    "system": "You are a helpful AI assistant for our platform...",
    "user_template": "User question: {question}\nContext: {context}",
    "avg_output_tokens": 200
  },
  {
    "name": "classification_prompt",
    "model": "gpt-4o-mini",
    "system": "Classify the following text into categories...",
    "user_template": "Text: {text}",
    "avg_output_tokens": 50
  }
]

Integrating cost monitoring into the pipeline

The cost estimation step in CI

  ai-checks:
    steps:
      - name: Cost estimation
        id: cost
        run: |
          python scripts/cost_estimation.py \
            --prompts prompts.json \
            --threshold 0.20 \
            --output cost-report.json

          COST=$(python -c "
          import json
          with open('cost-report.json') as f:
              r = json.load(f)
          print(r['projections']['per_request'])
          ")
          echo "estimate=$COST" >> $GITHUB_OUTPUT

      - name: Upload cost report
        uses: actions/upload-artifact@v4
        with:
          name: cost-report-${{ github.run_number }}
          path: cost-report.json
          retention-days: 90

The comparison step against the previous deploy

  cost-comparison:
    needs: [ai-checks, docker]
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'

    steps:
      - uses: actions/checkout@v4

      - name: Download previous cost report
        uses: actions/download-artifact@v4
        with:
          name: cost-baseline
          path: ./previous/
        continue-on-error: true

      - name: Generate current report with comparison
        run: |
          PREV_FLAG=""
          if [ -f previous/cost-report.json ]; then
            PREV_FLAG="--previous previous/cost-report.json"
          fi

          python scripts/cost_estimation.py \
            --prompts prompts.json \
            --threshold 0.20 \
            $PREV_FLAG \
            --output cost-report.json

      - name: Check for cost alert
        id: alert
        run: |
          ALERT=$(python -c "
          import json
          with open('cost-report.json') as f:
              r = json.load(f)
          print('true' if r['alert'] else 'false')
          ")
          echo "triggered=$ALERT" >> $GITHUB_OUTPUT

      - name: Cost summary
        run: |
          python -c "
          import json
          with open('cost-report.json') as f:
              r = json.load(f)
          p = r['projections']
          print('## 💰 Cost Report')
          print('')
          print(f'| Metric | Value |')
          print(f'|--------|-------|')
          print(f'| Per request | \${p[\"per_request\"]:.6f} |')
          print(f'| Per 1,000 req | \${p[\"per_1000_requests\"]:.4f} |')
          print(f'| Per 10,000 req | \${p[\"per_10000_requests\"]:.2f} |')
          if r.get('comparison'):
              c = r['comparison']
              print(f'| Change vs previous | {c[\"change_percentage\"]:+.1f}% |')
          if r['alert']:
              print('')
              print('⚠️ **Cost increase exceeds threshold!**')
          " >> $GITHUB_STEP_SUMMARY

      - name: Upload as new baseline
        uses: actions/upload-artifact@v4
        with:
          name: cost-baseline
          path: cost-report.json
          retention-days: 90

      - name: Alert on cost spike
        if: steps.alert.outputs.triggered == 'true'
        uses: slackapi/slack-github-action@v2.0.0
        with:
          webhook-type: incoming-webhook
          webhook: ${{ secrets.SLACK_WEBHOOK_URL }}
          payload: |
            {
              "text": "💰 Cost spike detected in AI pipeline! Review before deploying."
            }

The cost history with artifacts

Every deployment generates an artifact with the cost report. To analyze the trend:

# scripts/analyze_cost_history.py
"""Analyzes the history of cost reports."""

import json
import sys
from pathlib import Path


def analyze_history(reports_dir: str) -> dict:
    reports = sorted(Path(reports_dir).glob("cost-report-*.json"))

    history = []
    for report_path in reports:
        with open(report_path) as f:
            data = json.load(f)
        history.append({
            "file": report_path.name,
            "per_request": data["projections"]["per_request"],
            "per_10k": data["projections"]["per_10000_requests"],
        })

    if len(history) < 2:
        return {"trend": "INSUFFICIENT_DATA", "history": history}

    first = history[0]["per_request"]
    last = history[-1]["per_request"]
    total_change = ((last - first) / first * 100) if first > 0 else 0

    return {
        "trend": "INCREASING" if total_change > 10 else "STABLE" if total_change > -10 else "DECREASING",
        "total_change_pct": round(total_change, 1),
        "first_cost": first,
        "last_cost": last,
        "data_points": len(history),
        "history": history,
    }


if __name__ == "__main__":
    result = analyze_history(sys.argv[1] if len(sys.argv) > 1 else ".")
    print(json.dumps(result, indent=2))

Comparisons

Cost monitoring in the pipeline vs the monthly invoice

Token-based estimation vs a live API call

Troubleshooting

"The cost estimate doesn't reflect the real cost"

Cause: The estimator uses a character count (len/4) that isn't exact. Models like gpt-4o use more complex tokenizers.

Solution: For more precision, use tiktoken:

import tiktoken

enc = tiktoken.encoding_for_model("gpt-4o-mini")
tokens = len(enc.encode(text))

"The threshold is too sensitive — constant alerts"

Cause: A 10% threshold can trigger on minimal changes to the prompt.

Solution: Adjust the threshold to 20-30% or use an absolute threshold:

alert = change_pct > 0.20 and abs(total_cost - prev_cost) > 0.001

This only alerts if the change is >20% AND the absolute increase is >$0.001/request.

"I don't have a previous cost report to compare against"

Cause: It's the first deployment or the previous artifact expired.

Solution: The script handles this gracefully — if there's no previous report, it generates a new one with no comparison. The next deployment will have the baseline.

Exercises

Exercise 1: Add cost estimation to your pipeline

Integrate the cost estimation script into your CI pipeline. Generate an artifact with the report.

- name: Cost estimation
  run: |
    python scripts/cost_estimation.py \
      --prompts prompts.json \
      --threshold 0.20 \
      --output cost-report.json

- name: Upload cost report
  uses: actions/upload-artifact@v4
  with:
    name: cost-report-${{ github.run_number }}
    path: cost-report.json
    retention-days: 90

- name: Cost summary
  run: |
    python scripts/cost_estimation.py --prompts prompts.json

Exercise 2: Create a weekly cost report

Create a scheduled workflow that generates a weekly cost report.

name: Weekly Cost Report

on:
  schedule:
    - cron: "0 9 * * 1"
  workflow_dispatch:

jobs:
  cost-report:
    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

      - name: Generate cost report
        run: |
          python scripts/cost_estimation.py \
            --prompts prompts.json \
            --output weekly-cost.json

          echo "## 💰 Weekly Cost Report" >> $GITHUB_STEP_SUMMARY
          python scripts/cost_estimation.py --prompts prompts.json >> $GITHUB_STEP_SUMMARY

      - uses: actions/upload-artifact@v4
        with:
          name: weekly-cost-${{ github.run_number }}
          path: weekly-cost.json
          retention-days: 365

Exercise 3: Alert if the model changed

Modify the script to detect whether the model in use changed between deployments (e.g. from gpt-4o-mini to gpt-4o).

def detect_model_changes(current: dict, previous: dict) -> list[dict]:
    changes = []
    prev_estimates = {e["name"]: e for e in previous.get("estimates", [])}

    for est in current.get("estimates", []):
        name = est["name"]
        if name in prev_estimates:
            prev_model = prev_estimates[name]["model"]
            curr_model = est["model"]
            if prev_model != curr_model:
                changes.append({
                    "prompt": name,
                    "previous_model": prev_model,
                    "current_model": curr_model,
                    "cost_impact": est["total_cost_per_request"] - prev_estimates[name]["total_cost_per_request"],
                })

    return changes

Cost monitoring best practices

1. Token estimation vs token counting

Character-based estimation (len(text) / 4) is fast but imprecise. For real production, use the official tokenizer:

# Estimation (fast, imprecise ±20%)
estimated_tokens = len(text) / 4

# Counting (precise, it requires tiktoken)
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o")
exact_tokens = len(enc.encode(text))

In the pipeline, the estimate is enough. For financial reports, use tiktoken.

2. Alert at the right threshold

A 20% threshold seems reasonable, but if your base costs are $0.001, a 50% increase is still $0.0005 — it isn't worth an alert. Combine the percentage with an absolute minimum:

alert = (
    cost_increase_percent > threshold
    and cost_increase_absolute > 0.05  # A minimum $0.05 increase
)

3. Versioning the pricing

Providers change prices periodically. Version your pricing dictionary:

MODEL_PRICING = {
    "gpt-4o-mini": {"input_per_1m": 0.15, "output_per_1m": 0.60},
    "gpt-4o": {"input_per_1m": 2.50, "output_per_1m": 10.00},
}
PRICING_UPDATED = "2025-01-15"

Summary

• ✅ Cost monitoring in the pipeline prevents cost spikes by detecting them before the deploy
• ✅ Token-based estimation is fast and free — enough for CI checks
• ✅ The comparison against the previous deploy alerts if the cost rises above the threshold
• ✅ Artifacts with cost reports create a history that allows trend analysis
• ✅ The pricing per model varies enormously: gpt-4o-mini ($0.15/1M input) vs gpt-4o ($2.50/1M input)
• ✅ The script generates automatic alerts when the cost exceeds the configurable threshold
• ✅ Weekly cost reports give periodic visibility into the projected spend
• ✅ It complements the monthly invoice — the pipeline prevents, the invoice confirms

Additional resources

1. OpenAI Pricing - The models' current prices
2. tiktoken - OpenAI's official tokenizer for precise counting
3. GitHub Actions — Artifacts - Storing reports
4. OpenAI Usage API - The API for querying real usage
5. Anthropic Pricing - Claude's pricing (for comparison)
6. LLM Cost Calculator - A tool for comparing costs between models