from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
import time
# Setup tracing
provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)
def llm_call_with_tracing(prompt):
start = time.time()
with tracer.start_as_current_span("llm_completion") as span:
response = call_llm(prompt)
latency_ms = (time.time() - start) * 1000
span.set_attribute("prompt_length", len(prompt))
span.set_attribute("response_length", len(response))
span.set_attribute("latency_ms", latency_ms)
span.set_attribute("tokens_used", response.usage.total_tokens)
return response
import redis
import time
class BudgetManager:
def __init__(self, daily_budget_usd, redis_client):
self.daily_budget = daily_budget_usd
self.redis = redis_client
self.today = time.strftime("%Y-%m-%d")
def check_and_record(self, cost_usd):
key = f"daily_spend:{self.today}"
current = float(self.redis.get(key) or 0)
if current + cost_usd > self.daily_budget:
return False # Block request
else:
self.redis.incrbyfloat(key, cost_usd)
return True # Allow request
| Metric | Value | Alert Threshold |
|---|---|---|
| Daily spend | $42.50 | $100 (๐ข OK) |
| Tokens per request (avg) | 1,250 | >1,500 (๐ก Warn) |
| Cost per request (avg) | $0.0085 | >$0.01 (๐ข OK) |
| Most expensive user | User #1234: $12.20 | Alert if >$20 |
import json
from datetime import datetime
class PromptRegistry:
def __init__(self, storage_dir="prompts"):
self.storage_dir = storage_dir
def register(self, name, prompt_template, description):
version = self._next_version(name)
entry = {
"name": name,
"version": version,
"prompt": prompt_template,
"description": description,
"created_at": datetime.utcnow().isoformat(),
"deployed": False
}
with open(f"{self.storage_dir}/{name}_{version}.json", "w") as f:
json.dump(entry, f)
return version
def deploy(self, name, version):
# Mark as deployed in production
pass
def rollback(self, name):
# Revert to previous version
pass
class ABTest:
def __init__(self, config):
# config = {"variant_a": 50%, "variant_b": 50%}
self.config = config
self.results = {"variant_a": [], "variant_b": []}
def get_variant(self, user_id):
# Deterministic assignment based on user_id
hash_val = hash(user_id) % 100
if hash_val < self.config["variant_a"]:
return "variant_a"
return "variant_b"
def record_result(self, variant, score):
self.results[variant].append(score)
def get_winner(self):
# Statistical significance check
from scipy import stats
a_scores = self.results["variant_a"]
b_scores = self.results["variant_b"]
_, p_value = stats.ttest_ind(a_scores, b_scores)
if p_value < 0.05:
return "variant_b" if np.mean(b_scores) > np.mean(a_scores) else "variant_a"
return None # inconclusive
INCIDENT_RESPONSE = {
"detection": [
"Automated alert (latency > 10s for 5 min)",
"User report via feedback system"
],
"triage": [
"Check if issue affects all users or subset",
"Verify API provider status page",
"Compare metrics to baseline"
],
"mitigation": [
"Rollback to previous prompt version",
"Switch to fallback model",
"Enable rate limiting to protect system"
],
"resolution": [
"RCA within 24 hours",
"Deploy fix (prompt update, model update)",
"Post-mortem documentation"
]
}
List 5 metrics you would track for an AI-powered customer support chatbot. For each, define the healthy range and alert threshold.
You have two prompts for summarization. Design an A/B test to determine which is better. Include: metrics, sample size, duration.
You see a sudden 300% increase in token usage. Walk through your incident response steps.
You've finished Level 3 of the AI course. You now have professional-level knowledge to:
Certificate of completion available! ๐