선제적 알림 및 경고 시스템
요청받지 않아도 중요한 정보를 알려 주는 에이전트를 구축합니다.
선제적 알림 및 경고 시스템은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 2번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 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와 무료 요금제를 제공합니다. 한 명의 사용자에게 알림을 보내야 하는 개인 에이전트에 적합합니다.
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으로 보내십시오. 채널 메시지와 달리 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 또는 이메일로 알림을 보냅니다. 핵심 설계 요소는 알림 피로를 방지하는 대기 시간이 포함된 임계값 기반 규칙, 개인화를 위한 사용자 설정 필터링, 긴급도에 따른 다중 채널 전달, 방해 금지 시간대 지원, 중요도가 낮은 알림을 모아 보내는 일일 요약입니다.
자주 묻는 질문
“선제적 알림 및 경고 시스템” 강의는 무료인가요?
네 — “선제적 알림 및 경고 시스템” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
“선제적 알림 및 경고 시스템”에서 뭘 배우나요?
요청받지 않아도 중요한 정보를 알려 주는 에이전트를 구축합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Agents을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 2번째 강의입니다.
“선제적 알림 및 경고 시스템” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 상시 실행 에이전트 설계 패턴
- 선제적 알림 및 경고 시스템
- 세션 간 맥락 유지
- 일일 브리핑 에이전트 구축