[Promptflow] Support image in SingleNodeRun

src/promptflow/promptflow/_internal/__init__.py

@@ -102,3 +102,4 @@
     NotFoundException,
     SqliteClient,
 )
+from promptflow.storage._run_storage import DefaultRunStorage

src/promptflow/promptflow/_sdk/operations/_test_submitter.py

@@ -204,6 +204,7 @@ def node_test(
                 dependency_nodes_outputs=dependency_nodes_outputs,
                 connections=connections,
                 working_dir=self.flow.code,
+                output_dir=".promptflow/intermediate",
             )
             return result
 

src/promptflow/promptflow/executor/flow_executor.py

@@ -24,7 +24,7 @@
 from promptflow._core.tools_manager import ToolsManager
 from promptflow._utils.context_utils import _change_working_dir
 from promptflow._utils.logger_utils import logger
-from promptflow._utils.multimedia_utils import load_multimedia_data
+from promptflow._utils.multimedia_utils import load_multimedia_data, load_multimedia_data_recursively
 from promptflow._utils.utils import transpose
 from promptflow.contracts.flow import Flow, FlowInputDefinition, InputAssignment, InputValueType, Node
 from promptflow.contracts.run_info import FlowRunInfo, Status
@@ -48,7 +48,7 @@
 from promptflow.executor._tool_resolver import ToolResolver
 from promptflow.executor.flow_validator import FlowValidator
 from promptflow.storage import AbstractRunStorage
-from promptflow.storage._run_storage import DefaultRunStorage, DummyRunStorage
+from promptflow.storage._run_storage import DefaultRunStorage
 
 LINE_NUMBER_KEY = "line_number"  # Using the same key with portal.
 LINE_TIMEOUT_SEC = 600
@@ -242,6 +242,7 @@ def load_and_exec_node(
         flow_file: Path,
         node_name: str,
         *,
+        output_dir: Optional[str] = None,
         flow_inputs: Optional[Mapping[str, Any]] = None,
         dependency_nodes_outputs: Optional[Mapping[str, Any]] = None,
         connections: Optional[dict] = None,
@@ -294,8 +295,10 @@ def load_and_exec_node(
                 flow_file=flow_file,
             )
 
-        flow_inputs = FlowExecutor._apply_default_value_for_input(flow.inputs, flow_inputs)
-        converted_flow_inputs_for_node = FlowValidator.convert_flow_inputs_for_node(flow, node, flow_inputs)
+        inputs_with_default_value = FlowExecutor._apply_default_value_for_input(flow.inputs, flow_inputs)
+        inputs = load_multimedia_data(flow.inputs, inputs_with_default_value, working_dir)
+        dependency_nodes_outputs = load_multimedia_data_recursively(dependency_nodes_outputs)
+        converted_flow_inputs_for_node = FlowValidator.convert_flow_inputs_for_node(flow, node, inputs)
         package_tool_keys = [node.source.tool] if node.source and node.source.tool else []
         tool_resolver = ToolResolver(working_dir, connections, package_tool_keys)
         resolved_node = tool_resolver.resolve_tool_by_node(node)
@@ -320,7 +323,9 @@ def load_and_exec_node(
         resolved_inputs = {k: v for k, v in resolved_inputs.items() if k not in resolved_node.init_args}
 
         # TODO: Simplify the logic here
-        run_tracker = RunTracker(DummyRunStorage())
+        sub_dir = "." if output_dir is None else output_dir
+        storage = DefaultRunStorage(base_dir=Path("."), sub_dir=Path(sub_dir))
+        run_tracker = RunTracker(storage)
         with run_tracker.node_log_manager:
             ToolInvoker.activate(DefaultToolInvoker())
 

src/promptflow/tests/executor/e2etests/test_executor_happypath.py

@@ -1,4 +1,5 @@
 import uuid
+import os
 from types import GeneratorType
 
 import pytest
@@ -19,6 +20,7 @@
     get_flow_expected_status_summary,
     get_flow_sample_inputs,
     get_yaml_file,
+    get_yaml_working_dir
 )
 
 SAMPLE_FLOW = "web_classification_no_variants"
@@ -203,6 +205,7 @@ def test_executor_exec_bulk_with_openai_metrics(self, dev_connections):
             "script_with_import",
             "package_tools",
             "connection_as_input",
+            "python_tool_with_multiple_image_nodes"
         ],
     )
     def test_executor_exec_line(self, flow_folder, dev_connections):
