forked from FlowiseAI/Flowise
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Feature/add endpoints for realtime api (FlowiseAI#3318)
add endpoints for realtime api
- Loading branch information
1 parent
10bfba7
commit 14b7148
Showing
5 changed files
with
304 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| } |