LoginSignup
8
6
音声認識APIを使ってみよう！
@TomK

音声認識API AmiVoiceAPIとChatGPTを使ってリアルタイム音声処理を試してみた

Posted at

構想中なので深くは言えませんが、AIによる音声認識とChatGPTによる言語処理を組み合わせて、アプリを作りたいと思っています。

そう、ちょうど思っていたんですよ！そんなときに記事投稿キャンペーンが。AmiVoiceさんありがとうございますクーポンコード使わせていただきました:cherries: もっとくれてもいいんですよ！！

すみません興奮しました。「AIによる音声認識とChatGPTによる言語処理を組み合わせる」というアイデアは十分に実現可能と思われますが(というかChatGPTのアプリにその機能ある)、レスポンスの速度や精度について検証したかったので簡単なPoCを行いました。音声の取り扱い方や、WebSocketの使い方などの初歩的なところでいくつか躓いたので、ここでシェアします。

PoC概要

Untitled_Artwork 4.png

やったこと

今回はPythonを使用してPoC用のプログラムを作ります。
次のような実装を組み込みました。

  • ① PyAudioを使って音声入力を受け取る
  • ②③ 音声データをAmiVoice APIに送り、音声認識結果をテキストで受け取る
  • ④ 受け取ったテキストが一定量(50文字くらい)たまったら、ChatGPTに送ってテキスト処理させる
  • ⑤⑥ 要約などにテキスト処理結果を受け取ったら、画面に表示する
  • 一連の処理は断続的に行い、出力結果を待たずして次の発話を行える
    ...ちょうどこういうことやりたいと思っていたのは本当で、もともとWhisperを使ってお試し開発していたという経緯があります。Whisper使用版では、15秒程度で強制的に音声を区切って断続的に処理してみました。AmiVoice APIでは、WebSocketを使ってリアルタイムに音声を送信できるため、連続的な処理ができそうです。結果は発話区間ごとに取り出すことができるため、「15秒程度」という恣意的な時間で区切らずに済む！

はじめてのWebSocket & 音声認識でハマったところ

今回WebSocketも音声処理もChatGPTのAPIも初めてチャレンジしました。おそらくめちゃくちゃ基本的(というかあるある？) な錯誤をたくさん起こしましたが、ChatGPTが手取り足取り教えてくれたのでなんとか目的のものを作って検証することができました。たくさんハマったので、その箇所を皆様にシェアします。

WebSocketを使った音声処理API呼び出しの流れ

WebSocketとは何かについての正確な説明は私にはできないので各自ググってください。

WebSocketを使って、Pythonアプリ(クライアント側)とAmiVoice API(サーバー側) の間に双方向通信を確立します。WebSocketによる通信ができるライブラリはいくつかありますが、今回は websockets を使用します。ちなみに当初別のものをつかっていて、非同期対応していなくて詰みました。。

接続を確立したら、 ws.send() によってメッセージを送るか、 ws.recv() によってメッセージを受け取ります。今回は送受信するメッセージのフォーマットが定められているので、それに従って送ります。音声認識のフォーマット情報などを、 s コマンドのフォーマット に合わせて送信します。

その後はAmiVoice API側のタイミングでメッセージが送られてくるので、それを受け取れるような記述をします。イベントループを作り、メッセージが来るまで待機し、来たら適切に処理します。

PyAudioを使った音声ストリームの処理

入力ラインが複数ある

通常デバイスには音声の入出力機器が複数備わっています。この基本的な事実に気づかずちょっとハマりました(厳密には知ってはいたけど関係ないと思ってしまった) Macで開発しているとき、近くにあるiPhoneのマイクも使うことができますがこれが罠！

入力デバイスの一覧を表示して、適切なものを選びましょう。

list_pyaudio.py
import pyaudio

# PyAudioのインスタンスを作成
pa = pyaudio.PyAudio()

# 使用可能な音声チャンネルの一覧を表示
for i in range(pa.get_device_count()):
    device_info = pa.get_device_info_by_index(i)
    print(device_info)

