Slackで独自カスタマイズしたAIと会話してみた〜Lambda(AWS)とPython使用編〜

AI Python Slack AWS Lambda 生成 AI tech はじめに

LLMの進歩著しい昨今、AIを使ってキャラクターとの会話ができるサービスも数多く出てきました。 今回の記事では、Spiral.AI株式会社のTwinRoomというサービスを使い、Slackで独自カスタマイズしたAIと簡単に会話してみます。

この記事で紹介する機能を活用すれば、開発者・企業独自のアプリケーションやサービスに独自カスタマイズしたAIキャラクターを簡単に組み込むことができます。

一例として、今回は自分たちで作ったAIキャラをTwinRoomのAPI経由でSlackに登場させて、会話させてみます。

記事監修： わいけい（@yk_llm_gpt）

https://zenn.dev/spiralai/articles/8af7cbf526c2e1

最初に、ユーザーがAIに対するメッセージをSlackに投稿してから最終的にSlack上で返信が返ってくるまでの仕組みについて説明します。

まず、Slackにメッセージが投稿されてからのデータの流れについてです。

今回作成するアーキテクチャでは、ユーザーがSlackに投稿したメッセージがSlack Events APIからのイベントとしてAPI Gatewayを通じてLambda関数に転送され、そのLambda関数内でSlackのイベントを処理する流れになっています（下図）。

下記のアーキテクチャでは、API GatewayがHTTP(S)リクエストを受け取るエンドポイントとして機能し、それをLambda関数に渡しています。

Lambda関数はその後、受け取ったイベントからメッセージをTwinRoomに転送し、レスポンスをSlack APIにHTTPリクエストとして送信してSlackチャンネルにメッセージを投稿します。

それぞれのリソースの説明です。

Slack Events API: SlackのイベントAPIを使用して、ユーザーのメッセージやアクションを検知。

特定のイベント（メッセージがチャンネルに投稿されたときなど）に対して、SlackからHTTP POSTリクエストが送信される。

API Gateway: Slackからのリクエストを受け取るエンドポイントとして機能。

リクエストを受けた後、それをLambda関数にルーティング。

Lambda (Python): SlackからAPI Gatewayを通して送られてきたイベントを処理するサーバレス関数。

環境変数として指定されたトークンやAPIキーを使用して、セキュリティを担保する。

Slackのイベントに対する応答やTwinroom APIとの通信を行う。

Twinroom API: Lambda関数がキャラクターAIの処理や会話の継続に必要な情報を要求する外部API。

TwinRoom上でカスタマイズしたAIの応答をLambdaに返す

ざっくり以上です。 なお、パラメータやSecretsを厳重に管理するのであれば、AWS Secrets Manager または Parameter Storeを使います。(今回はクイックに実装するため、使っていません。)

SpiralAI株式会社が開発しているTwinRoomとは、さまざまなAIキャラクターを作成可能なLLMを使ったコミュニケーションプラットフォームです。

自然な会話や発声ができるAIキャラクターを簡単に作成でき、（今回は使用しませんが）テキストだけでなく、音声の入出力による対話も可能です。

TwinRoom上で作成したAIキャラは以下のクライアントで利用が可能です。

TwinRoom上で作成したAIキャラクターをLINE公式アカウントとして運用することができます。 LINEを介してリアルタイムでの対話が可能なAIキャラクターを設置し、友だちからのメッセージに返信するだけでなく自発的に話しかけることもできます。 採用向けや販促の公式LINE等でも活用可能です。

iPadなどのタブレットを使い、AIキャラクターとの音声通話によるコミュニケーションを可能にします。憧れのキャラクターや芸能人とタブレット通話ができる感覚です。

開発者や企業が独自のアプリケーションやサービスに、TwinRoomで発行したAPIキーを連携させることで独自AIキャラクターを組み込むことができます。 最も自由度が高い連携方法で、サービスにあわせて色々な使い方が可能です。 （今回の記事でもSlackにAIキャラを登場させるにあたり、このAPI連携機能を使っています。）

初期設定時にAIの話し方について自由記述で指定できますが、設定後もいつでも変更が可能です。 設定したら「保存」ボタンをクリックするのを忘れないようにしましょう。

