diff --git a/server-deno/main.ts b/server-deno/main.ts index 6da631d..0ce5de2 100644 --- a/server-deno/main.ts +++ b/server-deno/main.ts @@ -59,40 +59,37 @@ wss.on("connection", async (ws: WSWebSocket, payload: IPayload) => { ws.send( JSON.stringify({ type: "auth", - volume_control: user.device?.volume ?? 100, + volume_control: user.device?.volume ?? 20, is_ota: user.device?.is_ota ?? false, is_reset: user.device?.is_reset ?? false, pitch_factor: user.personality?.pitch_factor ?? 1, }), ); + // Common close handler for cleanup + const closeHandler = async () => { + // Add any common cleanup logic here + }; + + // Common provider args + const providerArgs: ProviderArgs = { + ws, + payload, + connectionPcmFile, + firstMessage, + systemPrompt, + closeHandler, + }; + switch (provider) { case "openai": - await connectToOpenAI( - ws, - payload, - connectionPcmFile, - firstMessage, - systemPrompt, - ); + await connectToOpenAI(providerArgs); break; case "gemini": - await connectToGemini( - ws, - payload, - connectionPcmFile, - firstMessage, - systemPrompt, - ); + await connectToGemini(providerArgs); break; case "grok": - await connectToGrok( - ws, - payload, - connectionPcmFile, - firstMessage, - systemPrompt, - ); + await connectToGrok(providerArgs); break; case "elevenlabs": const agentId = user.personality?.oai_voice ?? ""; @@ -107,11 +104,11 @@ wss.on("connection", async (ws: WSWebSocket, payload: IPayload) => { connectionPcmFile, agentId, elevenLabsApiKey, + closeHandler, ); break; case "hume": - await connectToHume(ws, payload, - connectionPcmFile, firstMessage, systemPrompt, () => Promise.resolve()); + await connectToHume(providerArgs); break; default: throw new Error(`Unknown provider: ${provider}`); diff --git a/server-deno/models/elevenlabs.ts b/server-deno/models/elevenlabs.ts index 397ea2c..85ae3ee 100644 --- a/server-deno/models/elevenlabs.ts +++ b/server-deno/models/elevenlabs.ts @@ -9,7 +9,7 @@ import { } from "npm:@elevenlabs/client"; import { addConversation, getDeviceInfo } from "../supabase.ts"; -import { encoder, FRAME_SIZE, isDev } from "../utils.ts"; +import { createOpusPacketizer, isDev } from "../utils.ts"; // Calculate audio level for debugging function calculateAudioLevel(audioData: any): number { @@ -32,10 +32,13 @@ export const connectToElevenLabs = async ( connectionPcmFile: Deno.FsFile | null, agentId: string, apiKey: string, + closeHandler: () => Promise, ) => { console.log(apiKey, agentId); const { user, supabase } = payload; + const opus = createOpusPacketizer((packet) => ws.send(packet)); + // Queue messages until ElevenLabs connection is ready const messageQueue: RawData[] = []; let isElevenLabsConnected = false; @@ -128,13 +131,6 @@ export const connectToElevenLabs = async ( isElevenLabsConnected = true; console.log(`ElevenLabs connection ready - conversation_initiation_metadata already processed by SDK`); - // Send initial RESPONSE.CREATED for the first message - // console.log("Sending initial RESPONSE.CREATED to ESP32"); - // ws.send(JSON.stringify({ - // type: "server", - // msg: "RESPONSE.CREATED" - // })); - // Set up ElevenLabs event handlers elevenLabsConnection.onMessage(async (event: IncomingSocketEvent) => { console.log("ElevenLabs message type:", event); @@ -142,7 +138,6 @@ export const connectToElevenLabs = async ( switch (event.type) { case "conversation_initiation_metadata": console.log("ElevenLabs conversation initiated (metadata received)"); - // RESPONSE.CREATED already sent when connection was established break; case "ping": @@ -158,9 +153,10 @@ export const connectToElevenLabs = async ( case "audio": if (event.audio_event?.audio_base_64) { - // Send RESPONSE.CREATED only for the first audio chunk of each response + // Send RESPONSE.CREATED before first audio chunk if (!hasResponseStarted) { console.log("Sending RESPONSE.CREATED to ESP32 (agent audio starting)"); + opus.reset(); ws.send(JSON.stringify({ type: "server", msg: "RESPONSE.CREATED" @@ -169,23 +165,10 @@ export const connectToElevenLabs = async ( } const audioBuffer = Buffer.from(event.audio_event.audio_base_64, "base64"); - console.log(`Received audio from ElevenLabs: ${audioBuffer.length} bytes, processing into ${Math.ceil(audioBuffer.length / FRAME_SIZE)} frames`); + console.log(`Received audio from ElevenLabs: ${audioBuffer.length} bytes`); - let framesSent = 0; - // Process audio in frames for Opus encoding - for (let offset = 0; offset < audioBuffer.length; offset += FRAME_SIZE) { - const frame = audioBuffer.subarray(offset, offset + FRAME_SIZE); - - try { - const encodedPacket = encoder.encode(frame); - ws.send(encodedPacket); - framesSent++; - } catch (_e) { - // Skip this frame but continue with others - console.log(`Failed to encode frame at offset ${offset}`); - } - } - console.log(`Sent ${framesSent} audio frames to ESP32`); + // Use Opus packetizer to encode and send audio + opus.push(audioBuffer); } break; @@ -199,23 +182,15 @@ export const connectToElevenLabs = async ( user, ); - // Send audio committed message like OpenAI does - // console.log("Sending AUDIO.COMMITTED to ESP32"); - // ws.send(JSON.stringify({ - // type: "server", - // msg: "AUDIO.COMMITTED" - // })); - if (!hasResponseStarted) { console.log("Sending RESPONSE.CREATED to ESP32 (agent audio starting)"); + opus.reset(); ws.send(JSON.stringify({ type: "server", msg: "RESPONSE.CREATED" })); hasResponseStarted = true; } - - } break; @@ -229,6 +204,9 @@ export const connectToElevenLabs = async ( user, ); + // Flush any remaining audio before sending complete + opus.flush(true); + // Send response complete with device info like OpenAI does console.log("Sending RESPONSE.COMPLETE to ESP32"); hasResponseStarted = false; // Reset for next response @@ -305,6 +283,8 @@ export const connectToElevenLabs = async ( ws.on("close", async (code: number, reason: string) => { console.log(`ESP32 WebSocket closed with code ${code}, reason: ${reason}`); + await closeHandler(); + opus.close(); elevenLabsConnection?.close(); if (isDev && connectionPcmFile) { @@ -329,4 +309,4 @@ export const connectToElevenLabs = async ( msg: errorMessage })); } -}; \ No newline at end of file +}; diff --git a/server-deno/models/gemini.ts b/server-deno/models/gemini.ts index 0093746..d72dd01 100644 --- a/server-deno/models/gemini.ts +++ b/server-deno/models/gemini.ts @@ -1,281 +1,253 @@ import { Buffer } from "node:buffer"; import type { WebSocketServer as _WebSocketServer } from "npm:@types/ws"; import { - EndSensitivity, - GoogleGenAI, - LiveConnectConfig, - LiveServerMessage, - Modality, - Session, + EndSensitivity, + GoogleGenAI, + LiveConnectConfig, + LiveServerMessage, + Modality, + Session, } from "npm:@google/genai"; -import { encoder, FRAME_SIZE, geminiApiKey, isDev } from "../utils.ts"; +import { createOpusPacketizer, geminiApiKey, isDev, defaultGeminiVoice } from "../utils.ts"; import { addConversation } from "../supabase.ts"; -export const connectToGemini = async ( - ws: WebSocket, - payload: IPayload, - connectionPcmFile: Deno.FsFile | null, - firstMessage: string, - systemPrompt: string, -) => { - const { user, supabase } = payload; - const { oai_voice } = user.personality ?? { oai_voice: "Sadachbia" }; +export const connectToGemini = async ({ + ws, + payload, + connectionPcmFile, + firstMessage, + systemPrompt, + closeHandler, +}: ProviderArgs) => { + const { user, supabase } = payload; + const voiceName = user.personality?.oai_voice ?? defaultGeminiVoice; - console.log(`Connecting with Gemini key "${geminiApiKey.slice(0, 3)}..."`); + const opus = createOpusPacketizer((packet) => ws.send(packet)); - // Initialize Google GenAI - const ai = new GoogleGenAI({ apiKey: geminiApiKey }); - const model = "gemini-2.5-flash-native-audio-preview-09-2025"; - const config: LiveConnectConfig = { - responseModalities: [Modality.AUDIO], - systemInstruction: systemPrompt, - speechConfig: { - voiceConfig: { - prebuiltVoiceConfig: { - voiceName: oai_voice, - }, - }, - }, - realtimeInputConfig: { - automaticActivityDetection: { - disabled: false, // Keep VAD enabled - endOfSpeechSensitivity: EndSensitivity.END_SENSITIVITY_LOW, // How sensitive to detect speech ending - silenceDurationMs: 100, // How much silence before considering speech ended - }, - }, - outputAudioTranscription: {}, - inputAudioTranscription: {}, - }; + console.log(`Connecting with Gemini key "${geminiApiKey?.slice(0, 3)}..."`); - // Response queue for handling Google's callback-based responses - const responseQueue: LiveServerMessage[] = []; - let geminiSession: Session | null = null; + // Initialize Google GenAI + const ai = new GoogleGenAI({ apiKey: geminiApiKey }); + const model = "gemini-2.5-flash-native-audio-preview-09-2025"; + const config: LiveConnectConfig = { + responseModalities: [Modality.AUDIO], + systemInstruction: systemPrompt, + speechConfig: { + voiceConfig: { + prebuiltVoiceConfig: { + voiceName: voiceName, + }, + }, + }, + realtimeInputConfig: { + automaticActivityDetection: { + disabled: false, // Keep VAD enabled + endOfSpeechSensitivity: EndSensitivity.END_SENSITIVITY_LOW, + silenceDurationMs: 100, + }, + }, + outputAudioTranscription: {}, + inputAudioTranscription: {}, + }; - async function waitMessage() { - let done = false; - let message: LiveServerMessage | undefined = undefined; - while (!done) { - message = responseQueue.shift(); - if (message) { - done = true; - } else { - await new Promise((resolve) => setTimeout(resolve, 10)); - } - } - return message; - } + // Response queue for handling Google's callback-based responses + const responseQueue: LiveServerMessage[] = []; + let geminiSession: Session | null = null; - async function handleTurn() { - const turns: any[] = []; - let done = false; - while (!done) { - const message = await waitMessage(); - turns.push(message); + async function waitMessage() { + let done = false; + let message: LiveServerMessage | undefined = undefined; + while (!done) { + message = responseQueue.shift(); + if (message) { + done = true; + } else { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + } + return message; + } - if ( - message.serverContent - ) { - if (message.serverContent.generationComplete) { - ws.send(JSON.stringify({ - type: "server", - msg: "RESPONSE.CREATED", - })); - done = true; - } + async function handleTurn() { + const turns: any[] = []; + let done = false; + while (!done) { + const message = await waitMessage(); + turns.push(message); - // if (message.serverContent.turnComplete) { - // ws.send( - // JSON.stringify({ - // type: "server", - // msg: "AUDIO.COMMITTED", - // }), - // ); - // } - } - } - return turns; - } + if (message.serverContent) { + if (message.serverContent.generationComplete) { + opus.reset(); + ws.send(JSON.stringify({ + type: "server", + msg: "RESPONSE.CREATED", + })); + done = true; + } + } + } + return turns; + } - async function processGeminiTurns() { - try { - console.log("Processing Gemini turns"); - while (geminiSession) { - const turns = await handleTurn(); + async function processGeminiTurns() { + try { + console.log("Processing Gemini turns"); + while (geminiSession) { + const turns = await handleTurn(); - // Combine all audio data from this turn - const combinedAudio = turns.reduce( - (acc: number[], turn: any) => { - if (turn.data) { - const buffer = Buffer.from(turn.data, "base64"); - const intArray = new Int16Array( - buffer.buffer, - buffer.byteOffset, - buffer.byteLength / - Int16Array.BYTES_PER_ELEMENT, - ); - return acc.concat(Array.from(intArray)); - } - return acc; - }, - [], - ); + // Combine all audio data from this turn + const combinedAudio = turns.reduce( + (acc: number[], turn: any) => { + if (turn.data) { + const buffer = Buffer.from(turn.data, "base64"); + const intArray = new Int16Array( + buffer.buffer, + buffer.byteOffset, + buffer.byteLength / + Int16Array.BYTES_PER_ELEMENT, + ); + return acc.concat(Array.from(intArray)); + } + return acc; + }, + [], + ); - if (combinedAudio.length > 0) { - // Convert back to buffer and send to client - const audioBuffer = new Int16Array(combinedAudio); - const buffer = Buffer.from(audioBuffer.buffer); + if (combinedAudio.length > 0) { + // Convert back to buffer and send to client + const audioBuffer = new Int16Array(combinedAudio); + const buffer = Buffer.from(audioBuffer.buffer); - // PREVIEW AUDIO - // const wf = new WaveFile(); - // wf.fromScratch(1, SAMPLE_RATE, "16", audioBuffer); + // Use Opus packetizer to encode and send audio + opus.push(buffer); + opus.flush(true); + } - // const filename = `gemini_response_${Date.now()}.wav`; - // await Deno.writeFile(filename, wf.toBuffer()); - // console.log(`Audio saved as ${filename}`); + // Handle text responses if any + let outputTranscriptionText = ""; + let inputTranscriptionText = ""; + for (const turn of turns as LiveServerMessage[]) { + if ( + turn.serverContent && + turn.serverContent.outputTranscription + ) { + outputTranscriptionText += + turn.serverContent.outputTranscription.text; + } - // SEND TO ESP32 - // Send audio in chunks to client - for ( - let offset = 0; - offset < buffer.length; - offset += FRAME_SIZE - ) { - const frame = buffer.subarray( - offset, - offset + FRAME_SIZE, - ); - try { - const encodedPacket = encoder.encode(frame); - ws.send(encodedPacket); - } catch (_e) { - // Skip this frame but continue with others - } - } - } + if ( + turn.serverContent && + turn.serverContent.inputTranscription + ) { + inputTranscriptionText += + turn.serverContent.inputTranscription.text; + } + } - // // Handle text responses if any - let outputTranscriptionText = ""; - let inputTranscriptionText = ""; - for (const turn of turns as LiveServerMessage[]) { - if ( - turn.serverContent && - turn.serverContent.outputTranscription - ) { - outputTranscriptionText += - turn.serverContent.outputTranscription.text; - } + // Send completion signal + ws.send(JSON.stringify({ + type: "server", + msg: "RESPONSE.COMPLETE", + })); - if ( - turn.serverContent && - turn.serverContent.inputTranscription - ) { - inputTranscriptionText += - turn.serverContent.inputTranscription.text; - } - } + // Add user transcription to supabase + await addConversation( + supabase, + "user", + inputTranscriptionText, + user, + ); - // Send completion signal - ws.send(JSON.stringify({ - type: "server", - msg: "RESPONSE.COMPLETE", - })); + // Add assistant transcription to supabase + await addConversation( + supabase, + "assistant", + outputTranscriptionText, + user, + ); + } + } catch (error) { + console.error("Error processing Gemini turns:", error); + } + } - // Add user transcription to supabase - await addConversation( - supabase, - "user", - inputTranscriptionText, - user, - ); + // Connect to Google Gemini Live + try { + geminiSession = await ai.live.connect({ + model: model, + callbacks: { + onopen: function () { + console.log("Gemini session opened"); + }, + onmessage: function (message: LiveServerMessage) { + responseQueue.push(message); + }, + onerror: function (e: any) { + console.error("Gemini error:", e.message); + ws.send( + JSON.stringify({ + type: "server", + msg: "RESPONSE.ERROR", + }), + ); + }, + onclose: function (e: any) { + console.log("Gemini session closed:", e.reason); + }, + }, + config: config, + }); - // Add assistant transcription to supabase - await addConversation( - supabase, - "assistant", - outputTranscriptionText, - user, - ); - } - } catch (error) { - console.error("Error processing Gemini turns:", error); - } - } + console.log("Connected to Gemini successfully!"); + // Send first message if available + const inputTurns = [{ + role: "user", + parts: [{ text: firstMessage }], + }]; + geminiSession?.sendClientContent({ turns: inputTurns }); + processGeminiTurns(); + } catch (e: unknown) { + console.log(`Error connecting to Gemini: ${e}`); + ws.close(); + return; + } - // Connect to Google Gemini Live - try { - geminiSession = await ai.live.connect({ - model: model, - callbacks: { - onopen: function () { - console.log("Gemini session opened"); - }, - onmessage: function (message: LiveServerMessage) { - responseQueue.push(message); - }, - onerror: function (e: any) { - console.error("Gemini error:", e.message); - ws.send( - JSON.stringify({ - type: "server", - msg: "RESPONSE.ERROR", - }), - ); - }, - onclose: function (e: any) { - console.log("Gemini session closed:", e.reason); - }, - }, - config: config, - }); + ws.on("message", (data: any, isBinary: boolean) => { + try { + if (isBinary) { + // Handle binary audio data from ESP32 + const base64Data = data.toString("base64"); - console.log("Connected to Gemini successfully!"); - // Send first message if available - const inputTurns = [{ - role: "user", - parts: [{ text: firstMessage }], - }]; - geminiSession?.sendClientContent({ turns: inputTurns }); - processGeminiTurns(); - } catch (e: unknown) { - console.log(`Error connecting to Gemini: ${e}`); - ws.close(); - return; - } + if (isDev && connectionPcmFile) { + connectionPcmFile.write(data); + } - ws.on("message", (data: any, isBinary: boolean) => { - try { - if (isBinary) { - // Handle binary audio data from ESP32 - const base64Data = data.toString("base64"); + // Send audio to Gemini + geminiSession?.sendRealtimeInput({ + audio: { + data: base64Data, + mimeType: "audio/pcm;rate=24000", + }, + }); + } + } catch (e: unknown) { + console.error("Error handling message:", (e as Error).message); + } + }); - if (isDev && connectionPcmFile) { - connectionPcmFile.write(data); - } + ws.on("error", (error: any) => { + console.error("WebSocket error:", error); + geminiSession?.close(); + }); - // Send audio to Gemini - geminiSession?.sendRealtimeInput({ - audio: { - data: base64Data, - mimeType: "audio/pcm;rate=24000", // Gemini expects 16kHz but 24kHz is fine - }, - }); - } - } catch (e: unknown) { - console.error("Error handling message:", (e as Error).message); - } - }); - - ws.on("error", (error: any) => { - console.error("WebSocket error:", error); - geminiSession?.close(); - }); - - ws.on("close", async (code: number, reason: string) => { - console.log(`WebSocket closed with code ${code}, reason: ${reason}`); - geminiSession?.close(); - if (isDev && connectionPcmFile) { - connectionPcmFile.close(); - console.log("Closed debug audio file."); - } - }); + ws.on("close", async (code: number, reason: string) => { + console.log(`WebSocket closed with code ${code}, reason: ${reason}`); + await closeHandler(); + opus.close(); + geminiSession?.close(); + if (isDev && connectionPcmFile) { + connectionPcmFile.close(); + console.log("Closed debug audio file."); + } + }); }; diff --git a/server-deno/models/grok.ts b/server-deno/models/grok.ts index c134ab4..e3d71fa 100644 --- a/server-deno/models/grok.ts +++ b/server-deno/models/grok.ts @@ -1,32 +1,33 @@ +import { Buffer } from "node:buffer"; +import type { RawData } from "npm:@types/ws"; +import { WebSocket } from "npm:ws"; +import { addConversation, getDeviceInfo } from "../supabase.ts"; +import { createOpusPacketizer, isDev, xaiApiKey, defaultGrokVoice } from "../utils.ts"; -import { Buffer } from 'node:buffer'; -import type { RawData } from 'npm:@types/ws'; -import { WebSocket } from 'npm:ws'; -import { addConversation, getDeviceInfo } from '../supabase.ts'; -import { encoder, FRAME_SIZE, isDev, xaiApiKey } from '../utils.ts'; +const XAI_REALTIME_URL = "wss://api.x.ai/v1/realtime"; -const XAI_REALTIME_URL = 'wss://api.x.ai/v1/realtime'; -const DEFAULT_GROK_VOICE = 'Ara'; - -export const connectToGrok = async ( - ws: WebSocket, - payload: IPayload, - connectionPcmFile: Deno.FsFile | null, - firstMessage: string, - systemPrompt: string, -) => { +export const connectToGrok = async ({ + ws, + payload, + connectionPcmFile, + firstMessage, + systemPrompt, + closeHandler, +}: ProviderArgs) => { const { user, supabase } = payload; if (!xaiApiKey) { - throw new Error('XAI_API_KEY is not set'); + throw new Error("XAI_API_KEY is not set"); } - const voice = user.personality?.oai_voice ?? DEFAULT_GROK_VOICE; + const voice = user.personality?.oai_voice ?? defaultGrokVoice; + + const opus = createOpusPacketizer((packet) => ws.send(packet)); const grokWs = new WebSocket(XAI_REALTIME_URL, { headers: { Authorization: `Bearer ${xaiApiKey}`, - 'Content-Type': 'application/json', + "Content-Type": "application/json", }, }); @@ -34,21 +35,21 @@ export const connectToGrok = async ( const messageQueue: RawData[] = []; let createdSent = false; - let outputTranscript = ''; - let audioRemainder = Buffer.alloc(0); + let outputTranscript = ""; const sendResponseCreated = async () => { try { const device = await getDeviceInfo(supabase, user.user_id); + opus.reset(); ws.send( JSON.stringify({ - type: 'server', - msg: 'RESPONSE.CREATED', + type: "server", + msg: "RESPONSE.CREATED", volume_control: device?.volume ?? 100, }), ); } catch { - ws.send(JSON.stringify({ type: 'server', msg: 'RESPONSE.CREATED' })); + ws.send(JSON.stringify({ type: "server", msg: "RESPONSE.CREATED" })); } }; @@ -56,30 +57,30 @@ export const connectToGrok = async ( if (!firstMessage) return; grokWs.send( JSON.stringify({ - type: 'conversation.item.create', + type: "conversation.item.create", item: { - type: 'message', - role: 'user', - content: [{ type: 'input_text', text: firstMessage }], + type: "message", + role: "user", + content: [{ type: "input_text", text: firstMessage }], }, }), ); - grokWs.send(JSON.stringify({ type: 'response.create' })); + grokWs.send(JSON.stringify({ type: "response.create" })); }; - grokWs.on('open', () => { + grokWs.on("open", () => { isConnected = true; grokWs.send( JSON.stringify({ - type: 'session.update', + type: "session.update", session: { voice, instructions: systemPrompt, turn_detection: { type: "server_vad" }, audio: { - input: { format: { type: 'audio/pcm', rate: 16000 } }, - output: { format: { type: 'audio/pcm', rate: 24000 } }, + input: { format: { type: "audio/pcm", rate: 16000 } }, + output: { format: { type: "audio/pcm", rate: 24000 } }, }, }, }), @@ -95,91 +96,84 @@ export const connectToGrok = async ( } }); - grokWs.on('message', async (data: Buffer) => { + grokWs.on("message", async (data: Buffer) => { let event: any; try { - event = JSON.parse(data.toString('utf-8')); + event = JSON.parse(data.toString("utf-8")); } catch { return; } try { switch (event.type) { - case 'response.created': + case "response.created": if (!createdSent) { await sendResponseCreated(); createdSent = true; } break; - case 'response.output_audio_transcript.delta': - if (typeof event.delta === 'string') { + case "response.output_audio_transcript.delta": + if (typeof event.delta === "string") { outputTranscript += event.delta; } break; - case 'response.output_audio.delta': - if (typeof event.delta === 'string') { - const pcmChunk = Buffer.from(event.delta, 'base64'); - audioRemainder = Buffer.concat([audioRemainder, pcmChunk]); - - while (audioRemainder.length >= FRAME_SIZE) { - const frame = audioRemainder.subarray(0, FRAME_SIZE); - audioRemainder = audioRemainder.subarray(FRAME_SIZE); - try { - const packet = encoder.encode(frame); - ws.send(packet); - } catch { - // Skip frame - } - } + case "response.output_audio.delta": + if (typeof event.delta === "string") { + const pcmChunk = Buffer.from(event.delta, "base64"); + // Use Opus packetizer to encode and send audio + opus.push(pcmChunk); } break; - case 'conversation.item.input_audio_transcription.completed': - if (typeof event.transcript === 'string' && event.transcript.length > 0) { - await addConversation(supabase, 'user', event.transcript, user); + case "conversation.item.input_audio_transcription.completed": + if (typeof event.transcript === "string" && event.transcript.length > 0) { + await addConversation(supabase, "user", event.transcript, user); } break; - case 'input_audio_buffer.committed': - ws.send(JSON.stringify({ type: 'server', msg: 'AUDIO.COMMITTED' })); + case "input_audio_buffer.committed": + ws.send(JSON.stringify({ type: "server", msg: "AUDIO.COMMITTED" })); break; - case 'response.done': + case "response.done": + // Flush any remaining audio + opus.flush(true); + if (outputTranscript) { - await addConversation(supabase, 'assistant', outputTranscript, user); - outputTranscript = ''; + await addConversation(supabase, "assistant", outputTranscript, user); + outputTranscript = ""; } - ws.send(JSON.stringify({ type: 'server', msg: 'RESPONSE.COMPLETE' })); + ws.send(JSON.stringify({ type: "server", msg: "RESPONSE.COMPLETE" })); createdSent = false; break; - case 'error': - ws.send(JSON.stringify({ type: 'server', msg: 'RESPONSE.ERROR' })); + case "error": + ws.send(JSON.stringify({ type: "server", msg: "RESPONSE.ERROR" })); createdSent = false; break; } } catch (err) { - console.error('Error processing Grok event:', err); - ws.send(JSON.stringify({ type: 'server', msg: 'RESPONSE.ERROR' })); + console.error("Error processing Grok event:", err); + ws.send(JSON.stringify({ type: "server", msg: "RESPONSE.ERROR" })); createdSent = false; } }); - grokWs.on('close', () => { + grokWs.on("close", () => { ws.close(); }); - grokWs.on('error', (error: any) => { - console.error('Grok WebSocket error:', error); - ws.send(JSON.stringify({ type: 'server', msg: 'RESPONSE.ERROR' })); + grokWs.on("error", (error: any) => { + console.error("Grok WebSocket error:", error); + ws.send(JSON.stringify({ type: "server", msg: "RESPONSE.ERROR" })); }); const messageHandler = async (data: RawData, isBinary: boolean) => { if (isBinary) { - const base64Data = (data as Buffer).toString('base64'); - grokWs.send(JSON.stringify({ type: 'input_audio_buffer.append', audio: base64Data })); + const base64Data = (data as Buffer).toString("base64"); + grokWs.send(JSON.stringify({ type: "input_audio_buffer.append", audio: base64Data })); if (isDev && connectionPcmFile) { await connectionPcmFile.write(data as Buffer); @@ -189,23 +183,23 @@ export const connectToGrok = async ( let message: any; try { - message = JSON.parse((data as Buffer).toString('utf-8')); + message = JSON.parse((data as Buffer).toString("utf-8")); } catch { return; } - if (message?.type !== 'instruction') return; + if (message?.type !== "instruction") return; - if (message.msg === 'end_of_speech') { - grokWs.send(JSON.stringify({ type: 'input_audio_buffer.commit' })); - grokWs.send(JSON.stringify({ type: 'response.create' })); - grokWs.send(JSON.stringify({ type: 'input_audio_buffer.clear' })); - } else if (message.msg === 'INTERRUPT') { - grokWs.send(JSON.stringify({ type: 'input_audio_buffer.clear' })); + if (message.msg === "end_of_speech") { + grokWs.send(JSON.stringify({ type: "input_audio_buffer.commit" })); + grokWs.send(JSON.stringify({ type: "response.create" })); + grokWs.send(JSON.stringify({ type: "input_audio_buffer.clear" })); + } else if (message.msg === "INTERRUPT") { + grokWs.send(JSON.stringify({ type: "input_audio_buffer.clear" })); } }; - ws.on('message', (data: RawData, isBinary: boolean) => { + ws.on("message", (data: RawData, isBinary: boolean) => { if (!isConnected) { messageQueue.push(data); } else { @@ -213,13 +207,15 @@ export const connectToGrok = async ( } }); - ws.on('error', (error: any) => { - console.error('ESP32 WebSocket error:', error); + ws.on("error", (error: any) => { + console.error("ESP32 WebSocket error:", error); grokWs.close(); }); - ws.on('close', async (code: number, reason: string) => { + ws.on("close", async (code: number, reason: string) => { console.log(`ESP32 WebSocket closed with code ${code}, reason: ${reason}`); + await closeHandler(); + opus.close(); grokWs.close(); if (isDev && connectionPcmFile) { connectionPcmFile.close(); @@ -227,12 +223,12 @@ export const connectToGrok = async ( }); return new Promise((resolve, reject) => { - const timeout = setTimeout(() => reject(new Error('Grok connection timeout')), 10000); - grokWs.on('open', () => { + const timeout = setTimeout(() => reject(new Error("Grok connection timeout")), 10000); + grokWs.on("open", () => { clearTimeout(timeout); resolve(); }); - grokWs.on('error', (error: any) => { + grokWs.on("error", (error: any) => { clearTimeout(timeout); reject(error); }); diff --git a/server-deno/models/hume.ts b/server-deno/models/hume.ts index 0d48082..9127cb4 100644 --- a/server-deno/models/hume.ts +++ b/server-deno/models/hume.ts @@ -1,317 +1,302 @@ -import { Buffer } from 'node:buffer'; -import type { RawData } from 'npm:@types/ws'; -import { WebSocket } from 'npm:ws'; -import { addConversation, getDeviceInfo } from '../supabase.ts'; -import { encoder, FRAME_SIZE, isDev, humeApiKey, downsamplePcm, extractPcmFromWav, boostLimitPCM16LEInPlace } from '../utils.ts'; +import { Buffer } from "node:buffer"; +import type { RawData } from "npm:@types/ws"; +import { WebSocket } from "npm:ws"; +import { addConversation, getDeviceInfo } from "../supabase.ts"; +import { createOpusPacketizer, isDev, humeApiKey, downsamplePcm, extractPcmFromWav, boostLimitPCM16LEInPlace } from "../utils.ts"; -export const connectToHume = ( - ws: WebSocket, - payload: IPayload, - connectionPcmFile:Deno.FsFile | null, - firstMessage: string, - systemPrompt: string, - closeHandler: () => Promise, -) => { - const { user, supabase } = payload; - const { personality } = user; +export const connectToHume = ({ + ws, + payload, + connectionPcmFile, + firstMessage, + systemPrompt, + closeHandler, +}: ProviderArgs) => { + const { user, supabase } = payload; + const { personality } = user; - console.log(`Connecting to Hume with key "${humeApiKey?.slice(0, 3)}..."`); + const opus = createOpusPacketizer((packet) => ws.send(packet)); - // Build Hume WebSocket URL - const queryParams = new URLSearchParams({ - api_key: humeApiKey!, - config_id: personality!.oai_voice, - }); + console.log(`Connecting to Hume with key "${humeApiKey?.slice(0, 3)}..."`); - const humeWsUrl = `wss://api.hume.ai/v0/evi/chat?${queryParams.toString()}`; - - console.log(`Connecting to Hume WebSocket at: ${humeWsUrl.replace(humeApiKey!, 'API_KEY_HIDDEN')}`); - const humeWs = new WebSocket(humeWsUrl); - - let isConnected = false; - const messageQueue: RawData[] = []; - let createdSent = false; - - // Handle Hume WebSocket connection - humeWs.on('open', () => { - console.log('✅ Connected to Hume WebSocket API successfully'); - isConnected = true; - - // Configure Hume session settings for input audio format - // This tells Hume what format we're sending TO them, not what we want back - humeWs.send(JSON.stringify({ - type: 'session_settings', - audio: { - encoding: "linear16", - channels: 1, - sample_rate: 16000, - }, - system_prompt: systemPrompt, - })); - - // Send simple first message if provided - humeWs.send(JSON.stringify({ - type: 'user_input', - text: firstMessage, - })); - - // Process queued messages - while (messageQueue.length > 0) { - const queuedMessage = messageQueue.shift(); - if (queuedMessage) { - messageHandler(queuedMessage, true); // Assume binary for queued audio - } - } - }); - - // Handle messages from Hume - humeWs.on('message', async (data: Buffer) => { - try { - const message: HumeMessage = JSON.parse(data.toString()); - console.log(`Received from Hume: ${message.type}`); - - switch (message.type) { - case 'assistant_end': - - // Send RESPONSE.COMPLETE when assistant message is done - ws.send(JSON.stringify({ - type: 'server', - msg: 'RESPONSE.COMPLETE', - })); - - // Reset for next turn - createdSent = false; - - break; - - case 'assistant_message': - const assistantMsg = message as HumeAssistantMessage; - - // Store conversation in database - await addConversation( - supabase, - 'assistant', - assistantMsg.message.content, - user, - ); - - break; - - case 'audio_output': - const audioMsg = message as HumeAudioOutput; - - // Send RESPONSE.CREATED before first audio chunk - if (!createdSent) { - try { - const device = await getDeviceInfo(supabase, user.user_id); - - if (device) { - ws.send(JSON.stringify({ - type: 'server', - msg: 'RESPONSE.CREATED', - volume_control: device.volume ?? 70, - })); - } else { - ws.send(JSON.stringify({ - type: 'server', - msg: 'RESPONSE.CREATED', - })); - } - } catch (error) { - console.error('Error fetching device info:', error); - ws.send(JSON.stringify({ - type: 'server', - msg: 'RESPONSE.CREATED', - })); - } - createdSent = true; - } - - try { - // Decode base64 audio data from Hume (this is a WAV file, not raw PCM!) - const wavBuffer = Buffer.from(audioMsg.data, 'base64'); - - // Extract PCM data from WAV file - const pcmData = extractPcmFromWav(wavBuffer); - - if (!pcmData) { - console.error('Failed to extract PCM data from WAV'); - return; - } - - // Downsample from 48kHz to 24kHz to match our system - const downsampledPcm = downsamplePcm(pcmData, 48000, 24000); - boostLimitPCM16LEInPlace(downsampledPcm, /*gainDb=*/6.0, /*ceiling=*/0.89); - - - // Process the downsampled PCM data in frames - let audioBuffer = downsampledPcm; - - // Process complete frames using the standard FRAME_SIZE - while (audioBuffer.length >= FRAME_SIZE) { - const frame = audioBuffer.subarray(0, FRAME_SIZE); - audioBuffer = audioBuffer.subarray(FRAME_SIZE); - - try { - const packet = encoder.encode(frame); - ws.send(packet); - } catch (err) { - console.error('Hume Opus encode failed:', err); - } - } - - // Store remaining bytes for next chunk (if any) - if (audioBuffer.length > 0) { - console.log(`Hume audio remainder: ${audioBuffer.length} bytes`); - } - } catch (audioError) { - console.error('Error processing Hume audio output:', audioError); - } - break; - - case 'chat_metadata': - console.log('Chat metadata received:', message); - break; - - case 'user_message': - console.log('User message acknowledged:', message); - await addConversation( - supabase, - 'user', - message.message.content, - user, - ); - break; - - case 'user_input': - // This is an echo of our own input, we can log it but don't need to store it again - console.log('User input acknowledged by Hume'); - break; - - case 'error': - const errorMsg = message as HumeError; - console.error(`Hume error: ${errorMsg.code} - ${errorMsg.message}`); - - ws.send(JSON.stringify({ - type: 'server', - msg: 'RESPONSE.ERROR', - error: errorMsg.message, - })); - break; - - case 'session_created': - console.log('Hume session created'); - ws.send(JSON.stringify({ - type: 'server', - msg: 'SESSION.CREATED', - })); - break; - - case 'session_ended': - console.log('Hume session ended'); - ws.send(JSON.stringify({ - type: 'server', - msg: 'SESSION.END', - })); - break; - - default: - console.log(`Unhandled Hume message type: ${message.type}`); - } - } catch (error) { - console.error('Error processing Hume message:', error); - } - }); - - humeWs.on('close', (code: number, reason: Buffer) => { - console.log(`Hume WebSocket closed: ${code} - ${reason.toString()}`); - ws.send(JSON.stringify({ - type: 'server', - msg: 'SESSION.END', - })); - isConnected = false; - ws.close(); - }); - - humeWs.on('error', (error: Error) => { - console.error('Hume WebSocket error:', error); - console.error('Error details:', { - message: error.message, - stack: error.stack, - name: error.name + // Build Hume WebSocket URL + const configId = personality?.oai_voice ?? ""; + const queryParams = new URLSearchParams({ + api_key: humeApiKey!, + config_id: configId, }); - ws.send(JSON.stringify({ - type: 'server', - msg: 'RESPONSE.ERROR', - error: 'Connection to Hume failed', - })); - }); - // Handle messages from ESP32 client - const messageHandler = async (data: RawData, isBinary: boolean) => { - try { - if (isBinary) { - // Handle audio data from ESP32 - const base64Audio = data.toString('base64'); + const humeWsUrl = `wss://api.hume.ai/v0/evi/chat?${queryParams.toString()}`; - const audioMessage: HumeAudioInput = { - type: 'audio_input', - data: base64Audio, - }; + console.log(`Connecting to Hume WebSocket at: ${humeWsUrl.replace(humeApiKey!, "API_KEY_HIDDEN")}`); + const humeWs = new WebSocket(humeWsUrl); - if (isConnected) { - humeWs.send(JSON.stringify(audioMessage)); + let isConnected = false; + const messageQueue: RawData[] = []; + let createdSent = false; + + // Handle Hume WebSocket connection + humeWs.on("open", () => { + console.log("✅ Connected to Hume WebSocket API successfully"); + isConnected = true; + + // Configure Hume session settings for input audio format + humeWs.send(JSON.stringify({ + type: "session_settings", + audio: { + encoding: "linear16", + channels: 1, + sample_rate: 16000, + }, + system_prompt: systemPrompt, + })); + + // Send simple first message if provided + humeWs.send(JSON.stringify({ + type: "user_input", + text: firstMessage, + })); + + // Process queued messages + while (messageQueue.length > 0) { + const queuedMessage = messageQueue.shift(); + if (queuedMessage) { + messageHandler(queuedMessage, true); // Assume binary for queued audio + } } + }); + + // Handle messages from Hume + humeWs.on("message", async (data: Buffer) => { + try { + const message: HumeMessage = JSON.parse(data.toString()); + console.log(`Received from Hume: ${message.type}`); + + switch (message.type) { + case "assistant_end": + // Flush any remaining audio + opus.flush(true); + + // Send RESPONSE.COMPLETE when assistant message is done + ws.send(JSON.stringify({ + type: "server", + msg: "RESPONSE.COMPLETE", + })); + + // Reset for next turn + createdSent = false; + break; + + case "assistant_message": + const assistantMsg = message as HumeAssistantMessage; + + // Store conversation in database + await addConversation( + supabase, + "assistant", + assistantMsg.message.content, + user, + ); + break; + + case "audio_output": + const audioMsg = message as HumeAudioOutput; + + // Send RESPONSE.CREATED before first audio chunk + if (!createdSent) { + try { + const device = await getDeviceInfo(supabase, user.user_id); + opus.reset(); + + if (device) { + ws.send(JSON.stringify({ + type: "server", + msg: "RESPONSE.CREATED", + volume_control: device.volume ?? 70, + })); + } else { + ws.send(JSON.stringify({ + type: "server", + msg: "RESPONSE.CREATED", + })); + } + } catch (error) { + console.error("Error fetching device info:", error); + ws.send(JSON.stringify({ + type: "server", + msg: "RESPONSE.CREATED", + })); + } + createdSent = true; + } + + try { + // Decode base64 audio data from Hume (this is a WAV file, not raw PCM!) + const wavBuffer = Buffer.from(audioMsg.data, "base64"); + + // Extract PCM data from WAV file + const pcmData = extractPcmFromWav(wavBuffer); + + if (!pcmData) { + console.error("Failed to extract PCM data from WAV"); + return; + } + + // Downsample from 48kHz to 24kHz to match our system + const downsampledPcm = downsamplePcm(pcmData, 48000, 24000); + boostLimitPCM16LEInPlace(downsampledPcm, /*gainDb=*/6.0, /*ceiling=*/0.89); + + // Use Opus packetizer to encode and send audio + opus.push(downsampledPcm); + } catch (audioError) { + console.error("Error processing Hume audio output:", audioError); + } + break; + + case "chat_metadata": + console.log("Chat metadata received:", message); + break; + + case "user_message": + console.log("User message acknowledged:", message); + await addConversation( + supabase, + "user", + message.message.content, + user, + ); + break; + + case "user_input": + // This is an echo of our own input, we can log it but don't need to store it again + console.log("User input acknowledged by Hume"); + break; + + case "error": + const errorMsg = message as HumeError; + console.error(`Hume error: ${errorMsg.code} - ${errorMsg.message}`); + + ws.send(JSON.stringify({ + type: "server", + msg: "RESPONSE.ERROR", + error: errorMsg.message, + })); + break; + + case "session_created": + console.log("Hume session created"); + ws.send(JSON.stringify({ + type: "server", + msg: "SESSION.CREATED", + })); + break; + + case "session_ended": + console.log("Hume session ended"); + ws.send(JSON.stringify({ + type: "server", + msg: "SESSION.END", + })); + break; + + default: + console.log(`Unhandled Hume message type: ${message.type}`); + } + } catch (error) { + console.error("Error processing Hume message:", error); + } + }); + + humeWs.on("close", (code: number, reason: Buffer) => { + console.log(`Hume WebSocket closed: ${code} - ${reason.toString()}`); + ws.send(JSON.stringify({ + type: "server", + msg: "SESSION.END", + })); + isConnected = false; + ws.close(); + }); + + humeWs.on("error", (error: Error) => { + console.error("Hume WebSocket error:", error); + console.error("Error details:", { + message: error.message, + stack: error.stack, + name: error.name + }); + ws.send(JSON.stringify({ + type: "server", + msg: "RESPONSE.ERROR", + error: "Connection to Hume failed", + })); + }); + + // Handle messages from ESP32 client + const messageHandler = async (data: RawData, isBinary: boolean) => { + try { + if (isBinary) { + // Handle audio data from ESP32 + const base64Audio = data.toString("base64"); + + const audioMessage: HumeAudioInput = { + type: "audio_input", + data: base64Audio, + }; + + if (isConnected) { + humeWs.send(JSON.stringify(audioMessage)); + } + + // Write to debug file if enabled + if (isDev && connectionPcmFile) { + await connectionPcmFile.write(data as Buffer); + } + } + } catch (error) { + console.error("Error handling message:", error); + } + }; + + // Set up ESP32 WebSocket handlers + ws.on("message", (data: RawData, isBinary: boolean) => { + if (!isConnected) { + messageQueue.push(data); + } else { + messageHandler(data, isBinary); + } + }); + + ws.on("error", (error: Error) => { + console.error("ESP32 WebSocket error:", error); + humeWs.close(); + }); + + ws.on("close", async (code: number, reason: string) => { + console.log(`ESP32 WebSocket closed: ${code} - ${reason}`); + await closeHandler(); + opus.close(); + humeWs.close(); - // Write to debug file if enabled if (isDev && connectionPcmFile) { - await connectionPcmFile.write(data as Buffer); + connectionPcmFile.close(); + console.log("Closed debug audio file"); } - } - } catch (error) { - console.error('Error handling message:', error); - } - }; - - // Set up ESP32 WebSocket handlers - ws.on('message', (data: RawData, isBinary: boolean) => { - if (!isConnected) { - messageQueue.push(data); - } else { - messageHandler(data, isBinary); - } - }); - - ws.on('error', (error: Error) => { - console.error('ESP32 WebSocket error:', error); - humeWs.close(); - }); - - ws.on('close', async (code: number, reason: string) => { - console.log(`ESP32 WebSocket closed: ${code} - ${reason}`); - humeWs.close(); - await closeHandler(); - - if (isDev && connectionPcmFile) { - connectionPcmFile.close(); - console.log('Closed debug audio file'); - } - }); - - // Wait for Hume connection to be established - return new Promise((resolve, reject) => { - const timeout = setTimeout(() => { - reject(new Error('Hume connection timeout')); - }, 10000); - - humeWs.on('open', () => { - clearTimeout(timeout); - resolve(); }); - humeWs.on('error', (error) => { - clearTimeout(timeout); - reject(error); + // Wait for Hume connection to be established + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + reject(new Error("Hume connection timeout")); + }, 10000); + + humeWs.on("open", () => { + clearTimeout(timeout); + resolve(); + }); + + humeWs.on("error", (error) => { + clearTimeout(timeout); + reject(error); + }); }); - }); }; diff --git a/server-deno/models/openai.ts b/server-deno/models/openai.ts index cbc1664..3a557c8 100644 --- a/server-deno/models/openai.ts +++ b/server-deno/models/openai.ts @@ -1,364 +1,332 @@ import { Buffer } from "node:buffer"; -import type { - RawData, - WebSocket as WSWebSocket, - WebSocketServer as _WebSocketServer, -} from "npm:@types/ws"; - +import type { RawData } from "npm:@types/ws"; import { RealtimeClient } from "../realtime/client.js"; import { RealtimeUtils } from "../realtime/utils.js"; import { addConversation, getDeviceInfo } from "../supabase.ts"; -import { encoder, FRAME_SIZE, isDev, openaiApiKey } from "../utils.ts"; +import { createOpusPacketizer, isDev, openaiApiKey, defaultOpenAIVoice } from "../utils.ts"; const sendFirstMessage = (client: RealtimeClient, firstMessage: string) => { - const event = { - event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID - type: "conversation.item.create", - previous_item_id: "root", - item: { - type: "message", - role: "system", - content: [{ - type: "input_text", - text: firstMessage, - }], - }, - }; + const event = { + event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID + type: "conversation.item.create", + previous_item_id: "root", + item: { + type: "message", + role: "system", + content: [{ + type: "input_text", + text: firstMessage, + }], + }, + }; - client.realtime.send(event.type, event); - client.realtime.send("response.create", { - event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID - type: "response.create", - }); + client.realtime.send(event.type, event); + client.realtime.send("response.create", { + event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID + type: "response.create", + }); }; -export const connectToOpenAI = async ( - ws: WebSocket, - payload: IPayload, - connectionPcmFile: Deno.FsFile | null, - firstMessage: string, - systemPrompt: string, -) => { - const { user, supabase } = payload; +export const connectToOpenAI = async ({ + ws, + payload, + connectionPcmFile, + firstMessage, + systemPrompt, + closeHandler, +}: ProviderArgs) => { + const { user, supabase } = payload; - let currentItemId: string | null = null; - let currentCallId: string | null = null; + const opus = createOpusPacketizer((packet) => ws.send(packet)); - // Instantiate new client - console.log(`Connecting with key "${openaiApiKey.slice(0, 3)}..."`); - const client = new RealtimeClient({ apiKey: openaiApiKey }); + let currentItemId: string | null = null; + let currentCallId: string | null = null; - // ADD TOOL CALLS HERE - client.addTool( - { - type: "function", - name: "end_session", - description: - 'Call this if the user says bye or needs to leave or suggests they want to end the session. (e.g. "I gotta to go", "I have to work", "I have to sleep", "I have to do something else")', - parameters: { - type: "object", - strict: true, - properties: { - reason: { - type: "string", - description: "Short reason for ending the session.", - }, - }, - required: ["reason"], - }, - }, - (args: any) => { - console.log("end session", args); + // Instantiate new client + console.log(`Connecting with key "${openaiApiKey?.slice(0, 3)}..."`); + const client = new RealtimeClient({ apiKey: openaiApiKey }); - // Send your custom message to the client - ws.send(JSON.stringify({ type: "server", msg: "SESSION.END" })); + // ADD TOOL CALLS HERE + client.addTool( + { + type: "function", + name: "end_session", + description: + 'Call this if the user says bye or needs to leave or suggests they want to end the session. (e.g. "I gotta to go", "I have to work", "I have to sleep", "I have to do something else")', + parameters: { + type: "object", + strict: true, + properties: { + reason: { + type: "string", + description: "Short reason for ending the session.", + }, + }, + required: ["reason"], + }, + }, + (args: any) => { + console.log("end session", args); - // Send the function result back to OpenAI - const functionResult = { - event_id: RealtimeUtils.generateId("evt_"), - type: "conversation.item.create", - item: { - id: RealtimeUtils.generateId("item_"), - type: "function_call_output", - call_id: currentCallId, - output: JSON.stringify({ - success: true, - message: `Session ended: ${args.reason}`, - }), - }, - }; + // Send your custom message to the client + ws.send(JSON.stringify({ type: "server", msg: "SESSION.END" })); - client.realtime.send(functionResult.type, functionResult); + // Return the result for the callback + return { success: true, message: `Session ended: ${args.reason}` }; + }, + ); - // Return the result for the callback - return { success: true, message: `Session ended: ${args.reason}` }; - }, - ); + // Relay: OpenAI Realtime API Event -> Browser Event + client.realtime.on("server.*", async (event: any) => { + // Check if the event is session.created + if (event.type === "session.created") { + console.log("session created", event); + sendFirstMessage(client, firstMessage); + } else if (event.type === "session.updated") { + console.log("session updated", event); + } else if (event.type === "error") { + console.log("error", event); + } else if (event.type === "response.done") { + console.log("response.done", event); + const hasNoAudio = event.response?.usage?.output_token_details?.audio_tokens === 0; + opus.flush(true); + if (!hasNoAudio) { + ws.send( + JSON.stringify({ + type: "server", + msg: "RESPONSE.COMPLETE", + }), + ); + } + } else if (event.type === "response.audio_transcript.done") { + console.log("response.audio_transcript.done", event); + await addConversation( + supabase, + "assistant", + event.transcript, + user, + ); + } else if (event.type === "input_audio_buffer.committed") { + ws.send(JSON.stringify({ type: "server", msg: "AUDIO.COMMITTED" })); + } - // Relay: OpenAI Realtime API Event -> Browser Event - client.realtime.on("server.*", async (event: any) => { - // console.log(`Relaying "${event.type}" to Client`); - // Check if the event is session.created - if (event.type === "session.created") { - console.log("session created", event); - sendFirstMessage(client, firstMessage); - } else if (event.type === "session.updated") { - console.log("session updated", event); - } else if (event.type === "error") { - console.log("error", event); - } else if (event.type === "response.done") { - // Fetch the latest device info when response is complete - try { - const device = await getDeviceInfo(supabase, user.user_id); + if (event.type in client.conversation.EventProcessors) { + try { + switch (event.type) { + case "response.created": + console.log("response.created", event); + opus.reset(); + try { + const device = await getDeviceInfo(supabase, user.user_id); - if (device) { - // Send the updated volume data along with the response complete message - ws.send(JSON.stringify({ - type: "server", - msg: "RESPONSE.COMPLETE", - volume_control: device.volume ?? 100, - })); - } else { - // Fall back to just sending the complete message if there's an error - ws.send( - JSON.stringify({ - type: "server", - msg: "RESPONSE.COMPLETE", - }), - ); - } - } catch (error) { - console.error("Error fetching updated device info:", error); - ws.send( - JSON.stringify({ - type: "server", - msg: "RESPONSE.COMPLETE", - }), - ); - } - } else if (event.type === "response.audio_transcript.done") { - console.log("response.audio_transcript.done", event); - await addConversation( - supabase, - "assistant", - event.transcript, - user, - ); - } else if (event.type === "input_audio_buffer.committed") { - ws.send(JSON.stringify({ type: "server", msg: "AUDIO.COMMITTED" })); - } + if (device) { + // Send the updated volume data along with the response complete message + ws.send(JSON.stringify({ + type: "server", + msg: "RESPONSE.CREATED", + volume_control: device.volume ?? 100, + })); + } else { + // Fall back to just sending the complete message if there's an error + ws.send( + JSON.stringify({ + type: "server", + msg: "RESPONSE.CREATED", + }), + ); + } + } catch (error) { + console.error("Error fetching updated device info:", error); + ws.send( + JSON.stringify({ + type: "server", + msg: "RESPONSE.CREATED", + }), + ); + } + break; + case "response.output_item.added": + console.log("response.output_item.added", event); + if (event.item.id) { + console.log("foobar", event.item.id); + currentItemId = event.item.id; + currentCallId = event.item.call_id; + } + break; + case "response.audio.delta": + { + const { delta } = client.conversation.processEvent( + event, + ); + try { + if (delta?.audio?.buffer) { + const pcmBuffer = Buffer.from( + delta.audio.buffer, + ); + opus.push(pcmBuffer); + } + } catch (audioError) { + console.error( + "Error processing audio delta:", + audioError, + ); + // Don't send any audio data if there's an error at this level + } + } + break; + case "conversation.item.created": + console.log("user said: ", event.item); + break; + case "conversation.item.input_audio_transcription.completed": + console.log("user transcription:", event); + await addConversation( + supabase, + "user", + event.transcript, + user, + ); + break; + } + } catch (error) { + console.error("Error processing event:", error); + console.error("Event that caused the error:", event); + ws.send( + JSON.stringify({ type: "server", msg: "RESPONSE.ERROR" }), + ); + } + } + }); - if (event.type in client.conversation.EventProcessors) { - try { - switch (event.type) { - case "response.created": - console.log("response.created", event); - ws.send( - JSON.stringify({ - type: "server", - msg: "RESPONSE.CREATED", - }), - ); - break; - case "response.output_item.added": - console.log("response.output_item.added", event); - if (event.item.id) { - console.log("foobar", event.item.id); - currentItemId = event.item.id; - currentCallId = event.item.call_id; - } - break; - case "response.audio.delta": - { - const { delta } = client.conversation.processEvent( - event, - ); - try { - if (delta?.audio?.buffer) { - const pcmBuffer = Buffer.from( - delta.audio.buffer, - ); - for ( - let offset = 0; - offset < pcmBuffer.length; - offset += FRAME_SIZE - ) { - // Get one frame of PCM data. - const frame = pcmBuffer.subarray( - offset, - offset + FRAME_SIZE, - ); + client.realtime.on("close", () => ws.close()); - try { - const encodedPacket = encoder - .encode(frame); - ws.send(encodedPacket); - } catch (_e) { - // Skip this frame but continue with others - } - } - } - } catch (audioError) { - console.error( - "Error processing audio delta:", - audioError, - ); - // Don't send any audio data if there's an error at this level - } - } - break; - case "conversation.item.created": - console.log("user said: ", event.item); - break; - case "conversation.item.input_audio_transcription.completed": - console.log("user transcription:", event); - await addConversation( - supabase, - "user", - event.transcript, - user, - ); - break; - } - } catch (error) { - console.error("Error processing event:", error); - console.error("Event that caused the error:", event); - ws.send( - JSON.stringify({ type: "server", msg: "RESPONSE.ERROR" }), - ); - } - } - }); + // Relay: Browser Event -> OpenAI Realtime API Event + // We need to queue data waiting for the OpenAI connection + const messageQueue: RawData[] = []; - client.realtime.on("close", () => ws.close()); + const messageHandler = async (data: any, isBinary: boolean) => { + try { + let event; - // Relay: Browser Event -> OpenAI Realtime API Event - // We need to queue data waiting for the OpenAI connection - const messageQueue: RawData[] = []; + // for esp32 + if (isBinary) { + const base64Data = data.toString("base64"); - const messageHandler = async (data: any, isBinary: boolean) => { - try { - let event; + // Convert binary PCM16 data to base64 for OpenAI Realtime API + event = { + event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID + type: "input_audio_buffer.append", + audio: base64Data, + }; + // Write the raw PCM data to file for debugging if enabled. + // Also write the base64 data to a separate file + if (isDev) { + if (connectionPcmFile) { + await connectionPcmFile.write(data); + } + } + client.realtime.send(event.type, event); + } else { // Manual VAD + const message = JSON.parse(data.toString("utf-8")); - // for esp32 - if (isBinary) { - const base64Data = data.toString("base64"); + // commit user audio and create response + if ( + message.type === "instruction" && + message.msg === "end_of_speech" + ) { + console.log("end_of_speech detected"); - // Convert binary PCM16 data to base64 for OpenAI Realtime API - event = { - event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID - type: "input_audio_buffer.append", - audio: base64Data, - }; - // Write the raw PCM data to file for debugging if enabled. - // Also write the base64 data to a separate file - if (isDev) { - if (connectionPcmFile) { - await connectionPcmFile.write(data); - } - } - client.realtime.send(event.type, event); - } else { // Manual VAD - const message = JSON.parse(data.toString("utf-8")); + client.realtime.send("input_audio_buffer.commit", { + event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID + type: "input_audio_buffer.commit", + }); - // commit user audio and create response - if ( - message.type === "instruction" && - message.msg === "end_of_speech" - ) { - console.log("end_of_speech detected"); + client.realtime.send("response.create", { + event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID + type: "response.create", + }); - client.realtime.send("input_audio_buffer.commit", { - event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID - type: "input_audio_buffer.commit", - }); + client.realtime.send("input_audio_buffer.clear", { + event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID + type: "input_audio_buffer.clear", + }); + } else if ( + message.type === "instruction" && + message.msg === "INTERRUPT" + ) { + console.log("interrupt detected", message); + const audioEndMs = message.audio_end_ms; - client.realtime.send("response.create", { - event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID - type: "response.create", - }); + client.realtime.send("conversation.item.truncate", { + event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID + type: "conversation.item.truncate", + item_id: currentItemId, + content_index: 0, + audio_end_ms: audioEndMs, + }); - client.realtime.send("input_audio_buffer.clear", { - event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID - type: "input_audio_buffer.clear", - }); - } else if ( - message.type === "instruction" && - message.msg === "INTERRUPT" - ) { - console.log("interrupt detected", message); - const audioEndMs = message.audio_end_ms; + client.realtime.send("input_audio_buffer.clear", { + event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID + type: "input_audio_buffer.clear", + }); + } + } + } catch (e: unknown) { + console.error((e as Error).message); + console.log(`Error parsing event from client: ${data}`); + } + }; - client.realtime.send("conversation.item.truncate", { - event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID - type: "conversation.item.truncate", - item_id: currentItemId, - content_index: 0, - audio_end_ms: audioEndMs, - }); + ws.on("message", (data: any, isBinary: boolean) => { + if (!client.isConnected()) { + messageQueue.push(data); + } else { + messageHandler(data, isBinary); + } + }); - client.realtime.send("input_audio_buffer.clear", { - event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID - type: "input_audio_buffer.clear", - }); - } - } - } catch (e: unknown) { - console.error((e as Error).message); - console.log(`Error parsing event from client: ${data}`); - } - }; + // Add error handler + ws.on("error", (error: any) => { + console.error("WebSocket error:", error); + client.disconnect(); + }); - ws.on("message", (data: any, isBinary: boolean) => { - if (!client.isConnected()) { - messageQueue.push(data); - } else { - messageHandler(data, isBinary); - } - }); + // Add more detailed close handling + ws.on("close", async (code: number, reason: string) => { + console.log(`WebSocket closed with code ${code}, reason: ${reason}`); + await closeHandler(); + opus.close(); + client.disconnect(); + if (isDev) { + if (connectionPcmFile) { + connectionPcmFile.close(); + console.log(`Closed debug audio file.`); + } + } + }); - // Add error handler - ws.on("error", (error: any) => { - console.error("WebSocket error:", error); - client.disconnect(); - }); - - // Add more detailed close handling - ws.on("close", async (code: number, reason: string) => { - console.log(`WebSocket closed with code ${code}, reason: ${reason}`); - client.disconnect(); - if (isDev) { - if (connectionPcmFile) { - connectionPcmFile.close(); - console.log(`Closed debug audio file.`); - } - } - }); - - // Connect to the OpenAI Realtime API - try { - console.log(`Connecting to OpenAI...`); - const sessionOptions = { - model: "gpt-4o-mini-realtime-preview-2024-12-17", - // turn_detection: null, - turn_detection: { - type: "server_vad", - threshold: 0.4, - prefix_padding_ms: 400, - silence_duration_ms: 1000, - }, - voice: user.personality?.oai_voice ?? "ash", - instructions: systemPrompt, - input_audio_transcription: { model: "whisper-1" }, - }; - await client.connect(sessionOptions as any); - } catch (e: unknown) { - console.log(`Error connecting to OpenAI: ${e as Error}`); - ws.close(); - return; - } - console.log(`Connected to OpenAI successfully!`); - while (messageQueue.length) { - messageHandler(messageQueue.shift(), false); - } + // Connect to the OpenAI Realtime API + try { + console.log(`Connecting to OpenAI...`); + const sessionOptions = { + model: "gpt-4o-mini-realtime-preview-2024-12-17", + turn_detection: { + type: "server_vad", + threshold: 0.4, + prefix_padding_ms: 400, + silence_duration_ms: 1000, + }, + voice: user.personality?.oai_voice ?? defaultOpenAIVoice, + instructions: systemPrompt, + input_audio_transcription: { model: "whisper-1" }, + }; + await client.connect(sessionOptions as any); + } catch (e: unknown) { + console.log(`Error connecting to OpenAI: ${e as Error}`); + ws.close(); + return; + } + console.log(`Connected to OpenAI successfully!`); + while (messageQueue.length) { + messageHandler(messageQueue.shift(), false); + } }; diff --git a/server-deno/types.d.ts b/server-deno/types.d.ts index 28cbce8..3938370 100644 --- a/server-deno/types.d.ts +++ b/server-deno/types.d.ts @@ -199,4 +199,13 @@ declare global { code: string; message: string; } + + interface ProviderArgs { + ws: WebSocket; + payload: IPayload; + connectionPcmFile: Deno.FsFile | null; + firstMessage: string; + systemPrompt: string; + closeHandler: () => Promise; + } } diff --git a/server-deno/utils.ts b/server-deno/utils.ts index a9a3c7a..ac2a230 100644 --- a/server-deno/utils.ts +++ b/server-deno/utils.ts @@ -7,6 +7,10 @@ import { Encoder } from "@evan/opus"; export const defaultVolume = 50; +export const defaultGeminiVoice = "Sadachbia"; +export const defaultOpenAIVoice = "ash"; +export const defaultGrokVoice = "Ara"; + // Define your audio parameters export const SAMPLE_RATE = 24000; // For example, 24000 Hz const CHANNELS = 1; // Mono (set to 2 if you have stereo) @@ -15,6 +19,79 @@ const BYTES_PER_SAMPLE = 2; // 16-bit PCM: 2 bytes per sample const FRAME_SIZE = (SAMPLE_RATE * FRAME_DURATION / 1000) * CHANNELS * BYTES_PER_SAMPLE; // 960 bytes for 24000 Hz mono 16-bit +export function createOpusEncoder() { + const enc = new Encoder({ + channels: CHANNELS, + sample_rate: SAMPLE_RATE, + application: "voip", + }); + + enc.expert_frame_duration = FRAME_DURATION; + enc.bitrate = 24000; + return enc; +} + +export function createOpusPacketizer( + sendPacket: (packet: Uint8Array) => void, +) { + const enc = createOpusEncoder(); + let pending = Buffer.alloc(0); + let closed = false; + + const push = (pcm: Uint8Array) => { + if (closed) return; + if (!pcm || pcm.length === 0) return; + + pending = Buffer.concat([pending, Buffer.from(pcm)]); + + while (pending.length >= FRAME_SIZE) { + const frame = pending.subarray(0, FRAME_SIZE); + pending = pending.subarray(FRAME_SIZE); + try { + const packet = enc.encode(frame); + sendPacket(packet); + } catch (err) { + console.error("Opus encode failed:", err); + } + } + }; + + const flush = (padFinalFrame = false) => { + if (closed) return; + if (pending.length === 0) return; + + if (!padFinalFrame) { + pending = Buffer.alloc(0); + return; + } + + const padded = Buffer.alloc(FRAME_SIZE); + pending.copy(padded, 0, 0, pending.length); + pending = Buffer.alloc(0); + + try { + const packet = enc.encode(padded); + sendPacket(packet); + } catch (err) { + console.error("Opus encode failed:", err); + } + }; + + const reset = () => { + pending = Buffer.alloc(0); + }; + + const close = () => { + closed = true; + pending = Buffer.alloc(0); + }; + + const bufferedBytes = () => pending.length; + + return { push, flush, reset, close, bufferedBytes }; +} + +// Legacy encoder for backwards compatibility during migration const encoder = new Encoder({ channels: CHANNELS, sample_rate: SAMPLE_RATE, @@ -22,7 +99,7 @@ const encoder = new Encoder({ }); encoder.expert_frame_duration = FRAME_DURATION; -encoder.bitrate = 12000; +encoder.bitrate = 24000; export const openaiApiKey = Deno.env.get("OPENAI_API_KEY"); export const geminiApiKey = Deno.env.get("GEMINI_API_KEY");