はじめに
こんばんは。べんと申します。
生成AIの台頭以降、技術の進歩が早すぎて情報を追いきれていません。日々新しいニュースが発表されますが、毎日大量の記事をチェックするのは時間がかかります。
そこで、ニュースの情報を元に、真に自分に必要な情報を整理してくれるアシスタントがいればと思い、生成AIを使ったアプリケーションを作ってみました。
本記事では、アプリケーションの機能ブロックごとに解説します。
機能ブロック
全5ブロックに分けて解説します。
- RSSフィードを取得（ニュースサイトから最新記事を取得）
- 今日の新着ニュースを選別（日本時間での日付判定）
- 生成AIで記事を選別
- 選別したニュースを使って情報を整理
- Gmail APIを使ってメール通知
1. RSSフィードを取得
ニュースなどの最新情報を効率的に取得するため、RSSフィードを活用します。pythonにはfeedparserというライブラリがあり、これを使うとRSS（Really Simple Syndication）から簡単に情報を取得できます。
まずはfeedparserをインストールします。
pip install feedparser
pythonコード内では下記のようにしてRSSフィードの情報を取得できます。
import feedparser
# RSSフィードのURLを指定
rss_url = "https://example.com/rss"
# フィードを解析
feed = feedparser.parse(rss_url)
for entry in feed.entries:
print(f"タイトル: {entry.title}")
print(f"リンク: {entry.link}")
print(f"記事の要約: {entry.summary}")
print(f"公開日: {entry.published}")
feed.entries（リスト）の各要素は、RSSフィード内に記述された一つ一つの記事です。
以降のパートでは、各記事のタイトル、リンク、記事の要約、公開日を利用します。
ただし、RSSフィードによっては、例えば記事の要約を要素としてもたない場合があります。今回は、簡単のため記事の要約を要素としてもつRSSフィードを選んで使うことにします。
2. 今日の新着ニュースを選別
取得した記事には、今日の新着ニュースの他、昨日までに公開された記事が含まれることがあります。本アプリで毎日情報収集することを想定すると、昨日までの記事は重複になるので、本日公開されたものだけを選んで使うことにします。
現在の日付（日本時間）を取得し、feedparserで取得した記事の公開日をチェックします。
from datetime import datetime, timezone, timedelta
# 今日の日付を確認
jst = timezone(timedelta(hours=9))
today = datetime.now(jst).date()
# 今日のニュースだけを格納する変数を定義
news_entries = []
# 各記事を探索
for entry in feed.entries:
# 公開日情報を取得
entry_date = datetime(*entry.published_parsed[:6], tzinfo=jst.astimezone(jst)).date()
# 公開日が今日の日付の場合のみ、news_entriesに格納
if entry_date == today:
news_entries.append({
"title": entry.title,
"link": entry.link,
"summary": entry.summary,
})
ただし、実際には公開日（published_parsed）ではなく更新日（updated_parsed）に日付が入っている場合や、記事の要約（summary）をもたない記事もあります。
また、一つのRSSフィードではなく、複数のRSSフィードから情報取得することも考慮し、書き直したコードがこちら。
rss_urls = {
"RSS1": "URL1",
"RSS2": "URL2"
}
def get_today_news(rss_urls):
# 本日の新着記事のみを格納する変数
news_entries = []
# 今日の日付（日本時間）を取得
jst = timezone(timedelta(hours=9))
today = datetime.now(jst).date()
# RSSフィードを順に巡回
for rss_source, rss_url in rss_urls.items():
feed = feedparser.parse(rss_url)
# RSSフィード内の各記事を巡回
for entry in feed.entries:
entry_date = None
# 公開日または更新日を取得
if hasattr(entry, "published_parsed") and entry.published_parsed:
entry_date = datetime(*entry.published_parsed[:6], tzinfo=jst).date()
elif hasattr(entry, "updated_parsed") and entry.updated_parsed:
entry_date = datetime(*entry.updated_parsed[:6], tzinfo=jst).date()
# 公開日または更新日が本日の日付である、かつ記事の要約がある場合のみ、記事を保存
if entry_date and entry_date == today and entry.summary:
news_entries.append({
"title": entry.title,
"link": entry.link,
"summary": entry.summary,
"source": rss_source
})
return news_entries
news_entries = get_today_news(rss_urls)
3. 生成AIで記事を選別
ここからは生成AIを使います。今回は無料枠が存在するGoogleのGemini（gemini-1.5-flash）を使います。使い方はこちら。
from dotenv import load_dotenv
load_dotenv()
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(
model="gemini-1.5-flash",
temperature=0,
max_tokens=None,
timeout=60,
max_retries=2,
)
output = llm.invoke("こんにちは。")
print(output.content) # 生成AIからの応答を表示
Geminiを使うためには、APIキーが必要になります。今回は、このAPIキーを.envファイルに保存し、コードと同じ階層に格納していることを前提としています。
# .env
API_KEY=<APIキー>
さて、先程取得した本日の新着記事とGeminiを使って関心のある記事だけを選びましょう。
# ニュースのリストを入力すると、必要なニュースだけを選んで返す関数
def _filter_relevant_news(news_entries):
# ニュースのリストを、生成AIに入力できるテキスト形式に整形
# ニュースのリストは項番を付与され下記の形式に。
# 1. タイトル：<ニュース1のタイトル>
# 要約：<ニュース1の要約>
# 2. タイトル：<ニュース2のタイトル>
# ......
news_text = "\n\n".join([
f"{i+1}. タイトル: {news['title']}\n 要約: {news['summary']}"
for i, news in enumerate(news_entries)
])
# テキスト形式に整形されたニュースのリストと、指示文を合わせてプロンプトに変換
prompt = f"""
あなたはAIに関連する情報を収集するためのAIアシスタントです。以下のニュースリストの中から、私の関心に合致するものだけを選んでください。
# 関心のある分野
- AI技術の最新動向
- AIの活用事例
- AIに関連する法規制、ガバナンス
# ニュースリスト
{news_text}
# 出力フォーマット
関心のあるニュースの番号をカンマ区切りで挙げてください（例：1,3,5）
関連するニュースがない場合、「なし」と答えてください。
"""
# プロンプトをGeminiに入力、関心のあるニュースの項番を返す
response = llm.invoke(prompt)
selected_indices = [
int(i.strip()) - 1 for i in response.content.split(",") if i.strip().isdigit()
]
# 項番の情報を元に、関心のあるニュースのみを集めたリストを返す
return [news_list[i] for i in selected_indices if 0<= i < len(news_list)]
filtered_news = _filter_relevant_news(news_entries)
プロンプトについては、欲しい情報の種類に応じて工夫してみてください。
最後に、RSSフィードから取得した情報があまりにも多いとGeminiが処理しきれないので、news_entries内のニュースをバッチに分割して処理する関数を用意しておきます。
# news_entriesの要素を100個ずつに分割して処理する関数
def filter_relevant_news(news_entries, batch_size=100):
# 関心の高いニュースだけを格納する変数
filtered_news = []
# 100ニュースずつ処理
for i in range(0, len(news_entries), batch_size):
batch = [{"title": entry["title"], "summary": entry["summary"], "link": entry["link"]} for entry in news_entries[i:i+batch_size]]
filtered_news.extend(_filter_relevant_news(batch))
return filtered_news
filtered_news = filter_relevant_news(news_entries)
4. 選別したニュースを使って情報を整理
さて、ここまでで、本日の新着ニュースのうち、関心の高いものだけを選別できました。次に、厳選されたニュースの内容を、Geminiに整理・要約してもらいましょう。
def summarize_news(filtered_news):
# 関心の高いニュースがなかった場合の処理
if not filtered_news:
return "新しいニュースはありませんでした。"
# ニュースのリストを、生成AIに入力できるテキスト形式に整形
# ニュースのリストは項番を付与され下記の形式に。
# 1. タイトル：<ニュース1のタイトル>
# 要約：<ニュース1の要約>
# 2. タイトル：<ニュース2のタイトル>
# ......
news_text = "\n\n".join([
f"{i+1}. タイトル: {news['title']}\n 要約: {news['summary']}\n リンク: {news['link']}"
for i, news in enumerate(filtered_news)
])
# テキスト形式に整形されたニュースのリストと、指示文を合わせてプロンプトに変換
prompt = f"""
以下のニュースリストを要約してください。
# ニュースリスト
{news_text}
# 注意事項
- 複数のニュースが関連していることがあります。関連性を考慮し、重要なポイントを整理してください。
- 不要な情報を削除し、簡潔にまとめてください。
- 各トピックの最後に、関連ニュースのタイトルとそのリンクを掲示してください。
"""
# Geminiにプロンプトを入力し、要約された情報を取得
response = llm.invoke(prompt)
return response.content
こちらもプロンプトはほしい形式に応じて工夫してください。日本語を少し変えるだけで、アウトプットの形が結構変わります。日本語で指示を出すというのがいかに難しいことかわかります。
5. Gmail APIを使ってメール通知
新着ニュースの中から関心の高い情報を抽出し、要約するところまでができました。最後はこれをメールで通知します。
いろいろ方法はありますが、よく使うGmailからの送信を選択します。Gmailを使ったプログラムからのメール送信は、従来アプリパスワードを使うことでできたようですが、このアプリパスワード、セキュリティ上の都合により、Googleがサポートを終了しています。そのため、今後はOAuth 2.0認証を使って実行することが必要です。
OAuth 2.0認証に必要なGoogle Cloud上の手続きについてはここでは省略します。私はアルファシステムズさんの記事の途中までを参考にさせていただきました。
ざっくり紹介すると、
- Gmail APIを有効化
- OAuthクライアントIDを作成
- 認証情報が記されたjsonファイルをダウンロード
までを実行します。このjsonファイルを動作させたいpythonコードと同階層に配置し、下記のプログラムを実行することで、gmailからのメール送信が可能になります。
import os
import base64
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
def authenticate_gmail():
creds = None
SCOPES = ['https://www.googleapis.com/auth/gmail.send']
if os.path.exists("token.json"):
creds = Credentials.from_authorized_user_file("token.json", SCOPES)
else:
flow = InstalledAppFlow.from_client_secrets_file(
"credentials.json", SCOPES
)
creds = flow.run_local_server(port=0)
with open("token.json", "w") as token:
token.write(creds.to_json())
return creds
gmailからのメール送信には、認証情報が格納されたtoken.jsonファイルが必要になります。上記のコードを初めて実行すると、ウェブブラウザが立ち上がり、Googleのアクセス許可に進みます。許可するとtoken.jsonがプログラムと同階層に配置され、これを使うことでメール送信が可能になります。
メール送信のプログラムは下記の通り。
from googleapiclient.discovery import build
from email.mime.text import MIMEText
# メール件名（subject）、メール本文（body）を指定して、メールを送信する関数
def send_email(subject, body):
# 認証情報を取得
creds = authenticate_gmail()
service = build("gmail", "v1", credentials=creds)
# メッセージを構成
message = MIMEText(body)
message["to"] = "送信先メールアドレス"
message["from"] = "送信元メールアドレス（gmail）"
message["subject"] = subject
raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
try:
message = service.users().messages().send(userId="me", body={"raw": raw}).execute()
print(f"Email sent: {message['id']}")
except Exception as e:
print(f"Failed to send: {e}")
まとめ
以上、本記事では、ニュースの情報を元に、真に自分に必要な情報を整理して、さらにメールで通知をしてくれるアシスタントについて、順を追って解説しました。
このアシスタントを使えば、自分の関心のあるニュースだけを毎日時間をかけずにチェックすることができます。
最後に、これまでに定義した関数を呼び出すコードを記述しておきます。
すでに世の中には便利なツールが溢れていますが、仕組みを理解していて、信頼できるツールは意外と少ないかもしれません。自分のアイデアを、簡単なところからでもいいから自分で実装してみることで、今後も理解を深めていきたいと思います。
べん
# まとめ
if __name__ == "__main__":
news_list = get_today_news(rss_urls)
filtered_news = filter_relevant_news(news_list)
news_summary = summarize_news(filtered_news)
jst = timezone(timedelta(hours=9))
today = datetime.now(jst).strftime('%Y-%m-%d')
subject = f"{today}のAIニュース"
body = f"""{today}のAIニュースをお届けします。
{news_summary}
"""
send_email(subject, body)