「優しい口調」「クールでエレガント」「フランクでカジュアル」など自由に設定が可能です。 それぞれの設定ごとに「仕事疲れた」とメッセージを打ってみると、以下のように返答が変わるイメージです。

「優しい口調」 ... 大変ですね。お疲れ様です。リラックスする方法はありますか？

「クールでエレガント」...それは大変だったね。でも、あなたの頑張りはきっと報われるさ。少し休んで、リフレッシュする時間を持つことも大切だよ。

「フランクでカジュアル」...私もよくわかるよ。仕事って本当に疲れるよね。何をしているの？

これ以外にも、具体的な語尾を指定したり、どういったトピックに興味があるかなど、用途に合わせて設定をカスタマイズできます。

AIが保持する知識や情報は、例えば以下のような知識形式・応答形式で設定が可能です。 こういった知識を設定しておくと、関連した話題になった際にキャラクターがその知識を踏まえた回答をしてくれるようになります。よりキャラクターの深みが増し、ユーザーとのコネクションを強化することができます。

「知識」...「自動車免許を保持している。」 「QA応答」...「好きな食べ物はなんですか？」「カレーライスです。」

数が多い場合は、ファイル形式で一括アップロードすることも可能です。

※こちらはいわゆるRAGに反映されるための検索対象データになっています。 RAGについて知りたい方は下記の記事も参考にしてください。

https://zenn.dev/spiralai/articles/8af7cbf526c2e1

AIに話させてはいけない項目（NG項目）の設定

TwinRoomには、NG項目設定という機能があります。 これは例えば「AIキャラに性的トピックについては話させたくない」といったニーズに対応するためのものです。 これを使うと、性的な話を振られた時にAIキャラに「その話題には返答できません」といった対応を取らせることができます。

（NG項目の設定イメージ） 宗教について知りたいです。 政治について知りたいです。 性的な冗談を言いたいです。

NG項目はよく使うものがデフォルトで設定されていますが、ユーザー独自のカスタマイズも可能です。

左のメニュー「会話履歴」からLINE・Tablet・APIそれぞれのインターフェースごとに履歴の閲覧が可能です。

API連携の場合、会話履歴はsession_idとuser_idによって管理されます。

session_id … 会話を行っている部屋(一連のメッセージを束ねる箱のような存在で、グループチャットなども可)を管理するためのIDです。

user_id … メッセージを送ったユーザーのID。

（これらはいずれも、APIと連携したクライアント側で設定することが出来ます。）

また上記の履歴から「QAに追加」を押すと、ダイアログが出ます。 ここで「ユーザーにXXと答えていたのを次からはOOと答えるように修正する」ということができます。 QA追加すると、記憶情報としてAI設定に保存され、次回の会話以降に反映されます。

API連携サービスの場合、カスタマイズしたAIの応答を管理画面で簡易的にチェックできます。

チャット画面で気軽に試せるので、動作確認も簡単です。

ちょっとしたやり取りを行なってみます。

ここで、以下のような設定を加えてみます。

これを踏まえて、会話を続けてみます。

自然な形で情報を参照し、会話を続けてくれました！

後述しますが、このやりとりも会話履歴に保存されますので、履歴を踏まえてさらにキャラクターの返答をカスタマイズしていくことが可能です。

TwinRoomでAPIキーを発行

さて、ここからは実際にTwinRoomで独自カスタマイズしたAIキャラのAPIキーを発行し、LambdaおよびSlackに連携する流れを解説します。

キャラクター作成後、左側のメニューから「API -> APIキー管理」でAPIキーの作成をします。

ラベル名を設定して「追加」ボタンを押すと、APIキーが表示されます。

一度しか表示されないので、コピペして保存しておきます。

API配信設定で、公開設定を「公開する」にして「保存」します。

APIドキュメントがあるので、後ほどこれに沿ってリクエストしていきます。

https://{{TwinRoomAPIのリクエスト先ドメイン}}/api-docs

AIキャラに発話させるには、api/v1/messageにリクエストを送ります。

※APIのリクエスト先は、TwinRoom利用開始後に共有されます。

