AIとボイスチャット(したかった)

GPUがあればできそう

ちょっと前に書いた https://yakimaguro.com/p/voicevox-streaming-trial voicevox のストリーミングと音声認識を組み合わせて
Discord で AI とボイスチャットができないか調べた。

前提条件が CPU なのが厳しいところ。

Discord から音声を拾う

これは pycord を使えばできた。

とりあえず全体のコードを

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class VoiceListener:
    def __init__(self, bot: discord.Bot):
        self.bot = bot
        self.is_listening = False
        self.voice_client: discord.VoiceClient = None

    def start_listening(self, voice_client: discord.VoiceClient):
        if self.is_listening:
            print("Already listening.")
            return

        self.voice_client = voice_client

        self.voice_client.empty_socket()
        self.voice_client.decoder = discord.opus.DecodeManager(self)
        self.voice_client.decoder.start()
        self.is_listening = True

        t = threading.Thread(target=self.recv_audio)
        t.start()

    def recv_decoded_audio(self, data: discord.sinks.core.RawData):
        while data.ssrc not in self.voice_client.ws.ssrc_map:
            time.sleep(0.01)

        pcm_audio = data.decoded_data

    def recv_audio(self):
        while self.is_listening:
            ready, _, err = select.select(
                [self.voice_client.socket], [], [self.voice_client.socket], 0.01
            )
            if not ready:
                if err:
                    print(f"Socket error: {err}")
                continue

            try:
                data = self.voice_client.socket.recv(4096)
            except OSError:
                continue

            self.voice_client.unpack_audio(data)

    async def stop_listening(self):
        if not self.is_listening:
            print("Not listening.")
            return

        self.is_listening = False
        self.voice_client.decoder.stop()
        await self.voice_client.disconnect()
        self.voice_client = None
        print("Stopped listening.")

まずボイスチャンネルに接続されたら start_listening を呼び出だす。 そして recv_audic で音声を拾う。

ちなみに、recv_decoded_audio はどこで呼び出されるのかというと、これは pycord のソースコード(opus.py)を読めばわかるが、

1
self.voice_client.decoder = discord.opus.DecodeManager(self)

まず DecodeManager のインスタンスを生成するときに、self を引数で渡しているのがポイントで、こうすると

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
class DecodeManager(threading.Thread, _OpusStruct):
    def __init__(self, client):
        super().__init__(daemon=True, name="DecodeManager")

        self.client = client
        self.decode_queue = []

        self.decoder = {}

        self._end_thread = threading.Event()

(opus.py 521-530)
の client に VoiceListener 自身が入る。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
def run(self):
    while not self._end_thread.is_set():
        try:
            data = self.decode_queue.pop(0)
        except IndexError:
            time.sleep(0.001)
            continue

        try:
            if data.decrypted_data is None:
                continue
            else:
                data.decoded_data = self.get_decoder(data.ssrc).decode(
                    data.decrypted_data
                )
        except OpusError:
            print("Error occurred while decoding opus frame.")
            continue

        self.client.recv_decoded_audio(data)

一番下に

1
self.client.recv_decoded_audio(data)

と書いてあるが、この時 self.client は VoiceListener のことなので VoiceListener の中にある recv_decoded_audio が実行される。

これでひとまず Discord の音声を拾うことができた。

音声認識

色々探したが以下が見つかった。

とりあえず CPU で軽そうな vosk を使ってみることにした。

まず Discord から拾った音声を PCM 16bit のモノラル音声にしないといけないので

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
def downsample_to_16k_mono(self, pcm_audio: bytes) -> bytes:
    audio_np = np.frombuffer(pcm_audio, dtype=np.int16)

    if len(audio_np) % 2 == 0:
        audio_np = audio_np.reshape(-1, 2)
        audio_np = audio_np.mean(axis=1).astype(np.int16)

    downsampled = resample_poly(audio_np, up=1, down=3)

    return downsampled.astype(np.int16).tobytes()

こんな関数を ChatGPT に作ってもらった。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
def recv_decoded_audio(self, data: discord.sinks.core.RawData):
    while data.ssrc not in self.voice_client.ws.ssrc_map:
        time.sleep(0.01)

    pcm_audio = data.decoded_data
    pcm_audio = self.downsample_to_16k_mono(pcm_audio)

    if self.recognizer.AcceptWaveform(pcm_audio):
        result = self.recognizer.Result()
        if result:
            print(f"result: {result}")
    else:
        partial_result = self.recognizer.PartialResult()
        if partial_result:
            print(f"Partial result: {partial_result}")

こんな感じに Discord から拾った音声を変換して vosk にいれる。(recognizer は vosk のドキュメントなどを見ながら書いてください)

ここまで書くと Discord で喋ると文字起こしされるが、small モデルだからだかすごく精度が悪い!

これじゃ実用的とは言えないな…というお話でした。

(GPU で Whisper を使えばできると思います。)

Hugo で構築されています。
テーマ StackJimmy によって設計されています。