feat: 工作流编排支持并行 #1154 (#1362)

Author: shaohuzhang1

apps/application/flow/i_step_node.py

@@ -28,7 +28,7 @@ def write_context(step_variable: Dict, global_variable: Dict, node, workflow):
     if step_variable is not None:
         for key in step_variable:
             node.context[key] = step_variable[key]
-        if workflow.is_result() and 'answer' in step_variable:
+        if workflow.is_result(node, NodeResult(step_variable, global_variable)) and 'answer' in step_variable:
             answer = step_variable['answer']
             yield answer
             workflow.answer += answer
@@ -166,6 +166,7 @@ def get_flow_params_serializer_class(self) -> Type[serializers.Serializer]:
     def get_write_error_context(self, e):
         self.status = 500
         self.err_message = str(e)
+        self.context['run_time'] = time.time() - self.context['start_time']
 
         def write_error_context(answer, status=200):
             pass

apps/application/flow/step_node/ai_chat_step_node/impl/base_chat_node.py

@@ -31,7 +31,7 @@ def _write_context(node_variable: Dict, workflow_variable: Dict, node: INode, wo
     node.context['history_message'] = node_variable['history_message']
     node.context['question'] = node_variable['question']
     node.context['run_time'] = time.time() - node.context['start_time']
-    if workflow.is_result():
+    if workflow.is_result(node, NodeResult(node_variable, workflow_variable)):
         workflow.answer += answer
 
 

apps/application/flow/step_node/function_lib_node/impl/base_function_lib_node.py

@@ -26,7 +26,7 @@ def write_context(step_variable: Dict, global_variable: Dict, node, workflow):
     if step_variable is not None:
         for key in step_variable:
             node.context[key] = step_variable[key]
-        if workflow.is_result() and 'result' in step_variable:
+        if workflow.is_result(node, NodeResult(step_variable, global_variable)) and 'result' in step_variable:
             result = str(step_variable['result']) + '\n'
             yield result
             workflow.answer += result