curl -X 'POST' \ 'http://{{TwinRoomAPIのリクエスト先ドメイン}}/api/v1/message' \ -H 'accept: application/json' \ -H 'api-key: skey-...' \ # apiキーを入れる -H 'Content-Type: application/json' \ -d '{ "content": "こんにちは" }'

このエンドポイントはSSEでの通信を行い、下記形式のレスポンスが返ってきます。

data: message_id='d3f3a7f7-1432-4617-b999-2e81eb62bdc9' msg='こんにちは！\n' msg_type='text' created_at='2024-04-01T03:17:19.330150' is_last=False voice=None is_error=False error_code=None data: message_id='d3f3a7f7-1432-4617-b999-2e81eb62bdc9' msg='キミは今日どうだい？\n' msg_type='text' created_at='2024-04-01T03:17:19.330150' is_last=True voice=None is_error=False error_code=None

次にSlackでのアプリケーション設定を行います。

まず、事前に対象とするチャンネルを作成しておきます。 (チャンネルIDはチャンネル詳細の表示から何度でも確認できるので、コピーは先でOKです。)

次に、Slackの設定画面 ( https://api.slack.com/apps )から「Create an App」ボタンを押して、Slack APIで新しいアプリを作成します。

今回は「From scratch」で作成しました。

アプリ名は分かりやすい名前を自由に名付けてください。

最後に「Create App」を押すとアプリができ、以下のようなアプリの設定画面になります。

必要なSlackのスコープ（権限）をアプリに追加します。

例えば、メンションを読み取ったり、発言を投稿する権限( chat:write など)を与えます。

左のメニューから「Basic Information」に進むと「Install your app」というリンクが出てきますので、これをクリックします。

こんな画面になります。Slackワークスペース側として、このアプリに権限を許可するかどうかの画面です。「Allow」を押してください。

アクセス可能な情報・実行できる内容を確認し、許可します。 (権限がない場合、ワークスペースの管理者に承認が必要ですので、依頼します。)

Slackから提供されるAPIトークン（Bot User OAuth Tokenなど）を保存しておきます。

TwinRoomでAIをカスタマイズ

次にTwinRoom側でAIを作成＆カスタマイズしていきましょう。

ログイン後のダッシュボード画面から、「新しいキャラクターを作成」します。

すると、以下のような設定画面になります。

必須はキャラクター名のみですが、アイコンやメモ等も設定できます。

その後、あなたの名前、キャラクターの一人称、二人称、話し方を入力します。

この辺りの設定がキャラクターとの会話に効いてきます。

その後のキャラクター固有設定は任意です。

NG設定もデフォルトで登録されているものがありますが、後ほど設定画面から変更が可能です。

作成後、AIに固有の情報を設定してみます。 この例では、このAIは朝5時に起きる習慣を持っている、という設定にしてみました。

これで一旦設定は完了です。とても簡単に設定できましたね！

「試す」機能でも設定が反映されていることが確認できました。 「朝何時に起きる？」という質問に対し、ちゃんと「朝５時におきている」と答えていますね。

Lambda用のPython関数を実装

さて、ここからはLambdaで使用するPython関数のコードを書いていきます。

main処理では、イベントのtypeに応じて適切なアクションを行います。 正常動作であれば、AIのレスポンスを取得してSlackチャネルに投稿します。

servide.pyではAIサービスからのレスポンスを解析し、メッセージを抽出します。

slack_post.pyでは、Slackからのリクエストを検証し、指定されたSlackチャネルにメッセージを投稿します。

import os import json import logging from slack_post import post_to_slack, verify_slack_request from service import get_ai_response logger = logging.getLogger() logger.setLevel(logging.INFO) SLACK_CHANNEL_ID = os.getenv("SLACK_CHANNEL_ID") def lambda_handler(event, context): if not verify_slack_request(event): logger.error("Invalid request signature") return {"statusCode": 400, "body": "Invalid request signature"} body = json.loads(event["body"]) if body.get("type") == "url_verification": return {"statusCode": 200, "body": json.dumps({"challenge": body["challenge"]})} if body.get("type") == "event_callback": process_event(body) return {"statusCode": 200, "body": "Event processed"} return {"statusCode": 200, "body": "No action taken"} def process_event(event_body): event = event_body.get("event", {}) if event.get("type") == "app_mention": handle_app_mention(event) def handle_app_mention(event): text = event.get("text", "") response = get_ai_response(text) post_to_slack(SLACK_CHANNEL_ID, response)

import os import requests import logging import re logger = logging.getLogger() logger.setLevel(logging.INFO) SERVICE_API_KEY = os.getenv("SERVICE_API_KEY") SERVICE_API_URL_BASE = os.getenv("SERVICE_API_URL_BASE") SERVICE_SESSION_ID = os.getenv("SERVICE_SESSION_ID") def extract_msg_from_response(response_text): pattern = re.compile(r"msg='([^']*)'") matches = pattern.findall(response_text) return [match.replace('\\n', '\n') for match in matches] def parse_sse_response(response_text): messages = [] for line in response_text.splitlines(): if line.startswith('data: '): message_data = line[6:] messages.append(message_data) return messages def get_ai_response(query): headers = { 'accept': 'application/json', 'api-key': SERVICE_API_KEY, 'Content-Type': 'application/json', } payload = { "session_id": SERVICE_SESSION_ID, # "user_id": user_id, 今回はユーザーを区別せずに同一チャンネル内を同一セッション管理する "content": query } service_endpoint = f"{SERVICE_API_URL_BASE}/api/v1/message" try: response = requests.post(service_endpoint, headers=headers, json=payload) parsed_messages = parse_sse_response(response.text) msgs = [] for message in parsed_messages: msgs.extend(extract_msg_from_response(message)) full_message = '\n'.join(msgs) return full_message if full_message else "No response" except requests.RequestException as e: logger.error("Service API request failed: %s", e) return "Error processing the AI response"

import os from slack_sdk import WebClient from slack_sdk.errors import SlackApiError from slack_sdk.signature import SignatureVerifier import logging logger = logging.getLogger() logger.setLevel(logging.INFO) SLACK_SIGNING_SECRET = os.getenv("SLACK_SIGNING_SECRET") SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN") slack_client = WebClient(token=SLACK_BOT_TOKEN) signature_verifier = SignatureVerifier(SLACK_SIGNING_SECRET) def post_to_slack(channel_id, message): try: response = slack_client.chat_postMessage(channel=channel_id, text=message) logger.info("Message posted to %s", response["channel"]) except SlackApiError as e: logger.error("Slack API Error: %s", e.response['error']) def verify_slack_request(event): body = event.get("body", "") headers = {key.lower(): value for key, value in event["headers"].items()} return signature_verifier.is_valid_request(body, headers)

aiohttp requests slack_sdk

※今回は省略していますが、Slackでは3秒以内に応答しないとwebhookが再送されてLambdaが重複起動してしまいます。 再送webhookには固有のヘッダー値が設定されているので、それを読み取って適切に処理するようにしましょう。

Lambda関数のデプロイ手順

CloudFormationやTerraformを使ったデプロイが一般的ですが、今回はコンソールからZIPアーカイブをアップロードする形でやってみようと思います。

まずは、Lambda関数で使用する全てのファイルが同じディレクトリ内にあることを確認します。

この例では requirements.txt、service.py、main.py、slack_post.py が含まれます。

Lambda関数で外部ライブラリ( requests など )を使用する場合、これらの依存関係を含める必要があります。

依存関係は、requirements.txtファイルにリストされています。

まず、Pythonの仮想環境を作成します。これにより、必要な依存関係をシステムのPython環境から分離して管理できます。

python3 -m venv lambda-build source lambda-build/bin/activate

次に、requirements.txtにリストされたライブラリをインストールします。

pip install -r requirements.txt

依存関係がインストールされたら、それらとLambda関数のコードを含むZIPファイルを作成します。

以下のコマンドは、依存関係がインストールされたsite-packagesディレクトリと、Lambda関数のコードファイルを現在のディレクトリにZIPアーカイブとしてパッケージする方法を示しています。 (今回Python 3.9環境でパッケージ