⚡ (openai) Use Vercel's AI SDK for streaming

apps/viewer/package.json

@@ -19,6 +19,7 @@
     "@typebot.io/js": "workspace:*",
     "@typebot.io/prisma": "workspace:*",
     "@typebot.io/react": "workspace:*",
+    "ai": "^2.1.3",
     "aws-sdk": "2.1384.0",
     "bot-engine": "workspace:*",
     "cors": "2.8.5",
@@ -30,6 +31,7 @@
     "nextjs-cors": "2.1.2",
     "nodemailer": "6.9.2",
     "openai": "3.2.1",
+    "openai-edge": "^1.1.0",
     "qs": "6.11.2",
     "react": "18.2.0",
     "react-dom": "18.2.0",

apps/viewer/src/features/blocks/integrations/openai/getChatCompletionStream.ts

@@ -6,10 +6,12 @@ import {
   OpenAICredentials,
 } from '@typebot.io/schemas/features/blocks/integrations/openai'
 import { SessionState } from '@typebot.io/schemas/features/chat'
-import type {
+import { OpenAIStream } from 'ai'
+import {
   ChatCompletionRequestMessage,
-  CreateChatCompletionRequest,
-} from 'openai'
+  Configuration,
+  OpenAIApi,
+} from 'openai-edge'
 
 export const getChatCompletionStream =
   (conn: Connection) =>
@@ -37,19 +39,18 @@ export const getChatCompletionStream =
       options.advancedSettings?.temperature
     )
 
-    const res = await fetch('https://api.openai.com/v1/chat/completions', {
-      headers: {
-        'Content-Type': 'application/json',
-        Authorization: `Bearer ${apiKey}`,
-      },
-      method: 'POST',
-      body: JSON.stringify({
-        messages,
-        model: options.model,
-        temperature,
-        stream: true,
-      } satisfies CreateChatCompletionRequest),
+    const config = new Configuration({
+      apiKey,
+    })
+
+    const openai = new OpenAIApi(config)
+
+    const response = await openai.createChatCompletion({
+      model: options.model,
+      temperature,
+      stream: true,
+      messages,
     })
 
-    return res.body
+    return OpenAIStream(response)
   }

apps/viewer/src/pages/api/integrations/openai/streamer.ts

@@ -1,6 +1,7 @@
 import { getChatCompletionStream } from '@/features/blocks/integrations/openai/getChatCompletionStream'
 import { connect } from '@planetscale/database'
 import { IntegrationBlockType, SessionState } from '@typebot.io/schemas'
+import { StreamingTextResponse } from 'ai'
 import { ChatCompletionRequestMessage } from 'openai'
 
 export const config = {
@@ -15,8 +16,7 @@ const handler = async (req: Request) => {
         'Access-Control-Allow-Origin': '*',
         'Access-Control-Allow-Methods': 'POST',
         'Access-Control-Expose-Headers': 'Content-Length, X-JSON',
-        'Access-Control-Allow-Headers':
-          'apikey,X-Client-Info, Content-Type, Authorization, Accept, Accept-Language, X-Authorization',
+        'Access-Control-Allow-Headers': '*',
       },
     })
   }
@@ -66,12 +66,10 @@ const handler = async (req: Request) => {
     messages
   )
 
-  if (!stream) return new Response('Missing credentials', { status: 400 })
+  if (!stream) return new Response('Could not create stream', { status: 400 })
 
