[prompty] prompty support submit to remote

src/promptflow-azure/promptflow/azure/_entities/_flow.py

@@ -15,7 +15,12 @@
 from promptflow._sdk._constants import SERVICE_FLOW_TYPE_2_CLIENT_FLOW_TYPE, AzureFlowSource, FlowType
 from promptflow._sdk._utilities.general_utils import PromptflowIgnoreFile, load_yaml, remove_empty_element_from_dict
 from promptflow._sdk._utilities.signature_utils import update_signatures
-from promptflow._utils.flow_utils import dump_flow_yaml_to_existing_path, load_flow_dag, resolve_flow_path
+from promptflow._utils.flow_utils import (
+    dump_flow_yaml_to_existing_path,
+    is_prompty_flow,
+    load_flow_dag,
+    resolve_flow_path,
+)
 from promptflow._utils.logger_utils import LoggerFactory
 from promptflow.azure._ml import AdditionalIncludesMixin, Code
 from promptflow.core._model_configuration import MODEL_CONFIG_NAME_2_CLASS
@@ -54,14 +59,20 @@ def __init__(
         self.flow_portal_url = kwargs.get("flow_portal_url", None)
         # flow's environment, used to calculate session id, value will be set after flow is resolved to code.
         self._environment = {}
+        self._is_prompty_flow = is_prompty_flow(path)
 
         if self._flow_source == AzureFlowSource.LOCAL:
             absolute_path = self._validate_flow_from_source(path)
             # flow snapshot folder
             self.code = absolute_path.parent.as_posix()
             self._code_uploaded = False
             self.path = absolute_path.name
-            self._flow_dict = self._load_flow_yaml(absolute_path)
+            if self._is_prompty_flow:
+                from promptflow.core._flow import Prompty
+
+                self._flow_dict = Prompty.load(source=absolute_path)._data
+            else:
+                self._flow_dict = self._load_flow_yaml(absolute_path)
             self.display_name = self.display_name or absolute_path.parent.name
             self.description = description or self._flow_dict.get("description", None)
             self.tags = tags or self._flow_dict.get("tags", None)
@@ -148,7 +159,10 @@ def _try_build_local_code(self) -> Optional[Code]:
             dag_updated = False
             if isinstance(code, Code):
                 flow_dir = Path(code.path)
-                _, flow_dag = load_flow_dag(flow_path=flow_dir)
+                if self._is_prompty_flow:
+                    flow_dag = self._flow_dict
+                else:
+                    _, flow_dag = load_flow_dag(flow_path=flow_dir)
                 original_flow_dag = copy.deepcopy(flow_dag)
                 if self._get_all_additional_includes_configs():
                     # Remove additional include in the flow yaml.
@@ -159,22 +173,23 @@ def _try_build_local_code(self) -> Optional[Code]:
                 code.datastore = DEFAULT_STORAGE
                 dag_updated = self._resolve_requirements(flow_dir, flow_dag) or dag_updated
 
-                # generate .promptflow/flow.json for csharp flow as it's required to infer signature for csharp flow
-                flow_directory, flow_file = resolve_flow_path(code.path)
-                ProxyFactory().create_inspector_proxy(self.language).prepare_metadata(
-                    flow_file=flow_directory / flow_file, working_dir=flow_directory, init_kwargs=self._init_kwargs
-                )
+                if not self._is_prompty_flow:
+                    # generate .promptflow/flow.json for csharp flow as it's required to infer signature for csharp flow
+                    flow_directory, flow_file = resolve_flow_path(code.path)
+                    ProxyFactory().create_inspector_proxy(self.language).prepare_metadata(
+                        flow_file=flow_directory / flow_file, working_dir=flow_directory, init_kwargs=self._init_kwargs
+                    )
                 dag_updated = update_signatures(code=flow_dir, data=flow_dag) or dag_updated
                 # validate init kwargs with signature
                 self._validate_init_kwargs(init_signatures=flow_dag.get("init"), init_kwargs=self._init_kwargs)
                 # validate and resolve environment
                 self._environment = self._resolve_environment(flow_dir, flow_dag)
-                if dag_updated:
+                if dag_updated and not self._is_prompty_flow:
                     dump_flow_yaml_to_existing_path(flow_dag, flow_dir)
             try:
                 yield code
             finally:
-                if dag_updated:
+                if dag_updated and not self._is_prompty_flow:
                     dump_flow_yaml_to_existing_path(original_flow_dag, flow_dir)
 
     def _get_base_path_for_code(self) -> Path:

src/promptflow-azure/promptflow/azure/operations/_run_operations.py

@@ -735,6 +735,7 @@ def _resolve_flow_and_session_id(self, run: Run) -> Tuple[str, Optional[str]]:
         # for remote flow case, leave session id to None and let service side resolve
         if run._use_remote_flow:
             return self._resolve_flow_definition_resource_id(run=run), None
+
         flow = load_flow(run.flow)
         # set init kwargs for validation
         flow._init_kwargs = run.init

src/promptflow-azure/tests/sdk_cli_azure_test/e2etests/test_cli_with_azure.py

@@ -4,7 +4,9 @@
 import contextlib
 import os
 import sys
+import tempfile
 import uuid
+from pathlib import Path
 from typing import Callable
 
 import pytest
@@ -14,6 +16,7 @@
 
 from promptflow._constants import PF_USER_AGENT
 from promptflow._sdk.entities import Run
+from promptflow._utils.context_utils import _change_working_dir
 from promptflow._utils.user_agent_utils import ClientUserAgentUtil
 from promptflow._utils.utils import environment_variable_overwrite, parse_ua_to_dict
 from promptflow.azure import PFClient
@@ -22,6 +25,7 @@
 from .._azure_utils import DEFAULT_TEST_TIMEOUT, PYTEST_TIMEOUT_METHOD
 
 RUNS_DIR = PROMPTFLOW_ROOT / "tests/test_configs/runs"
+PROMPTY_DIR = PROMPTFLOW_ROOT / "tests/test_configs/prompty"
 
 
 # TODO: move this to a shared utility module
@@ -205,3 +209,52 @@ def check_workspace_info(*args, **kwargs):
                 if "obj must be an instance or subtype of type" in str(e):
                     pass
                 raise
+
+    @pytest.mark.skipif(pytest.is_replay, reason="Skip to avoid expose secret in record.")
+    @pytest.mark.usefixtures("use_secrets_config_file", "setup_local_connection")
+    def test_azure_run_prompty(self, pf, runtime: str, azure_open_ai_connection, randstr: Callable[[str], str]) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            env_file = Path(temp_dir) / ".env"
+            env = {
+                "MOCK_AZURE_DEVELOPMENT": "gpt-35-turbo",
+                "MOCK_AZURE_API_KEY": azure_open_ai_connection.api_key,
+                "MOCK_AZURE_API_VERSION": azure_open_ai_connection.api_version,
+                "MOCK_AZURE_ENDPOINT": azure_open_ai_connection.api_base,
+            }
+            with open(env_file, "w") as f:
+                f.writelines([f"{key}={value}\n" for key, value in env.items()])
+
+            with _change_working_dir(temp_dir):
+                # Submit with .env
+                name = randstr("name")
+                run_pf_command(
+                    "run",
+                    "create",
+                    "--flow",
+                    f"{PROMPTY_DIR}/prompty_with_env.prompty",
+                    "--data",
+                    f"{DATAS_DIR}/prompty_inputs.jsonl",
+                    "--name",
+                    name,
+                    "--env",
+                    str(env_file),
+                    pf=pf,
+                )
+                run = pf.runs.get(run=name)
+                assert isinstance(run, Run)
+
+                # Submit prompty with connection
+                name = randstr("name")
+                run_pf_command(
+                    "run",
+                    "create",
+                    "--flow",
+                    f"{PROMPTY_DIR}/prompty_example.prompty",
+                    "--data",
+                    f"{DATAS_DIR}/prompty_inputs.jsonl",
+                    "--name",
+                    name,
+                    pf=pf,
+                )
+                run = pf.runs.get(run=name)
+                assert isinstance(run, Run)

src/promptflow-devkit/promptflow/_sdk/_utilities/general_utils.py

@@ -70,7 +70,7 @@
     UnsecureConnectionError,
 )
 from promptflow._sdk._vendor import IgnoreFile, get_ignore_file, get_upload_files_from_folder
-from promptflow._utils.flow_utils import is_flex_flow, resolve_flow_path
+from promptflow._utils.flow_utils import is_flex_flow, is_prompty_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
@@ -236,8 +236,10 @@ def _sanitize_python_variable_name(name: str):
 
 
 def _get_additional_includes(yaml_path):
-    flow_dag = load_yaml(yaml_path)
-    return flow_dag.get("additional_includes", [])
+    if not is_prompty_flow(yaml_path):
+        flow_dag = load_yaml(yaml_path)
+        return flow_dag.get("additional_includes", [])
+    return []
 
 
 def _is_folder_to_compress(path: Path) -> bool: