[SDK]flow as func POC

.pre-commit-config.yaml

@@ -1,7 +1,7 @@
 # See https://pre-commit.com for more information
 # See https://pre-commit.com/hooks.html for more hooks
 
-exclude: '(^docs/)|flows|examples|scripts|src/promptflow/promptflow/azure/_restclient/|src/promptflow/tests/test_configs|src/promptflow-tools'
+exclude: '(^docs/)|flows|scripts|src/promptflow/promptflow/azure/_restclient/|src/promptflow/tests/test_configs|src/promptflow-tools'
 
 repos:
     - repo: https://github.com/pre-commit/pre-commit-hooks
@@ -24,11 +24,11 @@ repos:
       hooks:
           - id: flake8
     # Temporary disable this since it gets stuck when updating env
-#    - repo: https://github.com/streetsidesoftware/cspell-cli
-#      rev: v7.3.0
-#      hooks:
-#      - id: cspell
-#        args: ['--config', '.cspell.json', "--no-must-find-files"]
+    - repo: https://github.com/streetsidesoftware/cspell-cli
+      rev: v7.3.0
+      hooks:
+      - id: cspell
+        args: ['--config', '.cspell.json', "--no-must-find-files"]
     - repo: https://github.com/hadialqattan/pycln
       rev: v2.1.2 # Possible releases: https://github.com/hadialqattan/pycln/tags
       hooks:

examples/tutorials/flow-deploy/create-service-with-flow/readme.md

@@ -0,0 +1,6 @@
+# Create service with flow
+
+You can create your own service with a flow.
+This folder contains a example on how to build a service with a flow.
+Reference [here](./simple_score.py) for a minimal service example.
+Reference [here](./sample.ipynb) fro sample usages of flow-as-a-function.

examples/tutorials/flow-deploy/create-service-with-flow/sample.ipynb

@@ -0,0 +1,128 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Example1: Load flow as a function with inputs"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from promptflow import load_flow\n",
+    "\n",
+    "\n",
+    "flow_path=\"../../flows/standard/web-classification\"\n",
+    "sample_url = \"https://www.youtube.com/watch?v=o5ZQyXaAv1g\"\n",
+    "\n",
+    "f = load_flow(source=flow_path)\n",
+    "result = f(url=sample_url)\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": [
+    "f = load_flow(\n",
+    "    source=flow_path,\n",
+    ")\n",
+    "# need to create the connection\n",
+    "f.context.connections={\"classify_with_llm\": {\"connection\": \"new_ai_connection\"}}\n",
+    "\n",
+    "result = f(url=sample_url)\n",
+    "\n",
+    "print(result)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Example 3: Local flow as a function with flow inputs overwrite"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from promptflow.entities import FlowContext\n",
+    "\n",
+    "f = load_flow(source=flow_path)\n",
+    "f.context = FlowContext(\n",
+    "    # node \"fetch_text_content_from_url\" will take inputs from the following command instead of from flow input\n",
+    "    overrides={\"nodes.fetch_text_content_from_url.inputs.url\": sample_url},\n",
+    ")\n",
+    "# the url=\"unknown\" will not take effect\n",
+    "result = f(url=\"unknown\")\n",
+    "print(result)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Example 4: Load flow as a function with streaming output"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "f = load_flow(source=\"../../flows/chat/basic-chat\")\n",
+    "f.context.streaming = True\n",
+    "result = f(\n",
+    "    chat_history=[\n",
+    "        {\n",
+    "            \"inputs\": {\"chat_input\": \"Hi\"}, \n",
+    "            \"outputs\": {\"chat_output\": \"Hello! How can I assist you today?\"}\n",
+    "        }\n",
+    "    ]\n",
+    ")\n",
+    "\n",
+    "# the result will be a generator, iterate it to get the result\n",
+    "for r in result:\n",
+    "    print(r)"
+   ]
+  }
+ ],
+ "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
+}

examples/tutorials/flow-deploy/create-service-with-flow/simple_score.py