$ python list_pyaudio.py 
{'index': 0, 'structVersion': 2, 'name': 'BenQ GL2580', 'hostApi': 0, 'maxInputChannels': 0, 'maxOutputChannels': 2, 'defaultLowInputLatency': 0.01, 'defaultLowOutputLatency': 0.009833333333333333, 'defaultHighInputLatency': 0.1, 'defaultHighOutputLatency': 0.019166666666666665, 'defaultSampleRate': 48000.0}
{'index': 1, 'structVersion': 2, 'name': '\u200e俺のiPhoneのマイク', 'hostApi': 0, 'maxInputChannels': 1, 'maxOutputChannels': 0, 'defaultLowInputLatency': 0.12841666666666668, 'defaultLowOutputLatency': 0.01, 'defaultHighInputLatency': 0.13775, 'defaultHighOutputLatency': 0.1, 'defaultSampleRate': 48000.0}
{'index': 2, 'structVersion': 2, 'name': 'MacBook Proのマイク', 'hostApi': 0, 'maxInputChannels': 1, 'maxOutputChannels': 0, 'defaultLowInputLatency': 0.03285416666666666, 'defaultLowOutputLatency': 0.01, 'defaultHighInputLatency': 0.0421875, 'defaultHighOutputLatency': 0.1, 'defaultSampleRate': 48000.0}
{'index': 3, 'structVersion': 2, 'name': 'MacBook Proのスピーカー', 'hostApi': 0, 'maxInputChannels': 0, 'maxOutputChannels': 2, 'defaultLowInputLatency': 0.01, 'defaultLowOutputLatency': 0.018708333333333334, 'defaultHighInputLatency': 0.1, 'defaultHighOutputLatency': 0.028041666666666666, 'defaultSampleRate': 48000.0}
{'index': 4, 'structVersion': 2, 'name': 'Microsoft Teams Audio', 'hostApi': 0, 'maxInputChannels': 2, 'maxOutputChannels': 2, 'defaultLowInputLatency': 0.01, 'defaultLowOutputLatency': 0.0013333333333333333, 'defaultHighInputLatency': 0.1, 'defaultHighOutputLatency': 0.010666666666666666, 'defaultSampleRate': 48000.0}

今回は検証用なので事前に選びますが、普通はデバイスを選べるような機能を作るでしょう。

音声はストリームにたまる。オーバーフローに気をつけて

pa.open() すると、音声ストリームが開きます。開いた音声ストリームには音声データがどんどんたまっていくので、これを pa.read() により定期的にくみ出していきます。当初ほどほどのペースでくみ出せばいいかなーと思っていたのですがそういうものではないようで、いいペースでくみ出さないとあふれるみたいです。

蛇口の下にバケツがあり、あなたはバケツからコップで水をくみ出します。蛇口を開くと、バケツに水が入り始めます。蛇口から水が出るのと同じくらいのペースでコップから水をくみ出しましょう。早すぎるとバケツは空っぽになります(別に構いません)が、遅すぎるとバケツがいっぱいになりあふれてしまいます(これはダメ)。　・・・意味のある比喩なのかはわかりませんがそんな感じ。

AmiVoice API WebSocketのイベント処理

AmiVoice API(WebSocketインターフェース)の中核となる処理です。

s コマンド送信&イベント処理

AmiVoice APIとWebSocketで接続したら、 s コマンドを送信することで音声認識のオプションを渡します。

async def on_open(ws):
    """WebSocketのイベントハンドラ"""
    logger.info("WebSocket Opened")
    audio_format = "lsb16k"
    grammer_file_name="-a-general"
    authorization = "ひみつ"
    await ws.send(f"s {audio_format} {grammer_file_name} authorization={authorization}")

s コマンドのフォーマット詳細は公式ドキュメントを参照してください。特に音声フォーマットやサンプリングレートはよく確かめて記載しましょう！

  • サンプリングレート: PyAudioの初期化時に設定するサンプリングレートと合わせます。音声認識には16kHzあれば十分ということなので、今回は素直にそれでいきます。
  • ヘッダの有無: PyAudioのストリームから取り出したデータにはヘッダはありません。
  • リトルエンディアンとビッグエンディアン: 動作するシステムにより異なります。Copilotが判別方法を教えてくれたのでこれで確認して設定。
import sys

