From 55e8e500d7fd88969ac0d5a1b0791b7ce5909600 Mon Sep 17 00:00:00 2001 From: shaohuzhang1 Date: Mon, 6 Jul 2026 16:07:30 +0800 Subject: [PATCH] feat: Migration break, continue nodes --- apps/application/workflow/i_node.py | 13 +++++- .../nodes/loop_break_node/__init__.py | 9 ++++ .../nodes/loop_break_node/loop_break_node.py | 45 +++++++++++++++++++ .../nodes/loop_continue_node/__init__.py | 9 ++++ .../loop_continue_node/loop_continue_node.py | 45 +++++++++++++++++++ .../workflow/nodes/loop_node/loop_node.py | 12 ++++- apps/application/workflow/workflow_manage.py | 1 + 7 files changed, 132 insertions(+), 2 deletions(-) create mode 100644 apps/application/workflow/nodes/loop_break_node/__init__.py create mode 100644 apps/application/workflow/nodes/loop_break_node/loop_break_node.py create mode 100644 apps/application/workflow/nodes/loop_continue_node/__init__.py create mode 100644 apps/application/workflow/nodes/loop_continue_node/loop_continue_node.py diff --git a/apps/application/workflow/i_node.py b/apps/application/workflow/i_node.py index 93b649c07b7..2f40dcdbabe 100644 --- a/apps/application/workflow/i_node.py +++ b/apps/application/workflow/i_node.py @@ -7,6 +7,7 @@ @desc: """ import time +from enum import Enum from typing import Optional, Type, Callable from rest_framework import serializers @@ -17,6 +18,12 @@ from common.utils.logger import maxkb_logger +class Signal(str, Enum): + BREAK = 'BREAK' + CONTINUE = 'CONTINUE' + FORM = 'FORM' + + class INode: # 当前节点支持的工作流类型 supported_workflow_type_list = [] @@ -92,13 +99,14 @@ def _run(self): self.execute() self.complete(Status.SUCCESS) - def complete(self, status, anchors=None, error=None): + def complete(self, status, anchors=None, error=None, signal: Optional[Signal] = None): """ 节点结束调用函数 @param status: 状态 @param anchors 锚点信息 @param error: 错误信息 + @param signal: 信号 @return: """ if self._completed: @@ -108,6 +116,9 @@ def complete(self, status, anchors=None, error=None): if error: self.data['error'] = str(error) self.data['run_time'] = time.time() - self.data['start_time'] + if signal: + self.workflow_manage.signal = signal + anchors = [] if anchors is None: anchors = [self.success_anchor() if status == Status.SUCCESS else self.fail_anchor()] self._dispatch(anchors) diff --git a/apps/application/workflow/nodes/loop_break_node/__init__.py b/apps/application/workflow/nodes/loop_break_node/__init__.py new file mode 100644 index 00000000000..8bc0177353d --- /dev/null +++ b/apps/application/workflow/nodes/loop_break_node/__init__.py @@ -0,0 +1,9 @@ +# coding=utf-8 +""" + @project: MaxKB + @Author：虎虎虎 + @file： __init__.py + @date：2026/7/6 15:00 + @desc: +""" +from .loop_break_node import LoopBreakNode diff --git a/apps/application/workflow/nodes/loop_break_node/loop_break_node.py b/apps/application/workflow/nodes/loop_break_node/loop_break_node.py new file mode 100644 index 00000000000..ae67636c97d --- /dev/null +++ b/apps/application/workflow/nodes/loop_break_node/loop_break_node.py @@ -0,0 +1,45 @@ +# coding=utf-8 +""" + @project: MaxKB + @Author：虎虎虎 + @file： loop_break_node.py + @date：2026/7/6 15:00 + @desc: +""" +from django.utils.translation import gettext_lazy as _ +from rest_framework import serializers + +from application.workflow.compare import do_assertion +from application.workflow.common import WorkflowType +from application.workflow.i_node import INode, Signal +from application.workflow.status import Status + + +class ConditionSerializer(serializers.Serializer): + compare = serializers.CharField(required=True, label=_("Comparator")) + value = serializers.CharField(required=False, allow_null=True, allow_blank=True, label=_("value")) + field = serializers.ListField(required=True, label=_("Fields")) + + +class LoopBreakNodeSerializer(serializers.Serializer): + condition = serializers.CharField(required=True, label=_("Condition or|and")) + condition_list = ConditionSerializer(many=True) + + +class LoopBreakNode(INode): + serializer_class = LoopBreakNodeSerializer + supported_workflow_type_list = [WorkflowType.APPLICATION, WorkflowType.KNOWLEDGE, WorkflowType.TOOL] + type = 'loop-break-node' + + def execute(self): + node_params = self.get_parameters() + condition = node_params.get('condition') + condition_list = node_params.get('condition_list', []) + + is_break = do_assertion(self.workflow_manage, condition, condition_list) + self.write_context('is_break', is_break) + + if is_break: + self.complete(Status.SUCCESS, signal=Signal.BREAK) + return + self.complete(Status.SUCCESS) diff --git a/apps/application/workflow/nodes/loop_continue_node/__init__.py b/apps/application/workflow/nodes/loop_continue_node/__init__.py new file mode 100644 index 00000000000..a7733ae87d4 --- /dev/null +++ b/apps/application/workflow/nodes/loop_continue_node/__init__.py @@ -0,0 +1,9 @@ +# coding=utf-8 +""" + @project: MaxKB + @Author：虎虎虎 + @file： __init__.py + @date：2026/7/6 15:10 + @desc: +""" +from .loop_continue_node import LoopContinueNode diff --git a/apps/application/workflow/nodes/loop_continue_node/loop_continue_node.py b/apps/application/workflow/nodes/loop_continue_node/loop_continue_node.py new file mode 100644 index 00000000000..ac589deec88 --- /dev/null +++ b/apps/application/workflow/nodes/loop_continue_node/loop_continue_node.py @@ -0,0 +1,45 @@ +# coding=utf-8 +""" + @project: MaxKB + @Author：虎虎虎 + @file： loop_continue_node.py + @date：2026/7/6 15:10 + @desc: +""" +from django.utils.translation import gettext_lazy as _ +from rest_framework import serializers + +from application.workflow.compare import do_assertion +from application.workflow.common import WorkflowType +from application.workflow.i_node import INode, Signal +from application.workflow.status import Status + + +class ConditionSerializer(serializers.Serializer): + compare = serializers.CharField(required=True, label=_("Comparator")) + value = serializers.CharField(required=False, allow_null=True, allow_blank=True, label=_("value")) + field = serializers.ListField(required=True, label=_("Fields")) + + +class LoopContinueNodeSerializer(serializers.Serializer): + condition = serializers.CharField(required=True, label=_("Condition or|and")) + condition_list = ConditionSerializer(many=True) + + +class LoopContinueNode(INode): + serializer_class = LoopContinueNodeSerializer + supported_workflow_type_list = [WorkflowType.APPLICATION, WorkflowType.KNOWLEDGE, WorkflowType.TOOL] + type = 'loop-continue-node' + + def execute(self): + node_params = self.get_parameters() + condition = node_params.get('condition') + condition_list = node_params.get('condition_list', []) + + is_continue = do_assertion(self.workflow_manage, condition, condition_list) + self.write_context('is_continue', is_continue) + + if is_continue: + self.complete(Status.SUCCESS, signal=Signal.CONTINUE) + return + self.complete(Status.SUCCESS) diff --git a/apps/application/workflow/nodes/loop_node/loop_node.py b/apps/application/workflow/nodes/loop_node/loop_node.py index d2b7accea0c..8aeb3f158b1 100644 --- a/apps/application/workflow/nodes/loop_node/loop_node.py +++ b/apps/application/workflow/nodes/loop_node/loop_node.py @@ -12,7 +12,7 @@ from rest_framework import serializers from application.workflow.common import WorkflowType, new_instance -from application.workflow.i_node import INode +from application.workflow.i_node import INode, Signal from application.workflow.message.struct.content import NodeInfo from application.workflow.message.struct.text_content import TextContent @@ -121,6 +121,16 @@ def on_complete(wf_manage, error): self.write_context('index', index) self.write_context('item', item) + if wf_manage.signal == Signal.BREAK: + self.write_context('answer', self._answer_text) + self.write_context('run_time', time.time() - self.data.get('start_time', time.time())) + self.complete(Status.SUCCESS) + return + + if wf_manage.signal == Signal.CONTINUE: + self._run_next() + return + if error: self.write_context('error_message', str(error)) return diff --git a/apps/application/workflow/workflow_manage.py b/apps/application/workflow/workflow_manage.py index b83fd11a82c..cd978e541e0 100644 --- a/apps/application/workflow/workflow_manage.py +++ b/apps/application/workflow/workflow_manage.py @@ -55,6 +55,7 @@ def __init__(self, self.parameters = parameters self.context = {} self.nodes = [] + self.signal = None self.start_node = get_start_node(workflow, self) def run(self):