Merge branch 'feat/workflow-backend' into deploy/dev

langgenius · Mar 13, 2024 · 9b81b2b · 9b81b2b
2 parents 4e4b07c + db299a8
commit 9b81b2b
Show file tree

Hide file tree

Showing 9 changed files with 31 additions and 14 deletions.
diff --git a/api/core/app/apps/advanced_chat/generate_task_pipeline.py b/api/core/app/apps/advanced_chat/generate_task_pipeline.py
@@ -226,6 +226,7 @@ def _process_stream_response(self) -> Generator:
                     'data': {
                         'id': workflow_run.id,
                         'workflow_id': workflow_run.workflow_id,
+                        'sequence_number': workflow_run.sequence_number,
                         'created_at': int(workflow_run.created_at.timestamp())
                     }
                 }

diff --git a/api/core/app/apps/advanced_chat/workflow_event_trigger_callback.py b/api/core/app/apps/advanced_chat/workflow_event_trigger_callback.py
@@ -96,7 +96,9 @@ def on_workflow_node_execute_succeeded(self, node_id: str,
     def on_workflow_node_execute_failed(self, node_id: str,
                                         node_type: NodeType,
                                         node_data: BaseNodeData,
-                                        error: str) -> None:
+                                        error: str,
+                                        inputs: Optional[dict] = None,
+                                        process_data: Optional[dict] = None) -> None:
         """
         Workflow node execute failed
         """
@@ -105,6 +107,8 @@ def on_workflow_node_execute_failed(self, node_id: str,
                 node_id=node_id,
                 node_type=node_type,
                 node_data=node_data,
+                inputs=inputs,
+                process_data=process_data,
                 error=error
             ),
             PublishFrom.APPLICATION_MANAGER

diff --git a/api/core/app/apps/workflow/generate_task_pipeline.py b/api/core/app/apps/workflow/generate_task_pipeline.py
@@ -195,6 +195,7 @@ def _process_stream_response(self) -> Generator:
                     'data': {
                         'id': workflow_run.id,
                         'workflow_id': workflow_run.workflow_id,
+                        'sequence_number': workflow_run.sequence_number,
                         'created_at': int(workflow_run.created_at.timestamp())
                     }
                 }

diff --git a/api/core/app/apps/workflow/workflow_event_trigger_callback.py b/api/core/app/apps/workflow/workflow_event_trigger_callback.py
@@ -96,7 +96,9 @@ def on_workflow_node_execute_succeeded(self, node_id: str,
     def on_workflow_node_execute_failed(self, node_id: str,
                                         node_type: NodeType,
                                         node_data: BaseNodeData,
-                                        error: str) -> None:
+                                        error: str,
+                                        inputs: Optional[dict] = None,
+                                        process_data: Optional[dict] = None) -> None:
         """
         Workflow node execute failed
         """
@@ -105,6 +107,8 @@ def on_workflow_node_execute_failed(self, node_id: str,
                 node_id=node_id,
                 node_type=node_type,
                 node_data=node_data,
+                inputs=inputs,
+                process_data=process_data,
                 error=error
             ),
             PublishFrom.APPLICATION_MANAGER

diff --git a/api/core/app/entities/queue_entities.py b/api/core/app/entities/queue_entities.py
@@ -158,6 +158,9 @@ class QueueNodeFailedEvent(AppQueueEvent):
     node_type: NodeType
     node_data: BaseNodeData
 
+    inputs: Optional[dict] = None
+    process_data: Optional[dict] = None
+
     error: str
 
 

diff --git a/api/core/workflow/callbacks/base_workflow_callback.py b/api/core/workflow/callbacks/base_workflow_callback.py
@@ -55,7 +55,9 @@ def on_workflow_node_execute_succeeded(self, node_id: str,
     def on_workflow_node_execute_failed(self, node_id: str,
                                         node_type: NodeType,
                                         node_data: BaseNodeData,
-                                        error: str) -> None:
+                                        error: str,
+                                        inputs: Optional[dict] = None,
+                                        process_data: Optional[dict] = None) -> None:
         """
         Workflow node execute failed
         """

diff --git a/api/core/workflow/workflow_engine_manager.py b/api/core/workflow/workflow_engine_manager.py
@@ -420,7 +420,9 @@ def _run_workflow_node(self, workflow_run_state: WorkflowRunState,
                         node_id=node.node_id,
                         node_type=node.node_type,
                         node_data=node.node_data,
-                        error=node_run_result.error
+                        error=node_run_result.error,
+                        inputs=node_run_result.inputs,
+                        process_data=node_run_result.process_data,
                     )
 
             raise ValueError(f"Node {node.node_data.title} run failed: {node_run_result.error}")

diff --git a/api/models/workflow.py b/api/models/workflow.py
@@ -123,11 +123,11 @@ def updated_by_account(self):
 
     @property
     def graph_dict(self):
-        return self.graph if not self.graph else json.loads(self.graph)
+        return json.loads(self.graph) if self.graph else None
 
     @property
     def features_dict(self):
-        return self.features if not self.features else json.loads(self.features)
+        return json.loads(self.features) if self.features else None
 
     def user_input_form(self) -> list:
         # get start node from graph
@@ -270,15 +270,15 @@ def created_by_end_user(self):
 
     @property
     def graph_dict(self):
-        return self.graph if not self.graph else json.loads(self.graph)
+        return json.loads(self.graph) if self.graph else None
 
     @property
     def inputs_dict(self):
-        return self.inputs if not self.inputs else json.loads(self.inputs)
+        return json.loads(self.inputs) if self.inputs else None
 
     @property
     def outputs_dict(self):
-        return self.outputs if not self.outputs else json.loads(self.outputs)
+        return json.loads(self.outputs) if self.outputs else None
 
 
 class WorkflowNodeExecutionTriggeredFrom(Enum):
@@ -419,19 +419,19 @@ def created_by_end_user(self):
 
     @property
     def inputs_dict(self):
-        return self.inputs if not self.inputs else json.loads(self.inputs)
+        return json.loads(self.inputs) if self.inputs else None
 
     @property
     def outputs_dict(self):
-        return self.outputs if not self.outputs else json.loads(self.outputs)
+        return json.loads(self.outputs) if self.outputs else None
 
     @property
     def process_data_dict(self):
-        return self.process_data if not self.process_data else json.loads(self.process_data)
+        return json.loads(self.process_data) if self.process_data else None
 
     @property
     def execution_metadata_dict(self):
-        return self.execution_metadata if not self.execution_metadata else json.loads(self.execution_metadata)
+        return json.loads(self.execution_metadata) if self.execution_metadata else None
 
 
 class WorkflowAppLogCreatedFrom(Enum):

diff --git a/api/tests/integration_tests/workflow/nodes/test_llm.py b/api/tests/integration_tests/workflow/nodes/test_llm.py
@@ -36,7 +36,7 @@ def test_execute_llm(setup_openai_mock):
                 'type': 'llm',
                 'model': {
                     'provider': 'openai',
-                    'name': 'gpt-3.5.turbo',
+                    'name': 'gpt-3.5-turbo',
                     'mode': 'chat',
                     'completion_params': {}
                 },