Pi Agent: 事件系统与命令注册
最后更新：2026-08-31
事件系统让 Agent 不再是"你问它答"的单线模式，而是"有来有往"的双向通信。
1. 事件系统概览
Pi Agent 采用事件驱动架构：
TEXT
📖 仅展示
事件流
触发源 → 事件总线 → 处理器
用户输入 分发路由 工具调用
工具返回 过滤排序 UI 更新
定时器 优先级排序 日志记录
外部消息 错误路由 通知发送
2. 内置事件
| 事件名 | 触发时机 | 数据 |
|---|---|---|
| on_chat_start | 对话开始 | session_id |
| on_chat_end | 对话结束 | session_id, summary |
| on_user_message | 用户发送消息 | message |
| on_agent_response | Agent 回复 | response |
| on_tool_call | 工具调用 | tool_name, params |
| on_tool_result | 工具返回结果 | tool_name, result |
| on_error | 发生错误 | error, context |
| on_model_switch | 切换模型 | old_model, new_model |
| on_context_overflow | 上下文溢出 | size, limit |
3. 事件监听
(1) 装饰器方式
PYTHON
from pi_agent import Agent
agent = Agent(name="monitored")
@agent.on("tool_call")
def log_tool_call(event):
print(f"调用工具: {event.tool_name}({event.params})")
@agent.on("error")
def handle_error(event):
print(f"错误: {event.error}")
with open("error_log.txt", "a") as f:
f.write(f"{event.error}\n")
@agent.on("agent_response")
def log_response(event):
print(f"Token 消耗: {event.response.usage}")
(2) 类方式
PYTHON
from pi_agent import Agent, EventHandler
class MyHandler(EventHandler):
def on_tool_call(self, event):
print(f"工具: {event.tool_name}")
def on_tool_result(self, event):
print(f"结果: {event.result}")
def on_error(self, event):
print(f"错误: {event.error}")
agent = Agent(name="monitored", event_handler=MyHandler())
4. 自定义事件
(1) 定义事件
PYTHON
from pi_agent import Event
class DeployEvent(Event):
name = "deploy"
fields = ["environment", "status", "url"]
(2) 触发事件
PYTHON
agent.emit("deploy", {
"environment": "production",
"status": "success",
"url": "https://myapp.example.com"
})
(3) 监听自定义事件
PYTHON
@agent.on("deploy")
def on_deploy(event):
if event.status == "success":
send_notification(f"部署成功: {event.url}")
5. 命令注册
(1) 注册交互模式命令
PYTHON
from pi_agent import Agent
agent = Agent(name="custom_cmd")
@agent.command("/deploy", description="部署当前项目到指定环境")
def deploy_cmd(args: str):
env = args.strip() or "staging"
result = agent.run(f"将当前项目部署到 {env} 环境")
print(result)
@agent.command("/review", description="审查指定文件的代码")
def review_cmd(args: str):
filename = args.strip()
result = agent.run(skill="code_review", file=filename)
print(result)
@agent.command("/cost", description="显示当前会话的 token 消耗统计")
def cost_cmd(args: str):
usage = agent.session.get_usage()
print(f"Token 消耗: {usage.total_tokens}")
print(f"预估费用: {usage.estimated_cost:.2f}")
(2) 命令参数解析
PYTHON
@agent.command("/search", description="搜索代码")
def search_cmd(args: str):
parts = args.split(maxsplit=1)
if len(parts) < 2:
print("用法: /search <目录> <关键词>")
return
directory, keyword = parts
result = agent.run(f"在 {directory} 中搜索 {keyword}")
print(result)
6. 事件过滤器
(1) 按条件过滤
PYTHON
@agent.on("tool_call", filter=lambda e: e.tool_name == "shell")
def log_shell_calls(event):
print(f"Shell 命令: {event.params.get('cmd')}")
(2) 优先级排序
PYTHON
@agent.on("error", priority=10) # 高优先级
def critical_error(event):
send_alert(f"严重错误: {event.error}")
@agent.on("error", priority=1) # 低优先级
def log_error(event):
with open("errors.log", "a") as f:
f.write(f"{event.error}\n")
7. 事件重放
▶ 示例 1：调试事件链（难度⭐⭐）
PYTHON
from pi_agent import Agent
agent = Agent(name="debug_agent", debug=True)
# 记录所有事件
events = []
@agent.on("*")
def capture_all(event):
events.append(event)
# 执行任务
agent.run("分析 main.py 并生成测试")
# 重放事件链
for e in events:
print(f"[{e.timestamp}] {e.name}: {e.data}")
Alice 用事件重放功能追踪了 Agent 的完整执行链："原来它先读了文件，又搜索了最佳实践，然后才生成测试——这个推理过程一目了然。"
❓ 常见问题
Q 事件处理器能修改事件数据吗？
A 可以。在处理器中修改 event.data，后续处理器会看到修改后的数据。但建议只读取，不修改，避免副作用。
Q 事件处理是同步还是异步？
A 默认同步，按优先级顺序执行。异步处理器用
@agent.on("event", async_handler=True) 注册。Q 命令和斜杆命令有什么区别？
A 命令是用
@agent.command() 注册的自定义扩展，斜杆命令是交互模式的内置命令（如 /help、/exit）。两者格式一致但来源不同。📖 小节
- 事件驱动架构：触发源 → 事件总线 → 处理器
- 9 种内置事件覆盖 Agent 完整生命周期
- 两种监听方式：装饰器和 EventHandler 类
- 自定义事件和命令注册扩展交互能力
- 过滤器和优先级控制事件处理逻辑
📝 作业
- 基础题（难度⭐）：监听 tool_call 事件，记录所有工具调用到日志文件。
- 进阶题（难度⭐⭐）：创建一个自定义命令 /summarize，生成当前会话的摘要。
- 挑战题（难度⭐⭐⭐）：实现一个事件驱动的部署流水线：代码审查 → 测试 → 部署，每个阶段通过事件触发下一阶段。
