feat: Refactoring workflow executor (#6328)

Author: shaohuzhang1

apps/application/workflow/i_node.py

@@ -14,6 +14,7 @@
 from application.workflow.common import Node
 from application.workflow.message.struct.content import Content
 from application.workflow.status import Status
+from common.utils.logger import maxkb_logger
 
 
 class INode:
@@ -36,22 +37,37 @@ def __init__(self, node, workflow_manage, get_node_parameters: Callable[[Node],
         self.parameters = get_node_parameters(node)
         # 节点运行时产生的数据
         self.data = {}
+        self._completed = False
+        # ---- 锚点构造,全项目唯一的拼接点 ----
 
-    def next_success_nodes(self):
-        edge_node_list = self.workflow_manage.workflow.get_next_edge_nodes(self.node.id)
-        if edge_node_list is None:
-            self.workflow_manage.next_nodes([])
-            return
-        self.workflow_manage.next_nodes(
-            [en.node for en in edge_node_list if en.edge.sourceAnchorId == self.node.id + '_right'])
+    def anchor(self, *parts):
+        """
+        通用锚点: anchor('right') → '{id}_right',
+                anchor(branch_id, 'right') → '{id}_{branch_id}_right'
+        """
+        return '_'.join([self.node.id, *map(str, parts)])
 
-    def next_fail_nodes(self):
-        edge_node_list = self.workflow_manage.workflow.get_next_edge_nodes(self.node.id)
-        if edge_node_list is None:
-            self.workflow_manage.next_nodes([])
-            return
-        self.workflow_manage.next_nodes(
-            [en.node for en in edge_node_list if en.edge.sourceAnchorId == self.node.id + '_exception_right'])
+    def success_anchor(self):
+        """
+        成功锚点
+        @return: 成功锚点
+        """
+        return self.anchor('right')
+
+    def fail_anchor(self):
+        """
+        失败锚点
+        @return: 失败锚点
+        """
+        return self.branch_anchor('exception')
+
+    def branch_anchor(self, branch_id):
+        """
+        自定义锚点
+        @param branch_id: 自定义分支id
+        @return: 自定义锚点
+        """
+        return self.anchor(branch_id, 'right')
 
     def execute(self):
         pass
@@ -61,18 +77,55 @@ def run(self):
         运行节点
         @return: 不响应数据
         """
-        start_time = time.time()
+        self.data['start_time'] = time.time()
         self.status = Status.RUNNING
-        self.data['start_time'] = start_time
-        self._run()
-        self.data['run_time'] = time.time() - start_time
+        try:
+            self._run()
+        except Exception as e:
+            self.complete(Status.FAIL, error=e)
 
     def _run(self):
         """
         执行节点
         @return:
         """
-        return self.execute()
+        self.execute()
+        self.complete(Status.SUCCESS)
+
+    def complete(self, status, anchors=None, error=None):
+        """
+        节点结束调用函数
+
+        @param status: 状态
+        @param anchors 锚点信息
+        @param error:  错误信息
+        @return:
+        """
+        if self._completed:
+            return
+        self._completed = True
+        self.status = status
+        if error:
+            self.data['error'] = str(error)
+        self.data['run_time'] = time.time() - self.data['start_time']
+        if anchors is None:
+            anchors = [self.success_anchor() if status == Status.SUCCESS else self.fail_anchor()]
+        self._dispatch(anchors)
+        self.workflow_manage.assertion_end()
+
+    def _dispatch(self, anchors):
+        """
+        根据锚点执行下一个节点
+        @param anchors: 锚点列表
+        @return:不返回
+        """
+        edge_node_list = self.workflow_manage.workflow.get_next_edge_nodes(self.node.id) or []
+        known = {en.edge.sourceAnchorId for en in edge_node_list}
+        unknown = set(anchors) - known
+        if unknown and known:
+            maxkb_logger.warning(f'node {self.node.id}: anchors {unknown} matched no edges, known={known}')
+        self.workflow_manage.next_nodes(
+            [en.node for en in edge_node_list if en.edge.sourceAnchorId in anchors])
 
     def get_node_id(self):
         """

apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py

@@ -262,8 +262,6 @@ def execute(self):
             answer = self.get_context('answer')
             node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.SUCCESS)
             self.write(TextContent(text_content_id, answer, Status.SUCCESS, node_info))
-        self.status = Status.SUCCESS
-        self.next_success_nodes()
 
     def _generate_prompt_question(self, prompt, model, vision, image_list, video_list):
         images = []

apps/application/workflow/nodes/start_node/start_node.py

@@ -107,5 +107,3 @@ def execute(self):
                 key = field.get('value')
                 if key:
                     self.workflow_manage.context[key] = workflow_variable.get(key, '')
-        self.status = Status.SUCCESS
-        self.next_success_nodes()

apps/application/workflow/workflow_manage.py

@@ -47,6 +47,7 @@ def __init__(self,
         @param workflow_type: 工作流类型
         @param parameters:    工作流使用到的其他数据
         """
+        self._lock = threading.Lock()
         self.done = False
         self.call_back = call_back
         self.workflow = workflow
@@ -65,10 +66,7 @@ def run(self):
 
     def _run(self, node):
         self.nodes.append(node)
-        try:
-            node.run()
-        except Exception as e:
-            node.status = Status.FAIL
+        node.run()
 
     def next_nodes(self, nodes: Optional[List[Node]]):
         """
@@ -77,18 +75,26 @@ def next_nodes(self, nodes: Optional[List[Node]]):
         @return:
         """
         if nodes is None or len(nodes) == 0:
-            self.assertion_end()
             return
-        for node in nodes:
-            self._run_async(get_node_class(node.type, self.workflow_type)(node, self, get_node_parameters))
+        with self._lock:
+            instances = [get_node_class(n.type, self.workflow_type)(n, self, get_node_parameters)
+                         for n in nodes]
+            self.nodes.extend(instances)
+        for inst in instances:
+            self._run_async(inst)
 
     def assertion_end(self):
         """
         如果节点执行结束没有下一个节点  就结束工作流
         @return:
         """
-        if self.is_end():
-            self.end()
+        with self._lock:
+            if self.done:
+                return
+            if not self.is_end():
+                return
+            self.done = True
+        self.end()
 
     def is_end(self):
         """