AWSサービス(Lambda + API Gateway + AWS Polly)とLINE、ChatGPTとを連携させて、LINEで質問した内容の返信を音声で受け取るシステムを作った時のメモを残します。
最終目標として、RVC(Retrieval-based-Voice-Conversion)やbark(以下)といった技術を使用して、流ちょうなしゃべり方、特定のキャラクターのしゃべり方への音声変換して使われることを目標としています。
とりあえずは、その前段階の作成記録として、順々にまとめていきます。
github.com
目次
使用環境
・AWS Lambda ランタイム python3.9
・AWS API Gateway
・Amazon Polly
・Amazon S3
・LINE (developers)
・ChatGPT API
実装
前準備[省略:別ページで紹介]
これらの手順は過去のブログ(以下)で説明をしているので、そちらをご参照ください。
具体的な作業は、AWS API Gateway + LINE API + ChatGPT APIを使うための準備になります。
dango-study.hatenablog.jp
dango-study.hatenablog.jp
Amazon Polly + Amazon S3を使用する環境設定
Amazon Polly はAWSのサービスの一つで、深層学習技術を使用し、人間の声のような音声を合成します。Text-to-Speech (TTS)の方法は他にも色々ありますが、他の方法をAWSで実装するよりも既存のものを使った方が安いし早いと思ったので、こちらのサービスを使いました。
今回は、このサービスにChatGPTの返答結果(Text)を入れて、音声ファイルを出力させます。
この音声ファイルをLINEで送る為に、一度どこかに保管が必要であり、その保管場所として、Amazon S3を使用します。
Amazon S3の設定について
- 各種設定についての特記事項
LINEにデータを渡すための方法として、パブリックアクセスができるような設定にしています。
S3コンソールを使用したバケット作成時に、パブリックアクセスのブロック設定の無効化に s3:PutBucketPublicAccessBlock 許可が必要となる為、IAMのポリシーに追加する必要がある。
- パブリックアクセスができるのかどうかの確認
試しに、buffer.mp3というファイルをアップロードして、そのファイルにアクセスできるかどうかを確認する。
ファイルのタブをクリックして"URLをコピー"を押すと、ファイルのアクセスパスを入手できる。
そのパスにブラウザでアクセスをした時に、ファイルをダウンロードをすることができれば、パブリックアクセス設定ができている。
Amazon Pollyの設定について
ロールを追加したら、あとはコードを書くだけ。
以下を参考にして、コードを作成する。
開始方法 - Amazon Polly | AWS
AWS Lambdaでコード実行(動作確認)
コード作成
Lambda関数では、以下コードを作成/実行する。
(こちらがフルのコードとなります)
import json
import urllib.request
import os
import openai
from linebot import LineBotApi
from linebot.models import TextSendMessage, AudioSendMessage
import boto3
from botocore.exceptions import ClientError
#######LINE Bot関係
# 環境変数からLINE Botのチャネルアクセストークンを取得
LINE_CHANNEL_ACCESS_TOKEN = os.environ['LINE_CHANNEL_ACCESS_TOKEN']
# チャネルアクセストークンを使用して、LINE BotのAPIインスタンスを作成
LINE_BOT_API = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)
#######ChatGPT
openai.api_key = os.environ["Openai_ACCESS_TOKEN"]
#Token数
MAX_TOKENS = 150
#モデルを指定
MODEL_ENGINE = "gpt-3.5-turbo"
#######Dynamo DB
def get_system_prompts():
try:
# Reverse messages
system_prompts = {"role": "system", "content": "質問を返してください"}
# Join the lists
return system_prompts
except Exception as e:
# Raise the exception
raise e
def use_chatgpt(messages):
try:
response = openai.ChatCompletion.create(
model=MODEL_ENGINE,
messages=messages,
max_tokens=MAX_TOKENS
)
answer = response.__dict__["_previous"]["choices"][0]["message"]["content"]
return answer
except Exception as e:
# Raise the exception
raise e
def reply_message_for_line(reply_token, completed_text, original_content_url):
#LINEで入力されたテキストは ['events’][0]['message’]['text’] に入っています。
#また、受信したメッセージに対してリプライを行うには、replyTokenの値を使用します。
try:
# Reply the message using the LineBotApi instance
print(original_content_url)
#テキストで返す場合は、以下を使用する
#LINE_BOT_API.reply_message(reply_token, TextSendMessage(text=completed_text))
#音声で返す場合は、以下を使用する
LINE_BOT_API.reply_message(reply_token, AudioSendMessage(original_content_url=original_content_url, duration=30000))
except Exception as e:
# Raise the exception
raise e
def create_completed_text(line_user_id, prompt_text):
# Create the list of a current user prompt
current_prompts = {"role": "user", "content": prompt_text}
system_prompts = get_system_prompts()
messages = [system_prompts] + [current_prompts]
completed_text = use_chatgpt(messages)
return completed_text
def lambda_handler(event, context):
try:
if 1:
#S3_Polly
session = boto3.Session()
polly = boto3.client("polly")
s3 = session.resource('s3')
bucket = s3.Bucket("polly.speech")
# Parse the event body as a JSON object
#if json.loads(event)["events"][0]["message"]["type"] == "text":# textの場合のみ対応
if event["events"][0]["message"]["type"] == "text":# textの場合のみ対応
reply_token = event["events"][0]['replyToken']# リプライ用トークン
prompt_text = event["events"][0]["message"]["text"]# 受信メッセージ
line_user_id = event["events"][0]['source']['userId']# userID
# Check if the event is a message type and is of text type
if prompt_text is None or line_user_id is None or reply_token is None:
raise Exception('Elements of the event body are not found.')
# Create the completed text by Chat-GPT 3.5 turbo
completed_text, history_text, conversation_num = create_completed_text(line_user_id, prompt_text)
print("completed_text", completed_text)
#Polly output
if 1:
response = polly.synthesize_speech(
Text=completed_text,
#Engine="neural",
OutputFormat="mp3",
VoiceId="Mizuki")
with closing(response["AudioStream"]) as stream:
bucket.put_object(Key="<任意のファイル名>.mp3", Body=stream.read(), ACL="public-read")
#バケット内の音声
original_content_url=<ここにS3に保管したmp3ファイルのアクセスパスを入れる>
# Reply the message using the LineBotApi instance
reply_message_for_line(reply_token, completed_text, original_content_url)
except Exception as e:
# Log the error
# Return 200 even when an error occurs as mentioned in Line API documentation
return {'statusCode': 200, 'body': json.dumps(f'Exception occurred: {e}')}
# Return a success message if the reply was sent successfully
return {'statusCode': 200, 'body': json.dumps('Reply ended normally.')}
Amazon Pollyに渡す引数設定や、S3からファイルを取り出すためのコードは、AWSのドキュメントが一番詳しく書いてあるが、簡単に以下説明を追加した。
#Polly output
response = polly.synthesize_speech(
Text=completed_text,
OutputFormat="mp3",#出力ファイル形式設定
VoiceId="Mizuki") #出力ボイスの選択
with closing(response["AudioStream"]) as stream:
bucket.put_object(Key="<任意のファイル名>.mp3", Body=stream.read(), ACL="public-read") #キーとファイルの組み合わせで格納する。ACL="public-read"とすることで、更新したファイルのパブリックアクセスできるようになる
#バケット内の音声
original_content_url=<ここにS3に保管したmp3ファイルのアクセスパスを入れる>#保存したファイルのアクセスパス
実行結果
ブログなので、結果を表現しずらいですが、、あいさつに対して、ChatGPTが適切な挨拶を考えて、その文言が音声として帰ってきます。
実行結果
