AI Agents · レッスン

プロアクティブな通知・アラートシステム

Q: 「プロアクティブな通知・アラートシステム」レッスンは無料ですか？

はい。「プロアクティブな通知・アラートシステム」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。

Q: 「プロアクティブな通知・アラートシステム」で何を学びますか？

要求されなくても重要な情報を提示するエージェントを構築します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

Q: AI Agentsを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。

Q: 「プロアクティブな通知・アラートシステム」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

Q: このAI Agentsレッスンでコードを書いて実行できますか？

はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

要求されなくても重要な情報を提示するエージェントを構築します。

「プロアクティブな通知・アラートシステム」はCoddyKit上の無料AI Agentsレッスンです。これはレッスン2/4です。下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。

プロアクティブエージェントとリアクティブエージェント

ほとんどのエージェントはリアクティブで、リクエストに応答します。一方、プロアクティブなエージェントは状況を監視し、依頼されていなくても注目すべきことが起きたときにユーザーへ通知します。これは、パーソナルアシスタントを持つことに近い仕組みです。

バックグラウンドポーリングループ

最もシンプルなプロアクティブパターンは、N分ごとにチェックを実行し、条件を満たした場合にアラートを送信するバックグラウンドループです。ブロッキングを避けるには、スレッドまたは非同期ループを使用します。

import asyncio
import httpx
from datetime import datetime

async def poll_price(ticker: str, alert_threshold: float) -> None:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f'https://api.finance.example.com/quote/{ticker}',
            headers={'Authorization': 'Bearer your-api-key'}
        )
        data = response.json()
        price = float(data.get('price', 0))
        
        if price < alert_threshold:
            await send_push_notification(
                title=f'{ticker} Price Alert',
                message=f'{ticker} is now ${price:.2f}, below your threshold of ${alert_threshold:.2f}'
            )

async def price_alert_loop(ticker: str, threshold: float, interval_seconds: int = 300):
    print(f'Monitoring {ticker} every {interval_seconds}s, alert below ${threshold}')
    while True:
        try:
            await poll_price(ticker, threshold)
        except Exception as e:
            print(f'Polling error: {e}')
        await asyncio.sleep(interval_seconds)

print('Price alert loop defined')

しきい値ベースのアラート

しきい値アラートは、メトリクスが定義した境界を超えたときに発生します。適切なしきい値には、絶対値（価格 < $100）、変化率（5%超の下落）、過去の平均値との比較などがあります。

from dataclasses import dataclass
from typing import Optional, Callable

@dataclass
class AlertRule:
    name: str
    check_fn: Callable  # Returns a float value
    condition: str      # 'lt', 'gt', 'lte', 'gte'
    threshold: float
    cooldown_minutes: int = 60  # Don't re-alert for this long
    last_alerted: Optional[float] = None

def evaluate_rule(rule: AlertRule) -> Optional[str]:
    import time
    
    # Respect cooldown
    if rule.last_alerted:
        elapsed = time.time() - rule.last_alerted
        if elapsed < rule.cooldown_minutes * 60:
            return None
    
    current_value = rule.check_fn()
    
    triggered = (
        (rule.condition == 'lt' and current_value < rule.threshold) or
        (rule.condition == 'gt' and current_value > rule.threshold) or
        (rule.condition == 'lte' and current_value <= rule.threshold) or
        (rule.condition == 'gte' and current_value >= rule.threshold)
    )
    
    if triggered:
        rule.last_alerted = time.time()
        return f'Alert: {rule.name} = {current_value:.2f} ({rule.condition} {rule.threshold})'
    return None

# Example rule
rule = AlertRule(
    name='CPU Usage',
    check_fn=lambda: 85.0,  # In reality: psutil.cpu_percent()
    condition='gt',
    threshold=80.0,
    cooldown_minutes=30
)
print(evaluate_rule(rule))

条件に一致する新着メール

プロアクティブなメールエージェントは受信トレイを監視し、特定の条件に一致するメッセージが届いたときに通知します。条件には、VIP送信者からのメール、キーワードを含むメール、重要度フラグが付いたメールなどがあります。