if sys.byteorder == "little":
    print("System is Little Endian")
else:
    print("System is Big Endian")

s コマンドの送信後、サーバー側から s コマンド応答が返ってきます。

  • うまくいった場合には、 文字列 s だけが返ってきます。
  • うまく行かなかった場合には、文字列 s の後に、エラーメッセージもついて返ってきます。

サーバー側から成功の s イベントが返ってきたら、音声認識対象のデータをサーバーに送ります。PyAudioの音声ストリームからデータを取り出して、(後述の音声レベル調整を行ってから)AmiVoice APIに送ります。AmiVoiceでは、送信対象のデータ(バイト列)の先頭に文字列 p を結合してから送ります。

async def send_audio(ws, stream):
    """音声データをWebSocket経由で送信します。"""
    try:
        while ws.open:
            data = stream.read(frames_per_buffer * 4)
            data = auto_gain_control(data) # 音量調整
            await ws.send(b'p' + data)
            await asyncio.sleep(0.1)
    except Exception as e:
        logger.error(f"Exception during websocket communication: {e}")
        raise e

A イベント処理

p コマンドを使ってサーバーに音声データを送信してしばらく経ち、音声認識に成功した場合にはサーバーから A イベントが返ってきます。 A イベントのフォーマットの詳細は公式ドキュメントをみてください。 A の他にも、音声認識の状況や途中経過がリアルタイムに送られてくるので、ほしいイベントを拾って処理すれば良いと思います。

全体的に、 s や A のイベントを拾って処理する部分は次のようになります。1

async def process_message(ws, message, stream, buffered_processor):
    """受信したAmiVoice Apiのメッセージを処理します。"""
    event = message[0]
    content = message[2:].rstrip()
    try:
        if event == "s":
			asyncio.create_task(send_audio(ws, stream))
        elif event == "A":
            json_content = json.loads(content) if content else ''
            await buffered_processor.append(json_content['text']) # buffered_processorは、認識結果のテキストをためておくところ (この後の節で説明)
        elif event == "e":
            return False
    except Exception as e:
        logger.error(f"Error in processing AmiVoice Api Message: {e}")
        return False
    return True

この関数では、音声認識を継続するべきかどうかを戻り値として返しています。 s A のほかに e イベントも拾っていて、 e を受け取ったり例外発生時には False を返します。なおこの関数はイベントを受け取った後の処理に注目しているので、イベントを待ち受ける部分(イベントループ)は別の箇所にあります。

音声データの事前処理: マイクの音声レベル調整

これも知識としては知っていたものの、実装する際には見落としていたことです。

無事に音声データをAmiVoice APIに送り、Aイベントから文字起こし結果を取得することができるようになりましたが、どうも精度が安定しません。「音声が不明瞭」と言われてしまうこともしばしば。念のため、そもそもうまく送れているのかを確認するために、音声ストリームからのデータを貯めて、ファイルに保存してみることにしました。すると明らかに音量が小さい！たまたま良く聞こえる部分だけ、音声認識がうまく行っていたようでした。

音量を上げるには、単純に音声データのサンプリングされた各値に倍率を掛けて大きくすればいいらしいです。ただし、上限値・下限値を超えた値は切る必要があります。 np.clip() を使います。

def amplify_audio(data, gain_factor=1.5):
    """音量変更"""
    audio_data = np.frombuffer(data, dtype=np.int16)
    amplified_audio = np.clip(audio_data * gain_factor, -32768, 32767)
    return amplified_audio.astype(np.int16).tobytes()

音量を上げる倍率は、適切に設定する必要があります。いかにもそれっぽいなのは、音声データ全体を見て、もっとも大きな振幅にあわせて、(それよりも気持ち少なめに)倍率を設定するというものです。

def analyze_peak(data):
    """ピーク振幅を分析し、推奨される倍率を返す"""
    max_amplitude = np.max(np.abs(data))
    max_gain = 32767 / max_amplitude if max_amplitude != 0 else 1.0
    return max_gain * 0.8

ただし今回は、音量を上げたあとの音声の音量が、チャンクごと(先ほどのバケツの例のコップに相当)にバラバラでも問題ありません。AmiVoice APIの中で一生懸命文字起こしをしてくれている小人さんが「音量がバラバラだよぉ」と困るだけです。それは構わないので、チャンクごとに音量倍率を計算して、都度音量を上げていきます。

