Feature/Add bullmq redis for message queue processing

package.json

@@ -17,6 +17,9 @@
         "start": "run-script-os",
         "start:windows": "cd packages/server/bin && run start",
         "start:default": "cd packages/server/bin && ./run start",
+        "start-worker": "run-script-os",
+        "start-worker:windows": "cd packages/server/bin && run worker",
+        "start-worker:default": "cd packages/server/bin && ./run worker",
         "clean": "pnpm --filter \"./packages/**\" clean",
         "nuke": "pnpm --filter \"./packages/**\" nuke && rimraf node_modules .turbo",
         "format": "prettier --write \"**/*.{ts,tsx,md}\"",

packages/components/nodes/agents/ConversationalAgent/ConversationalAgent.ts

@@ -21,7 +21,7 @@ import {
 } from '../../../src/Interface'
 import { AgentExecutor } from '../../../src/agents'
 import { addImagesToMessages, llmSupportsVision } from '../../../src/multiModalUtils'
-import { checkInputs, Moderation } from '../../moderation/Moderation'
+import { checkInputs, Moderation, streamResponse } from '../../moderation/Moderation'
 import { formatResponse } from '../../outputparsers/OutputParserHelpers'
 
 const DEFAULT_PREFIX = `Assistant is a large language model trained by OpenAI.
@@ -124,10 +124,9 @@ class ConversationalAgent_Agents implements INode {
                 input = await checkInputs(moderations, input)
             } catch (e) {
                 await new Promise((resolve) => setTimeout(resolve, 500))
-                // if (options.shouldStreamResponse) {
-                //     streamResponse(options.sseStreamer, options.chatId, e.message)
-                // }
-                //streamResponse(options.socketIO && options.socketIOClientId, e.message, options.socketIO, options.socketIOClientId)
+                if (options.shouldStreamResponse) {
+                    streamResponse(sseStreamer, chatId, e.message)
+                }
                 return formatResponse(e.message)
             }
         }

packages/components/nodes/cache/InMemoryCache/InMemoryCache.ts

@@ -27,17 +27,17 @@ class InMemoryCache implements INode {
     }
 
     async init(nodeData: INodeData, _: string, options: ICommonObject): Promise<any> {
-        const memoryMap = options.cachePool.getLLMCache(options.chatflowid) ?? new Map()
+        const memoryMap = (await options.cachePool.getLLMCache(options.chatflowid)) ?? new Map()
         const inMemCache = new InMemoryCacheExtended(memoryMap)
 
         inMemCache.lookup = async (prompt: string, llmKey: string): Promise<any | null> => {
-            const memory = options.cachePool.getLLMCache(options.chatflowid) ?? inMemCache.cache
+            const memory = (await options.cachePool.getLLMCache(options.chatflowid)) ?? inMemCache.cache
             return Promise.resolve(memory.get(getCacheKey(prompt, llmKey)) ?? null)
         }
 
         inMemCache.update = async (prompt: string, llmKey: string, value: any): Promise<void> => {
             inMemCache.cache.set(getCacheKey(prompt, llmKey), value)
-            options.cachePool.addLLMCache(options.chatflowid, inMemCache.cache)
+            await options.cachePool.addLLMCache(options.chatflowid, inMemCache.cache)
         }
         return inMemCache
     }

packages/components/nodes/cache/InMemoryCache/InMemoryEmbeddingCache.ts

@@ -43,26 +43,26 @@ class InMemoryEmbeddingCache implements INode {
     async init(nodeData: INodeData, _: string, options: ICommonObject): Promise<any> {
         const namespace = nodeData.inputs?.namespace as string
         const underlyingEmbeddings = nodeData.inputs?.embeddings as Embeddings
-        const memoryMap = options.cachePool.getEmbeddingCache(options.chatflowid) ?? {}
+        const memoryMap = (await options.cachePool.getEmbeddingCache(options.chatflowid)) ?? {}
         const inMemCache = new InMemoryEmbeddingCacheExtended(memoryMap)
 
         inMemCache.mget = async (keys: string[]) => {
-            const memory = options.cachePool.getEmbeddingCache(options.chatflowid) ?? inMemCache.store
+            const memory = (await options.cachePool.getEmbeddingCache(options.chatflowid)) ?? inMemCache.store
             return keys.map((key) => memory[key])
         }
 
         inMemCache.mset = async (keyValuePairs: [string, any][]): Promise<void> => {
             for (const [key, value] of keyValuePairs) {
                 inMemCache.store[key] = value
             }
-            options.cachePool.addEmbeddingCache(options.chatflowid, inMemCache.store)
+            await options.cachePool.addEmbeddingCache(options.chatflowid, inMemCache.store)
         }
 
         inMemCache.mdelete = async (keys: string[]): Promise<void> => {
             for (const key of keys) {
                 delete inMemCache.store[key]
             }
-            options.cachePool.addEmbeddingCache(options.chatflowid, inMemCache.store)
+            await options.cachePool.addEmbeddingCache(options.chatflowid, inMemCache.store)
         }
 
         return CacheBackedEmbeddings.fromBytesStore(underlyingEmbeddings, inMemCache, {

packages/components/nodes/cache/RedisCache/RedisCache.ts

@@ -1,47 +1,10 @@
-import { Redis, RedisOptions } from 'ioredis'
-import { isEqual } from 'lodash'
+import { Redis } from 'ioredis'
 import hash from 'object-hash'
 import { RedisCache as LangchainRedisCache } from '@langchain/community/caches/ioredis'
 import { StoredGeneration, mapStoredMessageToChatMessage } from '@langchain/core/messages'
 import { Generation, ChatGeneration } from '@langchain/core/outputs'
 import { getBaseClasses, getCredentialData, getCredentialParam, ICommonObject, INode, INodeData, INodeParams } from '../../../src'
 
-let redisClientSingleton: Redis
-let redisClientOption: RedisOptions
-let redisClientUrl: string
-
-const getRedisClientbyOption = (option: RedisOptions) => {
-    if (!redisClientSingleton) {
-        // if client doesn't exists
-        redisClientSingleton = new Redis(option)
-        redisClientOption = option
-        return redisClientSingleton
-    } else if (redisClientSingleton && !isEqual(option, redisClientOption)) {
-        // if client exists but option changed
-        redisClientSingleton.quit()
-        redisClientSingleton = new Redis(option)
-        redisClientOption = option
-        return redisClientSingleton
-    }
-    return redisClientSingleton
-}
-
-const getRedisClientbyUrl = (url: string) => {
-    if (!redisClientSingleton) {
-        // if client doesn't exists
-        redisClientSingleton = new Redis(url)
-        redisClientUrl = url
-        return redisClientSingleton
-    } else if (redisClientSingleton && url !== redisClientUrl) {
-        // if client exists but option changed
-        redisClientSingleton.quit()
-        redisClientSingleton = new Redis(url)
-        redisClientUrl = url
-        return redisClientSingleton
-    }
-    return redisClientSingleton
-}
-
 class RedisCache implements INode {
     label: string
     name: string
@@ -85,33 +48,19 @@ class RedisCache implements INode {
     async init(nodeData: INodeData, _: string, options: ICommonObject): Promise<any> {
         const ttl = nodeData.inputs?.ttl as string
 
-        const credentialData = await getCredentialData(nodeData.credential ?? '', options)
-        const redisUrl = getCredentialParam('redisUrl', credentialData, nodeData)
-
-        let client: Redis
-        if (!redisUrl || redisUrl === '') {
-            const username = getCredentialParam('redisCacheUser', credentialData, nodeData)
-            const password = getCredentialParam('redisCachePwd', credentialData, nodeData)
-            const portStr = getCredentialParam('redisCachePort', credentialData, nodeData)
-            const host = getCredentialParam('redisCacheHost', credentialData, nodeData)
-            const sslEnabled = getCredentialParam('redisCacheSslEnabled', credentialData, nodeData)
-
-            const tlsOptions = sslEnabled === true ? { tls: { rejectUnauthorized: false } } : {}
-
-            client = getRedisClientbyOption({
-                port: portStr ? parseInt(portStr) : 6379,
-                host,
-                username,
-                password,
-                ...tlsOptions
-            })
-        } else {
-            client = getRedisClientbyUrl(redisUrl)
-        }
-
+        let client = await getRedisClient(nodeData, options)
         const redisClient = new LangchainRedisCache(client)
 
         redisClient.lookup = async (prompt: string, llmKey: string) => {
+            try {
+                const pingResp = await client.ping()
+                if (pingResp !== 'PONG') {
+                    client = await getRedisClient(nodeData, options)
+                }
+            } catch (error) {
+                client = await getRedisClient(nodeData, options)
+            }
+
             let idx = 0
             let key = getCacheKey(prompt, llmKey, String(idx))
             let value = await client.get(key)
@@ -125,10 +74,21 @@ class RedisCache implements INode {
                 value = await client.get(key)
             }
 
+            client.quit()
+
             return generations.length > 0 ? generations : null
         }
 
         redisClient.update = async (prompt: string, llmKey: string, value: Generation[]) => {
+            try {
+                const pingResp = await client.ping()
+                if (pingResp !== 'PONG') {
+                    client = await getRedisClient(nodeData, options)
+                }
+            } catch (error) {
+                client = await getRedisClient(nodeData, options)
+            }
+
             for (let i = 0; i < value.length; i += 1) {
                 const key = getCacheKey(prompt, llmKey, String(i))
                 if (ttl) {
@@ -137,12 +97,43 @@ class RedisCache implements INode {
                     await client.set(key, JSON.stringify(serializeGeneration(value[i])))
                 }
             }
+
+            client.quit()
         }
 
+        client.quit()
+
         return redisClient
     }
 }
+const getRedisClient = async (nodeData: INodeData, options: ICommonObject) => {
+    let client: Redis
+
+    const credentialData = await getCredentialData(nodeData.credential ?? '', options)
+    const redisUrl = getCredentialParam('redisUrl', credentialData, nodeData)
+
+    if (!redisUrl || redisUrl === '') {
+        const username = getCredentialParam('redisCacheUser', credentialData, nodeData)
+        const password = getCredentialParam('redisCachePwd', credentialData, nodeData)
+        const portStr = getCredentialParam('redisCachePort', credentialData, nodeData)
+        const host = getCredentialParam('redisCacheHost', credentialData, nodeData)
+        const sslEnabled = getCredentialParam('redisCacheSslEnabled', credentialData, nodeData)
+
+        const tlsOptions = sslEnabled === true ? { tls: { rejectUnauthorized: false } } : {}
+
+        client = new Redis({
+            port: portStr ? parseInt(portStr) : 6379,
+            host,
+            username,
+            password,
+            ...tlsOptions
+        })
+    } else {
+        client = new Redis(redisUrl)
+    }
 
+    return client
+}
 const getCacheKey = (...strings: string[]): string => hash(strings.join('_'))
 const deserializeStoredGeneration = (storedGeneration: StoredGeneration) => {
     if (storedGeneration.message !== undefined) {