Lesson 6: Capstone — Build an AI Observability Stack

Instrument a multi-operation AI API (chat, RAG Q&A, summarization) with full observability: distributed tracing, structured cost logs, async quality sampling, drift detection, and three SLOs with automated alerting.

Every request
 └─ traced_llm_call() → span with model, tokens, latency
 └─ make_request_log() → structured JSON to stdout / log aggregator
 └─ maybe_score_response() → async 5% sample → quality_scores table
 └─ SLO checker (runs every 5 min) → alert if breach detected
 └─ Nightly drift job → KS test baseline vs current → Slack alert if drifting

Phase 1 — Distributed Tracing

Instrument all three operations (chat, rag, summarize) with Langfuse @observe() decorator or manual Span objects. Each span must capture: model, input_tokens, output_tokens, latency_ms, operation.
Add a parent trace for multi-step RAG (retrieval span + generation span as children). View the waterfall in Langfuse — confirm retrieval time is separated from generation time.
Introduce an artificial error in the RAG pipeline and verify it appears in the trace with status="error" and the exception message.

Phase 2 — Cost Tracking

Instrument every LLM call with CostRecord: model, input_tokens, output_tokens, operation, cost_usd. Write to a SQLite table (local dev) or PostgreSQL (production).
Run 100 test requests across all three operations and query: total cost, average cost per operation, and which operation has the highest cost-per-request. Identify the optimization opportunity.
Implement prompt caching for the RAG system prompt (the large stable instructions block). Verify input_tokens drops by >50% on the second call to the same conversation.
Set up a cost budget alert: if daily cost exceeds $10 (test threshold), log a cost_budget_exceeded event and print a warning. Simulate triggering it by sending 1,000 test requests.

Phase 3 — Quality Sampling & Drift Detection

Implement maybe_score_response() with 5% sampling rate, running asynchronously after the response has been sent to the user (background task or asyncio).
Run 200 test requests through the chat endpoint. Query the quality_scores table: mean score, p10, p25, p75, p90. Build a histogram (even just print the distribution in 10 buckets).
Simulate quality drift: modify your prompt to intentionally produce shorter, lower-quality responses. Run 50 more requests. Run detect_quality_drift(baseline, current) and verify it flags the drift as statistically significant.
Build a 10-case golden test set for each operation type. Run regression_gate(threshold=0.90). Confirm it passes on the good prompt and fails on the degraded prompt.

Phase 4 — SLOs & Alerting

Define three SLOs: latency_p95 < 5s (target: 95%), quality_score >= 0.7 (target: 90%), error_rate < 0.5% (target: 99.5%).
Implement check_all_slos() that reads from your cost and quality DB tables and returns compliance ratios. Schedule it to run every 5 minutes (use a simple while loop with time.sleep(300) for local dev).
Simulate an SLO breach: inject 10% HTTP 500 errors into your endpoint. Run the SLO checker and confirm it reports error_rate as non-compliant. Print the alert payload.
Compute error budgets for all three SLOs over a 30-day window. Determine: at current performance, how many minutes of budget remain for each SLO this month?

# Final integration test — run all observability checks
import asyncio

async def observability_health_check() -> dict:
    slo_status = check_all_slos(PRODUCTION_SLOS)
    quality_status = detect_quality_drift(
        get_quality_scores(days_ago=14, days=7),
        get_quality_scores(days_ago=7, days=7),
    )
    cost_status = {
        "daily_cost": get_daily_cost(),
        "budget_remaining": DAILY_BUDGET - get_daily_cost(),
    }
    return {
        "slos": slo_status,
        "quality_drift": quality_status["is_drifting"],
        "cost": cost_status,
        "healthy": (
            all(v["compliant"] for v in slo_status.values())
            and not quality_status["is_drifting"]
            and cost_status["budget_remaining"] > 0
        ),
    }

status = asyncio.run(observability_health_check())
print(f"System healthy: {status['healthy']}")
for slo_name, slo_data in status['slos'].items():
    marker = "✓" if slo_data["compliant"] else "✗"
    print(f"  {marker} {slo_name}: {slo_data['current']:.1%} (target: {slo_data['target']:.1%})")

Graduation criteria:All three SLOs tracked and alertable, quality drift detector successfully flags the degraded prompt, regression gate blocks the degraded configuration, and cost tracking correctly attributes spend to each operation with <5% error in cost calculation.

Every request └─ traced_llm_call() → span with model, tokens, latency └─ make_request_log() → structured JSON to stdout / log aggregator └─ maybe_score_response() → async 5% sample → quality_scores table └─ SLO checker (runs every 5 min) → alert if breach detected └─ Nightly drift job → KS test baseline vs current → Slack alert if drifting

# Final integration test — run all observability checks import asyncio async def observability_health_check() -> dict: slo_status = check_all_slos(PRODUCTION_SLOS) quality_status = detect_quality_drift( get_quality_scores(days_ago=14, days=7), get_quality_scores(days_ago=7, days=7), ) cost_status = { "daily_cost": get_daily_cost(), "budget_remaining": DAILY_BUDGET - get_daily_cost(), } return { "slos": slo_status, "quality_drift": quality_status["is_drifting"], "cost": cost_status, "healthy": ( all(v["compliant"] for v in slo_status.values()) and not quality_status["is_drifting"] and cost_status["budget_remaining"] > 0 ), } status = asyncio.run(observability_health_check()) print(f"System healthy: {status['healthy']}") for slo_name, slo_data in status['slos'].items(): marker = "✓" if slo_data["compliant"] else "✗" print(f" {marker} {slo_name}: {slo_data['current']:.1%} (target: {slo_data['target']:.1%})")