Merge pull request #758 from Portkey-AI/feat/openai-realtime-api

Feat: openai and azure-openai realtime api support

Author: VisargD

package-lock.json

@@ -42,11 +42,13 @@
   "dependencies": {
     "@aws-crypto/sha256-js": "^5.2.0",
     "@hono/node-server": "^1.3.3",
+    "@hono/node-ws": "^1.0.4",
     "@portkey-ai/mustache": "^2.1.2",
     "@smithy/signature-v4": "^2.1.1",
     "@types/mustache": "^4.2.5",
     "async-retry": "^1.3.3",
-    "hono": "^3.12.0",
+    "hono": "^4.6.10",
+    "ws": "^8.18.0",
     "zod": "^3.22.4"
   },
   "devDependencies": {
@@ -57,6 +59,7 @@
     "@types/async-retry": "^1.4.5",
     "@types/jest": "^29.5.12",
     "@types/node": "20.8.3",
+    "@types/ws": "^8.5.12",
     "husky": "^9.1.4",
     "jest": "^29.7.0",
     "prettier": "3.2.5",

package.json

@@ -262,6 +262,7 @@ export async function tryPostProxy(
     providerOptions: providerOption,
     fn,
     gatewayRequestBody: params,
+    gatewayRequestURL: c.req.url,
   });
 
   const url = endpoint
@@ -517,6 +518,7 @@ export async function tryPost(
     providerOptions: providerOption,
     fn,
     gatewayRequestBody: params,
+    gatewayRequestURL: c.req.url,
   });
   const url = `${baseUrl}${endpoint}`;
 