あとノイズ処理とかもやると、もっと認識精度が上がるかもしれません。今回は省きました。

非同期処理

とても基本的ですがこれが一番つまづきました。

今回のケースでは、次の3点を同時並行的に処理する必要があります。

  • AmiVoiceに常時音声データを送る。
  • AmiVoice側からの解析完了の通知を常時監視する。解析完了のメッセージを受け取ったら、バッファに解析結果をためる。
  • バッファが一定量たまったら、バッファをクリアしてテキストをChatGPTに送り、返答が返ってきたらそれを画面に表示する。

AmiVoiceへの音声データ送信

前述のように、AmiVoiceへの最初の音声データ送信は、 s メッセージを受け取った時に開始します。音声データ送信処理は、少し待って→送信して→少し待って を繰り返す必要がありますが、このループ処理のせいで他の処理(たとえばAmiVoiceからのメッセージを受け取る)を止めてはいけません。なので asyncio.create_task() を使って非同期的に行います。2

AmiVoiceからの解析完了の通知を監視

AmiVoiceからは音声認識の状況や結果などがリアルタイムに送られてきます。これを自作のイベントループ内で待ち受けます。 websockets ライブラリは、メッセージを待ち受けるために recv() メソッドを提供しているので、これを await で待つことで次のイベントの受け取りを待つことができます。

async with websockets.connect(uri) as ws:
    pa = pyaudio.PyAudio()
    stream = pa.open(format=pyaudio.paInt16, channels=1, rate=api_sample_rate, input=True, frames_per_buffer=frames_per_buffer, input_device_index=2)
    openai_client = AsyncOpenAI(api_key="ないしょ")

    async def async_get_chatgpt_response(buffer): # BufferedProcessorにためた文字数が閾値を超えたら実行される
        return await get_chatgpt_response(buffer, openai_client)

    buffered_processor = BufferedProcessor(async_get_chatgpt_response, max_length=50) # 音声認識結果のテキストをためておくところ (次の節で説明)

    await on_open(ws)  # WebSocket接続時の処理

    continue_flag = True
    while continue_flag:
        message = await ws.recv()  # メッセージ受信を待機
        continue_flag = await process_message(ws, message, stream, buffered_processor)  # メッセージ処理

ちなみに pa.open のときの input_device_index=2は、この記事の「入力ラインが複数ある」の節で調べたデバイスのインデックスです。

バッファがたまったらChatGPTにテキストを送る

AmiVoice APIからのイベント処理内で、 A イベントにより受け取った音声認識結果をバッファに貯めます。バッファは文字列のリストの形式を取っていて、追加された文字数の合計が一定数を超えたら、すべてを取り出してChatGPTに送ります3。ししおどしのようなイメージです。実装としては、バッファに新たに文字列を追加する際に文字数カウントのチェックを行い、一定値を超えていたら特定の処理Pを行う、のようにします。ただし、「バッファに文字列を追加する」の処理は、処理Pの完了を待たずに迅速に終わってほしいですね。

このししおどしは、再利用のために別クラスに分離してみました。
今回の利用ケースのようにゆるい条件、つまり「 append が頻繁に呼ばれず、同時にも呼ばれない」ことを前提にしている場合にはあまり問題はないと思われますが、再利用可能性を高めてそのような利用ケースに対応させるためには、スレッドセーフになるように改善を加える必要があります。(今回はやりません)

import asyncio

class BufferedProcessor:
    def __init__(self, process_callback, max_length=100):
        self.buffer = []
        self.max_length = max_length
        self.process_callback = process_callback

    async def append(self, text):
        self.buffer.append(text)

        # self.bufferに含まれる各要素の文字数の合計がmax_lengthを超えたら処理を実行
        if len(''.join(self.buffer)) >= self.max_length:
            asyncio.create_task(self.process_buffer())

    async def process_buffer(self):
        # コールバック関数を呼び出してバッファを処理
        asyncio.create_task(self.process_callback(self.buffer))
        self.buffer = []  # バッファをクリア

