4. Monitoring and Observability in Production

Overview

In the previous capsule you configured scaling and performance: async execution, connection pooling, caching, horizontal scaling. Your agent can handle load. But there's a problem scaling doesn't solve: you don't know what's happening. How many requests are failing? How long does each query take? How much are you spending on tokens? Without monitoring, your agent in production is a black box — it works until it stops working, and when it fails, you have no idea why.

Monitoring without alerts is logging nobody reads. Dashboards without thresholds are decoration. The difference between a production-ready system and one that's "deployed" is that the first one tells you before the user complains. It alerts when the error rate goes over 10%. It alerts when the average latency passes 10 seconds. It alerts when the daily cost goes over $50. Actionable alerts — each one with a runbook that says exactly what to do.

Connection to the module: This capsule connects everything you saw in capsule 02 (deployment) and 03 (scaling) with the reality of operating an agent in production. In capsule 05 you'll see error recovery and resilience — but first you need to detect the errors. In 06, cost control — but first you need to measure the costs. Monitoring is the foundation everything else is built on.

Production Logging

The problem with print() and logging.info()

In development, print(f"Agent responded: {result}") works. In production with 100 concurrent requests, print() is noise: thousands of lines with no context, no way to filter, no correlation between the logs of a single request.

# ❌ Development logging — useless in production
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

logger.info("Agent invoked")
logger.info(f"Result: {result}")
logger.info("Done")

Which request generated that "Agent invoked"? From which user? How long did it take? You can't tell. You need structured logging — logs in JSON format with semantic fields you can filter, aggregate, and alert on.

Structured Logging with JSON

import logging
import json
import uuid
import time
from datetime import datetime, timezone
from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default="no-request")
user_id_var: ContextVar[str] = ContextVar("user_id", default="anonymous")


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "request_id": request_id_var.get(),
            "user_id": user_id_var.get(),
        }
        if hasattr(record, "extra_fields"):
            log_data.update(record.extra_fields)
        if record.exc_info and record.exc_info[0] is not None:
            log_data["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_data)


def setup_production_logging():
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())
    root = logging.getLogger()
    root.handlers.clear()
    root.addHandler(handler)
    root.setLevel(logging.INFO)

Every log line is valid JSON. Tools like CloudWatch, Datadog, or Elasticsearch can index every field. You can search all the logs for a specific request_id, filter by user_id, aggregate by level.

Correlation IDs: Following a Request End to End

An agent runs multiple steps: it plans, executes tools, reflects, generates an answer. Without a correlation ID, those logs are loose lines. With one, you reconstruct the complete story:

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

class CorrelationIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        req_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request_id_var.set(req_id)
        user_id_var.set(request.headers.get("X-User-ID", "anonymous"))
        start = time.perf_counter()
        try:
            response = await call_next(request)
            logging.getLogger("api").info("Request completed", extra={"extra_fields": {
                "status_code": response.status_code,
                "elapsed_ms": round((time.perf_counter() - start) * 1000, 2),
            }})
            response.headers["X-Request-ID"] = req_id
            return response
        except Exception as e:
            logging.getLogger("api").error("Request failed", exc_info=True, extra={
                "extra_fields": {"error_type": type(e).__name__}
            })
            raise

app.add_middleware(CorrelationIDMiddleware)

If a user reports an error, you ask them for the request_id from the response header and reconstruct the whole trace in your logs.

Log Levels: What to Log at Each Level

Not everything deserves the same level. In production, if everything is INFO, the volume drowns you:

In production, run at INFO. When you're investigating an incident, drop to DEBUG temporarily:

@app.post("/admin/log-level")
async def set_log_level(level: str):
    numeric_level = getattr(logging, level.upper(), None)
    if numeric_level is None:
        return {"error": f"Invalid level: {level}"}
    logging.getLogger().setLevel(numeric_level)
    return {"log_level": level.upper(), "status": "updated"}

Agent Logging: What to Capture

For an agent, standard web logs aren't enough. You need to capture the internal logic — which agent was selected, which tools were called, how many tokens each step consumed:

logger = logging.getLogger("agent")

