Agent Safety › Monitoring and Observability
Language AI HandbookAll chapters Agent Safety › Monitoring and ObservabilityOne section · Sections in this chapter5 / 12 ↓
Reading tools Definitions & inline links
Choose your expertise level to adjust how many terms are explained. Beginners see more tooltips, experts see fewer to maintain reading flow. Hover over underlined terms for instant definitions.
Article links
Make inline references clickable
Even with constraints and sandboxing in place, things can go wrong in ways that were not anticipated at design time. Monitoring provides visibility into what the agent is doing in real time, creating opportunities to detect anomalies, trigger alerts, and intervene before harm accumulates.
The core insight behind agent monitoring is that harmful behavior usually has observable precursors. Before an agent causes serious damage, it typically exhibits intermediate signals: repeated failed actions suggesting it is stuck, unusual sequences of tool calls inconsistent with the stated task, access to data outside the expected scope of the task, or durations and resource consumption that deviate from normal patterns. Monitoring that catches these signals early can stop harm before it becomes severe.
What to Monitor
Effective agent monitoring tracks several layers of the agent's behavior:
- Action log: Every tool call, recording its arguments and result along with timing and success status. This is the fundamental audit trail and the minimum viable monitoring requirement.
- Reasoning trace: The agent's intermediate thoughts or plans, if the framework exposes them. ReAct-style agents (which we covered earlier) produce explicit reasoning steps that can be inspected and logged, with suspicious steps flagged.
- Anomaly indicators: Patterns that suggest the agent is off-track, such as repeated failed tool calls, escalating resource consumption, or actions inconsistent with the original task.
- Output inspection: The final or intermediate outputs the agent produces, checked against expected content types and lengths as well as required formats.
- Cross-session patterns: Behavior across multiple sessions, to detect gradual drift or persistent misconfigurations that might not be obvious within a single session.
import time
from dataclasses import dataclass
from enum import Enum
from typing import Callable, List, Optional
class AlertSeverity(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
@dataclass
class ActionEvent:
timestamp: float
action_name: str
args: Dict[str, Any]
result: Optional[Any]
success: bool
duration_ms: float
session_id: str
@dataclass
class Alert:
severity: AlertSeverity
message: str
event: ActionEvent
rule_name: str
class AgentMonitor:
def __init__(self, session_id: str, alert_handlers: List[Callable] = None):
self.session_id = session_id
self.action_log: List[ActionEvent] = []
self.alert_handlers = alert_handlers or []
self._rules = self._build_default_rules()
def _build_default_rules(self):
return [
self._rule_repeated_failures,
self._rule_high_duration,
self._rule_sensitive_action,
]
def record_action(
self,
action_name: str,
args: Dict[str, Any],
result: Any,
success: bool,
duration_ms: float,
) -> None:
event = ActionEvent(
timestamp=time.time(),
action_name=action_name,
args=args,
result=result,
success=success,
duration_ms=duration_ms,
session_id=self.session_id,
)
self.action_log.append(event)
self._evaluate_rules(event)
def _evaluate_rules(self, event: ActionEvent) -> None:
for rule in self._rules:
alert = rule(event)
if alert:
self._fire_alert(alert)
def _rule_repeated_failures(self, event: ActionEvent) -> Optional[Alert]:
recent = self.action_log[-5:]
failures = sum(1 for e in recent if not e.success)
if failures >= 3:
return Alert(
severity=AlertSeverity.WARNING,
message="3+ consecutive failures detected in last 5 actions",
event=event,
rule_name="repeated_failures",
)
return None
def _rule_high_duration(self, event: ActionEvent) -> Optional[Alert]:
if event.duration_ms > 30_000:
return Alert(
severity=AlertSeverity.WARNING,
message=f"Action '{event.action_name}' took {event.duration_ms:.0f}ms (threshold: 30s)",
event=event,
rule_name="high_duration",
)
return None
def _rule_sensitive_action(self, event: ActionEvent) -> Optional[Alert]:
sensitive = {"delete_file", "send_email", "execute_code", "payment_api"}
if event.action_name in sensitive:
return Alert(
severity=AlertSeverity.INFO,
message=f"Sensitive action '{event.action_name}' executed",
event=event,
rule_name="sensitive_action",
)
return None
def _fire_alert(self, alert: Alert) -> None:
for handler in self.alert_handlers:
handler(alert)
def summary(self) -> Dict[str, Any]:
total = len(self.action_log)
failures = sum(1 for e in self.action_log if not e.success)
return {
"session_id": self.session_id,
"total_actions": total,
"successful": total - failures,
"failed": failures,
"failure_rate": failures / total if total > 0 else 0.0,
"avg_duration_ms": sum(e.duration_ms for e in self.action_log)
/ total
if total > 0
else 0.0,
}Monitoring agent session: [INFO] sensitive_action: Sensitive action 'execute_code' executed [WARNING] repeated_failures: 3+ consecutive failures detected in last 5 actions [WARNING] repeated_failures: 3+ consecutive failures detected in last 5 actions [INFO] sensitive_action: Sensitive action 'delete_file' executed Session summary: session_id: demo-session-001 total_actions: 7 successful: 4 failed: 3 failure_rate: 0.429 avg_duration_ms: 189.000
The monitor records every action and evaluates rules against the accumulating history. This creates both a real-time alert channel (for the alert handlers) and a post-hoc audit trail (the action log). The three default rules capture three distinct failure modes: cascading errors (repeated failures), performance degradation (high duration), and sensitive operation logging (sensitive actions).
Designing Alert Rules
Alert rules should be designed with two failure modes in mind. False positives (alerting when nothing is wrong) reduce the utility of the monitoring system because human reviewers learn to ignore alerts. False negatives (not alerting when something is wrong) defeat the purpose of monitoring. The right balance depends on the cost of the two types of errors.
For agent safety, false negatives are usually more costly than false positives because a missed harmful action can cause irreversible damage. This argues for erring on the side of more alerts, while investing in alert routing and prioritization so that the most critical alerts reach the right people quickly, and lower-severity informational alerts are logged for later review rather than interrupting operators.
A practical taxonomy for alert severity:
- CRITICAL: Stop the agent immediately and require human intervention. Examples include detecting a prompt injection attempt, an agent trying to access a disallowed resource, or an anomaly score exceeding a high threshold.
- WARNING: Flag for review but allow the agent to continue. Examples include repeated failures, slow actions, or unusual action sequences.
- INFO: Log for post-hoc analysis. Examples include sensitive action execution, task completion, or session start and end.
Anomaly Detection at Scale
Rule-based monitoring works well for known failure modes, but agents can fail in unforeseen ways. Statistical anomaly detection adds a second layer: establish baseline normal behavior from historical runs and flag sessions that deviate significantly from that baseline.
The intuition is simple. If the agent typically completes a task in 10 actions taking 5 seconds total, a session that takes 200 actions and 5 minutes is a strong signal that something unusual is happening. It might be a legitimate but unusual task. It might be a misbehaving agent. Without historical baseline data, you cannot distinguish the two. With that data, you can compute a deviation score and flag outliers for human review.
from collections import defaultdict
import numpy as np
class ActionProfiler:
"""
Builds a statistical baseline of agent behavior and
scores new sessions against that baseline.
"""
def __init__(self):
self.action_counts: Dict[str, List[int]] = defaultdict(list)
self.session_durations: List[float] = []
self.failure_rates: List[float] = []
def record_session(self, monitor: AgentMonitor) -> None:
summary = monitor.summary()
self.session_durations.append(
sum(e.duration_ms for e in monitor.action_log)
)
self.failure_rates.append(summary["failure_rate"])
action_counts_this_session: Dict[str, int] = defaultdict(int)
for event in monitor.action_log:
action_counts_this_session[event.action_name] += 1
for action, count in action_counts_this_session.items():
self.action_counts[action].append(count)
def anomaly_score(self, monitor: AgentMonitor) -> float:
"""
Returns a z-score for the session duration.
Higher scores indicate more unusual behavior.
"""
if len(self.session_durations) < 2:
return 0.0
total_duration = sum(e.duration_ms for e in monitor.action_log)
mean = np.mean(self.session_durations)
std = np.std(self.session_durations)
if std == 0:
return 0.0
return abs(total_duration - mean) / stdAnomaly detection results: Baseline sessions: 20 Baseline mean duration: 4302ms Baseline std deviation: 1505ms Current session duration: 1323ms Anomaly z-score: 1.98 Interpretation: Within normal range
A z-score above two suggests the session is behaving unusually compared to the historical baseline. This can trigger a deeper review or automatic escalation to a human operator. The score is not a perfect indicator, but it is a useful signal when combined with other monitoring signals.
The z-score on session duration is the simplest possible anomaly metric. More sophisticated baselines can track multivariate features: the joint distribution of action counts and failure rates over duration and action sequences. Isolation forests and autoencoders are two unsupervised anomaly detection methods that can be applied to the feature vectors extracted from session logs.
The plot below illustrates the detection in action: 20 baseline sessions cluster around a typical duration, while the demo session with repeated failures sits apart, producing a high anomaly score.