from dataclasses import dataclass
from typing import List
import re

@dataclass
class EmailAlertCriteria:
    from_domains: List[str] = None
    from_emails: List[str] = None
    subject_keywords: List[str] = None
    body_keywords: List[str] = None
    min_importance: str = None  # 'high', 'medium'

def matches_criteria(email: dict, criteria: EmailAlertCriteria) -> bool:
    sender = email.get('from', '').lower()
    subject = email.get('subject', '').lower()
    body = email.get('body', '').lower()
    
    if criteria.from_domains:
        domain_match = any(domain.lower() in sender for domain in criteria.from_domains)
        if not domain_match:
            return False
    
    if criteria.from_emails:
        email_match = any(e.lower() in sender for e in criteria.from_emails)
        if not email_match:
            return False
    
    if criteria.subject_keywords:
        keyword_match = any(kw.lower() in subject for kw in criteria.subject_keywords)
        if not keyword_match:
            return False
    
    return True

criteria = EmailAlertCriteria(
    from_domains=['@important-client.com', '@boss.company.com'],
    subject_keywords=['urgent', 'action required', 'ASAP']
)

email = {'from': 'john@important-client.com', 'subject': 'Urgent: Contract Issue'}
print('Matches:', matches_criteria(email, criteria))

Pushover通知

PushoverはiOSとAndroidにプッシュ通知を配信します。シンプルなREST APIと無料プランを備えています。1人のユーザーに通知する必要がある個人向けエージェントに適しています。

import httpx
import os

PUSHOVER_TOKEN = os.environ.get('PUSHOVER_APP_TOKEN', 'your-app-token')
PUSHOVER_USER = os.environ.get('PUSHOVER_USER_KEY', 'your-user-key')

async def send_pushover(title: str, message: str, priority: int = 0, url: str = None) -> bool:
    '''
    priority: -2 (lowest) to 2 (emergency with acknowledgment)
    0 = normal, 1 = high priority (bypass quiet hours)
    '''
    async with httpx.AsyncClient() as client:
        data = {
            'token': PUSHOVER_TOKEN,
            'user': PUSHOVER_USER,
            'title': title,
            'message': message,
            'priority': priority
        }
        if url:
            data['url'] = url
        
        response = await client.post('https://api.pushover.net/1/messages.json', data=data)
        result = response.json()
        
        if result.get('status') == 1:
            print(f'Push sent: {title}')
            return True
        else:
            print(f'Push failed: {result.get("errors")}')
            return False

async def send_push_notification(title: str, message: str):
    await send_pushover(title, message)

Slack DM通知

プロアクティブなアラートをSlackのダイレクトメッセージとして送信します。チャンネルメッセージとは異なり、DMならユーザーが特定のチャンネルにいなくても確実に通知できます。

from slack_sdk import WebClient
import os

slack_client = WebClient(token=os.environ.get('SLACK_BOT_TOKEN', 'xoxb-...'))

def send_slack_dm(user_id: str, title: str, message: str, urgency: str = 'normal') -> bool:
    # Build blocks for rich formatting
    blocks = [
        {
            'type': 'section',
            'text': {
                'type': 'mrkdwn',
                'text': f'*{title}*\n{message}'
            }
        }
    ]
    
    if urgency == 'high':
        # Add urgent emoji prefix
        blocks[0]['text']['text'] = '🚨 ' + blocks[0]['text']['text']
    
    try:
        response = slack_client.chat_postMessage(
            channel=user_id,  # DM: channel = user_id
            text=f'{title}: {message}',  # Fallback text
            blocks=blocks
        )
        return True
    except Exception as e:
        print(f'Slack DM failed: {e}')
        return False

print('Slack DM function defined')
print('Usage: send_slack_dm("U0123ABCD", "Price Alert", "AAPL dropped below $150")')

メール通知

メールは緊急性のないアラートに適した信頼性の高い手段です。エージェントからプログラムでメールを送信するには、PythonのsmtplibまたはSendGrid APIを使用します。

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import os

