fix workflow api return

api/core/app/apps/advanced_chat/app_generator.py

@@ -16,18 +16,19 @@ from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom
 from core.file.message_file_parser import MessageFileParser
 from core.model_runtime.errors.invoke import InvokeAuthorizationError, InvokeError
-from core.workflow.workflow_engine_manager import WorkflowEngineManager
 from extensions.ext_database import db
 from models.account import Account
 from models.model import App, Conversation, EndUser, Message
+from models.workflow import Workflow
 
 logger = logging.getLogger(__name__)
 
 
 class AdvancedChatAppGenerator(MessageBasedAppGenerator):
     def generate(self, app_model: App,
+                 workflow: Workflow,
                  user: Union[Account, EndUser],
-                 args: Any,
+                 args: dict,
                  invoke_from: InvokeFrom,
                  stream: bool = True) \
             -> Union[dict, Generator]:
@@ -35,6 +36,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         Generate App response.
 
         :param app_model: App
+        :param workflow: Workflow
         :param user: account or end user
         :param args: request args
         :param invoke_from: invoke from source
@@ -59,16 +61,6 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         if args.get('conversation_id'):
             conversation = self._get_conversation_by_user(app_model, args.get('conversation_id'), user)
 
-        # get workflow
-        workflow_engine_manager = WorkflowEngineManager()
-        if invoke_from == InvokeFrom.DEBUGGER:
-            workflow = workflow_engine_manager.get_draft_workflow(app_model=app_model)
-        else:
-            workflow = workflow_engine_manager.get_published_workflow(app_model=app_model)
-
-        if not workflow:
-            raise ValueError('Workflow not initialized')
-
         # parse files
         files = args['files'] if 'files' in args and args['files'] else []
         message_file_parser = MessageFileParser(tenant_id=app_model.tenant_id, app_id=app_model.id)

api/core/app/apps/advanced_chat/app_runner.py

@@ -1,15 +1,20 @@
 import logging
+import time
 from typing import cast
 
 from core.app.app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.base_app_runner import AppRunner
 from core.app.entities.app_invoke_entities import (
-    AdvancedChatAppGenerateEntity,
+    AdvancedChatAppGenerateEntity, InvokeFrom,
 )
+from core.app.entities.queue_entities import QueueStopEvent
 from core.moderation.base import ModerationException
+from core.workflow.entities.node_entities import SystemVariable
+from core.workflow.workflow_engine_manager import WorkflowEngineManager
 from extensions.ext_database import db
-from models.model import App, Conversation, Message
+from models.account import Account
+from models.model import App, Conversation, Message, EndUser
 
 logger = logging.getLogger(__name__)
 
@@ -38,66 +43,151 @@ class AdvancedChatAppRunner(AppRunner):
         if not app_record:
             raise ValueError("App not found")
 
+        workflow = WorkflowEngineManager().get_workflow(app_model=app_record, workflow_id=app_config.workflow_id)
+        if not workflow:
+            raise ValueError("Workflow not initialized")
+
         inputs = application_generate_entity.inputs
         query = application_generate_entity.query
         files = application_generate_entity.files
 
         # moderation
+        if self.handle_input_moderation(
+                queue_manager=queue_manager,
+                app_record=app_record,
+                app_generate_entity=application_generate_entity,
+                inputs=inputs,
+                query=query
+        ):
+            return
+
+        # annotation reply
+        if self.handle_annotation_reply(
+                app_record=app_record,
+                message=message,
+                query=query,
+                queue_manager=queue_manager,
+                app_generate_entity=application_generate_entity
+        ):
+            return
+
+        # fetch user
+        if application_generate_entity.invoke_from in [InvokeFrom.DEBUGGER, InvokeFrom.EXPLORE]:
+            user = db.session.query(Account).filter(Account.id == application_generate_entity.user_id).first()
+        else:
+            user = db.session.query(EndUser).filter(EndUser.id == application_generate_entity.user_id).first()
+
+        # RUN WORKFLOW
+        workflow_engine_manager = WorkflowEngineManager()
+        result_generator = workflow_engine_manager.run_workflow(
+            app_model=app_record,
+            workflow=workflow,
+            user=user,
+            user_inputs=inputs,
+            system_inputs={
+                SystemVariable.QUERY: query,
+                SystemVariable.FILES: files,
+                SystemVariable.CONVERSATION: conversation.id,
+            }
+        )
+
+        for result in result_generator:
+            # todo handle workflow and node event
+            pass
+
+
+    def handle_input_moderation(self, queue_manager: AppQueueManager,
+                                app_record: App,
+                                app_generate_entity: AdvancedChatAppGenerateEntity,
+                                inputs: dict,
+                                query: str) -> bool:
+        """
+        Handle input moderation
+        :param queue_manager: application queue manager
+        :param app_record: app record
+        :param app_generate_entity: application generate entity
+        :param inputs: inputs
+        :param query: query
+        :return:
+        """
         try:
             # process sensitive_word_avoidance
             _, inputs, query = self.moderation_for_inputs(
                 app_id=app_record.id,
-                tenant_id=app_config.tenant_id,
-                app_generate_entity=application_generate_entity,
+                tenant_id=app_generate_entity.app_config.tenant_id,
+                app_generate_entity=app_generate_entity,
                 inputs=inputs,
                 query=query,
             )
         except ModerationException as e:
