Agent中的Hook:简单示例带你十分钟搞懂实现原理
·
Agent中的Hook:简单示例带你十分钟搞懂实现原理
我们在Agent开发和使用中经常听到Hook,本文意在简介Hook,并以一个简单易理解的例子带你理解Hook的实现机制。
Hook是什么?
Hook 就是注册在特定事件上的回调函数。它是一种编程模式,允许在特定事件发生时插入自定义逻辑。在Agent开发中使用Hook优点如下:
- 解耦:将扩展逻辑从主流程中分离,保持核心代码简洁
- 可扩展:新增功能只需注册新 hook,无需修改原有代码
核心思想:事件发生 → 触发 hook → 执行hook函数 → 继续原流程
下面用一个简单示例来理解Hook
1. 定义一个Hook注册表
HOOKS = {"UserPromptSubmit": [], "PreToolUse": []}
其中 key 表示触发 Hook 的事件名称(可以理解为触发的时机,例如 “UserPromptSubmit” 指用户提交 prompt 之后,prompt 进入大模型之前),value 是一个列表,存储该事件触发时需要执行的 hook。这里的一个 KV 结构就是一个事件的 hook 注册表,一个事件可以注册多个 hook 函数。
2. 提供注册Hook的函数
def register_hook(event: str, callback):
HOOKS[event].append(callback)
register_hook 函数有两个参数:
event:字符串,指定要注册回调的事件名称callback:事件发生时要执行的函数
注册动作就是将 callback 函数添加到对应 event 的列表中。
示例:
# 定义 Hook 注册表
HOOKS = {"on_start": []}
# 提供 Hook 注册函数
def register_hook(event: str, callback):
HOOKS[event].append(callback)
# 测试函数
def say_hello():
print("Hello!")
# 将 say_hello 函数注册到 "on_start" 事件列表
register_hook("on_start", say_hello)
3. 提供触发Hook的函数
def trigger_hooks(event: str, *args):
for callback in HOOKS[event]:
result = callback(*args)
if result is not None:
return result
return None
这段代码是 register_hook() 的配套实现,用来触发某个事件的所有回调函数。
执行逻辑:
- 传入事件名称,函数遍历该事件下所有已注册的回调函数
- 依次执行每个回调,将
*args作为参数传入 - 阻断机制:如果某个回调返回非
None值,立即返回该结果,后续 hook 不再执行 - 如果所有回调都返回
None,最终返回None
阻断机制的应用:
def permission_check(block):
if block.command == "rm -rf /":
return "Permission denied" # 返回错误信息,阻断后续执行
return None # 允许继续
def log_action(block):
print(f"Executing: {block.command}")
return None # 不阻断
register_hook("PreToolUse", permission_check)
register_hook("PreToolUse", log_action)
当触发 “PreToolUse” 时:
- 如果
permission_check返回错误信息,log_action不会执行 - 如果
permission_check返回None,继续执行log_action
4. 在特定时机触发该Hook
补充刚才的例子,在合适的时机触发 “on_start” 事件的所有函数:
def main():
# 前置代码逻辑
trigger_hooks("on_start")
# 后置代码逻辑
下面举例一个完整示例
在 Agent Loop 循环中,通过 Hook 实现用户 prompt 日志记录和危险命令拦截:
# 1.定义注册表
HOOKS = {"UserPromptSubmit": [], "PreToolUse": []}
#注册hook的函数
def register_hook(event, callback):
HOOKS[event].append(callback)
#触发hook的函数
def trigger_hooks(event, *args):
for callback in HOOKS[event]:
result = callback(*args)
if result is not None:
return result
return None
# 2.定义 hook 函数
#记录用户prompt的函数
def log_user_prompt(prompt):
print(f"[LOG] User submitted: {prompt[:50]}...")
return None
# PreToolUse hook:拦截危险命令
DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot"]
#执行危险命令拦截的函数
def block_dangerous_commands(tool_name, command):
if tool_name == "bash":
for pattern in DENY_LIST:
if pattern in command:
return f"Blocked: dangerous command '{pattern}'"
return None
#记录工具调用的函数
def log_tool_call(tool_name, command):
print(f"[LOG] Tool called: {tool_name}({command[:30]}...)")
return None
# 3.注册 hook
register_hook("UserPromptSubmit", log_user_prompt)
register_hook("PreToolUse", block_dangerous_commands)
register_hook("PreToolUse", log_tool_call)
# 4.在Agent Loop 中使用 hook
def agent_loop():
# 模拟用户输入
user_prompt = "帮我删除根目录"
# UserPromptSubmit 事件 <--------------------------在prompt发LLM之前做日志记录
trigger_hooks("UserPromptSubmit", user_prompt)
# 模拟 LLM 决策调用 bash 工具
tool_name = "bash"
command = "rm -rf /"
# PreToolUse 事件 <------------------------在执行命令之前做危险命令拦截
error = trigger_hooks("PreToolUse", tool_name, command)
if error:
print(f"[ERROR] {error}")
return
# 正常执行工具
print(f"[EXEC] Running: {command}")
# 测试
agent_loop()
输出:
[LOG] User submitted: 帮我删除根目录...
[ERROR] Blocked: dangerous command 'rm -rf /'
用户的危险命令被“PreToolUse”事件的block_dangerous_commands函数拦截
执行流程:
用户输入 "帮我删除根目录"
↓
触发 UserPromptSubmit hook
→ log_user_prompt() 记录日志
↓
LLM 决策调用 bash("rm -rf /")
↓
触发 PreToolUse hooks
→ block_dangerous_commands() 检测到危险命令,返回错误信息
→ log_tool_call() 被阻断,不执行
↓
agent_loop 收到错误,终止执行
以下是Hook在Agent中一些实际应用场景
可以按照本文上述流程实现以下场景应用:
| 场景 | Hook 类型 | 示例 |
|---|---|---|
| 权限控制 | Pre-hook | 检查命令是否安全,拒绝危险操作 |
| 日志记录 | Pre/Post-hook | 记录每次工具调用的参数和结果 |
| 数据校验 | Pre-hook | 验证输入数据的合法性 |
| 通知提醒 | Post-hook | 操作完成后发送通知 |
| 状态追踪 | Pre/Post-hook | 更新任务进度、统计调用次数 |
| 环境检查 | Pre-hook | 检查依赖、权限、配置等 |
注:本文参考了 GitHub 开源学习项目:learn-claude-code。这是一个非常优秀的学习资源,推荐读者结合项目内容一同学习。
更多推荐
所有评论(0)