毎日Discordにオススメ記事を配信してくれるAI Botを一瞬で作る！

以下の様な最新記事を毎日配信してくれる兄貴がいたらありがたいですよね！
今回は、FastAPIとGemini APIを使用して、LambdaとEventBridgeでサクっとBotを作成する方法をまとめました！

説明はいいから、さっさとコードだけ見せろやって方はコチラ👇
https://github.com/mk668a/lambda-discord-rss-bot

使用する技術は以下になります。Gemini APIが無料なので今回はGeminiを選びました。

• Python3.12
• FastAPI
• Lambda
• EventBridge
• Docker
• Gemini API

事前準備

DiscordのWebhook URLを発行します。
Botを召喚したいチャンネルの設定から、Botを作成して、Webhook URLを保存してください。
アイコン画像は、ChatGPTに作成してもらいました。

構成

今回は、serverlessフレームワークを使用し、コマンド一発でデプロイできるような構成にします。
AWS上のWeb UIを触らずにローカルでもコマンドを２つ実行するだけで簡単にデプロイが完了します。
今回EventBridgeを使用して定期実行しますが、実行する時間帯の設定などもファイルに定義しておくことで、自動でデプロイされるようにします。

.
├── app
│   ├── .env
│   ├── __init__.py
│   ├── main.py
│   └── src
│       ├── logger.py
│       ├── service.py
│       └── settings.py
├── Dockerfile
├── requirements.txt
└── serverless.yml

serverless.ymlの設定

serverless.ymlファイルを作成します。
ここの設定さえしっかりすれば後はコマンドでデプロイするだけです。

serverless.yml

service: discord-rss-bot-test
frameworkVersion: "4"

plugins:
  - serverless-python-requirements

provider:
  name: aws
  runtime: python3.12
  memorySize: 512
  architecture: x86_64
  region: ap-northeast-1
  timeout: 30
  apiGateway:
    binaryMediaTypes:
      - "*/*"
  ecr:
    images:
      discord-rss-bot-test:
        path: ./
        platform: linux/amd64

functions:
  app:
    events:
      - eventBridge:
          schedule: cron(0 3 * * ? *)
          input:
            bot_name: developer
    image:
      name: discord-rss-bot-test
      command:
        - app.main.handler

package:
  individually: true

eventBridgeの設定を記述します。
scheduleにcronでチャンネルに投稿する時間を指定します。（cronについての詳しい解説は省きます。）
今回は、日本時間の午後12時に配信するために、UTCで3を指定しました。

bot_nameについては、LambdaのAPI発火時にコード上で受け取る変数になります。後ほど説明します。

Dockerfileを作成

Dockerfile

FROM --platform=linux/amd64 public.ecr.aws/lambda/python:3.12

COPY requirements.txt .
RUN pip install --upgrade pip
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["app.main.handler"]

serverless.ymlファイルのimageにて指定するDocker imageをbuildするために作成します。

main.pyの作成

appディレクトリを作成して、その中にmain.pyを作成します。

main.py

from app.src.logger import logger
from app.src.service import call_llm


def handler(event, context):
    logger.info(event)
    bot_name = event.get("bot_name")

    if bot_name == "developer":
        theme = "エンジニア"
        bot_name = "兄貴"
        urls = [
            "https://www.publickey1.jp/atom.xml",
            "https://ai-data-base.com/feed",
            "https://zenn.dev/topics/flutter/feed",
        ]
        discord_url = "https://discord.com/api/webhooks/hogehoge"

        call_main(urls, theme, bot_name, discord_url)
    else:
        logger.info("other")

    return {"message": event}


if __name__ == "__main__":
    # test
    handler({"bot_name": "developer"}, None)

theme、bot_nameは、Geminiへのプロンプトを作成する際に使用するために、定義しています。
また、urlsは配信してほしいRSSのURLのリストになります。