def log_agent_step(step_name: str, **kwargs):
    logger.info(f"Agent step: {step_name}", extra={"extra_fields": {
        "step": step_name, "agent_type": kwargs.get("agent_type"),
        "tokens_used": kwargs.get("tokens_used"),
        "tools_called": kwargs.get("tools_called", []),
    }})

def log_tool_call(tool_name: str, duration_ms: float, success: bool):
    level = logging.INFO if success else logging.WARNING
    logger.log(level, f"Tool call: {tool_name}", extra={"extra_fields": {
        "tool_name": tool_name, "duration_ms": round(duration_ms, 2), "success": success,
    }})

LangSmith in Production

From Development to Production

In module 09 you configured LangSmith for testing and evaluation. In production, LangSmith becomes your tracing system — every request generates a complete trace. But production has constraints development doesn't: volume, cost, and retention.

Configuration for Production

import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGSMITH_API_KEY")
os.environ["LANGCHAIN_PROJECT"] = "research-agent-prod"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"

Use separate projects for each environment: research-agent-dev, research-agent-staging, research-agent-prod. Never mix development traces with production — the noise makes it impossible to find real patterns.

Sampling: Don't Trace Everything

In production, tracing 100% has two problems: cost (LangSmith charges by volume above a certain tier) and noise. Use sampling:

import random
from langsmith import traceable

TRACE_SAMPLE_RATE = 0.2

@traceable(name="research_agent_invoke")
async def invoke_agent_traced(query: str, config: dict):
    return await graph_app.ainvoke({"messages": [HumanMessage(content=query)]}, config)

async def invoke_agent(query: str, config: dict):
    if random.random() < TRACE_SAMPLE_RATE:
        return await invoke_agent_traced(query, config)
    prev = os.environ.get("LANGCHAIN_TRACING_V2")
    os.environ["LANGCHAIN_TRACING_V2"] = "false"
    try:
        return await graph_app.ainvoke({"messages": [HumanMessage(content=query)]}, config)
    finally:
        if prev:
            os.environ["LANGCHAIN_TRACING_V2"] = prev

The exception to sampling: always trace requests that fail. If a request ends in an error, you need the complete trace to investigate.

Tags and Metadata for Filtering

Add context to each trace so you can filter in LangSmith:

@traceable(
    name="research_query",
    tags=["production", "research-agent"],
    metadata={"version": "2.1.0", "environment": "prod"}
)
async def handle_research_query(query: str, user_id: str):
    config = {
        "configurable": {"thread_id": user_id},
        "metadata": {"user_id": user_id},
        "tags": [f"user:{user_id}"],
    }
    return await graph_app.ainvoke({"messages": [HumanMessage(content=query)]}, config)

With this you can filter: "traces from user X", "version 2.1.0", "tag production".

The Cost of Tracing

LangSmith has a generous free tier for development, but in production the volume adds up. Check LangSmith's pricing and calculate based on your expected volume. For most agent applications, 10-20% sampling is enough visibility without blowing the budget.

Metrics and Dashboards

The 5 Metrics That Matter

You don't need 50 metrics. You need 5 that tell you whether your system is healthy:

┌──────────────────────────────────────────────────────────────────┐
│  DASHBOARD: Research Agent — Production                          │
│                                                                  │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐             │
│  │ Error Rate  │  │   Latency   │  │  Daily Cost │             │
│  │    2.3%     │  │  p50: 4.2s  │  │   $34.50    │             │
│  │  ✅ < 10%   │  │  p95: 12.1s │  │  ✅ < $50    │             │
│  └─────────────┘  │  p99: 28.4s │  └─────────────┘             │
│                    └─────────────┘                               │
│  ┌─────────────┐                    ┌─────────────┐             │
│  │ Throughput  │                    │ Completion  │             │
│  │  142 req/h  │                    │   96.2%     │             │
│  └─────────────┘                    └─────────────┘             │
└──────────────────────────────────────────────────────────────────┘

Collection with Prometheus

Use prometheus_client, the industry standard for metrics in Python:

from prometheus_client import Counter, Histogram, Gauge

