[SDK][Internal] Support flow test in experiment (#1883)

# Description

Please add an informative description that covers that changes made by
the pull request and link all relevant issues.

# All Promptflow Contribution checklist:
- [ ] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [ ] **I have read the [contribution guidelines](../CONTRIBUTING.md).**
- [ ] **Create an issue and link to the pull request to get dedicated
review from promptflow team. Learn more: [suggested
workflow](../CONTRIBUTING.md#suggested-workflow).**

## General Guidelines and Best Practices
- [ ] Title of the pull request is clear and informative.
- [ ] There are a small number of commits, each of which have an
informative message. This means that previously merged commits do not
appear in the history of the PR. For more information on cleaning up the
commits in your PR, [see this
page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).

### Testing Guidelines
- [ ] Pull request includes test coverage for the included changes.

---------

Signed-off-by: Brynn Yin <biyi@microsoft.com>
Co-authored-by: Philip Gao <yigao@microsoft.com>

src/promptflow/promptflow/_cli/_pf/_experiment.py

@@ -160,7 +160,7 @@ def create_experiment(args: argparse.Namespace):
     template_path = args.template
     logger.debug("Loading experiment template from %s", template_path)
     template = load_common(ExperimentTemplate, source=template_path)
-    logger.debug("Creating experiment from template %s", template.name)
+    logger.debug("Creating experiment from template %s", template.dir_name)
     experiment = Experiment.from_template(template, name=args.name)
     logger.debug("Creating experiment %s", experiment.name)
     exp = _get_pf_client()._experiments.create_or_update(experiment)

src/promptflow/promptflow/_cli/_pf/_flow.py

@@ -39,6 +39,7 @@
 from promptflow._cli._pf._run import exception_handler
 from promptflow._cli._utils import _copy_to_flow, activate_action, confirm, inject_sys_path, list_of_dict_to_dict
 from promptflow._constants import FlowLanguage
+from promptflow._sdk._configuration import Configuration
 from promptflow._sdk._constants import PROMPT_FLOW_DIR_NAME, ConnectionProvider
 from promptflow._sdk._pf_client import PFClient
 from promptflow._sdk.operations._flow_operations import FlowOperations
@@ -228,6 +229,9 @@ def add_parser_test_flow(subparsers):
     add_param_detail = lambda parser: parser.add_argument(  # noqa: E731
         "--detail", type=str, default=None, required=False, help=argparse.SUPPRESS
     )
+    add_param_experiment = lambda parser: parser.add_argument(  # noqa: E731
+        "--experiment", type=str, help="the experiment template path of flow."
+    )
 
     add_params = [
         add_param_flow,
@@ -242,6 +246,9 @@ def add_parser_test_flow(subparsers):
         add_param_config,
         add_param_detail,
     ] + base_params
+
+    if Configuration.get_instance().is_internal_features_enabled():
+        add_params.append(add_param_experiment)
     activate_action(
         name="test",
         description="Test the flow.",
@@ -371,7 +378,6 @@ def _init_flow_by_template(flow_name, flow_type, overwrite=False, connection=Non
 
 @exception_handler("Flow test")
 def test_flow(args):
-    from promptflow._sdk._load_functions import load_flow
 
     config = list_of_dict_to_dict(args.config)
     pf_client = PFClient(config=config)
@@ -380,6 +386,22 @@ def test_flow(args):
         environment_variables = list_of_dict_to_dict(args.environment_variables)
     else:
         environment_variables = {}
+    inputs = _build_inputs_for_flow_test(args)
+    # Select different test mode
+    if Configuration.get_instance().is_internal_features_enabled() and args.experiment:
+        _test_flow_experiment(args, pf_client, inputs, environment_variables)
+        return
+    if args.multi_modal or args.ui:
+        _test_flow_multi_modal(args, pf_client)
+        return
+    if args.interactive:
+        _test_flow_interactive(args, pf_client, inputs, environment_variables)
+        return
+    _test_flow_standard(args, pf_client, inputs, environment_variables)
+
+
+def _build_inputs_for_flow_test(args):
+    """Build inputs from --input and --inputs for flow test."""
     inputs = {}
     if args.input:
         from promptflow._utils.load_data import load_data
@@ -394,49 +416,78 @@ def test_flow(args):
         inputs = load_data(local_path=args.input)[0]
     if args.inputs:
         inputs.update(list_of_dict_to_dict(args.inputs))
+    return inputs
 
-    if args.multi_modal or args.ui:
-        with tempfile.TemporaryDirectory() as temp_dir:
-            flow = load_flow(args.flow)
-
-            script_path = [
-                os.path.join(temp_dir, "main.py"),
-                os.path.join(temp_dir, "utils.py"),
-                os.path.join(temp_dir, "logo.png"),
-            ]
-            for script in script_path:
-                StreamlitFileReplicator(
-                    flow_name=flow.display_name if flow.display_name else flow.name,
-                    flow_dag_path=flow.flow_dag_path,
-                ).generate_to_file(script)
-            main_script_path = os.path.join(temp_dir, "main.py")
-            pf_client.flows._chat_with_ui(script=main_script_path)
+
+def _test_flow_multi_modal(args, pf_client):
+    """Test flow with multi modality mode."""
+    from promptflow._sdk._load_functions import load_flow
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        flow = load_flow(args.flow)
+
+        script_path = [
+            os.path.join(temp_dir, "main.py"),
+            os.path.join(temp_dir, "utils.py"),
+            os.path.join(temp_dir, "logo.png"),
+        ]
+        for script in script_path:
+            StreamlitFileReplicator(
+                flow_name=flow.display_name if flow.display_name else flow.name,
+                flow_dag_path=flow.flow_dag_path,
+            ).generate_to_file(script)
+        main_script_path = os.path.join(temp_dir, "main.py")
+        pf_client.flows._chat_with_ui(script=main_script_path)
+
+
+def _test_flow_interactive(args, pf_client, inputs, environment_variables):
+    """Test flow with interactive mode."""
+    pf_client.flows._chat(
+        flow=args.flow,
+        inputs=inputs,
+        environment_variables=environment_variables,
+        variant=args.variant,
+        show_step_output=args.verbose,
+    )
+
+
+def _test_flow_standard(args, pf_client, inputs, environment_variables):
+    """Test flow with standard mode."""
+    result = pf_client.flows.test(
+        flow=args.flow,
+        inputs=inputs,
+        environment_variables=environment_variables,
+        variant=args.variant,
+        node=args.node,
+        allow_generator_output=False,
+        stream_output=False,
+        dump_test_result=True,
+        output_path=args.detail,
+    )
+    # Print flow/node test result
+    if isinstance(result, dict):
+        print(json.dumps(result, indent=4, ensure_ascii=False))
     else:
-        if args.interactive:
-            pf_client.flows._chat(
-                flow=args.flow,
-                inputs=inputs,
-                environment_variables=environment_variables,
-                variant=args.variant,
-                show_step_output=args.verbose,
-            )
-        else:
-            result = pf_client.flows.test(
-                flow=args.flow,
-                inputs=inputs,
-                environment_variables=environment_variables,
-                variant=args.variant,
-                node=args.node,
-                allow_generator_output=False,
-                stream_output=False,
-                dump_test_result=True,
-                detail=args.detail,
-            )
-            # Print flow/node test result
-            if isinstance(result, dict):
-                print(json.dumps(result, indent=4, ensure_ascii=False))
-            else:
-                print(result)
+        print(result)
+
+
+def _test_flow_experiment(args, pf_client, inputs, environment_variables):
+    """Test flow with experiment specified."""
+    if args.variant or args.node:
+        error = ValueError("--variant or --node is not supported experiment is specified.")
+        raise UserErrorException(
+            target=ErrorTarget.CONTROL_PLANE_SDK,
+            message=str(error),
+            error=error,
+        )
+    node_results = pf_client.flows.test(
+        flow=args.flow,
+        inputs=inputs,
+        environment_variables=environment_variables,
+        experiment=args.experiment,
+        output_path=args.detail,
+    )
+    print(json.dumps(node_results, indent=4, ensure_ascii=False))
 
 
 def serve_flow(args):

src/promptflow/promptflow/_internal/__init__.py

@@ -71,6 +71,7 @@
 )
 from promptflow._utils.execution_utils import handle_line_failures
 from promptflow._utils.feature_utils import Feature, FeatureState, get_feature_list
+from promptflow._utils.inputs_mapping_utils import apply_inputs_mapping
 from promptflow._utils.logger_utils import (
     DATETIME_FORMAT,
     LOG_FORMAT,
@@ -107,7 +108,6 @@
     transpose,
 )
 from promptflow._version import VERSION
-from promptflow.batch._batch_inputs_processor import apply_inputs_mapping
 from promptflow.executor._errors import InputNotFound
 from promptflow.executor._tool_invoker import DefaultToolInvoker
 from promptflow.storage._run_storage import DefaultRunStorage