discord_urlには、事前準備で取得したDiscordのWebhook URLを入れます。

call_mainにて、RSS記事取得からDiscordへのポストの処理を行います。後ほど、この関数について解説します。

event.get("bot_name")にて、serverless.ymlで指定したbot_nameの値を取得できます。serverless.ymlのeventBridgeの設定を増やす事で複数のBotを作成できるような仕様にしました。

ローカルでテストする場合は、以下のコマンドを実行してください。

python -m app.main

RSSを取得する関数を実装する

appディレクトリ配下にsrcディレクトリを作成します。
appディレクトリ配下にservice.pyを作成して、ここに細かい関数を実装していきます。

まずは、必要なライブラリをimportします。

service.py

import feedparser
from dataclasses import dataclass
from typing import List
import requests
import json
from datetime import datetime, timedelta, timezone
from dateutil import parser
import time
from app.src import logger
from app.src.settings import GOOGLE_AI_STUDIO_API_KEY

RSS記事を取得する関数を実装します。

service.py

@dataclass
class RssContent:
    title: str
    url: str
    published: str


def get_rss(endpoint: str) -> List[RssContent]:
    feed = feedparser.parse(endpoint)
    rss_list: List[RssContent] = []
    for entry in feed.entries:
        if not entry.get("link"):
            continue
        rss_content = RssContent(
            title=entry.title, url=entry.link, published=entry.published
        )
        rss_list.append(rss_content)
    return rss_list

Gemini APIを呼び出す関数

Geminiを呼び出す関数を実装します。

service.py

def call_gemini(page_content: str, theme: str, bot_name: str):
    prompt = f"""あなたはRSSを発信する{bot_name}です。{bot_name}のような口調でRSS(テーマ:{theme})の内容を500文字以内で要約してください。改行は含めないでください。
<info>
{page_content}
</info>
"""
    uri = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash-latest:generateContent?key={GOOGLE_AI_STUDIO_API_KEY}"
    headers = {"Content-Type": "application/json"}
    data = {
        "contents": [{"parts": [{"text": prompt}]}],
    }
    response = requests.post(uri, headers=headers, data=json.dumps(data))

    json_data = response.json()

    response_text = (
        json_data.get("candidates")[0].get("content").get("parts")[0].get("text")
    )

    return response_text

bot_name、themeはmain.pyで定義した値が入ります。

theme = "エンジニア"
bot_name = "兄貴"

環境変数から、GOOGLE_AI_STUDIO_API_KEYを取得しています。
.envファイルに設定されている変数をsettings.pyに定義しています。

settings.py

import os
from os.path import join, dirname
from dotenv import load_dotenv

load_dotenv(verbose=True)

dotenv_path = join(dirname(__file__), ".env")
load_dotenv(dotenv_path)

GOOGLE_AI_STUDIO_API_KEY = os.environ.get("GOOGLE_AI_STUDIO_API_KEY")

全体処理の実装

call_mainというmain.pyから呼び出す全体処理の関数を作成します。
今まで実装してきたget_rss()、call_gemini()関数はここで使用します。

service.py

def call_main(urls: List[str], theme: str, bot_name: str, discord_url: str):
    # URLsからRSSのリストを取得
    rss_list = []
    for url in urls:
        rss_list.extend(get_rss(url))

    # publishedが24時間以内のものを取得
    rss_list = [
        rss
        for rss in rss_list
        if datetime.now(timezone.utc) - parser.parse(rss.published) < timedelta(days=1)
    ]

    # rss_listを最新順に並び替える
    rss_list = sorted(
        rss_list,
        key=lambda x: parser.parse(x.published),
        reverse=False,
    )

    # rss_listを10件ずつ分割する
    rss_lists = [rss_list[i : i + 10] for i in range(0, len(rss_list), 10)]

    for rss_list in rss_lists:
        # RSSの記事のタイトル一覧を作成（リンクを含む）
        parsed_content = "".join(
            [f"{i+1}. [{rss.title}]({rss.url})\n" for i, rss in enumerate(rss_list)]
        )
        # RSSの記事のタイトル一覧を作成
        content = "".join([f"{i}. {rss.title}\n" for i, rss in enumerate(rss_list)])

        for _ in range(3):  # 最大3回リトライ
            try:
                res = call_gemini(content, theme, bot_name)
                break
            except Exception as e:
                logger.error(f"エラーが発生しました: {e}")
                time.sleep(1)
        else:
            res = "エラーが発生しました。"

        requests.post(discord_url, json={"content": f"{res}\n{parsed_content}"})