requests_total = Counter("agent_requests_total", "Total requests", ["status", "agent_type"])
request_duration = Histogram(
    "agent_request_duration_seconds", "Request duration",
    ["agent_type"], buckets=[1, 2, 5, 10, 15, 30, 60, 120]
)
active_requests = Gauge("agent_active_requests", "Currently processing")
tokens_used = Counter("agent_tokens_total", "Tokens consumed", ["model", "type"])
daily_cost = Gauge("agent_daily_cost_dollars", "Daily cost in USD")

Metrics Middleware

Wire in the collection so every request is measured automatically:

class MetricsMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path in ("/health", "/metrics"):
            return await call_next(request)
        active_requests.inc()
        start = time.perf_counter()
        try:
            response = await call_next(request)
            status = "success" if response.status_code < 400 else "error"
            requests_total.labels(status=status, agent_type="research").inc()
            request_duration.labels(agent_type="research").observe(time.perf_counter() - start)
            return response
        except Exception:
            requests_total.labels(status="error", agent_type="research").inc()
            raise
        finally:
            active_requests.dec()

Tracking Tokens and Costs

Tokens are the financial metric. Track them by model and type (input/output):

MODEL_COSTS = {
    "gpt-4o": {"input": 2.50 / 1_000_000, "output": 10.00 / 1_000_000},
    "gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
}

def track_llm_usage(model: str, input_tokens: int, output_tokens: int):
    tokens_used.labels(model=model, type="input").inc(input_tokens)
    tokens_used.labels(model=model, type="output").inc(output_tokens)
    costs = MODEL_COSTS.get(model, MODEL_COSTS["gpt-4o-mini"])
    cost = (input_tokens * costs["input"]) + (output_tokens * costs["output"])
    daily_cost.inc(cost)
    return cost

Expose a /metrics endpoint for Prometheus scraping. Connect Grafana for visualization:

from prometheus_client import generate_latest, CONTENT_TYPE_LATEST

@app.get("/metrics")
async def metrics():
    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)

Actionable Alerts

Alerts vs. Dashboards

A dashboard nobody looks at is useless. An alert that goes off every 5 minutes for something non-actionable is too. The key: every alert must have a runbook — a document that says exactly what to do when it fires.

┌──────────────────────────────────────────────────────────────────┐
│  ACTIONABLE ALERTS                                                │
│                                                                  │
│  🔴 CRITICAL (wake someone up)                                   │
│  ├── Error rate > 25% for 5 min                                 │
│  ├── Service completely down                                     │
│  └── Daily cost > $100                                          │
│                                                                  │
│  🟡 WARNING (review during work hours)                           │
│  ├── Error rate > 10% for 15 min                                │
│  ├── p95 latency > 10s for 10 min                               │
│  └── Daily cost > $50                                           │
│                                                                  │
│  🔵 INFO (daily review)                                          │
│  ├── p95 latency > 5s                                           │
│  ├── Throughput 50% below average                               │
│  └── New error types in the logs                                │
└──────────────────────────────────────────────────────────────────┘

Implementing Alerts

You don't need an enterprise tool. An in-app system with rules, time windows, and cooldowns:

from dataclasses import dataclass
from datetime import datetime, timezone, timedelta
from collections import deque
from enum import Enum


class AlertSeverity(Enum):
    WARNING = "warning"
    CRITICAL = "critical"


@dataclass
class AlertRule:
    name: str
    severity: AlertSeverity
    threshold: float
    window_minutes: int
    cooldown_minutes: int = 30
    last_fired: datetime | None = None


