ils15 Agent Observability & Monitoring Skill
When to Use
Implement this skill when your multi-agent system needs observability across agent spans, LLM calls, token costs, latency, error rates, and alerting. Covers the full stack: trace export → metrics collection → structured logging → dashboards → incident response.
OpenTelemetry Tracing
Span Creation & Context Propagation
from opentelemetry import trace
from opentelemetry.trace import SpanKind, Status, StatusCode
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.propagate import inject, extract
import json
resource = Resource.create({"service.name": "agent-orchestrator"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(
BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4317"))
)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)
class AgentTracer:
def __init__(self, agent_name: str):
self.agent_name = agent_name
self.tracer = trace.get_tracer(agent_name)
async def trace_agent_run(self, task: dict) -> dict:
with self.tracer.start_as_current_span(
f"agent.{self.agent_name}.run",
kind=SpanKind.SERVER,
attributes={
"agent.name": self.agent_name,
"task.id": task.get("id"),
"task.type": task.get("type", "unknown"),
},
) as span:
try:
result = await self._execute(task)
span.set_attribute("result.status", "success")
span.set_attribute("result.length", len(str(result)))
return result
except Exception as e:
span.set_status(Status(StatusCode.ERROR, str(e)))
span.record_exception(e)
raise
async def _execute(self, task: dict) -> dict:
with self.tracer.start_as_current_span(
f"agent.{self.agent_name}.execute",
kind=SpanKind.INTERNAL,
attributes={"task.priority": task.get("priority", 0)},
) as span:
spans = await self._llm_call(task["prompt"])
return spans
Agent Span Hierarchy
async def orchestrate_workflow(request: dict, headers: dict) -> dict:
ctx = extract(headers)
with tracer.start_as_current_span(
"orchestrator", context=ctx, kind=SpanKind.SERVER
) as root:
root.set_attribute("request.id", request["id"])
with tracer.start_as_current_span("agent.hermes.run") as hermes_span:
hermes_result = await hermes.process(request["query"])
hermes_span.set_attribute("tokens", hermes_result["tokens"])
with tracer.start_as_current_span("agent.aphrodite.run") as aphro_span:
aphro_result = await aphrodite.render(hermes_result["data"])
aphro_span.set_attribute("components", len(aphro_result["ui"]))
root.set_attribute("total.duration_ms", root.end_time - root.start_time)
return {"result": hermes_result, "ui": aphro_result}
LLM Call Tracing with LangSmith
from langsmith import Client, traceable
from langsmith.run_trees import RunTree
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
client = Client()
@traceable(
project_name="agent-hermes",
tags=["production", "hermes-v2"],
metadata={"agent": "hermes", "version": "2.1.0"},
)
async def agent_llm_call(prompt: str, context: dict) -> str:
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
messages = [
SystemMessage(content="You are Hermes, a backend agent."),
HumanMessage(content=prompt),
]
response = await llm.ainvoke(messages)
return response.content
@traceable(run_type="chain", name="hermes_full_flow")
async def hermes_plan_and_generate(task: str) -> dict:
plan = await agent_llm_call(f"Plan implementation for: {task}")
code = await agent_llm_call(f"Generate code for plan: {plan}")
return {"plan": plan, "code": code}
Custom Traces & Feedback
async def trace_with_feedback(task_id: str, user_input: str, output: str) -> None:
run = RunTree(
name="agent-qa",
run_type="chain",
inputs={"task_id": task_id, "query": user_input},
tags=["qa-review"],
)
with run:
child_llm = run.create_child(
name="llm_check",
run_type="llm",
inputs={"prompt": user_input},
outputs={"response": output},
)
child_llm.end()
run.end()
client.create_run(**run.dict())
client.create_feedback(
run_id=run.id,
key="user_rating",
score=0.95,
comment="Answer was accurate and well-structured",
)
Token & Cost Tracking
Token Counters
import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class TokenUsage:
input_tokens: int = 0
output_tokens: int = 0
total_tokens: int = 0
cost_usd: float = 0.0
MODEL_PRICING = {
"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},
"claude-sonnet-4": {"input": 3.00 / 1_000_000, "output": 15.00 / 1_000_000},
"claude-haiku": {"input": 0.25 / 1_000_000, "output": 1.25 / 1_000_000},
}
class TokenTracker:
def __init__(self):
self._usage: dict[str, dict[str, TokenUsage]] = defaultdict(
lambda: defaultdict(TokenUsage)
)
def record(
self,
agent: str,
model: str,
input_tokens: int,
output_tokens: int,
feature: str = "general",
) -> TokenUsage:
pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
cost = (input_tokens * pricing["input"]) + (output_tokens * pricing["output"])
usage = TokenUsage(
input_tokens=input_tokens,
output_tokens=output_tokens,
total_tokens=input_tokens + output_tokens,
cost_usd=round(cost, 6),
)
self._usage[agent][feature] = usage
with tracer.start_as_current_span("token.record") as span:
span.set_attribute("token.agent", agent)
span.set_attribute("token.model", model)
span.set_attribute("token.input", input_tokens)
span.set_attribute("token.output", output_tokens)
span.set_attribute("token.cost", cost)
return usage
def total_cost(self, agent: Optional[str] = None) -> float:
if agent:
return sum(u.cost_usd for u in self._usage[agent].values())
return sum(
u.cost_usd
for agent_usage in self._usage.values()
for u in agent_usage.values()
)
def budget_alert(self, budget_usd: float, agent: Optional[str] = None) -> list[str]:
alerts = []
if agent:
cost = self.total_cost(agent)
if cost > budget_usd:
alerts.append(f"{agent} cost ${cost:.2f} exceeds budget ${budget_usd:.2f}")
else:
for a in self._usage:
cost = self.total_cost(a)
if cost > budget_usd / len(self._usage):
alerts.append(f"{a} cost ${cost:.2f} exceeds proportional budget")
return alerts
tracker = TokenTracker()
usage = tracker.record("hermes", "gpt-4o", input_tokens=1500, output_tokens=420, feature="code-gen")
print(f"Cost: ${usage.cost_usd:.4f}, Budget alerts: {tracker.budget_alert(10.0)}")
Agent Performance Metrics
Prometheus Counters & Histograms
from prometheus_client import Counter, Histogram, Gauge, generate_latest, REGISTRY
from contextlib import asynccontextmanager
import time
agent_calls = Counter(
"agent_calls_total",
"Total agent LLM calls",
["agent", "model", "status"],
)
agent_latency = Histogram(
"agent_latency_seconds",
"Agent call latency in seconds",
["agent", "operation"],
buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0],
)
agent_tokens = Counter(
"agent_tokens_total",
"Total tokens consumed",
["agent", "direction"],
)
agent_cost = Counter(
"agent_cost_usd_total",
"Total cost in USD",
["agent", "model"],
)
agent_errors = Counter(
"agent_errors_total",
"Agent errors by type",
["agent", "error_type"],
)
agent_concurrent = Gauge(
"agent_concurrent_requests",
"Current concurrent agent requests",
["agent"],
)
@asynccontextmanager
async def track_agent_call(agent: str, model: str, operation: str):
agent_concurrent.labels(agent=agent).inc()
start = time.monotonic()
try:
yield
agent_calls.labels(agent=agent, model=model, status="success").inc()
except Exception as e:
agent_calls.labels(agent=agent, model=model, status="error").inc()
agent_errors.labels(agent=agent, error_type=type(e).__name__).inc()
raise
finally:
duration = time.monotonic() - start
agent_latency.labels(agent=agent, operation=operation).observe(duration)
agent_concurrent.labels(agent=agent).dec()
async def agent_handler(agent: str, model: str, prompt: str) -> str:
async with track_agent_call(agent, model, "llm_invoke"):
response = await llm_call(prompt)
token_count = len(response.split())
agent_tokens.labels(agent=agent, direction="output").inc(token_count)
agent_cost.labels(agent=agent, model=model).inc(
token_count * MODEL_PRICING[model]["output"]
)
return response
Drift Detection
from collections import deque
import statistics
class LatencyDriftDetector:
def __init__(self, window: int = 100, threshold: float = 2.0):
self.window = deque(maxlen=window)
self.threshold = threshold
def record(self, latency: float) -> Optional[float]:
self.window.append(latency)
if len(self.window) < 30:
return None
mean = statistics.mean(self.window)
std = statistics.stdev(self.window)
z_score = (latency - mean) / max(std, 1e-6)
if abs(z_score) > self.threshold:
return z_score
return None
drift = LatencyDriftDetector()
latency = 3.2
score = drift.record(latency)
if score:
logger.warning(f"Latency drift detected: z-score={score:.2f}, latency={latency:.2f}s")
Structured Logging
import json
import logging
import re
from datetime import datetime, timezone
from uuid import uuid4
SENSITIVE_PATTERNS = [
(re.compile(r'"api_key":\s*"[^"]+"'), '"api_key": "***REDACTED***"'),
(re.compile(r'"password":\s*"[^"]+"'), '"password": "***REDACTED***"'),
(re.compile(r'"token":\s*"[^"]+"'), '"token": "***REDACTED***"'),
(re.compile(r'\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b'), '***UUID-REDACTED***'),
]
class JSONFormatter(logging.Formatter):
def __init__(self, agent_name: str):
super().__init__()
self.agent_name = agent_name
def format(self, record: logging.LogRecord) -> str:
log_entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"agent": self.agent_name,
"correlation_id": getattr(record, "correlation_id", ""),
"message": record.getMessage(),
"module": record.module,
"function": record.funcName,
"line": record.lineno,
}
if record.exc_info and record.exc_info[0]:
log_entry["exception"] = {
"type": record.exc_info[0].__name__,
"message": str(record.exc_info[1]),
}
if hasattr(record, "extra_data"):
log_entry["extra"] = record.extra_data
raw = json.dumps(log_entry, default=str)
for pattern, replacement in SENSITIVE_PATTERNS:
raw = pattern.sub(replacement, raw)
return raw
def setup_agent_logger(agent_name: str, level: int = logging.INFO) -> logging.Logger:
logger = logging.getLogger(f"agent.{agent_name}")
logger.setLevel(level)
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter(agent_name))
logger.handlers.clear()
logger.addHandler(handler)
correlation_id = str(uuid4())
logger = logging.LoggerAdapter(logger, {"correlation_id": correlation_id})
return logger
hermes_logger = setup_agent_logger("hermes", logging.DEBUG)
hermes_logger.info("Agent initialized", extra={"extra_data": {"model": "gpt-4o", "version": "2.1.0"}})
hermes_logger.error("LLM call failed", extra={"extra_data": {"tokens_used": 1500, "retry": 3}})
Alerting
Alertmanager Configuration
# alertmanager.yml
route:
receiver: "pagerduty-critical"
routes:
- match:
severity: critical
receiver: "pagerduty-critical"
continue: true
- match:
severity: warning
receiver: "slack-alerts"
- match:
severity: info
receiver: "log-only"
receivers:
- name: "pagerduty-critical"
pagerduty_configs:
- routing_key: "${PAGERDUTY_ROUTING_KEY}"
severity: critical
description: "Agent system alert - {{ .GroupLabels.alertname }}"
- name: "slack-alerts"
slack_configs:
- api_url: "${SLACK_WEBHOOK_URL}"
channel: "#agent-alerts"
title: "{{ .GroupLabels.alertname }}"
text: "{{ .CommonAnnotations.summary }}"
Prometheus Alert Rules
# alerts/agent_alerts.yml
groups:
- name: agent_observability
rules:
- alert: HighAgentLatency
expr: histogram_quantile(0.95, rate(agent_latency_seconds_bucket[5m])) > 10
for: 2m
labels:
severity: critical
annotations:
summary: "P95 latency > 10s for agent {{ $labels.agent }}"
- alert: HighErrorRate
expr: |
rate(agent_calls_total{status="error"}[5m])
/
rate(agent_calls_total[5m]) > 0.05
for: 5m
labels:
severity: critical
annotations:
summary: "Error rate > 5% for agent {{ $labels.agent }}"
- alert: CostAnomaly
expr: |
rate(agent_cost_usd_total[1h])
/
avg(rate(agent_cost_usd_total[24h])) > 3
for: 15m
labels:
severity: warning
annotations:
summary: "Cost spike 3x above 24h average for {{ $labels.agent }}"
- alert: TokenRateSpike
expr: rate(agent_tokens_total[5m]) > 100000
for: 2m
labels:
severity: warning
annotations:
summary: "Token rate > 100K/5m for {{ $labels.agent }}"
- alert: AgentDeadlock
expr: |
agent_concurrent_requests > 20
and
rate(agent_calls_total[5m]) < 0.1
for: 30s
labels:
severity: critical
annotations:
summary: "Agent {{ $labels.agent }} has 20+ concurrent requests but no completions in 30s"
- alert: BudgetExceeded
expr: agent_cost_usd_total > 50
labels:
severity: warning
annotations:
summary: "Agent {{ $labels.agent }} exceeded $50 daily budget"
- alert: NoTraces
expr: rate(agent_calls_total[10m]) == 0
for: 5m
labels:
severity: critical
annotations:
summary: "No agent calls detected for 10 minutes - possible pipeline failure"
Dashboards
Grafana Panels
{
"title": "Agent Observability",
"panels": [
{
"title": "Agent Throughput",
"type": "timeseries",
"targets": [{
"expr": "rate(agent_calls_total{status=\"success\"}[5m])",
"legendFormat": "{{ agent }} - {{ status }}"
}],
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 0}
},
{
"title": "LLM Cost by Agent (24h)",
"type": "barchart",
"targets": [{
"expr": "sum by(agent) (increase(agent_cost_usd_total[24h]))",
"legendFormat": "{{ agent }}"
}],
"gridPos": {"h": 8, "w": 12, "x": 12, "y": 0}
},
{
"title": "Token Consumption Rate",
"type": "timeseries",
"targets": [{
"expr": "rate(agent_tokens_total[5m])",
"legendFormat": "{{ agent }} - {{ direction }}"
}],
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 8}
},
{
"title": "Latency Percentiles",
"type": "stat",
"targets": [
{"expr": "histogram_quantile(0.50, rate(agent_latency_seconds_bucket[5m]))", "legendFormat": "P50"},
{"expr": "histogram_quantile(0.95, rate(agent_latency_seconds_bucket[5m]))", "legendFormat": "P95"},
{"expr": "histogram_quantile(0.99, rate(agent_latency_seconds_bucket[5m]))", "legendFormat": "P99"}
],
"gridPos": {"h": 8, "w": 12, "x": 12, "y": 8}
},
{
"title": "Error Rate",
"type": "timeseries",
"targets": [{
"expr": "rate(agent_calls_total{status=\"error\"}[5m]) / rate(agent_calls_total[5m])",
"legendFormat": "{{ agent }}"
}],
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 16}
},
{
"title": "Concurrent Requests",
"type": "timeseries",
"targets": [{
"expr": "agent_concurrent_requests",
"legendFormat": "{{ agent }}"
}],
"gridPos": {"h": 8, "w": 12, "x": 12, "y": 16}
}
]
}
Incident Response
Trace → Log → Metrics Correlation
import structlog
from opentelemetry import trace
async def incident_correlation(alert_name: str, labels: dict) -> dict:
correlation_id = str(uuid4())
logger = structlog.get_logger()
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span(f"incident.{alert_name}") as span:
span.set_attribute("correlation_id", correlation_id)
span.set_attribute("alert.name", alert_name)
span.set_attributes(labels)
logger.error(
"incident_triggered",
correlation_id=correlation_id,
alert=alert_name,
agent=labels.get("agent"),
)
metrics = await query_prometheus(
f'rate(agent_calls_total{{agent="{labels.get("agent")}"}}[5m])'
)
span.set_attribute("metrics.query_result", str(metrics))
recent_logs = await query_loki(
f'{{agent="{labels.get("agent")}"}} |= "{correlation_id}"'
)
span.set_attribute("logs.matched", len(recent_logs))
root_cause = await diagnose_root_cause(alert_name, labels, metrics, recent_logs)
if root_cause["severity"] == "critical" and root_cause.get("auto_rollback"):
await rollback_deployment(labels.get("agent"))
logger.warning("auto_rollback_triggered", agent=labels.get("agent"))
return root_cause
async def diagnose_root_cause(
alert: str, labels: dict, metrics: dict, logs: list
) -> dict:
error_rate = metrics.get("error_rate", 0)
latency = metrics.get("p95_latency", 0)
diagnosis = {"severity": "info", "cause": "unknown", "recommendation": ""}
if error_rate > 0.1:
diagnosis.update(
severity="critical",
cause="high_error_rate",
recommendation="Rollback last agent deployment. Check model API status.",
)
elif latency > 30:
diagnosis.update(
severity="warning",
cause="latency_regression",
recommendation="Scale agent replicas. Check for prompt length increases.",
)
elif "rate_limit" in str(logs).lower():
diagnosis.update(
severity="warning",
cause="rate_limited",
recommendation="Implement exponential backoff. Reduce concurrency.",
)
return diagnosis
async def rollback_deployment(agent: str) -> None:
logger = structlog.get_logger()
logger.warning("rollback_initiated", agent=agent)
span = trace.get_current_span()
span.add_event("rollback.started", {"agent": agent})
Complete Monitoring Setup
from fastapi import FastAPI, Request
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from prometheus_client import make_asgi_app
import structlog
app = FastAPI()
FastAPIInstrumentor.instrument_app(app)
HTTPXClientInstrumentor().instrument()
metrics_app = make_asgi_app()
app.mount("/metrics", metrics_app)
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.JSONRenderer(),
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
logger = structlog.get_logger()
@app.middleware("http")
async def observability_middleware(request: Request, call_next):
correlation_id = request.headers.get("X-Correlation-Id", str(uuid4()))
with tracer.start_as_current_span(
f"http.{request.method}.{request.url.path}",
attributes={"correlation_id": correlation_id},
):
response = await call_next(request)
response.headers["X-Correlation-Id"] = correlation_id
return response
@app.get("/health")
async def health():
return {
"status": "healthy",
"metrics_endpoint": "/metrics",
"trace_endpoint": "http://jaeger:16686",
}
# Usage: app starts with OpenTelemetry SDK, Prometheus /metrics, structured logging
# Alertmanager watches Prometheus rules, PagerDuty/Slack on critical, Grafana dashboards
Real-World Checklist
- OpenTelemetry SDK configured with exporter (Jaeger/Honeycomb/OTLP)
- All agent spans created with proper parent-child hierarchy
- LangSmith tracing enabled for LLM calls with metadata tags
- Token counters per agent per model with cost attribution
- Budget alerts configured (per-agent and global thresholds)
- Prometheus histograms for P50/P95/P99 latency per operation
- Error counters by agent and error type
- Structured JSON logging with correlation IDs across all agents
- Sensitive data redaction in logs (API keys, tokens, passwords)
- Alertmanager rules for latency spikes, error rates, cost anomalies, deadlock detection
- Grafana dashboard with throughput, costs, error rates, concurrent requests
- Incident response playbook with trace-log-metrics correlation
- Auto-rollback triggers for critical error rate thresholds
- Metrics endpoint exposed and scraped by Prometheus