トリガー・アクション型エージェントのパターン
イベント検出 → 判断 → アクションという、自動化の中核ループを学びます。
「トリガー・アクション型エージェントのパターン」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン1/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
トリガー・アクションエージェントとは
トリガー・アクションエージェントはイベントを監視し、アクションで応答します。3つの段階からなるループは、イベントを検知 → LLMがアクションを判断 → アクションを実行です。
例：メールを受信 → 要約して返信、ファイルをアップロード → 検証して処理、毎日午前9時 → 日次レポートを生成。
トリガーの種類
トリガーは3つのカテゴリに分けられます。
- イベントベース：メールの到着やファイルのアップロード時にwebhookが発火します
- 時間ベース：cronスケジュールによって一定間隔でエージェントを実行します
- ポーリングベース：エージェントが新しいデータをAPIに繰り返し問い合わせます
適切なトリガーの種類を選ぶことが、エージェントの遅延とリソース使用量を左右します。
検知フェーズ
検知とは、イベントを受信または認識することです。webhookの場合、サーバーがPOSTリクエストを受信します。ポーリングの場合、エージェントがAPIにクエリを送り、最後に確認した状態と結果を比較します。
import json
def detect_new_email(current_emails, last_seen_id):
new_emails = [
e for e in current_emails
if e['id'] > last_seen_id
]
return new_emails
# Simulate detection
current = [{'id': 3, 'subject': 'Meeting'}, {'id': 4, 'subject': 'Invoice'}]
new = detect_new_email(current, last_seen_id=2)
print('New emails:', [e['subject'] for e in new])判断フェーズ
イベントを検知した後、エージェントはコンテキストをLLMに送り、どのアクションを実行するかを尋ねます。LLMはツールを選択するか、直接レスポンスを返します。
import openai
client = openai.OpenAI(api_key='sk-...')
def decide_action(event_description):
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[
{'role': 'system', 'content': 'You are an automation agent. Decide what action to take for the event.'},
{'role': 'user', 'content': f'Event: {event_description}'}
],
tools=[
{'type': 'function', 'function': {'name': 'send_reply', 'description': 'Reply to email', 'parameters': {'type': 'object', 'properties': {'message': {'type': 'string'}}, 'required': ['message']}}}
]
)
return response.choices[0].message
result = decide_action('New email: Invoice for $500 from supplier')
print(result)実行フェーズ
実行フェーズでは、選択されたアクションを実行します。APIの呼び出し、ファイルへの書き込み、メッセージの送信、別のワークフローの起動などを行います。必ずエラーを処理し、結果をログに記録してください。
import logging
import sys
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger('agent')
def execute_action(action_name, params):
try:
if action_name == 'send_reply':
# In real code, call Gmail API here
logger.info(f'Sending reply: {params["message"]}')
return {'status': 'success'}
elif action_name == 'create_task':
logger.info(f'Creating task: {params["title"]}')
return {'status': 'success'}
else:
raise ValueError(f'Unknown action: {action_name}')
except Exception as e:
logger.error(f'Action failed: {e}')
return {'status': 'error', 'message': str(e)}
if __name__ == '__main__':
result = execute_action('send_reply', {'message': 'Thanks for reaching out!'})
print('Result:', result)
ステートマシンモデル
ステートマシンは、自動化エージェントに適した強力なモデルです。状態には、IDLE、DETECTING、DECIDING、EXECUTING、ERRORなどがあります。状態の遷移は、イベントまたは条件によって発生します。
ステートマシンを使うと、エージェントの動作が予測しやすくなり、デバッグも容易になります。
from enum import Enum
class AgentState(Enum):
IDLE = 'idle'
DETECTING = 'detecting'
DECIDING = 'deciding'
EXECUTING = 'executing'
ERROR = 'error'
class AutomationAgent:
def __init__(self):
self.state = AgentState.IDLE
def transition(self, new_state):
print(f'State: {self.state.value} -> {new_state.value}')
self.state = new_state
def run_cycle(self, event=None):
self.transition(AgentState.DETECTING)
if event:
self.transition(AgentState.DECIDING)
self.transition(AgentState.EXECUTING)
self.transition(AgentState.IDLE)
agent = AutomationAgent()
agent.run_cycle(event={'type': 'email', 'subject': 'Test'})メール受信トリガーパターン
Gmailのプッシュ通知にはPub/Subを使用します。新しいメールが届くと、Googleがトピックに公開します。エージェントはwebhookを受信し、メールを取得して処理します。
from fastapi import FastAPI, Request
import base64, json
app = FastAPI()
@app.post('/gmail-push')
async def gmail_push(request: Request):
body = await request.json()
# Decode Pub/Sub message
message = body.get('message', {})
data = base64.b64decode(message.get('data', '')).decode('utf-8')
notification = json.loads(data)
email_address = notification.get('emailAddress')
history_id = notification.get('historyId')
print(f'New email for {email_address}, historyId: {history_id}')
# Fetch email details and run agent here
return {'status': 'ok'}ファイルアップロードトリガーパターン
S3のイベント通知やローカルファイルシステムのウォッチャーを使うと、ファイルが現れたときにエージェントを起動できます。watchdogライブラリは、新しいファイルが作成されていないかディレクトリを監視します。
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time
class UploadHandler(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
return
print(f'New file detected: {event.src_path}')
self.process_file(event.src_path)
def process_file(self, filepath):
# Run agent logic on new file
print(f'Processing: {filepath}')
observer = Observer()
handler = UploadHandler()
observer.schedule(handler, path='/tmp/uploads/', recursive=False)
observer.start()
try:
time.sleep(30) # Watch for 30 seconds
finally:
observer.stop()
observer.join()時間ベースのトリガーパターン
時間ベースのトリガーは、スケジュールに従ってエージェントを起動します。プロセス内のスケジューリングにはAPSchedulerを、プロセスレベルのスケジューリングにはシステムのcronジョブを使用します。
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
scheduler = BlockingScheduler()
def daily_report_agent():
print(f'Daily report running at {datetime.now()}')
# Fetch data, call LLM, send report
pass
def hourly_check_agent():
print(f'Hourly check at {datetime.now()}')
pass
# Run at 8am every day
scheduler.add_job(daily_report_agent, 'cron', hour=8, minute=0)
# Run every 30 minutes
scheduler.add_job(hourly_check_agent, 'interval', minutes=30)
print('Scheduler started')
scheduler.start()冪等なアクション実行
自動化エージェントは冪等でなければなりません。同じアクションを2回実行しても、重複した効果が発生しないようにします。冪等性キーと、実行前に確認するパターンを使用してください。
import hashlib
processed_events = set() # In production, use Redis or DB
def compute_event_id(event):
content = f"{event['type']}:{event['source_id']}:{event['timestamp']}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
def handle_event_idempotent(event):
event_id = compute_event_id(event)
if event_id in processed_events:
print(f'Skipping duplicate event: {event_id}')
return {'status': 'duplicate', 'event_id': event_id}
# Process event
print(f'Processing event: {event_id}')
processed_events.add(event_id)
return {'status': 'processed', 'event_id': event_id}
# Simulate duplicate event
event = {'type': 'email', 'source_id': 'abc123', 'timestamp': '2024-01-01T09:00:00'}
print(handle_event_idempotent(event))
print(handle_event_idempotent(event)) # Duplicate - skippedエラー状態とリカバリ
堅牢なエージェントは、障害を適切に処理します。実行に失敗した場合、エージェントは指数バックオフで再試行したり、人間にアラートを送ったり、手動確認用のデッドレターキューに移動したりできます。
import time
def execute_with_retry(action_fn, max_retries=3):
for attempt in range(max_retries):
try:
result = action_fn()
print(f'Success on attempt {attempt + 1}')
return result
except Exception as e:
wait = 2 ** attempt # Exponential backoff: 1s, 2s, 4s
print(f'Attempt {attempt + 1} failed: {e}. Retrying in {wait}s')
if attempt < max_retries - 1:
time.sleep(wait)
else:
print('All retries exhausted. Moving to dead-letter queue.')
raise
# Example usage
call_count = [0]
def flaky_action():
call_count[0] += 1
if call_count[0] < 3:
raise ConnectionError('Service unavailable')
return 'Done'
execute_with_retry(flaky_action)理解度チェック：トリガー・アクションパターン
トリガー・アクションエージェントのパターンについて、理解度を確認しましょう。
全体の組み立て
完全なトリガー・アクションエージェントは、トリガーソース（メール、ファイル、タイマー）、検知レイヤー、LLMベースの意思決定、冪等な実行、再試行ロジック、状態追跡をすべて組み合わせます。
まずはシンプルに、トリガーの種類を1つ、アクションを1つにします。エージェントの動作に自信がついたら、少しずつ複雑さを追加してください。
よくある質問
「トリガー・アクション型エージェントのパターン」レッスンは無料ですか？
はい。「トリガー・アクション型エージェントのパターン」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「トリガー・アクション型エージェントのパターン」で何を学びますか？
イベント検出 → 判断 → アクションという、自動化の中核ループを学びます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン1/4です。
「トリガー・アクション型エージェントのパターン」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- トリガー・アクション型エージェントのパターン
- エージェントとWebhookの接続
- スケジューリングとCronベースのエージェント
- 複数アプリの自動化パイプラインを構築する