class AlertManager:
    def __init__(self):
        self.request_log: deque[dict] = deque(maxlen=10000)
        self.rules = [
            AlertRule("high_error_rate", AlertSeverity.WARNING, 0.10, 15),
            AlertRule("critical_error_rate", AlertSeverity.CRITICAL, 0.25, 5),
            AlertRule("high_latency", AlertSeverity.WARNING, 10.0, 10),
            AlertRule("high_daily_cost", AlertSeverity.WARNING, 50.0, 1440),
        ]

    def record_request(self, success: bool, latency_s: float, cost_usd: float):
        self.request_log.append({
            "timestamp": datetime.now(timezone.utc),
            "success": success, "latency_s": latency_s, "cost_usd": cost_usd,
        })
        self._evaluate_rules()

    def _evaluate_rules(self):
        now = datetime.now(timezone.utc)
        for rule in self.rules:
            if rule.last_fired and (now - rule.last_fired) < timedelta(minutes=rule.cooldown_minutes):
                continue
            window_start = now - timedelta(minutes=rule.window_minutes)
            recent = [r for r in self.request_log if r["timestamp"] > window_start]
            if not recent:
                continue

            if "error_rate" in rule.name:
                val = sum(1 for r in recent if not r["success"]) / len(recent)
                if val > rule.threshold:
                    self._fire(rule, f"Error rate: {val:.1%}")
            elif "latency" in rule.name:
                p95 = sorted(r["latency_s"] for r in recent)[int(len(recent) * 0.95)]
                if p95 > rule.threshold:
                    self._fire(rule, f"P95 latency: {p95:.1f}s")
            elif "cost" in rule.name:
                total = sum(r["cost_usd"] for r in recent)
                if total > rule.threshold:
                    self._fire(rule, f"Cost: ${total:.2f}")

    def _fire(self, rule: AlertRule, details: str):
        rule.last_fired = datetime.now(timezone.utc)
        logger.warning(f"ALERT [{rule.severity.value}] {rule.name}: {details}")
        self._send_to_slack(rule, details)

    def _send_to_slack(self, rule: AlertRule, details: str):
        """Send to Slack with a link to the runbook."""
        pass  # httpx.post(webhook_url, json={...})


alert_manager = AlertManager()

Notifications: Slack + Runbooks

Every alert must land somewhere someone will see it, and every alert must include a link to the runbook. Without a runbook, the alert says "there's a problem" but doesn't say what to do. With a runbook, the on-call engineer has step-by-step instructions:

import httpx

async def send_slack_alert(webhook_url: str, name: str, severity: str, details: str):
    emoji = {"warning": "⚠️", "critical": "🚨"}
    payload = {"text": (
        f"{emoji.get(severity, '❓')} *{name}*\n"
        f"Details: {details}\n"
        f"Runbook: https://wiki.internal/runbooks/{name}"
    )}
    async with httpx.AsyncClient() as client:
        await client.post(webhook_url, json=payload)

SLA Monitoring

Defining SLAs for Agents

An SLA (Service Level Agreement) is a measurable commitment. For an agent, your typical SLAs are:

Tracking SLA Compliance

@dataclass
class SLATracker:
    completion_target: float = 0.95
    latency_p95_target: float = 15.0
    cost_per_query_target: float = 0.10

    def __post_init__(self):
        self.requests: deque[dict] = deque(maxlen=50000)

    def record(self, success: bool, latency_s: float, cost_usd: float):
        self.requests.append({"success": success, "latency_s": latency_s, "cost_usd": cost_usd})

    def get_compliance(self) -> dict:
        if not self.requests:
            return {"error": "No data"}
        reqs = list(self.requests)
        total = len(reqs)
        completion = sum(1 for r in reqs if r["success"]) / total
        p95 = sorted(r["latency_s"] for r in reqs)[int(total * 0.95)]
        avg_cost = sum(r["cost_usd"] for r in reqs) / total
        return {
            "completion_rate": {"current": round(completion, 4), "target": 0.95, "compliant": completion >= 0.95},
            "latency_p95": {"current": round(p95, 2), "target": 15.0, "compliant": p95 <= 15.0},
            "cost_per_query": {"current": round(avg_cost, 4), "target": 0.10, "compliant": avg_cost <= 0.10},
            "total_requests": total,
        }

sla_tracker = SLATracker()

Expose a /sla endpoint to check the status in real time. A typical result:

{
  "status": "compliant",
  "sla": {
    "completion_rate": {"current": 0.962, "target": 0.95, "compliant": true},
    "latency_p95": {"current": 12.4, "target": 15.0, "compliant": true},
    "cost_per_query": {"current": 0.073, "target": 0.1, "compliant": true},
    "total_requests": 4250
  }
}

