feat: The knowledge base workflow uses a new workflow engine

apps/application/flow/tools.py

@@ -16,7 +16,6 @@
 import shutil
 import threading
 import zipfile
-from functools import reduce
 from typing import Iterator
 
 # ---------------------------------------------------------------------------
@@ -50,7 +49,7 @@
 from langgraph.checkpoint.memory import MemorySaver
 from maxkb.const import CONFIG
 from pydantic import Field, create_model
-from tools.models import Tool, ToolRecord, ToolScope, ToolType, ToolWorkflowVersion
+from tools.models import Tool, ToolRecord, ToolType, ToolWorkflowVersion
 
 from application.flow.backend.sandbox_shell import SandboxShellBackend
 from application.flow.common import Workflow, WorkflowMode
@@ -817,194 +816,6 @@ async def anext_async(agen):
     return await agen.__anext__()
 
 
-target_source_node_mapping = {
-    "TOOL": {
-        "tool-lib-node": lambda n: [n.get("properties").get("node_data").get("tool_lib_id")],
-        "ai-chat-node": lambda n: [
-            *(n.get("properties").get("node_data").get("mcp_tool_ids") or []),
-            *(n.get("properties").get("node_data").get("tool_ids") or []),
-            *(n.get("properties").get("node_data").get("skill_tool_ids") or []),
-        ],
-        "mcp-node": lambda n: [n.get("properties").get("node_data").get("mcp_tool_id")],
-        "tool-workflow-lib-node": lambda n: [n.get("properties").get("node_data").get("tool_lib_id")],
-    },
-    "MODEL": {
-        "ai-chat-node": lambda n: [n.get("properties").get("node_data").get("model_id")],
-        "question-node": lambda n: [n.get("properties").get("node_data").get("model_id")],
-        "speech-to-text-node": lambda n: [n.get("properties").get("node_data").get("stt_model_id")],
-        "text-to-speech-node": lambda n: [n.get("properties").get("node_data").get("tts_model_id")],
-        "image-to-video-node": lambda n: [n.get("properties").get("node_data").get("model_id")],
-        "image-generate-node": lambda n: [n.get("properties").get("node_data").get("model_id")],
-        "intent-node": lambda n: [n.get("properties").get("node_data").get("model_id")],
-        "image-understand-node": lambda n: [n.get("properties").get("node_data").get("model_id")],
-        "parameter-extraction-node": lambda n: [n.get("properties").get("node_data").get("model_id")],
-        "video-understand-node": lambda n: [n.get("properties").get("node_data").get("model_id")],
-        "reranker-node": lambda n: [n.get("properties").get("node_data").get("reranker_model_id")],
-    },
-    "KNOWLEDGE": {
-        "search-knowledge-node": lambda n: n.get("properties").get("node_data").get("knowledge_id_list"),
-        "search-document-node": lambda n: n.get("properties").get("node_data").get("knowledge_id_list"),
-    },
-    "APPLICATION": {
-        "application-node": lambda n: [n.get("properties").get("node_data").get("application_id")],
-        "ai-chat-node": lambda n: [*(n.get("properties").get("node_data").get("application_ids") or [])],
-    },
-}
-
-
-def get_node_handle_callback(source_type, source_id):
-    def node_handle_callback(node):
-        from system_manage.models.resource_mapping import ResourceMapping
-
-        response = []
-        for key, value in target_source_node_mapping.items():
-            if node.get("type") in value:
-                call = value.get(node.get("type"))
-                target_source_id_list = call(node)
-                for target_source_id in target_source_id_list:
-                    if target_source_id:
-                        response.append(
-                            ResourceMapping(
-                                source_type=source_type,
-                                target_type=key,
-                                source_id=source_id,
-                                target_id=target_source_id,
-                            )
-                        )
-        return response
-
-    return node_handle_callback
-
-
-def get_workflow_resource(workflow, node_handle):
-    response = []
-    if "nodes" in workflow:
-        for node in workflow.get("nodes"):
-            rs = node_handle(node)
-            if rs:
-                for r in rs:
-                    response.append(r)
-            if node.get("type") == "loop-node":
-                r = get_workflow_resource(node.get("properties", {}).get("node_data", {}).get("loop_body"), node_handle)
-                for rn in r:
-                    response.append(rn)
-        return list({(str(item.target_type) + str(item.target_id)): item for item in response}.values())
-    return []
-
-
-application_instance_field_call_dict = {
-    "TOOL": [
-        lambda instance: instance.mcp_tool_ids or [],
-        lambda instance: instance.skill_tool_ids or [],
-        lambda instance: instance.tool_ids or [],
-    ],
-    "APPLICATION": [
-        lambda instance: instance.application_ids or [],
-    ],
-    "MODEL": [
-        lambda instance: [instance.model_id] if instance.model_id else [],
-        lambda instance: [instance.long_term_model_id] if instance.long_term_model_id else [],
-        lambda instance: [instance.tts_model_id] if instance.tts_model_id else [],
-        lambda instance: [instance.stt_model_id] if instance.stt_model_id else [],
-    ],
-}
-knowledge_instance_field_call_dict = {
-    "MODEL": [lambda instance: [instance.embedding_model_id] if instance.embedding_model_id else []],
-}
-
-
-def get_instance_resource(instance, source_type, source_id, instance_field_call_dict):
-    response = []
-    from system_manage.models.resource_mapping import ResourceMapping
-
-    for target_type, call_list in instance_field_call_dict.items():
-        target_id_list = reduce(lambda x, y: [*x, *y], [call(instance) for call in call_list], [])
-        if target_id_list:
-            for target_id in target_id_list:
-                response.append(
-                    ResourceMapping(
-                        source_type=source_type, target_type=target_type, source_id=source_id, target_id=target_id
-                    )
-                )
-    return response
-
-
-def save_workflow_mapping(workflow, source_type, source_id, other_resource_mapping=None):
-    if not other_resource_mapping:
-        other_resource_mapping = []
-    from django.db.models import QuerySet
-    from system_manage.models.resource_mapping import ResourceMapping
-
-    QuerySet(ResourceMapping).filter(source_type=source_type, source_id=source_id).delete()
-    resource_mapping_list = get_workflow_resource(workflow, get_node_handle_callback(source_type, source_id))
-    resource_mapping_list += other_resource_mapping
-    if resource_mapping_list:
-        QuerySet(ResourceMapping).bulk_create(
-            {(str(item.target_type) + str(item.target_id)): item for item in resource_mapping_list}.values()
-        )
-
-
-def get_tool_id_list(workflow, with_deep=False):
-    from tools.models import ToolType, ToolWorkflow
-
-    _result = []
-    for node in workflow.get("nodes", []):
-        if node.get("type") == "tool-lib-node":
-            tool_id = node.get("properties", {}).get("node_data", {}).get("tool_lib_id")
-            if tool_id:
-                _result.append(tool_id)
-        elif node.get("type") == "loop-node":
-            r = get_tool_id_list(node.get("properties", {}).get("node_data", {}).get("loop_body", {}))
-            for item in r:
-                _result.append(item)
-        elif node.get("type") == "tool-workflow-lib-node":
-            tool_id = node.get("properties", {}).get("node_data", {}).get("tool_lib_id")
-            if tool_id:
-                _result.append(tool_id)
-        elif node.get("type") == "ai-chat-node":
-            node_data = node.get("properties", {}).get("node_data", {})
-            mcp_tool_ids = node_data.get("mcp_tool_ids") or []
-            skill_tool_ids = node_data.get("skill_tool_ids") or []
-            tool_ids = node_data.get("tool_ids") or []
-            for _id in mcp_tool_ids + tool_ids + skill_tool_ids:
-                _result.append(_id)
-        elif node.get("type") == "mcp-node":
-            mcp_tool_id = node.get("properties", {}).get("node_data", {}).get("mcp_tool_id")
-            if mcp_tool_id:
-                _result.append(mcp_tool_id)
-    if with_deep:
-        workflow_list = QuerySet(Tool).filter(id__in=_result, tool_type=ToolType.WORKFLOW)
-        tool_work_flow_list = QuerySet(ToolWorkflow).filter(tool_id__in=[wl.id for wl in workflow_list])
-        for tool_work_flow in tool_work_flow_list:
-            child_tool_id_list = get_child_tool_id_list(tool_work_flow.work_flow, [])
-            for c in child_tool_id_list:
-                _result.append(c)
-    return _result
-
-
-def get_child_tool_id_list(work_flow, response):
-    from tools.models import ToolType, ToolWorkflow
-
-    tool_id_list = get_tool_id_list(work_flow, False)
-    tool_id_list = [tool_id for tool_id in tool_id_list if len([r for r in response if r == tool_id]) == 0]
-    tool_list = []
-    if len(tool_id_list) > 0:
-        tool_list = QuerySet(Tool).filter(id__in=tool_id_list).exclude(scope=ToolScope.SHARED)
-        work_flow_tools = [tool for tool in tool_list if tool.tool_type == ToolType.WORKFLOW]
-        if len(work_flow_tools) > 0:
-            work_flow_tool_dict = {
-                tw.tool_id: tw for tw in QuerySet(ToolWorkflow).filter(tool_id__in=[t.id for t in work_flow_tools])
-            }
-            for tool in tool_list:
-                response.append(str(tool.id))
-                if tool.tool_type == ToolType.WORKFLOW:
-                    get_child_tool_id_list(work_flow_tool_dict.get(tool.id).work_flow, response)
-        else:
-            for tool in tool_list:
-                response.append(str(tool.id))
-    return response
-
-
 def build_schema(fields: dict):
     return create_model("dynamicSchema", **fields)
 