-  return new Response(stream, {
-    status: 200,
+  return new StreamingTextResponse(stream, {
     headers: {
-      'Content-Type': 'application/json; charset=utf-8',
       'Access-Control-Allow-Origin': '*',
     },
   })

packages/embeds/js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@typebot.io/js",
-  "version": "0.0.64",
+  "version": "0.0.65",
   "description": "Javascript library to display typebots on your website",
   "type": "module",
   "main": "dist/index.js",

packages/embeds/js/src/features/blocks/integrations/openai/streamChat.ts

@@ -1,10 +1,8 @@
-import { getOpenAiStreamerQuery } from '@/queries/getOpenAiStreamerQuery'
 import { ClientSideActionContext } from '@/types'
-import {
-  createParser,
-  ParsedEvent,
-  ReconnectInterval,
-} from 'eventsource-parser'
+import { guessApiHost } from '@/utils/guessApiHost'
+import { isNotEmpty } from '@typebot.io/lib/utils'
+
+let abortController: AbortController | null = null
 
 export const streamChat =
   (context: ClientSideActionContext) =>
@@ -13,59 +11,76 @@ export const streamChat =
       content?: string | undefined
       role?: 'system' | 'user' | 'assistant' | undefined
     }[],
-    {
-      onStreamedMessage,
-      isRetrying,
-    }: { onStreamedMessage?: (message: string) => void; isRetrying?: boolean }
+    { onStreamedMessage }: { onStreamedMessage?: (message: string) => void }
   ): Promise<{ message?: string; error?: object }> => {
-    const data = await getOpenAiStreamerQuery(context)(messages)
-
-    if (!data) return { error: { message: "Couldn't get streamer data" } }
+    try {
+      abortController = new AbortController()
 
-    let message = ''
+      const apiHost = context.apiHost
 
-    const reader = data.getReader()
-    const decoder = new TextDecoder()
+      const res = await fetch(
+        `${
+          isNotEmpty(apiHost) ? apiHost : guessApiHost()
+        }/api/integrations/openai/streamer`,
+        {
+          method: 'POST',
+          headers: {
+            'Content-Type': 'application/json',
+          },
+          body: JSON.stringify({
+            sessionId: context.sessionId,
+            messages,
+          }),
+          signal: abortController.signal,
+        }
+      )
 
-    const onParse = (event: ParsedEvent | ReconnectInterval) => {
-      if (event.type === 'event') {
-        const data = event.data
-        try {
-          const json = JSON.parse(data) as {
-            choices: { delta: { content: string } }[]
-          }
-          const text = json.choices.at(0)?.delta.content
-          if (!text) return
-          message += text
-          onStreamedMessage?.(message)
-        } catch (e) {
-          console.error(e)
+      if (!res.ok) {
+        return {
+          error: {
+            message: (await res.text()) || 'Failed to fetch the chat response.',
+          },
         }
       }
-    }
 
-    const parser = createParser(onParse)
+      if (!res.body) {
+        throw new Error('The response body is empty.')
+      }
+
+      let message = ''
+
+      const reader = res.body.getReader()
+      const decoder = new TextDecoder()
 
-    // eslint-disable-next-line no-constant-condition
-    while (true) {
-      const { value, done } = await reader.read()
-      if (done || !value) break
-      const dataString = decoder.decode(value)
-      if (dataString.includes('503 Service Temporarily Unavailable')) {
-        if (isRetrying)
-          return { error: { message: "Couldn't get streamer data" } }
-        await new Promise((resolve) => setTimeout(resolve, 3000))
-        return streamChat(context)(messages, {
-          onStreamedMessage,
-          isRetrying: true,
-        })
+      // eslint-disable-next-line no-constant-condition
+      while (true) {
+        const { done, value } = await reader.read()
+        if (done) {
+          break
+        }
+        const chunk = decoder.decode(value)
+        if (onStreamedMessage) onStreamedMessage(chunk)
+        message += chunk
+        if (abortController === null) {
+          reader.cancel()
+          break
+        }
       }
-      if (dataString.includes('[DONE]')) break
-      if (dataString.includes('"error":')) {
-        return { error: JSON.parse(dataString).error }
+
+      abortController = null
+
+      return { message }
+    } catch (err) {
+      console.error(err)
+      // Ignore abort errors as they are expected.
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      if ((err as any).name === 'AbortError') {
+        abortController = null
+        return { error: { message: 'Request aborted' } }
       }
-      parser.feed(dataString)
-    }
 
-    return { message }
+      if (err instanceof Error) return { error: { message: err.message } }
+
+      return { error: { message: 'Failed to fetch the chat response.' } }
+    }
   }

packages/embeds/react/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@typebot.io/react",
-  "version": "0.0.64",
+  "version": "0.0.65",
   "description": "React library to display typebots on your website",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",