毎日のブリーフィングエージェントを構築する
ニュース、カレンダー、メールの要約を朝のダイジェストとして自動配信します。
「毎日のブリーフィングエージェントを構築する」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
デイリーブリーフィングエージェントとは
デイリーブリーフィングエージェントは、毎朝決められた時刻に実行され、複数の情報源（カレンダー、メール、ニュース）から情報を収集し、LLMで統合して、パーソナライズされたサマリーを配信します。これは、スケジュール実行されるパーソナルエージェントの典型的な例です。
ブリーフィングのアーキテクチャ
ブリーフィングのパイプラインは、次の5段階で構成されます：
- トリガー：午前8時のcronジョブがエージェントを起動します
- 取得：カレンダーの予定、未読メール、ニュースの見出しを並列で収集します
- 統合：LLMがすべてのデータから一貫性のあるブリーフィングを作成します
- パーソナライズ：ユーザー設定に合わせてトーンと内容を調整します
- 配信：メールまたはSlack DMで送信します
from dataclasses import dataclass, field
from typing import List, Dict, Any
@dataclass
class DailyBriefingData:
date: str
calendar_events: List[Dict] = field(default_factory=list)
unread_emails: List[Dict] = field(default_factory=list)
news_headlines: List[Dict] = field(default_factory=list)
weather: Dict = field(default_factory=dict)
briefing_text: str = ''
delivery_status: str = 'pending'
print('Daily briefing data structure defined')APSchedulerによるスケジューリング
ブリーフィングが平日の午前8時に実行されるようスケジュールします。SQLiteによる永続化機能を備えたAPSchedulerを使用すると、サーバーを再起動してもジョブを維持できます。
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.triggers.cron import CronTrigger
import asyncio
scheduler = BackgroundScheduler(
jobstores={'default': SQLAlchemyJobStore(url='sqlite:///jobs.db')},
timezone='America/New_York'
)
def run_briefing_job():
print('Daily briefing job triggered')
asyncio.run(generate_and_deliver_briefing('user-42'))
# 8am Monday-Friday
scheduler.add_job(
run_briefing_job,
CronTrigger(day_of_week='mon-fri', hour=8, minute=0),
id='daily_briefing_user_42',
replace_existing=True
)
scheduler.start()
print('Briefing scheduler started: runs Mon-Fri at 8am ET')カレンダーイベントの取得
Google Calendar APIを使用して、当日のカレンダーイベントを取得します。ブリーフィングに予定を含めるには、エージェントがどの会議の予定が入っているかを把握する必要があります。
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
from datetime import datetime, timedelta
import pytz
def get_todays_calendar_events(token_path: str = 'token.json', timezone: str = 'America/New_York') -> list:
creds = Credentials.from_authorized_user_file(token_path)
service = build('calendar', 'v3', credentials=creds)
tz = pytz.timezone(timezone)
now = datetime.now(tz)
start_of_day = now.replace(hour=0, minute=0, second=0, microsecond=0)
end_of_day = now.replace(hour=23, minute=59, second=59)
result = service.events().list(
calendarId='primary',
timeMin=start_of_day.isoformat(),
timeMax=end_of_day.isoformat(),
singleEvents=True,
orderBy='startTime'
).execute()
events = []
for item in result.get('items', []):
start = item.get('start', {}).get('dateTime', item.get('start', {}).get('date'))
events.append({
'title': item.get('summary', 'Untitled'),
'start': start,
'attendees': len(item.get('attendees', [])),
'location': item.get('location', '')
})
return events未読メールの取得
最新の未読メールを取得します。ブリーフィングエージェントに必要なのは、LLMのコンテキストを管理しやすくするため、メール本文全体ではなく、タイトルと送信者などのサマリーだけです。
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
import base64
def get_email_summaries(token_path: str = 'token.json', max_results: int = 10) -> list:
creds = Credentials.from_authorized_user_file(token_path)
service = build('gmail', 'v1', credentials=creds)
results = service.users().messages().list(
userId='me',
q='is:unread newer_than:1d',
maxResults=max_results
).execute()
summaries = []
for msg in results.get('messages', []):
detail = service.users().messages().get(
userId='me',
id=msg['id'],
format='metadata',
metadataHeaders=['From', 'Subject', 'Date']
).execute()
headers = {h['name']: h['value'] for h in detail['payload']['headers']}
summaries.append({
'from': headers.get('From', ''),
'subject': headers.get('Subject', 'No Subject'),
'date': headers.get('Date', '')
})
return summariesニュース見出しの取得
ニュースAPIを使用して、関連するニュースの見出しを取得します。パーソナライズされたニュースセクションを作成するため、ユーザーが設定した関心のあるトピックでフィルタリングします。
import httpx
import os
NEWS_API_KEY = os.environ.get('NEWS_API_KEY', 'your-key')
async def get_news_headlines(topics: list, max_per_topic: int = 3) -> list:
headlines = []
async with httpx.AsyncClient() as client:
for topic in topics:
try:
response = await client.get(
'https://newsapi.org/v2/top-headlines',
params={
'q': topic,
'language': 'en',
'pageSize': max_per_topic,
'apiKey': NEWS_API_KEY
},
timeout=10.0
)
data = response.json()
for article in data.get('articles', []):
headlines.append({
'topic': topic,
'title': article.get('title', ''),
'source': article.get('source', {}).get('name', ''),
'url': article.get('url', '')
})
except Exception as e:
print(f'News fetch failed for {topic}: {e}')
return headlines[:10] # Cap totalデータの並列取得
カレンダーイベント、メール、ニュース、天気をすべて同時に取得します。これらは独立しているため、並列取得によって取得時間全体を約4秒から約1秒に短縮できます。
import asyncio
from datetime import datetime
async def fetch_all_briefing_data(user_prefs: dict) -> 'DailyBriefingData':
data = DailyBriefingData(date=datetime.now().strftime('%A, %B %d, %Y'))
async def safe_get_calendar():
try:
return get_todays_calendar_events()
except Exception as e:
print(f'Calendar fetch failed: {e}')
return []
async def safe_get_emails():
try:
return get_email_summaries()
except Exception as e:
print(f'Email fetch failed: {e}')
return []
async def safe_get_news():
topics = user_prefs.get('news_topics', ['technology', 'business'])
return await get_news_headlines(topics)
# Run all fetches in parallel
calendar_events, emails, news = await asyncio.gather(
safe_get_calendar(),
safe_get_emails(),
safe_get_news(),
return_exceptions=False
)
data.calendar_events = calendar_events
data.unread_emails = emails
data.news_headlines = news
return dataLLMによる統合
取得したすべてのデータをLLMに渡し、自然でパーソナライズされたブリーフィングを作成するよう依頼します。プロンプトの構成が重要です。データを明確に整理することで、LLMが構造化されたサマリーを生成しやすくなります。
import openai
import json
client = openai.OpenAI(api_key='sk-...')
def synthesize_briefing(data: 'DailyBriefingData', user_prefs: dict) -> str:
user_name = user_prefs.get('name', 'there')
tone = user_prefs.get('briefing_tone', 'professional') # 'casual', 'professional', 'concise'
calendar_text = json.dumps(data.calendar_events[:5], indent=2) if data.calendar_events else 'No meetings today.'
email_text = json.dumps(data.unread_emails[:5], indent=2) if data.unread_emails else 'No unread emails.'
news_text = '\n'.join([f'- [{h["topic"]}] {h["title"]} ({h["source"]})' for h in data.news_headlines[:6]])
prompt = (
f'Create a {tone} morning briefing for {user_name} for {data.date}.\n\n'
f'Today\'s calendar:\n{calendar_text}\n\n'
f'Unread emails:\n{email_text}\n\n'
f'News headlines:\n{news_text}\n\n'
'Write a concise, actionable briefing in 3-4 paragraphs. '
'Start with today\'s schedule, then email highlights, then relevant news. '
'End with one suggested priority for the day.'
)
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
temperature=0.3
)
return response.choices[0].message.contentメールによる配信
ブリーフィングをHTMLメールとしてフォーマットして送信します。多くの人は朝のサマリーをモバイル端末で読むため、適切なHTMLメールのフォーマットによってブリーフィングが読みやすくなります。
def format_briefing_email(briefing_text: str, data: 'DailyBriefingData') -> str:
paragraphs = briefing_text.split('\n\n')
html_paragraphs = ''.join([f'<p>{p}</p>' for p in paragraphs if p.strip()])
calendar_items = ''.join([
f'<li><strong>{e["start"][:16]}</strong> - {e["title"]}</li>'
for e in data.calendar_events[:5]
])
return f'''
<html><body style="font-family: Arial, sans-serif; max-width: 600px; margin: 0 auto;">
<h2 style="color: #333;">Morning Briefing — {data.date}</h2>
{html_paragraphs}
{'<h3>Today\'s Calendar</h3><ul>' + calendar_items + '</ul>' if calendar_items else ''}
<hr>
<p style="color: #666; font-size: 12px;">Generated by your Personal AI Assistant</p>
</body></html>
'''
def deliver_briefing_email(data: 'DailyBriefingData', user_prefs: dict):
html = format_briefing_email(data.briefing_text, data)
to_email = user_prefs.get('email', '')
send_email_alert(
to_email=to_email,
subject=f'Morning Briefing - {data.date}',
html_body=html
)
print('Email delivery function defined')Slack DMによる配信
Slackを常用するユーザーには、ブリーフィングをフォーマット済みのSlack DMとして配信するほうが、メールより好まれることが多くあります。リッチなフォーマットにはBlock Kitを使用します。
from slack_sdk import WebClient
import os
slack_client = WebClient(token=os.environ.get('SLACK_BOT_TOKEN', 'xoxb-...'))
def deliver_briefing_slack(data: 'DailyBriefingData', user_prefs: dict):
slack_user_id = user_prefs.get('slack_user_id')
if not slack_user_id:
print('No Slack user ID configured')
return
# Split briefing into sections for blocks
paragraphs = [p for p in data.briefing_text.split('\n\n') if p.strip()]
blocks = [
{'type': 'header', 'text': {'type': 'plain_text', 'text': f'Morning Briefing - {data.date}'}},
{'type': 'divider'}
]
for para in paragraphs[:4]: # Max 4 paragraphs
blocks.append({
'type': 'section',
'text': {'type': 'mrkdwn', 'text': para}
})
if data.calendar_events:
event_list = '\n'.join([f'• {e["start"][:16]} - {e["title"]}' for e in data.calendar_events[:3]])
blocks.append({'type': 'section', 'text': {'type': 'mrkdwn', 'text': f'*Meetings*\n{event_list}'}})
slack_client.chat_postMessage(
channel=slack_user_id,
text=f'Morning Briefing - {data.date}',
blocks=blocks
)
print(f'Briefing delivered to Slack user {slack_user_id}')メインブリーフィングオーケストレーター
オーケストレーターはすべての段階をつなぎます。データを並列で取得し、LLMで統合し、ユーザーが希望するチャンネルに配信し、監視用に結果を記録します。
import asyncio
from datetime import datetime
async def generate_and_deliver_briefing(user_id: str):
print(f'Generating briefing for {user_id} at {datetime.now()}')
# Load user preferences
conn = init_db()
user_prefs = load_user_preferences(conn, user_id)
# Step 1: Fetch all data in parallel
data = await fetch_all_briefing_data(user_prefs)
if not data.calendar_events and not data.unread_emails and not data.news_headlines:
print('No data fetched, skipping briefing')
return
# Step 2: Synthesize with LLM
data.briefing_text = synthesize_briefing(data, user_prefs)
# Step 3: Deliver
delivery_channel = user_prefs.get('delivery_channel', 'email')
if delivery_channel == 'slack':
deliver_briefing_slack(data, user_prefs)
else:
deliver_briefing_email(data, user_prefs)
data.delivery_status = 'delivered'
# Step 4: Log
print(f'Briefing delivered via {delivery_channel}')
print(f'Data: {len(data.calendar_events)} events, {len(data.unread_emails)} emails, {len(data.news_headlines)} headlines')
print('Daily briefing orchestrator defined')理解度チェック：デイリーブリーフィングエージェント
デイリーブリーフィングエージェントの構築について理解度を確認します。
デイリーブリーフィングエージェントのまとめ
完全なデイリーブリーフィングエージェントは、平日の午前8時にcron実行するAPScheduler、カレンダー・メール・ニュースAPIからの並列データ取得、パーソナライズされた一貫性のあるブリーフィングを作成するLLMによる統合、ユーザー設定に基づくパーソナライズ、メールまたはSlack DMによるマルチチャンネル配信を組み合わせます。これは、コースで扱ったすべての概念を示す、典型的な常時稼働型パーソナルエージェントのパターンです。
よくある質問
「毎日のブリーフィングエージェントを構築する」レッスンは無料ですか？
はい。「毎日のブリーフィングエージェントを構築する」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「毎日のブリーフィングエージェントを構築する」で何を学びますか？
ニュース、カレンダー、メールの要約を朝のダイジェストとして自動配信します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「毎日のブリーフィングエージェントを構築する」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- 常時稼働エージェントの設計パターン
- プロアクティブな通知・アラートシステム
- セッションをまたぐコンテキストの永続化
- 毎日のブリーフィングエージェントを構築する