def send_email_alert(to_email: str, subject: str, html_body: str) -> bool:
    smtp_host = os.environ.get('SMTP_HOST', 'smtp.gmail.com')
    smtp_port = int(os.environ.get('SMTP_PORT', '587'))
    smtp_user = os.environ.get('SMTP_USER', '')
    smtp_pass = os.environ.get('SMTP_PASS', '')
    
    try:
        msg = MIMEMultipart('alternative')
        msg['Subject'] = subject
        msg['From'] = smtp_user
        msg['To'] = to_email
        
        html_part = MIMEText(html_body, 'html')
        msg.attach(html_part)
        
        with smtplib.SMTP(smtp_host, smtp_port) as server:
            server.starttls()
            server.login(smtp_user, smtp_pass)
            server.sendmail(smtp_user, to_email, msg.as_string())
        
        print(f'Email sent to {to_email}: {subject}')
        return True
    except Exception as e:
        print(f'Email failed: {e}')
        return False

html = '<p>Your agent detected a price drop: <strong>AAPL is now $147.50</strong></p>'
print('Email alert function defined')

マルチチャンネルアラートルーター

緊急度に応じて、アラートを適切なチャンネルへ振り分けます。重大なアラートはプッシュ通知、情報提供を目的とするアラートはメール、毎日のサマリーはSlack DMに送信します。

import asyncio
from enum import Enum

class AlertLevel(Enum):
    INFO = 'info'        # Email/Daily summary
    WARNING = 'warning'  # Slack DM
    CRITICAL = 'critical'  # Push notification immediately

async def route_alert(title: str, message: str, level: AlertLevel, user_config: dict):
    user_id = user_config.get('user_id')
    email = user_config.get('email')
    push_enabled = user_config.get('push_enabled', True)
    
    if level == AlertLevel.CRITICAL and push_enabled:
        success = await send_pushover(title, message, priority=1)
        if not success:
            # Fallback to Slack DM
            slack_user = user_config.get('slack_user_id')
            if slack_user:
                send_slack_dm(slack_user, title, message, urgency='high')
    
    elif level == AlertLevel.WARNING:
        slack_user = user_config.get('slack_user_id')
        if slack_user:
            send_slack_dm(slack_user, title, message)
    
    else:  # INFO
        if email:
            send_email_alert(email, title, f'<p>{message}</p>')

user_config = {
    'user_id': 'user-42',
    'email': 'user@example.com',
    'slack_user_id': 'U0123ABCD',
    'push_enabled': True
}

asyncio.run(route_alert('Price Alert', 'AAPL dropped to $147', AlertLevel.WARNING, user_config))

アラートの重複排除とクールダウン

重複排除を行わないと、何時間も続く条件によって何百件ものアラートが送信されます。クールダウンを設定すると、指定した期間内に同じ条件について再度アラートを送信することを防げます。

import time
import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def should_send_alert(alert_key: str, cooldown_seconds: int = 3600) -> bool:
    redis_key = f'alert:cooldown:{alert_key}'
    if r.exists(redis_key):
        ttl = r.ttl(redis_key)
        print(f'Alert {alert_key} on cooldown. {ttl}s remaining')
        return False
    return True

def record_alert_sent(alert_key: str, cooldown_seconds: int = 3600):
    redis_key = f'alert:cooldown:{alert_key}'
    r.setex(redis_key, cooldown_seconds, '1')

def maybe_send_alert(alert_type: str, metric_value: float, user_id: str, cooldown_hours: int = 1):
    # Create unique key per alert type per user
    alert_key = f'{user_id}:{alert_type}'
    
    if not should_send_alert(alert_key, cooldown_seconds=cooldown_hours * 3600):
        return False
    
    # Send the alert
    print(f'Sending alert: {alert_type} = {metric_value} for user {user_id}')
    record_alert_sent(alert_key, cooldown_seconds=cooldown_hours * 3600)
    return True

