OpenAI の Realtime API を使ってAIと音声会話するWebアプリを実装してみる

npx nuxi@latest init <nuxt-app-name>
cd <nuxt-app-name>

# WebSocketクライアント(Nitro)
npm install ws

# 開発系ライブラリ、Nuxtモジュール等(eslint, tailwindcss...)
npm install -D @types/ws eslint @nuxt/eslint @nuxtjs/tailwindcss

export default defineNuxtConfig({
  compatibilityDate: '2024-04-03',
  devtools: { enabled: true },
  modules: ['@nuxtjs/tailwindcss', '@nuxt/eslint'],
  // NitroのWebSocket対応
  nitro: {
    experimental: {
      websocket: true,
    },
  },
  eslint: {
    config: {
      stylistic: {
        semi: true,
        braceStyle: '1tbs',
      },
    },
  },
});

import { WebSocket } from 'ws';

// 接続ユーザー(peer)単位にRealtime APIのセッションを管理
const connections: { [id: string]: WebSocket } = {};

export default defineWebSocketHandler({
  open(peer) {
    if (!connections[peer.id]) {
      // OpenAIのRealtime APIとの接続
      const url = 'wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01';
      connections[peer.id] = new WebSocket(url, {
        headers: {
          'Authorization': 'Bearer ' + process.env.OPENAI_API_KEY,
          'OpenAI-Beta': 'realtime=v1',
        },
      });
    }
    const instructions = '明るく元気に話してください。仲の良い友人のように振る舞い、敬語は使わないでください。出力は日本語でしてください。';

    connections[peer.id].on('open', () => {
      // Realtime APIのセッション設定
      connections[peer.id].send(JSON.stringify({
        type: 'session.update',
        session: {
          voice: 'shimmer',
          instructions: instructions,
          input_audio_transcription: { model: 'whisper-1' },
          turn_detection: { type: 'server_vad' },
        },
      }));
    });
    connections[peer.id].on('message', (message) => {
      // Realtime APIのサーバーイベントはそのままクライアントに返す
      peer.send(message.toString());
    });
  },
  message(peer, message) {
    // クライアントイベインとはそのままRealtime APIに中継する
    connections[peer.id].send(message.text());
  },
  close(peer) {
    connections[peer.id].close();
    connections[peer.id] = undefined;
    console.log('closed websocket');
  },
  error(peer, error) {
    console.log('error', { error, id: peer.id });
  },
});

type Params = {
  url: string;
  logMessage: (string) => void;
  onMessageCallback?: (message: MessageEvent) => void;
};

export const useRealtimeApi = ({ url, logMessage, onMessageCallback }: Params) => {
  let ws: WebSocket | null = null;
  const isConnected = ref(false);

  function connect() {
    if (isConnected.value) return;

    ws = new WebSocket(url);
    ws.onopen = () => {
      logMessage('Connected to server🍻');
      isConnected.value = true;
    };

    ws.onclose = () => {
      logMessage('Disconnected👋');
      isConnected.value = false;
    };

    ws.onerror = (error) => {
      logMessage('Error occurred😭: ' + error.message);
    };

    ws.onmessage = (message: MessageEvent) => {
      if (onMessageCallback) onMessageCallback(message); // Realtime APIのサーバーイベント
    };
  }

  function disconnect() {
    ws?.close();
  }

  function sendMessage(data: unknown) {
    if (isConnected.value) {
      ws?.send(JSON.stringify(data)); // Realtime APIのクライアントイベント
    }
  }

  return {
    connect,
    disconnect,
    sendMessage,
    isConnected,
  };
};

const BUFFER_SIZE = 8192;
const BUFFER_INTERVAL = 1000;

type Params = {
  audioCanvas: Ref<HTMLCanvasElement>;
  logMessage: (string) => void;
  onFlushCallback: (buffer: ArrayBuffer) => void;
};

export function useAudio({ audioCanvas, logMessage, onFlushCallback }: Params) {
  const isRecording = ref(false);
  const isPlaying = ref(false);

  // 省略

  /**
   * 音声録音開始関数
   * 音声はバッファリングして一定サイズ超過または一定時間経過後にコールバック実行
   */
  async function startRecording() {
    isRecording.value = true;
    try {
      // マイクの準備(許可要求)
      const mediaStream = await navigator.mediaDevices.getUserMedia({ audio: true });
      audioContext = new (window.AudioContext || window.webkitAudioContext)({ sampleRate: 24000 });

      // 省略(音声入力変換とコールバック処理実行)
    } catch (e) {
      // 省略
    }
  }

  /**
   * 音声録音停止関数
   */
  function stopRecording() {
    // 省略
  }

  /**
   * 再生する音声をキューに格納 -> 順次再生
   */
  function enqueueAudio(buffer: Float32Array) {
    audioQueue.push(buffer);
    if (!isPlaying.value) {
      playFromQueue();
    }
  }
  
  return {
    startRecording,
    stopRecording,
    enqueueAudio,
    isPlaying,
    isRecording,
  };
}

// 再生音声フラッシュ時のコールバック
function handleAudioFlush(buffer: ArrayBuffer) {
  // マイクからの音声入力をRealtime APIに送信
  sendMessage({ type: 'input_audio_buffer.append', audio: arrayBufferToBase64(buffer) });
}

const { startRecording, stopRecording, enqueueAudio, isRecording } = useAudio({
  audioCanvas, // 波形表示用のCanvs
  logMessage,  // ログ追加コールバック
  onFlushCallback: handleAudioFlush, // 音声フラッシュ時のコールバック
});

// RealtimeAPIのサーバーイベントハンドラ
function handleWebSocketMessage(message: MessageEvent) {
  const event = JSON.parse(message.data);
  logEvent(event.type);
  switch (event.type) {
    case 'response.audio.delta': {
      enqueueAudio(base64ToArrayBuffer(event.delta));
      break;
    }
    case 'response.audio_transcript.done':
      // レスポンスより早くイベントが発火することがあるのでロギングを遅延させる
      setTimeout(() => logMessage(`🤖: ${event.transcript}`), 100);
      break;
    case 'conversation.item.input_audio_transcription.completed':
      logMessage(`😄: ${event.transcript}`);
      break;
    case 'error':
      logEvent(event.error);
      if (isRecording.value) stopRecording();
      // Realtime APIは15分経過するとセッションタイムアウトする
      if (event.code === 'session_expired') disconnect();
      break;
    default:
      break;
  }
}

// Realtime API
const { connect, isConnected, disconnect, sendMessage } = useRealtimeApi({
  url: 'ws://localhost:3000/ws',
  logMessage,
  onMessageCallback: handleWebSocketMessage,
});