Pi Agent: Event System & Command Registration
最終更新：2026-08-31
--- title: "イベントシステムとコマンド登録" description: "Pi Agentのイベント駆動アーキテクチャとコマンド登録メカニズムをマスターします。" order: 16 lang: ja
イベントシステムは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"Tool called: {event.tool_name}({event.params})")
@agent.on("error")
def handle_error(event):
print(f"Error: {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 usage: {event.response.usage}")
(2) クラススタイル
PYTHON
from pi_agent import Agent, EventHandler
class MyHandler(EventHandler):
def on_tool_call(self, event):
print(f"Tool: {event.tool_name}")
def on_tool_result(self, event):
print(f"Result: {event.result}")
def on_error(self, event):
print(f"Error: {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"Deploy succeeded: {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="現在のセッションのトークン使用統計を表示")
def cost_cmd(args: str):
usage = agent.session.get_usage()
print(f"トークン使用量: {usage.total_tokens}")
print(f"推定コスト: ${usage.estimated_cost:.4f}")
6. イベントフィルター
(1) 条件フィルタリング
PYTHON
@agent.on("tool_call", filter=lambda e: e.tool_name == "shell")
def log_shell_calls(event):
print(f"Shell command: {event.params.get('cmd')}")
(2) 優先度
PYTHON
@agent.on("error", priority=10)
def critical_error(event):
send_alert(f"Critical error: {event.error}")
@agent.on("error", priority=1)
def log_error(event):
with open("errors.log", "a") as f:
f.write(f"{event.error}\n")
❓ よくある質問
Q イベントハンドラーはイベントデータを変更できますか？
A はい、ただし副作用を避けるため読み取りのみを推奨します。変更されたデータは後続のハンドラーに見えます。
Q 同期と非同期のイベント処理？
A デフォルトは同期で、優先度順に実行されます。非同期ハンドラーには
async_handler=True を使用してください。Q コマンドとスラッシュコマンドの違いは？
A コマンドは
@agent.command() によるカスタム拡張です。スラッシュコマンドはインタラクティブモードの組み込みコマンド（/help、/exit）です。同じ形式ですが、由来が異なります。📖 まとめ
- イベント駆動：トリガー→イベントバス→ハンドラー
- 9つの組み込みイベントがAgentの完全なライフサイクルをカバー
- 2つのリスニングスタイル：デコレータとEventHandlerクラス
- カスタムイベントとコマンド登録でインタラクティブ機能を拡張
- フィルターと優先度でイベント処理を制御
📝 練習問題
- 基礎（難易度：⭐）: tool_callイベントをリスニングし、すべてのツール呼び出しをファイルに記録してください。
- 中級（難易度：⭐⭐）: セッションサマリーを生成する /summarize コマンドを作成してください。
- 上級（難易度：⭐⭐⭐）: イベント駆動のデプロイパイプラインを実装してください：コードレビュー→テスト→デプロイ、各ステージがイベントでトリガーされる。
