主动通知与提醒系统
无需用户提出请求即可主动呈现重要信息的智能体
主动通知与提醒系统 是 CoddyKit 上的免费 AI Agents 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 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 私信通知
请将主动警报作为 Slack 直接消息发送。与频道消息不同，私信可以确保用户即使不在某个特定频道中，也能收到通知。
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 私信。
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 私信或电子邮件发送警报。关键设计要素包括：带冷却时间的基于阈值的规则，以避免通知疲劳；用于个性化的用户偏好筛选；根据紧急程度进行多渠道路由；支持勿扰时段；以及用于低优先级累积警报的每日摘要。
常见问题解答
「主动通知与提醒系统」课时是免费的吗？
是的 — 「主动通知与提醒系统」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「主动通知与提醒系统」这节课中我会学到什么？
无需用户提出请求即可主动呈现重要信息的智能体 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。
「主动通知与提醒系统」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗？
能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。