@@ -1033,7 +844,9 @@ def get_workflow_args(tool, qv):
                 {
                     field.get("field"): (
                         get_type(field.get("type")),
-                        Field(..., required=True, description=field.get("desc")) if field.get("is_required") else Field(default=None, required=False, description=field.get("desc"))
+                        Field(..., required=True, description=field.get("desc"))
+                        if field.get("is_required")
+                        else Field(default=None, required=False, description=field.get("desc")),
                     )
                     for field in input_field_list
                 }

apps/application/serializers/application.py

@@ -1170,7 +1170,7 @@ def export(self, with_valid=True):
                 self.is_valid()
             application_id = self.data.get("application_id")
             application = QuerySet(Application).filter(id=application_id).first()
-            from application.flow.tools import get_tool_id_list
+            from system_manage.services.resource_mapping import get_tool_id_list
 
             tool_id_list = get_tool_id_list(application.work_flow, True)
             if len(tool_id_list) > 0:

apps/application/serializers/common.py

@@ -575,7 +575,7 @@ def get_cache(chat_id):
 
 
 def update_resource_mapping_by_application(application_id: str, other_resource_mapping=None):
-    from application.flow.tools import (
+    from system_manage.services.resource_mapping import (
         application_instance_field_call_dict,
         get_instance_resource,
         save_workflow_mapping,

apps/application/workflow/i_node.py

@@ -15,7 +15,8 @@
 from rest_framework import serializers
 
 from application.workflow.common import Node
-from application.workflow.message.struct.content import Content
+from application.workflow.message.struct.content import Content, NodeInfo, Position
+from application.workflow.message.struct.progress_content import ProgressContent
 from application.workflow.status import Status
 from common.utils.logger import maxkb_logger
 
@@ -108,6 +109,14 @@ def _run(self):
         执行节点
         @return:
         """
+        self.write(
+            ProgressContent(
+                self.node.id,
+                Status.BEFORE_RUNNING,
+                NodeInfo(self.get_node_id(), self.get_node_name(), Status.BEFORE_RUNNING),
+                Position(self.get_node_id()),
+            )
+        )
         self.execute()
         self.complete(Status.SUCCESS)
 

apps/application/workflow/message/struct/progress_content.py

@@ -0,0 +1,21 @@
+# coding=utf-8
+"""
+@project: MaxKB
+@Author：虎虎
+@file： reasoning_content.py
+@date：2026/6/30 16:07
+@desc:
+"""
+
+from application.workflow.content_type import ContentType
+from application.workflow.message.struct.content import Content, NodeInfo, Position
+from application.workflow.status import Status
+
+
+class ProgressContent(Content):
+    def __init__(self, _id, status: Status, node_info: NodeInfo, position: Position, **kwargs):
+        super().__init__(_id, status, ContentType.REASONING, node_info, position, **kwargs)
+
+    def to_dict(self):
+        result = super().to_dict()
+        return result

apps/application/workflow/nodes/data_source_local_node/__init__.py

@@ -0,0 +1,10 @@
+# coding=utf-8
+"""
+@project: MaxKB
+@Author:  虎虎虎
+@file:    __init__.py
+@date:    2026/9/11
+@desc:    本地文件数据源节点（知识库工作流起始节点之一）
+"""
+
+from .data_source_local_node import DataSourceLocalNode

apps/application/workflow/nodes/data_source_local_node/data_source_local_node.py

@@ -0,0 +1,59 @@
+# coding=utf-8
+"""
+@project: MaxKB
+@Author:  虎虎虎
+@file:    data_source_local_node.py
+@date:    2026/9/11
+@desc:    本地文件数据源节点：知识库工作流的起始节点之一，把上传的文件列表写入节点输出供下游读取
+"""
+
+from django.utils.translation import gettext_lazy as _
+from rest_framework import serializers
+
+from application.workflow.common import WorkflowType
+from application.workflow.i_node import INode
+
+
+class DataSourceLocalNodeParamsSerializer(serializers.Serializer):
+    file_type_list = serializers.ListField(child=serializers.CharField(label=_("")), label=_(""))
+    file_size_limit = serializers.IntegerField(required=True, label=_("Upload file size"))
+    file_count_limit = serializers.IntegerField(required=True, label=_("Number of uploaded files"))
+
+
+class DataSourceLocalNode(INode):
+    serializer_class = DataSourceLocalNodeParamsSerializer
+    supported_workflow_type_list = [WorkflowType.KNOWLEDGE]
+    type = "data-source-local-node"
+
+    @staticmethod
+    def get_form_list(node):
+        node_data = node.get("properties").get("node_data")
+        return [
+            {
+                "field": "file_list",
+                "input_type": "LocalFileUpload",
+                "attrs": {
+                    "file_count_limit": node_data.get("file_count_limit") or 10,
+                    "file_size_limit": node_data.get("file_size_limit") or 100,
+                    "file_type_list": node_data.get("file_type_list"),
+                },
+                "label": "",
+            }
+        ]
+
+    def execute(self):
+        # 文件列表来自工作流入参 data_source.file_list，写入本节点输出供下游节点引用
+        workflow_params = self.get_workflow_parameters()
+        file_list = (workflow_params.get("data_source") or {}).get("file_list")
+        self.write_context("file_list", file_list)
+
+    def get_details(self, index: int = 0, position: dict = None, old_details: dict = None, **kwargs):
+        details = super().get_details(index, position, old_details, **kwargs)
+        details.update(
+            {
+                "file_list": self.get_context("file_list"),
+                "knowledge_base": self.get_workflow_parameters().get("knowledge_base"),
+                "enableException": self.node.properties.get("enableException"),
+            }
+        )
+        return details

apps/application/workflow/nodes/document_extract_node/__init__.py

@@ -0,0 +1,10 @@
+# coding=utf-8
+"""
+@project: MaxKB
+@Author:  虎虎虎
+@file:    __init__.py
+@date:    2026/9/11
+@desc:    文档内容提取节点
+"""
+
+from .document_extract_node import DocumentExtractNode