Feature/add endpoints for realtime api (#3318)

add endpoints for realtime api

packages/components/nodes/tools/Searxng/Searxng.ts

@@ -2,6 +2,10 @@ import { Tool } from '@langchain/core/tools'
 import { INode, INodeData, INodeParams } from '../../../src/Interface'
 import { getBaseClasses } from '../../../src/utils'
 
+const defaultDesc =
+    'A meta search engine. Useful for when you need to answer questions about current events. Input should be a search query. Output is a JSON array of the query results'
+const defaultName = 'searxng-search'
+
 interface SearxngResults {
     query: string
     number_of_results: number
@@ -65,7 +69,7 @@ class Searxng_Tools implements INode {
     constructor() {
         this.label = 'SearXNG'
         this.name = 'searXNG'
-        this.version = 2.0
+        this.version = 3.0
         this.type = 'SearXNG'
         this.icon = 'SearXNG.svg'
         this.category = 'Tools'
@@ -75,7 +79,20 @@ class Searxng_Tools implements INode {
                 label: 'Base URL',
                 name: 'apiBase',
                 type: 'string',
-                default: 'http://searxng:8080'
+                default: 'http://localhost:8080'
+            },
+            {
+                label: 'Tool Name',
+                name: 'toolName',
+                type: 'string',
+                default: defaultName
+            },
+            {
+                label: 'Tool Description',
+                name: 'toolDescription',
+                type: 'string',
+                rows: 4,
+                default: defaultDesc
             },
             {
                 label: 'Headers',
@@ -101,7 +118,7 @@ class Searxng_Tools implements INode {
                 ],
                 default: 'json',
                 description:
-                    'Format of the response. You need to enable search formats in settings.yml. Refer to <a target="_blank" href="https://docs.flowiseai.com/integrations/langchain/tools/searchapi">SearXNG Setup Guide</a> for more details.',
+                    'Format of the response. You need to enable search formats in settings.yml. Refer to <a target="_blank" href="https://docs.flowiseai.com/integrations/langchain/tools/searxng">SearXNG Setup Guide</a> for more details.',
                 additionalParams: true
             },
             {
@@ -170,6 +187,8 @@ class Searxng_Tools implements INode {
         const time_range = nodeData.inputs?.time_range as string
         const safesearch = nodeData.inputs?.safesearch as 0 | 1 | 2 | undefined
         const format = nodeData.inputs?.format as string
+        const toolName = nodeData.inputs?.toolName as string
+        const toolDescription = nodeData.inputs?.toolDescription as string
 
         const params: SearxngSearchParams = {}
 
@@ -189,7 +208,9 @@ class Searxng_Tools implements INode {
         const tool = new SearxngSearch({
             apiBase,
             params,
-            headers: customHeaders
+            headers: customHeaders,
+            toolName,
+            toolDescription
         })
 
         return tool
@@ -201,10 +222,9 @@ class SearxngSearch extends Tool {
         return 'SearxngSearch'
     }
 
-    name = 'searxng-search'
+    name = defaultName
 
-    description =
+    description = defaultDesc
-        'A meta search engine. Useful for when you need to answer questions about current events. Input should be a search query. Output is a JSON array of the query results'
 
     protected apiBase?: string
 
@@ -223,7 +243,19 @@ class SearxngSearch extends Tool {
         }
     }
 
-    constructor({ apiBase, params, headers }: { apiBase?: string; params?: SearxngSearchParams; headers?: SearxngCustomHeaders }) {
+    constructor({
+        apiBase,
+        params,
+        headers,
+        toolName,
+        toolDescription
+    }: {
+        apiBase?: string
+        params?: SearxngSearchParams
+        headers?: SearxngCustomHeaders
+        toolName?: string
+        toolDescription?: string
+    }) {
         super(...arguments)
 
         this.apiBase = apiBase
@@ -236,6 +268,14 @@ class SearxngSearch extends Tool {
         if (params) {
             this.params = { ...this.params, ...params }
         }
+
+        if (toolName) {
+            this.name = toolName
+        }
+
+        if (toolDescription) {
+            this.description = toolDescription
+        }
     }
 
     protected buildUrl<P extends SearxngSearchParams>(path: string, parameters: P, baseUrl: string): string {

packages/server/src/controllers/openai-realtime/index.ts

@@ -0,0 +1,68 @@
+import { Request, Response, NextFunction } from 'express'
+import openaiRealTimeService from '../../services/openai-realtime'
+import { InternalFlowiseError } from '../../errors/internalFlowiseError'
+import { StatusCodes } from 'http-status-codes'
+
+const getAgentTools = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+        if (typeof req.params === 'undefined' || !req.params.id) {
+            throw new InternalFlowiseError(
+                StatusCodes.PRECONDITION_FAILED,
+                `Error: openaiRealTimeController.getAgentTools - id not provided!`
+            )
+        }
+        const apiResponse = await openaiRealTimeService.getAgentTools(req.params.id)
+        return res.json(apiResponse)
+    } catch (error) {
+        next(error)
+    }
+}
+
+const executeAgentTool = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+        if (typeof req.params === 'undefined' || !req.params.id) {
+            throw new InternalFlowiseError(
+                StatusCodes.PRECONDITION_FAILED,
+                `Error: openaiRealTimeController.executeAgentTool - id not provided!`
+            )
+        }
+        if (!req.body) {
+            throw new InternalFlowiseError(
+                StatusCodes.PRECONDITION_FAILED,
+                `Error: openaiRealTimeController.executeAgentTool - body not provided!`
+            )
+        }
+        if (!req.body.chatId) {
+            throw new InternalFlowiseError(
+                StatusCodes.PRECONDITION_FAILED,
+                `Error: openaiRealTimeController.executeAgentTool - body chatId not provided!`
+            )
+        }
+        if (!req.body.toolName) {
+            throw new InternalFlowiseError(
+                StatusCodes.PRECONDITION_FAILED,
+                `Error: openaiRealTimeController.executeAgentTool - body toolName not provided!`
+            )
+        }
+        if (!req.body.inputArgs) {
+            throw new InternalFlowiseError(
+                StatusCodes.PRECONDITION_FAILED,
+                `Error: openaiRealTimeController.executeAgentTool - body inputArgs not provided!`
+            )
+        }
+        const apiResponse = await openaiRealTimeService.executeAgentTool(
+            req.params.id,
+            req.body.chatId,
+            req.body.toolName,
+            req.body.inputArgs
+        )
+        return res.json(apiResponse)
+    } catch (error) {
+        next(error)
+    }
+}
+
+export default {
+    getAgentTools,
+    executeAgentTool
+}

packages/server/src/routes/index.ts

@@ -28,6 +28,7 @@ import nodesRouter from './nodes'
 import openaiAssistantsRouter from './openai-assistants'
 import openaiAssistantsFileRouter from './openai-assistants-files'
 import openaiAssistantsVectorStoreRouter from './openai-assistants-vector-store'
+import openaiRealtimeRouter from './openai-realtime'
 import pingRouter from './ping'
 import predictionRouter from './predictions'
 import promptListsRouter from './prompts-lists'
@@ -73,6 +74,7 @@ router.use('/nodes', nodesRouter)
 router.use('/openai-assistants', openaiAssistantsRouter)
 router.use('/openai-assistants-file', openaiAssistantsFileRouter)
 router.use('/openai-assistants-vector-store', openaiAssistantsVectorStoreRouter)
+router.use('/openai-realtime', openaiRealtimeRouter)
 router.use('/prediction', predictionRouter)
 router.use('/prompts-list', promptListsRouter)
 router.use('/public-chatbotConfig', publicChatbotRouter)

packages/server/src/routes/openai-realtime/index.ts

@@ -0,0 +1,12 @@
+import express from 'express'
+import openaiRealTimeController from '../../controllers/openai-realtime'
+
+const router = express.Router()
+
+// GET
+router.get(['/', '/:id'], openaiRealTimeController.getAgentTools)
+
+// EXECUTE
+router.post(['/', '/:id'], openaiRealTimeController.executeAgentTool)
+
+export default router

packages/server/src/services/openai-realtime/index.ts

@@ -0,0 +1,174 @@
+import { StatusCodes } from 'http-status-codes'
+import { InternalFlowiseError } from '../../errors/internalFlowiseError'
+import { getErrorMessage } from '../../errors/utils'
+import { buildFlow, constructGraphs, databaseEntities, getEndingNodes, getStartingNodes, resolveVariables } from '../../utils'
+import { getRunningExpressApp } from '../../utils/getRunningExpressApp'
+import { ChatFlow } from '../../database/entities/ChatFlow'
+import { IDepthQueue, IReactFlowNode } from '../../Interface'
+import { ICommonObject, INodeData } from 'flowise-components'
+import { convertToOpenAIFunction } from '@langchain/core/utils/function_calling'
+import { v4 as uuidv4 } from 'uuid'
+
+const SOURCE_DOCUMENTS_PREFIX = '\n\n----FLOWISE_SOURCE_DOCUMENTS----\n\n'
+const ARTIFACTS_PREFIX = '\n\n----FLOWISE_ARTIFACTS----\n\n'
+
+const buildAndInitTool = async (chatflowid: string, _chatId?: string) => {
+    const appServer = getRunningExpressApp()
+    const chatflow = await appServer.AppDataSource.getRepository(ChatFlow).findOneBy({
+        id: chatflowid
+    })
+    if (!chatflow) {
+        throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Chatflow ${chatflowid} not found`)
+    }
+
+    const chatId = _chatId || uuidv4()
+    const flowData = JSON.parse(chatflow.flowData)
+    const nodes = flowData.nodes
+    const edges = flowData.edges
+
+    const toolAgentNode = nodes.find(
+        (node: IReactFlowNode) => node.data.inputAnchors.find((acr) => acr.type === 'Tool') && node.data.category === 'Agents'
+    )
+    if (!toolAgentNode) {
+        throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Agent with tools not found in chatflow ${chatflowid}`)
+    }
+
+    const { graph, nodeDependencies } = constructGraphs(nodes, edges)
+    const directedGraph = graph
+    const endingNodes = getEndingNodes(nodeDependencies, directedGraph, nodes)
+
+    /*** Get Starting Nodes with Reversed Graph ***/
+    const constructedObj = constructGraphs(nodes, edges, { isReversed: true })
+    const nonDirectedGraph = constructedObj.graph
+    let startingNodeIds: string[] = []
+    let depthQueue: IDepthQueue = {}
+    const endingNodeIds = endingNodes.map((n) => n.id)
+    for (const endingNodeId of endingNodeIds) {
+        const resx = getStartingNodes(nonDirectedGraph, endingNodeId)
+        startingNodeIds.push(...resx.startingNodeIds)
+        depthQueue = Object.assign(depthQueue, resx.depthQueue)
+    }
+    startingNodeIds = [...new Set(startingNodeIds)]
+
+    const reactFlowNodes = await buildFlow({
+        startingNodeIds,
+        reactFlowNodes: nodes,
+        reactFlowEdges: edges,
+        graph,
+        depthQueue,
+        componentNodes: appServer.nodesPool.componentNodes,
+        question: '',
+        chatHistory: [],
+        chatId: chatId,
+        sessionId: chatId,
+        chatflowid,
+        appDataSource: appServer.AppDataSource
+    })
+
+    const nodeToExecute =
+        endingNodeIds.length === 1
+            ? reactFlowNodes.find((node: IReactFlowNode) => endingNodeIds[0] === node.id)
+            : reactFlowNodes[reactFlowNodes.length - 1]
+
+    if (!nodeToExecute) {
+        throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Node not found`)
+    }
+
+    const flowDataObj: ICommonObject = { chatflowid, chatId }
+    const reactFlowNodeData: INodeData = await resolveVariables(
+        appServer.AppDataSource,
+        nodeToExecute.data,
+        reactFlowNodes,
+        '',
+        [],
+        flowDataObj
+    )
+    let nodeToExecuteData = reactFlowNodeData
+
+    const nodeInstanceFilePath = appServer.nodesPool.componentNodes[nodeToExecuteData.name].filePath as string
+    const nodeModule = await import(nodeInstanceFilePath)
+    const nodeInstance = new nodeModule.nodeClass()
+
+    const agent = await nodeInstance.init(nodeToExecuteData, '', {
+        chatflowid,
+        chatId,
+        appDataSource: appServer.AppDataSource,
+        databaseEntities,
+        analytic: chatflow.analytic
+    })
+
+    return agent
+}
+
+const getAgentTools = async (chatflowid: string): Promise<any> => {
+    try {
+        const agent = await buildAndInitTool(chatflowid)
+        const tools = agent.tools
+        return tools.map(convertToOpenAIFunction)
+    } catch (error) {
+        throw new InternalFlowiseError(
+            StatusCodes.INTERNAL_SERVER_ERROR,
+            `Error: openaiRealTimeService.getAgentTools - ${getErrorMessage(error)}`
+        )
+    }
+}
+
+const executeAgentTool = async (chatflowid: string, chatId: string, toolName: string, inputArgs: string): Promise<any> => {
+    try {
+        const agent = await buildAndInitTool(chatflowid, chatId)
+        const tools = agent.tools
+        const tool = tools.find((tool: any) => tool.name === toolName)
+
+        if (!tool) {
+            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Tool ${toolName} not found`)
+        }
+
+        const inputArgsObj = typeof inputArgs === 'string' ? JSON.parse(inputArgs) : inputArgs
+
+        let toolOutput = await tool.call(inputArgsObj, undefined, undefined, { chatId })
+
+        if (typeof toolOutput === 'object') {
+            toolOutput = JSON.stringify(toolOutput)
+        }
+
+        let sourceDocuments = []
+        if (typeof toolOutput === 'string' && toolOutput.includes(SOURCE_DOCUMENTS_PREFIX)) {
+            const _splitted = toolOutput.split(SOURCE_DOCUMENTS_PREFIX)
+            toolOutput = _splitted[0]
+            const _sourceDocuments = JSON.parse(_splitted[1].trim())
+            if (Array.isArray(_sourceDocuments)) {
+                sourceDocuments = _sourceDocuments
+            } else {
+                sourceDocuments.push(_sourceDocuments)
+            }
+        }
+
+        let artifacts = []
+        if (typeof toolOutput === 'string' && toolOutput.includes(ARTIFACTS_PREFIX)) {
+            const _splitted = toolOutput.split(ARTIFACTS_PREFIX)
+            toolOutput = _splitted[0]
+            const _artifacts = JSON.parse(_splitted[1].trim())
+            if (Array.isArray(_artifacts)) {
+                artifacts = _artifacts
+            } else {
+                artifacts.push(_artifacts)
+            }
+        }
+
+        return {
+            output: toolOutput,
+            sourceDocuments,
+            artifacts
+        }
+    } catch (error) {
+        throw new InternalFlowiseError(
+            StatusCodes.INTERNAL_SERVER_ERROR,
+            `Error: openaiRealTimeService.executeAgentTool - ${getErrorMessage(error)}`
+        )
+    }
+}
+
+export default {
+    getAgentTools,
+    executeAgentTool
+}