結果と感想

音声認識結果

認識精度について

まあ悪くないかな、といったところです4。自分の声で本を読み上げる場合にはそこそこ精度が出ますが、雑に会話を聞かせる場合にはあまり当てにならなかったりもします。カジュアルな会話に含まれるような固有名詞やくだけた語尾などの認識はぜんぜんできませんでした。音声認識エンジンは複数あるみたいなので、より適切なものがあれば切り換えてみるのもいいでしょう。また、文脈に合わせて単語を事前登録しておくのもいいかもしれません。

認識速度について

結構速いです5。ほとんど同時、とまではいきませんが、1秒待たずに結果が返ってくることもあります。ちなみに A イベントは、発話区間で区切った全体の認識が完了したら結果を返すというものなので、ずっとしゃべり続けていると結果が返ってこないこともあります。途中経過が U イベントとして返ってくるので、見た目の処理速度を上げるために必要に応じてこちらを使ってみるのもありかなと。

ChatGPTによる処理結果

参考までにプロンプト:

以下で示す入力は、ユーザーによる音声入力をテキストに変換したものです。音声入力のため、文法や表現が不明瞭な場合がある他、誤字脱字や意味不明な単語が含まれる可能性があります。これを修正し、修正後の内容を返却してください。あまりに内容が通らない場合には、当該部分は除去してください。レスポンスには結果のみを含め、前後の説明や補足は不要です。

検証中にたまたま見ていたこれを聞かせました。(1:36:09〜)

Spoken: 感動しました。
Spoken: 美味しかったです高良薬局長へ確認やこれもなんか美味しく焼いてくれましたとか太っ腹ありがとう、せっかくだからねせっかく名産ですからねしかしてまだ一方
[ChatGPT Request]: 感動しました。美味しかったです高良薬局長へ確認やこれもなんか美味しく焼いてくれましたとか太っ腹ありがとう、せっかくだからねせっかく名産ですからねしかしてまだ一方
[ChatGPT Summary]: 美味しかったです。高良薬局長、確認やこの料理も美味しく焼いてくれてありがとう。太っ腹ですね。せっかくの名産ですからね。

Spoken: ギャバン社長は違います序盤あれは皆さん右です半年ですこちら
Spoken: 近江牛、近江技師が県の
Spoken: 近江牛ですさっきの松崎は結構薄めに切ったんですよこれはちょっと厚めの美味しい
[ChatGPT Request]: ギャバン社長は違います序盤あれは皆さん右です半年ですこちら近江牛、近江技師が県の近江牛ですさっきの松崎は結構薄めに切ったんですよこれはちょっと厚めの美味しい
[ChatGPT Summary]: 近江牛です。さっきの松崎は結構薄めに切ったんです。これはちょっと厚めで美味しいです。

音声認識の結果次第、、ですね6。ChatGPTの能力的には、文脈内に正しく変換された部分があれば、その後表記がゆれていたり誤変換があっても直してくれそうに思いますが、今回は短い会話文しか与えていないのでそういう高度な推測はしてもらえませんでした。誤変換や固有名詞については、事前に単語登録することである程度解消しそうです(今回は試していません)。

WebSocket APIの使い勝手

双方向での通信や発話単位での結果取得ができるため便利だと思いました。音声認識の途中経過も取得できるし、認識の速度もじゅうぶん速いと思います。ただいろいろと初めてということもあり初見ではやや難しかったです。AmiVoice API自体が難しいというよりも、WebSocketの基本的な考え方や音声認識の基本的な理解によるところが大きかったです。

  1. sコマンド失敗時にもevent="s"となるので本当はきちんと拾わなきゃダメですねダメですよ ↩

  2. あっているのか自信ないのであまり参考にしないでください ↩

  3. ChatGPT APIの呼び出し部分は変わったことはないし記事の主題ではないので略 ↩

  4. 面倒なので数字では測っていません ↩

  5. 面倒なので数字では測っていません ↩

  6. 辛口コメントっぽくてすみません・・冒頭で書いた自分がやりたいことが雑談の認識っぽいことなので、それに近い利用ケースを想定してちょっとハードな条件での検証になっちゃいました ↩

8
6
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
8
6