第 5 章：用 Hooks 控制 Agent

在 YouTube 观看

关于本课

本课程的视频是某个时间点的快照。Strands 一直在活跃开发，所以本页的代码反映的是最新写法，视频里讲的概念依然适用。拿不准的时候，以代码为准。

本课代码：samples/05-hooks

不依赖 Model 的规则

Hooks 在生命周期事件上注入代码（Tool 调用前后、agent loop 前后），不改变 Agent 的逻辑。Tool 由 Model 决定用不用，Hook 不一样，它每次都会自动触发，不管 Model 推理出什么。关键区别就在这里：prompt 是请求，Hook 是保证。

一个失控的循环可能把同一个 Tool 调用几十次。Model 也可能不问一声就去执行破坏性操作。Prompting 是请 Model 好好表现，Hooks 则保证它这么表现。

Hook 架构

• 继承 HookProvider 来定义 Hook
• 在 register_hooks() 里为生命周期事件注册 callback
• 多个 Hook 可以监听同一个事件（可叠加）
• event.interrupt() 暂停循环，等待人工输入
• event.cancel_tool 阻止 Tool 执行，并把原因作为消息回传给 Model

示例：删除操作的人工审批

from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent, HookProvider, HookRegistry

class DeleteApprovalHook(HookProvider):
    """Intercepts delete operations for human approval."""

    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeToolCallEvent, self.check_delete)

    def check_delete(self, event: BeforeToolCallEvent) -> None:
        if event.tool_use["name"] != "delete_file":
            return

        approval = event.interrupt(
            "delete-approval",
            reason={"path": event.tool_use["input"]["path"]}
        )

        if approval.lower() != "y":
            event.cancel_tool = "User denied file deletion"

agent = Agent(
    tools=[list_files, read_file, write_file, delete_file],
    hooks=[DeleteApprovalHook()],
)

📂 approval_interrupt.py

event.interrupt() 会暂停 Agent 执行，把控制权交回调用方。审批类的工作流就靠它实现：人不回答，Agent 就不往下走。

示例：限制 Tool 调用频率

from strands.hooks import BeforeInvocationEvent, BeforeToolCallEvent, HookProvider, HookRegistry

class LimitToolCounts(HookProvider):
    def __init__(self, max_calls: int = 3):
        self.max_calls = max_calls
        self.counts: dict[str, int] = {}

    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeInvocationEvent, self.reset)
        registry.add_callback(BeforeToolCallEvent, self.check)

    def reset(self, event: BeforeInvocationEvent) -> None:
        self.counts = {}

    def check(self, event: BeforeToolCallEvent) -> None:
        name = event.tool_use["name"]
        self.counts[name] = self.counts.get(name, 0) + 1
        if self.counts[name] > self.max_calls:
            event.cancel_tool = (
                f"'{name}' hit the {self.max_calls}-call limit. "
                "Do NOT call this tool again."
            )

agent = Agent(tools=[get_weather], hooks=[LimitToolCounts(max_calls=3)])

📂 rate_limiter.py

注意计数器是在 BeforeInvocationEvent 时重置的。频率限制是每个请求一份，不是整个 Agent 生命周期共用。另外，因为 cancel_tool 会把消息回传给 Model，Model 就知道了这次调用为什么被拦，不会再反复重试。

何时使用 Hooks

• 限制 Tool 的使用频率，防止失控循环
• 破坏性操作前要求人工审批
• 记录每一次 Tool 调用，留作审计轨迹
• 按业务规则校验 Tool 的输入和输出
• 对敏感 Tool 做访问控制

Hooks 不碰 Tool，也不碰 prompt。它是一层独立的、可复用的东西，可以随手加到任何 Agent 上。

参考资源

• 📖 Hooks
• 📖 Lifecycle Controls