When an SLA isn't met, it should fire an alert. A violated SLA isn't an "I'll fix it tomorrow" — it's a broken commitment to your users or your organization.

Advanced Health Checks

Beyond {"status": "ok"}

A health check that always returns 200 is useless. Your agent depends on multiple components: the LLM provider, MCP servers, a database, API keys. A real health check verifies each dependency:

import asyncio
import httpx
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage


async def check_llm_health() -> dict:
    try:
        llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, max_tokens=5)
        start = time.perf_counter()
        await llm.ainvoke([HumanMessage(content="Say OK")])
        elapsed_ms = (time.perf_counter() - start) * 1000
        return {"component": "llm", "status": "healthy", "latency_ms": round(elapsed_ms, 2)}
    except Exception as e:
        return {"component": "llm", "status": "unhealthy", "error": str(e)}


async def check_mcp_server(name: str, url: str) -> dict:
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            resp = await client.get(f"{url}/health")
            status = "healthy" if resp.status_code == 200 else "degraded"
            return {"component": f"mcp_{name}", "status": status}
    except Exception as e:
        return {"component": f"mcp_{name}", "status": "unhealthy", "error": str(e)}


async def check_api_keys() -> dict:
    api_key = os.getenv("OPENAI_API_KEY", "")
    return {
        "component": "api_keys",
        "status": "healthy" if api_key else "unhealthy",
        "openai_key_present": bool(api_key),
    }

The Complete Health Check Endpoint

@app.get("/health")
async def health_check():
    checks = await asyncio.gather(
        check_llm_health(),
        check_mcp_server("search", "http://localhost:8001"),
        check_mcp_server("files", "http://localhost:8002"),
        check_api_keys(),
        return_exceptions=True,
    )

    results = [
        c if isinstance(c, dict) else {"component": "unknown", "status": "error", "error": str(c)}
        for c in checks
    ]

    any_unhealthy = any(r.get("status") == "unhealthy" for r in results)
    overall = "unhealthy" if any_unhealthy else "healthy"
    status_code = 503 if any_unhealthy else 200

    return Response(
        content=json.dumps({"status": overall, "checks": results}),
        status_code=status_code,
        media_type="application/json",
    )

If the LLM provider is down, the health check reports it and the load balancer can route traffic to another instance. Don't wait for a request to fail — run periodic health checks as a background task:

@app.on_event("startup")
async def startup():
    async def periodic_health(interval: int = 60):
        while True:
            checks = await asyncio.gather(check_llm_health(), check_api_keys(), return_exceptions=True)
            for c in checks:
                if isinstance(c, dict) and c.get("status") == "unhealthy":
                    logger.error(f"Health check failed: {c}")
            await asyncio.sleep(interval)
    asyncio.create_task(periodic_health())

Connection to the Project

The guide's Research Agent now needs complete monitoring:

1. Structured logging — Every request with a correlation ID, every tool call logged with its duration and result, every agent step with the tokens consumed.

2. LangSmith production — A research-agent-prod project with 20% sampling. Tags by user and version. 100% tracing on requests with errors.

3. Prometheus metrics — Error rate, latency (p50/p95/p99), tokens per request, daily cost. A /metrics endpoint for scraping.

4. Alerts — Error rate > 10% for 15 min → Slack warning. p95 latency > 15s → Slack warning. Daily cost > $50 → Slack warning. Error rate > 25% → PagerDuty critical.

5. SLA tracking — 95% completion rate, p95 < 15s, < $0.10/query. A /sla endpoint to check.

6. Health checks — Verify the LLM provider, each MCP server, the API keys. Periodic every 60 seconds. A /health endpoint with per-component detail.

The combination of logging + tracing + metrics + alerts is what separates a professionally operated service from one that's "running somewhere."

Troubleshooting

Problem 1: "The logs don't show up in CloudWatch/Datadog"

Likely cause: The formatter isn't producing valid JSON, or there are newlines inside the JSON splitting one entry across multiple lines. CloudWatch expects one JSON line per log entry.

Solution: Use json.dumps(log_data, default=str) to handle non-serializable objects. Test locally that every line is parseable JSON before deploying.

Problem 2: "LangSmith shows incomplete traces"

