はじめに
AWSの新機能リリースやアップデート情報、追いきれていませんか?
特にAmazon Connectのような特定サービスに注目している場合、関連する更新だけを効率的にキャッチアップしたいですよね。そこで、AWS公式の更新情報を自動収集し、AIで日本語要約してSlackに通知するシステムを作りました。
本記事では、GitHub ActionsとOpenAI APIを活用した自動通知システムの実装方法を詳しく解説します。
🎯 作ったもの
システム概要
- AWS公式RSSフィードから最新情報を自動収集
- OpenAI APIで英語記事を日本語3行要約に変換
- 毎週月曜日9時にSlackへ自動通知
- 重複防止機能で同じ記事の再通知を防止
実際の通知例
📢 Amazon Connect 週次更新(1件/一次情報リンク)
📅 2025年1月20日 | AI要約: ON
----------------------------------------
1. Amazon Connect launches new agent workspace
・エージェント向け新ワークスペース機能を提供開始
・複数チャネルの統合管理とリアルタイム分析が可能に
・生産性向上と顧客満足度改善への寄与が期待される
----------------------------------------
🏗️ システムアーキテクチャ
📦 必要な技術スタック
- Python 3.10: メイン処理言語
- GitHub Actions: 定期実行の自動化
- OpenAI API: GPT-4による要約生成
- Slack Incoming Webhooks: メッセージ投稿
🚀 実装方法
1. プロジェクト構成
aws-notifier/
├── .github/
│ └── workflows/
│ └── notify.yml # GitHub Actions定義
├── scripts/
│ ├── notify.py # メイン処理
│ ├── ai_providers.py # AI要約処理
│ └── config.py # 設定管理
├── data/
│ └── seen_links.json # 通知済みリンク履歴
├── requirements.txt # 依存関係
└── README.md
2. メイン処理の実装(notify.py)
import feedparser
import requests
import json
from datetime import datetime, timedelta
from typing import List, Dict
import os
class AWSUpdateNotifier:
def __init__(self):
self.slack_webhook_url = os.environ.get('SLACK_WEBHOOK_URL')
self.openai_api_key = os.environ.get('OPENAI_API_KEY')
self.seen_links_file = 'data/seen_links.json'
self.time_window_days = 7
def fetch_aws_feeds(self) -> List[Dict]:
"""AWS RSSフィードから更新情報を取得"""
feeds = [
'https://aws.amazon.com/new/feed/',
'https://aws.amazon.com/about-aws/whats-new/recent/feed/'
]
entries = []
cutoff_date = datetime.now() - timedelta(days=self.time_window_days)
for feed_url in feeds:
feed = feedparser.parse(feed_url)
for entry in feed.entries:
# 日付のパースと絞り込み
published = datetime.strptime(
entry.published, '%a, %d %b %Y %H:%M:%S %Z'
)
if published > cutoff_date:
entries.append({
'title': entry.title,
'link': entry.link,
'summary': entry.summary,
'published': published
})
return entries
def filter_connect_updates(self, entries: List[Dict]) -> List[Dict]:
"""Amazon Connect関連の更新のみをフィルタリング"""
keywords = [
'amazon connect', 'contact center', 'contact lens',
'voice id', 'customer profiles', 'contact flow'
]
filtered = []
for entry in entries:
text = f"{entry['title']} {entry['summary']}".lower()
if any(keyword in text for keyword in keywords):
filtered.append(entry)
return filtered
def remove_duplicates(self, entries: List[Dict]) -> List[Dict]:
"""通知済みのリンクを除外"""
seen_links = self.load_seen_links()
new_entries = [
entry for entry in entries
if entry['link'] not in seen_links
]
return new_entries
def generate_summary(self, entry: Dict) -> str:
"""OpenAI APIで日本語要約を生成"""
if not self.openai_api_key:
# APIキーがない場合は簡易要約
return self.fallback_summary(entry)
from openai import OpenAI
client = OpenAI(api_key=self.openai_api_key)
prompt = f"""
以下のAWS更新情報を日本語で3行に要約してください。
各行は80文字以内で、箇条書き形式(・で開始)としてください。
タイトル: {entry['title']}
内容: {entry['summary'][:500]}
"""
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
max_tokens=300,
temperature=0.3
)
return response.choices[0].message.content
except Exception as e:
print(f"OpenAI API error: {e}")
return self.fallback_summary(entry)
def send_to_slack(self, entries: List[Dict]):
"""Slackに通知を送信"""
if not entries:
print("No new updates to notify")
return
blocks = self.create_slack_blocks(entries)
response = requests.post(
self.slack_webhook_url,
json={"blocks": blocks}
)
if response.status_code == 200:
self.save_seen_links(entries)
print(f"Successfully notified {len(entries)} updates")
else:
print(f"Slack notification failed: {response.status_code}")
def create_slack_blocks(self, entries: List[Dict]) -> List[Dict]:
"""Slack用のブロックメッセージを作成"""
blocks = []
# ヘッダー
date_str = datetime.now().strftime('%Y年%m月%d日')
ai_status = "ON" if self.openai_api_key else "OFF"
blocks.append({
"type": "header",
"text": {
"type": "plain_text",
"text": f"📢 Amazon Connect 週次更新({len(entries)}件)"
}
})
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"📅 {date_str} | AI要約: {ai_status}"
}
})
# 各更新情報
for i, entry in enumerate(entries, 1):
summary = self.generate_summary(entry)
blocks.append({
"type": "divider"
})
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*{i}. <{entry['link']}|{entry['title']}>*\n{summary}"
}
})
return blocks
def run(self):
"""メイン処理の実行"""
print("Starting AWS update notification...")
# 1. フィード取得
entries = self.fetch_aws_feeds()
print(f"Fetched {len(entries)} entries")
# 2. フィルタリング
entries = self.filter_connect_updates(entries)
print(f"Filtered to {len(entries)} Connect-related entries")
# 3. 重複除去
entries = self.remove_duplicates(entries)
print(f"New entries: {len(entries)}")
# 4. Slack通知
self.send_to_slack(entries)
if __name__ == "__main__":
notifier = AWSUpdateNotifier()
notifier.run()
3. GitHub Actions の設定(.github/workflows/notify.yml)
name: AWS Update Notifier
on:
schedule:
# 毎週月曜日 9:00 JST (0:00 UTC)
- cron: '0 0 * * 1'
workflow_dispatch: # 手動実行も可能
jobs:
notify:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v3
with:
token: ${{ secrets.GITHUB_TOKEN }}
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.10'
- name: Install dependencies
run: |
pip install -r requirements.txt
- name: Run notification script
env:
SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
python scripts/notify.py
- name: Commit seen links
run: |
git config user.name "github-actions[bot]"
git config user.email "github-actions[bot]@users.noreply.github.com"
git add data/seen_links.json
if git diff --staged --quiet; then
echo "No changes to commit"
else
git commit -m "Update seen links [skip ci]"
git push
fi
4. 環境変数の設定
GitHub リポジトリの Settings > Secrets and variables > Actions で以下を設定:
-
SLACK_WEBHOOK_URL: Slack Incoming Webhook URL -
OPENAI_API_KEY: OpenAI API キー(オプション)
🔧 カスタマイズのポイント
監視対象サービスの変更
config.pyでキーワードを変更することで、他のAWSサービスも監視可能:
SERVICE_CONFIGS = {
'lambda': {
'keywords': ['aws lambda', 'serverless', 'function'],
'emoji': '⚡'
},
'bedrock': {
'keywords': ['amazon bedrock', 'generative ai', 'foundation model'],
'emoji': '🤖'
},
's3': {
'keywords': ['amazon s3', 'simple storage', 'object storage'],
'emoji': '🗄️'
}
}
通知頻度の変更
GitHub Actionsのcron式を変更:
# 毎日 10:00 JST
- cron: '0 1 * * *'
# 平日のみ 9:00 JST
- cron: '0 0 * * 1-5'
# 月水金 15:00 JST
- cron: '0 6 * * 1,3,5'
AI要約プロバイダーの切り替え
OpenAI以外にもClaude APIなどに対応:
class ClaudeProvider:
def generate_summary(self, text: str) -> str:
# Claude API実装
pass
class AzureOpenAIProvider:
def generate_summary(self, text: str) -> str:
# Azure OpenAI実装
pass
📊 運用してみた結果
効果
- 情報収集時間を週30分→0分に削減
- 重要な更新の見逃しがゼロに
- チーム全体での情報共有が自動化
課題と改善点
-
要約品質のばらつき
- プロンプトエンジニアリングで改善中
- Few-shot学習の導入を検討
-
API利用料金
- GPT-4o-miniで月額約$5程度
- キャッシュ機能の実装を検討
-
通知のノイズ
- 重要度スコアリング機能を追加予定
- ユーザーごとのフィルタリング機能を検討
🎯 今後の展望
- 複数チャンネル対応: サービスごとに異なるチャンネルへ通知
- インタラクティブ機能: Slackボタンで詳細表示/非表示
- トレンド分析: 更新頻度やカテゴリの可視化
- カスタムアラート: 特定キーワードの即時通知
まとめ
GitHub ActionsとOpenAI APIを活用することで、AWS更新情報の自動通知システムを簡単に構築できました。特定サービスに特化した情報収集が自動化され、チーム全体の情報共有が大幅に改善されました。