Amazon Connect + Whisper + GPT-4 Turboで、発話から個人情報（名前、住所、生年月日）を正しく認識できるか試してみた

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]
    print("Received fragment_number_array:" + json.dumps(fragment_number_array,default=decimal_to_int, ensure_ascii=False))


    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_extract_utterances(utterances_content):
    input_text = """
    ## 役割
    あなたはコールセンターの担当者です。ルールに遵守し、顧客の発話から住所のみを出力してください。
    ## ルール
     1. 顧客から住所を含めたお問い合わせが文字起こしされます。
     2. 推測される住所の部分のみを返してください
     3. 住所は、都道府県名、市区町村、丁目、番地、号、建物名、階数を含めてください。
     4. 住所に使用する数字は、全てアラビア数字（1, 2, 3, ...）を使用してください。
    """
    
    inquiry_text = utterances_content

    response = openai.ChatCompletion.create(
        model="gpt-4-1106-preview",
        messages=[
            {"role": "system", "content": input_text},
            {"role": "user", "content": inquiry_text}
        ],
        temperature=0,
    )
    return response["choices"][0]["message"]["content"]

def transcribe_audio_file(file_path):
    with open(file_path, "rb") as audio_file:
        return openai.Audio.transcribe("whisper-1", audio_file)

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)

    transcript = transcribe_audio_file("/tmp/output.wav")
    transcript = str(transcript["text"])
    print("Transcript result:" + transcript)
    
    utterance_extraction = get_extract_utterances(transcript)
    print("Utterance_extraction result:" + utterance_extraction)

    return {
        "transcript": utterance_extraction
    }

住所は、$.External.transcript 、ですね。

    input_text = """
    ## 役割
    あなたはコールセンターの担当者です。ルールに遵守し、顧客の発話から住所のみを出力してください。
    ## ルール
     1. 顧客から住所を含めたお問い合わせが文字起こしされます。
     2. 推測される住所の部分のみを返してください
     3. 住所は、都道府県名、市区町村、丁目、番地、号、建物名、階数を含めてください。
     4. 住所に使用する数字は、全てアラビア数字（1, 2, 3, ...）を使用してください。
    """

    input_text = """
    ## 役割
    あなたはコールセンターの担当者です。ルールに遵守し、顧客の発話から人名のみを出力してください。
    ## ルール
     1. 顧客から人名を含めたお問い合わせが文字起こしされます。
     2. 推測される人名を部分のみを返してください
     3. 人名はカタカナに変換してください
    """

    input_text = """
    ## 役割
    あなたはコールセンターの担当者です。ルールに遵守し、顧客の発話から会員番号である英字のみを出力してください。
    ## ルール
     1. 会員番号は、英字で構成されています
     2. 顧客から会員番号(英字)を含めたお問い合わせが文字起こしされます。
     3. 会員番号(英字)部分のみを返してください
     2. 英字を1文字ずつカンマで区切ってください
    """

    return {
        "transcript": utterance_extraction,
        "correct_value": utterance_extraction.replace(",", "")
    }

    input_text = """
    ## 役割
    あなたはコールセンターの担当者です。ルールに遵守し、顧客の発話から数字のみを出力してください。
    ## ルール
    1. 漢数字は、次のようにアラビア数字に変換してください
      - 〇: 0
      - 一: 1
      - 二: 2
      - 三: 3
      - 四: 4
      - 五: 5
      - 六: 6
      - 七: 7
      - 八: 8
      - 九: 9
    2. アラビア数字を1桁ずつカンマで区切ってください
      - 例：〇,〇,〇,〇,〇,〇
    3. 「以下のように数字を変換しました」などの文言は不要です。
    """

    input_text = """
    ## 役割
    あなたはコールセンターの担当者です。ルールに遵守し、顧客の発話から誕生日のみを出力してください。
    ## ルール
    1. 漢数字は、次のようにアラビア数字に変換してください
      - 〇: 0
      - 一: 1
      - 二: 2
      - 三: 3
      - 四: 4
      - 五: 5
      - 六: 6
      - 七: 7
      - 八: 8
      - 九: 9
    2. 「以下のように変換しました」などの文言は不要です。
    3. 誕生日の日付を以下のフォーマットで答えてください。
     - 例：〇〇〇〇年〇〇月〇〇日
    """