Merge branch 'main' into chore/address-hono-deprecations

src/globals.ts

@@ -39,5 +39,7 @@ export const PROXY_REQUEST_PATH_PREFIX:string = "/v1/proxy";
 export const CONTENT_TYPES = {
   APPLICATION_JSON: "application/json",
   MULTIPART_FORM_DATA: "multipart/form-data",
-  EVENT_STREAM: "text/event-stream"
+  EVENT_STREAM: "text/event-stream",
+  AUDIO_MPEG: "audio/mpeg",
+  APPLICATION_OCTET_STREAM: "application/octet-stream"
 }

src/handlers/handlerUtils.ts

@@ -1,13 +1,15 @@
 import { Context } from "hono";
-import { AZURE_OPEN_AI, GOOGLE, HEADER_KEYS, PALM, POWERED_BY, RESPONSE_HEADER_KEYS, RETRY_STATUS_CODES } from "../globals";
+import { AZURE_OPEN_AI, CONTENT_TYPES, GOOGLE, HEADER_KEYS, PALM, POWERED_BY, RESPONSE_HEADER_KEYS, RETRY_STATUS_CODES } from "../globals";
 import Providers from "../providers";
 import { ProviderAPIConfig, endpointStrings } from "../providers/types";
 import transformToProviderRequest from "../services/transformToProviderRequest";
 import { Config, Options, Params, RequestBody, ShortConfig, Targets } from "../types/requestBody";
 import { convertKeysToCamelCase } from "../utils";
 import { retryRequest } from "./retryHandler";
-import { handleAudioResponse, handleNonStreamingMode, handleOctetStreamResponse, handleStreamingMode } from "./streamHandler";
+import { handleAudioResponse, handleJSONToStreamResponse, handleNonStreamingMode, handleOctetStreamResponse, handleStreamingMode } from "./streamHandler";
 import { env } from "hono/adapter";
