Chapter 21 · Human-in-the-Loop and Guardrails

Learning objectives: Install "brakes" on your Agent — implement HITL via tool confirmation, intercept sensitive inputs and dangerous outputs with pre/post hooks and built-in Guardrails, and design sensible failure degradation strategies.

21.1 Why Agents Need "Brakes"

The risks of autonomous Agents come from two ends: the input (users may inject malicious instructions or submit private data) and the output (the model may call irreversible operations like dropping tables or transferring funds). The engineering countermeasure is a three-layer defense:

1. Tool confirmation (Human-in-the-Loop): pause before executing dangerous operations and wait for human approval;
2. Guardrails (pre_hooks): validate / sanitize / intercept before input reaches the model;
3. Output inspection (post_hooks): compliance review before the reply reaches the user.

Version note

Guardrails and hooks were introduced in Agno v2.1.0; tool confirmation (requires_confirmation) is likewise standard HITL practice. This tutorial is based on Agno 3.x.

21.2 Tool Confirmation: Letting a Dangerous Operation Wait for a Human

Add @tool(requires_confirmation=True) to a custom tool. When the Agent reaches that tool it pauses (is_paused=True), hands the pending tool name and arguments to you for approval, and continues with continue_run() once approved:

# hitl_confirm.py
import os
from agno.agent import Agent
from agno.db.sqlite import SqliteDb          # Database needed to persist run state for continuation
from agno.models.openai import OpenAIChat
from agno.tools import tool

@tool(requires_confirmation=True)
def refund_order(order_id: str, amount: float) -> str:
    """Issue a refund for the specified order — a sensitive financial operation."""
    return f"Order {order_id} refunded {amount} CNY"

agent = Agent(
    model=OpenAIChat(
        id="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    ),
    tools=[refund_order],
    db=SqliteDb(db_file="tmp/hitl.db"),      # Must persist run state to enable continuation
)

run_response = agent.run("Please refund order A1001 for 99 yuan")

# Approval step: iterate pending requirements
for requirement in run_response.active_requirements:
    if requirement.needs_confirmation:
        print(f"Pending confirmation: {requirement.tool_execution.tool_name}"
              f"{requirement.tool_execution.tool_args}")
        if input("Approve? (y/n): ").lower() == "y":
            requirement.confirm()
        else:
            requirement.reject()

# Continue execution with the approval result
response = agent.continue_run(
    run_id=run_response.run_id,
    requirements=run_response.requirements,
)
print(response.content)

For Toolkit built-in tools, use requires_confirmation_tools=["tool_name"] to protect only the specified function while others execute automatically.

21.3 Built-in Guardrails: PII and Prompt Injection Defense

Guardrails mount as pre_hooks, running before input reaches the model. Agno ships three built-in guardrails:

from agno.agent import Agent
from agno.guardrails import PIIDetectionGuardrail, PromptInjectionGuardrail

support_agent = Agent(
    model=model,
    pre_hooks=[
        PIIDetectionGuardrail(),        # Blocks national ID numbers, phone numbers, and other PII
        PromptInjectionGuardrail(),     # Detects injections like "ignore all previous instructions"
    ],
    markdown=True,
)
# Input containing a phone number will be blocked directly, raising InputCheckError

21.4 Custom Guardrails

Inherit BaseGuardrail and implement check() / async_check() (for sync .run() and async .arun() respectively); raise InputCheckError when a violation is detected:

import re
from agno.exceptions import CheckTrigger, InputCheckError
from agno.guardrails import BaseGuardrail
from agno.run.agent import RunInput

class SensitiveWordGuardrail(BaseGuardrail):
    """Block inbound咨询 inputs that contain competitor names."""

    def check(self, run_input: RunInput) -> None:
        if isinstance(run_input.input_content, str):
            if re.search(r"某友|某蝶", run_input.input_content):
                raise InputCheckError(
                    "Input contains content that cannot be discussed.",
                    check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
                )

    async def async_check(self, run_input: RunInput) -> None:
        self.check(run_input)   # For simple rules the async version can reuse the sync logic

# Usage is identical to built-in guardrails: pre_hooks=[SensitiveWordGuardrail()]

Output-side inspection uses a post_hook: it receives the generated result before it reaches the user, making it suitable for sensitive-word scanning, compliance rewriting, or tagging.

from datetime import datetime

def audit_log_hook(run_input, run_output):
    """post_hook: persist every input and output to disk for compliance audit."""
    with open("tmp/audit.log", "a", encoding="utf-8") as f:
        f.write(f"[{datetime.now():%F %T}] IN: {run_input.input_content}\n")
        f.write(f"[{datetime.now():%F %T}] OUT: {run_output.content}\n")

compliance_agent = Agent(
    model=model,
    pre_hooks=[SensitiveWordGuardrail()],   # Pre-check
    post_hooks=[audit_log_hook],            # Post-audit (can also rewrite/tag)
)

21.5 Failure Degradation Strategy

Blocking is not the end goal; a production system must answer "what happens after a block":

• Fixable errors (format issues): auto-rewrite and retry once in a post_hook;
• Requires human intervention: catch InputCheckError and route to a manual queue, logging the reason for audit;
• Reject without crashing: return a polite fallback response externally; preserve the full error code internally;
• Gradual rollout: new guardrails start in a "log-only, no-block" mode to measure false-positive rate, then gradually switch to enforced blocking.

Chapter Summary

• Three-layer defense: tool confirmation (during execution), pre_hooks (input-side, before), post_hooks (output-side, after);
• @tool(requires_confirmation=True) + active_requirements + continue_run() is the standard HITL flow; a db must be configured to persist run state;
• Built-in Guardrails mount via pre_hooks=[...]; custom ones inherit BaseGuardrail, implement check()/async_check(), and raise InputCheckError;
• After a block there must be a clear degradation path: retry, route to human, fallback response, gradual rollout.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. When a tool marked requires_confirmation=True is triggered, what happens to the Agent?

2. Why must you configure db when using the tool confirmation mechanism?

3. Which class should a custom Guardrail inherit from, and what should it do on violation?

4. When do pre_hooks execute?

🛠️ Hands-On Exercises

1. Write a "delete file" tool with requires_confirmation=True; run through the full pause → reject → continue sequence and observe how the Agent responds after rejection.
2. Implement a custom Guardrail that blocks inputs exceeding 500 characters; verify that InputCheckError is raised correctly with a long text payload.
3. Combine PIIDetectionGuardrail with a refund-confirmation tool for a customer-service Agent, simulating a dual-block scenario where the user provides a phone number and requests a refund.

Next chapter broadens the horizon to the external world: Chapter 22: MCP Protocol Integration.