fix: tool workflow bugs (#4962)

Author: shaohuzhang1

apps/application/flow/i_step_node.py

@@ -150,7 +150,12 @@ def handler(self, workflow):
                             source_type=self.chat_info.source_type,
                             source_id=self.chat_info.source_id,
                             state=state,
+                            run_time=time.time() - workflow.context.get('start_time') if workflow.context.get(
+                                'start_time') is not None else 0,
                             meta={
+                                'input_field_list': workflow.get_input_field_list(),
+                                'output_field_list': workflow.get_output_field_list(),
+                                'input': workflow.get_input(),
                                 'output': workflow.out_context,
                                 'details': workflow.get_runtime_details(),
                                 'answer_text_list': workflow.get_answer_text_list()

apps/application/flow/step_node/ai_chat_step_node/impl/base_chat_node.py

@@ -13,10 +13,15 @@
 from functools import reduce
 from typing import List, Dict
 
+import uuid_utils.compat as uuid
+from django.db.models import QuerySet, OuterRef, Subquery
+from django.utils.translation import gettext as _
+from langchain_core.messages import BaseMessage, AIMessage, HumanMessage, SystemMessage
 from langchain_core.tools import StructuredTool
+from pydantic import Field, create_model
 
 from application.flow.common import Workflow, WorkflowMode
-from application.flow.i_step_node import NodeResult, INode, ToolWorkflowPostHandler, ToolWorkflowCallPostHandler
+from application.flow.i_step_node import NodeResult, INode, ToolWorkflowPostHandler
 from application.flow.step_node.ai_chat_step_node.i_chat_node import IChatNode
 from application.flow.tools import Reasoning, mcp_response_generator
 from application.models import Application, ApplicationApiKey, ApplicationAccessToken
@@ -25,14 +30,9 @@
 from common.utils.rsa_util import rsa_long_decrypt
 from common.utils.shared_resource_auth import filter_authorized_ids
 from common.utils.tool_code import ToolExecutor
-from django.db.models import QuerySet, OuterRef, Subquery
-from django.utils.translation import gettext as _
-from langchain_core.messages import BaseMessage, AIMessage, HumanMessage, SystemMessage
 from models_provider.models import Model
 from models_provider.tools import get_model_credential, get_model_instance_by_model_workspace_id
 from tools.models import Tool, ToolWorkflowVersion, ToolType
-from pydantic import BaseModel, Field, create_model
-import uuid_utils.compat as uuid
 
 
 def build_schema(fields: dict):
@@ -66,14 +66,14 @@ def get_workflow_args(tool, qv):
     return build_schema({})
 
 
-def get_workflow_func(tool, qv, workspace_id):
+def get_workflow_func(node, tool, qv, workspace_id):
     tool_id = tool.id
     tool_record_id = str(uuid.uuid7())
     took_execute = ToolExecute(tool_id, tool_record_id,
                                workspace_id,
-                               None,
-                               None,
-                               True)
+                               node.workflow_manage.get_source_type(),
+                               node.workflow_manage.get_source_id(),
+                               False)
 
     def inner(**kwargs):
         from application.flow.tool_workflow_manage import ToolWorkflowManage
@@ -86,7 +86,7 @@ def inner(**kwargs):
                 'workspace_id': workspace_id,
                 **kwargs},
 
-            ToolWorkflowCallPostHandler(took_execute, tool_id),
+            ToolWorkflowPostHandler(took_execute, tool_id),
             is_the_task_interrupted=lambda: False,
             child_node=None,
             start_node_id=None,
@@ -101,7 +101,7 @@ def inner(**kwargs):
     return inner
 
 
