[ChatGPT] gpt-3.5-turbo でコップの音を聞き分けてみました〜Audioデータは、128バイトの文字列に変換してプロンプトに入れてます

SIN

2023.03.22

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

1 はじめに

CX 事業本部のデリバリー部の平内（SIN）です。

今回は、gpt-3.5-turbo を使用して、音を判別してみました。

最初に動作している様子です。

「私のコップ(my_cup)」をスプーンで叩いた音と、「別のコップ(unknown)」の音を聞いてChatGPTが、どちらを鳴らしたのか判別しています。

2 コップの音

２種類のコップを叩いた時の波形です。見た目ですが、結構、違いが出ていたので、「これぐらいなら判別できるかもしれない！」と思ったのが、今回のスタートです。

my_cup

unknown

3 データの正規化及び圧縮

Audioデータは、そのままでは、ちょっと無理があると思ったので、正規化で、0 から 255 の整数へ変換し、64バイトまで圧縮しています。また、先頭の無音部分をトリミングして、音の立ち上がりを揃えました。

正規化　(0..255 の整数で表現する)
- -1 〜 1 への正規化
- 絶対値への変換 0..1 の表現に変換する
- 256 倍して、0..255 の表現に変換する
- 整数化 float -> int
Compress 圧縮 1/256
Trim 先頭の無音部分の削除

以下の図は、データを変換していく過程の波形です。詳しくは、後述のコードで、 create_text(frames):をご参照ください。

なお、圧縮は、何種類かやってみたのですが、元の波形が認識できる程度ということで、1/256 としました。

4 文字列化

圧縮したデータは、プロンプトで使用できるよう 16 進数の文字列とし、コップを叩くと128バイトの文字列が生成されます。

# 16bit の int データを 00-FF の 2 バイトで表現
def to_string(array):
    str = ""
    for n in range(array.size):
        str += "{:02X}".format(array[n])
    return str

当初、「FF,FF,01,02,・・・」のように、カンマで区切って試していたのですが、トークンが　多くなってしまうので、データ間の区切りは、「無し」としました。「FFFF0102・・・」

5 プロンプト

ChatGPT に送るプロンプトは、以下のような形式となっています。

最初に、"system"で、データ形式と判定要領を定義し、その後は、事前にサンプリングしたデータで、"my_cup"と"unknown"を「判定例」を 3 回づつ繰り返しています。

「判定例」の繰り返しが、3回で充分というのは、ちょっと驚きでした。

[
    {
        "role": "system",
        "content": "From the input that expresses the change in sound in a hexadecimal number of 2 byte, Look at similarity and reply with [my_cup] or [unknown]."
    },
    { "role": "user", "content": "{既存のmy_cupのデータ}" },
    { "role": "assistant", "content": "my_cup" },
    ・・・略・・・
    { "role": "user", "content": "{既存のunknownのデータ}" },
    { "role": "assistant", "content": "unknown" },
    ・・・略・・・
   { "role": "user", "content": "{今回の録音データ}" }
]

6 実装

作成したコードです。

import os
import numpy as np
import pyaudio
import openai

openai.api_key = os.environ["OPENAI_API_KEY"]

# プロンプトの生成
def create_prompt(data):
    order = "From the input that expresses the change in sound in a hexadecimal number of 2 byte, Look at similarity and reply with [my_cup] or [unknown]."
    my_cup_list = [
        "FFFFDBB4B7897C685C504C3F3D312D282620201B1A1817141512121010100F0E0E0D0C0C0B0C0B0B0A0A0909080908080707070706060606060605050505050505040404040404030303030303020302000000000000",
        "FFFFFFC0B58B7A695C4E50403D302D2826211F1C1A1817161515121210100E0F0D0E0C0C0B0C0A0B0A0A0909090909080807070707070706060606050505050505040404040404030303030303030300000000000000",
        "FFFFD1BCAE86807995725F4C4D3B3629281C201C1815150E120E0E0C0B0B0A09090908080706060605050505040404050304030402030203020302020202020202020102010201010101010101010000000000000000",
    ]
    unknown_list = [
        "FF722D130E070706050405030302010000000001000100000001000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
        "FF651D100A060404030203010101000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
        "FF8727150C050404040203010101000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
    ]

    role = "role"
    content = "content"
    system = "system"
    user = "user"
    assistant = "assistant"

    prompt = []
    prompt.append({role: system, content: order})
    for line in my_cup_list:
        prompt.append({role: user, content: line})
        prompt.append({role: assistant, content: "my_cup"})
    for line in unknown_list:
        prompt.append({role: user, content: line})
        prompt.append({role: assistant, content: "unknown"})
    prompt.append({role: user, content: data})
    return prompt