@@ -0,0 +1,73 @@
+import json
+import logging
+
+from flask import Flask, jsonify, request
+
+from promptflow import load_flow
+from promptflow.entities import FlowContext
+from promptflow.exceptions import SystemErrorException, UserErrorException
+
+
+class SimpleScoreApp(Flask):
+    pass
+
+
+app = SimpleScoreApp(__name__)
+logger = logging.getLogger(__name__)
+
+
+@app.errorhandler(Exception)
+def handle_error(e):
+    if isinstance(e, UserErrorException):
+        return jsonify({"message": e.message}), 400
+    elif isinstance(e, SystemErrorException):
+        return jsonify({"message": e.message}), 500
+    else:
+        from promptflow._internal import ErrorResponse, ExceptionPresenter
+
+        # handle other unexpected errors, can use internal class to format them
+        # but interface may change in the future
+        presenter = ExceptionPresenter.create(e)
+        resp = ErrorResponse(presenter.to_dict())
+        response_code = resp.response_code
+        return jsonify(resp.to_simplified_dict()), response_code
+
+
+@app.route("/health", methods=["GET"])
+def health():
+    """Check if the runtime is alive."""
+    return {"status": "Healthy"}
+
+
+@app.route("/score", methods=["POST"])
+def score():
+    """process a flow request in the runtime."""
+    raw_data = request.get_data()
+    logger.info(f"Start loading request data '{raw_data}'.")
+    data = json.loads(raw_data)
+
+    # load flow as a function
+    f = load_flow("../../flows/standard/web-classification")
+    # configure flow contexts
+    f.context = FlowContext(
+        # override flow connections, the overrides may come from the request
+        # connections={"classify_with_llm.connection": "another_ai_connection"},
+        # override the flow nodes' inputs or other flow configs, the overrides may come from the request
+        # **Note**: after this change, node "fetch_text_content_from_url" will take inputs from the
+        # following command instead of from flow input
+        overrides={"nodes.fetch_text_content_from_url.inputs.url": data["url"]},
+    )
+    result_dict = f(url="not used")
+
+    # Note: if specified streaming=True in the flow context, the result will be a generator
+    # reference promptflow._sdk._serving.response_creator.ResponseCreator on how to handle it in app.
+    return jsonify(result_dict)
+
+
+def create_app(**kwargs):
+    return app
+
+
+if __name__ == "__main__":
+    # test this with curl -X POST http://127.0.0.1:5000/score --header "Content-Type: application/json" --data '{\"url\": \"https://www.youtube.com/watch?v=o5ZQyXaAv1g\"}'  # noqa: E501
+    create_app().run()

src/promptflow/promptflow/_core/connection_manager.py

@@ -12,6 +12,7 @@
 from promptflow._constants import CONNECTION_NAME_PROPERTY, CONNECTION_SECRET_KEYS, PROMPTFLOW_CONNECTIONS
 from promptflow._sdk._constants import CustomStrongTypeConnectionConfigs
 from promptflow._utils.utils import try_import
+from promptflow.connections import _Connection
 from promptflow.contracts.tool import ConnectionType
 from promptflow.contracts.types import Secret
 
@@ -41,6 +42,10 @@ def _build_connections(cls, _dict: Dict[str, dict]):
         cls.import_requisites(_dict)
         connections = {}  # key to connection object
         for key, connection_dict in _dict.items():
+            if isinstance(connection_dict, _Connection):
+                # support directly pass connection object to executor
+                connections[key] = connection_dict
+                continue
             typ = connection_dict.get("type")
             if typ not in cls_mapping:
                 supported = [key for key in cls_mapping.keys() if not key.startswith("_")]
@@ -109,6 +114,9 @@ def import_requisites(cls, _dict: Dict[str, dict]):
         """Import connection required modules."""
         modules = set()
         for key, connection_dict in _dict.items():
+            if isinstance(connection_dict, _Connection):
+                # support directly pass connection object to executor
+                continue
             module = connection_dict.get("module")
             if module:
                 modules.add(module)

