Merge branch 'main' into main

Author: sonichi

autogen/agentchat/conversable_agent.py

@@ -755,6 +755,77 @@ def check_termination_and_human_reply(
 
         return False, None
 
+    async def a_check_termination_and_human_reply(
+        self,
+        messages: Optional[List[Dict]] = None,
+        sender: Optional[Agent] = None,
+        config: Optional[Any] = None,
+    ) -> Tuple[bool, Union[str, Dict, None]]:
+        """(async) Check if the conversation should be terminated, and if human reply is provided."""
+        if config is None:
+            config = self
+        if messages is None:
+            messages = self._oai_messages[sender]
+        message = messages[-1]
+        reply = ""
+        no_human_input_msg = ""
+        if self.human_input_mode == "ALWAYS":
+            reply = await self.a_get_human_input(
+                f"Provide feedback to {sender.name}. Press enter to skip and use auto-reply, or type 'exit' to end the conversation: "
+            )
+            no_human_input_msg = "NO HUMAN INPUT RECEIVED." if not reply else ""
+            # if the human input is empty, and the message is a termination message, then we will terminate the conversation
+            reply = reply if reply or not self._is_termination_msg(message) else "exit"
+        else:
+            if self._consecutive_auto_reply_counter[sender] >= self._max_consecutive_auto_reply_dict[sender]:
+                if self.human_input_mode == "NEVER":
+                    reply = "exit"
+                else:
+                    # self.human_input_mode == "TERMINATE":
+                    terminate = self._is_termination_msg(message)
+                    reply = await self.a_get_human_input(
+                        f"Please give feedback to {sender.name}. Press enter or type 'exit' to stop the conversation: "
+                        if terminate
+                        else f"Please give feedback to {sender.name}. Press enter to skip and use auto-reply, or type 'exit' to stop the conversation: "
+                    )
+                    no_human_input_msg = "NO HUMAN INPUT RECEIVED." if not reply else ""
+                    # if the human input is empty, and the message is a termination message, then we will terminate the conversation
+                    reply = reply if reply or not terminate else "exit"
+            elif self._is_termination_msg(message):
+                if self.human_input_mode == "NEVER":
+                    reply = "exit"
+                else:
+                    # self.human_input_mode == "TERMINATE":
+                    reply = await self.a_get_human_input(
+                        f"Please give feedback to {sender.name}. Press enter or type 'exit' to stop the conversation: "
+                    )
+                    no_human_input_msg = "NO HUMAN INPUT RECEIVED." if not reply else ""
+                    # if the human input is empty, and the message is a termination message, then we will terminate the conversation
+                    reply = reply or "exit"
+
+        # print the no_human_input_msg
+        if no_human_input_msg:
+            print(colored(f"\n>>>>>>>> {no_human_input_msg}", "red"), flush=True)
+
+        # stop the conversation
+        if reply == "exit":
+            # reset the consecutive_auto_reply_counter
+            self._consecutive_auto_reply_counter[sender] = 0
+            return True, None
+
+        # send the human reply
+        if reply or self._max_consecutive_auto_reply_dict[sender] == 0:
+            # reset the consecutive_auto_reply_counter
+            self._consecutive_auto_reply_counter[sender] = 0
+            return True, reply
+
+        # increment the consecutive_auto_reply_counter
+        self._consecutive_auto_reply_counter[sender] += 1
+        if self.human_input_mode != "NEVER":
+            print(colored("\n>>>>>>>> USING AUTO REPLY...", "red"), flush=True)
+
+        return False, None
+
     def generate_reply(
         self,
         messages: Optional[List[Dict]] = None,
@@ -891,6 +962,20 @@ def get_human_input(self, prompt: str) -> str:
         reply = input(prompt)
         return reply
 
+    async def a_get_human_input(self, prompt: str) -> str:
+        """(Async) Get human input.
+
+        Override this method to customize the way to get human input.
+
+        Args:
+            prompt (str): prompt for the human input.
+
+        Returns:
+            str: human input.
+        """
+        reply = input(prompt)
+        return reply
+
     def run_code(self, code, **kwargs):
         """Run the code and return the result.
 

test/agentchat/test_async_get_human_input.py

@@ -0,0 +1,35 @@
+import asyncio
+import autogen
+import pytest
+from test_assistant_agent import KEY_LOC, OAI_CONFIG_LIST
+
+
+@pytest.mark.asyncio
+async def test_async_get_human_input():
+    try:
+        import openai
+    except ImportError:
+        return
+    config_list = autogen.config_list_from_json(OAI_CONFIG_LIST, KEY_LOC)
+
+    # create an AssistantAgent instance named "assistant"
+    assistant = autogen.AssistantAgent(
+        name="assistant",
+        max_consecutive_auto_reply=2,
+        llm_config={"request_timeout": 600, "seed": 41, "config_list": config_list, "temperature": 0},
+    )
+
+    user_proxy = autogen.UserProxyAgent(name="user", human_input_mode="ALWAYS", code_execution_config=False)
+
+    async def custom_a_get_human_input(prompt):
+        return "This is a test"
+
+    user_proxy.a_get_human_input = custom_a_get_human_input
+
+    user_proxy.register_reply([autogen.Agent, None], autogen.ConversableAgent.a_check_termination_and_human_reply)
+
+    await user_proxy.a_initiate_chat(assistant, clear_history=True, message="Hello.")
+
+
+if __name__ == "__main__":
+    test_async_get_human_input()