# 音量が敷居値を超えているかどうかの判定
def is_valid(data, threshold):
    np_array = np.frombuffer(data, dtype=np.int16)  # データは、16bit単位で判定する
    if np.amax(np_array) > threshold:
        return True
    return False


# データの圧縮
def compress(array, split):
    result = np.empty(0, dtype=np.uint16)
    size = np.size(array)
    # データを分割する
    sub_arrays = np.array_split(array, len(array) / split)
    for i in range(int(size / split)):
        # 分割した中で、最大値を使用する
        max_val = np.max(sub_arrays[i])
        result = np.append(result, max_val)
    return result


# 先頭の無音部分の削除
def trim(array):
    max = array.size
    result = np.empty(0, dtype=np.uint16)
    trim = True
    for n in range(max):
        if trim and array[n] > 10:
            trim = False
        if trim == False:
            result = np.append(result, array[n])
    for n in range(max - result.size):
        result = np.append(result, 0)
    return result


# 16進文字列への変換
def to_string(array):
    str = ""
    for n in range(array.size):
        str += "{:02X}".format(array[n])
    return str


# 録音データからテキストデータの生成する
def create_text(frames):
    # numpy arrayへの変換
    np_array = np.frombuffer(b"".join(frames), dtype=np.int16)
    # -1 〜 1への正規化
    np_array = (np_array) / (np.max(np_array) - np.mean(np_array))
    # 絶対値への変換 0..1の表現に変換する
    np_array = np.abs(np_array)
    # 256倍して、0..256の表現に変換する
    np_array = np_array * 256
    # 範囲制限 データを規定範囲内(0..255)に収める
    np_array = np.clip(np_array, 0, 255)
    # 整数化 float -> int
    np_array = np.array(np_array, dtype="int")
    # データの圧縮 1/256
    np_array = compress(np_array, 256)
    # 先頭の無音部分の削除
    np_array = trim(np_array)
    # 16進文字列に変換
    return to_string(np_array)


# Audio ストリームの初期化
def init_stream(py_audio, stream, CHUNK, SAMPLE_RATE, FORMAT, CHANNELS):
    if stream is not None:
        stream.stop_stream()
        stream.close()
    return py_audio.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=SAMPLE_RATE,
        input=True,
        frames_per_buffer=CHUNK,
    )


def main():

    FORMAT = pyaudio.paInt16
    CHANNELS = 1
    SAMPLE_RATE = 44100
    CHUNK = int(SAMPLE_RATE / 40)  # ループ毎の時間は、 SAMPLE_RATE/40 = 25ms
    threshold = 2000  # MacOSのマイク音量に依存する

    py_audio = pyaudio.PyAudio()
    stream = None

    stream = init_stream(py_audio, stream, CHUNK, SAMPLE_RATE, FORMAT, CHANNELS)
    buffers = []

    print("start.")

    while True:
        data = stream.read(CHUNK)
        buffers.append(data)

        # 一定の音量を超えたら、処理を開始する
        if is_valid(data, threshold):
            frames = []
            # 25ms前のデータから使用する
            frames.append(buffers[len(buffers) - 1])
            for _ in range(14):  # 25msec * 14 = 350msec 録音する
                frames.append(stream.read(CHUNK))
            # 録音データからテキストデータの生成する
            str = create_text(frames)

            print("gpt-3.5-turbo:", end=" ")

            prompt = create_prompt(str)
            try:
                response = openai.ChatCompletion.create(
                    model="gpt-3.5-turbo", messages=prompt, temperature=0
                )
                print(response.choices[0]["message"]["content"].strip())
            except Exception as e:
                print("Exception:", e.args)

            stream = init_stream(py_audio, stream, CHUNK, SAMPLE_RATE, FORMAT, CHANNELS)
            buffers = []


if __name__ == "__main__":
    main()