diff --git a/apps/application/flow/tools.py b/apps/application/flow/tools.py index 7ee3bd7b1b5..b3cf19968a5 100644 --- a/apps/application/flow/tools.py +++ b/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 } diff --git a/apps/application/serializers/application.py b/apps/application/serializers/application.py index 1925b0058c3..1a05b7506c7 100644 --- a/apps/application/serializers/application.py +++ b/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: diff --git a/apps/application/serializers/common.py b/apps/application/serializers/common.py index e874f33f421..c1da528f3dc 100644 --- a/apps/application/serializers/common.py +++ b/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, diff --git a/apps/application/workflow/i_node.py b/apps/application/workflow/i_node.py index 772020bcc14..07b83ba0c36 100644 --- a/apps/application/workflow/i_node.py +++ b/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) diff --git a/apps/application/workflow/message/struct/progress_content.py b/apps/application/workflow/message/struct/progress_content.py new file mode 100644 index 00000000000..cab37d98b59 --- /dev/null +++ b/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 diff --git a/apps/application/workflow/nodes/data_source_local_node/__init__.py b/apps/application/workflow/nodes/data_source_local_node/__init__.py new file mode 100644 index 00000000000..dc72f858db7 --- /dev/null +++ b/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 diff --git a/apps/application/workflow/nodes/data_source_local_node/data_source_local_node.py b/apps/application/workflow/nodes/data_source_local_node/data_source_local_node.py new file mode 100644 index 00000000000..8d55aff2f27 --- /dev/null +++ b/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 diff --git a/apps/application/workflow/nodes/document_extract_node/__init__.py b/apps/application/workflow/nodes/document_extract_node/__init__.py new file mode 100644 index 00000000000..ec08b854e4d --- /dev/null +++ b/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 diff --git a/apps/application/workflow/nodes/document_extract_node/document_extract_node.py b/apps/application/workflow/nodes/document_extract_node/document_extract_node.py new file mode 100644 index 00000000000..225fdb7e18a --- /dev/null +++ b/apps/application/workflow/nodes/document_extract_node/document_extract_node.py @@ -0,0 +1,119 @@ +# coding=utf-8 +""" +@project: MaxKB +@Author: 虎虎虎 +@file: document_extract_node.py +@date: 2026/9/11 +@desc: 文档内容提取节点：把引用到的文件解析为文本内容,并保存文档内嵌图片 +""" + +import io + +from django.db.models import QuerySet +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 +from knowledge.models import File, FileSourceType +from knowledge.serializers.document import FileBufferHandle, parse_table_handle_list, split_handles + +splitter = "\n`-----------------------------------`\n" + + +class DocumentExtractNodeSerializer(serializers.Serializer): + document_list = serializers.ListField(required=False, label=_("document")) + + +class DocumentExtractNode(INode): + serializer_class = DocumentExtractNodeSerializer + supported_workflow_type_list = [WorkflowType.APPLICATION, WorkflowType.KNOWLEDGE, WorkflowType.TOOL] + type = "document-extract-node" + + def execute(self): + node_params = self.get_parameters() + workflow_params = self.get_workflow_parameters() + document_reference = node_params.get("document_list") or [] + document = ( + self.workflow_manage.get_reference_field(document_reference[0], document_reference[1:]) + if document_reference + else None + ) + chat_id = workflow_params.get("chat_id") + + self.write_context("document_list", document) + if document is None or not isinstance(document, list): + self.write_context("content", "") + self.write_context("document_list", []) + return + + # 按工作流类型确定归属资源 id(知识库/应用/工具),均取自工作流入参 + application_id = None + tool_id = None + knowledge_id = None + workflow_type = self.get_workflow_type() + if workflow_type == WorkflowType.KNOWLEDGE: + knowledge_id = workflow_params.get("knowledge_id") + elif workflow_type == WorkflowType.APPLICATION: + application_id = workflow_params.get("application_id") + elif workflow_type == WorkflowType.TOOL: + tool_id = workflow_params.get("tool_id") + + # doc 文件中内嵌的图片另存为文件 + def save_image(image_list): + for image in image_list: + meta = { + "debug": False if (application_id or knowledge_id or tool_id) else True, + "chat_id": chat_id, + "application_id": str(application_id) if application_id else None, + "knowledge_id": str(knowledge_id) if knowledge_id else None, + "tool_id": str(tool_id) if tool_id else None, + "file_id": str(image.id), + } + file_bytes = image.meta.pop("content") + new_file = File( + id=meta["file_id"], + file_name=image.file_name, + file_size=len(file_bytes), + source_type=FileSourceType.APPLICATION.value + if application_id + else FileSourceType.KNOWLEDGE.value + if knowledge_id + else FileSourceType.TOOL.value, + source_id=application_id or knowledge_id or tool_id, + meta=meta, + ) + if not QuerySet(File).filter(id=new_file.id).exists(): + new_file.save(file_bytes) + + get_buffer = FileBufferHandle().get_buffer + content = [] + document_list = [] + for doc in document: + file = QuerySet(File).filter(id=doc["file_id"]).first() + buffer = io.BytesIO(file.get_bytes()) + buffer.name = doc["name"] # this is the important line + + for split_handle in parse_table_handle_list + split_handles: + if split_handle.support(buffer, get_buffer): + buffer.seek(0) + file_content = split_handle.get_content(buffer, save_image) + content.append("### " + doc["name"] + "\n" + file_content) + document_list.append({"id": str(file.id), "name": doc["name"], "content": file_content}) + break + + self.write_context("content", splitter.join(content)) + self.write_context("document_list", document_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) + content = (self.get_context("content") or "").split(splitter) + details.update( + { + # 不保存 content 全部内容,因为 content 可能非常大 + "content": [file_content[:500] for file_content in content], + "document_list": self.get_context("document_list"), + "enableException": self.node.properties.get("enableException"), + } + ) + return details diff --git a/apps/application/workflow/nodes/document_split_node/__init__.py b/apps/application/workflow/nodes/document_split_node/__init__.py new file mode 100644 index 00000000000..fb048472596 --- /dev/null +++ b/apps/application/workflow/nodes/document_split_node/__init__.py @@ -0,0 +1,10 @@ +# coding=utf-8 +""" +@project: MaxKB +@Author: 虎虎虎 +@file: __init__.py +@date: 2026/9/11 +@desc: 文档分段节点 +""" + +from .document_split_node import DocumentSplitNode diff --git a/apps/application/workflow/nodes/document_split_node/document_split_node.py b/apps/application/workflow/nodes/document_split_node/document_split_node.py new file mode 100644 index 00000000000..89d11721ac6 --- /dev/null +++ b/apps/application/workflow/nodes/document_split_node/document_split_node.py @@ -0,0 +1,311 @@ +# coding=utf-8 +""" +@project: MaxKB +@Author: 虎虎虎 +@file: document_split_node.py +@date: 2026/9/11 +@desc: 文档分段节点：把提取出的文档内容按策略切分为段落,供知识库写入 +""" + +import io +import mimetypes +from typing import List + +from django.core.files.uploadedfile import InMemoryUploadedFile +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 +from common.chunk import text_to_chunk +from knowledge.serializers.document import FileBufferHandle, default_split_handle, md_qa_split_handle + + +class DocumentSplitNodeSerializer(serializers.Serializer): + document_list = serializers.ListField(required=False, label=_("document list")) + split_strategy = serializers.ChoiceField( + choices=["auto", "custom", "qa"], required=False, label=_("split strategy"), default="auto" + ) + paragraph_title_relate_problem_type = serializers.ChoiceField( + choices=["custom", "referencing"], + required=False, + label=_("paragraph title relate problem type"), + default="custom", + ) + paragraph_title_relate_problem = serializers.BooleanField( + required=False, label=_("paragraph title relate problem"), default=False + ) + paragraph_title_relate_problem_reference = serializers.ListField( + required=False, label=_("paragraph title relate problem reference"), child=serializers.CharField(), default=[] + ) + document_name_relate_problem_type = serializers.ChoiceField( + choices=["custom", "referencing"], + required=False, + label=_("document name relate problem type"), + default="custom", + ) + document_name_relate_problem = serializers.BooleanField( + required=False, label=_("document name relate problem"), default=False + ) + document_name_relate_problem_reference = serializers.ListField( + required=False, label=_("document name relate problem reference"), child=serializers.CharField(), default=[] + ) + limit = serializers.IntegerField(required=False, label=_("limit"), default=4096) + limit_type = serializers.ChoiceField( + choices=["custom", "referencing"], + required=False, + label=_("document name relate problem type"), + default="custom", + ) + limit_reference = serializers.ListField( + required=False, label=_("limit reference"), child=serializers.CharField(), default=[] + ) + chunk_size = serializers.IntegerField(required=False, label=_("chunk size"), default=256) + chunk_size_type = serializers.ChoiceField( + choices=["custom", "referencing"], required=False, label=_("chunk size type"), default="custom" + ) + chunk_size_reference = serializers.ListField( + required=False, label=_("chunk size reference"), child=serializers.CharField(), default=[] + ) + patterns = serializers.ListField(required=False, label=_("patterns"), child=serializers.CharField(), default=[]) + patterns_type = serializers.ChoiceField( + choices=["custom", "referencing"], required=False, label=_("patterns type"), default="custom" + ) + patterns_reference = serializers.ListField( + required=False, label=_("patterns reference"), child=serializers.CharField(), default=[] + ) + with_filter = serializers.BooleanField(required=False, label=_("with filter"), default=False) + with_filter_type = serializers.ChoiceField( + choices=["custom", "referencing"], required=False, label=_("with filter type"), default="custom" + ) + with_filter_reference = serializers.ListField( + required=False, label=_("with filter reference"), child=serializers.CharField(), default=[] + ) + + +def bytes_to_uploaded_file(file_bytes, file_name="file.txt"): + if file_name.startswith("http"): + file_name = "file.txt" + content_type, _unused = mimetypes.guess_type(file_name) + if content_type is None: + # 如果未能识别，设置为默认的二进制文件类型 + content_type = "application/octet-stream" + # 创建一个内存中的字节流对象 + file_stream = io.BytesIO(file_bytes) + # 获取文件大小 + file_size = len(file_bytes) + # 创建 InMemoryUploadedFile 对象 + uploaded_file = InMemoryUploadedFile( + file=file_stream, + field_name=None, + name=file_name, + content_type=content_type, + size=file_size, + charset=None, + ) + return uploaded_file + + +class DocumentSplitNode(INode): + serializer_class = DocumentSplitNodeSerializer + supported_workflow_type_list = [WorkflowType.APPLICATION, WorkflowType.KNOWLEDGE, WorkflowType.TOOL] + type = "document-split-node" + + def get_reference_content(self, fields: List[str]): + return self.workflow_manage.get_reference_field(fields[0], fields[1:]) if fields else None + + def execute(self): + # 通过 serializer 应用默认值(新引擎不会对 node_data 自动校验/填默认) + serializer = DocumentSplitNodeSerializer(data=self.get_parameters()) + serializer.is_valid(raise_exception=True) + params = serializer.data + + knowledge_id = ( + self.get_workflow_parameters().get("knowledge_id") + if self.get_workflow_type() == WorkflowType.KNOWLEDGE + else None + ) + + document_list = params.get("document_list") + split_strategy = params.get("split_strategy") + paragraph_title_relate_problem_type = params.get("paragraph_title_relate_problem_type") + paragraph_title_relate_problem = params.get("paragraph_title_relate_problem") + paragraph_title_relate_problem_reference = params.get("paragraph_title_relate_problem_reference") + document_name_relate_problem_type = params.get("document_name_relate_problem_type") + document_name_relate_problem = params.get("document_name_relate_problem") + document_name_relate_problem_reference = params.get("document_name_relate_problem_reference") + limit = params.get("limit") + limit_type = params.get("limit_type") + limit_reference = params.get("limit_reference") + chunk_size = params.get("chunk_size") + chunk_size_type = params.get("chunk_size_type") + chunk_size_reference = params.get("chunk_size_reference") + patterns = params.get("patterns") + patterns_type = params.get("patterns_type") + patterns_reference = params.get("patterns_reference") + with_filter = params.get("with_filter") + with_filter_type = params.get("with_filter_type") + with_filter_reference = params.get("with_filter_reference") + + self.write_context("knowledge_id", knowledge_id) + file_list = self.get_reference_content(document_list) + + # 处理引用类型的参数 + if patterns_type == "referencing": + patterns = self.get_reference_content(patterns_reference) + if limit_type == "referencing": + limit = self.get_reference_content(limit_reference) + if chunk_size_type == "referencing": + chunk_size = self.get_reference_content(chunk_size_reference) + if with_filter_type == "referencing": + with_filter = self.get_reference_content(with_filter_reference) + + paragraph_list = [] + for doc in file_list: + get_buffer = FileBufferHandle().get_buffer + + file_mem = bytes_to_uploaded_file(doc["content"].encode("utf-8"), doc["name"]) + if split_strategy == "qa": + result = md_qa_split_handle.handle(file_mem, get_buffer, self._save_image) + else: + result = default_split_handle.handle( + file_mem, patterns, with_filter, limit, get_buffer, self._save_image + ) + # 统一处理结果为列表 + results = result if isinstance(result, list) else [result] + + for item in results: + self._process_split_result( + item, + knowledge_id, + doc.get("id"), + doc.get("name"), + split_strategy, + paragraph_title_relate_problem_type, + paragraph_title_relate_problem, + paragraph_title_relate_problem_reference, + document_name_relate_problem_type, + document_name_relate_problem, + document_name_relate_problem_reference, + chunk_size, + ) + + paragraph_list += results + + self.write_context("paragraph_list", paragraph_list) + self.write_context("document_list", file_list) + self.write_context("limit", limit) + self.write_context("chunk_size", chunk_size) + self.write_context("with_filter", with_filter) + self.write_context("patterns", patterns) + self.write_context("split_strategy", split_strategy) + + def _save_image(self, image_list): + pass + + def _process_split_result( + self, + item, + knowledge_id, + source_file_id, + file_name, + split_strategy, + paragraph_title_relate_problem_type, + paragraph_title_relate_problem, + paragraph_title_relate_problem_reference, + document_name_relate_problem_type, + document_name_relate_problem, + document_name_relate_problem_reference, + chunk_size, + ): + """处理文档分割结果""" + item["meta"] = { + "knowledge_id": knowledge_id, + "source_file_id": source_file_id, + "source_url": file_name, + } + if item.get("name", "file.txt") == "file.txt": + item["name"] = file_name + item["source_file_id"] = source_file_id + item["paragraphs"] = item.pop("content", item.get("paragraphs", [])) + + for paragraph in item["paragraphs"]: + paragraph["problem_list"] = self._generate_problem_list( + paragraph, + file_name, + split_strategy, + paragraph_title_relate_problem_type, + paragraph_title_relate_problem, + paragraph_title_relate_problem_reference, + document_name_relate_problem_type, + document_name_relate_problem, + document_name_relate_problem_reference, + ) + paragraph["is_active"] = True + paragraph["chunks"] = text_to_chunk(paragraph["content"], chunk_size) + + def _generate_problem_list( + self, + paragraph, + document_name, + split_strategy, + paragraph_title_relate_problem_type, + paragraph_title_relate_problem, + paragraph_title_relate_problem_reference, + document_name_relate_problem_type, + document_name_relate_problem, + document_name_relate_problem_reference, + ): + if paragraph_title_relate_problem_type == "referencing": + paragraph_title_relate_problem = self.get_reference_content(paragraph_title_relate_problem_reference) + if document_name_relate_problem_type == "referencing": + document_name_relate_problem = self.get_reference_content(document_name_relate_problem_reference) + + problem_list = [ + item + for p in paragraph.get("problem_list", []) + for item in p.get("content", "").split("
") + if item.strip() + ] + + if split_strategy == "auto": + if paragraph_title_relate_problem and paragraph.get("title"): + problem_list.append(paragraph.get("title")) + if document_name_relate_problem and document_name: + problem_list.append(document_name) + elif split_strategy == "custom": + if paragraph_title_relate_problem and paragraph.get("title"): + problem_list.append(paragraph.get("title")) + if document_name_relate_problem and document_name: + problem_list.append(document_name) + elif split_strategy == "qa": + if document_name_relate_problem and document_name: + problem_list.append(document_name) + + return list(set(problem_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) + paragraph_list = self.get_context("paragraph_list") or [] + # 每个文档保留前 5 个分段 + limited_paragraph_list = [] + for doc in paragraph_list: + if doc.get("paragraphs"): + doc_copy = doc.copy() + doc_copy["paragraphs"] = doc["paragraphs"][:5] + limited_paragraph_list.append(doc_copy) + else: + limited_paragraph_list.append(doc) + + details.update( + { + "paragraph_list": limited_paragraph_list, + "limit": self.get_context("limit"), + "chunk_size": self.get_context("chunk_size"), + "with_filter": self.get_context("with_filter"), + "patterns": self.get_context("patterns"), + "split_strategy": self.get_context("split_strategy"), + "enableException": self.node.properties.get("enableException"), + } + ) + return details diff --git a/apps/application/workflow/nodes/knowledge_write_node/__init__.py b/apps/application/workflow/nodes/knowledge_write_node/__init__.py new file mode 100644 index 00000000000..e81dda631fb --- /dev/null +++ b/apps/application/workflow/nodes/knowledge_write_node/__init__.py @@ -0,0 +1,10 @@ +# coding=utf-8 +""" +@project: MaxKB +@Author: 虎虎虎 +@file: __init__.py +@date: 2026/9/11 +@desc: 知识库写入节点 +""" + +from .knowledge_write_node import KnowledgeWriteNode diff --git a/apps/application/workflow/nodes/knowledge_write_node/knowledge_write_node.py b/apps/application/workflow/nodes/knowledge_write_node/knowledge_write_node.py new file mode 100644 index 00000000000..1a10e33558f --- /dev/null +++ b/apps/application/workflow/nodes/knowledge_write_node/knowledge_write_node.py @@ -0,0 +1,392 @@ +# coding=utf-8 +""" +@project: MaxKB +@Author: 虎虎虎 +@file: knowledge_write_node.py +@date: 2026/9/11 +@desc: 知识库写入节点：把上游产出的文档/段落写入知识库并触发向量化 +""" + +from functools import reduce +from typing import Any, Dict, List + +import uuid_utils.compat as uuid +from django.db.models import QuerySet +from django.db.models.aggregates import Max +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 +from common.chunk import text_to_chunk +from common.utils.common import bulk_create_in_batches, filter_special_character +from knowledge.models import ( + ContentOrigin, + Document, + DocumentResourceType, + DocumentTag, + File, + FileSourceType, + KnowledgeType, + Paragraph, + Problem, + ProblemParagraphMapping, + Tag, +) +from knowledge.serializers.common import ProblemParagraphManage, ProblemParagraphObject +from knowledge.serializers.document import DocumentSerializers +from knowledge.serializers.document_strategy import DocumentStrategySerializer +from knowledge.services.document_strategy import ( + document_source_hash, + normalize_document_strategy, + strategy_hashes, +) +from knowledge.services.incremental_sync import prepare_remote_paragraphs + + +class ParagraphInstanceSerializer(serializers.Serializer): + content = serializers.CharField( + required=True, label=_("content"), max_length=102400, min_length=1, allow_null=True, allow_blank=True + ) + title = serializers.CharField( + required=False, max_length=256, label=_("section title"), allow_null=True, allow_blank=True + ) + problem_list = serializers.ListField(required=False, child=serializers.CharField(required=False, allow_blank=True)) + is_active = serializers.BooleanField(required=False, label=_("Is active")) + chunks = serializers.ListField(required=False, child=serializers.CharField(required=True)) + + +class TagInstanceSerializer(serializers.Serializer): + key = serializers.CharField(required=True, max_length=64, label=_("Tag Key")) + value = serializers.CharField(required=True, max_length=128, label=_("Tag Value")) + + +class KnowledgeWriteParamSerializer(serializers.Serializer): + name = serializers.CharField( + required=True, label=_("document name"), max_length=128, min_length=1, source=_("document name") + ) + meta = serializers.DictField(required=False) + tags = serializers.ListField(required=False, label=_("Tags"), child=TagInstanceSerializer()) + paragraphs = ParagraphInstanceSerializer(required=False, many=True, allow_null=True) + source_file_id = serializers.UUIDField(required=False, allow_null=True) + user_id = serializers.UUIDField(required=False, allow_null=True) + doc_strategy = DocumentStrategySerializer(required=False, allow_null=True) + + +class KnowledgeWriteNodeParamSerializer(serializers.Serializer): + document_list = serializers.ListField( + required=True, child=serializers.CharField(required=True), allow_null=True, label=_("document list") + ) + + +def convert_uuid_to_str(obj): + if isinstance(obj, dict): + return {k: convert_uuid_to_str(v) for k, v in obj.items()} + elif isinstance(obj, list): + return [convert_uuid_to_str(i) for i in obj] + elif isinstance(obj, uuid.UUID): + return str(obj) + else: + return obj + + +def link_file(source_file_id, document_id): + if source_file_id is None: + return + source_file = QuerySet(File).filter(id=source_file_id).first() + if source_file: + file_content = source_file.get_bytes() + + new_file = File( + id=uuid.uuid7(), + file_name=source_file.file_name, + file_size=source_file.file_size, + source_type=FileSourceType.DOCUMENT, + source_id=document_id, # 更新为当前知识库ID + meta=source_file.meta.copy() if source_file.meta else {}, + ) + + # 保存文件内容和元数据 + new_file.save(file_content) + + +def get_paragraph_problem_model(knowledge_id: str, document_id: str, instance: Dict): + content = filter_special_character(instance.get("content")) + paragraph = Paragraph( + id=uuid.uuid7(), + document_id=document_id, + content=content, + knowledge_id=knowledge_id, + title=instance.get("title") if "title" in instance else "", + chunks=[ + filter_special_character(c) + for c in ( + instance.get("chunks") + if "chunks" in instance + else text_to_chunk(content, instance.get("child_length", 256)) + ) + ], + origin=instance.get("origin", ContentOrigin.SYNCED), + source_key=instance.get("source_key", ""), + source_hash=instance.get("source_hash", ""), + source_snapshot=instance.get("source_snapshot") + or { + "title": instance.get("title") or "", + "content": content, + }, + source_updated_at=instance.get("source_updated_at"), + ) + + problem_paragraph_object_list = [ + ProblemParagraphObject(knowledge_id, document_id, str(paragraph.id), problem) + for problem in (instance.get("problem_list") if "problem_list" in instance else []) + ] + + return { + "paragraph": paragraph, + "problem_paragraph_object_list": problem_paragraph_object_list, + } + + +def get_paragraph_model(document_model, paragraph_list: List): + knowledge_id = document_model.knowledge_id + paragraph_model_dict_list = [ + get_paragraph_problem_model(knowledge_id, document_model.id, paragraph) for paragraph in paragraph_list + ] + + paragraph_model_list = [] + problem_paragraph_object_list = [] + for paragraphs in paragraph_model_dict_list: + paragraph = paragraphs.get("paragraph") + for problem_model in paragraphs.get("problem_paragraph_object_list"): + problem_paragraph_object_list.append(problem_model) + paragraph_model_list.append(paragraph) + + return { + "document": document_model, + "paragraph_model_list": paragraph_model_list, + "problem_paragraph_object_list": problem_paragraph_object_list, + } + + +def get_document_paragraph_model(knowledge_id: str, instance: Dict): + source_meta = {"source_file_id": instance.get("source_file_id")} if instance.get("source_file_id") else {} + meta = {**instance.get("meta"), **source_meta} if instance.get("meta") is not None else source_meta + meta = {**convert_uuid_to_str(meta), "allow_download": True} + + strategy = normalize_document_strategy(instance.get("doc_strategy")) + normalized_paragraphs = prepare_remote_paragraphs( + [ + { + **paragraph, + "content": filter_special_character(paragraph.get("content")), + "origin": ContentOrigin.SYNCED, + "child_length": strategy["split"]["child_length"], + } + for paragraph in instance.get("paragraphs", []) + ] + ) + document_model = Document( + **{ + "knowledge_id": knowledge_id, + "id": uuid.uuid7(), + "name": instance.get("name"), + "char_length": reduce(lambda x, y: x + y, [len(p.get("content")) for p in normalized_paragraphs], 0), + "meta": meta, + "type": instance.get("type") if instance.get("type") is not None else KnowledgeType.WORKFLOW, + "resource_type": DocumentResourceType.DOCUMENT, + "doc_strategy": strategy, + "source_hash": document_source_hash(normalized_paragraphs), + "user_id": instance.get("user_id"), + **strategy_hashes(strategy), + } + ) + + return get_paragraph_model(document_model, normalized_paragraphs) + + +def save_knowledge_tags(knowledge_id: str, tags: List[Dict[str, Any]]): + existed_tags_dict = { + (key, value): str(tag_id) + for key, value, tag_id in QuerySet(Tag).filter(knowledge_id=knowledge_id).values_list("key", "value", "id") + } + + tag_model_list = [] + new_tag_dict = {} + for tag in tags: + key = tag.get("key") + value = tag.get("value") + + if (key, value) not in existed_tags_dict: + tag_model = Tag(id=uuid.uuid7(), knowledge_id=knowledge_id, key=key, value=value) + tag_model_list.append(tag_model) + new_tag_dict[(key, value)] = str(tag_model.id) + + if tag_model_list: + Tag.objects.bulk_create(tag_model_list) + + all_tag_dict = {**existed_tags_dict, **new_tag_dict} + + return all_tag_dict, new_tag_dict + + +def batch_add_document_tag(document_tag_map: Dict[str, List[str]]): + """ + 批量添加文档-标签关联 + document_tag_map: {document_id: [tag_id1, tag_id2, ...]} + """ + all_document_ids = list(document_tag_map.keys()) + all_tag_ids = list(set(tag_id for tag_ids in document_tag_map.values() for tag_id in tag_ids)) + + # 查询已存在的文档-标签关联 + existed_relations = set( + QuerySet(DocumentTag) + .filter(document_id__in=all_document_ids, tag_id__in=all_tag_ids) + .values_list("document_id", "tag_id") + ) + + new_relations = [ + DocumentTag( + id=uuid.uuid7(), + document_id=doc_id, + tag_id=tag_id, + ) + for doc_id, tag_ids in document_tag_map.items() + for tag_id in tag_ids + if (doc_id, tag_id) not in existed_relations + ] + + if new_relations: + QuerySet(DocumentTag).bulk_create(new_relations) + + +class KnowledgeWriteNode(INode): + serializer_class = KnowledgeWriteNodeParamSerializer + supported_workflow_type_list = [WorkflowType.KNOWLEDGE] + type = "knowledge-write-node" + + def save(self, document_list, user_id): + serializer = KnowledgeWriteParamSerializer(data=document_list, many=True) + serializer.is_valid(raise_exception=True) + document_list = serializer.data + + workflow_params = self.get_workflow_parameters() + knowledge_id = workflow_params.get("knowledge_id") + workspace_id = workflow_params.get("workspace_id") + + document_model_list = [] + paragraph_model_list = [] + problem_paragraph_object_list = [] + # 文档标签映射关系 + document_tags_map = {} + knowledge_tag_dict = {} + + for document in document_list: + document["user_id"] = user_id + document_paragraph_dict_model = get_document_paragraph_model(knowledge_id, document) + document_instance = document_paragraph_dict_model.get("document") + link_file(document.get("source_file_id"), document_instance.id) + document_model_list.append(document_instance) + # 收集标签 + single_document_tag_list = document.get("tags", []) + # 去重传入的标签 + for tag in single_document_tag_list: + tag_key = (tag["key"], tag["value"]) + if tag_key not in knowledge_tag_dict: + knowledge_tag_dict[tag_key] = tag + + if single_document_tag_list: + document_tags_map[str(document_instance.id)] = single_document_tag_list + + for paragraph in document_paragraph_dict_model.get("paragraph_model_list"): + paragraph_model_list.append(paragraph) + for problem_paragraph_object in document_paragraph_dict_model.get("problem_paragraph_object_list"): + problem_paragraph_object_list.append(problem_paragraph_object) + knowledge_tag_list = list(knowledge_tag_dict.values()) + # 保存所有文档中含有的标签到知识库 + if knowledge_tag_list: + all_tag_dict, new_tag_dict = save_knowledge_tags(knowledge_id, knowledge_tag_list) + # 构建文档-标签ID映射 + document_tag_id_map = {} + # 为每个文档添加其对应的标签 + for doc_id, doc_tags in document_tags_map.items(): + doc_tag_ids = [ + all_tag_dict[(tag.get("key"), tag.get("value"))] + for tag in doc_tags + if (tag.get("key"), tag.get("value")) in all_tag_dict + ] + if doc_tag_ids: + document_tag_id_map[doc_id] = doc_tag_ids + if document_tag_id_map: + batch_add_document_tag(document_tag_id_map) + + problem_model_list, problem_paragraph_mapping_list = ProblemParagraphManage( + problem_paragraph_object_list, knowledge_id + ).to_problem_model_list() + + QuerySet(Document).bulk_create(document_model_list) if len(document_model_list) > 0 else None + + if len(paragraph_model_list) > 0: + for document in document_model_list: + max_position = ( + Paragraph.objects.filter(document_id=document.id).aggregate(max_position=Max("position"))[ + "max_position" + ] + or 0 + ) + sub_list = [p for p in paragraph_model_list if p.document_id == document.id] + for i, paragraph in enumerate(sub_list): + paragraph.position = max_position + i + 1 + QuerySet(Paragraph).bulk_create(sub_list if len(sub_list) > 0 else []) + + bulk_create_in_batches(Problem, problem_model_list, batch_size=1000) + + bulk_create_in_batches(ProblemParagraphMapping, problem_paragraph_mapping_list, batch_size=1000) + + return document_model_list, knowledge_id, workspace_id + + @staticmethod + def post_embedding(document_model_list, knowledge_id, workspace_id): + for document in document_model_list: + DocumentSerializers.Operate( + data={"knowledge_id": knowledge_id, "document_id": document.id, "workspace_id": workspace_id} + ).refresh() + + def execute(self): + node_params = self.get_parameters() + workflow_params = self.get_workflow_parameters() + document_reference = node_params.get("document_list") or [] + documents = ( + self.workflow_manage.get_reference_field(document_reference[0], document_reference[1:]) + if document_reference + else [] + ) + user_id = workflow_params.get("user_id") + + document_model_list, knowledge_id, workspace_id = self.save(documents, user_id) + self.post_embedding(document_model_list, knowledge_id, workspace_id) + + write_content_list = [ + { + "name": document.get("name"), + "paragraphs": [ + { + "title": p.get("title"), + "content": p.get("content"), + } + for p in document.get("paragraphs")[0:5] + ], + } + for document in documents + ] + self.write_context("write_content", write_content_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( + { + "write_content": self.get_context("write_content"), + "enableException": self.node.properties.get("enableException"), + } + ) + return details diff --git a/apps/common/auth/handle/impl/application_key.py b/apps/common/auth/handle/impl/application_key.py index 48bc60d1681..661f3b5ca98 100644 --- a/apps/common/auth/handle/impl/application_key.py +++ b/apps/common/auth/handle/impl/application_key.py @@ -1,16 +1,18 @@ # coding=utf-8 """ - @project: MaxKB - @Author：虎虎 - @file： application_key.py - @date：2025/7/10 03:02 - @desc: 应用api key认证 +@project: MaxKB +@Author：虎虎 +@file： application_key.py +@date：2025/7/10 03:02 +@desc: 应用api key认证 """ + from django.db.models import QuerySet from django.utils import timezone from django.utils.translation import gettext_lazy as _ from application.models import ApplicationApiKey, ChatUserType, ApplicationAccessToken +from common.auth.constants.group_constants import Group from common.auth.handle.auth_base_handle import AuthBaseHandle from common.auth.struct.auth import Principal, Auth from common.exception.app_exception import AppAuthenticationFailed @@ -20,20 +22,25 @@ class ApplicationKey(AuthBaseHandle): def handle(self, request, token: str, get_token_details): application_api_key = QuerySet(ApplicationApiKey).filter(secret_key=token).first() if application_api_key is None: - raise AppAuthenticationFailed(500, _('Secret key is invalid')) + raise AppAuthenticationFailed(500, _("Secret key is invalid")) if not application_api_key.is_active: - raise AppAuthenticationFailed(500, _('Secret key is invalid')) + raise AppAuthenticationFailed(500, _("Secret key is invalid")) if application_api_key.is_permanent is False and application_api_key.expire_time < timezone.now(): - raise AppAuthenticationFailed(500, _('Secret key is expired')) - application_access_token = QuerySet(ApplicationAccessToken).filter( - application_id=application_api_key.application_id).first() + raise AppAuthenticationFailed(500, _("Secret key is expired")) + application_access_token = ( + QuerySet(ApplicationAccessToken).filter(application_id=application_api_key.application_id).first() + ) if application_access_token is not None: if application_access_token.authentication: - if application_access_token.authentication_value.get('type', - 'password') != 'password': - raise AppAuthenticationFailed(1002, _('Authentication information is incorrect')) - return Principal(str(application_api_key.id), ChatUserType.APPLICATION_API_KEY, - application_id=str(application_api_key.application_id)), Auth(set(), {}) + if application_access_token.authentication_value.get("type", "password") != "password": + raise AppAuthenticationFailed(1002, _("Authentication information is incorrect")) + + k = f"{Group.CHAT_USER}:r:{application_access_token.application_id}" + return Principal( + str(application_api_key.id), + ChatUserType.APPLICATION_API_KEY, + application_id=str(application_api_key.application_id), + ), Auth(set(), {k: 1}) def support(self, request, token: str, get_token_details): - return str(token).startswith("application-") or str(token).startswith('agent-') + return str(token).startswith("application-") or str(token).startswith("agent-") diff --git a/apps/knowledge/serializers/common.py b/apps/knowledge/serializers/common.py index 78d08198c72..4ba863e8576 100644 --- a/apps/knowledge/serializers/common.py +++ b/apps/knowledge/serializers/common.py @@ -17,7 +17,11 @@ from django.utils.translation import gettext_lazy as _ from rest_framework import serializers -from application.flow.tools import save_workflow_mapping, get_instance_resource, knowledge_instance_field_call_dict +from system_manage.services.resource_mapping import ( + save_workflow_mapping, + get_instance_resource, + knowledge_instance_field_call_dict, +) from common.config.embedding_config import ModelManage from common.db.search import native_search from common.db.sql_execute import sql_execute, update_execute diff --git a/apps/knowledge/serializers/knowledge.py b/apps/knowledge/serializers/knowledge.py index f5d7b935557..d3667e79ac8 100644 --- a/apps/knowledge/serializers/knowledge.py +++ b/apps/knowledge/serializers/knowledge.py @@ -97,7 +97,7 @@ from knowledge.task.embedding import delete_embedding_by_knowledge, embedding_by_knowledge from knowledge.task.generate import generate_related_by_knowledge_id from knowledge.task.sync import sync_replace_web_knowledge, sync_web_knowledge -from application.flow.tools import get_tool_id_list +from system_manage.services.resource_mapping import get_tool_id_list from tools.models import Tool, ToolScope, ToolType, ToolWorkflow diff --git a/apps/knowledge/serializers/knowledge_workflow.py b/apps/knowledge/serializers/knowledge_workflow.py index 49a6612bae9..1523ed29ca6 100644 --- a/apps/knowledge/serializers/knowledge_workflow.py +++ b/apps/knowledge/serializers/knowledge_workflow.py @@ -3,13 +3,13 @@ import base64 import json import pickle +import time from copy import deepcopy from functools import reduce from typing import Dict, List import requests import uuid_utils.compat as uuid -from django.core.cache import cache from django.db import transaction from django.db.models import QuerySet from django.http import HttpResponse @@ -17,13 +17,14 @@ from django.utils.translation import gettext_lazy as _ from rest_framework import serializers, status -from application.flow.common import Workflow, WorkflowMode -from application.flow.i_step_node import KnowledgeWorkflowPostHandler -from application.flow.knowledge_workflow_manage import KnowledgeWorkflowManage -from application.flow.step_node import get_node -from application.flow.tools import get_tool_id_list, save_workflow_mapping +from application.workflow.common import WorkflowType, new_instance +from application.workflow.i_node import Signal +from application.workflow.nodes import get_node_class +from system_manage.services.resource_mapping import get_tool_id_list, save_workflow_mapping +from application.workflow.status import Status +from application.workflow.workflow_manage import CallBack, WorkflowManage +from application.workflow.workflow_run_registry import WorkflowRunRegistry from application.mcp_tools import get_mcp_tools -from common.constants.cache_version import Cache_Version from common.db.search import page_search from common.exception.app_exception import AppApiException from common.field.common import UploadedFileField @@ -42,11 +43,17 @@ KnowledgeWorkflowVersion, File, FileSourceType, + Document, + DocumentResourceType, + KnowledgeSyncLog, + KnowledgeSyncStatus, + KnowledgeSyncType, ) from knowledge.models.knowledge_action import KnowledgeAction, State from knowledge.serializers.common import update_resource_mapping_by_knowledge from knowledge.serializers.knowledge_model import KnowledgeModelSerializer from knowledge.services.document_cleanup import delete_document_data +from knowledge.services.workflow_sync import merge_workflow_incremental_snapshot from system_manage.models import AuthTargetType from system_manage.models.resource_mapping import ResourceType from system_manage.serializers.user_resource_permission import UserResourcePermissionSerializer @@ -81,6 +88,56 @@ def hand_node(node, update_tool_map): node.get("properties", {}).get("node_data", {})["tool_lib_id"] = update_tool_map.get(tool_lib_id, tool_lib_id) +def finalize_knowledge_action(knowledge_action_id, state, run_time, sync_log_id=None, document_cleanup=None): + """ + 知识库工作流执行结束后的收尾:更新 KnowledgeAction 的最终状态/耗时,并在同步场景下收尾 KnowledgeSyncLog。 + 与具体执行引擎解耦——state/run_time 由调用方按各自引擎算好传入。 + """ + QuerySet(KnowledgeAction).filter(id=knowledge_action_id).update(state=state, run_time=run_time) + if sync_log_id is not None: + sync_log = QuerySet(KnowledgeSyncLog).filter(id=sync_log_id).first() + if sync_log is not None: + if ( + state == State.SUCCESS + and sync_log.sync_type == KnowledgeSyncType.INCREMENTAL + and document_cleanup is not None + ): + stats = merge_workflow_incremental_snapshot(sync_log) + else: + stats = { + "total_count": QuerySet(Document) + .filter( + knowledge_id=sync_log.knowledge_id, + resource_type=DocumentResourceType.DOCUMENT, + ) + .count(), + "synced_count": QuerySet(Document) + .filter( + knowledge_id=sync_log.knowledge_id, + type=KnowledgeType.WORKFLOW, + resource_type=DocumentResourceType.DOCUMENT, + create_time__gte=sync_log.create_time, + ) + .count(), + "skipped_count": 0, + "deleted_count": sync_log.deleted_count, + "failed_count": 0 if state == State.SUCCESS else 1, + } + is_success = state == State.SUCCESS + QuerySet(KnowledgeSyncLog).filter(id=sync_log.id).update( + status=KnowledgeSyncStatus.SUCCESS + if is_success and not stats["failed_count"] + else KnowledgeSyncStatus.FAILURE, + total_count=stats["total_count"], + synced_count=stats["synced_count"], + skipped_count=stats["skipped_count"], + deleted_count=stats["deleted_count"], + failed_count=stats["failed_count"], + duration_ms=max(0, round(run_time * 1000)), + message=f"Workflow action {knowledge_action_id}: {state}", + ) + + class KnowledgeWorkflowModelSerializer(serializers.ModelSerializer): class Meta: model = KnowledgeWorkflow @@ -199,26 +256,7 @@ def action(self, instance: Dict, user, with_valid=True, sync_log_id=None): "workspace_id": knowledge.workspace_id, }, } - work_flow_manage = KnowledgeWorkflowManage( - Workflow.new_instance(knowledge_workflow.work_flow, WorkflowMode.KNOWLEDGE), - { - "knowledge_id": self.data.get("knowledge_id"), - "knowledge_action_id": knowledge_action_id, - "stream": True, - "workspace_id": self.data.get("workspace_id"), - "user_id": str(user.id), - **instance, - }, - KnowledgeWorkflowPostHandler(None, knowledge_action_id, sync_log_id, delete_document_data), - is_the_task_interrupted=lambda: ( - cache.get( - Cache_Version.KNOWLEDGE_WORKFLOW_INTERRUPTED.get_key(action_id=knowledge_action_id), - version=Cache_Version.KNOWLEDGE_WORKFLOW_INTERRUPTED.get_version(), - ) - or False - ), - ) - work_flow_manage.run() + self._launch_knowledge_workflow(instance, user, knowledge_action_id, knowledge_workflow.work_flow, sync_log_id) # 需要把文件改成永久文件 data_source = instance.get("data_source") or {} file_ids = [item.get("file_id") for item in data_source.get("file_list") or [] if item.get("file_id")] @@ -239,6 +277,66 @@ def action(self, instance: Dict, user, with_valid=True, sync_log_id=None): "meta": meta, } + def _launch_knowledge_workflow(self, instance: Dict, user, knowledge_action_id, work_flow, sync_log_id=None): + """ + 在新引擎上异步启动知识库工作流(action/upload_document 共用): + 动态解析数据源起点 -> 注册到运行注册表(供停止)-> run() 每节点起线程立即返回, + 最终状态由 on_complete 通过 finalize_knowledge_action 落库。 + """ + parameters = { + "knowledge_id": self.data.get("knowledge_id"), + "knowledge_action_id": knowledge_action_id, + "stream": True, + "workspace_id": self.data.get("workspace_id"), + "user_id": str(user.id), + **instance, + } + workflow = new_instance(work_flow, WorkflowType.KNOWLEDGE) + start_time = time.time() + + def get_node_parameters(node): + return node.properties.get("node_data", {}) + + def get_start_node_fn(wf, wm): + # 知识库起点是数据源节点,由 data_source.node_id 指定(动态,非固定 start-node) + node_id = (instance.get("data_source") or {}).get("node_id") + node = wf.get_node(node_id) if node_id else None + if node is None: + raise AppApiException(500, _("The start node does not exist")) + return get_node_class(node.type, WorkflowType.KNOWLEDGE)(node, wm, get_node_parameters) + + def on_next(wf_manage, content): + # 实时刷新节点详情,供前端轮询 KnowledgeAction 展示进度 + QuerySet(KnowledgeAction).filter(id=knowledge_action_id).update(details=wf_manage.get_details()) + + def on_complete(wf_manage, error): + WorkflowRunRegistry.unregister(str(knowledge_action_id)) + details = wf_manage.get_details() + cancelled = wf_manage.signal == Signal.CANCELLED + state = self.compute_knowledge_state(details, error, cancelled) + run_time = time.time() - start_time + QuerySet(KnowledgeAction).filter(id=knowledge_action_id).update(details=details) + finalize_knowledge_action(knowledge_action_id, state, run_time, sync_log_id, delete_document_data) + + call_back = CallBack(on_next, on_complete) + work_flow_manage = WorkflowManage(workflow, parameters, WorkflowType.KNOWLEDGE, call_back, get_start_node_fn) + WorkflowRunRegistry.register(str(knowledge_action_id), None, work_flow_manage) + work_flow_manage.run() + return work_flow_manage + + @staticmethod + def compute_knowledge_state(details, error, cancelled): + if cancelled: + return State.REVOKED + details = details or [] + has_fail = any(d.get("status") == Status.FAIL.value and not d.get("enableException") for d in details) + if error or has_fail: + return State.FAILURE + write_exist = any(d.get("type") == "knowledge-write-node" for d in details) + if not write_exist: + return State.FAILURE + return State.SUCCESS + def upload_document(self, instance: Dict, user, with_valid=True): if with_valid: self.is_valid(raise_exception=True) @@ -266,26 +364,8 @@ def upload_document(self, instance: Dict, user, with_valid=True): "workspace_id": knowledge.workspace_id, }, } - work_flow_manage = KnowledgeWorkflowManage( - Workflow.new_instance(knowledge_workflow_version.work_flow, WorkflowMode.KNOWLEDGE), - { - "knowledge_id": self.data.get("knowledge_id"), - "knowledge_action_id": knowledge_action_id, - "stream": True, - "workspace_id": self.data.get("workspace_id"), - "user_id": str(user.id), - **instance, - }, - KnowledgeWorkflowPostHandler(None, knowledge_action_id), - is_the_task_interrupted=lambda: ( - cache.get( - Cache_Version.KNOWLEDGE_WORKFLOW_INTERRUPTED.get_key(action_id=knowledge_action_id), - version=Cache_Version.KNOWLEDGE_WORKFLOW_INTERRUPTED.get_version(), - ) - or False - ), - ) - work_flow_manage.run() + # 线上上传走已发布版本的 work_flow,执行链路与 action 一致(新引擎异步执行) + self._launch_knowledge_workflow(instance, user, knowledge_action_id, knowledge_workflow_version.work_flow) return { "id": knowledge_action_id, "knowledge_id": self.data.get("knowledge_id"), @@ -335,11 +415,8 @@ def cancel(self, is_valid=True): if is_valid: self.is_valid(raise_exception=True) knowledge_action_id = self.data.get("id") - cache.set( - Cache_Version.KNOWLEDGE_WORKFLOW_INTERRUPTED.get_key(action_id=knowledge_action_id), - True, - version=Cache_Version.KNOWLEDGE_WORKFLOW_INTERRUPTED.get_version(), - ) + # action / upload_document 均在新引擎执行,统一向运行注册表发送停止信号 + WorkflowRunRegistry.cancel_by_record_id(str(knowledge_action_id)) QuerySet(KnowledgeAction).filter( id=knowledge_action_id, knowledge_id=self.data.get("knowledge_id"), @@ -358,8 +435,9 @@ class Datasource(serializers.Serializer): def action(self): self.is_valid(raise_exception=True) if self.data.get("type") == "local": - node = get_node(self.data.get("id"), WorkflowMode.KNOWLEDGE) - return node.__getattribute__(node, self.data.get("function_name"))(**self.data.get("params")) + # self.data["id"] 为数据源节点类型,取新引擎该节点类上的同名静态方法(如 get_form_list) + node_class = get_node_class(self.data.get("id"), WorkflowType.KNOWLEDGE) + return getattr(node_class, self.data.get("function_name"))(**self.data.get("params")) elif self.data.get("type") == "tool": tool = QuerySet(Tool).filter(id=self.data.get("id")).first() init_params = json.loads(rsa_long_decrypt(tool.init_params)) diff --git a/apps/knowledge/tests.py b/apps/knowledge/tests.py index 20ba09ceaee..785f5674a57 100644 --- a/apps/knowledge/tests.py +++ b/apps/knowledge/tests.py @@ -1128,12 +1128,13 @@ def test_scheduled_workflow_uses_saved_input_and_starts_an_action(self, query_se str(sync_log.id), ) - @patch("knowledge.serializers.knowledge_workflow.Workflow.new_instance") - @patch("knowledge.serializers.knowledge_workflow.KnowledgeWorkflowManage") + @patch("knowledge.serializers.knowledge_workflow.WorkflowRunRegistry") + @patch("knowledge.serializers.knowledge_workflow.new_instance") + @patch("knowledge.serializers.knowledge_workflow.WorkflowManage") @patch("knowledge.serializers.knowledge_workflow.KnowledgeAction.save") @patch("knowledge.serializers.knowledge_workflow.QuerySet") def test_manual_action_saves_input_for_later_scheduled_runs( - self, query_set, _save_action, workflow_manage, _new_workflow + self, query_set, _save_action, workflow_manage, _new_instance, _registry ): workflow = MagicMock(work_flow={}) knowledge = MagicMock( diff --git a/apps/knowledge/views/knowledge_workflow.py b/apps/knowledge/views/knowledge_workflow.py index 43e33107937..5c3ca434c3d 100644 --- a/apps/knowledge/views/knowledge_workflow.py +++ b/apps/knowledge/views/knowledge_workflow.py @@ -131,7 +131,7 @@ class KnowledgeWorkflowActionView(APIView): def post(self, request: Request, workspace_id: str, knowledge_id: str): return result.success( KnowledgeWorkflowActionSerializer(data={"workspace_id": workspace_id, "knowledge_id": knowledge_id}).action( - request.data, request.user, True + request.data, request.user.profile, True ) ) diff --git a/apps/system_manage/migrations/0005_resourcemapping.py b/apps/system_manage/migrations/0005_resourcemapping.py index bafecc4a8d5..36d6737731c 100644 --- a/apps/system_manage/migrations/0005_resourcemapping.py +++ b/apps/system_manage/migrations/0005_resourcemapping.py @@ -10,56 +10,74 @@ def get_initialization_resource_mapping(): from django.db.models import QuerySet - from application.flow.tools import get_workflow_resource, get_node_handle_callback, \ - get_instance_resource + from system_manage.services.resource_mapping import ( + get_workflow_resource, + get_node_handle_callback, + get_instance_resource, + ) from system_manage.models.resource_mapping import ResourceType from application.models import Application from knowledge.models import KnowledgeWorkflow - from application.flow.tools import application_instance_field_call_dict, knowledge_instance_field_call_dict + from system_manage.services.resource_mapping import ( + application_instance_field_call_dict, + knowledge_instance_field_call_dict, + ) from application.models.application import ApplicationKnowledgeMapping from system_manage.models.resource_mapping import ResourceMapping + resource_mapping_list = [] - ids = list(Application.objects.values_list('id', flat=True)) + ids = list(Application.objects.values_list("id", flat=True)) for app_id in ids: try: application = Application.objects.get(id=app_id) - workflow_mapping = get_workflow_resource(application.work_flow, - get_node_handle_callback(ResourceType.APPLICATION, - application.id)) - instance_mapping = get_instance_resource(application, ResourceType.APPLICATION, str(application.id), - application_instance_field_call_dict) + workflow_mapping = get_workflow_resource( + application.work_flow, get_node_handle_callback(ResourceType.APPLICATION, application.id) + ) + instance_mapping = get_instance_resource( + application, ResourceType.APPLICATION, str(application.id), application_instance_field_call_dict + ) resource_mapping_list += workflow_mapping resource_mapping_list += instance_mapping except: pass - knowledge_ids = list(Knowledge.objects.values_list('id', flat=True)) + knowledge_ids = list(Knowledge.objects.values_list("id", flat=True)) for knowledge_id in knowledge_ids: try: knowledge = Knowledge.objects.get(id=knowledge_id) if knowledge.type == 4: knowledge_workflow = QuerySet(KnowledgeWorkflow).filter(knowledge_id=knowledge_id).first() if knowledge_workflow: - workflow_mapping = get_workflow_resource(knowledge_workflow.work_flow, - get_node_handle_callback(ResourceType.KNOWLEDGE, - str(knowledge_workflow.knowledge_id))) + workflow_mapping = get_workflow_resource( + knowledge_workflow.work_flow, + get_node_handle_callback(ResourceType.KNOWLEDGE, str(knowledge_workflow.knowledge_id)), + ) resource_mapping_list += workflow_mapping - instance_mapping = get_instance_resource(knowledge, ResourceType.KNOWLEDGE, str(knowledge.id), - knowledge_instance_field_call_dict) + instance_mapping = get_instance_resource( + knowledge, ResourceType.KNOWLEDGE, str(knowledge.id), knowledge_instance_field_call_dict + ) resource_mapping_list += instance_mapping except: pass application_knowledge_mapping = [ - ResourceMapping(source_type=ResourceType.APPLICATION, target_type=ResourceType.KNOWLEDGE, - source_id=str(akm.application_id), target_id=str(akm.knowledge_id)) for akm in - QuerySet(ApplicationKnowledgeMapping).all()] + ResourceMapping( + source_type=ResourceType.APPLICATION, + target_type=ResourceType.KNOWLEDGE, + source_id=str(akm.application_id), + target_id=str(akm.knowledge_id), + ) + for akm in QuerySet(ApplicationKnowledgeMapping).all() + ] resource_mapping_list += application_knowledge_mapping - return {(str(item.target_type) + str(item.target_id) + str(item.source_type) + str(item.source_id)): item for item - in resource_mapping_list}.values() + return { + (str(item.target_type) + str(item.target_id) + str(item.source_type) + str(item.source_id)): item + for item in resource_mapping_list + }.values() def resource_mapping(apps, schema_editor): from system_manage.models.resource_mapping import ResourceMapping + with ThreadPoolExecutor(max_workers=3) as executor: future = executor.submit(get_initialization_resource_mapping) resource_mapping_list = future.result() @@ -68,32 +86,49 @@ def resource_mapping(apps, schema_editor): class Migration(migrations.Migration): dependencies = [ - ('system_manage', '0004_alter_systemsetting_type_and_more'), - ('knowledge', '0007_remove_knowledgeworkflowversion_workflow_and_more'), - ('application', '0003_application_stt_model_params_setting_and_more'), + ("system_manage", "0004_alter_systemsetting_type_and_more"), + ("knowledge", "0007_remove_knowledgeworkflowversion_workflow_and_more"), + ("application", "0003_application_stt_model_params_setting_and_more"), ] operations = [ migrations.CreateModel( - name='ResourceMapping', + name="ResourceMapping", fields=[ - ('create_time', models.DateTimeField(auto_now_add=True, db_index=True, verbose_name='创建时间')), - ('update_time', models.DateTimeField(auto_now=True, db_index=True, verbose_name='修改时间')), - ('id', - models.UUIDField(default=uuid_utils.compat.uuid7, editable=False, primary_key=True, serialize=False, - verbose_name='主键id')), - ('source_type', models.CharField( - choices=[('KNOWLEDGE', '知识库'), ('APPLICATION', '应用'), ('TOOL', '工具'), ('MODEL', '模型')], - db_index=True, verbose_name='关联资源类型')), - ('target_type', models.CharField( - choices=[('KNOWLEDGE', '知识库'), ('APPLICATION', '应用'), ('TOOL', '工具'), ('MODEL', '模型')], - db_index=True, verbose_name='被关联资源类型')), - ('source_id', models.CharField(db_index=True, max_length=128, verbose_name='关联资源id')), - ('target_id', models.CharField(db_index=True, max_length=128, verbose_name='被关联资源id')), + ("create_time", models.DateTimeField(auto_now_add=True, db_index=True, verbose_name="创建时间")), + ("update_time", models.DateTimeField(auto_now=True, db_index=True, verbose_name="修改时间")), + ( + "id", + models.UUIDField( + default=uuid_utils.compat.uuid7, + editable=False, + primary_key=True, + serialize=False, + verbose_name="主键id", + ), + ), + ( + "source_type", + models.CharField( + choices=[("KNOWLEDGE", "知识库"), ("APPLICATION", "应用"), ("TOOL", "工具"), ("MODEL", "模型")], + db_index=True, + verbose_name="关联资源类型", + ), + ), + ( + "target_type", + models.CharField( + choices=[("KNOWLEDGE", "知识库"), ("APPLICATION", "应用"), ("TOOL", "工具"), ("MODEL", "模型")], + db_index=True, + verbose_name="被关联资源类型", + ), + ), + ("source_id", models.CharField(db_index=True, max_length=128, verbose_name="关联资源id")), + ("target_id", models.CharField(db_index=True, max_length=128, verbose_name="被关联资源id")), ], options={ - 'db_table': 'resource_mapping', + "db_table": "resource_mapping", }, ), - migrations.RunPython(resource_mapping, atomic=False) + migrations.RunPython(resource_mapping, atomic=False), ] diff --git a/apps/system_manage/services/__init__.py b/apps/system_manage/services/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/apps/system_manage/services/resource_mapping.py b/apps/system_manage/services/resource_mapping.py new file mode 100644 index 00000000000..4462b489fc9 --- /dev/null +++ b/apps/system_manage/services/resource_mapping.py @@ -0,0 +1,200 @@ +# coding=utf-8 +""" +@project: MaxKB +@Author: 虎虎虎 +@file: resource_mapping.py +@desc: 资源映射工具集(唯一来源)。 + +这些函数只解析 work_flow 的 JSON 结构(nodes/properties/node_data)或实例字段, +据此推导它引用的目标资源并登记到 system_manage.ResourceMapping。与执行引擎无关, +故归属于 ResourceMapping 所在的 system_manage,不放在工作流引擎(flow / workflow)里。 +""" + +from functools import reduce + +from django.db.models import QuerySet + +from tools.models import Tool, ToolScope, ToolType, ToolWorkflow + +# 节点类型 -> 其引用的目标资源 id 提取函数(按资源类型分组),供资源映射登记使用 +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 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): + _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): + 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 diff --git a/apps/tools/serializers/tool.py b/apps/tools/serializers/tool.py index 042490d9657..76f63da563b 100644 --- a/apps/tools/serializers/tool.py +++ b/apps/tools/serializers/tool.py @@ -493,7 +493,7 @@ def insert(self, instance, with_valid=True): try: download_callback_url = template_instance.get("downloadCallbackUrl", "") if not validate_trusted_url(download_callback_url, ALLOWED_CALLBACK_HOSTS): - raise AppApiException(500, _("Illegal download callback url")) + raise AppApiException(500, _("Illegal download callback url")) requests.get(download_callback_url, timeout=5, allow_redirects=False) except Exception as e: maxkb_logger.error(f"callback appstore tool download error: {e}") @@ -584,7 +584,7 @@ def debug(self, debug_instance): ] } # 合并初始化参数（默认值 → 已保存的启动参数 → 运行时入参） - init_params_default_value = {i["field"]: i.get('default_value') for i in init_field_list} + init_params_default_value = {i["field"]: i.get("default_value") for i in init_field_list} # 合并初始化参数 if init_params is not None: all_params = init_params_default_value | init_params | params @@ -649,9 +649,7 @@ def edit(self, instance, with_valid=True): if instance.get("tool_type") == ToolType.MCP: ToolExecutor().validate_mcp_transport(instance.get("code", "")) - if not QuerySet(Tool).filter( - id=self.data.get("id"), workspace_id=self.data.get("workspace_id") - ).exists(): + if not QuerySet(Tool).filter(id=self.data.get("id"), workspace_id=self.data.get("workspace_id")).exists(): raise serializers.ValidationError(_("Tool not found")) edit_field_list = [ @@ -671,11 +669,9 @@ def edit(self, instance, with_valid=True): if (field in instance and instance.get(field) is not None) } - tool = QuerySet(Tool).filter( - id=self.data.get("id"), workspace_id=self.data.get("workspace_id") - ).first() + tool = QuerySet(Tool).filter(id=self.data.get("id"), workspace_id=self.data.get("workspace_id")).first() if "init_params" in edit_dict: - if edit_dict["init_field_list"] is not None: + if edit_dict.get("init_field_list") is not None: rm_key = [] for key in edit_dict["init_params"]: if key not in [field["field"] for field in edit_dict["init_field_list"]]: @@ -690,9 +686,9 @@ def edit(self, instance, with_valid=True): edit_dict["init_params"] = rsa_long_encrypt(json.dumps(edit_dict["init_params"])) edit_dict["update_time"] = timezone.now() - QuerySet(Tool).filter( - id=self.data.get("id"), workspace_id=self.data.get("workspace_id") - ).update(**edit_dict) + QuerySet(Tool).filter(id=self.data.get("id"), workspace_id=self.data.get("workspace_id")).update( + **edit_dict + ) if "is_active" in instance: QuerySet(TriggerTask).filter(source_type="TOOL", source_id=self.data.get("id")).update( is_active=instance.get("is_active") @@ -714,9 +710,7 @@ def delete(self): from trigger.serializers.trigger import TriggerModelSerializer self.is_valid(raise_exception=True) - tool = QuerySet(Tool).filter( - id=self.data.get("id"), workspace_id=self.data.get("workspace_id") - ).first() + tool = QuerySet(Tool).filter(id=self.data.get("id"), workspace_id=self.data.get("workspace_id")).first() if tool is None: raise serializers.ValidationError(_("Tool not found")) if tool.template_id is None and tool.icon != "": @@ -724,9 +718,7 @@ def delete(self): if tool.tool_type == ToolType.SKILL: QuerySet(File).filter(id=tool.code).delete() QuerySet(WorkspaceUserResourcePermission).filter(target=tool.id).delete() - QuerySet(Tool).filter( - id=self.data.get("id"), workspace_id=self.data.get("workspace_id") - ).delete() + QuerySet(Tool).filter(id=self.data.get("id"), workspace_id=self.data.get("workspace_id")).delete() ResourceMapping.objects.filter(Q(target_id=self.data.get("id")) | Q(source_id=self.data.get("id"))).delete() QuerySet(ToolRecord).filter(tool_id=self.data.get("id")).delete() trigger_ids = list( @@ -735,9 +727,7 @@ def delete(self): .values("trigger_id") .distinct() ) - QuerySet(TriggerTask).filter( - source_type="TOOL", source_id=self.data.get("id") - ).delete() + QuerySet(TriggerTask).filter(source_type="TOOL", source_id=self.data.get("id")).delete() for trigger_id in trigger_ids: trigger = Trigger.objects.filter(id=trigger_id["trigger_id"]).first() if trigger and trigger.is_active: @@ -783,7 +773,7 @@ def one(self): } def get_child_tool_list(self, work_flow, response): - 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(work_flow, False) tool_id_list = [ @@ -889,9 +879,9 @@ def run(self, instance, is_valid=True): { "type": "error" if ( - item.get("code") == "E999" - or str(item.get("code") or "").startswith("E9") - or item.get("code") in ["F821", "F822", "F823"] + item.get("code") == "E999" + or str(item.get("code") or "").startswith("E9") + or item.get("code") in ["F821", "F822", "F823"] ) else "warning", "module": "", @@ -1017,7 +1007,7 @@ def import_workflow_tools(self, tool, workspace_id, user_id, folder_id, new_chil {**tool, "id": update_tool_map.get(tool.get("id"))} for tool in tool_list if not exits_tool_id_list.__contains__(tool.get("id")) - and not exits_tool_id_list.__contains__( + and not exits_tool_id_list.__contains__( new_uuid.generate_uuid(tool.get("id")) if new_child_policy == 2 else generate_uuid((tool.get("id") + workspace_id or "")) @@ -1229,9 +1219,7 @@ def add(self, instance, with_valid=True): self.is_valid(raise_exception=True) AddInternalToolRequest(data=instance).is_valid(raise_exception=True) - internal_tool = QuerySet(Tool).filter( - id=self.data.get("tool_id"), scope=ToolScope.INTERNAL - ).first() + internal_tool = QuerySet(Tool).filter(id=self.data.get("tool_id"), scope=ToolScope.INTERNAL).first() if internal_tool is None: raise AppApiException(500, _("Tool does not exist")) @@ -1403,9 +1391,9 @@ def update_tool(self, with_valid=True): self.is_valid(raise_exception=True) if not validate_trusted_url(self.data.get("download_url"), ALLOWED_DOWNLOAD_HOSTS): raise AppApiException(500, _("Illegal download url")) - tool = QuerySet(Tool).filter( - id=self.data.get("tool_id"), workspace_id=self.data.get("workspace_id") - ).first() + tool = ( + QuerySet(Tool).filter(id=self.data.get("tool_id"), workspace_id=self.data.get("workspace_id")).first() + ) if tool is None: raise AppApiException(500, _("Tool does not exist")) # 查找匹配的版本名称 @@ -1637,15 +1625,15 @@ def process(): ) try: for r in model.stream( - [ - # SystemMessage(content=SYSTEM_ROLE), - *[ - HumanMessage(content=m.get("content")) - if m.get("role") == "user" - else AIMessage(content=m.get("content")) - for m in messages - ] + [ + # SystemMessage(content=SYSTEM_ROLE), + *[ + HumanMessage(content=m.get("content")) + if m.get("role") == "user" + else AIMessage(content=m.get("content")) + for m in messages ] + ] ): yield "data: " + json.dumps({"content": r.content}) + "\n\n" except Exception as e: @@ -1809,8 +1797,9 @@ def is_x_pack_ee(): def page_tool_with_folders(self, current_page: int, page_size: int): self.is_valid(raise_exception=True) - workspace_manage = is_workspace_manage_permission_read(self.data.get("user_id"), - self.data.get("workspace_id"), 'TOOL:READ') + workspace_manage = is_workspace_manage_permission_read( + self.data.get("user_id"), self.data.get("workspace_id"), "TOOL:READ" + ) is_x_pack_ee = self.is_x_pack_ee() result = native_page_search( current_page, @@ -1838,8 +1827,9 @@ def page_tool_with_folders(self, current_page: int, page_size: int): def get_tools(self): self.is_valid(raise_exception=True) - workspace_manage = is_workspace_manage_permission_read(self.data.get("user_id"), - self.data.get("workspace_id"), 'TOOL:READ') + workspace_manage = is_workspace_manage_permission_read( + self.data.get("user_id"), self.data.get("workspace_id"), "TOOL:READ" + ) is_x_pack_ee = self.is_x_pack_ee() results = native_search( self.get_query_set(workspace_manage, is_x_pack_ee), diff --git a/apps/tools/serializers/tool_workflow.py b/apps/tools/serializers/tool_workflow.py index 2862f0b3f62..71f221b6f96 100644 --- a/apps/tools/serializers/tool_workflow.py +++ b/apps/tools/serializers/tool_workflow.py @@ -622,7 +622,7 @@ def get_appstore_templates(self): def update_resource_mapping_by_tool(tool_id: str, other_resource_mapping=None): - from application.flow.tools import get_instance_resource, save_workflow_mapping + from system_manage.services.resource_mapping import get_instance_resource, save_workflow_mapping from system_manage.models.resource_mapping import ResourceType if other_resource_mapping is None: diff --git a/ui/src/api/admin/workspace/knowledge/workflow.ts b/ui/src/api/admin/workspace/knowledge/workflow.ts index 00c81d32930..cead35d652a 100644 --- a/ui/src/api/admin/workspace/knowledge/workflow.ts +++ b/ui/src/api/admin/workspace/knowledge/workflow.ts @@ -1,6 +1,15 @@ import type LogicFlow from '@logicflow/core' -import { put } from '../../core/request' -import type { DefaultModelSettingPayload, KnowledgeWorkflowDetail } from '@/api/types' +import { get, post, put } from '../../core/request' +import type { LoadingTarget } from '../../core/types' +import type { + DefaultModelSettingPayload, + Dict, + KnowledgeItem, + KnowledgeType, + KnowledgeWorkflowAction, + KnowledgeWorkflowDebugPayload, + KnowledgeWorkflowDetail, +} from '@/api/types' import { getWorkspaceId } from '@/utils/resource-context' interface KnowledgeWorkflowPayload { @@ -8,11 +17,24 @@ interface KnowledgeWorkflowPayload { work_flow: LogicFlow.GraphConfigData } +interface CreateKnowledgeWorkflowPayload { + name: string + desc?: string + folder_id: string + type: KnowledgeType + work_flow: LogicFlow.GraphConfigData +} + const getPrefix = () => { const workspaceId = getWorkspaceId() return `/workspace/${workspaceId}/knowledge` } +/** 创建工作流知识库。 */ +const postKnowledgeWorkflow = (payload: CreateKnowledgeWorkflowPayload) => { + return post(`${getPrefix()}/workflow`, payload) +} + /** 保存知识库工作流。 */ const putKnowledgeWorkflow = (knowledgeId: string, payload: KnowledgeWorkflowPayload) => { return put(`${getPrefix()}/${knowledgeId}/workflow`, payload) @@ -23,4 +45,48 @@ const putKnowledgeWorkflowPublish = (knowledgeId: string) => { return put(`${getPrefix()}/${knowledgeId}/publish`) } -export default { putKnowledgeWorkflow, putKnowledgeWorkflowPublish } +/** 上传知识库调试文件，返回文件访问地址（末段为 file_id）。 */ +const postKnowledgeUploadFile = (knowledgeId: string, file: File, loading?: LoadingTarget) => { + const payload = new FormData() + payload.append('file', file) + payload.append('source_id', knowledgeId) + payload.append('source_type', 'KNOWLEDGE') + return post('/oss/file', payload, undefined, loading) +} + +/** 获取数据源节点的动态表单配置。 */ +const getKnowledgeWorkflowFormList = ( + knowledgeId: string, + type: 'local' | 'tool', + id: string, + node: Dict, + loading?: LoadingTarget, +) => { + return post<{ node: Dict }, Dict[]>(`${getPrefix()}/${knowledgeId}/datasource/${type}/${id}/form_list`, { node }, undefined, loading) +} + +/** 提交知识库工作流调试任务。 */ +const postKnowledgeWorkflowDebug = (knowledgeId: string, payload: KnowledgeWorkflowDebugPayload, loading?: LoadingTarget) => { + return post(`${getPrefix()}/${knowledgeId}/debug`, payload, undefined, loading) +} + +/** 轮询知识库工作流调试任务详情。 */ +const getKnowledgeWorkflowAction = (knowledgeId: string, actionId: string, loading?: LoadingTarget) => { + return get(`${getPrefix()}/${knowledgeId}/action/${actionId}`, undefined, loading) +} + +/** 取消知识库工作流调试任务。 */ +const postCancelKnowledgeWorkflowAction = (knowledgeId: string, actionId: string, loading?: LoadingTarget) => { + return post(`${getPrefix()}/${knowledgeId}/action/${actionId}/cancel`, undefined, undefined, loading) +} + +export default { + postKnowledgeWorkflow, + putKnowledgeWorkflow, + putKnowledgeWorkflowPublish, + postKnowledgeUploadFile, + getKnowledgeWorkflowFormList, + postKnowledgeWorkflowDebug, + getKnowledgeWorkflowAction, + postCancelKnowledgeWorkflowAction, +} diff --git a/ui/src/api/types/knowledge.ts b/ui/src/api/types/knowledge.ts index 6628533a252..ca8bf55ff93 100644 --- a/ui/src/api/types/knowledge.ts +++ b/ui/src/api/types/knowledge.ts @@ -42,6 +42,24 @@ export interface KnowledgeTagGroup { values: { id: string; value: string; create_time: string; update_time: string }[] } +/** 知识库工作流调试任务状态。 */ +export type KnowledgeWorkflowActionState = 'STARTED' | 'PENDING' | 'SUCCESS' | 'FAILURE' | 'REVOKE' | 'REVOKED' + +/** 知识库工作流调试任务，details 按节点 id 聚合各节点执行详情。 */ +export interface KnowledgeWorkflowAction { + id: string + knowledge_id: string + state: KnowledgeWorkflowActionState + details: Record> + meta?: Record +} + +/** 知识库工作流调试提交参数。 */ +export interface KnowledgeWorkflowDebugPayload { + data_source: Record + knowledge_base: Record +} + /** 知识库工作流详情。 */ export interface KnowledgeWorkflowDetail { id: string diff --git a/ui/src/components/mk-dynamics-form/items/tree/Tree.vue b/ui/src/components/mk-dynamics-form/items/tree/Tree.vue index dba17757b70..eb06b1ed603 100644 --- a/ui/src/components/mk-dynamics-form/items/tree/Tree.vue +++ b/ui/src/components/mk-dynamics-form/items/tree/Tree.vue @@ -68,8 +68,8 @@ const loadNode: LoadFunction = (node, resolve) => { ), body: { current_node: node.level === 0 ? undefined : node.data }, then: (res: DynamicFormValue) => { - resolve(res.data) - res.data.forEach((childNode: DynamicFormValue) => { + resolve(res) + res.forEach((childNode: DynamicFormValue) => { if (childNode.is_exist) { treeRef.value?.setChecked(childNode.token, true, false) } diff --git a/ui/src/views/knowledge/KnowledgeView.vue b/ui/src/views/knowledge/KnowledgeView.vue index d2c8900a389..f1cc3b2df57 100644 --- a/ui/src/views/knowledge/KnowledgeView.vue +++ b/ui/src/views/knowledge/KnowledgeView.vue @@ -11,6 +11,7 @@ import { RESOURCE_TYPE } from '@/api/enums' import { FOLDER_ENTRIES, FOLDER_ENTRY_ID } from '@/constants' import FolderTree from '@/components/business/folder-tree/index.vue' import KnowledgeCard from './knowledge-card/KnowledgeCard.vue' +import CreateWorkflowKnowledgeDialog from './create-knowledge/CreateWorkflowKnowledgeDialog.vue' import { AuthorizeKnowledgeAction, DeleteKnowledgeAction, @@ -49,6 +50,13 @@ function handleCreateFolder() { folderTreeRef.value?.openCreate() } +/* 创建知识库 */ +const createFolderId = computed(() => currentFolder.value.id || FOLDER_ENTRY_ID.ALL) +const createWorkflowKnowledgeDialogRef = useTemplateRef>('createWorkflowKnowledgeDialogRef') +function handleCreateWorkflowKnowledge() { + createWorkflowKnowledgeDialogRef.value?.open() +} + /* 知识库查询 */ const knowledgeData = ref([]) const infiniteScrollRef = useTemplateRef<{ reset: () => Promise }>('infiniteScrollRef') @@ -201,7 +209,7 @@ function handleBatchDelete() { 通用知识库 Web 站点知识库 - 工作流知识库 + 工作流知识库 导入创建 @@ -279,4 +287,6 @@ function handleBatchDelete() { + + diff --git a/ui/src/views/knowledge/create-knowledge/CreateWorkflowKnowledgeDialog.vue b/ui/src/views/knowledge/create-knowledge/CreateWorkflowKnowledgeDialog.vue new file mode 100644 index 00000000000..d2261f4a755 --- /dev/null +++ b/ui/src/views/knowledge/create-knowledge/CreateWorkflowKnowledgeDialog.vue @@ -0,0 +1,109 @@ + + + diff --git a/ui/src/views/knowledge/template.ts b/ui/src/views/knowledge/template.ts index b6d34d9544d..10de376f521 100644 --- a/ui/src/views/knowledge/template.ts +++ b/ui/src/views/knowledge/template.ts @@ -1,8 +1,8 @@ import type LogicFlow from '@logicflow/core' -import { defaultNodes } from '@/workflow-canvas/config/node-mapping' +import { defaultApplicationNodes } from '@/workflow-canvas/config/node-mapping' export const applicationTemplate = { - blank: { edges: [], nodes: defaultNodes }, + blank: { edges: [], nodes: defaultApplicationNodes }, assistant: { nodes: [ { diff --git a/ui/src/views/workflow/knowledge/KnowledgeWorkflowView.vue b/ui/src/views/workflow/knowledge/KnowledgeWorkflowView.vue index 5d068c70265..6be63ee02ab 100644 --- a/ui/src/views/workflow/knowledge/KnowledgeWorkflowView.vue +++ b/ui/src/views/workflow/knowledge/KnowledgeWorkflowView.vue @@ -14,6 +14,7 @@ import { defaultKnowledgeNodes } from '@/workflow-canvas/config/node-mapping' import { WorkflowMode } from '@/workflow-canvas/types' import WorkflowViewLayout from '../components/WorkflowViewLayout.vue' import DefaultModelSettingButton from '../components/default-model-setting/DefaultModelSettingButton.vue' +import DebugDrawer from './debug/DebugDrawer.vue' defineOptions({ name: 'KnowledgeWorkflowView' }) @@ -94,6 +95,24 @@ function handleSave() { return saveKnowledgeWorkflow(undefined, true).catch(() => {}) } +/* 调试：先落库未保存的画布改动，保证调试命中最新工作流 */ +const debugDrawerRef = useTemplateRef>('debugDrawerRef') + +function openDebug() { + const graphData = getGraphData() + if (graphData) debugDrawerRef.value?.open(graphData) +} + +function handleDebug() { + if (hasUnsavedChanges()) { + saveKnowledgeWorkflow(undefined, false) + .then(() => openDebug()) + .catch(() => {}) + return + } + openDebug() +} + function handlePublish() { if (!workflowRef.value) return @@ -180,6 +199,8 @@ onMounted(() => { /> 保存 + + 调试 发布 @@ -191,5 +212,8 @@ onMounted(() => { :loop-workflow-mode="WorkflowMode.KnowledgeLoop" :workflow-mode="WorkflowMode.Knowledge" /> + + + diff --git a/ui/src/views/workflow/knowledge/debug/DebugDrawer.vue b/ui/src/views/workflow/knowledge/debug/DebugDrawer.vue new file mode 100644 index 00000000000..daef56b0560 --- /dev/null +++ b/ui/src/views/workflow/knowledge/debug/DebugDrawer.vue @@ -0,0 +1,130 @@ + + + diff --git a/ui/src/views/workflow/knowledge/debug/action/DataSource.vue b/ui/src/views/workflow/knowledge/debug/action/DataSource.vue new file mode 100644 index 00000000000..d7d92e89270 --- /dev/null +++ b/ui/src/views/workflow/knowledge/debug/action/DataSource.vue @@ -0,0 +1,129 @@ + + + diff --git a/ui/src/views/workflow/knowledge/debug/action/KnowledgeBase.vue b/ui/src/views/workflow/knowledge/debug/action/KnowledgeBase.vue new file mode 100644 index 00000000000..fc4a81b607a --- /dev/null +++ b/ui/src/views/workflow/knowledge/debug/action/KnowledgeBase.vue @@ -0,0 +1,44 @@ + + + diff --git a/ui/src/views/workflow/knowledge/debug/action/Result.vue b/ui/src/views/workflow/knowledge/debug/action/Result.vue new file mode 100644 index 00000000000..6e2b58aa67a --- /dev/null +++ b/ui/src/views/workflow/knowledge/debug/action/Result.vue @@ -0,0 +1,65 @@ + + + diff --git a/ui/src/workflow-canvas/details/index.vue b/ui/src/workflow-canvas/details/index.vue index 47e12c53261..e562bd1fb04 100644 --- a/ui/src/workflow-canvas/details/index.vue +++ b/ui/src/workflow-canvas/details/index.vue @@ -35,9 +35,28 @@ function nodeDetailComponent(type?: WorkflowNodeType): Component | null { return type ? (nodeDetailComponentMap.get(type) ?? null) : null } -// 按 index 升序展示各节点执行详情。 -const sortedDetail = computed(() => - [...props.detail].sort((a, b) => ((a?.index as number) || 0) - ((b?.index as number) || 0)), +// 新引擎节点详情的 status 为枚举字符串，而 v3 详情契约（types/DetailContainer/BaseHeader）统一使用 +// v2 的数字状态码（200 成功 / 202 运行中 / 其余失败）。在详情进入 v3 渲染体系的唯一入口做一次适配， +// 已是数字（如历史数据或预览 mock）则原样透传。嵌套子节点会在递归渲染时经本入口再次适配。 +const STATUS_CODE_MAP: Record = { + SUCCESS: 200, + RUNNING: 202, + BEFORE_RUNNING: 202, + CANCELLED: 201, + FAIL: 500, +} + +function coerceStatus(status: unknown): number | undefined { + if (typeof status === 'number') return status + if (typeof status === 'string') return STATUS_CODE_MAP[status] ?? 500 + return status as undefined +} + +// 按 index 升序展示各节点执行详情，并把枚举状态适配为数字状态码。 +const sortedDetail = computed(() => + [...props.detail] + .sort((a, b) => ((a?.index as number) || 0) - ((b?.index as number) || 0)) + .map((item) => ({ ...item, status: coerceStatus(item.status) })), ) diff --git a/ui/src/workflow-canvas/nodes/data-source-local-node/details/index.vue b/ui/src/workflow-canvas/nodes/data-source-local-node/details/index.vue new file mode 100644 index 00000000000..326f28b010b --- /dev/null +++ b/ui/src/workflow-canvas/nodes/data-source-local-node/details/index.vue @@ -0,0 +1,55 @@ + + + diff --git a/ui/src/workflow-canvas/nodes/data-source-local-node/index.ts b/ui/src/workflow-canvas/nodes/data-source-local-node/index.ts index 6fc020a366c..5ef3b0abf9f 100644 --- a/ui/src/workflow-canvas/nodes/data-source-local-node/index.ts +++ b/ui/src/workflow-canvas/nodes/data-source-local-node/index.ts @@ -1,4 +1,5 @@ import DataSourceLocalNode from './index.vue' +import DataSourceLocalNodeDetail from './details/index.vue' import { WorkflowNodeModel, WorkflowNodeView } from '@/workflow-canvas/core/workflow-node' import { WorkflowNodeType } from '@/workflow-canvas/types' @@ -8,4 +9,9 @@ class DataSourceLocalNodeView extends WorkflowNodeView { } } -export default { type: WorkflowNodeType.DataSourceLocalNode, model: WorkflowNodeModel, view: DataSourceLocalNodeView } +export default { + type: WorkflowNodeType.DataSourceLocalNode, + model: WorkflowNodeModel, + view: DataSourceLocalNodeView, + details: DataSourceLocalNodeDetail, +} diff --git a/ui/src/workflow-canvas/nodes/document-split-node/details/index.vue b/ui/src/workflow-canvas/nodes/document-split-node/details/index.vue new file mode 100644 index 00000000000..1d5cf537c93 --- /dev/null +++ b/ui/src/workflow-canvas/nodes/document-split-node/details/index.vue @@ -0,0 +1,98 @@ + + + + + diff --git a/ui/src/workflow-canvas/nodes/document-split-node/index.ts b/ui/src/workflow-canvas/nodes/document-split-node/index.ts index 3c28e4f2831..f0035a85e35 100644 --- a/ui/src/workflow-canvas/nodes/document-split-node/index.ts +++ b/ui/src/workflow-canvas/nodes/document-split-node/index.ts @@ -1,4 +1,5 @@ import DocumentSplitNode from './index.vue' +import DocumentSplitNodeDetail from './details/index.vue' import { WorkflowNodeModel, WorkflowNodeView } from '@/workflow-canvas/core/workflow-node' import { WorkflowNodeType } from '@/workflow-canvas/types' @@ -15,4 +16,9 @@ class DocumentSplitNodeModel extends WorkflowNodeModel { } } -export default { type: WorkflowNodeType.DocumentSplitNode, model: DocumentSplitNodeModel, view: DocumentSplitNodeView } +export default { + type: WorkflowNodeType.DocumentSplitNode, + model: DocumentSplitNodeModel, + view: DocumentSplitNodeView, + details: DocumentSplitNodeDetail, +} diff --git a/ui/src/workflow-canvas/nodes/knowledge-write-node/details/index.vue b/ui/src/workflow-canvas/nodes/knowledge-write-node/details/index.vue new file mode 100644 index 00000000000..5e09e03a566 --- /dev/null +++ b/ui/src/workflow-canvas/nodes/knowledge-write-node/details/index.vue @@ -0,0 +1,77 @@ + + + + + diff --git a/ui/src/workflow-canvas/nodes/knowledge-write-node/index.ts b/ui/src/workflow-canvas/nodes/knowledge-write-node/index.ts index 84f815f374b..ae86ca33a8c 100644 --- a/ui/src/workflow-canvas/nodes/knowledge-write-node/index.ts +++ b/ui/src/workflow-canvas/nodes/knowledge-write-node/index.ts @@ -1,4 +1,5 @@ import KnowledgeWriteNode from './index.vue' +import KnowledgeWriteNodeDetail from './details/index.vue' import { WorkflowNodeModel, WorkflowNodeView } from '@/workflow-canvas/core/workflow-node' import { WorkflowNodeType } from '@/workflow-canvas/types' @@ -8,4 +9,9 @@ class KnowledgeWriteNodeView extends WorkflowNodeView { } } -export default { type: WorkflowNodeType.KnowledgeWriteNode, model: WorkflowNodeModel, view: KnowledgeWriteNodeView } +export default { + type: WorkflowNodeType.KnowledgeWriteNode, + model: WorkflowNodeModel, + view: KnowledgeWriteNodeView, + details: KnowledgeWriteNodeDetail, +}