-def get_tools(tool_workflow_ids, workspace_id):
+def get_tools(node, tool_workflow_ids, workspace_id):
     tools = QuerySet(Tool).filter(id__in=tool_workflow_ids, tool_type=ToolType.WORKFLOW, workspace_id=workspace_id)
     latest_subquery = ToolWorkflowVersion.objects.filter(
         tool_id=OuterRef('tool_id')
@@ -115,7 +115,7 @@ def get_tools(tool_workflow_ids, workspace_id):
     results = []
     for tool in tools:
         qv = qd.get(tool.id)
-        func = get_workflow_func(tool, qv, workspace_id)
+        func = get_workflow_func(node, tool, qv, workspace_id)
         args = get_workflow_args(tool, qv)
         tool = StructuredTool.from_function(
             func=func,
@@ -360,7 +360,7 @@ def _handle_mcp_request(self, mcp_source, mcp_servers, mcp_tool_id, mcp_tool_ids
                     mcp_servers_config = {**mcp_servers_config, **json.loads(mcp_tool['code'])}
                     mcp_servers_config = self.handle_variables(mcp_servers_config)
         tool_init_params = {}
-        tools = get_tools(tool_ids, workspace_id)
+        tools = get_tools(self, tool_ids, workspace_id)
         if tool_ids and len(tool_ids) > 0:  # 如果有工具ID，则将其转换为MCP
             self.context['tool_ids'] = tool_ids
             for tool_id in tool_ids:

apps/application/flow/step_node/image_generate_step_node/impl/base_image_generate_node.py

@@ -10,8 +10,8 @@
 from application.flow.step_node.image_generate_step_node.i_image_generate_node import IImageGenerateNode
 from common.utils.common import bytes_to_uploaded_file
 from knowledge.models import FileSourceType
-from oss.serializers.file import FileSerializer
 from models_provider.tools import get_model_instance_by_model_workspace_id
+from oss.serializers.file import FileSerializer
 
 
 class BaseImageGenerateNode(IImageGenerateNode):
@@ -117,6 +117,8 @@ def upload_file(self, file):
         if [WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP].__contains__(
                 self.workflow_manage.flow.workflow_mode):
             return self.upload_knowledge_file(file)
+        if [WorkflowMode.TOOL, WorkflowMode.TOOL_LOOP].__contains__(self.workflow_manage.flow.workflow_mode):
+            return self.upload_tool_file(file)
         return self.upload_application_file(file)
 
     def upload_knowledge_file(self, file):
@@ -133,6 +135,20 @@ def upload_knowledge_file(self, file):
         }).upload()
         return file_url
 
+    def upload_tool_file(self, file):
+        tool_id = self.workflow_params.get('tool_id')
+        meta = {
+            'debug': False,
+            'tool_id': tool_id,
+        }
+        file_url = FileSerializer(data={
+            'file': file,
+            'meta': meta,
+            'source_id': tool_id,
+            'source_type': FileSourceType.TOOL.value
+        }).upload()
+        return file_url
+
     def upload_application_file(self, file):
         application = self.workflow_manage.work_flow_post_handler.chat_info.application
         chat_id = self.workflow_params.get('chat_id')

apps/application/flow/step_node/image_to_video_step_node/impl/base_image_to_video_node.py

@@ -94,6 +94,8 @@ def upload_file(self, file):
         if [WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP].__contains__(
                 self.workflow_manage.flow.workflow_mode):
             return self.upload_knowledge_file(file)
+        if [WorkflowMode.TOOL, WorkflowMode.TOOL_LOOP].__contains__(self.workflow_manage.flow.workflow_mode):
+            return self.upload_tool_file(file)
         return self.upload_application_file(file)
 
     def upload_knowledge_file(self, file):
@@ -110,6 +112,20 @@ def upload_knowledge_file(self, file):
         }).upload()
         return file_url
 
+    def upload_tool_file(self, file):
+        tool_id = self.workflow_params.get('tool_id')
+        meta = {
+            'debug': False,
+            'tool_id': tool_id,
+        }
+        file_url = FileSerializer(data={
+            'file': file,
+            'meta': meta,
+            'source_id': tool_id,
+            'source_type': FileSourceType.TOOL.value
+        }).upload()
+        return file_url
+
     def upload_application_file(self, file):
         application = self.workflow_manage.work_flow_post_handler.chat_info.application
         chat_id = self.workflow_params.get('chat_id')

apps/application/flow/step_node/question_node/i_question_node.py

@@ -42,7 +42,12 @@ def get_node_params_serializer_class(self) -> Type[serializers.Serializer]:
         return QuestionNodeSerializer
 
     def _run(self):
-        return self.execute(**self.node_params_serializer.data, **self.flow_params_serializer.data)
+        if [WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP, WorkflowMode.TOOL,
+            WorkflowMode.TOOL_LOOP].__contains__(self.workflow_manage.flow.workflow_mode):
+            return self.execute(**self.node_params_serializer.data, **self.flow_params_serializer.data,
+                                **{'history_chat_record': [], 'stream': True, 'chat_id': None, 'chat_record_id': None})
+        else:
+            return self.execute(**self.node_params_serializer.data, **self.flow_params_serializer.data)
 
     def execute(self, model_id, system, prompt, dialogue_number, history_chat_record, stream, chat_id, chat_record_id,
                 model_params_setting=None, model_id_type=None, model_id_reference=None,

apps/application/flow/step_node/text_to_speech_step_node/impl/base_text_to_speech_node.py

@@ -111,6 +111,8 @@ def upload_file(self, file):
         if [WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP].__contains__(
                 self.workflow_manage.flow.workflow_mode):
             return self.upload_knowledge_file(file)
+        if [WorkflowMode.TOOL, WorkflowMode.TOOL_LOOP].__contains__(self.workflow_manage.flow.workflow_mode):
+            return self.upload_tool_file(file)
         return self.upload_application_file(file)
 
     def upload_knowledge_file(self, file):
@@ -127,6 +129,20 @@ def upload_knowledge_file(self, file):
         }).upload()
         return file_url
 
