Module 8: Unified AI Client — Final integrating project
Monitoring and metrics
Your client already does fallback and routing. You don't know what it's doing. You don't know how many requests are going to Mistral vs OpenAI, you don't know how much it's costing you, you don't know which provider fails most. Without that data, your client is a black box.
In this capsule we add a MetricsCollector that records every request and exposes aggregate metrics. It's the difference between "works on my machine" code and code you operate in production.
By the end you'll be able to:
- Record each request: provider used, latency, cost, error if any
- Expose aggregate metrics: counters per provider, P50/P95 latency, total cost
- Export snapshots to JSON or Prometheus for integration with your stack
- Reset metrics for tests or reports per time window
Mental model: collector as a passive observer
The MetricsCollector is a collaborator of the UnifiedClient. It doesn't change the fallback/routing logic — it only observes and records.
┌─────────────────┐
│ UnifiedClient │
│ │
│ chat() ───────┼──> tries adapter → OK → calls collector.record(...)
│ │ ↓ (on error)
│ │ tries next → OK → record(...)
│ │
│ get_metrics()──┼──> collector.summary()
└─────────────────┘
│
▼
┌──────────────────┐
│ MetricsCollector │
│ - events[] │
│ + record(...) │
│ + summary() │
│ + reset() │
└──────────────────┘
The collector lives in memory in the process. For cross-process tracking (multi-worker, multi-instance), you connect to external backends (Prometheus, Datadog) — out of scope, but the design allows it.
Implementation
Create unified_ai_client/metrics.py:
# unified_ai_client/metrics.py
import time
import json
from dataclasses import dataclass, field, asdict
from collections import deque
@dataclass
class RequestEvent:
timestamp: float
provider: str
duration_ms: int
tokens_input: int
tokens_output: int
cost_usd: float | None
success: bool
error_type: str | None = None # 'rate_limit', 'auth', 'timeout', 'provider', None
@dataclass
class MetricsSummary:
total_requests: int
total_successes: int
total_errors: int
total_cost_usd: float
total_tokens_input: int
total_tokens_output: int
latency_p50_ms: int
latency_p95_ms: int
latency_p99_ms: int
by_provider: dict
errors_by_type: dict
window_seconds: float
class MetricsCollector:
"""
In-memory collector. Keeps a rolling window of the last N events
(default 10,000) so memory doesn't grow unbounded.
"""
def __init__(self, max_events: int = 10_000):
self.events: deque[RequestEvent] = deque(maxlen=max_events)
self.period_start: float = time.time()
def record(
self,
provider: str,
duration_ms: int,
tokens_input: int = 0,
tokens_output: int = 0,
cost_usd: float | None = None,
success: bool = True,
error_type: str | None = None,
) -> None:
self.events.append(RequestEvent(
timestamp=time.time(),
provider=provider,
duration_ms=duration_ms,
tokens_input=tokens_input,
tokens_output=tokens_output,
cost_usd=cost_usd,
success=success,
error_type=error_type,
))
def summary(self) -> MetricsSummary:
events = list(self.events)
successes = [e for e in events if e.success]
errors = [e for e in events if not e.success]
latencies = sorted(e.duration_ms for e in successes)
def percentile(p: int) -> int:
if not latencies:
return 0
idx = int(len(latencies) * p / 100)
return latencies[min(idx, len(latencies) - 1)]
# Aggregated per provider
by_provider: dict[str, dict] = {}
for e in events:
if e.provider not in by_provider:
by_provider[e.provider] = {
"requests": 0, "successes": 0, "errors": 0,
"cost_usd": 0.0, "tokens_input": 0, "tokens_output": 0,
"duration_ms_total": 0,
}
p = by_provider[e.provider]
p["requests"] += 1
if e.success:
p["successes"] += 1
else:
p["errors"] += 1
if e.cost_usd:
p["cost_usd"] += e.cost_usd
p["tokens_input"] += e.tokens_input
p["tokens_output"] += e.tokens_output
p["duration_ms_total"] += e.duration_ms
# Compute average duration per provider
for p in by_provider.values():
p["duration_ms_avg"] = (
p["duration_ms_total"] // p["requests"] if p["requests"] else 0
)
# Errors by type
errors_by_type: dict[str, int] = {}
for e in errors:
etype = e.error_type or "unknown"
errors_by_type[etype] = errors_by_type.get(etype, 0) + 1
return MetricsSummary(
total_requests=len(events),
total_successes=len(successes),
total_errors=len(errors),
total_cost_usd=round(sum(e.cost_usd for e in events if e.cost_usd), 6),
total_tokens_input=sum(e.tokens_input for e in events),
total_tokens_output=sum(e.tokens_output for e in events),
latency_p50_ms=percentile(50),
latency_p95_ms=percentile(95),
latency_p99_ms=percentile(99),
by_provider=by_provider,
errors_by_type=errors_by_type,
window_seconds=time.time() - self.period_start,
)
def reset(self) -> None:
self.events.clear()
self.period_start = time.time()
def to_json(self) -> str:
return json.dumps(asdict(self.summary()), indent=2)Integrate into UnifiedClient
Modify client.py to inject the collector and call it on each attempt:
# unified_ai_client/client.py — only relevant changes
from .metrics import MetricsCollector
from .exceptions import RateLimitError, AuthError, TimeoutError, ProviderError
class UnifiedClient:
def __init__(
self,
config: ClientConfig,
circuit_threshold: int = 3,
circuit_duration_s: int = 60,
default_priority: Priority | None = None,
):
# ... rest the same
self.metrics = MetricsCollector() if config.metrics_enabled else None
# ...
def chat_with_messages(
self,
messages: list[Message],
max_tokens: int = 256,
temperature: float = 0.7,
use_fallback: bool = True,
priority: Priority | None = None,
) -> ChatResponse:
prio = priority or self.default_priority
ordered_adapters = (
sort_by_priority(self.adapters, prio) if prio else self.adapters
)
if not use_fallback:
ordered_adapters = ordered_adapters[:1]
errors: dict[str, Exception] = {}
for adapter in ordered_adapters:
if self._circuit_open(adapter.name):
errors[adapter.name] = ProviderError(adapter.name, "Circuit open")
continue
try:
response = self._try_with_retry(
adapter, messages, max_tokens, temperature
)
if self.metrics:
self.metrics.record(
provider=response.provider,
duration_ms=response.duration_ms,
tokens_input=response.tokens_input,
tokens_output=response.tokens_output,
cost_usd=response.cost_usd,
success=True,
)
return response
except (RateLimitError, AuthError, TimeoutError, ProviderError) as e:
if self.metrics:
self.metrics.record(
provider=adapter.name,
duration_ms=0,
success=False,
error_type=_error_type(e),
)
errors[adapter.name] = e
self._record_failure(adapter.name)
continue
raise AllProvidersFailedError(errors)
def get_metrics(self):
if not self.metrics:
return None
return self.metrics.summary()
def reset_metrics(self) -> None:
if self.metrics:
self.metrics.reset()
def _error_type(e: Exception) -> str:
if isinstance(e, RateLimitError):
return "rate_limit"
if isinstance(e, AuthError):
return "auth"
if isinstance(e, TimeoutError):
return "timeout"
if isinstance(e, ProviderError):
return "provider"
return "unknown"Verification: a simple dashboard
Create examples/dashboard.py:
# examples/dashboard.py
import time
from unified_ai_client import UnifiedClient
client = UnifiedClient.from_yaml("examples/clients_fallback.yaml")
# Generate sample traffic
prompts = [
"Explain REST in one sentence",
"Define microservices",
"What is a container?",
"Summarize HTTP/2",
"Explain DNS",
]
for i in range(20):
prompt = prompts[i % len(prompts)]
priority = "cost-first" if i % 2 == 0 else "quality-first"
try:
client.chat(prompt, max_tokens=80, priority=priority)
except Exception as e:
print(f"Request {i} failed: {e}")
time.sleep(0.3)
# Report
summary = client.get_metrics()
print("\n=== SUMMARY ===")
print(f"Total requests: {summary.total_requests}")
print(f"Successful: {summary.total_successes}")
print(f"Errors: {summary.total_errors}")
print(f"Total cost: ${summary.total_cost_usd:.6f}")
print(f"Tokens input: {summary.total_tokens_input:,}")
print(f"Tokens output: {summary.total_tokens_output:,}")
print(f"Latency P50: {summary.latency_p50_ms}ms")
print(f"Latency P95: {summary.latency_p95_ms}ms")
print(f"Latency P99: {summary.latency_p99_ms}ms")
print(f"Window: {summary.window_seconds:.1f}s")
print("\n=== BY PROVIDER ===")
for provider, stats in summary.by_provider.items():
print(f"\n {provider}")
print(f" Requests: {stats['requests']} ({stats['successes']} ok, {stats['errors']} err)")
print(f" Cost: ${stats['cost_usd']:.6f}")
print(f" Avg duration: {stats['duration_ms_avg']}ms")
if summary.errors_by_type:
print("\n=== ERRORS BY TYPE ===")
for etype, count in summary.errors_by_type.items():
print(f" {etype}: {count}")
# Export JSON
with open("metrics_snapshot.json", "w") as f:
f.write(client.metrics.to_json())
print("\n→ Snapshot saved to metrics_snapshot.json")Expected output (depends on your config and traffic):
=== SUMMARY ===
Total requests: 20
Successful: 20
Errors: 0
Total cost: $0.000632
Tokens input: 240
Tokens output: 1452
Latency P50: 1842ms
Latency P95: 2911ms
Latency P99: 3045ms
=== BY PROVIDER ===
ollama-mistral
Requests: 10 (10 ok, 0 err)
Cost: $0.000000
Avg duration: 4521ms
openai-mini
Requests: 10 (10 ok, 0 err)
Cost: $0.000632
Avg duration: 1834ms
Exporting for Prometheus / Datadog
Common pattern: your app exposes a /metrics endpoint that returns Prometheus format. Your collector translates to that format:
def to_prometheus(self) -> str:
s = self.summary()
lines = [
f"# HELP llm_requests_total Total LLM requests",
f"# TYPE llm_requests_total counter",
f"llm_requests_total {s.total_requests}",
f"# HELP llm_cost_usd_total Total cost in USD",
f"# TYPE llm_cost_usd_total counter",
f"llm_cost_usd_total {s.total_cost_usd}",
f"# HELP llm_latency_p95_ms P95 latency in ms",
f"# TYPE llm_latency_p95_ms gauge",
f"llm_latency_p95_ms {s.latency_p95_ms}",
]
# Per provider
for provider, stats in s.by_provider.items():
lines.append(
f'llm_requests_by_provider{{provider="{provider}"}} {stats["requests"]}'
)
return "\n".join(lines)If your app is FastAPI:
@app.get("/metrics")
def metrics():
return Response(client.metrics.to_prometheus(), media_type="text/plain")Prometheus scrapes that endpoint and you get dashboards for free.
Advanced patterns
Pattern 1 — Metrics-based alerts
summary = client.get_metrics()
error_rate = summary.total_errors / max(summary.total_requests, 1)
if error_rate > 0.05:
send_alert(f"High error rate: {error_rate:.1%}")
if summary.total_cost_usd > 100:
send_alert(f"Cost in last 10k requests: ${summary.total_cost_usd:.2f}")Pattern 2 — Hourly reset
import schedule
def snapshot_and_reset():
s = client.metrics.to_json()
save_to_s3(f"metrics_{datetime.now().isoformat()}.json", s)
client.reset_metrics()
schedule.every().hour.do(snapshot_and_reset)Your metrics become historical with hourly resolution.
Pattern 3 — Metrics by tag
If your app has additional context (user_tier, feature, etc.), extend record() to accept tags and group by them.
Common traps
Trap 1 — "Memory grows unbounded."
Without maxlen on the deque, events accumulate until a crash. That's why I use deque(maxlen=10_000). For cases where you need the full history, export and reset periodically.
Trap 2 — "Metrics lost on restart." The metrics are in memory. A restart loses everything. If you need durability, persist every N requests (e.g., to SQLite) or use an external backend.
Trap 3 — "Multi-process with a single collector." If your app is multi-worker (gunicorn with 4 workers), each worker has its own collector. In the dashboard you only sum what each worker saw. For a unified view, export to a shared backend.
Trap 4 — "Cost_usd is null because I didn't configure pricing."
Without price_input_per_1m and price_output_per_1m in your ProviderConfig, cost_usd is None. Your summary sums it as zero. Configure prices for real tracking.
Trap 5 — "I measure wall-clock latency but wanted provider-only."
The duration_ms the adapter measures includes wall-clock (network + provider processing). If you want to separate them, you need more granular timing (not trivial).
Exercise
Extend the MetricsCollector to:
- Support aggregation by time window —
summary_last_minutes(n: int) -> MetricsSummarythat filters events to the lastnminutes - Add a
top_costs_by_provider(n: int)method that returns thenproviders with the highest cost in the current window - Add a
pytesttest that mocks 100 events (50 successful, 50 with an error) and verifies that the summary counts them correctly
See solution
# In metrics.py
def summary_last_minutes(self, n: int) -> MetricsSummary:
"""Same as summary() but filters events from the last n minutes."""
cutoff = time.time() - n * 60
filtered_events = [e for e in self.events if e.timestamp >= cutoff]
# Reuse summary()'s logic over filtered_events
# ... (refactor the main method to accept a filtered list)
def top_costs_by_provider(self, n: int = 5) -> list[tuple[str, float]]:
summary = self.summary()
sorted_providers = sorted(
summary.by_provider.items(),
key=lambda kv: kv[1]["cost_usd"],
reverse=True,
)
return [(name, stats["cost_usd"]) for name, stats in sorted_providers[:n]]
# Test
def test_collector_counts_events():
c = MetricsCollector()
for _ in range(50):
c.record(provider="A", duration_ms=100, tokens_input=10, tokens_output=20,
cost_usd=0.001, success=True)
for _ in range(50):
c.record(provider="B", duration_ms=0, success=False, error_type="timeout")
s = c.summary()
assert s.total_requests == 100
assert s.total_successes == 50
assert s.total_errors == 50
assert s.errors_by_type == {"timeout": 50}
assert s.by_provider["A"]["successes"] == 50
assert s.by_provider["B"]["errors"] == 50
assert abs(s.total_cost_usd - 0.05) < 1e-9Summary
You learned:
- ✅
MetricsCollectoras a passive collaborator of theUnifiedClient - ✅ Per-request tracking: provider, latency, tokens, cost, success/error
- ✅ Aggregate summary with percentiles, totals, breakdown per provider and by error type
- ✅ Export to JSON and Prometheus for integration with an existing stack
- ✅ Patterns: alerts, periodic reset, metrics by tag
- ✅ Trade-offs of in-memory vs persistent implementation
Checkpoint: if your client.get_metrics() returns real data and you can build a basic dashboard with those numbers, you're ready.
Next capsule
07 — Testing and validation. Your client has many features now: fallback, routing, circuit breaker, metrics. Without tests, a refactor breaks something and you don't notice. We're going to write a test suite that validates each feature with mocks and edge cases.
Resources
- Prometheus client for Python — official library.
- Langfuse — Self-hosted observability — LLM-specific alternative.
- Helicone — proxy with built-in observability.
- Datadog APM for LLMs — if your team already has Datadog.
- OpenTelemetry Python — emerging standard for distributed tracing.