@@ -1004,6 +1006,7 @@ export function updateResponseHeaders(
   // Delete content-length header to avoid conflicts with hono compress middleware
   // workerd environment handles this authomatically
   response.headers.delete('content-length');
+  response.headers.delete('transfer-encoding');
 }
 
 export function constructConfigFromRequestHeaders(
@@ -1021,6 +1024,9 @@ export function constructConfigFromRequestHeaders(
       requestHeaders[`x-${POWERED_BY}-azure-entra-client-secret`],
     azureEntraTenantId: requestHeaders[`x-${POWERED_BY}-azure-entra-tenant-id`],
     azureModelName: requestHeaders[`x-${POWERED_BY}-azure-model-name`],
+    openaiBeta:
+      requestHeaders[`x-${POWERED_BY}-openai-beta`] ||
+      requestHeaders[`openai-beta`],
   };
 
   const stabilityAiConfig = {
@@ -1058,6 +1064,9 @@ export function constructConfigFromRequestHeaders(
   const openAiConfig = {
     openaiOrganization: requestHeaders[`x-${POWERED_BY}-openai-organization`],
     openaiProject: requestHeaders[`x-${POWERED_BY}-openai-project`],
+    openaiBeta:
+      requestHeaders[`x-${POWERED_BY}-openai-beta`] ||
+      requestHeaders[`openai-beta`],
   };
 
   const huggingfaceConfig = {

src/handlers/handlerUtils.ts

@@ -0,0 +1,91 @@
+import { Context } from 'hono';
+import { constructConfigFromRequestHeaders } from './handlerUtils';
+import { ProviderAPIConfig } from '../providers/types';
+import Providers from '../providers';
+import { Options } from '../types/requestBody';
+import {
+  addListeners,
+  getOptionsForOutgoingConnection,
+  getURLForOutgoingConnection,
+} from './websocketUtils';
+import { RealtimeLlmEventParser } from '../services/realtimeLlmEventParser';
+
+const getOutgoingWebSocket = async (url: string, options: RequestInit) => {
+  let outgoingWebSocket: WebSocket | null = null;
+  try {
+    let response = await fetch(url, options);
+    outgoingWebSocket = response.webSocket;
+  } catch (error) {
+    console.log(error);
+  }
+
+  if (!outgoingWebSocket) {
+    throw new Error('WebSocket connection failed');
+  }
+
+  outgoingWebSocket.accept();
+  return outgoingWebSocket;
+};
+
+export async function realTimeHandler(c: Context): Promise<Response> {
+  try {
+    const requestHeaders = Object.fromEntries(c.req.raw.headers);
+
+    const providerOptions = constructConfigFromRequestHeaders(
+      requestHeaders
+    ) as Options;
+    const provider = providerOptions.provider ?? '';
+    const apiConfig: ProviderAPIConfig = Providers[provider].api;
+    const url = getURLForOutgoingConnection(
+      apiConfig,
+      providerOptions,
+      c.req.url
+    );
+    const options = await getOptionsForOutgoingConnection(
+      apiConfig,
+      providerOptions,
+      url,
+      c
+    );
+
+    const sessionOptions = {
+      id: crypto.randomUUID(),
+      providerOptions: {
+        ...providerOptions,
+        requestURL: url,
+        rubeusURL: 'realtime',
+      },
+      requestHeaders,
+      requestParams: {},
+    };
+
+    const webSocketPair = new WebSocketPair();
+    const client = webSocketPair[0];
+    const server = webSocketPair[1];
+
+    server.accept();
+
+    let outgoingWebSocket: WebSocket = await getOutgoingWebSocket(url, options);
+    const eventParser = new RealtimeLlmEventParser();
+    addListeners(outgoingWebSocket, eventParser, server, c, sessionOptions);
+
+    return new Response(null, {
+      status: 101,
+      webSocket: client,
+    });
+  } catch (err: any) {
+    console.log('realtimeHandler error', err.message);
+    return new Response(
+      JSON.stringify({
+        status: 'failure',
+        message: 'Something went wrong',
+      }),
+      {
+        status: 500,
+        headers: {
+          'content-type': 'application/json',
+        },
+      }
+    );
+  }
+}

src/handlers/realtimeHandler.ts

@@ -0,0 +1,97 @@
+import { Context } from 'hono';
+import { constructConfigFromRequestHeaders } from './handlerUtils';
+import WebSocket from 'ws';
+import { ProviderAPIConfig } from '../providers/types';
+import Providers from '../providers';
+import { Options } from '../types/requestBody';
+import { RealtimeLlmEventParser } from '../services/realtimeLlmEventParser';
+import { WSContext, WSEvents } from 'hono/ws';
+
+export async function realTimeHandlerNode(
+  c: Context
+): Promise<WSEvents<unknown>> {
+  try {
+    let incomingWebsocket: WSContext<unknown> | null = null;
+    const requestHeaders = Object.fromEntries(c.req.raw.headers);
+    const camelCaseConfig = constructConfigFromRequestHeaders(requestHeaders);
+
+    const provider = camelCaseConfig?.provider ?? '';
+    const apiConfig: ProviderAPIConfig = Providers[provider].api;
+    const providerOptions = camelCaseConfig as Options;
+    const baseUrl = apiConfig.getBaseURL({ providerOptions });
+    const endpoint = apiConfig.getEndpoint({
+      providerOptions,
+      fn: 'realtime',
+      gatewayRequestBody: {},
+      gatewayRequestURL: c.req.url,
+    });
+    let url = `${baseUrl}${endpoint}`;
+    url = url.replace('https://', 'wss://');
+    const headers = await apiConfig.headers({
+      c,
+      providerOptions,
+      fn: 'realtime',
+      transformedRequestUrl: url,
+      transformedRequestBody: {},
+    });
+
+    const sessionOptions = {
+      id: crypto.randomUUID(),
+      providerOptions: {
+        ...providerOptions,
+        requestURL: url,
+        rubeusURL: 'realtime',
+      },
+      requestHeaders,
+      requestParams: {},
+    };
+
+    const outgoingWebSocket = new WebSocket(url, {
+      headers,
+    });
+    const eventParser = new RealtimeLlmEventParser();
+
+    outgoingWebSocket.addEventListener('message', (event) => {
+      incomingWebsocket?.send(event.data as string);
+      try {
+        const parsedData = JSON.parse(event.data as string);
+        eventParser.handleEvent(c, parsedData, sessionOptions);
+      } catch (err: any) {
+        console.error(`eventParser.handleEvent error: ${err.message}`);
+      }
+    });
+
+    outgoingWebSocket.addEventListener('close', (event) => {
+      incomingWebsocket?.close(event.code, event.reason);
+    });
+
+    outgoingWebSocket.addEventListener('error', (event) => {
+      console.error(`outgoingWebSocket error: ${event.message}`);
+      incomingWebsocket?.close();
+    });
+
+    return {
+      onOpen(evt, ws) {
+        incomingWebsocket = ws;
+      },
+      onMessage(event) {
+        outgoingWebSocket?.send(event.data as string);
+      },
+      onError(evt) {
+        console.error(`incomingWebsocket error: ${evt.type}`);
+        outgoingWebSocket?.close();
+      },
+      onClose() {
+        outgoingWebSocket?.close();
+      },
+    };
+  } catch (err) {
+    c.set('websocketError', true);
+    return {
+      onOpen() {},
+      onMessage() {},
+      onError() {},
+      onClose() {},
+    };
+  }
+}