[Executor] Raise readable error message for wrong entry (#2078)

# Description

Raise readable error message for wrong entry. If users provide an
invalid entry in flex flow, we should raise a user error to hint them
the correct format.

# 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.**
- [x] **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.
- [x] 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.

---------

Co-authored-by: Lina Tang <[email protected]>

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

@@ -319,3 +319,7 @@ def __init__(self, init_kwargs, ex):
             init_kwargs=init_kwargs,
             ex=ex,
         )
+
+
+class InvalidFlexFlowEntry(ValidationException):
+    pass

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

@@ -22,6 +22,8 @@
 from promptflow.connections import ConnectionProvider
 from promptflow.contracts.flow import Flow
 from promptflow.contracts.tool import ConnectionType
+from promptflow.exceptions import ErrorTarget
+from promptflow.executor._errors import InvalidFlexFlowEntry
 from promptflow.executor._result import LineResult
 from promptflow.storage import AbstractRunStorage
 from promptflow.storage._run_storage import DefaultRunStorage
@@ -238,8 +240,8 @@ def _parse_entry_func(self):
             if inspect.isfunction(self._entry):
                 return self._entry
             return self._entry.__call__
+        module_name, func_name = self._parse_flow_file()
         try:
-            module_name, func_name = self._parse_flow_file()
             module = importlib.import_module(module_name)
         except Exception as e:
             raise PythonLoadError(
@@ -289,5 +291,12 @@ def _parse_flow_file(self):
         with open(self._working_dir / self._flow_file, "r", encoding="utf-8") as fin:
             flow_dag = load_yaml(fin)
         entry = flow_dag.get("entry", "")
-        module_name, func_name = entry.split(":")
+        try:
+            module_name, func_name = entry.split(":")
+        except Exception as e:
+            raise InvalidFlexFlowEntry(
+                message_format="Invalid entry '{entry}'.The entry should be in the format of '<module>:<function>'.",
+                entry=entry,
+                target=ErrorTarget.EXECUTOR,
+            ) from e
         return module_name, func_name

src/promptflow-core/tests/core/e2etests/test_eager_flow.py

@@ -5,7 +5,7 @@
 
 from promptflow._core.tool_meta_generator import PythonLoadError
 from promptflow.contracts.run_info import Status
-from promptflow.executor._errors import FlowEntryInitializationError
+from promptflow.executor._errors import FlowEntryInitializationError, InvalidFlexFlowEntry
 from promptflow.executor._result import LineResult
 from promptflow.executor._script_executor import ScriptExecutor
 from promptflow.executor.flow_executor import FlowExecutor
@@ -133,7 +133,7 @@ def test_execute_init_func_with_user_error(self):
         [
             ("callable_flow_with_init_exception", FlowEntryInitializationError, "Failed to initialize flow entry with"),
             ("invalid_illegal_entry", PythonLoadError, "Failed to load python module for"),
-            ("incorrect_entry", PythonLoadError, "Failed to load python module for"),
+            ("incorrect_entry", InvalidFlexFlowEntry, "Invalid entry"),
         ],
     )
     def test_execute_func_with_user_error(self, flow_folder, expected_exception, expected_error_msg):

src/promptflow/tests/test_configs/eager_flows/dummy_flow_with_invalid_entry/flow.dag.yaml

@@ -0,0 +1 @@
+entry: invalid_entry