feat: knowledge workflow

Author: shaohuzhang1

apps/application/flow/common.py

@@ -6,7 +6,7 @@
     @date：2024/12/11 17:57
     @desc:
 """
-
+from enum import Enum
 from typing import List, Dict
 
 from django.db.models import QuerySet
@@ -90,6 +90,16 @@ def __init__(self, edge, node):
         self.node = node
 
 
+class WorkflowMode(Enum):
+    APPLICATION = "application"
+
+    APPLICATION_LOOP = "application-loop"
+
+    KNOWLEDGE = "knowledge"
+
+    KNOWLEDGE_LOOP = "knowledge-loop"
+
+
 class Workflow:
     """
     节点列表
@@ -112,7 +122,10 @@ class Workflow:
     """
     next_node_map: Dict[str, List[EdgeNode]]
 
-    def __init__(self, nodes: List[Node], edges: List[Edge]):
+    workflow_mode: WorkflowMode
+
+    def __init__(self, nodes: List[Node], edges: List[Edge],
+                 workflow_mode: WorkflowMode = WorkflowMode.APPLICATION.value):
         self.nodes = nodes
         self.edges = edges
         self.node_map = {node.id: node for node in nodes}
@@ -125,6 +138,7 @@ def __init__(self, nodes: List[Node], edges: List[Edge]):
         self.next_node_map = {key: [EdgeNode(edge, self.node_map.get(edge.targetNodeId)) for edge in edges] for
                               key, edges in
                               group_by(edges, key=lambda edge: edge.sourceNodeId).items()}
+        self.workflow_mode = workflow_mode
 
     def get_node(self, node_id):
         """
@@ -167,13 +181,13 @@ def get_next_nodes(self, node_id) -> List[Node]:
         return [en.node for en in self.next_node_map.get(node_id, [])]
 
     @staticmethod
-    def new_instance(flow_obj: Dict):
+    def new_instance(flow_obj: Dict, workflow_mode: WorkflowMode = WorkflowMode.APPLICATION.value):
         nodes = flow_obj.get('nodes')
         edges = flow_obj.get('edges')
         nodes = [Node(node.get('id'), node.get('type'), **node)
                  for node in nodes]
         edges = [Edge(edge.get('id'), edge.get('type'), **edge) for edge in edges]
-        return Workflow(nodes, edges)
+        return Workflow(nodes, edges, workflow_mode)
 
     def get_start_node(self):
         return self.get_node('start-node')
@@ -190,10 +204,9 @@ def is_valid(self):
         self.is_valid_base_node()
         self.is_valid_work_flow()
 
-    @staticmethod
-    def is_valid_node_params(node: Node):
+    def is_valid_node_params(self, node: Node):
         from application.flow.step_node import get_node
-        get_node(node.type)(node, None, None)
+        get_node(node.type, self.workflow_mode)(node, None, None)
 
     def is_valid_node(self, node: Node):
         self.is_valid_node_params(node)

apps/application/flow/i_step_node.py

@@ -96,6 +96,14 @@ def handler(self, workflow):
                 application_public_access_client.save()
 
 
+class KnowledgeWorkflowPostHandler(WorkFlowPostHandler):
+    def __init__(self, chat_info):
+        super().__init__(chat_info)
+
+    def handler(self, workflow):
+        pass
+
+
 class NodeResult:
     def __init__(self, node_variable: Dict, workflow_variable: Dict,
                  _write_context=write_context, _is_interrupt=is_interrupt):
@@ -152,6 +160,12 @@ class FlowParamsSerializer(serializers.Serializer):
     debug = serializers.BooleanField(required=True, label="是否debug")
 
 
+class KnowledgeFlowParamsSerializer(serializers.Serializer):
+    knowledge_id = serializers.CharField(required=True, label="知识库id")
+    data_source = serializers.DictField(required=True, label="数据源")
+    knowledge_base = serializers.DictField(required=False, label="知识库设置")
+
+
 class INode:
     view_type = 'many_view'
 
@@ -221,7 +235,7 @@ def get_node_params_serializer_class(self) -> Type[serializers.Serializer]:
         pass
 
     def get_flow_params_serializer_class(self) -> Type[serializers.Serializer]:
-        return FlowParamsSerializer
+        return self.workflow_manage.get_params_serializer_class()
 
     def get_write_error_context(self, e):
         self.status = 500

apps/application/flow/knowledge_workflow_manage.py

@@ -0,0 +1,35 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎虎
+    @file： Knowledge_workflow_manage.py
+    @date：2025/11/13 19:02
+    @desc:
+"""
+
+from application.flow.common import Workflow
+from application.flow.i_step_node import WorkFlowPostHandler, KnowledgeFlowParamsSerializer
+from application.flow.workflow_manage import WorkflowManage
+from common.handle.base_to_response import BaseToResponse
+from common.handle.impl.response.system_to_response import SystemToResponse
+
+
+class KnowledgeWorkflowManage(WorkflowManage):
+
+    def __init__(self, flow: Workflow,
+                 params,
+                 work_flow_post_handler: WorkFlowPostHandler,
+                 base_to_response: BaseToResponse = SystemToResponse(),
+                 start_node_id=None,
+                 start_node_data=None, chat_record=None, child_node=None):
+        super().__init__(flow, params, work_flow_post_handler, base_to_response, None, None, None,
+                         None,
+                         None, None, start_node_id, start_node_data, chat_record, child_node)
+
+    def get_params_serializer_class(self):
+        return KnowledgeFlowParamsSerializer
+
+    def get_start_node(self):
+        start_node_list = [node for node in self.flow.nodes if
+                           self.params.get('data_source', {}).get('node_id') == node.id]
+        return start_node_list[0]

apps/application/flow/loop_workflow_manage.py

@@ -105,10 +105,10 @@ def get_node_cls_by_id(self, node_id, up_node_id_list=None,
                            get_node_params=lambda node: node.properties.get('node_data')):
         for node in self.flow.nodes:
             if node.id == node_id:
-                node_instance = get_node(node.type)(node,
-                                                    self.params, self, up_node_id_list,
-                                                    get_node_params,
-                                                    salt=self.get_index())
+                node_instance = get_node(node.type, self.flow.workflow_mode)(node,
+                                                                             self.params, self, up_node_id_list,
+                                                                             get_node_params,
+                                                                             salt=self.get_index())
                 return node_instance
         return None
 

apps/application/flow/step_node/__init__.py

@@ -50,11 +50,14 @@
              BaseIntentNode, BaseLoopNode, BaseLoopStartStepNode,
              BaseLoopContinueNode,
              BaseLoopBreakNode, BaseVariableSplittingNode, BaseParameterExtractionNode, BaseVariableAggregationNode,
+<<<<<<< Updated upstream
              BaseDataSourceLocalNode,BaseDataSourceWebNode,BaseKnowledgeWriteNode]
+=======
+             BaseDataSourceLocalNode, BaseDataSourceWebNode]
 
+node_map = {n.type: {w: n for w in n.support} for n in node_list}
+>>>>>>> Stashed changes
 
-def get_node(node_type):
-    find_list = [node for node in node_list if node.type == node_type]
-    if len(find_list) > 0:
-        return find_list[0]
-    return None
+
+def get_node(node_type, workflow_model):
+    return node_map.get(node_type).get(workflow_model)

apps/application/flow/step_node/ai_chat_step_node/i_chat_node.py

@@ -11,6 +11,7 @@
 from django.utils.translation import gettext_lazy as _
 from rest_framework import serializers
 
+from application.flow.common import WorkflowMode
 from application.flow.i_step_node import INode, NodeResult
 
 
@@ -34,22 +35,31 @@ class ChatNodeSerializer(serializers.Serializer):
     mcp_enable = serializers.BooleanField(required=False, label=_("Whether to enable MCP"))
     mcp_servers = serializers.JSONField(required=False, label=_("MCP Server"))
     mcp_tool_id = serializers.CharField(required=False, allow_blank=True, allow_null=True, label=_("MCP Tool ID"))
-    mcp_tool_ids = serializers.ListField(child=serializers.UUIDField(), required=False, allow_empty=True, label=_("MCP Tool IDs"), )
+    mcp_tool_ids = serializers.ListField(child=serializers.UUIDField(), required=False, allow_empty=True,
+                                         label=_("MCP Tool IDs"), )
     mcp_source = serializers.CharField(required=False, allow_blank=True, allow_null=True, label=_("MCP Source"))
 
     tool_enable = serializers.BooleanField(required=False, default=False, label=_("Whether to enable tools"))
     tool_ids = serializers.ListField(child=serializers.UUIDField(), required=False, allow_empty=True,
                                      label=_("Tool IDs"), )
     mcp_output_enable = serializers.BooleanField(required=False, default=True, label=_("Whether to enable MCP output"))
 
+
 class IChatNode(INode):
     type = 'ai-chat-node'
+    support = [WorkflowMode.APPLICATION, WorkflowMode.APPLICATION_LOOP, WorkflowMode.KNOWLEDGE_LOOP,
+               WorkflowMode.KNOWLEDGE]
 
     def get_node_params_serializer_class(self) -> Type[serializers.Serializer]:
         return ChatNodeSerializer
 
     def _run(self):
-        return self.execute(**self.node_params_serializer.data, **self.flow_params_serializer.data)
+        if [WorkflowMode.KNOWLEDGE, WorkflowMode.APPLICATION_LOOP].__contains__(
+                self.workflow_manage.flow.workflow_mode):
+            return self.execute(**self.node_params_serializer.data, **self.flow_params_serializer.data,
+                                **{'history_chat_record': [], 'stream': True, 'chat_id': None, 'chat_record_id': None})
+        else:
+            return self.execute(**self.node_params_serializer.data, **self.flow_params_serializer.data)
 
     def execute(self, model_id, system, prompt, dialogue_number, history_chat_record, stream, chat_id,
                 chat_record_id,

apps/application/flow/step_node/application_node/i_application_node.py

@@ -3,6 +3,7 @@
 
 from rest_framework import serializers
 
+from application.flow.common import WorkflowMode
 from application.flow.i_step_node import INode, NodeResult
 
 from django.utils.translation import gettext_lazy as _
@@ -25,6 +26,7 @@ class ApplicationNodeSerializer(serializers.Serializer):
 
 class IApplicationNode(INode):
     type = 'application-node'
+    support = [WorkflowMode.APPLICATION, WorkflowMode.APPLICATION_LOOP]
 
     def get_node_params_serializer_class(self) -> Type[serializers.Serializer]:
         return ApplicationNodeSerializer

apps/application/flow/step_node/condition_node/i_condition_node.py

@@ -11,6 +11,7 @@
 from django.utils.translation import gettext_lazy as _
 from rest_framework import serializers
 
+from application.flow.common import WorkflowMode
 from application.flow.i_step_node import INode
 
 
@@ -36,3 +37,5 @@ def get_node_params_serializer_class(self) -> Type[serializers.Serializer]:
         return ConditionNodeParamsSerializer
 
     type = 'condition-node'
+
+    support = [WorkflowMode.APPLICATION_LOOP]

apps/application/flow/step_node/data_source_local_node/i_data_source_local_node.py

@@ -12,6 +12,7 @@
 from django.utils.translation import gettext_lazy as _
 from rest_framework import serializers
 
+from application.flow.common import WorkflowMode
 from application.flow.i_step_node import INode, NodeResult
 
 
@@ -35,5 +36,7 @@ def get_node_params_serializer_class(self) -> Type[serializers.Serializer]:
     def _run(self):
         return self.execute(**self.node_params_serializer.data, **self.flow_params_serializer.data)
 
-    def execute(self, file_format, max_file_number, file_max_size, **kwargs) -> NodeResult:
+    def execute(self, file_type_list, file_size_limit, file_count_limit, **kwargs) -> NodeResult:
         pass
+
+    support = [WorkflowMode.KNOWLEDGE]

apps/application/flow/step_node/data_source_local_node/impl/base_data_source_local_node.py

@@ -34,5 +34,6 @@ def get_form_list(node):
             'label': '',
         }]
 
-    def execute(self, file_format, max_file_number, file_max_size, **kwargs) -> NodeResult:
-        pass
+    def execute(self, file_type_list, file_size_limit, file_count_limit, **kwargs) -> NodeResult:
+        return NodeResult({'file_list': self.workflow_manage.params.get('data_source', {}).get('file_list')},
+                          self.workflow_manage.params.get('knowledge_base') or {})