Voice AIエージェントをゼロから作る — WebSocket + STT + LLM + TTS 構築記

// AudioWorkletProcessor: オーディオスレッドで動作
class MicProcessor extends AudioWorkletProcessor {
  process(inputs, outputs, parameters) {
    const input = inputs[0][0]; // モノラルチャンネル
    if (input) {
      // Float32 → Int16 PCM に変換（STTサービスが期待する形式）
      const pcm16 = new Int16Array(input.length);
      for (let i = 0; i < input.length; i++) {
        pcm16[i] = Math.max(-32768, Math.min(32767, input[i] * 32768));
      }
      this.port.postMessage(pcm16.buffer, [pcm16.buffer]);
    }
    return true; // trueを返してプロセッサを継続
  }
}
registerProcessor('mic-processor', MicProcessor);

// メインスレッド: マイクを取得してAudioWorkletに接続
const stream = await navigator.mediaDevices.getUserMedia({
  audio: { sampleRate: 16000, channelCount: 1, echoCancellation: true }
});

const audioCtx = new AudioContext({ sampleRate: 16000 });
await audioCtx.audioWorklet.addModule('/mic-processor.js');

const source = audioCtx.createMediaStreamSource(stream);
const workletNode = new AudioWorkletNode(audioCtx, 'mic-processor');

workletNode.port.onmessage = (event) => {
  // PCMデータをWebSocketに送信（ステップ2で実装）
  sendToWebSocket(event.data);
};

source.connect(workletNode);

# Python サーバー側（FastAPI + WebSocket）
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws/voice")
async def voice_endpoint(websocket: WebSocket):
    await websocket.accept()

    # 各ステージを非同期タスクとして起動
    stt_queue = asyncio.Queue()    # 音声 → STT
    llm_queue = asyncio.Queue()    # テキスト → LLM
    tts_queue = asyncio.Queue()    # 応答テキスト → TTS

    async def receive_audio():
        """ブラウザからの音声を受信してSTTキューに流す"""
        while True:
            data = await websocket.receive_bytes()
            await stt_queue.put(data)

    async def run_stt():
        """STTで書き起こしてLLMキューに流す"""
        # ステップ3で実装

    async def run_llm():
        """LLMで応答生成してTTSキューに流す"""
        # ステップ4で実装

    async def run_tts():
        """TTSで音声合成してブラウザに送信"""
        # ステップ5で実装

    await asyncio.gather(
        receive_audio(), run_stt(), run_llm(), run_tts()
    )

# ストリーミングSTTの実装例（Deepgram WebSocket）
import websockets
import json

async def run_stt(stt_queue, llm_queue):
    url = "wss://api.deepgram.com/v1/listen?model=nova-2&language=ja"
    headers = {"Authorization": f"Token {DEEPGRAM_API_KEY}"}

    async with websockets.connect(url, extra_headers=headers) as dg_ws:
        async def send_audio():
            while True:
                audio_chunk = await stt_queue.get()
                await dg_ws.send(audio_chunk)

        async def receive_transcript():
            async for message in dg_ws:
                result = json.loads(message)
                # is_final=True の結果のみLLMに渡す
                if result.get("is_final"):
                    transcript = result["channel"]["alternatives"][0]["transcript"]
                    if transcript.strip():
                        await llm_queue.put(transcript)

        await asyncio.gather(send_audio(), receive_transcript())

# LLM応答のストリーミング＋文単位チャンク分割
async def run_llm(llm_queue, tts_queue):
    conversation_history = [
        {"role": "system", "content": "あなたは親切な日本語の音声アシスタントです。回答は簡潔に。"}
    ]

    while True:
        user_text = await llm_queue.get()
        conversation_history.append({"role": "user", "content": user_text})

        # ストリーミングでLLM応答を取得
        stream = openai.chat.completions.create(
            model="gpt-4o",
            messages=conversation_history,
            stream=True
        )

        # 文単位でバッファリングしてTTSに送信
        sentence_buffer = ""
        sentence_delimiters = {"。", "！", "？", ".", "!", "?", "\n"}

        for chunk in stream:
            token = chunk.choices[0].delta.content or ""
            sentence_buffer += token

            # 文末記号が見つかったら1文をTTSに送信
            if any(d in token for d in sentence_delimiters):
                await tts_queue.put(sentence_buffer.strip())
                sentence_buffer = ""

        # 残りのバッファも送信
        if sentence_buffer.strip():
            await tts_queue.put(sentence_buffer.strip())

# ストリーミングTTSの実装例
async def run_tts(tts_queue, websocket):
    while True:
        text = await tts_queue.get()

        # ストリーミングTTS APIを呼び出し
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST",
                "https://api.openai.com/v1/audio/speech",
                headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
                json={
                    "model": "tts-1",
                    "voice": "nova",
                    "input": text,
                    "response_format": "pcm"  # 生PCMで受信
                }
            ) as response:
                async for audio_chunk in response.aiter_bytes(chunk_size=4096):
                    # 音声チャンクを即座にブラウザに送信
                    await websocket.send_bytes(audio_chunk)

// ブラウザ側: チャンクの連続再生
class AudioChunkPlayer {
  constructor(sampleRate = 24000) {
    this.audioCtx = new AudioContext({ sampleRate });
    this.nextStartTime = 0;
  }

  playChunk(pcmData) {
    // Int16 PCM → Float32 に変換
    const float32 = new Float32Array(pcmData.length);
    for (let i = 0; i < pcmData.length; i++) {
      float32[i] = pcmData[i] / 32768;
    }

    // AudioBufferを生成
    const buffer = this.audioCtx.createBuffer(1, float32.length, this.audioCtx.sampleRate);
    buffer.getChannelData(0).set(float32);

    const source = this.audioCtx.createBufferSource();
    source.buffer = buffer;
    source.connect(this.audioCtx.destination);

    // 次のチャンクの再生開始時刻を計算
    const now = this.audioCtx.currentTime;
    const startTime = Math.max(now, this.nextStartTime);
    source.start(startTime);

    // 次のチャンクはこのチャンクの終了直後に開始
    this.nextStartTime = startTime + buffer.duration;
  }

  // 割り込み（ユーザーが話し始めた時に再生を中断）
  interrupt() {
    this.audioCtx.close();
    this.audioCtx = new AudioContext({ sampleRate: 24000 });
    this.nextStartTime = 0;
  }
}