Amazon Connectでお問い合わせ内容をWhisper APIで文字起こしし、ChatGPTで要約して音声出力してみた（一次対応の無人化）
はじめに
Amazon Connectを使用して、お問い合わせ内容をOpenAIのWhisper APIで文字起こしとChatGPTで要約し、通話中に音声出力する方法をまとめました。
Connectで無人対応の場合、顧客からの発話を聞き取る方法としては、チャットボットサービスであるAmazon Lexもしくは、Kinesis Video Stream(KVS)で音声のストリーミングなどがあります。
Amazon Lexを利用する場合は、１度に15秒以上は聞き取ることができない点や文字起こしにはAmazon Transcribeを利用する制約があります。
今回は、文字起こしにWhisper APIを利用し、ChatGPTで要約した内容をConnectで音声出力させてみました。
利用用途としては、顧客の一次対応を無人で行う場合、ユーザーからのお問い合わせ内容の確認に使います。
確認後は、下記の記事のように、担当者に振り分けしたり、AIチャットボットで回答させたりすることができます。
構成
システムの流れは次の通りです。
- コンタクトフロー内で「メディアストリーミングの開始」ブロックを使って、KVSへの音声のストリーミングを開始します。顧客はお問い合わせ内容を伝えます。
- 「顧客の入力を保存する」ブロックで、顧客が特定の番号を押すと、ストリーミングを終了します。
- 「AWS Lambda関数を呼び出す」ブロックを使い、LambdaでKVSからデータを取得します。
- Lambdaで取得したデータをWAV形式に変換し、Whisper APIで文字起こしとChatGPTで要約を行います
- 「プロンプトの再生」ブロックでお問い合わせ内容の要約を音声出力します。
今回のフローは、音声出力までにしていますが、その後は先程の参考ブログの通りにフローを追加することで、担当者に振り分けしたり、AIチャットボットで回答させたりすることができます。
Lambdaの作成
Lambdaで利用するOpenAIアカウントAPIキーの発行方法やOpenAIが提供するPython向けのライブラリをLambdaにアップロードする方法は下記の記事を参考ください。
ランタイムPython 3.11を選択し、Lambda関数を作成します。
作成後、下記の設定を行います
- タイムアウトは3秒から20秒に変更しました。
- 環境変数では、キーはAPI_Key、値はAPIキーの値を入力
- Lambdaのレイヤーを追加します
- IAMポリシーに
AmazonKinesisVideoStreamsReadOnlyAccessを追加します。
コードは以下の通りです。
import boto3, os, struct, json, openai
from ebmlite import loadSchema
from enum import Enum
from datetime import datetime
from botocore.config import Config
from decimal import Decimal
openai.api_key = os.environ['API_Key']
def decimal_to_int(obj):
if isinstance(obj, Decimal):
return int(obj)
class Mkv(Enum):
SEGMENT = 0x18538067
CLUSTER = 0x1F43B675
SIMPLEBLOCK = 0xA3
class Ebml(Enum):
EBML = 0x1A45DFA3
class KVSParser:
def __init__(self, media_content):
self.__stream = media_content['Payload']
self.__schema = loadSchema('matroska.xml')
self.__buffer = bytearray()
@property
def fragments(self):
return [fragment for chunk in self.__stream if (fragment := self.__parse(chunk))]
def __parse(self, chunk):
self.__buffer.extend(chunk)
header_elements = [e for e in self.__schema.loads(self.__buffer) if e.id == Ebml.EBML.value]
if header_elements:
fragment_dom = self.__schema.loads(self.__buffer[:header_elements[0].offset])
self.__buffer = self.__buffer[header_elements[0].offset:]
return fragment_dom
def get_simple_blocks(media_content):
parser = KVSParser(media_content)
return [b.value for document in parser.fragments for b in
next(filter(lambda c: c.id == Mkv.CLUSTER.value,
next(filter(lambda s: s.id == Mkv.SEGMENT.value, document))))
if b.id == Mkv.SIMPLEBLOCK.value]
def create_audio_sample(simple_blocks, margin=4):
position = 0
total_length = sum(len(block) - margin for block in simple_blocks)
combined_samples = bytearray(total_length)
for block in simple_blocks:
temp = block[margin:]
combined_samples[position:position+len(temp)] = temp
position += len(temp)
return combined_samples
def convert_bytearray_to_wav(samples):
length = len(samples)
channel = 1
bit_par_sample = 16
format_code = 1
sample_rate = 8000
header_size = 44
wav = bytearray(header_size + length)
wav[0:4] = b'RIFF'
wav[4:8] = struct.pack('<I', 36 + length)
wav[8:12] = b'WAVE'
wav[12:16] = b'fmt '
wav[16:20] = struct.pack('<I', 16)
wav[20:22] = struct.pack('<H', format_code)
wav[22:24] = struct.pack('<H', channel)
wav[24:28] = struct.pack('<I', sample_rate)
wav[28:32] = struct.pack('<I', sample_rate * channel * bit_par_sample // 8)
wav[32:34] = struct.pack('<H', channel * bit_par_sample // 8)
wav[34:36] = struct.pack('<H', bit_par_sample)
wav[36:40] = b'data'
wav[40:44] = struct.pack('<I', length)
wav[44:] = samples
return wav
def create_kvs_client():
region_name = 'ap-northeast-1'
return boto3.client('kinesisvideo', region_name=region_name)
def create_archive_media_client(ep):
region_name = 'ap-northeast-1'
return boto3.client('kinesis-video-archived-media', endpoint_url=ep, config=Config(region_name=region_name))
def get_media_data(arn, start_timestamp, end_timestamp):
kvs_client = create_kvs_client()
list_frags_ep = kvs_client.get_data_endpoint(StreamARN=arn, APIName='LIST_FRAGMENTS')['DataEndpoint']
list_frags_client = create_archive_media_client(list_frags_ep)
fragment_list = list_frags_client.list_fragments(
StreamARN = arn,
FragmentSelector = {
'FragmentSelectorType': 'PRODUCER_TIMESTAMP',
'TimestampRange': {'StartTimestamp': datetime.fromtimestamp(start_timestamp), 'EndTimestamp': datetime.fromtimestamp(end_timestamp)}
}
)
sorted_fragments = sorted(fragment_list['Fragments'], key = lambda fragment: fragment['ProducerTimestamp'])
fragment_number_array = [fragment['FragmentNumber'] for fragment in sorted_fragments]
get_media_ep = kvs_client.get_data_endpoint(StreamARN=arn, APIName='GET_MEDIA_FOR_FRAGMENT_LIST')['DataEndpoint']
get_media_client = create_archive_media_client(get_media_ep)
media = get_media_client.get_media_for_fragment_list(StreamARN = arn, Fragments = fragment_number_array)
return media
def save_audio_to_tmp(wav_audio, filename='output.wav'):
with open(f"/tmp/{filename}", 'wb') as out_file:
out_file.write(wav_audio)
def get_inquiry_summary(inquiry_content):
input_text = """
## 役割
あなたはコールセンターの担当者です。ルールに遵守し、顧客のお問い合わせ内容を確認して下さい。
## ルール
1. お問い合わせに対して、要約してください。
2. 要約内容をお伝えしてください。伝え方は、顧客に確認する言い方にしてください。
例. パスワードが分からないということですね。
3. 顧客に伝えるとき、最初に「要約すると、」などの枕言葉は不要です。確認だけして下さい。
"""
inquiry_text = inquiry_content
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo-1106",
messages=[
{"role": "system", "content": input_text},
{"role": "user", "content": inquiry_text}
],
temperature=0,
)
print("Received get_chatgpt_response:" + json.dumps(response,default=decimal_to_int, ensure_ascii=False))
return response["choices"][0]["message"]["content"]
def lambda_handler(event, context):
print('Received event:' + json.dumps(event,default=decimal_to_int, ensure_ascii=False))
media_streams = event['Details']['ContactData']['MediaStreams']['Customer']['Audio']
stream_arn = media_streams['StreamARN']
start_time = float(media_streams['StartTimestamp']) / 1000
end_time = float(media_streams['StopTimestamp']) / 1000
combined_samples = create_audio_sample(
get_simple_blocks(get_media_data(stream_arn, start_time, end_time)))
wav_audio = convert_bytearray_to_wav(combined_samples)
save_audio_to_tmp(wav_audio)
with open("/tmp/output.wav", "rb") as audio_file:
transcript = openai.Audio.transcribe("whisper-1", audio_file)
transcript = str(transcript["text"])
print('Transcript result:' + transcript)
inquiry_summary = get_inquiry_summary(transcript)
print('Inquiry_summary result:' + inquiry_summary)
return {
"transcript": inquiry_summary
}
最後に、お問い合わせの要約を値として、キーをtranscriptとしたJSON形式でreturnしています。
後述しますが、この値は、コンタクフローにおいて、Lambdaを呼び出したブロックの次のブロックで音声出力する際に利用します。
モデルは、記事執筆時の最新であるgpt-3.5-turbo-1106を利用しています。
コンタクフロー
Connectのコンタクフローは、下記の通りです。
Lambdaを呼び出したブロックの後に、「プロンプトの再生」で下記の通りにテキストを入力しています。
transcriptは先程、Lambdaのレスポンスで設定したキーですので、$.External.transcriptによって、要約内容が音声出力されます。
$.External.transcript
あとは特に変わった点はありませんが、「顧客の入力を保存する」ブロックでは、読み上げるテキストは不要なので、句読点を入れています。
試してみた
実際に電話をかけて質問し、お問い合わせ内容が要約された状態で音声出力されるか確認します。
お問い合わせ内容は、５秒、20秒、70秒の３つで試して、Lambdaの実行時間も確認します。
録音時間5秒
お問い合わせ内容：「パスワードを忘れてしまったのですが、対応方法がわかりません。教えてください」文字起こし内容：「パスワードを忘れてしまったの ですが対応方法がわかりません 教えてください」音声出力（内容の要約）：「パスワードを忘れてしまったということですね。」
録音時間は5秒ほどで、10回ほど試したところ、Lambdaの実行時間は、2~4秒程度でした。
Lambdaで処理する内容としては以下です。
- KVSからメディアデータを取得し、WAV形式に変換する
- WAVファイルからWhisper APIをリクエストし、文字起こしする
- 文字起こした内容をChatGPTで要約する
これらを4~5秒で処理するのは早いなと思いつつ、ユーザーからすると4~5秒待つのは遅いと感じる方もいらっしゃいますね。
録音時20秒
お問い合わせ内容：「もしもし、私の名前はクラスメソッドです。いつもそちらのサービスをログインページからログインしているのですが、パスワードを忘れてしまいました。パスワードのリセットを試してみても、新しいパスワードがシステムに反映されないようです。どのような対応すればよいでしょうか」文字起こし内容：「私の名前はClassMethodです いつもそちらのサービスをログインページ からログインしているのですが パスワードを忘れてしまいました パスワードのリセットを試して みても 新しいパスワードがシステム に反映されないようです どのような 対応をすればいいのか」音声出力（内容の要約）：「パスワードをリセットしても新しいパスワードが反映されないということですね。」
録音時間は20秒で、10回ほど試したところ、Lambdaの実行時間は、4秒~5秒程度でした。
録音時間70秒
さらに長文の70秒の録音をしてみました。
ーーーお問い合わせ内容ここからーーー
もしもし、私の名前はクラスメソッドで、クラスメソッド株式会社に所属しています。お忙しいところ恐れ入りますが、伺いたいことが１点ございます。貴社のシステムのログインプロセスにおいて、パスワードに関する問題が発生しています。具体的には、ユーザーが正しいパスワードを入力しても、システムがそれを認識しないという状況です。この問題によって私たちの業務に大きな支障をきたしており、解決が急務となっています。さらに、パスワードのリセットを試みても、新しいパスワードがシステムに反映されないという別の問題も発生しています。これにより、ユーザーがシステムにアクセスできない時間長くなってしまい、業務効率に影響を及ぼしています。私たちは、これらの問題について、何か解決策があるかどうかをお聞きしたいしたいです。ご支援をいただければ幸いです。どのようなアドバイスも大歓迎です。お忙しい中、大変恐縮ですが、ご確認のほどよろしくお願いいたします。
ーーーここまでーーー
ーーー文字起こし内容ここからーーー
もしもし 私の名前はクラスメソッドで クラスメソッド株式会社に所属 しています お忙しいところを恐れ 入りますが 伺いたいことが1点 ございます 記者のシステムのログイン プロセスにおいて パスワードに関する 問題が発生しています 具体的には ユーザーが正しいパスワードを 入力しても システムがそれを認識 しないという状況です この問題 によって 私たちの業務に大きな 支障をきたしており 解決が急務 となっております さらに パスワード のリセットを試みても 新しいパスワード がシステムに反映されないという 別の問題も発生しています これ により ユーザーがシステムにアクセス できない時間が長くなってしまい 業務効率に影響を及ぼしています 私たちは これらの問題について 何か解決策があるかどうかをお聞き したいです ご支援をいただければ 幸いです どのようなアドバイス も大歓迎です お忙しい中 大変恐縮 ですが ご確認のほどよろしくお願いします
ーーーここまでーーー
ーーー音声出力（内容の要約）ここからーーー
記者のシステムのログインに関する問題が発生しており、具体的には正しいパスワードを入力してもシステムが認識しない状況と、パスワードのリセットを試みても新しいパスワードが反映されないという別の問題が発生しています。これにより業務に大きな支障をきたしており、解決が急務となっています。何か解決策があるかどうかをお聞きしたいです。
ーーーここまでーーー
録音時間70秒で、10回ほど試したところ、Lambdaの実行時間は、10~12秒程度でした。
注意
Connectのコンタクトフローで呼び出すLambdaのタイムアウトは最大8秒です。
そのため、今回の結果であれば、発話時間が20秒程度であれば、通話中に要約内容が音声出力できますが、録音時間が70秒の場合、Lambdaの実行時間が8秒を超えるため、音声出力できません。
最後に
Amazon Connectでお問い合わせ内容をWhisper APIで文字起こしとChatGPTで要約し、通話の中で音声出力してみました。
発話時間が20秒程度であれば、音声出力まで可能であることが分かりました。あくまでも今回の検証の場合です。
![[速報] OpenAIのビデオ生成AIモデル「Sora」が利用可能に](https://images.ctfassets.net/ct0aopd36mqt/wp-thumbnail-5285ca250d0503bc4e8ca44630ac36f2/a5da264e8e8b2da687153a6b18c0c8db/eyecatch_openai)