今回は、毎日決まった時間にポストされるようにcronで設定しました。なので、24時間以内の記事だけを取得するようにして、毎日のポストで重複した記事が含まれないようにしています。

また、Discordの無課金状態では、投稿の文字数制限があるため、10記事ずつ分割してポストするようにしました。

parsed_contentは、Discordに投稿するための、リンクを含むタイトルをマークダウン形式で記述したものになります。
contentは、Geminiのプロンプトで使用する記事のタイトル一覧になります。

call_gemini()関数にリトライ処理を入れてるのは、Gemini APIのエラー（レート制限やLLMによる出力のブレ）対策です。

その他(logger)

CloudWatchにてログを見やすくするために、loggerを実装しました。

import logging
from logging.handlers import TimedRotatingFileHandler
import os
import sys


def setup_logging(log_dir="logs", log_file_name="info.log"):
    aws_lambda_path = "/tmp/"
    log_dir = os.path.join(aws_lambda_path, log_dir)

    if not os.path.exists(log_dir):
        os.makedirs(log_dir)

    log_file_path = os.path.join(log_dir, log_file_name)

    logger = logging.getLogger("Logger")
    logger.setLevel(logging.INFO)  # ログレベルの設定

    handler = TimedRotatingFileHandler(
        filename=log_file_path,
        when="midnight",
        interval=1,
        encoding="utf-8",
        backupCount=200,
    )

    formatter = logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )
    handler.setFormatter(formatter)

    # StreamHandlerの追加
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(formatter)

    # ハンドラーをロガーに追加
    logger.addHandler(handler)
    logger.addHandler(console_handler)

    return logger


logger = setup_logging()

デプロイ

実装は以上なので、実際にデプロイしてみましょう。

AWSの設定をします。

awsコマンドのインストールなどの解説は省きます。

aws configure

serverlessコマンドのインストール

serverlessコマンドのインストールをします。

npm install -g serverless

venvによる仮想環境の構築

Pythonのバージョン管理としてvenvを使用しました。
特にこだわりが無い場合は、この手順はスキップしても大丈夫かと思います。

python -m venv ./venv
source ./venv/bin/activate

ライブラリのインストール

必要なライブラリをインストールします。

requirements.txt

boto3==1.35.54
botocore==1.35.54
certifi==2024.8.30
charset-normalizer==3.4.0
feedparser==6.0.11
idna==3.10
jmespath==1.0.1
python-dateutil==2.9.0.post0
python-dotenv==1.0.1
requests==2.32.3
s3transfer==0.10.3
sgmllib3k==1.0.0
six==1.16.0
urllib3==2.2.3

pip install -r requirements.txt

Lambdaへのデプロイ

1. Docker Imageをbuildします。

docker build -t discord-rss-bot-test .

※Apple Siliconを使用する場合

docker build --platform linux/amd64 -t discord-rss-bot-test .

2. デプロイします。

sls deploy

終わりに

今回実装したコードは以下になります。
cloneして該当箇所を変更すれば、すぐにデプロイできるような状態になっています。
爆速で作りたい方は、以下のコードを変更するのが手っ取り早いと思います。
https://github.com/mk668a/lambda-discord-rss-bot

コメントにて質問があれば、なるべく早く回答いたします。