name: hook-event-architecture description: 为 ADW 可观察性设计基于钩子的事件系统。用于实现实时事件广播、创建钩子管道或构建代理活动监控。 allowed-tools: 阅读、查找、全局
钩子事件架构
设计基于钩子的事件系统,以捕获和广播AI开发者工作流中的代理活动。
强制:文档管理委托
文档验证: 此技能引用Claude Code钩子事件和配置,这些可能随版本而变化。在实现之前,请调用
hook-management技能以验证当前事件类型和钩子配置模式。
验证检查点:
- [ ] 我是否调用了
hook-management以获取当前事件类型? - [ ] 官方文档是否确认了事件模式?
- [ ] 我的实现是否基于官方文档?
何时使用
- 实现实时事件广播
- 构建可观察性基础设施
- 创建泳道可视化
- 将代理活动记录到数据库
- 生成AI驱动的事件摘要
先决条件
- 了解Claude Code钩子 (@hook-event-patterns.md)
- 熟悉WebSocket模式 (@websocket-architecture.md)
- 访问Claude Agent SDK以进行完整实现
SDK要求
实现说明: 完整的钩子事件架构需要具有自定义工具Claude Agent SDK。此技能提供设计模式和规范。
事件类型
文档验证: 此处列出的事件类型是Claude Code内部类型,可能随版本而变化。要获取权威的当前事件类型,请查询
hook-management技能,该技能委托docs-management获取官方文档。
ADW系统捕获以下事件类型:
| 事件类型 | 图标 | 来源 | 载荷 |
|---|---|---|---|
PreToolUse |
🪝 | 钩子 | 工具名称、输入、会话 |
PostToolUse |
🪝 | 钩子 | 工具名称、输出、持续时间 |
TextBlock |
💬 | 代理 | 响应文本、令牌 |
ToolUseBlock |
🛠️ | 代理 | 工具调用记录 |
ThinkingBlock |
🧠 | 代理 | 扩展思考内容 |
StepStart |
⚙️ | 系统 | 步骤名称、输入 |
StepEnd |
⚙️ | 系统 | 步骤名称、输出、持续时间 |
架构设计流程
步骤 1:定义事件模式
创建事件的Pydantic模型:
class ADWEvent(BaseModel):
type: str # 事件类型来自表格
adw_id: str # 8字符相关ID
step: str # 当前步骤名称
timestamp: datetime
payload: dict # 类型特定数据
summary: str | None # AI生成的摘要
步骤 2:配置钩子触发器
设置Claude Code钩子:
{
"hooks": {
"PreToolUse": [{
"matcher": ".*",
"command": "python hooks/pre_tool.py"
}],
"PostToolUse": [{
"matcher": ".*",
"command": "python hooks/post_tool.py"
}]
}
}
步骤 3:设计事件管道
代理执行
│
├── PreToolUse ──► 钩子脚本 ──┬── 记录到数据库
│ ├── 摘要(Haiku)
▼ └── 广播(WebSocket)
工具执行
│
├── PostToolUse ──► 钩子脚本 ──┬── 记录到数据库
│ ├── 摘要(Haiku)
▼ └── 广播(WebSocket)
继续...
步骤 4:实现摘要生成
使用Haiku生成AI事件摘要:
async def summarize_event(event: ADWEvent) -> str:
prompt = f"""用15个词或更少总结:
事件:{event.type}
工具:{event.payload.get('tool_name', 'N/A')}
数据:{str(event.payload)[:500]}
"""
return await claude.complete(prompt, model="haiku")
步骤 5:设计广播模式
事件分发到客户端:
class EventBroadcaster:
def __init__(self, ws_manager, db_client):
self.ws = ws_manager
self.db = db_client
async def broadcast(self, event: ADWEvent):
# 首先记录到数据库
await self.db.log_event(event)
# 广播到WebSocket客户端
await self.ws.broadcast(event.dict())
钩子脚本模板
PreToolUse钩子
#!/usr/bin/env python
import sys, json, asyncio
from adw_modules import broadcast, summarize
async def main():
data = json.load(sys.stdin)
event = {
"type": "PreToolUse",
"adw_id": data.get("adw_id"),
"step": data.get("step"),
"payload": {
"tool_name": data["tool_name"],
"tool_input": data["tool_input"]
}
}
event["summary"] = await summarize(event)
await broadcast(event)
if __name__ == "__main__":
asyncio.run(main())
PostToolUse钩子
#!/usr/bin/env python
import sys, json, asyncio
from adw_modules import broadcast, summarize
async def main():
data = json.load(sys.stdin)
event = {
"type": "PostToolUse",
"adw_id": data.get("adw_id"),
"step": data.get("step"),
"payload": {
"tool_name": data["tool_name"],
"tool_output": data.get("tool_output", "")[:1000],
"duration_ms": data.get("duration_ms", 0)
}
}
event["summary"] = await summarize(event)
await broadcast(event)
if __name__ == "__main__":
asyncio.run(main())
输出格式
设计钩子事件架构时:
## 钩子事件架构设计
### 事件类型
| 类型 | 触发器 | 载荷模式 |
| --- | --- | --- |
| [类型] | [何时触发] | [字段] |
### 钩子配置
```json
[hooks.json配置]
事件管道
[ASCII流程图]
摘要策略
[Haiku如何生成摘要]
广播模式
[WebSocket或其他广播机制]
数据库模式
[事件记录表]
实现检查清单
- [ ] [步骤 1]
- [ ] [步骤 2] …
## 设计检查清单
- [ ] 定义了事件类型和载荷
- [ ] 指定了钩子配置
- [ ] 记录了事件管道
- [ ] 设计了摘要提示
- [ ] 选择了广播机制
- [ ] 定义了数据库模式
- [ ] 考虑了错误处理
- [ ] 评估了性能影响
## 反模式
| 反模式 | 问题 | 解决方案 |
| --- | --- | --- |
| 同步广播 | 阻塞代理执行 | 异步分发 |
| 无相关ID | 无法跟踪工作流 | 使用 adw_id |
| 原始载荷记录 | 令牌浪费 | 截断大数据 |
| 缺少摘要 | 难以扫描 | 始终生成摘要 |
| 无错误处理 | 静默失败 | 记录并恢复 |
## 交叉引用
- @hook-event-patterns.md - 事件类型详情
- @websocket-architecture.md - 广播模式
- @production-patterns.md - 数据库记录
- @adw-framework.md - ADW概述
## 版本历史
- **v1.0.0** (2026-01-01): 初始发布(第14课)
---
## 最后更新
**日期:** 2026-01-01
**模型:** claude-opus-4-5-20251101