-            # TODO
-            self.direct_output(
+            self._stream_output(
                 queue_manager=queue_manager,
-                app_generate_entity=application_generate_entity,
-                prompt_messages=prompt_messages,
                 text=str(e),
-                stream=application_generate_entity.stream
+                stream=app_generate_entity.stream,
+                stopped_by=QueueStopEvent.StopBy.INPUT_MODERATION
             )
-            return
+            return True
 
-        if query:
-            # annotation reply
-            annotation_reply = self.query_app_annotations_to_reply(
-                app_record=app_record,
-                message=message,
-                query=query,
-                user_id=application_generate_entity.user_id,
-                invoke_from=application_generate_entity.invoke_from
-            )
+        return False
 
-            if annotation_reply:
-                queue_manager.publish_annotation_reply(
-                    message_annotation_id=annotation_reply.id,
-                    pub_from=PublishFrom.APPLICATION_MANAGER
-                )
-
-                # TODO
-                self.direct_output(
-                    queue_manager=queue_manager,
-                    app_generate_entity=application_generate_entity,
-                    prompt_messages=prompt_messages,
-                    text=annotation_reply.content,
-                    stream=application_generate_entity.stream
-                )
-                return
-
-        # check hosting moderation
-        # TODO
-        hosting_moderation_result = self.check_hosting_moderation(
-            application_generate_entity=application_generate_entity,
-            queue_manager=queue_manager,
-            prompt_messages=prompt_messages
+    def handle_annotation_reply(self, app_record: App,
+                                message: Message,
+                                query: str,
+                                queue_manager: AppQueueManager,
+                                app_generate_entity: AdvancedChatAppGenerateEntity) -> bool:
+        """
+        Handle annotation reply
+        :param app_record: app record
+        :param message: message
+        :param query: query
+        :param queue_manager: application queue manager
+        :param app_generate_entity: application generate entity
+        """
+        # annotation reply
+        annotation_reply = self.query_app_annotations_to_reply(
+            app_record=app_record,
+            message=message,
+            query=query,
+            user_id=app_generate_entity.user_id,
+            invoke_from=app_generate_entity.invoke_from
         )
 
-        if hosting_moderation_result:
-            return
+        if annotation_reply:
+            queue_manager.publish_annotation_reply(
+                message_annotation_id=annotation_reply.id,
+                pub_from=PublishFrom.APPLICATION_MANAGER
+            )
 
-        # todo RUN WORKFLOW
+            self._stream_output(
+                queue_manager=queue_manager,
+                text=annotation_reply.content,
+                stream=app_generate_entity.stream,
+                stopped_by=QueueStopEvent.StopBy.ANNOTATION_REPLY
+            )
+            return True
+
+        return False
+
+    def _stream_output(self, queue_manager: AppQueueManager,
+                       text: str,
+                       stream: bool,
+                       stopped_by: QueueStopEvent.StopBy) -> None:
+        """
+        Direct output
+        :param queue_manager: application queue manager
+        :param text: text
+        :param stream: stream
+        :return:
+        """
+        if stream:
+            index = 0
+            for token in text:
+                queue_manager.publish_text_chunk(token, PublishFrom.APPLICATION_MANAGER)
+                index += 1
+                time.sleep(0.01)
+
+        queue_manager.publish(
+            QueueStopEvent(stopped_by=stopped_by),
+            PublishFrom.APPLICATION_MANAGER
+        )
+        queue_manager.stop_listen()

api/core/app/entities/queue_entities.py

@@ -165,6 +165,7 @@ class QueueStopEvent(AppQueueEvent):
         USER_MANUAL = "user-manual"
         ANNOTATION_REPLY = "annotation-reply"
         OUTPUT_MODERATION = "output-moderation"
+        INPUT_MODERATION = "input-moderation"
 
     event = QueueEvent.STOP
     stopped_by: StopBy