Amazon Connectでお問い合わせ内容をWhisper APIで文字起こしし、ChatGPTで要約して音声出力してみた（一次対応の無人化）

平井裕二

2023.11.09

はじめに

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秒程度であれば、音声出力まで可能であることが分かりました。あくまでも今回の検証の場合です。

Amazon Connectでお問い合わせ内容をWhisper APIで文字起こしし、ChatGPTで要約して音声出力してみた（一次対応の無人化）

はじめに

構成

Lambdaの作成

コンタクフロー