+import { OpenAIChatCompleteJSONToStreamResponseTransform } from "../providers/openai/chatComplete";
+import { OpenAICompleteJSONToStreamResponseTransform } from "../providers/openai/complete";
 
 /**
  * Constructs the request options for the API call.
@@ -374,15 +376,12 @@ export async function tryPost(c: Context, providerOption:Options, inputParams: P
       );
       if (cacheResponse) {
           response = await responseHandler(
-              new Response(cacheResponse, {
-                  headers: {
-                      "content-type": "application/json",
-                  },
-              }),
-              false,
+              new Response(cacheResponse, { headers: {"content-type": "application/json"}}),
+              isStreamingMode,
               provider,
-              undefined,
-              url
+              fn,
+              url,
+              true
           );
           c.set("requestOptions", [
               ...requestOptions,
@@ -471,21 +470,43 @@ export async function tryProvidersInSequence(c: Context, providers:Options[], pa
   throw new Error(JSON.stringify(errors));
 }
 
-// Response Handlers for streaming & non-streaming
-export function responseHandler(response: Response, streamingMode: boolean, proxyProvider: string, responseTransformer: string | undefined, requestURL: string): Promise<Response> {
-  // Checking status 200 so that errors are not considered as stream mode.
+/**
+ * Handles various types of responses based on the specified parameters
+ * and returns a mapped response
+ * @param {Response} response - The HTTP response recieved from LLM.
+ * @param {boolean} streamingMode - Indicates whether streaming mode is enabled.
+ * @param {string} proxyProvider - The provider string.
+ * @param {string | undefined} responseTransformer - The response transformer to determine type of call.
+ * @param {string} requestURL - The URL of the original LLM request.
+ * @param {boolean} [isCacheHit=false] - Indicates whether the response is a cache hit.
+ * @returns {Promise<Response>} - A promise that resolves to the processed response.
+ */
+export function responseHandler(response: Response, streamingMode: boolean, proxyProvider: string, responseTransformer: string | undefined, requestURL: string, isCacheHit: boolean = false): Promise<Response> {
   let responseTransformerFunction: Function | undefined;
+  const responseContentType = response.headers?.get("content-type");
+
+  // Checking status 200 so that errors are not considered as stream mode.
   if (responseTransformer && streamingMode && response.status === 200) {
     responseTransformerFunction = Providers[proxyProvider]?.responseTransforms?.[`stream-${responseTransformer}`];
   } else if (responseTransformer) {
     responseTransformerFunction = Providers[proxyProvider]?.responseTransforms?.[responseTransformer];
   }
 
-  if (streamingMode && response.status === 200) {
-      return handleStreamingMode(response, proxyProvider, responseTransformerFunction, requestURL)
-  } else if (response.headers?.get("content-type") === "audio/mpeg") {
+  // JSON to text/event-stream conversion is only allowed for unified routes: chat completions and completions.
+  // Set the transformer to OpenAI json to stream convertor function in that case.
+  if (responseTransformer && streamingMode && isCacheHit) {
+    responseTransformerFunction = responseTransformer === "chatComplete" ? OpenAIChatCompleteJSONToStreamResponseTransform : OpenAICompleteJSONToStreamResponseTransform;
+  } else if (responseTransformer && !streamingMode && isCacheHit) {
+    responseTransformerFunction = undefined;
+  }
+
+  if (streamingMode && response.status === 200 && isCacheHit && responseTransformerFunction) {
+      return handleJSONToStreamResponse(response, proxyProvider, responseTransformerFunction)
+  } else if (streamingMode && response.status === 200) {
+    return handleStreamingMode(response, proxyProvider, responseTransformerFunction, requestURL)
+  } else if (responseContentType === CONTENT_TYPES.AUDIO_MPEG) {
       return handleAudioResponse(response)
-  } else if (response.headers?.get("content-type") === "application/octet-stream") {
+  } else if (responseContentType === CONTENT_TYPES.APPLICATION_OCTET_STREAM) {
       return handleOctetStreamResponse(response)
   } else {
       return handleNonStreamingMode(response, responseTransformerFunction)

src/handlers/streamHandler.ts

@@ -1,4 +1,6 @@
-import { AZURE_OPEN_AI, GOOGLE } from "../globals";
+import { AZURE_OPEN_AI, CONTENT_TYPES, GOOGLE } from "../globals";
+import { OpenAIChatCompleteResponse } from "../providers/openai/chatComplete";
+import { OpenAICompleteResponse } from "../providers/openai/complete";
 import { getStreamModeSplitPattern } from "../utils";
 
 export async function* readStream(reader: ReadableStreamDefaultReader, splitPattern: string, transformFunction: Function | undefined, isSleepTimeRequired: boolean, fallbackChunkId: string) {
@@ -104,3 +106,25 @@ export async function handleStreamingMode(response: Response, proxyProvider: str
     return new Response(readable, response);
 }
 
+export async function handleJSONToStreamResponse(response: Response, provider: string, responseTransformerFunction: Function): Promise<Response> {
+    const { readable, writable } = new TransformStream();
+    const writer = writable.getWriter();
+    const encoder = new TextEncoder();
+    const responseJSON: OpenAIChatCompleteResponse | OpenAICompleteResponse = await response.clone().json();
+    const streamChunkArray = responseTransformerFunction(responseJSON, provider);
+
+    (async () => {
+        for (const chunk of streamChunkArray) {
+            await writer.write(encoder.encode(chunk));
+        }
+        writer.close();
+    })();
+
+    return new Response(readable, {
+        headers: new Headers({
+            ...Object.fromEntries(response.headers),
+            'content-type': CONTENT_TYPES.EVENT_STREAM
+        })
+    });
+}
+

src/providers/mistral-ai/chatComplete.ts

@@ -40,8 +40,14 @@ export const MistralAIChatCompleteConfig: ProviderConfig = {
         param: "random_seed",
         default: null,
     },
+    safe_prompt: {
+        param: "safe_prompt",
+        default: false
+    },
+    // TODO: deprecate this and move to safe_prompt in next release
     safe_mode: {
-        param: "safe_mode",
+        param: "safe_prompt",
+        default: false
     },
 };
 

src/providers/openai/chatComplete.ts

@@ -83,8 +83,119 @@ export const OpenAIChatCompleteConfig: ProviderConfig = {
   }
 };
 
-interface OpenAIChatCompleteResponse extends ChatCompletionResponse {
+export interface OpenAIChatCompleteResponse extends ChatCompletionResponse {
   system_fingerprint: string;
 }
 
 export const OpenAIChatCompleteResponseTransform: (response: OpenAIChatCompleteResponse) => ChatCompletionResponse = (response) => response;
+
+/**
+ * Transforms an OpenAI-format chat completions JSON response into an array of formatted OpenAI compatible text/event-stream chunks.
+ *
+ * @param {Object} response - The OpenAIChatCompleteResponse object.
+ * @param {string} provider - The provider string.
+ * @returns {Array<string>} - An array of formatted stream chunks.
+ */
+export const OpenAIChatCompleteJSONToStreamResponseTransform: (response: OpenAIChatCompleteResponse, provider: string) => Array<string> = (response, provider) => {
+  const streamChunkArray: Array<string> = [];
+  const { id, model, system_fingerprint, choices } = response;
+  const streamChunkTemplate = {
+    id,
+    object: "chat.completion.chunk",
+    created: Date.now(),
+    model: model || "",
+    system_fingerprint: system_fingerprint || null,
+    provider,
+    usage: {}
+  }
+
+  if (response.usage?.completion_tokens) {
+    streamChunkTemplate.usage = {
+      completion_tokens: response.usage?.completion_tokens
+    }
+  }
+
+  for (const [index, choice] of choices.entries()) {
+    if (choice.message && choice.message.tool_calls) {
+      const currentToolCall = choice.message.tool_calls[0];
+      const toolCallNameChunk = {
+        index: 0,
+        id: currentToolCall.id,
+        type: "function",
+        function: {
+          name: currentToolCall.function.name,
+          arguments: ""
+        }
+      }
+
+      const toolCallArgumentChunk = {
+        index: 0,
+        function: {
+          arguments: currentToolCall.function.arguments
+        }
+      }
+
+      streamChunkArray.push(`data: ${JSON.stringify({
+        ...streamChunkTemplate,
+        choices: [
+          {
+            index: index,
+            delta: {
+              role: "assistant",
+              content: null,
+              tool_calls: [toolCallNameChunk]
+            }
+          }
+        ]
+      })}\n\n`)
+
+      streamChunkArray.push(`data: ${JSON.stringify({
+        ...streamChunkTemplate,
+        choices: [
+          {
+            index: index,
+            delta: {
+              role: "assistant",
+              tool_calls: [toolCallArgumentChunk]
+            }
+          }
+        ]
+      })}\n\n`)
+    }
+
+    if (choice.message && choice.message.content && typeof choice.message.content === "string") {
+      const inidividualWords: Array<string> = [];
+      for (let i = 0; i < choice.message.content.length; i += 4) {
+        inidividualWords.push(choice.message.content.slice(i, i+4));
+      }
+      inidividualWords.forEach((word: string) => {
+        streamChunkArray.push(`data: ${JSON.stringify({
+          ...streamChunkTemplate,
+          choices: [
+            {
+              index: index,
+              delta: {
+                role: "assistant",
+                content: word
+              }
+            }
+          ]
+        })
+      }\n\n`)}) 
+    }
+
+    streamChunkArray.push(`data: ${JSON.stringify({
+      ...streamChunkTemplate,
+      choices: [
+        {
+          index: index,
+          delta: {},
+          finish_reason: choice.finish_reason
+        }
+      ]
+    })}\n\n`)
+  }
+
+  streamChunkArray.push(`data: [DONE]\n\n`)
+  return streamChunkArray;
+}

src/providers/openai/complete.ts

@@ -75,8 +75,68 @@ export const OpenAICompleteConfig: ProviderConfig = {
   }
 };
 
-interface OpenAICompleteResponse extends CompletionResponse {
+export interface OpenAICompleteResponse extends CompletionResponse {
   system_fingerprint: string;
 }
 
 export const OpenAICompleteResponseTransform: (response: OpenAICompleteResponse) => CompletionResponse = (response) => response;
+
+/**
+ * Transforms an OpenAI-format completions JSON into an array of formatted OpenAI compatible text/event-stream chunks.
+ *
+ * @param {Object} response - The OpenAICompleteResponse object.
+ * @param {string} provider - The provider string.
+ * @returns {Array<string>} - An array of formatted stream chunks.
+ */
+export const OpenAICompleteJSONToStreamResponseTransform: (response: OpenAICompleteResponse, provider: string) => Array<string> = (response, provider) => {
+  const streamChunkArray: Array<string> = [];
+  const { id, model, choices } = response;
+  const streamChunkTemplate = {
+    id: id,
+    object: "text_completion",
+    created: Date.now(),
+    model: model ?? "",
+    provider,
+    usage: {}
+  }
+
+  if (response.usage?.completion_tokens) {
+    streamChunkTemplate.usage = {
+      completion_tokens: response.usage?.completion_tokens
+    }
+  }
+
+  for (const [index, choice] of choices.entries()) {
+    if (choice.text) {
+      const inidividualWords = [];
+      for (let i = 0; i < choice.text.length; i += 4) {
+        inidividualWords.push(choice.text.slice(i, i+4));
+      }
+      inidividualWords.forEach((word: string) => {
+        streamChunkArray.push(`data: ${JSON.stringify({
+          ...streamChunkTemplate,
+          choices: [
+            {
+              index: index,
+              text: word
+            }
+          ]
+        })
+      }\n\n`)}) 
+    }
+
+    streamChunkArray.push(`data: ${JSON.stringify({
+      ...streamChunkTemplate,
+      choices: [
+        {
+          index: index,
+          text: "",
+          finish_reason: choice.finish_reason
+        }
+      ]
+    })}\n\n`)
+  }
+
+  streamChunkArray.push(`data: [DONE]\n\n`)
+  return streamChunkArray;
+}