Migrate the yaml used by the eager flow to flow.flex.yaml, code support both flow.flex.yaml and flow.dag.yaml. (#2548)

# Description

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

1. Modify the **yaml** used for all **eager tests** to
**flow.flex.yaml.**
2. Modify the **yaml** of the **eager flow** in the code to
**flow.flex.yaml**.
3. Add **flow** when saving run to/retrieving run from the **database**

# All Promptflow Contribution checklist:
- [x] **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
- [x] 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
- [x] Pull request includes test coverage for the included changes.

Author: Stephen1993

src/promptflow-core/promptflow/_constants.py

@@ -27,6 +27,7 @@
 FLOW_DAG_YAML = "flow.dag.yaml"
 FLOW_FLEX_YAML = "flow.flex.yaml"
 PROMPTY_EXTENSION = ".prompty"
+FLOW_FILE_SUFFIX = (".yaml", ".yml", PROMPTY_EXTENSION)
 
 CHAT_HISTORY = "chat_history"
 

src/promptflow-core/promptflow/_utils/flow_utils.py

@@ -13,6 +13,7 @@
     CHAT_HISTORY,
     DEFAULT_ENCODING,
     FLOW_DAG_YAML,
+    FLOW_FILE_SUFFIX,
     FLOW_FLEX_YAML,
     PROMPT_FLOW_DIR_NAME,
     PROMPTY_EXTENSION,
@@ -95,13 +96,19 @@ def resolve_flow_path(
                 f"Please specify a file or remove the extra YAML.",
                 privacy_info=[str(flow_path)],
             )
-    elif flow_path.is_file() or flow_path.suffix in (".yaml", ".yml"):
+    elif flow_path.is_file() or flow_path.suffix.lower() in FLOW_FILE_SUFFIX:
         flow_folder = flow_path.parent
         flow_file = flow_path.name
     else:  # flow_path doesn't exist
         flow_folder = flow_path
         flow_file = FLOW_DAG_YAML
 
+    file_path = flow_folder / flow_file
+    if file_path.suffix.lower() not in FLOW_FILE_SUFFIX:
+        raise UserErrorException(
+            error_format=f"The flow file suffix must be yaml or yml, " f"and cannot be {file_path.suffix}"
+        )
+
     if not check_flow_exist:
         return flow_folder.resolve().absolute(), flow_file
 
@@ -111,10 +118,10 @@ def resolve_flow_path(
             privacy_info=[flow_path.absolute().as_posix()],
         )
 
-    if not (flow_folder / flow_file).is_file():
+    if not file_path.is_file():
         raise UserErrorException(
-            f"Can't find file {flow_file}, " f"in the flow path {flow_folder.absolute().as_posix()}.",
-            privacy_info=[flow_folder.absolute().as_posix()],
+            f"Flow file {file_path.absolute().as_posix()} does not exist.",
+            privacy_info=[file_path.absolute().as_posix()],
         )
 
     return flow_folder.resolve().absolute(), flow_file
@@ -141,20 +148,25 @@ def dump_flow_dag(flow_dag: dict, flow_path: Path):
 
 
 def is_flex_flow(
-    *, file_path: Union[str, Path, None] = None, yaml_dict: Optional[dict] = None, working_dir: Optional[Path] = None
+    *,
+    flow_path: Union[str, Path, PathLike, None] = None,
+    yaml_dict: Optional[dict] = None,
+    working_dir: Union[str, Path, PathLike, None] = None,
+    check_flow_exist=True,
 ):
     """Check if the flow is a flex flow."""
-    if file_path is None and yaml_dict is None:
+    if flow_path is None and yaml_dict is None:
         raise UserErrorException("Either file_path or yaml_dict should be provided.")
-    if file_path is not None and yaml_dict is not None:
+    if flow_path is not None and yaml_dict is not None:
         raise UserErrorException("Only one of file_path and yaml_dict should be provided.")
-    if file_path is not None:
-        file_path = Path(file_path)
-        if working_dir is not None and not file_path.is_absolute():
-            file_path = working_dir / file_path
-        if file_path.suffix.lower() not in [".yaml", ".yml"]:
-            return False
-        yaml_dict = load_yaml(file_path)
+    if flow_path is not None:
+        flow_path, flow_file = resolve_flow_path(flow_path, base_path=working_dir, check_flow_exist=False)
+        file_path = flow_path / flow_file
+        if file_path.is_file() and file_path.suffix.lower() in (".yaml", ".yml"):
+            yaml_dict = load_yaml(file_path)
+        elif not check_flow_exist:
+            return flow_file == FLOW_FLEX_YAML
+
     return isinstance(yaml_dict, dict) and "entry" in yaml_dict
 
 

src/promptflow-core/promptflow/executor/flow_executor.py

@@ -211,7 +211,7 @@ def create(
             return ScriptExecutor(flow_file, storage=storage)
         if not isinstance(flow_file, (Path, str)):
             raise NotImplementedError("Only support Path or str for flow_file.")
-        if is_flex_flow(file_path=flow_file, working_dir=working_dir):
+        if is_flex_flow(flow_path=flow_file, working_dir=working_dir):
             from ._script_executor import ScriptExecutor
 
             return ScriptExecutor(

src/promptflow-core/tests/conftest.py

@@ -1,5 +1,7 @@
 from pathlib import Path
 
+from promptflow._utils.flow_utils import resolve_flow_path
+
 TEST_CONFIG_ROOT = Path(__file__).parent.parent.parent / "promptflow" / "tests" / "test_configs"
 FLOW_ROOT = TEST_CONFIG_ROOT / "flows"
 EAGER_FLOW_ROOT = TEST_CONFIG_ROOT / "eager_flows"
@@ -10,6 +12,11 @@ def get_flow_folder(folder_name, root: str = FLOW_ROOT) -> Path:
     return flow_folder_path
 
 
-def get_yaml_file(folder_name, root: str = FLOW_ROOT, file_name: str = "flow.dag.yaml") -> Path:
-    yaml_file = get_flow_folder(folder_name, root) / file_name
+def get_yaml_file(folder_name, root: str = FLOW_ROOT, file_name: str = None) -> Path:
+    if file_name is None:
+        flow_path, flow_file = resolve_flow_path(get_flow_folder(folder_name, root), check_flow_exist=False)
+        yaml_file = flow_path / flow_file
+    else:
+        yaml_file = get_flow_folder(folder_name, root) / file_name
+
     return yaml_file

src/promptflow-devkit/promptflow/_proxy/_base_executor_proxy.py

@@ -11,7 +11,7 @@
 
 import httpx
 
-from promptflow._constants import DEFAULT_ENCODING, LINE_TIMEOUT_SEC
+from promptflow._constants import DEFAULT_ENCODING, FLOW_FLEX_YAML, LINE_TIMEOUT_SEC
 from promptflow._core._errors import NotSupported, UnexpectedError
 from promptflow._proxy._errors import ExecutorServiceUnhealthy
 from promptflow._sdk._constants import (
@@ -23,7 +23,7 @@
 )
 from promptflow._utils.async_utils import async_run_allowing_running_loop
 from promptflow._utils.exception_utils import ErrorResponse, ExceptionPresenter
-from promptflow._utils.flow_utils import is_flex_flow, read_json_content, resolve_flow_path
+from promptflow._utils.flow_utils import is_flex_flow, read_json_content
 from promptflow._utils.logger_utils import bulk_logger
 from promptflow._utils.utils import load_json
 from promptflow.contracts.run_info import FlowRunInfo
@@ -52,7 +52,7 @@ def generate_flow_tools_json(
         load_in_subprocess: bool = True,
     ) -> dict:
         """Generate flow.tools.json for the specified flow."""
-        if is_flex_flow(file_path=flow_file, working_dir=working_dir):
+        if is_flex_flow(flow_path=flow_file, working_dir=working_dir):
             return {}
         else:
             return cls._generate_flow_tools_json(flow_file, working_dir, dump, timeout, load_in_subprocess)
@@ -93,7 +93,7 @@ def generate_flow_json(
         :return: The metadata of the flow.
         :rtype: Dict[str, Any]
         """
-        if is_flex_flow(file_path=flow_file, working_dir=working_dir):
+        if is_flex_flow(flow_path=flow_file, working_dir=working_dir):
             return cls._generate_flow_json(flow_file, working_dir, dump, timeout, load_in_subprocess)
         else:
             return {}
@@ -239,9 +239,8 @@ def get_inputs_definition(self):
         """Get the inputs definition of an eager flow"""
         from promptflow.contracts.flow import FlowInputDefinition
 
-        _, flow_file = resolve_flow_path(self.working_dir, check_flow_exist=False)
         flow_meta = self.generate_flow_json(
-            flow_file=self.working_dir / flow_file,
+            flow_file=self.working_dir / FLOW_FLEX_YAML,
             working_dir=self.working_dir,
             dump=False,
         )

src/promptflow-devkit/promptflow/_proxy/_csharp_executor_proxy.py

@@ -85,7 +85,7 @@ def dump_metadata(cls, flow_file: Path, working_dir: Path) -> NoReturn:
     @classmethod
     def get_outputs_definition(cls, flow_file: Path, working_dir: Path) -> dict:
         # TODO: no outputs definition for eager flow for now
-        if is_flex_flow(file_path=flow_file, working_dir=working_dir):
+        if is_flex_flow(flow_path=flow_file, working_dir=working_dir):
             return {}
 
         # TODO: get this from self._get_flow_meta for both eager flow and non-eager flow then remove

src/promptflow-devkit/promptflow/_sdk/_constants.py

@@ -489,6 +489,10 @@ class OSType:
     LINUX = "Linux"
 
 
+class RunMode:
+    EAGER = "Eager"
+
+
 # Note: Keep these for backward compatibility
 CustomStrongTypeConnectionConfigs = CustomStrongTypeConnectionConfigs
 ConnectionType = ConnectionType

src/promptflow-devkit/promptflow/_sdk/_service/apis/ui.py

@@ -168,7 +168,7 @@ def get(self):
         flow_path = get_set_flow_yaml(flow, experiment)
         flow_path_dir, flow_path_file = resolve_flow_path(flow_path)
         flow_info = load_yaml(flow_path_dir / flow_path_file)
-        if is_flex_flow(file_path=flow_path_dir / flow_path_file):
+        if is_flex_flow(flow_path=flow_path_dir / flow_path_file):
             # call api provided by han to get flow input
             flow_input = {}
             flow_info.update(flow_input)

src/promptflow-devkit/promptflow/_sdk/_utils.py

@@ -32,7 +32,7 @@
 from keyring.errors import NoKeyringError
 from marshmallow import ValidationError
 
-from promptflow._constants import ENABLE_MULTI_CONTAINER_KEY, EXTENSION_UA, FLOW_DAG_YAML, FlowLanguage
+from promptflow._constants import ENABLE_MULTI_CONTAINER_KEY, EXTENSION_UA, FLOW_FLEX_YAML, FlowLanguage
 from promptflow._core.entry_meta_generator import generate_flow_meta as _generate_flow_meta
 from promptflow._sdk._constants import (
     AZURE_WORKSPACE_REGEX_FORMAT,
@@ -54,6 +54,8 @@
     VARIANTS,
     AzureMLWorkspaceTriad,
     CommonYamlFields,
+    RunInfoSources,
+    RunMode,
 )
 from promptflow._sdk._errors import (
     DecryptConnectionError,
@@ -62,7 +64,7 @@
     UnsecureConnectionError,
 )
 from promptflow._sdk._vendor import IgnoreFile, get_ignore_file, get_upload_files_from_folder
-from promptflow._utils.flow_utils import resolve_flow_path
+from promptflow._utils.flow_utils import is_flex_flow, resolve_flow_path
 from promptflow._utils.logger_utils import get_cli_sdk_logger
 from promptflow._utils.user_agent_utils import ClientUserAgentUtil
 from promptflow._utils.yaml_utils import dump_yaml, load_yaml, load_yaml_string
@@ -984,6 +986,7 @@ def generate_yaml_entry(entry: Union[str, PathLike, Callable], code: Path = None
 @contextmanager
 def create_temp_flex_flow_yaml(entry: Union[str, PathLike, Callable], code: Path = None):
     """Create a temporary flow.dag.yaml in code folder"""
+
     logger.info("Create temporary entry for flex flow.")
     if callable(entry):
         entry = callable_to_entry_string(entry)
@@ -994,7 +997,7 @@ def create_temp_flex_flow_yaml(entry: Union[str, PathLike, Callable], code: Path
         code = Path(code)
         if not code.exists():
             raise UserErrorException(f"Code path {code.as_posix()} does not exist.")
-    flow_yaml_path = code / FLOW_DAG_YAML
+    flow_yaml_path = code / FLOW_FLEX_YAML
     existing_content = None
 
     try:
@@ -1048,6 +1051,22 @@ def callable_to_entry_string(callable_obj: Callable) -> str:
     return f"{module_str}:{func_str}"
 
 
+def is_flex_run(run: "Run") -> bool:
+    if run._run_source == RunInfoSources.LOCAL:
+        try:
+            # The flow yaml may have been temporarily generated and deleted after creating a run.
+            # So check_flow_exist=False.
+            return is_flex_flow(flow_path=run.flow, check_flow_exist=False)
+        except Exception as e:
+            # For run with incomplete flow snapshot, ignore load flow error to make sure it can still show.
+            logger.debug(f"Failed to check is flex flow from {run.flow} due to {e}.")
+            return False
+    elif run._run_source in [RunInfoSources.INDEX_SERVICE, RunInfoSources.RUN_HISTORY]:
+        return run._properties.get("azureml.promptflow.run_mode") == RunMode.EAGER
+    # TODO(2901279): support eager mode for run created from run folder
+    return False
+
+
 generate_flow_meta = _generate_flow_meta
 # DO NOT remove the following line, it's used by the runtime imports from _sdk/_utils directly
 get_used_connection_names_from_dict = get_used_connection_names_from_dict

src/promptflow-devkit/promptflow/_sdk/entities/_flows/base.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 from typing import Union
 
-from promptflow._constants import DEFAULT_ENCODING, PROMPTY_EXTENSION
+from promptflow._constants import DEFAULT_ENCODING, FLOW_FILE_SUFFIX
 from promptflow._sdk.entities._validation import SchemaValidatableMixin
 from promptflow._utils.flow_utils import is_flex_flow, is_prompty_flow, resolve_flow_path
 from promptflow._utils.yaml_utils import load_yaml_string
@@ -182,7 +182,7 @@ def _load_prepare(cls, source: Union[str, PathLike]):
         flow_dir, flow_filename = resolve_flow_path(source)
         flow_path = flow_dir / flow_filename
 
-        if flow_path.suffix not in [".yaml", ".yml", PROMPTY_EXTENSION]:
+        if flow_path.suffix not in FLOW_FILE_SUFFIX:
             raise UserErrorException("Source must be a directory or a 'flow.dag.yaml' file or a prompty file")
         return flow_dir, flow_path
 

src/promptflow-devkit/promptflow/_sdk/operations/_flow_operations.py

@@ -19,7 +19,7 @@
 
 import pydash
 
-from promptflow._constants import FLOW_DAG_YAML, LANGUAGE_KEY, PROMPT_FLOW_DIR_NAME, FlowLanguage
+from promptflow._constants import FLOW_FLEX_YAML, LANGUAGE_KEY, PROMPT_FLOW_DIR_NAME, FlowLanguage
 from promptflow._proxy import ProxyFactory
 from promptflow._sdk._configuration import Configuration
 from promptflow._sdk._constants import (
@@ -1109,7 +1109,7 @@ def _save(
         if LANGUAGE_KEY in kwargs:
             data[LANGUAGE_KEY] = language
 
-        target_flow_file = target_flow_directory / FLOW_DAG_YAML
+        target_flow_file = target_flow_directory / FLOW_FLEX_YAML
         # schema validation, here target_flow_file doesn't exist actually
         FlexFlow(path=target_flow_file, code=code, data=data, entry=data["entry"])._validate(raise_error=True)
 

src/promptflow-devkit/promptflow/_sdk/operations/_local_storage_operations.py

@@ -14,20 +14,19 @@
 
 from filelock import FileLock
 
-from promptflow._constants import FLOW_DAG_YAML, OUTPUT_FILE_NAME, OutputsFolderName
+from promptflow._constants import FLOW_DAG_YAML, FLOW_FLEX_YAML, OUTPUT_FILE_NAME, OutputsFolderName
 from promptflow._sdk._constants import (
     HOME_PROMPT_FLOW_DIR,
     LINE_NUMBER,
     LOCAL_STORAGE_BATCH_SIZE,
     PROMPT_FLOW_DIR_NAME,
     LocalStorageFilenames,
-    RunInfoSources,
 )
 from promptflow._sdk._errors import BulkRunException, InvalidRunError
-from promptflow._sdk._load_functions import load_flow
 from promptflow._sdk._utils import (
     PromptflowIgnoreFile,
     generate_flow_tools_json,
+    is_flex_run,
     json_dump,
     json_load,
     json_loads_parse_const_as_str,
@@ -36,7 +35,7 @@
     write_open,
 )
 from promptflow._sdk.entities import Run
-from promptflow._sdk.entities._flows import FlexFlow, Flow
+from promptflow._sdk.entities._flows import Flow
 from promptflow._utils.exception_utils import PromptflowExceptionPresenter
 from promptflow._utils.flow_utils import is_prompty_flow
 from promptflow._utils.logger_utils import LogContext, get_cli_sdk_logger
@@ -201,9 +200,10 @@ def __init__(self, run: Run, stream=False, run_mode=RunMode.Test):
             run_mode=run_mode,
             flow_logs_folder=self.path / LocalStorageFilenames.FLOW_LOGS_FOLDER,
         )
+        self._eager_mode = is_flex_run(run)
         # snapshot
         self._snapshot_folder_path = prepare_folder(self.path / LocalStorageFilenames.SNAPSHOT_FOLDER)
-        self._dag_path = self._snapshot_folder_path / FLOW_DAG_YAML
+        self._dag_path = self._snapshot_folder_path / (FLOW_FLEX_YAML if self._eager_mode else FLOW_DAG_YAML)
         self._flow_tools_json_path = (
             self._snapshot_folder_path / PROMPT_FLOW_DIR_NAME / LocalStorageFilenames.FLOW_TOOLS_JSON
         )
@@ -229,28 +229,12 @@ def __init__(self, run: Run, stream=False, run_mode=RunMode.Test):
         self._exception_path = self.path / LocalStorageFilenames.EXCEPTION
 
         self._dump_meta_file()
-        self._eager_mode = self._calculate_eager_mode(run)
         self._is_prompty_flow = is_prompty_flow(run.flow)
 
     @property
     def eager_mode(self) -> bool:
         return self._eager_mode
 
-    @classmethod
-    def _calculate_eager_mode(cls, run: Run) -> bool:
-        if run._run_source == RunInfoSources.LOCAL:
-            try:
-                flow_obj = load_flow(source=run.flow)
-                return isinstance(flow_obj, FlexFlow)
-            except Exception as e:
-                # For run with incomplete flow snapshot, ignore load flow error to make sure it can still show.
-                logger.debug(f"Failed to load flow from {run.flow} due to {e}.")
-                return False
-        elif run._run_source in [RunInfoSources.INDEX_SERVICE, RunInfoSources.RUN_HISTORY]:
-            return run._properties.get("azureml.promptflow.run_mode") == "Eager"
-        # TODO(2901279): support eager mode for run created from run folder
-        return False
-
     def delete(self) -> None:
         def on_rmtree_error(func, path, exc_info):
             raise InvalidRunError(f"Failed to delete run {self.path} due to {exc_info[1]}.")

src/promptflow-devkit/promptflow/_sdk/operations/_run_operations.py

@@ -16,6 +16,7 @@
     FlowRunProperties,
     ListViewType,
     RunInfoSources,
+    RunMode,
     RunStatus,
 )
 from promptflow._sdk._errors import InvalidRunStatusError, RunExistsError, RunNotFoundError, RunOperationParameterError
@@ -360,7 +361,7 @@ def _visualize(self, runs: List[Run], html_path: Optional[str] = None) -> None:
                 metrics=local_storage.load_metrics(parse_const_as_str=True),
                 dag=local_storage.load_dag_as_string(),
                 flow_tools_json=local_storage.load_flow_tools_json(),
-                mode="eager" if local_storage.eager_mode else "",
+                mode=RunMode.EAGER.lower() if local_storage.eager_mode else "",
             )
             details.append(copy.deepcopy(detail))
             metadatas.append(asdict(metadata))