+    def upload_tool_file(self, file):
+        tool_id = self.workflow_params.get('tool_id')
+        meta = {
+            'debug': False,
+            'tool_id': tool_id,
+        }
+        file_url = FileSerializer(data={
+            'file': file,
+            'meta': meta,
+            'source_id': tool_id,
+            'source_type': FileSourceType.TOOL.value
+        }).upload()
+        return file_url
+
     def upload_application_file(self, file):
         application = self.workflow_manage.work_flow_post_handler.chat_info.application
         chat_id = self.workflow_params.get('chat_id')

apps/application/flow/step_node/text_to_video_step_node/impl/base_text_to_video_node.py

@@ -68,6 +68,8 @@ def upload_file(self, file):
         if [WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP].__contains__(
                 self.workflow_manage.flow.workflow_mode):
             return self.upload_knowledge_file(file)
+        if [WorkflowMode.TOOL, WorkflowMode.TOOL_LOOP].__contains__(self.workflow_manage.flow.workflow_mode):
+            return self.upload_tool_file(file)
         return self.upload_application_file(file)
 
     def upload_knowledge_file(self, file):
@@ -84,6 +86,20 @@ def upload_knowledge_file(self, file):
         }).upload()
         return file_url
 
+    def upload_tool_file(self, file):
+        tool_id = self.workflow_params.get('tool_id')
+        meta = {
+            'debug': False,
+            'tool_id': tool_id,
+        }
+        file_url = FileSerializer(data={
+            'file': file,
+            'meta': meta,
+            'source_id': tool_id,
+            'source_type': FileSourceType.TOOL.value
+        }).upload()
+        return file_url
+
     def upload_application_file(self, file):
         application = self.workflow_manage.work_flow_post_handler.chat_info.application
         chat_id = self.workflow_params.get('chat_id')

apps/application/flow/step_node/tool_start_node/impl/base_tool_start_node.py

@@ -34,7 +34,7 @@ def execute(self, **kwargs) -> NodeResult:
         global_value = {}
         params = self.workflow_manage.get_body()
         for item in base_node.properties.get('user_input_field_list', []):
-            global_value[item.get('field')] = params[item.get('field')]
+            global_value[item.get('field')] = params.get(item.get('field'))
 
         self.workflow_manage.out_context = {
             item.get('field'): None
@@ -48,7 +48,7 @@ def get_details(self, index: int, **kwargs):
         for field in self.node.properties.get('config')['globalFields']:
             key = field['value']
             global_fields.append({
-                'label': field['label'],
+                'label': field.get('label'),
                 'key': key,
                 'value': self.workflow_manage.context[key] if key in self.workflow_manage.context else ''
             })

apps/application/flow/tool_workflow_manage.py

@@ -6,6 +6,7 @@
     @date：2026/3/12 15:17
     @desc:
 """
+import time
 from concurrent.futures import ThreadPoolExecutor
 
 from django.db import close_old_connections
@@ -32,6 +33,14 @@ def __init__(self, flow: Workflow, params, work_flow_post_handler: WorkFlowPostH
     def get_params_serializer_class(self):
         return ToolFlowParamsSerializer
 
+    def run(self):
+        self.context['start_time'] = time.time()
+        close_old_connections()
+        language = get_language()
+        if self.params.get('stream'):
+            return self.run_stream(self.start_node, None, language)
+        return self.run_block(language)
+
     def stream(self):
         close_old_connections()
         language = get_language()
@@ -48,6 +57,30 @@ def get_base_node(self):
         """
         return self.flow.get_node('tool-base-node')
 
+    def get_input_field_list(self):
+        """
+        获取输入字段列表
+        @return: 输入字段配置
+        """
+        base_node = self.get_base_node()
+        return base_node.properties.get("user_input_field_list") or []
+
+    def get_output_field_list(self):
+        """
+        获取输出字段列表配置
+        @return:  输出字段列表配置
+        """
+        base_node = self.get_base_node()
+        return base_node.properties.get("user_output_field_list") or []
+
+    def get_input(self):
+        """
+        获取用户输入
+        @return: 用户输入
+        """
+        input_field_list = self.get_input_field_list()
+        return {f.get('field'): self.params.get(f.get('field')) for f in input_field_list}
+
     def get_source_type(self):
         return "TOOL"
 

apps/application/serializers/common.py

@@ -78,6 +78,7 @@ def set_record(self, tool_record):
             QuerySet(ToolRecord).update_or_create(id=tool_record.id,
                                                   create_defaults={'id': tool_record.id,
                                                                    'tool_id': tool_record.tool_id,
+                                                                   'state': tool_record.state,
                                                                    'workspace_id': tool_record.workspace_id,
                                                                    "source_type": tool_record.source_type,
                                                                    'source_id': tool_record.source_id,
@@ -88,6 +89,7 @@ def set_record(self, tool_record):
                                                       'tool_id': tool_record.tool_id,
                                                       "source_type": tool_record.source_type,
                                                       'source_id': tool_record.source_id,
+                                                      'state': tool_record.state,
                                                       'meta': tool_record.meta,
                                                       'run_time': tool_record.run_time
                                                   })