@@ -233,17 +236,24 @@ def test_executor_exec_line(self, flow_folder, dev_connections):
             ("connection_as_input", "conn_node", None, None),
             ("simple_aggregation", "accuracy", {"text": "A"}, {"passthrough": "B"}),
             ("script_with_import", "node1", {"text": "text"}, None),
+            ("python_tool_with_multiple_image_nodes", "python_node_2", {"logo_content": "Microsoft and four squares"},
+             {"python_node": {"image": "logo.jpg", "image_name": "Microsoft's logo", "image_list": ["logo.jpg"]}}),
+            ("python_tool_with_multiple_image_nodes", "python_node", {
+             "image": "logo.jpg", "image_name": "Microsoft's logo"}, {},)
         ],
     )
     def test_executor_exec_node(self, flow_folder, node_name, flow_inputs, dependency_nodes_outputs, dev_connections):
         self.skip_serp(flow_folder, dev_connections)
         yaml_file = get_yaml_file(flow_folder)
+        working_dir = get_yaml_working_dir(flow_folder)
+        os.chdir(working_dir)
         run_info = FlowExecutor.load_and_exec_node(
             yaml_file,
             node_name,
             flow_inputs=flow_inputs,
             dependency_nodes_outputs=dependency_nodes_outputs,
             connections=dev_connections,
+            output_dir=("./temp"),
             raise_ex=True,
         )
         assert run_info.output is not None

src/promptflow/tests/executor/utils.py

@@ -14,6 +14,11 @@ def get_yaml_file(folder_name, root: str = FLOW_ROOT, file_name: str = "flow.dag
     return yaml_file
 
 
+def get_yaml_working_dir(folder_name, root: str = FLOW_ROOT):
+    flow_folder_path = Path(root) / folder_name
+    return flow_folder_path
+
+
 def get_flow_inputs(folder_name, root: str = FLOW_ROOT):
     flow_folder_path = Path(root) / folder_name
     inputs = load_json(flow_folder_path / "inputs.json")

src/promptflow/tests/test_configs/flows/python_tool_with_multiple_image_nodes/flow.dag.yaml

@@ -2,6 +2,12 @@ inputs:
   image:
     type: image
     default: logo.jpg
+  image_name:
+    type: string
+    default: Microsoft's logo
+  logo_content:
+    type: string
+    default: Microsoft and four squares
 outputs:
   output:
     type: image
@@ -14,10 +20,12 @@ nodes:
     path: python_with_image.py
   inputs:
     image: ${inputs.image}
+    image_name: ${inputs.image_name}
 - name: python_node_2
   type: python
   source:
     type: code
-    path: python_with_image.py
+    path: python_node_2.py
   inputs:
-    image: ${python_node.output}
+    image_dict: ${python_node.output}
+    logo_content: ${inputs.logo_content}

src/promptflow/tests/test_configs/flows/python_tool_with_multiple_image_nodes/python_node_2.py

@@ -0,0 +1,9 @@
+from promptflow.contracts.multimedia import Image
+from promptflow import tool
+
+
+@tool
+def python_with_image(image_dict: dict, logo_content: str) -> Image:
+    image_dict["image_list2"] = [image_dict["image"], image_dict["image"]]
+    image_dict["logo_content"] = logo_content
+    return image_dict

src/promptflow/tests/test_configs/flows/python_tool_with_multiple_image_nodes/python_with_image.py

@@ -3,5 +3,5 @@
 
 
 @tool
-def python_with_image(image: Image) -> Image:
-    return image
+def python_with_image(image: Image, image_name: str) -> Image:
+    return {"image": image, "image_name": image_name, "image_list": [image, image]}