feat: chat UI debug service (#2927)

# Description

This PR is targeting to change the experience of Chat UI.

Previously, running `pf flow test --ui` will generate an url only, which
means:
1. need to initialize executor for each request
2. hard to track logs and tracebacks
3. depends on multiple local pfs

Now we will start a monitor on flow directory, which will try to
maintain a serve app align with latest flow directory:
1) the serve app won't be restarted unless there are changes in flow
definition or init parameters
2) logs and tracebacks will be redirect to the main cli process
3) Chat UI talk to serve app /score directly instead of calling local
pfs for each request


![image](https://github.com/microsoft/promptflow/assets/37076709/71426c0e-88a5-4a1b-9a17-a70097047599)

Limitations:
1. Signature change in flow definition require manual refresh to update
settings tab in Chat UI
2. `pf flow serve` doesn't support specifying flow file name for now, so
give below error message if specified flow file is not picked.

![image](https://github.com/microsoft/promptflow/assets/37076709/6bf9c591-abf3-4daa-b508-67712b896802)

basic flex flow chat ui

![image](https://github.com/microsoft/promptflow/assets/37076709/ca4cc0a4-028f-4a3f-a1f2-da404b8b405b)

basic prompty chat ui:

![image](https://github.com/microsoft/promptflow/assets/37076709/74fe7c53-04ff-42d1-8a2f-699c4cd25014)

dag flow:

![image](https://github.com/microsoft/promptflow/assets/37076709/47199129-a27a-475c-ac68-9fdf004f8c6f)

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

src/promptflow-core/promptflow/_utils/flow_utils.py

@@ -93,6 +93,7 @@ def resolve_flow_path(
     else:
         flow_path = Path(flow_path)
 
+    prompty_count = -1
     if flow_path.is_dir():
         flow_folder = flow_path
         flow_file = default_flow_file
@@ -112,6 +113,7 @@ def resolve_flow_path(
             )
         elif allow_prompty_dir and check_flow_exist:
             candidates = list(flow_folder.glob(f"*{PROMPTY_EXTENSION}"))
+            prompty_count = len(candidates)
             if len(candidates) == 1:
                 flow_file = candidates[0].name
     elif flow_path.is_file() or flow_path.suffix.lower() in FLOW_FILE_SUFFIX:
@@ -137,18 +139,23 @@ def resolve_flow_path(
             privacy_info=[flow_path.absolute().as_posix()],
         )
 
-    if not file_path.is_file():
-        if flow_folder == flow_path:
+    if not file_path.is_file() and flow_folder == flow_path:
+        msg = f"Have found neither flow.dag.yaml nor flow.flex.yaml in {flow_path.absolute().as_posix()}"
+        if prompty_count == 0 or not allow_prompty_dir:
             raise UserErrorException(
-                f"Flow path {flow_path.absolute().as_posix()} "
-                f"must have postfix either {FLOW_DAG_YAML} or {FLOW_FLEX_YAML}",
+                msg,
                 privacy_info=[flow_path.absolute().as_posix()],
             )
         else:
             raise UserErrorException(
-                f"Flow file {file_path.absolute().as_posix()} does not exist.",
-                privacy_info=[file_path.absolute().as_posix()],
+                msg + " and there are more than 1 prompty file.",
+                privacy_info=[flow_path.absolute().as_posix()],
             )
+    if not file_path.is_file():
+        raise UserErrorException(
+            f"Flow file {file_path.absolute().as_posix()} does not exist.",
+            privacy_info=[file_path.absolute().as_posix()],
+        )
 
     return flow_folder.resolve().absolute(), flow_file
 

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

@@ -8,7 +8,6 @@
 import os
 import sys
 import tempfile
-import webbrowser
 from pathlib import Path
 
 from promptflow._cli._params import (
@@ -41,9 +40,9 @@
 from promptflow._sdk._configuration import Configuration
 from promptflow._sdk._constants import PROMPT_FLOW_DIR_NAME
 from promptflow._sdk._pf_client import PFClient
-from promptflow._sdk._utilities.chat_utils import construct_chat_page_url
+from promptflow._sdk._utilities.chat_utils import start_chat_ui_service_monitor
 from promptflow._sdk._utilities.general_utils import generate_yaml_entry_without_delete
-from promptflow._sdk._utilities.serve_utils import start_flow_service
+from promptflow._sdk._utilities.serve_utils import find_available_port, start_flow_service
 from promptflow._utils.flow_utils import is_flex_flow
 from promptflow._utils.logger_utils import get_cli_sdk_logger
 from promptflow.exceptions import ErrorTarget, UserErrorException
@@ -298,6 +297,8 @@ def add_parser_test_flow(subparsers):
     add_param_url_params = lambda parser: parser.add_argument(  # noqa: E731
         "--url-params", action=AppendToDictAction, help=argparse.SUPPRESS, nargs="+"
     )
+    # add a private param to support specifying port for chat debug service
+    add_param_port = lambda parser: parser.add_argument("--port", type=str, help=argparse.SUPPRESS)  # noqa: E731
 
     add_params = [
         add_param_flow,
@@ -315,6 +316,7 @@ def add_parser_test_flow(subparsers):
         add_param_skip_browser,
         add_param_init,
         add_param_url_params,
+        add_param_port,
     ] + base_params
 
     if Configuration.get_instance().is_internal_features_enabled():
@@ -519,18 +521,19 @@ def _test_flow_multi_modal(args, pf_client):
         from promptflow._sdk._tracing import _invoke_pf_svc
 
         pfs_port = _invoke_pf_svc()
+        serve_app_port = args.port or find_available_port()
         flow = generate_yaml_entry_without_delete(entry=args.flow)
         # flex flow without yaml file doesn't support /eval in chat window
-        enable_internal_features = Configuration.get_instance().is_internal_features_enabled() or flow != args.flow
-        chat_page_url = construct_chat_page_url(
-            flow,
-            pfs_port,
-            list_of_dict_to_dict(args.url_params),
+        enable_internal_features = Configuration.get_instance().is_internal_features_enabled() and flow == args.flow
+        start_chat_ui_service_monitor(
+            flow=flow,
+            serve_app_port=serve_app_port,
+            pfs_port=pfs_port,
+            url_params=list_of_dict_to_dict(args.url_params),
+            init=list_of_dict_to_dict(args.init),
             enable_internal_features=enable_internal_features,
+            skip_open_browser=args.skip_open_browser,
         )
-        print(f"You can begin chat flow on {chat_page_url}")
-        if not args.skip_open_browser:
-            webbrowser.open(chat_page_url)
 
 
 def _test_flow_interactive(args, pf_client, inputs, environment_variables):

src/promptflow-devkit/promptflow/_sdk/_constants.py

@@ -159,6 +159,7 @@ def _prepare_home_dir() -> Path:
 SPAN_EVENTS_ATTRIBUTE_PAYLOAD = "payload"
 
 UX_INPUTS_JSON = "ux.inputs.json"
+UX_INPUTS_INIT_KEY = "init"
 AzureMLWorkspaceTriad = namedtuple("AzureMLWorkspace", ["subscription_id", "resource_group_name", "workspace_name"])
 
 # chat group

src/promptflow-devkit/promptflow/_sdk/_service/static/chat-window/index.html

@@ -14,7 +14,7 @@
         padding: 0;
       }
     </style>
-    <script type="module" crossorigin src="/v1.0/ui/chat/assets/index-iVCtt0ds.js"></script>
+    <script type="module" crossorigin src="/v1.0/ui/chat/assets/index-Wrqw_iL4.js"></script>
 
   </head>
   <body>

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

@@ -1,21 +1,170 @@
+import json
+import webbrowser
+from pathlib import Path
+from typing import Any, Dict
 from urllib.parse import urlencode, urlunparse
 
+from promptflow._constants import FlowLanguage
+from promptflow._sdk._constants import DEFAULT_ENCODING, PROMPT_FLOW_DIR_NAME, UX_INPUTS_INIT_KEY, UX_INPUTS_JSON
 from promptflow._sdk._service.utils.utils import encrypt_flow_path
+from promptflow._sdk._utilities.general_utils import resolve_flow_language
+from promptflow._sdk._utilities.monitor_utils import (
+    DirectoryModificationMonitorTarget,
+    JsonContentMonitorTarget,
+    Monitor,
+)
+from promptflow._sdk._utilities.serve_utils import CSharpServeAppHelper, PythonServeAppHelper, ServeAppHelper
 from promptflow._utils.flow_utils import resolve_flow_path
 
 
+def print_log(text):
+    print(text)
+
+
 def construct_flow_absolute_path(flow: str) -> str:
     flow_dir, flow_file = resolve_flow_path(flow)
     return (flow_dir / flow_file).absolute().resolve().as_posix()
 
 
 # Todo: use base64 encode for now, will consider whether need use encryption or use db to store flow path info
-def construct_chat_page_url(flow, port, url_params, enable_internal_features=False):
-    flow_path_dir, flow_path_file = resolve_flow_path(flow)
-    flow_path = str(flow_path_dir / flow_path_file)
+def construct_chat_page_url(flow_path, port, url_params):
     encrypted_flow_path = encrypt_flow_path(flow_path)
-    query_dict = {"flow": encrypted_flow_path}
-    if enable_internal_features:
-        query_dict.update({"enable_internal_features": "true", **url_params})
+    query_dict = {"flow": encrypted_flow_path, **url_params}
     query_params = urlencode(query_dict)
     return urlunparse(("http", f"127.0.0.1:{port}", "/v1.0/ui/chat", "", query_params, ""))
+
+
+def _try_restart_service(
+    *, last_result: ServeAppHelper, flow_file_name: str, flow_dir: Path, serve_app_port: int, ux_input_path: Path
+):
+    if last_result is not None:
+        print_log("Changes detected, stopping current serve app...")
+        last_result.terminate()
+
+    # init must be always loaded from ux_inputs.json
+    if not ux_input_path.is_file():
+        init = {}
+    else:
+        ux_inputs = json.loads(ux_input_path.read_text(encoding=DEFAULT_ENCODING))
+        init = ux_inputs.get(UX_INPUTS_INIT_KEY, {}).get(flow_file_name, {})
+
+    language = resolve_flow_language(flow_path=flow_file_name, working_dir=flow_dir)
+    if language == FlowLanguage.Python:
+        # additional includes will always be called by the helper.
+        # This is expected as user will change files in original locations only
+        helper = PythonServeAppHelper(
+            flow_file_name=flow_file_name,
+            flow_dir=flow_dir,
+            init=init,
+            port=serve_app_port,
+        )
+    else:
+        helper = CSharpServeAppHelper(
+            flow_file_name=flow_file_name,
+            flow_dir=flow_dir,
+            init=init,
+            port=serve_app_port,
+        )
+
+    print_log("Starting serve app...")
+    try:
+        helper.start()
+    except Exception:
+        print_log("Failed to start serve app, please check the error message above.")
+    return helper
+
+
+def update_init_in_ux_inputs(*, ux_input_path: Path, flow_file_name: str, init: Dict[str, Any]):
+    # ensure that ux_inputs.json is created or updated so that we can always load init from it in monitor
+    if not ux_input_path.is_file():
+        ux_input_path.parent.mkdir(exist_ok=True, parents=True)
+        ux_input_path.write_text(json.dumps({UX_INPUTS_INIT_KEY: {flow_file_name: init}}, indent=4, ensure_ascii=False))
+        return
+
+    # avoid updating init if it's not provided
+    if not init:
+        return
+
+    # update init to ux_inputs.json
+    current_ux_inputs = json.loads(ux_input_path.read_text(encoding=DEFAULT_ENCODING))
+    if UX_INPUTS_INIT_KEY not in current_ux_inputs:
+        current_ux_inputs[UX_INPUTS_INIT_KEY] = {}
+    # save init with different key given there can be multiple prompty flow in one directory
+    current_ux_inputs[UX_INPUTS_INIT_KEY][flow_file_name] = init
+    ux_input_path.parent.mkdir(exist_ok=True, parents=True)
+    ux_input_path.write_text(json.dumps(current_ux_inputs, indent=4, ensure_ascii=False), encoding="utf-8")
+
+
+touch_iter_count = 0
+
+
+def touch_local_pfs():
+    from promptflow._sdk._tracing import _invoke_pf_svc
+
+    global touch_iter_count
+
+    touch_iter_count += 1
+    # invoke every 20-30 minutes for now, so trigger every 12000
+    # iterations given 1 iteration takes around 0.1s if no change
+    if touch_iter_count % 12000 == 0:
+        _invoke_pf_svc()
+
+
+def start_chat_ui_service_monitor(
+    flow,
+    *,
+    serve_app_port: str,
+    pfs_port: str,
+    url_params: Dict[str, str],
+    init: Dict[str, Any],
+    enable_internal_features: bool = False,
+    skip_open_browser: bool = False,
+):
+    flow_dir, flow_file_name = resolve_flow_path(flow, allow_prompty_dir=True)
+
+    ux_input_path = flow_dir / PROMPT_FLOW_DIR_NAME / UX_INPUTS_JSON
+    update_init_in_ux_inputs(ux_input_path=ux_input_path, flow_file_name=flow_file_name, init=init)
+
+    # show url for chat UI
+    url_params["serve_app_port"] = serve_app_port
+    if "enable_internal_features" not in url_params:
+        url_params["enable_internal_features"] = "true" if enable_internal_features else "false"
+    chat_page_url = construct_chat_page_url(
+        str(flow_dir / flow_file_name),
+        pfs_port,
+        url_params=url_params,
+    )
+    print_log(f"You can begin chat flow on {chat_page_url}")
+    if not skip_open_browser:
+        webbrowser.open(chat_page_url)
+
+    monitor = Monitor(
+        targets=[
+            DirectoryModificationMonitorTarget(
+                target=flow_dir,
+                relative_root_ignores=[PROMPT_FLOW_DIR_NAME, "__pycache__"],
+            ),
+            JsonContentMonitorTarget(
+                target=ux_input_path,
+                node_path=[UX_INPUTS_INIT_KEY, flow_file_name],
+            ),
+        ],
+        target_callback=_try_restart_service,
+        target_callback_kwargs={
+            "flow_file_name": flow_file_name,
+            "flow_dir": flow_dir,
+            "serve_app_port": int(serve_app_port),
+            "ux_input_path": ux_input_path,
+        },
+        inject_last_callback_result=True,
+        extra_logic_in_loop=touch_local_pfs,
+    )
+
+    try:
+        monitor.start_monitor()
+    except KeyboardInterrupt:
+        print_log("Stopping monitor and attached serve app...")
+        serve_app_helper = monitor.last_callback_result
+        if serve_app_helper is not None:
+            serve_app_helper.terminate()
+        print_log("Stopped monitor and attached serve app.")