[CLI] Resolve additional includes for flow serve (#1733)

# 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 <[email protected]>

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

@@ -6,6 +6,7 @@
 import importlib
 import json
 import os
+import shutil
 import subprocess
 import sys
 import tempfile
@@ -40,6 +41,7 @@
 from promptflow._constants import LANGUAGE_KEY, FlowLanguage
 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
 from promptflow._utils.logger_utils import get_cli_sdk_logger
 
 DEFAULT_CONNECTION = "open_ai_connection"
@@ -467,14 +469,32 @@ def serve_flow_csharp(args, source):
         pass
 
 
+def _resolve_python_flow_additional_includes(source) -> Path:
+    # Resolve flow additional includes
+    from promptflow import load_flow
+
+    flow = load_flow(source)
+    with FlowOperations._resolve_additional_includes(flow.path) as resolved_flow_path:
+        if resolved_flow_path == flow.path:
+            return source
+        # Copy resolved flow to temp folder if additional includes exists
+        # Note: DO NOT use resolved flow path directly, as when inner logic raise exception,
+        # temp dir will fail due to file occupied by other process.
+        temp_flow_path = Path(tempfile.TemporaryDirectory().name)
+        shutil.copytree(src=resolved_flow_path.parent, dst=temp_flow_path, dirs_exist_ok=True)
+
+    return temp_flow_path
+
+
 def serve_flow_python(args, source):
     from promptflow._sdk._serving.app import create_app
 
     static_folder = args.static_folder
     if static_folder:
         static_folder = Path(static_folder).absolute().as_posix()
     config = list_of_dict_to_dict(args.config)
-    # Change working directory to model dir
+    source = _resolve_python_flow_additional_includes(source)
+    os.environ["PROMPTFLOW_PROJECT_PATH"] = source.absolute().as_posix()
     logger.info(f"Change working directory to model dir {source}")
     os.chdir(source)
     app = create_app(

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

@@ -611,12 +611,18 @@ def build(
                 env_var_names=env_var_names,
             )
 
+    @classmethod
     @contextlib.contextmanager
     def _resolve_additional_includes(cls, flow_dag_path: Path) -> Iterable[Path]:
         # TODO: confirm if we need to import this
         from promptflow._sdk._submitter import remove_additional_includes
 
-        if _get_additional_includes(flow_dag_path):
+        # Eager flow may not contain a yaml file, skip resolving additional includes
+        def is_yaml_file(file_path):
+            _, file_extension = os.path.splitext(file_path)
+            return file_extension.lower() in (".yaml", ".yml")
+
+        if is_yaml_file(flow_dag_path) and _get_additional_includes(flow_dag_path):
             # Merge the flow folder and additional includes to temp folder.
             # TODO: support a flow_dag_path with a name different from flow.dag.yaml
             with _merge_local_code_and_additional_includes(code_path=flow_dag_path.parent) as temp_dir:

src/promptflow/tests/sdk_cli_test/unittests/test_flow_serve.py

@@ -0,0 +1,22 @@
+from pathlib import Path
+
+import pytest
+from sdk_cli_test.conftest import MODEL_ROOT
+
+from promptflow._cli._pf._flow import _resolve_python_flow_additional_includes
+
+
+@pytest.mark.unittest
+def test_flow_serve_resolve_additional_includes():
+    # Assert flow path not changed if no additional includes
+    flow_path = (Path(MODEL_ROOT) / "web_classification").resolve().absolute().as_posix()
+    resolved_flow_path = _resolve_python_flow_additional_includes(flow_path)
+    assert flow_path == resolved_flow_path
+
+    # Assert additional includes are resolved correctly
+    flow_path = (Path(MODEL_ROOT) / "web_classification_with_additional_include").resolve().absolute().as_posix()
+    resolved_flow_path = _resolve_python_flow_additional_includes(flow_path)
+
+    assert (Path(resolved_flow_path) / "convert_to_dict.py").exists()
+    assert (Path(resolved_flow_path) / "fetch_text_content_from_url.py").exists()
+    assert (Path(resolved_flow_path) / "summarize_text_content.jinja2").exists()