[SDK]flow as func POC

examples/tutorials/flow-as-a-function/sample.ipynb

@@ -0,0 +1,104 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Example1: Load flow as a function with inputs"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 3,
+   "metadata": {},
+   "outputs": [
+    {
+     "ename": "ImportError",
+     "evalue": "cannot import name 'load_flow' from 'promptflow' (D:\\code\\PromptFlow\\src\\promptflow-sdk\\promptflow\\__init__.py)",
+     "output_type": "error",
+     "traceback": [
+      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
+      "\u001b[1;31mImportError\u001b[0m                               Traceback (most recent call last)",
+      "\u001b[1;32md:\\code\\prompt-flow\\examples\\tutorials\\flow-as-a-function\\sample.ipynb Cell 2\u001b[0m line \u001b[0;36m1\n\u001b[1;32m----> <a href='vscode-notebook-cell:/d%3A/code/prompt-flow/examples/tutorials/flow-as-a-function/sample.ipynb#W1sZmlsZQ%3D%3D?line=0'>1</a>\u001b[0m \u001b[39mfrom\u001b[39;00m \u001b[39mpromptflow\u001b[39;00m \u001b[39mimport\u001b[39;00m load_flow\n\u001b[0;32m      <a href='vscode-notebook-cell:/d%3A/code/prompt-flow/examples/tutorials/flow-as-a-function/sample.ipynb#W1sZmlsZQ%3D%3D?line=3'>4</a>\u001b[0m flow_path\u001b[39m=\u001b[39m\u001b[39m\"\u001b[39m\u001b[39m../../flows/standard/web-classification\u001b[39m\u001b[39m\"\u001b[39m\n\u001b[0;32m      <a href='vscode-notebook-cell:/d%3A/code/prompt-flow/examples/tutorials/flow-as-a-function/sample.ipynb#W1sZmlsZQ%3D%3D?line=5'>6</a>\u001b[0m f \u001b[39m=\u001b[39m load_flow(source\u001b[39m=\u001b[39mflow_path)\n",
+      "\u001b[1;31mImportError\u001b[0m: cannot import name 'load_flow' from 'promptflow' (D:\\code\\PromptFlow\\src\\promptflow-sdk\\promptflow\\__init__.py)"
+     ]
+    }
+   ],
+   "source": [
+    "from promptflow import load_flow\n",
+    "\n",
+    "\n",
+    "flow_path=\"../../flows/standard/web-classification\"\n",
+    "\n",
+    "f = load_flow(source=flow_path)\n",
+    "\n",
+    "result = f(url=\"https://www.youtube.com/watch?v=o5ZQyXaAv1g\")\n",
+    "\n",
+    "print(result)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Example2: Load flow as a function with connection overwrite"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "flow_path=\"../../flows/standard/web-classification\"\n",
+    "\n",
+    "f = load_flow(\n",
+    "    source=flow_path,\n",
+    "    # need to create the connection\n",
+    "    connections={\"classify_with_llm\": {\"connection\": \"not_exist\"}},\n",
+    ")\n",
+    "\n",
+    "result = f(url=\"https://www.youtube.com/watch?v=o5ZQyXaAv1g\")\n",
+    "\n",
+    "print(result)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "f = load_flow(\n",
+    "    source=flow_path,\n",
+    "    variant=\"${summarize_text_content.variant_0}\",\n",
+    ")\n",
+    "\n",
+    "result = f(url=\"https://www.youtube.com/watch?v=o5ZQyXaAv1g\")\n",
+    "\n",
+    "print(result)"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "github_v2",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.8.13"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

src/promptflow/promptflow/_sdk/_load_functions.py

@@ -62,9 +62,19 @@ def load_common(
 
 def load_flow(
     source: Union[str, PathLike, IO[AnyStr]],
+    connections=None,
+    variant=None,
     **kwargs,
 ):
-    return ProtectedFlow.load(source, **kwargs)
+    """Load a flow from source file. The loaded flow can be used as a function.
+
+    :param source: Source file path.
+    :param connections: Connections for the run.
+    :param variant: Variant of the run.
+    :param kwargs:
+    :return:
+    """
+    return ProtectedFlow.load(source, connections=connections, variant=variant, **kwargs)
 
 
 def load_run(

src/promptflow/promptflow/_sdk/entities/_flow.py

@@ -52,6 +52,9 @@ def __init__(
         self._code = Path(code)
         path = kwargs.pop("path", None)
         self._path = Path(path) if path else None
+        # TODO: store here or in FlowExecutor?
+        self.connections = kwargs.pop("connections", None) or {}
+        self.variant = kwargs.pop("variant", None) or {}
         super().__init__(**kwargs)
 
     @property
@@ -190,3 +193,17 @@ def outputs(self):
         return {k: v.type.value for k, v in self._executable.outputs.items()}
 
     # endregion
+
+    def __call__(self, *args, **kwargs):
+        # run flow test here
+        # TODO: cache the submitter?
+        from promptflow import PFClient
+        from promptflow._sdk.operations._test_submitter import TestSubmitter
+
+        if args:
+            raise UserErrorException("Flow can only be called with keyword arguments.")
+        client = PFClient()
+        # TODO: gap: variants, environment_variables
+        result = client.flows._test_flow(flow=self, inputs=kwargs, variant=self.variant)
+        TestSubmitter._raise_error_when_test_failed(result, show_trace=False)
+        return result.output

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

@@ -104,7 +104,30 @@ def _test(
 
         inputs = inputs or {}
         flow = load_flow(flow)
+        self._test_flow(
+            flow=flow,
+            variant=variant,
+            node=node,
+            inputs=inputs,
+            environment_variables=environment_variables,
+            stream_log=stream_log,
+            allow_generator_output=allow_generator_output,
+            **kwargs,
+        )
+
+    def _test_flow(
+        self,
+        flow,
+        variant=None,
+        node=None,
+        inputs=None,
+        environment_variables=None,
+        stream_log=True,
+        allow_generator_output: bool = True,
+        **kwargs,
+    ):
         config = kwargs.get("config", None)
+
         with TestSubmitter(flow=flow, variant=variant, config=config).init() as submitter:
             is_chat_flow, chat_history_input_name, _ = self._is_chat_flow(submitter.dataplane_flow)
             flow_inputs, dependency_nodes_outputs = submitter._resolve_data(

src/promptflow/promptflow/_sdk/operations/_test_submitter.py

@@ -48,7 +48,9 @@ def init(self):
             tuning_node, node_variant = parse_variant(self._variant)
         else:
             tuning_node, node_variant = None, None
-        with variant_overwrite_context(self._origin_flow.code, tuning_node, node_variant) as temp_flow:
+        with variant_overwrite_context(
+            self._origin_flow.code, tuning_node, node_variant, connections=self.flow.connections
+        ) as temp_flow:
             # TODO execute flow test in a separate process.
             with _change_working_dir(temp_flow.code):
                 self.flow = temp_flow

src/promptflow/tests/sdk_cli_test/e2etests/test_flow_test.py

@@ -4,6 +4,7 @@
 
 import pytest
 
+from promptflow import load_flow
 from promptflow._sdk._constants import LOGGER_NAME
 from promptflow._sdk._pf_client import PFClient
 from promptflow.exceptions import UserErrorException
@@ -193,3 +194,36 @@ def test_generate_tool_meta_in_additional_folder(self):
     def test_pf_test_with_non_english_input(self):
         result = _client.test(flow=f"{FLOWS_DIR}/flow_with_non_english_input")
         assert result["output"] == "Hello 日本語"
+
+    def test_flow_as_a_func(self):
+        f = load_flow(f"{FLOWS_DIR}/flow_with_non_english_input")
+        result = f()
+        assert result["output"] == "Hello 日本語"
+
+    def test_flow_as_a_func_with_inputs(self):
+        flow_path = Path(f"{FLOWS_DIR}/classification_accuracy_evaluation").absolute()
+        f = load_flow(flow_path)
+        result = f(variant_id="variant_0", groundtruth="Pdf", prediction="PDF")
+        assert result["grade"] == "Correct"
+
+    def test_flow_as_a_func_with_connection_overwrite(self):
+        from promptflow._sdk._errors import ConnectionNotFoundError
+
+        flow_path = Path(f"{FLOWS_DIR}/web_classification").absolute()
+        f = load_flow(
+            flow_path,
+            connections={"classify_with_llm": {"connection": "not_exist"}},
+        )
+        with pytest.raises(ConnectionNotFoundError) as e:
+            f(url="https://www.youtube.com/watch?v=o5ZQyXaAv1g")
+        assert "Connection 'not_exist' required for flow" in str(e.value)
+
+    def test_flow_as_a_func_with_variant(self):
+
+        flow_path = Path(f"{FLOWS_DIR}/web_classification").absolute()
+        f = load_flow(
+            flow_path,
+            variant="${summarize_text_content.variant_0}",
+        )
+
+        f(url="https://www.youtube.com/watch?v=o5ZQyXaAv1g")