diff --git a/src/globals.ts b/src/globals.ts index 036ec9ff..a867ef9e 100644 --- a/src/globals.ts +++ b/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" } \ No newline at end of file diff --git a/src/handlers/handlerUtils.ts b/src/handlers/handlerUtils.ts index 38baa547..d36da115 100644 --- a/src/handlers/handlerUtils.ts +++ b/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 { - // 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} - 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 { 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) diff --git a/src/handlers/streamHandler.ts b/src/handlers/streamHandler.ts index d4ae6efc..0cb7e30f 100644 --- a/src/handlers/streamHandler.ts +++ b/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 { + 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 + }) + }); +} + diff --git a/src/providers/mistral-ai/chatComplete.ts b/src/providers/mistral-ai/chatComplete.ts index bbea47c6..a64f0c4b 100644 --- a/src/providers/mistral-ai/chatComplete.ts +++ b/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 }, }; diff --git a/src/providers/openai/chatComplete.ts b/src/providers/openai/chatComplete.ts index 927e9b8b..fdc236fa 100644 --- a/src/providers/openai/chatComplete.ts +++ b/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} - An array of formatted stream chunks. + */ +export const OpenAIChatCompleteJSONToStreamResponseTransform: (response: OpenAIChatCompleteResponse, provider: string) => Array = (response, provider) => { + const streamChunkArray: Array = []; + 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 = []; + 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; +} diff --git a/src/providers/openai/complete.ts b/src/providers/openai/complete.ts index 26c7ecdf..dc73bb6b 100644 --- a/src/providers/openai/complete.ts +++ b/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} - An array of formatted stream chunks. + */ +export const OpenAICompleteJSONToStreamResponseTransform: (response: OpenAICompleteResponse, provider: string) => Array = (response, provider) => { + const streamChunkArray: Array = []; + 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; +} \ No newline at end of file