maybe_send_alert('price_drop_AAPL', 147.50, 'user-42')
maybe_send_alert('price_drop_AAPL', 146.00, 'user-42')  # Blocked by cooldown

ユーザー設定に基づくフィルタリング

すべてのユーザーがすべてのアラートを必要としているわけではありません。ユーザーが受け取りたいアラートの種類、しきい値、配信チャンネルを設定として保存します。エージェントは送信前に設定を確認します。

from dataclasses import dataclass, field
from typing import Dict, List

@dataclass
class UserAlertPreferences:
    user_id: str
    enabled_channels: List[str] = field(default_factory=lambda: ['push'])
    alert_rules: Dict[str, dict] = field(default_factory=dict)
    quiet_hours_start: int = 22  # 10pm
    quiet_hours_end: int = 8    # 8am

def is_quiet_hours(prefs: UserAlertPreferences) -> bool:
    from datetime import datetime
    hour = datetime.now().hour
    start = prefs.quiet_hours_start
    end = prefs.quiet_hours_end
    if start > end:  # Spans midnight
        return hour >= start or hour < end
    return start <= hour < end

def should_alert_user(prefs: UserAlertPreferences, alert_type: str, value: float) -> bool:
    rule = prefs.alert_rules.get(alert_type)
    if not rule:
        return False  # User hasn't set up this alert type
    
    threshold = rule.get('threshold')
    condition = rule.get('condition', 'lt')
    urgent = rule.get('urgent', False)
    
    if is_quiet_hours(prefs) and not urgent:
        print(f'Suppressing non-urgent alert during quiet hours')
        return False
    
    return (
        (condition == 'lt' and value < threshold) or
        (condition == 'gt' and value > threshold)
    )

prefs = UserAlertPreferences(
    user_id='user-42',
    alert_rules={'price_drop': {'threshold': 150.0, 'condition': 'lt'}}
)
print('Should alert:', should_alert_user(prefs, 'price_drop', 147.50))

アラートのダイジェストへの集約

イベントごとに個別のアラートを送信する代わりに、アラートを蓄積してダイジェストとして送信します。監視対象の条件が多いユーザーに対する通知疲れを軽減できます。

import redis
import json
from datetime import datetime

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def add_to_digest(user_id: str, alert: dict):
    key = f'digest:{user_id}:{datetime.now().strftime("%Y%m%d")}'
    r.rpush(key, json.dumps(alert))
    r.expire(key, 86400 * 2)  # Keep for 2 days

def send_and_clear_digest(user_id: str) -> int:
    key = f'digest:{user_id}:{datetime.now().strftime("%Y%m%d")}'
    raw_alerts = r.lrange(key, 0, -1)
    if not raw_alerts:
        print(f'No alerts for user {user_id} today')
        return 0
    
    alerts = [json.loads(a) for a in raw_alerts]
    digest_text = f'Daily Digest ({len(alerts)} alerts):\n'
    digest_text += '\n'.join([f'- {a["title"]}: {a["message"]}' for a in alerts])
    
    # Send as single notification
    print(f'Sending digest to {user_id}:\n{digest_text}')
    # send_email_alert(user_email, 'Daily Alert Digest', digest_text)
    
    r.delete(key)
    return len(alerts)

add_to_digest('user-42', {'title': 'AAPL Alert', 'message': 'Price at $147.50'})
add_to_digest('user-42', {'title': 'GOOG Alert', 'message': 'Price at $175.00'})
send_and_clear_digest('user-42')

理解度チェック：プロアクティブ通知

プロアクティブな通知とアラートシステムについて理解度を確認します。

プロアクティブ通知のまとめ

プロアクティブエージェントは、バックグラウンドポーリングループで状況を監視し、プッシュ通知（Pushover）、Slack DM、メールでアラートを送信します。主な設計要素は、通知疲れを防ぐクールダウン付きのしきい値ベースのルール、パーソナライズのためのユーザー設定フィルタリング、緊急度に基づくマルチチャンネル振り分け、通知を控える時間帯への対応、優先度の低いアラートをまとめる毎日のダイジェストです。

よくある質問