Likely cause: The request failed before completing (OOM, timeout), and LangSmith never got the closing event. The SDK uses a background thread to send events.

Solution: Configure a delay on shutdown to give the flush time:

@app.on_event("shutdown")
async def shutdown():
    import time
    time.sleep(2)

Problem 3: "The alerts go off constantly (alert fatigue)"

Likely cause: Thresholds that are too aggressive or time windows that are too short. If your normal error rate is 5% and you alert at 6%, statistical variation fires alerts all day.

Solution: Analyze your baseline for a week with no alerts. Define thresholds based on real data. Use windows of at least 10-15 minutes. Implement cooldowns of 30+ minutes.

Problem 4: "The LLM health check is slow and expensive"

Likely cause: A real LLM invocation every 60 seconds that adds up in tokens and latency.

Solution: Use max_tokens=5 and a minimal prompt. Raise the interval to 5 minutes. To verify the API key with no tokens, do a GET to OpenAI's /v1/models.

Problem 5: "The Prometheus metrics reset on redeploy"

Likely cause: The counters live in memory. On restart, they begin at 0.

Solution: Prometheus handles counter resets correctly. In Grafana, use rate(agent_requests_total[5m]) instead of the raw value. The rate() function understands resets.

Exercises

Exercise 1: Design the log levels for your agent

Your Research Agent has these events: receive a query, select an agent, execute a tool call, tool call succeeded, tool call failed (with retry), tool call failed (no retries left), generate the final answer, send the answer to the user. Assign a log level to each event and justify it.

Exercise 2: Calculate the cost of tracing

Your agent processes 500 requests/day. Each request generates 8 spans in LangSmith (1 root + 3 LLM calls + 4 tool calls). If LangSmith charges $0.01/1000 spans after 50K free spans/month, how much does it cost to trace 100% vs. 20%?

Exercise 3: Write a runbook for "error rate > 10%"

Your high_error_rate alert fires. Write a 5-7 step runbook to diagnose and resolve it.

Exercise 4: Implement "time to first token" tracking

For streaming, users perceive speed through TTFT (time to first token), not total latency. Implement a tracker with Prometheus that measures TTFT and alerts if the p95 is > 3s.

ttft_histogram = Histogram(
    "agent_time_to_first_token_seconds",
    "Time to first token in streaming responses",
    ["agent_type"],
    buckets=[0.5, 1.0, 1.5, 2.0, 3.0, 5.0, 10.0]
)


async def stream_with_ttft_tracking(query: str, config: dict):
    start = time.perf_counter()
    first_token_recorded = False

    async for event in graph_app.astream_events(
        {"messages": [HumanMessage(content=query)]}, config, version="v2",
    ):
        if not first_token_recorded and event.get("event") == "on_chat_model_stream":
            ttft = time.perf_counter() - start
            ttft_histogram.labels(agent_type="research").observe(ttft)
            first_token_recorded = True
        yield event

Exercise 5: Design a 4-panel dashboard

If you could only have 4 panels on your dashboard, which would they be? Define which metric, which chart, and which visual thresholds.

Summary

• Structured logging with JSON, correlation IDs, and appropriate log levels. Every request traceable end to end.
• LangSmith in production with separate projects per environment, 20% sampling, and 100% tracing on errors.
• Prometheus metrics for the 5 metrics that matter: error rate, latency, cost, throughput, completion rate.
• Actionable alerts with data-based thresholds, cooldowns to avoid fatigue, and a runbook for each alert.
• SLA monitoring with measurable targets: 95% completion, p95 < 15s, < $0.10/query.
• Advanced health checks that verify each dependency: the LLM provider, MCP servers, API keys.

Monitoring isn't the glamorous part of building agents. But it's the difference between "my agent works" and "I know my agent works, and I know immediately when it stops."

Additional Resources

1. LangSmith Documentation — Tracing, evaluation, production setup
2. OpenTelemetry Python — The observability standard
3. Prometheus Python Client — The official library for metrics
4. Grafana Dashboards — Metrics visualization
5. SRE Book — Monitoring Distributed Systems — Google's reference on monitoring
6. structlog — An alternative to Python's standard logging