s2:Permission（执行前权限判断）
s03：Permission（执行前权限判断）
核心模式
三道闸门串联成责任链：Gate 1 (硬黑名单) → Gate 2 (规则匹配) → Gate 3 (用户确认) → 执行
任何工具执行前，都要先过这三关；任何一关说"不"，就直接拦截，不会走到真正的执行。
Gate 1：硬性拒绝名单
check_deny_list，一旦命中立即拦截。基于关键字匹配，覆盖强制删除根目录、格式化磁盘等致命操作。
# 基于关键字匹配，强制删除根目录，格式化磁盘等
DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"]
def check_deny_list(command: str) -> str | None:
for pattern in DENY_LIST:
if pattern in command:
return f"Blocked: '{pattern}' is on the deny list"
return None
Gate 2：上下文规则匹配
check_rules 是一个规则引擎。每条规则是一个字典（含 tools、check、message），新增一条安全策略只需要往 PERMISSION_RULES 里追加一个字典对象，无需修改任何控制流代码。
# 规则列表，每个规则都是一个字典，包括 tools、check、message
PERMISSION_RULES = [
# 规则1: 文件操作越界
# tools（适用哪些工具）
{"tools": ["read_file", "write_file", "edit_file"],
# check（一个可执行的 lambda 函数，接收工具参数作为输入，返回布尔值，规则大脑，能够进行动态逻辑判断）
"check": lambda args: not (WORKDIR / args.get("path", "")).resolve().is_relative_to(WORKDIR),
# 规则命中时向用户展示的原因
"message": "Writing outside workspace"},
# 规则2: 危险 Shell 命令模式
{"tools": ["bash"],
"check": lambda args: any(kw in args.get("command", "") for kw in ["rm ", "> /etc/", "chmod 777"]),
"message": "Potentially destructive command"},
]
def check_rules(tool_name: str, args: dict) -> str | None:
for rule in PERMISSION_RULES:
if tool_name in rule["tools"] and rule["check"](args):
return rule["message"]
return None
Gate 3：用户审批
ask_user，最后一道防线，将最终决定权交给人类。命令行交互，向用户展示了风险原因（来自 Gate 2 的 message）和具体操作（工具名和参数），然后要求输入 y/N 确认。默认选项是 N（拒绝），这是一个安全的默认值。
人在回路中（Human-in-the-loop）：这个函数会同步阻塞整个 Agent 循环，直到用户作出决定。
def ask_user(tool_name: str, args: dict, reason: str) -> str:
print(f"\n\033[33m⚠ {reason}\033[0m")
print(f" Tool: {tool_name}({args})")
choice = input(" Allow? [y/N] ").strip().lower()
return "allow" if choice in ("y", "yes") else "deny"
权限检查总控：check_permission
将三个独立的 Gate 串联成一个完整的责任链。
def check_permission(block) -> bool:
# 1. Gate 1: 检查硬性黑名单（仅 bash）
if block.name == "bash":
reason = check_deny_list(block.input.get("command", ""))
if reason:
print(f"\n\033[31m⛔ {reason}\033[0m")
return False
# 2. Gate 2: 检查上下文规则
reason = check_rules(block.name, block.input)
# 3. Gate 3: 命中规则则触发用户审批
if reason:
# 获取用户审批结果
decision = ask_user(block.name, block.input, reason)
if decision == "deny":
return False
# 4. 所有关口通过，放行
return True
最后在核心循环中的工具执行前调用：
# 权限判断
if not check_permission(block):
results.append({"type": "tool_result", "tool_use_id": block.id, "content": "Permission denied."})
continue
# 工具执行
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown: {block.name}"
三道闸门总览
| 闸门 | 机制 | 作用 |
|---|---|---|
| Gate 1 硬黑名单 | check_deny_list 关键字匹配 |
命中立即拦截（如 rm -rf /） |
| Gate 2 规则引擎 | check_rules 基于 PERMISSION_RULES 字典 |
新增策略只加字典，不改控制流 |
| Gate 3 用户审批 | ask_user 同步阻塞 |
最后防线，默认拒绝 N（人在回路） |
本模块知识点
- 执行前权限判断
check_permission，三道闸门串联成责任链。 - 默认拒绝：
ask_user默认选项N，同步阻塞整个循环（人在回路）。 - 新增安全策略只改
PERMISSION_RULES/DENY_LIST字典，不改控制流代码。
s04：Hooks（钩子系统）
核心思想
将核心循环与扩展逻辑彻底分离：s01~s03 的 Agent 运行流程是线性和固定的，引入钩子后，生命周期变成可编程的、可编排的，可以精确控制在生命周期节点上插入什么行为。
钩子基础设施
HOOKS：事件注册表（字典），将事件名称映射到一个回调函数列表，相当于系统的事件总线。register_hook：订阅机制，把 callback 添加到某事件的监听器列表中。trigger_hooks：发布机制，事件发生时，遍历执行该事件的所有回调函数。
# 字典，系统的事件总线
# 预定义了四个 agent 生命周期中的关键锚点：
# 用户提交问题后 工具调用执行前 工具调用执行后 循环即将结束时
HOOKS = {"UserPromptSubmit": [], "PreToolUse": [], "PostToolUse": [], "Stop": []}
# 将 event 事件添加到 HOOKS 字典对应时机 key 的 value 中
def register_hook(event: str, callback):
HOOKS[event].append(callback)
# FIFO 顺序执行回调链
def trigger_hooks(event: str, *args):
for callback in HOOKS[event]:
result = callback(*args)
# 若返回了非空结果，整个链条立即中断，将该结果作为阻断信号返回
if result is not None:
return result
return None
关键机制：回调按 FIFO（先进先出） 顺序执行；任一回调返回非空结果，整个链条立即中断，该结果作为"阻断信号"返回（例如权限被拒）。返回
None表示"放行，继续执行下一个"。
四大生命周期锚点
| 事件 | 触发时机 | 典型用途 |
|---|---|---|
UserPromptSubmit |
用户提交问题后、送入 LLM 前 | 注入上下文、内容审核 |
PreToolUse |
工具调用执行前 | 权限检查（替代 s03）、日志 |
PostToolUse |
工具调用执行后 | 大输出警告、监控 |
Stop |
循环即将结束时 | 统计、清理、动态注入新任务 |
预置钩子实现
permission_hook — PreToolUse
s03 中 check_permission 的平替，逻辑完全一致，但作为一个插件存在。把权限逻辑移出核心循环，极大净化了 agent_loop。
def permission_hook(block):
# 1. 检查 bash 工具的黑名单
if block.name == "bash":
for pattern in DENY_LIST: ...
for kw in DESTRUCTIVE: ...
# 2. 检查文件工具的工作区越界
if block.name in ("read_file", "write_file", "edit_file"):
if not (WORKDIR / path).resolve().is_relative_to(WORKDIR): ...
return None
log_hook(block) — PreToolUse
纯粹的观察者，不干预流程。在每个工具执行前打印一条日志，展示了钩子最常见的用途：透明地织入非功能性需求（如日志、监控）。
def log_hook(block):
args_preview = str(list(block.input.values())[:2])[:60]
print(f"\033[90m[HOOK] {block.name}({args_preview})\033[0m")
return None
large_output_hook(block, output) — PostToolUse
事后检查器：工具执行后，检查输出是否过大并发出警告。不影响数据流（输出没有被截断，完整地返回给了 LLM），但通过日志提醒开发者可能需要关注上下文大小的风险。这是监控能力的一种优雅实现。
def large_output_hook(block, output):
if len(str(output)) > 100000:
print(f"\033[33m[HOOK] ⚠ Large output from ...\033[0m")
return None
context_inject_hook(query) — UserPromptSubmit
用户输入被送入 LLM 之前进行拦截。目前只是记录日志，但其潜力巨大：未来可以在这里动态修改用户查询、注入上下文信息，或作为内容审核的关口。
def context_inject_hook(query: str):
print(f"\033[90m[HOOK] UserPromptSubmit: working in {WORKDIR}\033[0m")
return None
summary_hook(messages) — Stop
后置生命周期处理器，在 Agent 停止时执行清理或统计。它遍历整个对话历史 messages，统计工具调用次数。
def summary_hook(messages: list):
tool_count = sum(1 for m in messages ...)
print(f"\033[90m[HOOK] Stop: session used {tool_count} tool calls\033[0m")
return None
钩子系统注册
# 启动时注册所有钩子
register_hook("UserPromptSubmit", context_inject_hook)
register_hook("PreToolUse", permission_hook)
register_hook("PreToolUse", log_hook)
register_hook("PostToolUse", large_output_hook)
register_hook("Stop", summary_hook)
循环中的集成点
def agent_loop(messages: list):
while True:
...
if response.stop_reason != "tool_use":
# 循环结束前调用
force = trigger_hooks("Stop", messages)
if force:
# 非空结果，将其作为新的用户消息追加，并 continue，
# 提供了一个在循环结束前动态注入新任务的扩展点
messages.append({"role": "user", "content": force})
continue
return
...
# PreToolUse 事件钩子（替代了 s03 的 check_permission）
blocked = trigger_hooks("PreToolUse", block)
if blocked:
results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(blocked)})
continue
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown: {block.name}"
# PostToolUse 事件钩子
trigger_hooks("PostToolUse", block, output) # s04: post hook
...
执行入口的集成点
if __name__ == "__main__":
...
while True:
... # 获取 query
# 用户输入 prompt 后调用
trigger_hooks("UserPromptSubmit", query)
history.append({"role": "user", "content": query})
agent_loop(history)
...
本模块知识点
- 基础设施：
HOOKS事件总线 +register_hook订阅 +trigger_hooks发布（FIFO，非空结果即中断链条）。 - 预置钩子：
permission_hook（替代 s03）、log_hook（观察者）、large_output_hook（事后检查）、context_inject_hook、summary_hook。 - 四大锚点：
UserPromptSubmit/PreToolUse/PostToolUse/Stop。 - 钩子把权限、日志、监控等非功能性需求从核心循环里彻底抽离。
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来自 茯茶养生人的博客！
