Monitoring AI Agents in Production: A Complete Guide
AI agents—autonomous systems that reason, call tools, and interact with APIs—are increasingly deployed in production environments. Unlike traditional deterministic services, these agents exhibit non-deterministic behavior, consume variable amounts of tokens, and can fail in novel and unexpected ways. Monitoring them requires a dedicated approach that combines classical observability with LLM-specific telemetry.
What Is AI Agent Monitoring?
AI agent monitoring is the practice of collecting, aggregating, and analyzing telemetry data from autonomous agent systems to ensure they operate reliably, cost-effectively, and correctly. It extends beyond standard application performance monitoring (APM) to capture:
- Reasoning traces — the step-by-step thought process the agent follows
- Tool call sequences — which external functions were invoked, in what order, and with what parameters
- Token consumption — exact counts of input and output tokens per LLM call
- Latency breakdowns — time spent in each phase: planning, tool execution, reflection
- Error classifications — whether failures stem from model refusals, tool timeouts, rate limits, or malformed outputs
Why It Matters
Without proper monitoring, AI agents become black boxes. A seemingly functional agent can silently drift into generating expensive, redundant tool calls, or start returning degraded responses due to upstream model changes. The business impact is direct: runaway token costs, missed SLAs, and frustrated users who receive incorrect or incomplete results. Effective monitoring gives teams the visibility needed to debug failures, optimize costs, and build trust in autonomous systems.
The Three Pillars: Logs, Costs, and Error Tracking
A robust monitoring strategy for AI agents rests on three interconnected pillars. Let's examine each in detail with practical implementation examples.
1. Structured Logging for AI Agents
Logging for agents must capture the full execution graph—not just input and output, but every intermediate reasoning step and tool interaction. Plain text logs are insufficient; structured JSON logs allow you to query, filter, and visualize agent behavior across thousands of runs.
What to Log
- Agent run metadata: run ID, agent name/version, session ID, user ID, timestamp
- LLM call details: model name, prompt template used, full message array, completion tokens, latency
- Tool executions: tool name, arguments, response, duration, success/failure status
- Decision points: the agent's chosen action at each step, including any reasoning or plan text
- Final output: the complete response returned to the user or calling system
Implementation Example: A Logging Decorator
Below is a production-ready logging decorator for an agent loop. It wraps each step, captures all relevant data, and emits structured JSON logs using Python's logging module with a custom formatter.
import logging
import json
import time
import uuid
from datetime import datetime, timezone
from functools import wraps
from typing import Any, Dict, Optional
# Configure structured JSON logging
class StructuredFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
log_entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
}
if isinstance(record.msg, dict):
log_entry.update(record.msg)
else:
log_entry["message"] = record.getMessage()
if record.exc_info and record.exc_info[1]:
log_entry["exception"] = self.formatException(record.exc_info)
return json.dumps(log_entry, default=str)
logger = logging.getLogger("agent_monitor")
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
handler.setFormatter(StructuredFormatter())
logger.handlers = [handler]
def log_agent_step(step_name: str):
"""Decorator that logs inputs, outputs, timing, and errors for each agent step."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
step_id = str(uuid.uuid4())[:8]
start_time = time.perf_counter()
# Log step start
logger.info({
"event": "step_start",
"step_name": step_name,
"step_id": step_id,
"input_kwargs": {k: str(v)[:500] for k, v in kwargs.items() if k != 'messages'},
})
try:
result = func(*args, **kwargs)
elapsed_ms = (time.perf_counter() - start_time) * 1000
# Log step completion
logger.info({
"event": "step_complete",
"step_name": step_name,
"step_id": step_id,
"duration_ms": round(elapsed_ms, 2),
"success": True,
"output_preview": str(result)[:500] if result is not None else None,
})
return result
except Exception as e:
elapsed_ms = (time.perf_counter() - start_time) * 1000
logger.error({
"event": "step_error",
"step_name": step_name,
"step_id": step_id,
"duration_ms": round(elapsed_ms, 2),
"success": False,
"error_type": type(e).__name__,
"error_message": str(e)[:1000],
})
raise
return wrapper
return decorator
# Usage in an agent loop
class ToolCallingAgent:
def __init__(self, model: str, tools: list):
self.model = model
self.tools = {t.__name__: t for t in tools}
self.run_id = str(uuid.uuid4())
@log_agent_step("llm_call")
def call_llm(self, messages: list, tools: list) -> Dict[str, Any]:
"""Simulate an LLM call — replace with actual API call."""
# In production, this calls OpenAI / Anthropic / etc.
# Log token counts from the API response
response = {
"content": "I'll search for that information.",
"tool_calls": [{"name": "search", "args": {"query": "latest docs"}}],
"usage": {"prompt_tokens": 450, "completion_tokens": 120},
}
# Emit a dedicated token usage log
logger.info({
"event": "token_usage",
"run_id": self.run_id,
"model": self.model,
"prompt_tokens": response["usage"]["prompt_tokens"],
"completion_tokens": response["usage"]["completion_tokens"],
"total_tokens": response["usage"]["prompt_tokens"] + response["usage"]["completion_tokens"],
})
return response
@log_agent_step("tool_execution")
def execute_tool(self, tool_name: str, args: dict) -> str:
"""Execute a tool and return its result."""
tool_fn = self.tools.get(tool_name)
if not tool_fn:
raise ValueError(f"Unknown tool: {tool_name}")
return tool_fn(**args)
def run(self, user_query: str) -> str:
"""Main agent loop with full logging."""
logger.info({
"event": "run_start",
"run_id": self.run_id,
"model": self.model,
"user_query": user_query,
})
messages = [{"role": "user", "content": user_query}]
max_steps = 10
for step_num in range(1, max_steps + 1):
logger.info({
"event": "agent_step",
"run_id": self.run_id,
"step_number": step_num,
"message_count": len(messages),
})
llm_response = self.call_llm(messages, list(self.tools.keys()))
if llm_response.get("tool_calls"):
for tool_call in llm_response["tool_calls"]:
tool_result = self.execute_tool(
tool_call["name"],
tool_call["args"]
)
messages.append({
"role": "tool",
"tool_call_id": str(uuid.uuid4())[:8],
"content": str(tool_result),
})
else:
logger.info({
"event": "run_complete",
"run_id": self.run_id,
"total_steps": step_num,
"final_output": llm_response["content"][:500],
})
return llm_response["content"]
logger.warning({
"event": "max_steps_reached",
"run_id": self.run_id,
"step_limit": max_steps,
})
return "Agent reached step limit without final answer."2. Tracking and Managing Costs
Cost tracking is arguably the most critical operational concern for AI agents. A single agent run can consume thousands of tokens across multiple LLM calls, and costs compound quickly at scale. Without visibility, teams can face unexpected bills running into thousands of dollars.
Cost Tracking Architecture
A cost tracking system should:
- Capture token counts per model call — both prompt and completion tokens
- Map tokens to dollar amounts — using up-to-date pricing from each model provider
- Aggregate by dimensions — per agent, per user, per session, per tool, per day
- Alert on anomalies — sudden spikes, runs exceeding budget thresholds
- Provide cost attribution — which part of the agent's reasoning chain is most expensive
Implementation: Real-Time Cost Calculator
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, List, Optional
import threading
@dataclass
class ModelPricing:
"""Pricing per 1M tokens (input / output) in USD."""
input_cost_per_million: float
output_cost_per_million: float
# Pricing as of mid-2024 — update regularly
PRICING_TABLE: Dict[str, ModelPricing] = {
"gpt-4o": ModelPricing(5.00, 15.00),
"gpt-4o-mini": ModelPricing(0.15, 0.60),
"gpt-4-turbo": ModelPricing(10.00, 30.00),
"claude-3-opus": ModelPricing(15.00, 75.00),
"claude-3-sonnet": ModelPricing(3.00, 15.00),
"claude-3-haiku": ModelPricing(0.25, 1.25),
"claude-3.5-sonnet": ModelPricing(3.00, 15.00),
}
@dataclass
class CostRecord:
"""A single cost entry for an LLM call."""
timestamp: datetime
run_id: str
step_name: str
model: str
prompt_tokens: int
completion_tokens: int
cost_usd: float
class CostTracker:
"""Thread-safe cost tracker with aggregation and alerting."""
def __init__(self, budget_threshold_usd: float = 5.0):
self.records: List[CostRecord] = []
self._lock = threading.Lock()
self.budget_threshold = budget_threshold_usd
self._run_costs: Dict[str, float] = {} # run_id -> total cost
def record(self, run_id: str, step_name: str, model: str,
prompt_tokens: int, completion_tokens: int) -> float:
"""Record a token usage event and return the cost in USD."""
pricing = PRICING_TABLE.get(model)
if not pricing:
# Unknown model — log warning and use a safe default estimate
pricing = ModelPricing(10.0, 30.0)
prompt_cost = (prompt_tokens / 1_000_000) * pricing.input_cost_per_million
completion_cost = (completion_tokens / 1_000_000) * pricing.output_cost_per_million
total_cost = round(prompt_cost + completion_cost, 6)
record = CostRecord(
timestamp=datetime.now(timezone.utc),
run_id=run_id,
step_name=step_name,
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
cost_usd=total_cost,
)
with self._lock:
self.records.append(record)
self._run_costs[run_id] = self._run_costs.get(run_id, 0) + total_cost
current_run_cost = self._run_costs[run_id]
# Check budget threshold
if current_run_cost > self.budget_threshold:
self._alert_budget_exceeded(run_id, current_run_cost)
return total_cost
def _alert_budget_exceeded(self, run_id: str, cost: float):
"""Emit alert when a single run exceeds the budget threshold."""
print(f"🚨 BUDGET ALERT: Run {run_id[:8]} has cost ${cost:.4f} "
f"(threshold: ${self.budget_threshold:.2f})")
# In production, send to PagerDuty, Slack, email, etc.
def get_run_cost(self, run_id: str) -> float:
with self._lock:
return self._run_costs.get(run_id, 0.0)
def get_total_cost_since(self, since: datetime) -> float:
with self._lock:
return sum(r.cost_usd for r in self.records if r.timestamp >= since)
def get_cost_by_model(self, since: datetime) -> Dict[str, float]:
with self._lock:
result: Dict[str, float] = {}
for r in self.records:
if r.timestamp >= since:
result[r.model] = result.get(r.model, 0) + r.cost_usd
return result
def get_cost_by_step(self, run_id: str) -> Dict[str, float]:
"""Break down cost by agent step for a specific run."""
with self._lock:
result: Dict[str, float] = {}
for r in self.records:
if r.run_id == run_id:
result[r.step_name] = result.get(r.step_name, 0) + r.cost_usd
return result
# Integration example
cost_tracker = CostTracker(budget_threshold_usd=3.0)
# After each LLM call in your agent loop:
# cost_tracker.record(
# run_id=run_id,
# step_name="planning_step",
# model="gpt-4o",
# prompt_tokens=response.usage.prompt_tokens,
# completion_tokens=response.usage.completion_tokens,
# )
# Query costs at end of day
today_start = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
print(f"Total cost today: ${cost_tracker.get_total_cost_since(today_start):.4f}")
print(f"Cost by model: {cost_tracker.get_cost_by_model(today_start)}")Cost Dashboard Metrics
Beyond raw tracking, expose these key metrics in your monitoring dashboard:
- Cost per successful run — helps identify if certain query types are disproportionately expensive
- Token efficiency ratio — completion tokens / (prompt tokens + completion tokens); low ratios may indicate overly verbose system prompts
- Cost per tool call — reveals which tools drive the most LLM usage
- Daily/weekly burn rate — projected against your monthly budget
3. Error Tracking and Classification
AI agents fail differently than traditional software. Errors can arise from model refusals, malformed structured outputs, tool timeouts, rate limit exhaustion, context window overflows, or infinite reasoning loops. A dedicated error tracking system must classify, aggregate, and help you recover gracefully.
Error Categories for AI Agents
- Model Errors: rate limits (429), content filter blocks, model overload (503), token limit exceeded
- Tool Errors: timeouts, invalid arguments, permission denials, external API failures
- Parsing Errors: the LLM returned malformed JSON, unparseable function call arguments, or incomplete structured output
- Agent Logic Errors: infinite loops, contradictory reasoning, exceeding maximum steps, empty final responses
- Policy Violations: the agent attempted a disallowed action or accessed restricted resources
Implementation: Error Classification and Recovery System
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable, Dict
import asyncio
import random
class AgentErrorCategory(Enum):
MODEL_RATE_LIMIT = "model_rate_limit"
MODEL_CONTENT_FILTER = "model_content_filter"
MODEL_OVERLOAD = "model_overload"
MODEL_TOKEN_LIMIT = "model_token_limit"
TOOL_TIMEOUT = "tool_timeout"
TOOL_INVALID_ARGS = "tool_invalid_args"
TOOL_PERMISSION = "tool_permission"
TOOL_EXTERNAL_FAILURE = "tool_external_failure"
PARSE_MALFORMED_JSON = "parse_malformed_json"
PARSE_INCOMPLETE_OUTPUT = "parse_incomplete_output"
AGENT_MAX_STEPS = "agent_max_steps"
AGENT_EMPTY_RESPONSE = "agent_empty_response"
AGENT_LOOP_DETECTED = "agent_loop_detected"
POLICY_VIOLATION = "policy_violation"
UNKNOWN = "unknown"
@dataclass
class AgentError:
category: AgentErrorCategory
message: str
run_id: str
step_number: int
step_name: str
retryable: bool
original_exception: Optional[Exception] = None
context: Dict = None
def __post_init__(self):
if self.context is None:
self.context = {}
class ErrorClassifier:
"""Classifies raw exceptions into structured AgentError categories."""
def classify(self, exception: Exception, run_id: str,
step_number: int, step_name: str) -> AgentError:
exc_name = type(exception).__name__
exc_msg = str(exception)
# Model-level errors
if "rate_limit" in exc_msg.lower() or "429" in exc_msg:
return AgentError(
category=AgentErrorCategory.MODEL_RATE_LIMIT,
message=exc_msg,
run_id=run_id,
step_number=step_number,
step_name=step_name,
retryable=True,
original_exception=exception,
)
if "content_filter" in exc_msg.lower() or "content policy" in exc_msg.lower():
return AgentError(
category=AgentErrorCategory.MODEL_CONTENT_FILTER,
message=exc_msg,
run_id=run_id,
step_number=step_number,
step_name=step_name,
retryable=False, # Content filter blocks are typically not retryable without changes
original_exception=exception,
)
if "token" in exc_msg.lower() and ("limit" in exc_msg.lower() or "exceed" in exc_msg.lower()):
return AgentError(
category=AgentErrorCategory.MODEL_TOKEN_LIMIT,
message=exc_msg,
run_id=run_id,
step_number=step_number,
step_name=step_name,
retryable=False,
original_exception=exception,
)
# Tool-level errors
if "timeout" in exc_msg.lower() or "timed out" in exc_msg.lower():
return AgentError(
category=AgentErrorCategory.TOOL_TIMEOUT,
message=exc_msg,
run_id=run_id,
step_number=step_number,
step_name=step_name,
retryable=True,
original_exception=exception,
)
# Parse errors
if "json" in exc_msg.lower() or "parse" in exc_msg.lower():
return AgentError(
category=AgentErrorCategory.PARSE_MALFORMED_JSON,
message=exc_msg,
run_id=run_id,
step_number=step_number,
step_name=step_name,
retryable=True, # Can retry with a corrected prompt
original_exception=exception,
)
# Default
return AgentError(
category=AgentErrorCategory.UNKNOWN,
message=exc_msg,
run_id=run_id,
step_number=step_number,
step_name=step_name,
retryable=False,
original_exception=exception,
)
class ErrorRecoveryEngine:
"""Attempts recovery strategies based on error category."""
def __init__(self, max_retries: int = 3):
self.max_retries = max_retries
self.classifier = ErrorClassifier()
self.recovery_strategies: Dict[AgentErrorCategory, Callable] = {
AgentErrorCategory.MODEL_RATE_LIMIT: self._handle_rate_limit,
AgentErrorCategory.TOOL_TIMEOUT: self._handle_tool_timeout,
AgentErrorCategory.PARSE_MALFORMED_JSON: self._handle_parse_error,
}
# Track error counts per category for alerting
self.error_counts: Dict[AgentErrorCategory, int] = {}
async def handle_error(self, exception: Exception, run_id: str,
step_number: int, step_name: str,
context: Dict) -> AgentError:
"""Classify and attempt recovery for an error."""
error = self.classifier.classify(
exception, run_id, step_number, step_name
)
error.context = context
# Increment error counter
self.error_counts[error.category] = \
self.error_counts.get(error.category, 0) + 1
# Check if we've exceeded thresholds
if self.error_counts.get(error.category, 0) > 50:
print(f"🚨 ERROR SURGE: {error.category.value} has occurred "
f"{self.error_counts[error.category]} times — investigate immediately")
# Attempt recovery if strategy exists and error is retryable
if error.retryable and error.category in self.recovery_strategies:
strategy = self.recovery_strategies[error.category]
return await strategy(error)
return error
async def _handle_rate_limit(self, error: AgentError) -> AgentError:
"""Wait and signal retry for rate limit errors."""
wait_time = 2.0 + random.uniform(0, 3.0) # 2-5 second jitter
print(f"Rate limit hit — waiting {wait_time:.1f}s before retry")
await asyncio.sleep(wait_time)
# Return the error with a flag indicating retry should proceed
error.context["retry_after_seconds"] = wait_time
return error
async def _handle_tool_timeout(self, error: AgentError) -> AgentError:
"""For tool timeouts, mark for retry with reduced payload if possible."""
error.context["retry_with_backoff"] = True
return error
async def _handle_parse_error(self, error: AgentError) -> AgentError:
"""For parse errors, we'll retry with a stronger formatting prompt."""
error.context["add_formatting_reminder"] = True
return error
def get_error_report(self) -> Dict:
"""Return a summary of recent error patterns."""
return {
"total_errors": sum(self.error_counts.values()),
"by_category": {cat.value: count for cat, count in self.error_counts.items()},
"most_common": max(self.error_counts, key=self.error_counts.get).value
if self.error_counts else None,
}
# Usage in the agent loop
error_engine = ErrorRecoveryEngine(max_retries=3)
async def resilient_agent_step(agent, messages, step_num, run_id):
"""Execute a single agent step with error classification and recovery."""
retries = 0
while retries <= error_engine.max_retries:
try:
response = await agent.call_llm_async(messages)
return response
except Exception as e:
error = await error_engine.handle_error(
exception=e,
run_id=run_id,
step_number=step_num,
step_name="llm_call",
context={"messages_count": len(messages)},
)
if not error.retryable:
raise # Non-retryable — propagate up
retries += 1
if retries > error_engine.max_retries:
raise Exception(f"Max retries ({error_engine.max_retries}) exceeded "
f"for error: {error.category.value}")
# Apply recovery context (e.g., add formatting reminder to messages)
if error.context.get("add_formatting_reminder"):
messages.append({
"role": "system",
"content": "IMPORTANT: Respond ONLY with valid JSON. No markdown, no commentary."
})
return NoneBringing It All Together: A Unified Monitoring Pipeline
In production, logs, cost data, and error events should flow into a centralized observability platform. Here's a practical integration that wires all three pillars together and exports metrics in a format compatible with Prometheus or any time-series database.
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Dict, List, Optional
import json
import threading
import time
from collections import defaultdict
@dataclass
class AgentRunMetrics:
"""Complete metrics snapshot for a single agent run."""
run_id: str
agent_name: str
agent_version: str
user_id: str
start_time: datetime
end_time: Optional[datetime] = None
total_steps: int = 0
total_tool_calls: int = 0
total_prompt_tokens: int = 0
total_completion_tokens: int = 0
total_cost_usd: float = 0.0
errors: List[Dict] = field(default_factory=list)
success: bool = False
final_output_length: int = 0
def to_prometheus_metrics(self) -> str:
"""Export metrics in Prometheus exposition format."""
labels = f'agent="{self.agent_name}",version="{self.agent_version}",success="{str(self.success).lower()}"'
lines = [
f"# HELP agent_run_duration_seconds Total run duration",
f"# TYPE agent_run_duration_seconds gauge",
f"agent_run_duration_seconds{{{labels}}} {self._duration_seconds()}",
f"# HELP agent_run_steps_total Steps per run",
f"# TYPE agent_run_steps_total gauge",
f"agent_run_steps_total{{{labels}}} {self.total_steps}",
f"# HELP agent_run_tool_calls_total Tool calls per run",
f"# TYPE agent_run_tool_calls_total gauge",
f"agent_run_tool_calls_total{{{labels}}} {self.total_tool_calls}",
f"# HELP agent_run_tokens_total Tokens consumed per run",
f"# TYPE agent_run_tokens_total gauge",
f"agent_run_tokens_total{{{labels},type=\"prompt\"}} {self.total_prompt_tokens}",
f"agent_run_tokens_total{{{labels},type=\"completion\"}} {self.total_completion_tokens}",
f"# HELP agent_run_cost_usd Cost per run in USD",
f"# TYPE agent_run_cost_usd gauge",
f"agent_run_cost_usd{{{labels}}} {self.total_cost_usd:.6f}",
f"# HELP agent_run_errors_total Error count per run",
f"# TYPE agent_run_errors_total gauge",
f"agent_run_errors_total{{{labels}}} {len(self.errors)}",
]
return "\n".join(lines) + "\n"
def _duration_seconds(self) -> float:
if self.end_time:
return (self.end_time - self.start_time).total_seconds()
return (datetime.now(timezone.utc) - self.start_time).total_seconds()
class UnifiedAgentMonitor:
"""Central monitor combining logs, costs, and error tracking."""
def __init__(self):
self.runs: Dict[str, AgentRunMetrics] = {}
self._lock = threading.Lock()
self.cost_tracker = CostTracker()
self.error_classifier = ErrorClassifier()
# Aggregate metrics across all runs
self.aggregates = defaultdict(lambda: {"count": 0, "total_cost": 0.0, "errors": 0})
def start_run(self, agent_name: str, agent_version: str, user_id: str) -> str:
"""Begin monitoring a new agent run. Returns the run_id."""
run_id = f"run_{int(time.time() * 1000)}_{agent_name}"
metrics = AgentRunMetrics(
run_id=run_id,
agent_name=agent_name,
agent_version=agent_version,
user_id=user_id,
start_time=datetime.now(timezone.utc),
)
with self._lock:
self.runs[run_id] = metrics
return run_id
def record_step(self, run_id: str, step_name: str, model: str,
prompt_tokens: int, completion_tokens: int,
tool_calls_count: int = 0):
"""Record metrics from a single agent step."""
cost = self.cost_tracker.record(run_id, step_name, model,
prompt_tokens, completion_tokens)
with self._lock:
run = self.runs.get(run_id)
if run:
run.total_steps += 1
run.total_tool_calls += tool_calls_count
run.total_prompt_tokens += prompt_tokens
run.total_completion_tokens += completion_tokens
run.total_cost_usd += cost
def record_error(self, run_id: str, exception: Exception,
step_number: int, step_name: str):
"""Record a classified error for the run."""
error = self.error_classifier.classify(
exception, run_id, step_number, step_name
)
error_dict = {
"category": error.category.value,
"message": error.message,
"step_number": step_number,
"step_name": step_name,
"retryable": error.retryable,
"timestamp": datetime.now(timezone.utc).isoformat(),
}
with self._lock:
run = self.runs.get(run_id)
if run:
run.errors.append(error_dict)
# Update aggregates
key = f"{self.runs[run_id].agent_name}:{error.category.value}"
self.aggregates[key]["errors"] += 1
def complete_run(self, run_id: str, success: bool, final_output: str):
"""Mark a run as complete and finalize metrics."""
with self._lock:
run = self.runs.get(run_id)
if run:
run.end_time = datetime.now(timezone.utc)
run.success = success
run.final_output_length = len(final_output)
# Update aggregate
agg_key = run.agent_name
self.aggregates[agg_key]["count"] += 1
self.aggregates[agg_key]["total_cost"] += run.total_cost_usd
def get_prometheus_metrics(self) -> str:
"""Export all recent run metrics in Prometheus format."""
with self._lock:
all_metrics = []
for run in self.runs.values():
if run.end_time: # Only completed runs
all_metrics.append(run.to_prometheus_metrics())
# Add aggregate metrics
for agent_key, agg in self.aggregates.items():
labels = f'agent="{agent_key}"'
all_metrics.append(
f"agent_runs_total{{{labels}}} {agg['count']}\n"
f"agent_total_cost_usd{{{labels}}} {agg['total_cost']:.4f}\n"
f"agent_total_errors{{{labels}}} {agg['errors']}\n"
)
return "\n".join(all_metrics)
def get_dashboard_summary(self) -> Dict:
"""Return a JSON-serializable summary for dashboards."""
with self._lock:
completed = [r for r in self.runs.values() if r.end_time]
success_rate = sum(1 for r in completed if r.success) / max(len(completed), 1)
avg_cost = sum(r.total_cost_usd for r in completed) / max(len(completed), 1)
avg_steps = sum(r.total_steps for r in completed) / max(len(completed), 1)
error_categories = defaultdict(int)
for r in completed:
for e in r.errors:
error_categories[e["category"]] += 1
return {
"timestamp": datetime.now(timezone.utc).isoformat(),
"total_runs_completed": len(completed),
"success_rate": round(success_rate, 4),
"average_cost_usd": round(avg_cost, 6),
"average_steps": round(avg_steps, 1),
"total_cost_all_runs": round(sum(r.total_cost_usd for r in completed), 4),
"top_error_categories": sorted(
error_categories.items(), key=lambda x: x[1], reverse=True
)[:5],
"active_runs": len([r for r in self.runs.values() if not r.end_time]),
}
# Example: wiring the monitor into an agent service
monitor = UnifiedAgentMonitor()
def run_agent_with_monitoring(agent_name: str, user_query: str, user_id: str):
"""Run an agent with full monitoring instrumentation."""
run_id = monitor.start_run(agent_name, "v2.1.0", user_id)
try:
# ... agent execution logic ...
# After each LLM call:
monitor.record_step(
run_id=run_id,
step_name="reasoning",
model="gpt-4o",
prompt_tokens=450