admin-valentin/AIvoices
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

opus permits large chunks

Browse source
This commit is contained in:
akdeb 2025-12-26 19:51:15 +07:00
parent 31b1ad6d58
commit f1ed5653d5
8 changed files with 1013 additions and 1029 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
server-deno/main.ts
View file

@ -59,40 +59,37 @@ wss.on("connection", async (ws: WSWebSocket, payload: IPayload) => {
ws.send( ws.send(
JSON.stringify({ JSON.stringify({
type: "auth", type: "auth",
volume_control: user.device?.volume ?? 100, volume_control: user.device?.volume ?? 20,
is_ota: user.device?.is_ota ?? false, is_ota: user.device?.is_ota ?? false,
is_reset: user.device?.is_reset ?? false, is_reset: user.device?.is_reset ?? false,
pitch_factor: user.personality?.pitch_factor ?? 1, 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) { switch (provider) {
case "openai": case "openai":
await connectToOpenAI( await connectToOpenAI(providerArgs);
ws,
payload,
connectionPcmFile,
firstMessage,
systemPrompt,
);
break; break;
case "gemini": case "gemini":
await connectToGemini( await connectToGemini(providerArgs);
ws,
payload,
connectionPcmFile,
firstMessage,
systemPrompt,
);
break; break;
case "grok": case "grok":
await connectToGrok( await connectToGrok(providerArgs);
ws,
payload,
connectionPcmFile,
firstMessage,
systemPrompt,
);
break; break;
case "elevenlabs": case "elevenlabs":
const agentId = user.personality?.oai_voice ?? ""; const agentId = user.personality?.oai_voice ?? "";
@ -107,11 +104,11 @@ wss.on("connection", async (ws: WSWebSocket, payload: IPayload) => {
connectionPcmFile, connectionPcmFile,
agentId, agentId,
elevenLabsApiKey, elevenLabsApiKey,
closeHandler,
); );
break; break;
case "hume": case "hume":
await connectToHume(ws, payload, await connectToHume(providerArgs);
connectionPcmFile, firstMessage, systemPrompt, () => Promise.resolve());
break; break;
default: default:
throw new Error(`Unknown provider: ${provider}`); throw new Error(`Unknown provider: ${provider}`);

52
server-deno/models/elevenlabs.ts
View file

@ -9,7 +9,7 @@ import {
} from "npm:@elevenlabs/client"; } from "npm:@elevenlabs/client";
import { addConversation, getDeviceInfo } from "../supabase.ts"; 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 // Calculate audio level for debugging
function calculateAudioLevel(audioData: any): number { function calculateAudioLevel(audioData: any): number {
@ -32,10 +32,13 @@ export const connectToElevenLabs = async (
connectionPcmFile: Deno.FsFile | null, connectionPcmFile: Deno.FsFile | null,
agentId: string, agentId: string,
apiKey: string, apiKey: string,
closeHandler: () => Promise<void>,
) => { ) => {
console.log(apiKey, agentId); console.log(apiKey, agentId);
const { user, supabase } = payload; const { user, supabase } = payload;
const opus = createOpusPacketizer((packet) => ws.send(packet));
// Queue messages until ElevenLabs connection is ready // Queue messages until ElevenLabs connection is ready
const messageQueue: RawData[] = []; const messageQueue: RawData[] = [];
let isElevenLabsConnected = false; let isElevenLabsConnected = false;
@ -128,13 +131,6 @@ export const connectToElevenLabs = async (
isElevenLabsConnected = true; isElevenLabsConnected = true;
console.log(`ElevenLabs connection ready - conversation_initiation_metadata already processed by SDK`); 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 // Set up ElevenLabs event handlers
elevenLabsConnection.onMessage(async (event: IncomingSocketEvent) => { elevenLabsConnection.onMessage(async (event: IncomingSocketEvent) => {
console.log("ElevenLabs message type:", event); console.log("ElevenLabs message type:", event);
@ -142,7 +138,6 @@ export const connectToElevenLabs = async (
switch (event.type) { switch (event.type) {
case "conversation_initiation_metadata": case "conversation_initiation_metadata":
console.log("ElevenLabs conversation initiated (metadata received)"); console.log("ElevenLabs conversation initiated (metadata received)");
// RESPONSE.CREATED already sent when connection was established
break; break;
case "ping": case "ping":
@ -158,9 +153,10 @@ export const connectToElevenLabs = async (
case "audio": case "audio":
if (event.audio_event?.audio_base_64) { 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) { if (!hasResponseStarted) {
console.log("Sending RESPONSE.CREATED to ESP32 (agent audio starting)"); console.log("Sending RESPONSE.CREATED to ESP32 (agent audio starting)");
opus.reset();
ws.send(JSON.stringify({ ws.send(JSON.stringify({
type: "server", type: "server",
msg: "RESPONSE.CREATED" msg: "RESPONSE.CREATED"
@ -169,23 +165,10 @@ export const connectToElevenLabs = async (
} }
const audioBuffer = Buffer.from(event.audio_event.audio_base_64, "base64"); 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; // Use Opus packetizer to encode and send audio
// Process audio in frames for Opus encoding opus.push(audioBuffer);
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`);
} }
break; break;
@ -199,23 +182,15 @@ export const connectToElevenLabs = async (
user, 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) { if (!hasResponseStarted) {
console.log("Sending RESPONSE.CREATED to ESP32 (agent audio starting)"); console.log("Sending RESPONSE.CREATED to ESP32 (agent audio starting)");
opus.reset();
ws.send(JSON.stringify({ ws.send(JSON.stringify({
type: "server", type: "server",
msg: "RESPONSE.CREATED" msg: "RESPONSE.CREATED"
})); }));
hasResponseStarted = true; hasResponseStarted = true;
} }
} }
break; break;
@ -229,6 +204,9 @@ export const connectToElevenLabs = async (
user, user,
); );
// Flush any remaining audio before sending complete
opus.flush(true);
// Send response complete with device info like OpenAI does // Send response complete with device info like OpenAI does
console.log("Sending RESPONSE.COMPLETE to ESP32"); console.log("Sending RESPONSE.COMPLETE to ESP32");
hasResponseStarted = false; // Reset for next response hasResponseStarted = false; // Reset for next response
@ -305,6 +283,8 @@ export const connectToElevenLabs = async (
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}`); console.log(`ESP32 WebSocket closed with code ${code}, reason: ${reason}`);
await closeHandler();
opus.close();
elevenLabsConnection?.close(); elevenLabsConnection?.close();
if (isDev && connectionPcmFile) { if (isDev && connectionPcmFile) {
@ -329,4 +309,4 @@ export const connectToElevenLabs = async (
msg: errorMessage msg: errorMessage
})); }));
} }
}; };

474
server-deno/models/gemini.ts
View file

@ -1,281 +1,253 @@
import { Buffer } from "node:buffer"; import { Buffer } from "node:buffer";
import type { WebSocketServer as _WebSocketServer } from "npm:@types/ws"; import type { WebSocketServer as _WebSocketServer } from "npm:@types/ws";
import { import {
EndSensitivity, EndSensitivity,
GoogleGenAI, GoogleGenAI,
LiveConnectConfig, LiveConnectConfig,
LiveServerMessage, LiveServerMessage,
Modality, Modality,
Session, Session,
} from "npm:@google/genai"; } 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"; import { addConversation } from "../supabase.ts";
export const connectToGemini = async ( export const connectToGemini = async ({
ws: WebSocket, ws,
payload: IPayload, payload,
connectionPcmFile: Deno.FsFile | null, connectionPcmFile,
firstMessage: string, firstMessage,
systemPrompt: string, systemPrompt,
) => { closeHandler,
const { user, supabase } = payload; }: ProviderArgs) => {
const { oai_voice } = user.personality ?? { oai_voice: "Sadachbia" }; 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 console.log(`Connecting with Gemini key "${geminiApiKey?.slice(0, 3)}..."`);
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: {},
};
// Response queue for handling Google's callback-based responses // Initialize Google GenAI
const responseQueue: LiveServerMessage[] = []; const ai = new GoogleGenAI({ apiKey: geminiApiKey });
let geminiSession: Session | null = null; 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() { // Response queue for handling Google's callback-based responses
let done = false; const responseQueue: LiveServerMessage[] = [];
let message: LiveServerMessage | undefined = undefined; let geminiSession: Session | null = null;
while (!done) {
message = responseQueue.shift();
if (message) {
done = true;
} else {
await new Promise((resolve) => setTimeout(resolve, 10));
}
}
return message;
}
async function handleTurn() { async function waitMessage() {
const turns: any[] = []; let done = false;
let done = false; let message: LiveServerMessage | undefined = undefined;
while (!done) { while (!done) {
const message = await waitMessage(); message = responseQueue.shift();
turns.push(message); if (message) {
done = true;
} else {
await new Promise((resolve) => setTimeout(resolve, 10));
}
}
return message;
}
if ( async function handleTurn() {
message.serverContent const turns: any[] = [];
) { let done = false;
if (message.serverContent.generationComplete) { while (!done) {
ws.send(JSON.stringify({ const message = await waitMessage();
type: "server", turns.push(message);
msg: "RESPONSE.CREATED",
}));
done = true;
}
// if (message.serverContent.turnComplete) { if (message.serverContent) {
// ws.send( if (message.serverContent.generationComplete) {
// JSON.stringify({ opus.reset();
// type: "server", ws.send(JSON.stringify({
// msg: "AUDIO.COMMITTED", type: "server",
// }), msg: "RESPONSE.CREATED",
// ); }));
// } done = true;
} }
} }
return turns; }
} return turns;
}
async function processGeminiTurns() { async function processGeminiTurns() {
try { try {
console.log("Processing Gemini turns"); console.log("Processing Gemini turns");
while (geminiSession) { while (geminiSession) {
const turns = await handleTurn(); const turns = await handleTurn();
// Combine all audio data from this turn // Combine all audio data from this turn
const combinedAudio = turns.reduce( const combinedAudio = turns.reduce(
(acc: number[], turn: any) => { (acc: number[], turn: any) => {
if (turn.data) { if (turn.data) {
const buffer = Buffer.from(turn.data, "base64"); const buffer = Buffer.from(turn.data, "base64");
const intArray = new Int16Array( const intArray = new Int16Array(
buffer.buffer, buffer.buffer,
buffer.byteOffset, buffer.byteOffset,
buffer.byteLength / buffer.byteLength /
Int16Array.BYTES_PER_ELEMENT, Int16Array.BYTES_PER_ELEMENT,
); );
return acc.concat(Array.from(intArray)); return acc.concat(Array.from(intArray));
} }
return acc; return acc;
}, },
[], [],
); );
if (combinedAudio.length > 0) { if (combinedAudio.length > 0) {
// Convert back to buffer and send to client // Convert back to buffer and send to client
const audioBuffer = new Int16Array(combinedAudio); const audioBuffer = new Int16Array(combinedAudio);
const buffer = Buffer.from(audioBuffer.buffer); const buffer = Buffer.from(audioBuffer.buffer);
// PREVIEW AUDIO // Use Opus packetizer to encode and send audio
// const wf = new WaveFile(); opus.push(buffer);
// wf.fromScratch(1, SAMPLE_RATE, "16", audioBuffer); opus.flush(true);
}
// const filename = `gemini_response_${Date.now()}.wav`; // Handle text responses if any
// await Deno.writeFile(filename, wf.toBuffer()); let outputTranscriptionText = "";
// console.log(`Audio saved as ${filename}`); let inputTranscriptionText = "";
for (const turn of turns as LiveServerMessage[]) {
if (
turn.serverContent &&
turn.serverContent.outputTranscription
) {
outputTranscriptionText +=
turn.serverContent.outputTranscription.text;
}
// SEND TO ESP32 if (
// Send audio in chunks to client turn.serverContent &&
for ( turn.serverContent.inputTranscription
let offset = 0; ) {
offset < buffer.length; inputTranscriptionText +=
offset += FRAME_SIZE turn.serverContent.inputTranscription.text;
) { }
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
}
}
}
// // Handle text responses if any // Send completion signal
let outputTranscriptionText = ""; ws.send(JSON.stringify({
let inputTranscriptionText = ""; type: "server",
for (const turn of turns as LiveServerMessage[]) { msg: "RESPONSE.COMPLETE",
if ( }));
turn.serverContent &&
turn.serverContent.outputTranscription
) {
outputTranscriptionText +=
turn.serverContent.outputTranscription.text;
}
if ( // Add user transcription to supabase
turn.serverContent && await addConversation(
turn.serverContent.inputTranscription supabase,
) { "user",
inputTranscriptionText += inputTranscriptionText,
turn.serverContent.inputTranscription.text; user,
} );
}
// Send completion signal // Add assistant transcription to supabase
ws.send(JSON.stringify({ await addConversation(
type: "server", supabase,
msg: "RESPONSE.COMPLETE", "assistant",
})); outputTranscriptionText,
user,
);
}
} catch (error) {
console.error("Error processing Gemini turns:", error);
}
}
// Add user transcription to supabase // Connect to Google Gemini Live
await addConversation( try {
supabase, geminiSession = await ai.live.connect({
"user", model: model,
inputTranscriptionText, callbacks: {
user, 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 console.log("Connected to Gemini successfully!");
await addConversation( // Send first message if available
supabase, const inputTurns = [{
"assistant", role: "user",
outputTranscriptionText, parts: [{ text: firstMessage }],
user, }];
); geminiSession?.sendClientContent({ turns: inputTurns });
} processGeminiTurns();
} catch (error) { } catch (e: unknown) {
console.error("Error processing Gemini turns:", error); console.log(`Error connecting to Gemini: ${e}`);
} ws.close();
} return;
}
// Connect to Google Gemini Live ws.on("message", (data: any, isBinary: boolean) => {
try { try {
geminiSession = await ai.live.connect({ if (isBinary) {
model: model, // Handle binary audio data from ESP32
callbacks: { const base64Data = data.toString("base64");
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,
});
console.log("Connected to Gemini successfully!"); if (isDev && connectionPcmFile) {
// Send first message if available connectionPcmFile.write(data);
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;
}
ws.on("message", (data: any, isBinary: boolean) => { // Send audio to Gemini
try { geminiSession?.sendRealtimeInput({
if (isBinary) { audio: {
// Handle binary audio data from ESP32 data: base64Data,
const base64Data = data.toString("base64"); mimeType: "audio/pcm;rate=24000",
},
});
}
} catch (e: unknown) {
console.error("Error handling message:", (e as Error).message);
}
});
if (isDev && connectionPcmFile) { ws.on("error", (error: any) => {
connectionPcmFile.write(data); console.error("WebSocket error:", error);
} geminiSession?.close();
});
// Send audio to Gemini ws.on("close", async (code: number, reason: string) => {
geminiSession?.sendRealtimeInput({ console.log(`WebSocket closed with code ${code}, reason: ${reason}`);
audio: { await closeHandler();
data: base64Data, opus.close();
mimeType: "audio/pcm;rate=24000", // Gemini expects 16kHz but 24kHz is fine geminiSession?.close();
}, if (isDev && connectionPcmFile) {
}); connectionPcmFile.close();
} console.log("Closed debug audio file.");
} 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.");
}
});
}; };

168
server-deno/models/grok.ts
View file

@ -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'; const XAI_REALTIME_URL = "wss://api.x.ai/v1/realtime";
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'; export const connectToGrok = async ({
const DEFAULT_GROK_VOICE = 'Ara'; ws,
payload,
export const connectToGrok = async ( connectionPcmFile,
ws: WebSocket, firstMessage,
payload: IPayload, systemPrompt,
connectionPcmFile: Deno.FsFile | null, closeHandler,
firstMessage: string, }: ProviderArgs) => {
systemPrompt: string,
) => {
const { user, supabase } = payload; const { user, supabase } = payload;
if (!xaiApiKey) { 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, { const grokWs = new WebSocket(XAI_REALTIME_URL, {
headers: { headers: {
Authorization: `Bearer ${xaiApiKey}`, Authorization: `Bearer ${xaiApiKey}`,
'Content-Type': 'application/json', "Content-Type": "application/json",
}, },
}); });
@ -34,21 +35,21 @@ export const connectToGrok = async (
const messageQueue: RawData[] = []; const messageQueue: RawData[] = [];
let createdSent = false; let createdSent = false;
let outputTranscript = ''; let outputTranscript = "";
let audioRemainder = Buffer.alloc(0);
const sendResponseCreated = async () => { const sendResponseCreated = async () => {
try { try {
const device = await getDeviceInfo(supabase, user.user_id); const device = await getDeviceInfo(supabase, user.user_id);
opus.reset();
ws.send( ws.send(
JSON.stringify({ JSON.stringify({
type: 'server', type: "server",
msg: 'RESPONSE.CREATED', msg: "RESPONSE.CREATED",
volume_control: device?.volume ?? 100, volume_control: device?.volume ?? 100,
}), }),
); );
} catch { } 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; if (!firstMessage) return;
grokWs.send( grokWs.send(
JSON.stringify({ JSON.stringify({
type: 'conversation.item.create', type: "conversation.item.create",
item: { item: {
type: 'message', type: "message",
role: 'user', role: "user",
content: [{ type: 'input_text', text: firstMessage }], 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; isConnected = true;
grokWs.send( grokWs.send(
JSON.stringify({ JSON.stringify({
type: 'session.update', type: "session.update",
session: { session: {
voice, voice,
instructions: systemPrompt, instructions: systemPrompt,
turn_detection: { type: "server_vad" }, turn_detection: { type: "server_vad" },
audio: { audio: {
input: { format: { type: 'audio/pcm', rate: 16000 } }, input: { format: { type: "audio/pcm", rate: 16000 } },
output: { format: { type: 'audio/pcm', rate: 24000 } }, 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; let event: any;
try { try {
event = JSON.parse(data.toString('utf-8')); event = JSON.parse(data.toString("utf-8"));
} catch { } catch {
return; return;
} }
try { try {
switch (event.type) { switch (event.type) {
case 'response.created': case "response.created":
if (!createdSent) { if (!createdSent) {
await sendResponseCreated(); await sendResponseCreated();
createdSent = true; createdSent = true;
} }
break; break;
case 'response.output_audio_transcript.delta': case "response.output_audio_transcript.delta":
if (typeof event.delta === 'string') { if (typeof event.delta === "string") {
outputTranscript += event.delta; outputTranscript += event.delta;
} }
break; break;
case 'response.output_audio.delta': case "response.output_audio.delta":
if (typeof event.delta === 'string') { if (typeof event.delta === "string") {
const pcmChunk = Buffer.from(event.delta, 'base64'); const pcmChunk = Buffer.from(event.delta, "base64");
audioRemainder = Buffer.concat([audioRemainder, pcmChunk]); // Use Opus packetizer to encode and send audio
opus.push(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
}
}
} }
break; break;
case 'conversation.item.input_audio_transcription.completed': case "conversation.item.input_audio_transcription.completed":
if (typeof event.transcript === 'string' && event.transcript.length > 0) { if (typeof event.transcript === "string" && event.transcript.length > 0) {
await addConversation(supabase, 'user', event.transcript, user); await addConversation(supabase, "user", event.transcript, user);
} }
break; break;
case 'input_audio_buffer.committed': case "input_audio_buffer.committed":
ws.send(JSON.stringify({ type: 'server', msg: 'AUDIO.COMMITTED' })); ws.send(JSON.stringify({ type: "server", msg: "AUDIO.COMMITTED" }));
break; break;
case 'response.done': case "response.done":
// Flush any remaining audio
opus.flush(true);
if (outputTranscript) { if (outputTranscript) {
await addConversation(supabase, 'assistant', outputTranscript, user); await addConversation(supabase, "assistant", outputTranscript, user);
outputTranscript = ''; outputTranscript = "";
} }
ws.send(JSON.stringify({ type: 'server', msg: 'RESPONSE.COMPLETE' })); ws.send(JSON.stringify({ type: "server", msg: "RESPONSE.COMPLETE" }));
createdSent = false; createdSent = false;
break; break;
case 'error': case "error":
ws.send(JSON.stringify({ type: 'server', msg: 'RESPONSE.ERROR' })); ws.send(JSON.stringify({ type: "server", msg: "RESPONSE.ERROR" }));
createdSent = false; createdSent = false;
break; break;
} }
} catch (err) { } catch (err) {
console.error('Error processing Grok event:', err); console.error("Error processing Grok event:", err);
ws.send(JSON.stringify({ type: 'server', msg: 'RESPONSE.ERROR' })); ws.send(JSON.stringify({ type: "server", msg: "RESPONSE.ERROR" }));
createdSent = false; createdSent = false;
} }
}); });
grokWs.on('close', () => { grokWs.on("close", () => {
ws.close(); ws.close();
}); });
grokWs.on('error', (error: any) => { grokWs.on("error", (error: any) => {
console.error('Grok WebSocket error:', error); console.error("Grok WebSocket error:", error);
ws.send(JSON.stringify({ type: 'server', msg: 'RESPONSE.ERROR' })); ws.send(JSON.stringify({ type: "server", msg: "RESPONSE.ERROR" }));
}); });
const messageHandler = async (data: RawData, isBinary: boolean) => { const messageHandler = async (data: RawData, isBinary: boolean) => {
if (isBinary) { if (isBinary) {
const base64Data = (data as Buffer).toString('base64'); const base64Data = (data as Buffer).toString("base64");
grokWs.send(JSON.stringify({ type: 'input_audio_buffer.append', audio: base64Data })); grokWs.send(JSON.stringify({ type: "input_audio_buffer.append", audio: base64Data }));
if (isDev && connectionPcmFile) { if (isDev && connectionPcmFile) {
await connectionPcmFile.write(data as Buffer); await connectionPcmFile.write(data as Buffer);
@ -189,23 +183,23 @@ export const connectToGrok = async (
let message: any; let message: any;
try { try {
message = JSON.parse((data as Buffer).toString('utf-8')); message = JSON.parse((data as Buffer).toString("utf-8"));
} catch { } catch {
return; return;
} }
if (message?.type !== 'instruction') return; if (message?.type !== "instruction") return;
if (message.msg === 'end_of_speech') { if (message.msg === "end_of_speech") {
grokWs.send(JSON.stringify({ type: 'input_audio_buffer.commit' })); grokWs.send(JSON.stringify({ type: "input_audio_buffer.commit" }));
grokWs.send(JSON.stringify({ type: 'response.create' })); grokWs.send(JSON.stringify({ type: "response.create" }));
grokWs.send(JSON.stringify({ type: 'input_audio_buffer.clear' })); grokWs.send(JSON.stringify({ type: "input_audio_buffer.clear" }));
} else if (message.msg === 'INTERRUPT') { } else if (message.msg === "INTERRUPT") {
grokWs.send(JSON.stringify({ type: 'input_audio_buffer.clear' })); 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) { if (!isConnected) {
messageQueue.push(data); messageQueue.push(data);
} else { } else {
@ -213,13 +207,15 @@ export const connectToGrok = async (
} }
}); });
ws.on('error', (error: any) => { ws.on("error", (error: any) => {
console.error('ESP32 WebSocket error:', error); console.error("ESP32 WebSocket error:", error);
grokWs.close(); 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}`); console.log(`ESP32 WebSocket closed with code ${code}, reason: ${reason}`);
await closeHandler();
opus.close();
grokWs.close(); grokWs.close();
if (isDev && connectionPcmFile) { if (isDev && connectionPcmFile) {
connectionPcmFile.close(); connectionPcmFile.close();
@ -227,12 +223,12 @@ export const connectToGrok = async (
}); });
return new Promise<void>((resolve, reject) => { return new Promise<void>((resolve, reject) => {
const timeout = setTimeout(() => reject(new Error('Grok connection timeout')), 10000); const timeout = setTimeout(() => reject(new Error("Grok connection timeout")), 10000);
grokWs.on('open', () => { grokWs.on("open", () => {
clearTimeout(timeout); clearTimeout(timeout);
resolve(); resolve();
}); });
grokWs.on('error', (error: any) => { grokWs.on("error", (error: any) => {
clearTimeout(timeout); clearTimeout(timeout);
reject(error); reject(error);
}); });

587
server-deno/models/hume.ts
View file

@ -1,317 +1,302 @@
import { Buffer } from 'node:buffer'; import { Buffer } from "node:buffer";
import type { RawData } from 'npm:@types/ws'; import type { RawData } from "npm:@types/ws";
import { WebSocket } from 'npm:ws'; import { WebSocket } from "npm:ws";
import { addConversation, getDeviceInfo } from '../supabase.ts'; import { addConversation, getDeviceInfo } from "../supabase.ts";
import { encoder, FRAME_SIZE, isDev, humeApiKey, downsamplePcm, extractPcmFromWav, boostLimitPCM16LEInPlace } from '../utils.ts'; import { createOpusPacketizer, isDev, humeApiKey, downsamplePcm, extractPcmFromWav, boostLimitPCM16LEInPlace } from "../utils.ts";
export const connectToHume = ( export const connectToHume = ({
ws: WebSocket, ws,
payload: IPayload, payload,
connectionPcmFile:Deno.FsFile | null, connectionPcmFile,
firstMessage: string, firstMessage,
systemPrompt: string, systemPrompt,
closeHandler: () => Promise<void>, closeHandler,
) => { }: ProviderArgs) => {
const { user, supabase } = payload; const { user, supabase } = payload;
const { personality } = user; 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 console.log(`Connecting to Hume with key "${humeApiKey?.slice(0, 3)}..."`);
const queryParams = new URLSearchParams({
api_key: humeApiKey!,
config_id: personality!.oai_voice,
});
const humeWsUrl = `wss://api.hume.ai/v0/evi/chat?${queryParams.toString()}`; // Build Hume WebSocket URL
const configId = personality?.oai_voice ?? "";
console.log(`Connecting to Hume WebSocket at: ${humeWsUrl.replace(humeApiKey!, 'API_KEY_HIDDEN')}`); const queryParams = new URLSearchParams({
const humeWs = new WebSocket(humeWsUrl); api_key: humeApiKey!,
config_id: configId,
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
}); });
ws.send(JSON.stringify({
type: 'server',
msg: 'RESPONSE.ERROR',
error: 'Connection to Hume failed',
}));
});
// Handle messages from ESP32 client const humeWsUrl = `wss://api.hume.ai/v0/evi/chat?${queryParams.toString()}`;
const messageHandler = async (data: RawData, isBinary: boolean) => {
try {
if (isBinary) {
// Handle audio data from ESP32
const base64Audio = data.toString('base64');
const audioMessage: HumeAudioInput = { console.log(`Connecting to Hume WebSocket at: ${humeWsUrl.replace(humeApiKey!, "API_KEY_HIDDEN")}`);
type: 'audio_input', const humeWs = new WebSocket(humeWsUrl);
data: base64Audio,
};
if (isConnected) { let isConnected = false;
humeWs.send(JSON.stringify(audioMessage)); 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) { 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<void>((resolve, reject) => {
const timeout = setTimeout(() => {
reject(new Error('Hume connection timeout'));
}, 10000);
humeWs.on('open', () => {
clearTimeout(timeout);
resolve();
}); });
humeWs.on('error', (error) => { // Wait for Hume connection to be established
clearTimeout(timeout); return new Promise<void>((resolve, reject) => {
reject(error); const timeout = setTimeout(() => {
reject(new Error("Hume connection timeout"));
}, 10000);
humeWs.on("open", () => {
clearTimeout(timeout);
resolve();
});
humeWs.on("error", (error) => {
clearTimeout(timeout);
reject(error);
});
}); });
});
}; };

628
server-deno/models/openai.ts
View file

@ -1,364 +1,332 @@
import { Buffer } from "node:buffer"; import { Buffer } from "node:buffer";
import type { import type { RawData } from "npm:@types/ws";
RawData,
WebSocket as WSWebSocket,
WebSocketServer as _WebSocketServer,
} from "npm:@types/ws";
import { RealtimeClient } from "../realtime/client.js"; import { RealtimeClient } from "../realtime/client.js";
import { RealtimeUtils } from "../realtime/utils.js"; import { RealtimeUtils } from "../realtime/utils.js";
import { addConversation, getDeviceInfo } from "../supabase.ts"; 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 sendFirstMessage = (client: RealtimeClient, firstMessage: string) => {
const event = { const event = {
event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID
type: "conversation.item.create", type: "conversation.item.create",
previous_item_id: "root", previous_item_id: "root",
item: { item: {
type: "message", type: "message",
role: "system", role: "system",
content: [{ content: [{
type: "input_text", type: "input_text",
text: firstMessage, text: firstMessage,
}], }],
}, },
}; };
client.realtime.send(event.type, event); client.realtime.send(event.type, event);
client.realtime.send("response.create", { client.realtime.send("response.create", {
event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID
type: "response.create", type: "response.create",
}); });
}; };
export const connectToOpenAI = async ( export const connectToOpenAI = async ({
ws: WebSocket, ws,
payload: IPayload, payload,
connectionPcmFile: Deno.FsFile | null, connectionPcmFile,
firstMessage: string, firstMessage,
systemPrompt: string, systemPrompt,
) => { closeHandler,
const { user, supabase } = payload; }: ProviderArgs) => {
const { user, supabase } = payload;
let currentItemId: string | null = null; const opus = createOpusPacketizer((packet) => ws.send(packet));
let currentCallId: string | null = null;
// Instantiate new client let currentItemId: string | null = null;
console.log(`Connecting with key "${openaiApiKey.slice(0, 3)}..."`); let currentCallId: string | null = null;
const client = new RealtimeClient({ apiKey: openaiApiKey });
// ADD TOOL CALLS HERE // Instantiate new client
client.addTool( console.log(`Connecting with key "${openaiApiKey?.slice(0, 3)}..."`);
{ const client = new RealtimeClient({ apiKey: openaiApiKey });
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 your custom message to the client // ADD TOOL CALLS HERE
ws.send(JSON.stringify({ type: "server", msg: "SESSION.END" })); 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 // Send your custom message to the client
const functionResult = { ws.send(JSON.stringify({ type: "server", msg: "SESSION.END" }));
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}`,
}),
},
};
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 // Relay: OpenAI Realtime API Event -> Browser Event
return { success: true, message: `Session ended: ${args.reason}` }; 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 if (event.type in client.conversation.EventProcessors) {
client.realtime.on("server.*", async (event: any) => { try {
// console.log(`Relaying "${event.type}" to Client`); switch (event.type) {
// Check if the event is session.created case "response.created":
if (event.type === "session.created") { console.log("response.created", event);
console.log("session created", event); opus.reset();
sendFirstMessage(client, firstMessage); try {
} else if (event.type === "session.updated") { const device = await getDeviceInfo(supabase, user.user_id);
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 (device) { if (device) {
// Send the updated volume data along with the response complete message // Send the updated volume data along with the response complete message
ws.send(JSON.stringify({ ws.send(JSON.stringify({
type: "server", type: "server",
msg: "RESPONSE.COMPLETE", msg: "RESPONSE.CREATED",
volume_control: device.volume ?? 100, volume_control: device.volume ?? 100,
})); }));
} else { } else {
// Fall back to just sending the complete message if there's an error // Fall back to just sending the complete message if there's an error
ws.send( ws.send(
JSON.stringify({ JSON.stringify({
type: "server", type: "server",
msg: "RESPONSE.COMPLETE", msg: "RESPONSE.CREATED",
}), }),
); );
} }
} catch (error) { } catch (error) {
console.error("Error fetching updated device info:", error); console.error("Error fetching updated device info:", error);
ws.send( ws.send(
JSON.stringify({ JSON.stringify({
type: "server", type: "server",
msg: "RESPONSE.COMPLETE", msg: "RESPONSE.CREATED",
}), }),
); );
} }
} else if (event.type === "response.audio_transcript.done") { break;
console.log("response.audio_transcript.done", event); case "response.output_item.added":
await addConversation( console.log("response.output_item.added", event);
supabase, if (event.item.id) {
"assistant", console.log("foobar", event.item.id);
event.transcript, currentItemId = event.item.id;
user, currentCallId = event.item.call_id;
); }
} else if (event.type === "input_audio_buffer.committed") { break;
ws.send(JSON.stringify({ type: "server", msg: "AUDIO.COMMITTED" })); 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) { client.realtime.on("close", () => ws.close());
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,
);
try { // Relay: Browser Event -> OpenAI Realtime API Event
const encodedPacket = encoder // We need to queue data waiting for the OpenAI connection
.encode(frame); const messageQueue: RawData[] = [];
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" }),
);
}
}
});
client.realtime.on("close", () => ws.close()); const messageHandler = async (data: any, isBinary: boolean) => {
try {
let event;
// Relay: Browser Event -> OpenAI Realtime API Event // for esp32
// We need to queue data waiting for the OpenAI connection if (isBinary) {
const messageQueue: RawData[] = []; const base64Data = data.toString("base64");
const messageHandler = async (data: any, isBinary: boolean) => { // Convert binary PCM16 data to base64 for OpenAI Realtime API
try { event = {
let 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 // commit user audio and create response
if (isBinary) { if (
const base64Data = data.toString("base64"); message.type === "instruction" &&
message.msg === "end_of_speech"
) {
console.log("end_of_speech detected");
// Convert binary PCM16 data to base64 for OpenAI Realtime API client.realtime.send("input_audio_buffer.commit", {
event = { event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID
event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID type: "input_audio_buffer.commit",
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"));
// commit user audio and create response client.realtime.send("response.create", {
if ( event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID
message.type === "instruction" && type: "response.create",
message.msg === "end_of_speech" });
) {
console.log("end_of_speech detected");
client.realtime.send("input_audio_buffer.commit", { client.realtime.send("input_audio_buffer.clear", {
event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID
type: "input_audio_buffer.commit", 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", { client.realtime.send("conversation.item.truncate", {
event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID
type: "response.create", type: "conversation.item.truncate",
}); item_id: currentItemId,
content_index: 0,
audio_end_ms: audioEndMs,
});
client.realtime.send("input_audio_buffer.clear", { client.realtime.send("input_audio_buffer.clear", {
event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID
type: "input_audio_buffer.clear", type: "input_audio_buffer.clear",
}); });
} else if ( }
message.type === "instruction" && }
message.msg === "INTERRUPT" } catch (e: unknown) {
) { console.error((e as Error).message);
console.log("interrupt detected", message); console.log(`Error parsing event from client: ${data}`);
const audioEndMs = message.audio_end_ms; }
};
client.realtime.send("conversation.item.truncate", { ws.on("message", (data: any, isBinary: boolean) => {
event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID if (!client.isConnected()) {
type: "conversation.item.truncate", messageQueue.push(data);
item_id: currentItemId, } else {
content_index: 0, messageHandler(data, isBinary);
audio_end_ms: audioEndMs, }
}); });
client.realtime.send("input_audio_buffer.clear", { // Add error handler
event_id: RealtimeUtils.generateId("evt_"), // Generate unique ID ws.on("error", (error: any) => {
type: "input_audio_buffer.clear", console.error("WebSocket error:", error);
}); client.disconnect();
} });
}
} catch (e: unknown) {
console.error((e as Error).message);
console.log(`Error parsing event from client: ${data}`);
}
};
ws.on("message", (data: any, isBinary: boolean) => { // Add more detailed close handling
if (!client.isConnected()) { ws.on("close", async (code: number, reason: string) => {
messageQueue.push(data); console.log(`WebSocket closed with code ${code}, reason: ${reason}`);
} else { await closeHandler();
messageHandler(data, isBinary); opus.close();
} client.disconnect();
}); if (isDev) {
if (connectionPcmFile) {
connectionPcmFile.close();
console.log(`Closed debug audio file.`);
}
}
});
// Add error handler // Connect to the OpenAI Realtime API
ws.on("error", (error: any) => { try {
console.error("WebSocket error:", error); console.log(`Connecting to OpenAI...`);
client.disconnect(); const sessionOptions = {
}); model: "gpt-4o-mini-realtime-preview-2024-12-17",
turn_detection: {
// Add more detailed close handling type: "server_vad",
ws.on("close", async (code: number, reason: string) => { threshold: 0.4,
console.log(`WebSocket closed with code ${code}, reason: ${reason}`); prefix_padding_ms: 400,
client.disconnect(); silence_duration_ms: 1000,
if (isDev) { },
if (connectionPcmFile) { voice: user.personality?.oai_voice ?? defaultOpenAIVoice,
connectionPcmFile.close(); instructions: systemPrompt,
console.log(`Closed debug audio file.`); input_audio_transcription: { model: "whisper-1" },
} };
} await client.connect(sessionOptions as any);
}); } catch (e: unknown) {
console.log(`Error connecting to OpenAI: ${e as Error}`);
// Connect to the OpenAI Realtime API ws.close();
try { return;
console.log(`Connecting to OpenAI...`); }
const sessionOptions = { console.log(`Connected to OpenAI successfully!`);
model: "gpt-4o-mini-realtime-preview-2024-12-17", while (messageQueue.length) {
// turn_detection: null, messageHandler(messageQueue.shift(), false);
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);
}
}; };

9
server-deno/types.d.ts vendored
View file

@ -199,4 +199,13 @@ declare global {
code: string; code: string;
message: string; message: string;
} }
interface ProviderArgs {
ws: WebSocket;
payload: IPayload;
connectionPcmFile: Deno.FsFile | null;
firstMessage: string;
systemPrompt: string;
closeHandler: () => Promise<void>;
}
} }

79
server-deno/utils.ts
View file

@ -7,6 +7,10 @@ import { Encoder } from "@evan/opus";
export const defaultVolume = 50; export const defaultVolume = 50;
export const defaultGeminiVoice = "Sadachbia";
export const defaultOpenAIVoice = "ash";
export const defaultGrokVoice = "Ara";
// Define your audio parameters // Define your audio parameters
export const SAMPLE_RATE = 24000; // For example, 24000 Hz export const SAMPLE_RATE = 24000; // For example, 24000 Hz
const CHANNELS = 1; // Mono (set to 2 if you have stereo) 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 * const FRAME_SIZE = (SAMPLE_RATE * FRAME_DURATION / 1000) * CHANNELS *
BYTES_PER_SAMPLE; // 960 bytes for 24000 Hz mono 16-bit 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({ const encoder = new Encoder({
channels: CHANNELS, channels: CHANNELS,
sample_rate: SAMPLE_RATE, sample_rate: SAMPLE_RATE,
@ -22,7 +99,7 @@ const encoder = new Encoder({
}); });
encoder.expert_frame_duration = FRAME_DURATION; encoder.expert_frame_duration = FRAME_DURATION;
encoder.bitrate = 12000; encoder.bitrate = 24000;
export const openaiApiKey = Deno.env.get("OPENAI_API_KEY"); export const openaiApiKey = Deno.env.get("OPENAI_API_KEY");
export const geminiApiKey = Deno.env.get("GEMINI_API_KEY"); export const geminiApiKey = Deno.env.get("GEMINI_API_KEY");