検索ワードからLLMで専門家を見つけるSlack bot「Navigator」の開発
はじめに
株式会社松尾研究所でインターンをしているyukiです。
Slack上でユーザーのメッセージを分析し、特定の分野に詳しい人を提示するbotを開発したのでその紹介をします。
これは松尾研究所で行われたAIハッカソンで発表された「Navigator」をアップグレードして社内でリリースしたものになります。
大規模な組織のSlackには多くのユーザー、チャンネルが存在し、日々膨大な量のメッセージが交わされています。そのような環境では誰が何に詳しいのかを完全に把握することは難しく、分からないことを誰に聞けばよいか分からないことも多いです。
今回開発したbot、Navigatorはそのような問題を解決するためのツールとなっています。
従来の課題と解決手法
Slackなどのチャットツールは、チーム間でのコミュニケーションをスムーズにする一方で、以下のような問題が存在します。
・メッセージの増加による重要情報の埋没
・似た内容を扱うチャンネルの乱立
このような中でメッセージの検索をしても、検索結果が膨大すぎるため、ある分野に詳しい人に相談したいと思っても「どこで」、「誰に」聞けばよいのか判断するのが難しいです。
例えば松尾研究所のSlackワークスペースで「テックブログ」と検索すると以下のような結果になります。
メッセージが多く、チャンネルも複数あるためここからテックブログに詳しい人を見つけるのは手間がかかりそうです。
そこで、この課題を解決するため「メッセージの分析と可視化を行うSlack bot」の開発を行いました。このbotは以下のような機能を持っています:
・メッセージの発信回数と発信したチャンネルの可視化
・指定キーワードだけでなく、似た内容のメッセージも抽出するセマンティック検索
・発言回数だけでなく、メッセージ内容を分析することで本当にその知見に詳しいユーザを提示
これらにより、重要な情報や課題を迅速に把握し、チームの効率を向上できるようにします。
botの開発
開発環境
Slackアプリ開発用フレームワークである、Bolt for Pythonを使って開発しました。
-
言語: Python
-
主要ライブラリ: Slack SDK, Slack Bolt
-
デプロイ環境: AWS EC2
-
API: OpenAI, Gemini
準備
コードを書く前に、まずはbotをワークスペースにインストールする必要があります。
Slack APIでbotのインストールと、コードを実行するのに必要なtokenが発行できます。設定の"OAuth & Permissions"でbotに必要な権限を与えておきます。
APIでOpenAIとGeminiを使い分けている理由は、下図のようにGeminiの方が入力できるトークンの上限が大きいからです。今回は後述の理由でLLMへの入力文が長くなる箇所があるため、その部分でGemini 2.0 Flashを使用しています。
Model | 入力トークン上限 | 出力トークン上限 |
---|---|---|
gpt-4o | 128,000 | 16,384 |
gpt-o1 | 200,000 | 100,000 |
Gemini 2.0 Flash | 1,048,576 | 8,192 |
動作プロセス
botの動作は以下のような流れになっています。
- ユーザーが検索ワードをボットに送信
- OpenAI APIを使ってLLMに送信し、検索ワードの関連ワードを生成
- Slack APIにある関数を用いて検索ワード、関連ワードを含むメッセージを取得
- 取得したメッセージから各ユーザーの送信回数、送信したチャンネルをカウント
- ユーザーIDをSlackでの表示名に変更して検索結果を送信
- メッセージをLLMで分析し結果を送信
各過程について詳しく説明します。
1. ユーザーが検索ワードをボットに送信
"@navigator 検索ワード"のようにユーザーがBotをメンションした場合に動作するようにします。これはBoltにあるevent() メソッドを使うことで実装出来ます。今回はメンションされたときに応答するようにしたいので、Event APIのmessage
を使い、メッセージに"@navigator"が含まれるかどうかをif文で判断します。
@app.event("message")
def handle_message_events(body, logger, say):
logger.info(body)
event = body["event"]
if event.get("subtype") == "bot_message":
return
channel = event["channel"]
thread_ts = event["ts"]
text = event.get("text", "")
if f"<@{slack_bot_id}>" not in text:
return
input_message = text.replace(f"<@{slack_bot_id}>", "").strip()
if text.strip() == f"<@{slack_bot_id}>":
send_function_selection(channel, thread_ts)
return
if input_message.startswith("フィードバック”): #フィードバック取得用
word = input_message[7:].strip()
collect_feedback(event, say, word)
else:
say("入力ありがとうございます。少々お待ちください...", thread_ts=thread_ts)
# Handle different search types
if input_message.startswith("OR"): #OR検索
words = input_message[2:].split(',')
search_with_multiple_words(event, say, words, cnt= 0.2)
elif input_message.startswith("一致"): #一致検索
word = input_message[2:].strip()
search_user_with_words_input(event, say, word)
else: #セマンティック検索
word = input_message
search_with_semantic_words(event, say, word, cnt= 0.2)
def send_function_selection(channel, thread_ts):
# 機能選択のメッセージを作成
message_text = (
"以下のように検索ワードを入力してください。\n"
"@navigator word (セマンティック検索)\n"
"@navigator 一致 word (一致検索)\n"
"@navigator OR word1,word2 (OR検索)\n"
)
# メッセージとボタンを送信
app.client.chat_postMessage(
channel=channel,
text=message_text,
thread_ts=thread_ts,
blocks=[
{
"type": "section",
"text": {"type": "mrkdwn", "text": message_text}
},
]
)
2. OpenAI APIを使ってLLMに送信し、検索ワードの関連ワードを生成
セマンティック検索を実装するため、入力ワード以外の検索ワードを生成します。LLMに渡すプロンプトは以下のようにしました。
あなたは検索アシスタントです。 入力された検索ワードに関連する単語(表記ゆれや一部の単語など)をカンマ区切りで出力してください。
単語は4つだけ出力してください。
例
入力
LLM
出力
llm,大規模言語モデル,言語モデル,生成AI
入力
stable diffusion
出力
diffusion,diffusionモデル,画像生成,生成AI
これによって入力ワードの関連語の検索も可能になります。また、Few Shot promptingを使うことでワードの質が格段に向上しました。
検索の際には生成された単語と入力ワードの5単語で検索を行います。
3. Slack APIにある関数を用いて検索ワード、関連ワードを含むメッセージを取得
Slack API にはsearch.messageという関数があります。
これを使って指定されたワードを含むメッセージを検索します。メッセージ本文に加えて、投稿したチャンネル、ユーザーIDなども取得できます。メッセージは一度に100件しか取得できないので操作を繰り返すことで任意の数だけメッセージを取得します。以下の例では500件取得しています。
def search_messages(webclient, query):
max_results=int(500)
results = []
page = 1
while True:
response = webclient.search_messages(query=query, count=100, page=page)
messages = response["messages"]["matches"]
# 検索結果自体が空になった場合に終了
if not messages:
break
filtered_messages = [
msg for msg in messages
if (msg.get('user') and
msg.get('channel',{}).get('is_private')==False)
]
results.extend(filtered_messages)
# 最大結果数に達した場合に終了
if len(results) >= max_results or page >=max_results/100:
break
page += 1
return results[:max_results]
4. 取得したメッセージから各ユーザーの送信回数、送信したチャンネルをカウント
Counterを使ってそれぞれカウントします。例えば以下のような処理です。ここでユーザー、チャンネルのそれぞれ上位5件を抽出しておきます。
search_results = search_messages(webclient, query)
user_id_list = []
channel_list = []
for message in search_results:
try:
user_id = message.get('user', '')
if user_id:
user_id_list.append(user_id)
channel_list.append(message['channel']['name'])
except:
continue
user_count = Counter(user_id_list)
channel_count = Counter(channel_list)
top_users = user_count.most_common(5)
top_channels = channel_count.most_common(5)
5. ユーザーIDをslackでの表示名に変更して検索結果を送信
search.messageでは送信者のslack上の表示名(Display name)を取得できません。そのため、U00ABCDEFGのようなUser IDから表示名を取得する必要があります。ページネーションを使ってユーザーIDと表示名を対応させる辞書を作ります。
def get_user_display_name_map(webclient):
user_display_name_map = {}
cursor = None
while True:
try:
# cursorを使用してページネーション
if cursor:
response = webclient.users_list(limit=1000, cursor=cursor)
else:
response = webclient.users_list(limit=1000)
users_info = response['members']
# ユーザー情報の処理
for user in users_info:
if not user['is_bot'] and user['id'] != 'USLACKBOT':
user_display_name_map[user['id']] = user['profile'].get('display_name',
user['profile'].get('real_name', 'Unknown'))
# 次のページがあるかチェック
cursor = response['response_metadata'].get('next_cursor')
# カーソルが空の場合は全データを取得完了
if not cursor:
break
except Exception as e:
print(f"Error fetching user information: {e}")
break
return user_display_name_map
6. メッセージをLLMで分析し結果を送信
これまで取得した情報をもとに、ユーザー名がキー、検索ワードを含むメッセージとリンクが値となる辞書を作成してGeminiに渡します。メッセージのリンクは以下のようにして作成することが出来ます。
https://{ワークスペースID}.slack.com/archives/{チャンネルID}/p{メッセージのタイムスタンプ}
user_messages_dict = {}
workspace_id = os.getenv("SLACK_WORKSPACE_ID") # 環境変数からワークスペースIDを取得
for message in search_results:
user_id = message.get("user")
text = message.get("text", "")
if user_id and text and user_id in top_user_ids:
display_name = user_display_name_map.get(user_id, "Unknown User")
channel_id = message["channel"]["id"]
message_ts = message["ts"]
# Slackメッセージリンクを生成
message_link = f"https://{workspace_id}.slack.com/archives/{channel_id}/p{message_ts.replace('.', '')}"
user_messages_dict.setdefault(display_name, []).append({
"text": text,
"link": message_link
})
formatted_messages = ""
for user, messages in user_messages_dict.items():
formatted_messages += f"ユーザー: {user}\n"
for msg in messages:
formatted_messages += f"- {msg['text']}\n Link: {msg['link']}\n"
formatted_messages += "\n"
system_prompt = load_prompt("system_prompt.txt")
#Gemini
combined_prompt = f"""
{system_prompt}
以下が{', '.join(query)}の検索結果のメッセージ一覧です。メッセージにはリンクが含まれているので、分析結果と共にリンクも示してください。
{formatted_messages}
"""
try:
response = model.generate_content(combined_prompt)
response_text = response.text
say(response_text, thread_ts=thread_ts)
Geminiに入力するプロンプトは以下のようにしました。
あなたはSlackメッセージの検索結果を基に、特定の事柄に詳しいユーザーを抽出するアシスタントです。
以下はキーワード検索結果のメッセージ一覧です。辞書形式になっており、キーが発信者、値がそのユーザーのメッセージです。
各ユーザーのメッセージを分析し、キーワードに詳しいと思われるユーザーを順序づけしてください。
その際に根拠を述べ、根拠としてふさわしいメッセージの概要とリンクを出力してください。
マークダウン記法は使わずに、Slackのメッセージ記法に従ってください。
#出力例
1. *ユーザー名*
・キーワードについての発言の傾向
~
・メッセージ概要
>[Link1]
>[Link2]
2. *ユーザー名*
デフォルトだとマークダウン記法で出力されがちですが、Slackは対応していないのでSlackの記法に合わせた出力をさせる必要があります。また、入力文が長くなるのでここではGemini 2.0 Flashを使用しています。
結果
以下が実際にbotを動かした時の出力です。
入力ワードから関連ワードを生成してメッセージを検索し、その分析結果まで出力できています。
メッセージの検索結果
検索結果の分析
出力部分のコードに追加することで、Google Formのリンクを貼ったり、ボタンを実装したりしてフィードバックをもらうこともできます。「@navigator フィードバック」のようにbotに直接フィードバックを送れるようにすることも可能です。
グッドボタンを押したときの例
開発したSlackbotを社内で試験運用して社員やインターンの方に使っていただいたところ、以下のようなフィードバックをいただきました。
・入力から検索ワードを生成するので曖昧検索やタイプミスにも対応している
・所属やチャンネルの性質,特定のプロジェクトチャンネルだけでの検索などもあると良い。
・検索語句の文脈上の違いを理解できるとより良い。(例:ウェブの文脈におけるE2Eと深層学習の文脈でのE2E)
まとめ
今回、メッセージの検索から分析まで行うSlackbotを開発し、実際の運用に組み込むことで効率よく組織内の情報が得られるようにすることを目指しました。
さらなる改善としては、
・生成ワードや分析の精度向上
・処理の高速化
・より詳細な検索条件の実装
などが考えられます。
この記事が同様の課題に取り組む方々の参考になれば幸いです。最後までお読みいただきありがとうございました。
Discussion