はじめに
こんにちは！レバレジーズのデータ戦略室に所属しているブライソンです。弊社は、コミュニケーションツールとしてSlackを使用しているのですが、最近便利なSlackボットを作成したので、その紹介をしようと思います。メンションをすると、スレッドの先頭にあるURLを抽出して、記事の内容を要約してくれるボット、通称「要約くん（Mr.Summary）」です。
要約くんの作成にあたって、権限周りの問題で乗り越える壁が多かったのですが、それらの話は今回の記事のスコープに入れません。純粋に、なぜ作りたいと思ったのか、どのように実装したか、導入した結果どうだったのか、などをまとめようと思います。この記事が皆さんのSlackボット作成の一助になれば嬉しいです。
なぜ作ったのか
皆さんは、技術ブログや、他社のプレスリリース、テック企業の発表記事などから情報収集をしていませんでしょうか。僕が所属しているデータ戦略室は、データにまつわる記事が自動的に連携されてくるSlackチャンネルを利用したり、メンバー同士で興味のある記事を共有しあったりすることにより、最新情報をキャッチアップしたり、他社の情報をインプットしています。
データ戦略室に限らず、会社としてそのような文化があるのですが、正直に言って僕は長い記事を読むことがあまり得意ではありません。また、業務中に時間がないが故に、共有された記事を読まずにスルーしてしまうことも多々あります。わざわざリンクに飛ばずとも、記事の内容が要約されていたらささっと読むことができるのになーと思っていました。と、いうわけで、メンションしたら要約してくれるボットを作ったろう！！という考えに至りました。
どのように作ったのか
アーキテクチャは、至ってシンプルです。Slackから受け取ったメンションをCloud Run Functions(旧Cloud Functions)にリダイレクトさせます。受け取ったメンションから、スレッドの先頭のメッセージを取得し、URLを抽出、HTMLを取得します。HTML内のbodyタグもしくはarticleタグにある記事を抽出し、Vertex AIでGeminiに要約を依頼します。最後に得られた要約をスレッドに返信します。
実装詳細
ここからは、ソースコードを交えて実装方法の詳細を紹介したいと思います。Slackのボットを作ったのは初めてなので、SDKの使い方や仕様を理解していませんでしたが、これ自体もGeminiに色々聞いたら即キャッチアップできました。なんならソースコードもほとんどはGeminiとCursorが書いているので、僕の頑張りはほとんどないと言っても過言ではないですね。挙動もシンプルなので、コードも短めです。皆さんの参考になればと思います。
ディレクトリ構造ですが、以下のような4ファイルのみです。
slack_bot
├── requirements.txt
├── main.py
├── slack_handler.py
└── utils.py
requirements.txt
必要なライブラリですが、特にバージョンは工夫していません。テキトーに選んだバージョンを使っています。
functions-framework==3.* google-cloud-logging==3.10.0 requests==2.32.3 slack-bolt==1.19.0 slack-sdk==3.29.0 vertexai==1.49.0 beautifulsoup4==4.12.2
main.py
こちらが、main.pyです。HTTPリクエストがあれば、slackbot関数が発火するように登録しています。app.event('app_mention')(handle_mention)にて、メンションイベントに対して、handle_mention関数を使用するように登録しています。また、弊社のSlackワークスペースからのリクエスト以外はエラーを返すようにしています。
import json import logging import os import functions_framework import google.cloud.logging from slack_bolt import App from slack_bolt.adapter.google_cloud_functions import SlackRequestHandler from slack_handler import handle_mention # Google Cloud Logging クライアント ライブラリを設定 logging_client = google.cloud.logging.Client() logging_client.setup_logging(log_level=logging.DEBUG) SLACK_BOT_TOKEN = os.environ['SLACK_BOT_TOKEN'] SLACK_SIGNING_SECRET = os.environ['SLACK_SIGNING_SECRET'] app = App( token=SLACK_BOT_TOKEN, signing_secret=SLACK_SIGNING_SECRET, process_before_response=True, request_verification_enabled=True, ) # メンションイベントを処理する関数を登録 app.event('app_mention')(handle_mention) # Cloud Functions 用のハンドラを作成 @functions_framework.http def slackbot(request): """Slackからのリクエストを処理するCloud FunctionsのHTTPエンドポイント。 Args: request (flask.Request): Slackからのリクエストオブジェクト。 Returns: flask.Response: Slackへのレスポンスオブジェクト。 """ header = request.headers if header.get('x-slack-retry-num'): logging.info('slack retry received') return { 'statusCode': 200, 'body': json.dumps({'message': 'No need to resend'}), } slack_handler = SlackRequestHandler(app) return slack_handler.handle(request)
slack_handler.py
これは、main.pyにて登録されているhandle_mention関数を定義したものです。メンションイベントを処理します。utils.pyで定義している各関数を使って、記事の要約内容をスレッドで返信しています。ここでは特に複雑な処理はしていません。
import logging from utils import ( get_first_message, get_urls, get_articles, get_article_summary ) def handle_mention(body, client, say): """メンションイベントを処理し、URLを含む最初のメッセージから記事を要約して返信します。 Args: body (dict): イベント情報を含む辞書。 client: Slack クライアントオブジェクト。 say: Slack メッセージを送信する関数。 Raises: Exception: エラーが発生した場合、ログに記録し、例外を発生させます。 """ try: # イベント情報を取得 event = body['event'] logging.debug(event) # スレッドの情報を取得 channel_id = event.get('channel') thread_ts = event.get('thread_ts', event['ts']) # スレッドの最初のメッセージを取得 first_message = get_first_message(client, channel_id, thread_ts) logging.debug(first_message) # メッセージ内にあるURLをリストで取得 url_list = get_urls(first_message) logging.debug(url_list) if not url_list: say(text='URLが見つかりませんでした。', thread_ts=thread_ts, channel=channel_id) else: # URLのリストから、各記事を取得 article_list = get_articles(url_list) logging.debug(article_list) if not article_list: say(text='記事が見つかりませんでした。', thread_ts=thread_ts, channel=channel_id) # メッセージにあった記事の数だけ要約をスレッドに送信 for article in article_list: summary = get_article_summary(article) logging.debug(summary) # 得られた要約結果をスレッドに送信 say(text=summary, thread_ts=thread_ts, channel=channel_id) except Exception as e: logging.error(f'Error handling mention: {e}')
utils.py
utils.pyは、文量が多いので、関数ごとに分けて紹介します。
import
各関数で利用するライブラリをインポートします。Vertex AIのSDKは初めて使いましたが、こんなに使いやすくて便利なんですね。
from vertexai.generative_models import GenerativeModel, GenerationConfig import vertexai from bs4 import BeautifulSoup import logging import os import re import requests import time
get_first_message関数
これは、受け取ったメンションを元に、スレッドの先頭メッセージを取得する関数です。client.conversations_historyをするにあたって、OAuthのスコープを適切に設定する必要があるので、注意が必要ですね。
def get_first_message(client, channel_id, message_ts): """スレッドの最初のメッセージを取得します。 Args: client: Slack クライアントオブジェクト。 channel_id (str): SlackチャンネルのID。 message_ts (str): メッセージのタイムスタンプ。 Returns: str: スレッドの最初のメッセージのテキストコンテンツ。メッセージが見つからない場合は空文字列を返します。 """ result = client.conversations_history( channel=channel_id, inclusive=True, latest=message_ts, limit=1 # 最新の1件のみを取得 ) if result["messages"]: first_message = result["messages"][0] return first_message.get("text", "") # メッセージ本文を返す。本文がない場合は空文字列を返す else: return "" # メッセージが見つからない場合は空文字列を返す
get_urls関数
メッセージの文字列から正規表現を使って、URLを抽出します。複数ある場合もあるので、リストのまま返します。
def get_urls(message): """メッセージからURLを抽出します。 Args: message (str): URLを抽出するメッセージ文字列。 Returns: list: メッセージ内で見つかったURLのリスト。 """ url_pattern = re.compile(r"https?://(?:www\.)?[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b(?:[-a-zA-Z0-9()@:%_\+.~#?&//=]*)") urls = url_pattern.findall(message) return urls
get_articles関数
先ほど抽出した各URLについて、HTMLを取得します。また、HTMLの中からarticleタグもしくはbodyタグのみを抽出し、記事の本文のみを返します。
def get_articles(url_list): """URLリストのHTMLコンテンツを取得します。 Args: url_list (list): 取得するURLのリスト。 Returns: list: 各URLに対応する記事内容のリスト。URLの取得中にエラーが発生した場合、警告がログに記録され、そのURLはスキップする。また記事内容が取得できなかった場合、HTMLを丸ごとリストに追加する。 """ article_list = [] for url in url_list: time.sleep(1) try: response = requests.get(url, timeout=20) # タイムアウトを設定 response.raise_for_status() soup = BeautifulSoup(response.content, 'html.parser') # article と body タグの中身を探す article_body = soup.find('article') or soup.find('body') if article_body: # 不要な要素 (header, footer, aside, navなど) を削除 for element in article_body.find_all(['header', 'footer', 'aside', 'nav', 'script', 'style']): element.decompose() article_list.append(article_body.get_text(strip=True, separator='\n')) else: article_list.append(response.text) except requests.exceptions.RequestException as e: logging.warn(f"Error fetching URL {url}: {e}") return article_list
get_article_summary関数
記事の本文をgemini-1.5-pro-002に投げ、要約してもらいます。その際、Slackでも見やすいように書式設定も提示しています。「太字: 半角スペース*太字*半角スペース」と言っているのが、最高にダサいですが、こうでもしないと半角スペースを空けてくれません。これでも空けないことがしょっちゅうですが...。
def get_article_summary(article): """Geminiを使用して記事の内容を要約します。 Args: article (str): 記事内容。 Returns: str: Slack用にフォーマットされた記事の要約テキスト。 """ # プロジェクト名とリージョンを環境変数から取得 project_id = os.environ.get('PROJECT_ID') region = os.environ.get('REGION') # Vertex AIの初期化 vertexai.init(project=project_id, location=region) # Gemini Proモデルの初期化 model = GenerativeModel(model_name='gemini-1.5-pro-002', generation_config=GenerationConfig(temperature=0.5) ) # 得られた記事の内容を要約して、Slackの書式設定に合わせてまとめる prompt = f'''あなたは、要約のスペシャリストです。 下に示す記事を要約してください。なお、長くても500文字以内に収めてください。 要約した内容は、次の書式設定を効果的に使い、見やすい形でまとめてください。 太字: 半角スペース*太字*半角スペース 番号付きリスト: 1. アイテム1\n2. アイテム2 箇条書きリスト: • アイテム1\n• アイテム2 引用: > 引用文 記事内容はこちら ---------------------------------------- {article} ---------------------------------------- ''' # Geminiにプロンプトを投げる response = model.generate_content(prompt) return response.text
導入
試しに前回自分が書いた記事を要約させてみました。
元々自分のために作った要約くんでしたが、所属するデータ戦略室のチャンネルで宣伝してみました。
さいごに
今回は、要約くんの紹介をしました。実はこの記事を書いている途中で、全く同じボットを作成した人の記事を見つけたので、共有しておきます。この方は、スタンプに反応して要約を返信するボットを作っていて、なるほどその方法もあったかと思いました。要約くんの改善をする際に参考にしようと思います。 zenn.dev
ボット作成に関して、プログラム自体は早々に出来上がったのですが、プロンプトをいじくることに時間を使ってしまいました。例えば、元々HTMLを丸々含めていたのですが、レスポンスが遅すぎて、articleタグとbodyタグのみを含めるように変更しました。また、gemini-1.5-pro-001を使っていると、書式設定をなかなか反映してくれなかったので、gemini-1.5-pro-002に変えたところ、正しく書式設定を使ってくれるようになりました。Slackの書式設定もバカ正直に全て書いていたのですが、今度はGeminiが全ての書式設定を無理矢理使ってしまう現象に悩まされ、少なくとも必要な書式だけに絞ったりもしました。本質ではないなーと思いつつ、今後もプロンプトエンジニアリングが重要になってくることを肌で感じることになりました。とはいえ、このようなボットを他にも作りたいので、思いついたらまた作ろうと思います！
