add test case for image aggregation

src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py

@@ -2206,6 +2206,61 @@ def test_flow_run_resume_from_token(self, capfd, local_client) -> None:
             <= resume_run.properties["system_metrics"]["total_tokens"]
         )
 
+    def test_flow_run_resume_from_image_aggregation(self, capfd, local_client) -> None:
+        run_id = str(uuid.uuid4())
+        # fetch std out
+        run_pf_command(
+            "run",
+            "create",
+            "--flow",
+            f"{FLOWS_DIR}/eval_flow_with_image_resume_random_fail",
+            "--data",
+            f"{FLOWS_DIR}/eval_flow_with_image_resume_random_fail/data.jsonl",
+            "--column-mapping",
+            "input_image='${data.input_image}'",
+            "--name",
+            run_id,
+        )
+        out, _ = capfd.readouterr()
+        assert "Completed" in out
+        original_run = local_client.runs.get(name=run_id)
+        output_path = os.path.join(original_run.properties["output_path"], "flow_outputs", "output.jsonl")
+        with open(output_path, "r") as file:
+            original_output = [json.loads(line) for line in file]
+
+        new_run_id = str(uuid.uuid4())
+        display_name = "test"
+        description = "new description"
+        run_pf_command(
+            "run",
+            "create",
+            "--resume-from",
+            run_id,
+            "--name",
+            new_run_id,
+            "--set",
+            f"display_name={display_name}",
+            f"description={description}",
+            "tags.A=A",
+            "tags.B=B",
+        )
+        resume_run = local_client.runs.get(name=new_run_id)
+        output_path = os.path.join(resume_run.properties["output_path"], "flow_outputs", "output.jsonl")
+
+        with open(output_path, "r") as file:
+            resume_output = [json.loads(line) for line in file]
+
+        # assert original_output in resume_output
+        original_output_line_numbers = {line["line_number"] for line in original_output}
+        resume_output_line_numbers = {line["line_number"] for line in resume_output}
+        assert original_output_line_numbers.issubset(resume_output_line_numbers)
+        assert len(resume_output) >= len(original_output)
+        assert resume_run.name == new_run_id
+        assert resume_run.display_name == display_name
+        assert resume_run.description == description
+        assert resume_run.tags == {"A": "A", "B": "B"}
+        assert resume_run._resume_from == run_id
+
     def test_flow_run_exclusive_param(self, capfd) -> None:
         # fetch std out
         with pytest.raises(SystemExit):

src/promptflow/tests/test_configs/flows/eval_flow_with_image_resume_random_fail/count_image.py

@@ -0,0 +1,17 @@
+from typing import List
+from promptflow import tool
+from promptflow.contracts.multimedia import Image
+
+
+@tool
+def aggregate(images: List[Image]):
+    from promptflow import log_metric
+    image_count = 0
+    for image in images:
+        if not isinstance(image, Image):
+            print(f"Invalid image: {image}")
+        else:
+            image_count += 1
+    log_metric(key="image_count", value=image_count)
+
+    return image_count

src/promptflow/tests/test_configs/flows/eval_flow_with_image_resume_random_fail/data.jsonl

@@ -0,0 +1,9 @@
+{"input_image": {"data:image/jpg;path": "logo.gif"}}
+{"input_image": {"data:image/png;path": "logo.jpg"}}
+{"input_image": {"data:image/png;path": "logo.png"}}
+{"input_image": {"data:image/jpg;path": "logo.gif"}}
+{"input_image": {"data:image/png;path": "logo.jpg"}}
+{"input_image": {"data:image/png;path": "logo.png"}}
+{"input_image": {"data:image/jpg;path": "logo.gif"}}
+{"input_image": {"data:image/png;path": "logo.jpg"}}
+{"input_image": {"data:image/png;path": "logo.png"}}

src/promptflow/tests/test_configs/flows/eval_flow_with_image_resume_random_fail/flip_image.py

@@ -0,0 +1,18 @@
+import io
+import random
+from pathlib import Path
+from promptflow import tool
+from promptflow.contracts.multimedia import Image
+from PIL import Image as PIL_Image
+
+
+@tool
+def passthrough(input_image: Image) -> Image:
+    if random.random() < 0.5:
+        raise ValueError("Random failure")
+    image_stream = io.BytesIO(input_image)
+    pil_image = PIL_Image.open(image_stream)
+    flipped_image = pil_image.transpose(PIL_Image.FLIP_LEFT_RIGHT)
+    buffer = io.BytesIO()
+    flipped_image.save(buffer, format="PNG")
+    return Image(buffer.getvalue(), mime_type="image/png")

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

@@ -0,0 +1,25 @@
+$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Flow.schema.json
+inputs:
+  input_image:
+    type: image
+    default: https://developer.microsoft.com/_devcom/images/logo-ms-social.png
+outputs:
+  output_image:
+    type: string
+    reference: ${flip_image.output}
+nodes:
+- name: flip_image
+  type: python
+  source:
+    type: code
+    path: flip_image.py
+  inputs:
+    input_image: ${inputs.input_image}
+- name: count_image
+  type: python
+  source:
+    type: code
+    path: count_image.py
+  inputs:
+    images: ${flip_image.output}
+  aggregation: true

src/promptflow/tests/test_configs/flows/eval_flow_with_image_resume_random_fail/question_on_image.jinja2

@@ -0,0 +1,7 @@
+# system:
+As an AI assistant, your task involves interpreting images and responding to questions about the image.
+Remember to provide accurate answers based on the information present in the image.
+
+# user:
+{{question}}
+![image]({{test_image}})

src/promptflow/tests/test_configs/flows/eval_flow_with_image_resume_random_fail/requirements.txt

@@ -0,0 +1,2 @@
+promptflow
+promptflow-tools