src/promptflow/promptflow/_sdk/_load_functions.py

@@ -10,7 +10,7 @@
 from ._utils import load_yaml
 from .entities import Run
 from .entities._connection import CustomConnection, _Connection
-from .entities._flow import ProtectedFlow
+from .entities._flow import Flow, ProtectedFlow
 
 
 def load_common(
@@ -62,28 +62,27 @@ def load_common(
 
 def load_flow(
     source: Union[str, PathLike, IO[AnyStr]],
-    **kwargs,
-):
+) -> Flow:
     """Load flow from YAML file.
 
-    :param source: The local yaml source of a flow. Must be either a path to a local file.
+    :param source: The local yaml source of a flow. Must be a path to a local file.
         If the source is a path, it will be open and read.
         An exception is raised if the file does not exist.
     :type source: Union[PathLike, str]
     :return: A Flow object
     :rtype: Flow
     """
-    return ProtectedFlow.load(source, **kwargs)
+    return ProtectedFlow.load(source)
 
 
 def load_run(
     source: Union[str, PathLike, IO[AnyStr]],
     params_override: Optional[list] = None,
     **kwargs,
-):
+) -> Run:
     """Load run from YAML file.
 
-    :param source: The local yaml source of a run. Must be either a path to a local file.
+    :param source: The local yaml source of a run. Must be a path to a local file.
         If the source is a path, it will be open and read.
         An exception is raised if the file does not exist.
     :type source: Union[PathLike, str]

src/promptflow/promptflow/_sdk/_utils.py

@@ -53,7 +53,6 @@
     CommonYamlFields,
 )
 from promptflow._sdk._errors import (
-    ConnectionNotFoundError,
     DecryptConnectionError,
     GenerateFlowToolsJsonError,
     StoreConnectionEncryptionKeyError,
@@ -639,7 +638,6 @@ def _gen_dynamic_list(function_config: Dict) -> List:
     from promptflow._cli._utils import get_workspace_triad_from_local
 
     workspace_triad = get_workspace_triad_from_local()
-
     if workspace_triad.subscription_id and workspace_triad.resource_group_name and workspace_triad.workspace_name:
         return gen_dynamic_list(func_path, func_kwargs, workspace_triad._asdict())
     # if no workspace triple available, just skip.
@@ -802,21 +800,21 @@ def is_template(path: str):
             )
 
 
-def get_local_connections_from_executable(executable, client):
+def get_local_connections_from_executable(executable, client, connections_to_ignore: List[str] = None):
     """Get local connections from executable.
 
-    Please avoid using this function anymore, and we should remove this function once all references are removed.
+    executable: The executable flow object.
+    client: Local client to get connections.
+    connections_to_ignore: The connection names to ignore when getting connections.
     """
 
     connection_names = executable.get_connection_names()
+    connections_to_ignore = connections_to_ignore or []
     result = {}
     for n in connection_names:
-        try:
+        if n not in connections_to_ignore:
             conn = client.connections.get(name=n, with_secrets=True)
             result[n] = conn._to_execution_connection_dict()
-        except ConnectionNotFoundError:
-            # ignore when connection not found since it can be configured with env var.
-            raise Exception(f"Connection {n!r} required for flow {executable.name!r} is not found.")
     return result
 
 

src/promptflow/promptflow/_sdk/entities/__init__.py

@@ -19,9 +19,10 @@
 )
 from ._run import Run
 from ._validation import ValidationResult
+from ._flow import FlowContext
 
 __all__ = [
-    # Connection
+    # region: Connection
     "AzureContentSafetyConnection",
     "AzureOpenAIConnection",
     "OpenAIConnection",
@@ -32,7 +33,12 @@
     "QdrantConnection",
     "WeaviateConnection",
     "FormRecognizerConnection",
-    # Run
+    # endregion
+    # region Run
     "Run",
     "ValidationResult",
+    # endregion
+    # region Flow
+    "FlowContext",
+    # endregion
 ]