Agent Safety › Human-in-the-Loop Intervention
Language AI HandbookAll chapters Agent Safety › Human-in-the-Loop InterventionOne section · Sections in this chapter6 / 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
Monitoring detects problems. Intervention prevents or limits their consequences. The key design question is when to pause execution and require human input, versus when to let the agent continue autonomously.
This question involves a fundamental tradeoff. More human involvement means greater safety but lower throughput. An agent that pauses for confirmation on every action is barely more autonomous than a human clicking through a workflow manually. An agent that never pauses runs faster but accumulates more risk. The right calibration depends on the domain, the stakes, and the maturity of the system.
The answer for most production agent deployments is selective intervention: the agent runs autonomously for most actions but pauses for a specific, well-defined set of high-stakes actions. The criteria for "high-stakes" are typically irreversibility (cannot be undone), scope (affects many records or people), cost (involves money or significant compute), and external visibility (triggers communication outside the system).
Confirmation Gates
A confirmation gate is a mechanism that pauses agent execution and requests explicit human approval before proceeding with a high-risk action. Gates are placed around actions that are irreversible, expensive, or sensitive. The agent presents the proposed action, its arguments, and the context that led to it, and waits for a human to approve or deny it.
Effective confirmation gates present enough context for an informed decision. A gate that simply says "Approve delete_file?" is less useful than one that says "The agent is about to delete config.yaml as part of the cleanup step in the data pipeline task. The file was last modified 3 days ago. Approve?" The latter gives the reviewer enough information to understand what the agent is trying to do and whether the action is appropriate.
from typing import Protocol
class ConfirmationProvider(Protocol):
async def request_confirmation(
self, action_name: str, args: Dict[str, Any], context: str
) -> bool: ...
class ConsoleConfirmationProvider:
"""Simple confirmation provider for development and testing."""
async def request_confirmation(
self, action_name: str, args: Dict[str, Any], context: str
) -> bool:
print("\n[CONFIRMATION REQUIRED]")
print(f" Action: {action_name}")
print(f" Arguments: {args}")
print(f" Context: {context}")
response = input(" Allow? (y/n): ").strip().lower()
return response == "y"
class SafeActionExecutor:
"""
Wraps tool execution with validation, sandboxing checks,
and human confirmation for gated actions.
"""
def __init__(
self,
permissions: AgentPermissions,
validator: ActionValidator,
monitor: AgentMonitor,
confirmation_provider: Optional[ConfirmationProvider] = None,
):
self.permissions = permissions
self.validator = validator
self.monitor = monitor
self.confirmation_provider = confirmation_provider
async def execute(
self, action_name: str, args: Dict[str, Any], context: str = ""
) -> Any:
# Step 1: Validate
self.validator.validate(action_name, args)
# Step 2: Request confirmation if required
if self.validator.requires_confirmation(action_name):
if self.confirmation_provider is None:
raise ActionValidationError(
f"Action '{action_name}' requires confirmation but no "
f"confirmation provider is configured"
)
approved = await self.confirmation_provider.request_confirmation(
action_name, args, context
)
if not approved:
raise ActionValidationError(
f"Action '{action_name}' was denied by the human operator"
)
# Step 3: Execute (actual tool call happens here)
start = time.time()
try:
result = self._dispatch(action_name, args)
duration_ms = (time.time() - start) * 1000
self.monitor.record_action(
action_name, args, result, True, duration_ms
)
return result
except Exception:
duration_ms = (time.time() - start) * 1000
self.monitor.record_action(
action_name, args, None, False, duration_ms
)
raise
def _dispatch(self, action_name: str, args: Dict[str, Any]) -> Any:
# Stub: real implementation would call actual tools
return f"Result of {action_name}({args})"SafeActionExecutor workflow: 1. Validate: Check action against permissions and path safety rules 2. Confirm?: If action is in requires_confirmation set, pause and ask human 3. Execute: Dispatch to actual tool implementation 4. Monitor: Record event (success/failure, duration) in action log 5. Alert?: Evaluate monitoring rules, fire alerts if triggered Actions requiring confirmation in STANDARD profile: - write_file
The SafeActionExecutor integrates validation and confirmation before execution, with monitoring throughout, into a single composable component. Each layer has a distinct responsibility and can be updated independently. The confirmation provider is injected as a dependency, making it easy to swap between a console prompt in development, a Slack message in production, and a mock approval in tests.
Timeout-Bounded Confirmation
One practical concern with confirmation gates is latency. If a human reviewer is not watching the system, the agent can stall indefinitely waiting for approval. Production implementations typically pair confirmation requests with a timeout: if no response arrives within a specified window, the agent either cancels the action (conservative default) or proceeds with it (for actions where the cost of cancellation is high). The right default depends on the action's risk profile.
A second concern is confirmation fatigue. If the system asks for approval too frequently, reviewers start rubber-stamping requests without meaningful review. This is worse than no confirmation gate because it creates the illusion of oversight without the substance. Restricting confirmation to high-risk actions improves both throughput and the quality of human review.
Graceful Degradation and Abort
Not all interventions require human input. When an agent detects that it is stuck in a failure loop, has exceeded a resource budget, or is about to take an action inconsistent with its original task, it can trigger a graceful abort: stopping execution cleanly, preserving state, and reporting what happened.
Graceful abort is different from unhandled failure. An unhandled failure might leave the system in an inconsistent state: files half-written, database transactions uncommitted, external services partially notified. A graceful abort ensures that the system ends in a clean, known state that can be safely inspected and recovered from.
@dataclass
class AgentExecutionResult:
success: bool
completed_steps: int
total_steps: int
abort_reason: Optional[str]
action_summary: Dict[str, Any]
class SafeAgentRunner:
"""
Runs an agent plan with safety guardrails including
failure limits, resource budgets, and clean abort.
"""
def __init__(
self,
max_failures: int = 3,
max_actions: int = 50,
monitor: Optional[AgentMonitor] = None,
):
self.max_failures = max_failures
self.max_actions = max_actions
self.monitor = monitor or AgentMonitor(session_id="default")
self._consecutive_failures = 0
def should_abort(self) -> Optional[str]:
if self._consecutive_failures >= self.max_failures:
return (
f"Too many consecutive failures ({self._consecutive_failures})"
)
if len(self.monitor.action_log) >= self.max_actions:
return f"Action budget exceeded ({self.max_actions} actions)"
return None
def run_step(self, step_fn) -> bool:
"""Returns True if step succeeded, False otherwise."""
try:
step_fn()
self._consecutive_failures = 0
return True
except Exception:
self._consecutive_failures += 1
return False
def run_plan(self, steps: List[Callable]) -> AgentExecutionResult:
for i, step in enumerate(steps):
abort_reason = self.should_abort()
if abort_reason:
return AgentExecutionResult(
success=False,
completed_steps=i,
total_steps=len(steps),
abort_reason=abort_reason,
action_summary=self.monitor.summary(),
)
self.run_step(step)
return AgentExecutionResult(
success=True,
completed_steps=len(steps),
total_steps=len(steps),
abort_reason=None,
action_summary=self.monitor.summary(),
)Agent execution result: Success: False Completed steps: 5/6 Abort reason: Too many consecutive failures (3) Actions in log: 2
The runner aborts cleanly after three consecutive failures, preventing an agent from thrashing repeatedly on a broken step or consuming unbounded resources. The AgentExecutionResult captures enough information to reconstruct what happened: how many steps completed, where it stopped, and why. This information is essential for debugging and for communicating with operators.