move chat group constants to orchestrator

Author: Xiaodi Bi

src/promptflow-core/promptflow/_constants.py

@@ -30,12 +30,6 @@
 
 CHAT_HISTORY = "chat_history"
 
-# chat group reserved
-CONVERSATION_HISTORY_EXPRESSION_KEY = "parent.conversation_history"
-CONVERSATION_HISTORY_EXPRESSION = f"${{{CONVERSATION_HISTORY_EXPRESSION_KEY}}}"
-CONVERSATION_HISTORY_OUTPUT_KEY = "conversation_history"
-CHAT_ROLE_KEY = "role"
-
 # Tool meta info
 ICON_DARK = "icon_dark"
 ICON_LIGHT = "icon_light"

src/promptflow-devkit/promptflow/_orchestrator/_init__.py renamed to src/promptflow-devkit/promptflow/_orchestrator/__init__.py

@@ -11,7 +11,11 @@
 from promptflow._utils.execution_utils import apply_default_value_for_input
 from promptflow._proxy._proxy_factory import ProxyFactory
 from promptflow._utils.logger_utils import bulk_logger
-from promptflow._constants import CONVERSATION_HISTORY_EXPRESSION, CONVERSATION_HISTORY_OUTPUT_KEY, CHAT_ROLE_KEY
+from promptflow._orchestrator._constants import (
+    CONVERSATION_HISTORY_EXPRESSION,
+    CONVERSATION_HISTORY_OUTPUT_KEY,
+    CHAT_ROLE_KEY
+)
 from promptflow._orchestrator._errors import (
     InvalidChatRoleCount,
     MissingConversationHistoryExpression,
@@ -84,7 +88,7 @@ def _create_executor_proxy(self, **kwargs) -> List[AbstractExecutorProxy]:
         for chat_role in self._chat_group_roles:
             executor_proxy = executor_proxy_factory.create_executor_proxy(
                 flow_file=chat_role._flow_file,
-                working_dir=chat_role.working_dir,
+                working_dir=chat_role._working_dir,
                 connections=chat_role._connections,
                 storage=self._storage,
                 language=chat_role.check_language_from_yaml(),
@@ -97,7 +101,7 @@ def _create_executor_proxy(self, **kwargs) -> List[AbstractExecutorProxy]:
                 init_kwargs=chat_role._init_kwargs,
                 **kwargs
             )
-            bulk_logger.info(f"Created executor proxy for role:{chat_role.role}. name: {chat_role.name}")
+            bulk_logger.info(f"Created executor proxy for role:{chat_role.role}. name: {chat_role._name}")
             executor_proxy_list.append(executor_proxy)
         return executor_proxy_list
 
@@ -145,17 +149,17 @@ async def _schedule_line_runs(
             if conversation_history_key is None:
                 bulk_logger.error(
                     f"Cannot find conversation expression mapping for "
-                    f"chat role: {chat_role.role}. name: {chat_role.name}"
+                    f"chat role: {chat_role.role}. name: {chat_role._name}"
                 )
                 message = (
                     f"Cannot find conversation expression mapping for "
-                    f"chat role: {chat_role.role}. name: {chat_role.name} "
+                    f"chat role: {chat_role.role}. name: {chat_role._name} "
                     f"Please use define {CONVERSATION_HISTORY_EXPRESSION} for a flow input."
                 )
                 raise MissingConversationHistoryExpression(message=message)
             chat_role_input[conversation_history_key] = conversation_history
             bulk_logger.info(
-                f"Start to execute turn {turn}. role: {chat_role.role}. name: {chat_role.name}"
+                f"Start to execute turn {turn}. role: {chat_role.role}. name: {chat_role._name}"
             )
 
             current_line_result = await executor_proxy.exec_line_async(chat_role_input, line_index, run_id)
@@ -168,13 +172,13 @@ async def _schedule_line_runs(
                 aggregation_inputs)
             bulk_logger.info(
                 f"Finish process line result for "
-                f"line number: {line_index}, turn:{turn}. role:{chat_role.role}, name: {chat_role.name}"
+                f"line number: {line_index}, turn:{turn}. role:{chat_role.role}, name: {chat_role._name}"
             )
 
-            if any(value == chat_role.stop_signal for value in current_line_result.output.values()):
+            if any(value == chat_role._stop_signal for value in current_line_result.output.values()):
                 bulk_logger.info(
                     f"Stop chat since current turn align with stop signal. "
-                    f"line number: {line_index}, turn:{turn}. role:{chat_role.role}, name: {chat_role.name}"
+                    f"line number: {line_index}, turn:{turn}. role:{chat_role.role}, name: {chat_role._name}"
                 )
                 break
 
@@ -222,19 +226,19 @@ def _process_batch_inputs(self, inputs: Dict[str, Any]):
             if len(conversation_history_mapping) == 0:
                 bulk_logger.error(
                     f"Cannot find conversation expression mapping for "
-                    f"chat role: {chat_role.role}. name: {chat_role.name}"
+                    f"chat role: {chat_role.role}. name: {chat_role._name}"
                 )
                 message = (
                     f"Cannot find conversation expression mapping for "
-                    f"chat role: {chat_role.role}. name: {chat_role.name} "
+                    f"chat role: {chat_role.role}. name: {chat_role._name} "
                     f"Please mapping {CONVERSATION_HISTORY_EXPRESSION} for a flow input."
                 )
                 raise MissingConversationHistoryExpression(message=message)
 
             if len(conversation_history_mapping) > 1:
                 bulk_logger.error(f"Multiple inputs mapping of {CONVERSATION_HISTORY_EXPRESSION}")
                 message = (
-                    f"chat role: {chat_role.role}. name: {chat_role.name} "
+                    f"chat role: {chat_role.role}. name: {chat_role._name} "
                     f"only accepts 1 inputs mapping for {CONVERSATION_HISTORY_EXPRESSION}"
                 )
                 raise MultipleConversationHistoryInputsMapping(message=message)
@@ -245,7 +249,7 @@ def _process_batch_inputs(self, inputs: Dict[str, Any]):
             }
 
             batch_input_processor = BatchInputsProcessor(
-                chat_role.working_dir,
+                chat_role._working_dir,
                 chat_role._flow_definition.inputs,
                 self._max_lines_count)
             batch_input = batch_input_processor._process_batch_inputs_line(inputs, cleaned_inputs_mapping)

src/promptflow-devkit/promptflow/_orchestrator/_chat_group_orchestrator.py

@@ -0,0 +1,9 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+# chat group reserved
+CONVERSATION_HISTORY_EXPRESSION_KEY = "parent.conversation_history"
+CONVERSATION_HISTORY_EXPRESSION = f"${{{CONVERSATION_HISTORY_EXPRESSION_KEY}}}"
+CONVERSATION_HISTORY_OUTPUT_KEY = "conversation_history"
+CHAT_ROLE_KEY = "role"

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

@@ -48,12 +48,8 @@ def __init__(self,
                  init_kwargs: Optional[Dict[str, Any]] = None,
                  **kwargs):
         self._role = role
-        if working_dir is None:
-            flow_path = Path(flow).resolve()
-        else:
-            flow_path = (working_dir / Path(flow)).resolve()
-        self._flow, self._flow_object = self._validate_flow(flow_path)
-        self._inputs, self._outputs = self._build_role_io(flow_path, inputs)
+        self._flow, self._flow_object = self._validate_flow(flow, working_dir)
+        self._inputs, self._outputs = self._build_role_io(self._flow, inputs)
 
         # Below properties are used for cloud chat group. It may have some duplicate with above ones
         # Will evaluate and refine in the second step.
@@ -75,7 +71,7 @@ def __init__(self,
             self._line_timeout_sec = line_timeout_sec
             self._init_kwargs = init_kwargs
 
-        logger.info(f"Created chat role {self.role!r} with flow {flow_path.as_posix()!r}")
+        logger.info(f"Created chat role {self.role!r} with flow {self._flow.as_posix()!r}")
 
     @property
     def role(self):
@@ -92,32 +88,13 @@ def outputs(self):
         """Outputs of the chat role"""
         return self._outputs
 
-    @property
-    def name(self):
-        """Name of the chat role"""
-        return self._name
-
-    @property
-    def stop_signal(self):
-        """Stop signal of the role"""
-        return self._stop_signal
-
-    @property
-    def flow(self):
-        """Flow definition of the chat role"""
-        return self._flow_definition
-
-    @property
-    def working_dir(self):
-        return self._working_dir
-
-    @property
-    def output_dir(self):
-        return self._output_dir
-
-    def _validate_flow(self, flow_path: Path):
+    def _validate_flow(self, flow: Union[str, PathLike], working_dir: Optional[Path] = None):
         """Validate flow"""
-        logger.debug(f"Validating chat role flow source {flow_path!r}")
+        logger.debug(f"Validating chat role flow source {flow!r}")
+        if working_dir is None:
+            flow_path = Path(flow).resolve()
+        else:
+            flow_path = (working_dir / Path(flow)).resolve()
         try:
             flow_object = load_flow(flow_path)
         except Exception as e:
@@ -188,7 +165,7 @@ def invoke(self, *args, **kwargs):
         return result
 
     def check_language_from_yaml(self):
-        flow_file = self._working_dir / self._flow_file if self.working_dir else self._flow_file
+        flow_file = self._working_dir / self._flow_file if self._working_dir else self._flow_file
         if flow_file.suffix.lower() == ".dll":
             return FlowLanguage.CSharp
         with open(flow_file, "r", encoding="utf-8") as fin:

src/promptflow-devkit/promptflow/_sdk/entities/_chat_group/_chat_role.py

@@ -68,4 +68,4 @@ def test_chat_role_flow_dag_file(self):
             name="copilot",
             working_dir=FLOWS_DIR / "chat_group_copilot",
         )
-        assert copilot.flow is not None
+        assert copilot._flow_definition is not None