add image file as markdown stream outupt

api/core/app/task_pipeline/message_cycle_manage.py

@@ -97,6 +97,11 @@ class MessageCycleManage:
         )
 
         if message_file:
+            # get tool file id
+            tool_file_id = message_file.url.split('/')[-1]
+            # trim extension
+            tool_file_id = tool_file_id.split('.')[0]
+
             # get extension
             if '.' in message_file.url:
                 extension = f'.{message_file.url.split(".")[-1]}'
@@ -105,7 +110,7 @@ class MessageCycleManage:
             else:
                 extension = '.bin'
             # add sign url
-            url = ToolFileManager.sign_file(file_id=message_file.id, extension=extension)
+            url = ToolFileManager.sign_file(tool_file_id=tool_file_id, extension=extension)
 
             return MessageFileStreamResponse(
                 task_id=self._application_generate_entity.task_id,

api/core/app/task_pipeline/workflow_cycle_manage.py

@@ -21,6 +21,7 @@ from core.app.entities.task_entities import (
     WorkflowStartStreamResponse,
     WorkflowTaskState,
 )
+from core.file.file_obj import FileVar
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.workflow.entities.node_entities import NodeRunMetadataKey, NodeType, SystemVariable
 from extensions.ext_database import db
@@ -93,7 +94,7 @@ class WorkflowCycleManage:
                               start_at: float,
                               total_tokens: int,
                               total_steps: int,
-                              outputs: Optional[dict] = None) -> WorkflowRun:
+                              outputs: Optional[str] = None) -> WorkflowRun:
         """
         Workflow run success
         :param workflow_run: workflow run
@@ -244,7 +245,8 @@ class WorkflowCycleManage:
 
         return workflow_node_execution
 
-    def _workflow_start_to_stream_response(self, task_id: str, workflow_run: WorkflowRun) -> WorkflowStartStreamResponse:
+    def _workflow_start_to_stream_response(self, task_id: str,
+                                           workflow_run: WorkflowRun) -> WorkflowStartStreamResponse:
         """
         Workflow start to stream response.
         :param task_id: task id
@@ -262,7 +264,8 @@ class WorkflowCycleManage:
             )
         )
 
-    def _workflow_finish_to_stream_response(self, task_id: str, workflow_run: WorkflowRun) -> WorkflowFinishStreamResponse:
+    def _workflow_finish_to_stream_response(self, task_id: str,
+                                            workflow_run: WorkflowRun) -> WorkflowFinishStreamResponse:
         """
         Workflow finish to stream response.
         :param task_id: task id
@@ -283,7 +286,8 @@ class WorkflowCycleManage:
                 total_tokens=workflow_run.total_tokens,
                 total_steps=workflow_run.total_steps,
                 created_at=int(workflow_run.created_at.timestamp()),
-                finished_at=int(workflow_run.finished_at.timestamp())
+                finished_at=int(workflow_run.finished_at.timestamp()),
+                files=self._fetch_files_from_node_outputs(workflow_run.outputs_dict)
             )
         )
 
@@ -310,7 +314,7 @@ class WorkflowCycleManage:
         )
 
     def _workflow_node_finish_to_stream_response(self, task_id: str, workflow_node_execution: WorkflowNodeExecution) \
-        -> NodeFinishStreamResponse:
+            -> NodeFinishStreamResponse:
         """
         Workflow node finish to stream response.
         :param task_id: task id
@@ -334,7 +338,8 @@ class WorkflowCycleManage:
                 elapsed_time=workflow_node_execution.elapsed_time,
                 execution_metadata=workflow_node_execution.execution_metadata_dict,
                 created_at=int(workflow_node_execution.created_at.timestamp()),
-                finished_at=int(workflow_node_execution.finished_at.timestamp())
+                finished_at=int(workflow_node_execution.finished_at.timestamp()),
+                files=self._fetch_files_from_node_outputs(workflow_node_execution.outputs_dict)
             )
         )
 
@@ -465,3 +470,48 @@ class WorkflowCycleManage:
         db.session.close()
 
         return workflow_run
+
+    def _fetch_files_from_node_outputs(self, outputs_dict: dict) -> list[dict]:
+        """
+        Fetch files from node outputs
+        :param outputs_dict: node outputs dict
+        :return:
+        """
+        files = []
+        for output_var, output_value in outputs_dict.items():
+            file_vars = self._fetch_files_from_variable_value(output_value)
+            if file_vars:
+                files.extend(file_vars)
+
+        return files
+
+    def _fetch_files_from_variable_value(self, value: Union[dict, list]) -> list[dict]:
+        """
+        Fetch files from variable value
+        :param value: variable value
+        :return:
+        """
+        files = []
+        if isinstance(value, list):
+            for item in value:
+                file_var = self._get_file_var_from_value(item)
+                if file_var:
+                    files.append(file_var)
+        elif isinstance(value, dict):
+            file_var = self._get_file_var_from_value(value)
+            if file_var:
+                files.append(file_var)
+
+        return files
+
+    def _get_file_var_from_value(self, value: Union[dict, list]) -> Optional[dict]:
+        """
+        Get file var from value
+        :param value: variable value
+        :return:
+        """
+        if isinstance(value, dict):
+            if '__variant' in value and value['__variant'] == FileVar.__class__.__name__:
+                return value
+
+        return None