🚸 (dify) Improve error display when streaming

apps/viewer/src/app/api/integrations/openai/streamer/route.ts

@@ -153,7 +153,7 @@ export async function POST(req: Request) {
       // eslint-disable-next-line @typescript-eslint/no-unused-vars
       set: (_1: string, _2: unknown) => {},
     }
-    const stream = await action.run.stream.run({
+    const { stream } = await action.run.stream.run({
       credentials: decryptedCredentials,
       options: block.options,
       variables,

packages/bot-engine/apiHandlers/getMessageStream.ts

@@ -23,7 +23,14 @@ type Props = {
   messages: OpenAI.Chat.ChatCompletionMessage[] | undefined
 }
 
-export const getMessageStream = async ({ sessionId, messages }: Props) => {
+export const getMessageStream = async ({
+  sessionId,
+  messages,
+}: Props): Promise<{
+  stream?: ReadableStream<any>
+  status?: number
+  message?: string
+}> => {
   const session = await getSession(sessionId)
 
   if (!session?.state || !session.state.currentBlockId)
@@ -130,13 +137,15 @@ export const getMessageStream = async ({ sessionId, messages }: Props) => {
         })
       },
     }
-    const stream = await action.run.stream.run({
+    const { stream, httpError } = await action.run.stream.run({
       credentials: decryptedCredentials,
       options: deepParseVariables(
         session.state.typebotsQueue[0].typebot.variables
       )(block.options),
       variables,
     })
+    if (httpError) return httpError
+
     if (!stream) return { status: 500, message: 'Could not create stream' }
 
     return { stream }

packages/forge/blocks/anthropic/actions/createChatMessage.tsx

@@ -168,7 +168,7 @@ export const createChatMessage = createAction({
           stream: true,
         })
 
-        return AnthropicStream(response)
+        return { stream: AnthropicStream(response) }
       },
     },
   },

packages/forge/blocks/difyAi/actions/createChatMessage.ts

@@ -3,7 +3,7 @@ import { isDefined, isEmpty, isNotEmpty } from '@typebot.io/lib'
 import { auth } from '../auth'
 import { defaultBaseUrl } from '../constants'
 import { Chunk } from '../types'
-import ky from 'ky'
+import ky, { HTTPError } from 'ky'
 import { deprecatedCreateChatMessageOptions } from '../deprecated'
 
 export const createChatMessage = createAction({
@@ -68,77 +68,94 @@ export const createChatMessage = createAction({
         const existingDifyConversationId = conversationVariableId
           ? variables.get(conversationVariableId)
           : conversation_id
-        const response = await ky(
+        try {
-          (apiEndpoint ?? defaultBaseUrl) + '/v1/chat-messages',
+          const response = await ky(
-          {
+            (apiEndpoint ?? defaultBaseUrl) + '/v1/chat-messages',
-            method: 'POST',
+            {
-            headers: {
+              method: 'POST',
-              Authorization: `Bearer ${apiKey}`,
+              headers: {
-              'Content-Type': 'application/json',
+                Authorization: `Bearer ${apiKey}`,
-            },
+                'Content-Type': 'application/json',
-            body: JSON.stringify({
+              },
-              inputs:
+              body: JSON.stringify({
-                inputs?.reduce((acc, { key, value }) => {
+                inputs:
-                  if (isEmpty(key) || isEmpty(value)) return acc
+                  inputs?.reduce((acc, { key, value }) => {
-                  return {
+                    if (isEmpty(key) || isEmpty(value)) return acc
-                    ...acc,
+                    return {
-                    [key]: value,
+                      ...acc,
-                  }
+                      [key]: value,
-                }, {}) ?? {},
+                    }
-              query,
+                  }, {}) ?? {},
-              response_mode: 'streaming',
+                query,
-              conversation_id: existingDifyConversationId,
+                response_mode: 'streaming',
-              user,
+                conversation_id: existingDifyConversationId,
-              files: [],
+                user,
+                files: [],
+              }),
+            }
+          )
+          const reader = response.body?.getReader()
+
+          if (!reader) return {}
+
+          return {
+            stream: new ReadableStream({
+              async start(controller) {
+                try {
+                  await processDifyStream(reader, {
+                    onDone: () => {
+                      controller.close()
+                    },
+                    onMessage: (message) => {
+                      controller.enqueue(
+                        new TextEncoder().encode(
+                          '0:"' + message.replace(/"/g, '\\"') + '"\n'
+                        )
+                      )
+                    },
+                    onMessageEnd({ totalTokens, conversationId }) {
+                      if (
+                        conversationVariableId &&
+                        isNotEmpty(conversationId) &&
+                        isEmpty(existingDifyConversationId?.toString())
+                      )
+                        variables.set(conversationVariableId, conversationId)
+
+                      if ((responseMapping?.length ?? 0) === 0) return
+                      responseMapping?.forEach((mapping) => {
+                        if (!mapping.variableId) return
+
+                        if (
+                          mapping.item === 'Conversation ID' &&
+                          isNotEmpty(conversationId) &&
+                          isEmpty(existingDifyConversationId?.toString())
+                        )
+                          variables.set(mapping.variableId, conversationId)
+
+                        if (mapping.item === 'Total Tokens')
+                          variables.set(mapping.variableId, totalTokens)
+                      })
+                    },
+                  })
+                } catch (e) {
+                  console.error(e)
+                  controller.error(e) // Properly closing the stream with an error
+                }
+              },
             }),
           }
-        )
+        } catch (err) {
-        const reader = response.body?.getReader()
+          if (err instanceof HTTPError) {
-
+            return {
-        if (!reader) return
+              httpError: {
-
+                status: err.response.status,
-        return new ReadableStream({
+                message: await err.response.text(),
-          async start(controller) {
+              },
-            try {
-              await processDifyStream(reader, {
-                onDone: () => {
-                  controller.close()
-                },
-                onMessage: (message) => {
-                  controller.enqueue(
-                    new TextEncoder().encode('0:"' + message + '"\n')
-                  )
-                },
-                onMessageEnd({ totalTokens, conversationId }) {
-                  if (
-                    conversationVariableId &&
-                    isNotEmpty(conversationId) &&
-                    isEmpty(existingDifyConversationId?.toString())
-                  )
-                    variables.set(conversationVariableId, conversationId)
-
-                  if ((responseMapping?.length ?? 0) === 0) return
-                  responseMapping?.forEach((mapping) => {
-                    if (!mapping.variableId) return
-
-                    if (
-                      mapping.item === 'Conversation ID' &&
-                      isNotEmpty(conversationId) &&
-                      isEmpty(existingDifyConversationId?.toString())
-                    )
-                      variables.set(mapping.variableId, conversationId)
-
-                    if (mapping.item === 'Total Tokens')
-                      variables.set(mapping.variableId, totalTokens)
-                  })
-                },
-              })
-            } catch (e) {
-              console.error(e)
-              controller.error(e) // Properly closing the stream with an error
             }
-          },
+          }
-        })
+          console.error(err)
+          return {}
+        }
       },
     },
     server: async ({
@@ -243,12 +260,15 @@ export const createChatMessage = createAction({
           if (item === 'Total Tokens')
             variables.set(mapping.variableId, totalTokens)
         })
-      } catch (error) {
+      } catch (err) {
-        logs.add({
+        if (err instanceof HTTPError) {
-          status: 'error',
+          return logs.add({
-          description: 'Failed to create chat message',
+            status: 'error',
-        })
+            description: err.message,
-        console.error(error)
+            details: await err.response.text(),
+          })
+        }
+        console.error(err)
       }
     },
   },

packages/forge/blocks/mistral/actions/createChatCompletion.ts

@@ -139,7 +139,7 @@ export const createChatCompletion = createAction({
           (res) => res.item === 'Message content' || !res.item
         )?.variableId,
       run: async ({ credentials: { apiKey }, options, variables }) => {
-        if (!options.model) return
+        if (!options.model) return {}
         const model = createMistral({
           apiKey,
         })(options.model)
@@ -149,7 +149,7 @@ export const createChatCompletion = createAction({
           messages: parseMessages({ options, variables }),
         })
 
-        return response.toAIStream()
+        return { stream: response.toAIStream() }
       },
     },
   },

packages/forge/blocks/openRouter/actions/createChatCompletion.tsx

@@ -62,11 +62,12 @@ export const createChatCompletion = createAction({
       }),
     stream: {
       getStreamVariableId: getChatCompletionStreamVarId,
-      run: (params) =>
+      run: async (params) => ({
-        runChatCompletionStream({
+        stream: await runChatCompletionStream({
           ...params,
           config: { baseUrl: defaultOpenRouterOptions.baseUrl },
         }),
+      }),
     },
   },
 })

packages/forge/blocks/openai/actions/askAssistant.tsx

@@ -142,8 +142,8 @@ export const askAssistant = createAction({
       getStreamVariableId: ({ responseMapping }) =>
         responseMapping?.find((m) => !m.item || m.item === 'Message')
           ?.variableId,
-      run: async ({ credentials, options, variables }) =>
+      run: async ({ credentials, options, variables }) => ({
-        createAssistantStream({
+        stream: await createAssistantStream({
           apiKey: credentials.apiKey,
           assistantId: options.assistantId,
           message: options.message,
@@ -154,6 +154,7 @@ export const askAssistant = createAction({
           functions: options.functions,
           responseMapping: options.responseMapping,
         }),
+      }),
     },
     server: async ({
       credentials: { apiKey },

packages/forge/blocks/openai/actions/createChatCompletion.tsx

@@ -86,14 +86,15 @@ export const createChatCompletion = createAction({
       }),
     stream: {
       getStreamVariableId: getChatCompletionStreamVarId,
-      run: (params) =>
+      run: async (params) => ({
-        runChatCompletionStream({
+        stream: await runChatCompletionStream({
           ...params,
           config: {
             baseUrl: defaultOpenAIOptions.baseUrl,
             defaultModel: defaultOpenAIOptions.model,
           },
         }),
+      }),
     },
   },
 })

packages/forge/blocks/togetherAi/actions/createChatCompletion.tsx

@@ -45,11 +45,12 @@ export const createChatCompletion = createAction({
       }),
     stream: {
       getStreamVariableId: getChatCompletionStreamVarId,
-      run: (params) =>
+      run: async (params) => ({
-        runChatCompletionStream({
+        stream: await runChatCompletionStream({
           ...params,
           config: { baseUrl: defaultTogetherOptions.baseUrl },
         }),
+      }),
     },
   },
 })

packages/forge/core/types.ts

@@ -64,7 +64,10 @@ export type ActionDefinition<
         credentials: CredentialsFromAuthDef<A>
         options: z.infer<BaseOptions> & z.infer<Options>
         variables: VariableStore
-      }) => Promise<ReadableStream<any> | undefined>
+      }) => Promise<{
+        stream?: ReadableStream<any>
+        httpError?: { status: number; message: string }
+      }>
     }
     web?: {
       displayEmbedBubble?: {