functionality of manual history cleaning by user proxy added

autogen/agentchat/conversable_agent.py

@@ -759,16 +759,30 @@ def reset_consecutive_auto_reply_counter(self, sender: Optional[Agent] = None):
         else:
             self._consecutive_auto_reply_counter[sender] = 0
 
-    def clear_history(self, agent: Optional[Agent] = None):
+    def clear_history(self, recipient: Optional[Agent] = None, nr_messages_to_preserve: Optional[int] = None):
         """Clear the chat history of the agent.
 
         Args:
-            agent: the agent with whom the chat history to clear. If None, clear the chat history with all agents.
+            recipient: the agent with whom the chat history to clear. If None, clear the chat history with all agents.
+            nr_messages_to_preserve: the number of newest messages to preserve in the chat history.
         """
-        if agent is None:
-            self._oai_messages.clear()
+        if recipient is None:
+            if nr_messages_to_preserve:
+                for key in self._oai_messages:
+                    # Remove messages from history except last `nr_messages_to_preserve` messages.
+                    self._oai_messages[key] = self._oai_messages[key][-nr_messages_to_preserve:]
+            else:
+                self._oai_messages.clear()
         else:
-            self._oai_messages[agent].clear()
+            self._oai_messages[recipient].clear()
+            if nr_messages_to_preserve:
+                print(
+                    colored(
+                        "WARNING: `nr_preserved_messages` is ignored when clearing chat history with a specific agent.",
+                        "yellow",
+                    ),
+                    flush=True,
+                )
 
     def generate_oai_reply(
         self,

autogen/agentchat/groupchat.py

@@ -388,6 +388,12 @@ def run_chat(
                     raise
             if reply is None:
                 break
+
+            # check for "clear history" phrase in reply and activate clear history function if found
+            if type(reply) is dict and reply["role"] == "user":
+                if "CLEAR HISTORY" in reply["content"].upper():
+                    reply["content"] = self.clear_agents_history(reply["content"], groupchat)
+
             # The speaker sends the message without requesting a reply
             speaker.send(reply, self, request_reply=False)
             message = self.last_message(speaker)
@@ -464,3 +470,55 @@ def _raise_exception_on_async_reply_functions(self) -> None:
 
         for agent in self._groupchat.agents:
             agent._raise_exception_on_async_reply_functions()
+
+    def clear_agents_history(self, reply: str, groupchat: GroupChat) -> str:
+        """Clears history of messages for all agents or selected one. Can preserve selected number of last messages.
+
+        Args:
+            reply (str): Admin reply to analyse.
+            groupchat (GroupChat): GroupChat object.
+        """
+        # Split the reply into words
+        words = reply.split()
+        # Find the position of "clear" to determine where to start processing
+        clear_word_index = next(i for i in reversed(range(len(words))) if words[i].upper() == "CLEAR")
+        # Extract potential agent name and steps
+        words_to_check = words[clear_word_index + 2 : clear_word_index + 4]
+        nr_messages_to_preserve = None
+        agent_to_memory_clear = None
+
+        for word in words_to_check:
+            if word.isdigit():
+                nr_messages_to_preserve = int(word)
+            elif word[:-1].isdigit():  # for the case when number of messages is followed by dot or other sign
+                nr_messages_to_preserve = int(word[:-1])
+            else:
+                for agent in groupchat.agents:
+                    if agent.name == word:
+                        agent_to_memory_clear = agent
+                        break
+                    elif agent.name == word[:-1]:  # for the case when agent name is followed by dot or other sign
+                        agent_to_memory_clear = agent
+                        break
+        # clear history
+        if agent_to_memory_clear:
+            if nr_messages_to_preserve:
+                print(
+                    f"Clearing history for {agent_to_memory_clear.name} except last {nr_messages_to_preserve} messages."
+                )
+            else:
+                print(f"Clearing history for {agent_to_memory_clear.name}.")
+            agent_to_memory_clear.clear_history(nr_messages_to_preserve=nr_messages_to_preserve)
+        else:
+            if nr_messages_to_preserve:
+                print(f"Clearing history for all agents except last {nr_messages_to_preserve} messages.")
+            else:
+                print("Clearing history for all agents.")
+            for agent in groupchat.agents:
+                agent.clear_history(nr_messages_to_preserve=nr_messages_to_preserve)
+
+        # Reconstruct the reply without the "clear history" command and parameters
+        skip_words_number = 2 + int(bool(agent_to_memory_clear)) + int(bool(nr_messages_to_preserve))
+        reply = " ".join(words[:clear_word_index] + words[clear_word_index + skip_words_number :])
+
+        return reply

test/agentchat/test_groupchat.py

@@ -504,6 +504,80 @@ def test_selection_helpers():
         groupchat.manual_select_speaker()
 
 
+def test_clear_agents_history():
+    agent1 = autogen.ConversableAgent(
+        "alice",
+        max_consecutive_auto_reply=10,
+        human_input_mode="NEVER",
+        llm_config=False,
+        default_auto_reply="This is alice speaking.",
+    )
+    agent2 = autogen.ConversableAgent(
+        "bob",
+        max_consecutive_auto_reply=10,
+        human_input_mode="NEVER",
+        llm_config=False,
+        default_auto_reply="This is bob speaking.",
+    )
+    agent3 = autogen.ConversableAgent(
+        "sam",
+        max_consecutive_auto_reply=10,
+        human_input_mode="ALWAYS",
+        llm_config=False,
+    )
+    groupchat = autogen.GroupChat(agents=[agent1, agent2, agent3], messages=[], max_round=3)
+    group_chat_manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=False)
+
+    # testing pure "clear history" statement
+    with mock.patch.object(builtins, "input", lambda _: "clear history. How you doing?"):
+        agent1.initiate_chat(group_chat_manager, message="hello")
+    agent1_history = list(agent1._oai_messages.values())[0]
+    agent2_history = list(agent2._oai_messages.values())[0]
+    assert agent1_history == [{"content": "How you doing?", "name": "sam", "role": "user"}]
+    assert agent2_history == [{"content": "How you doing?", "name": "sam", "role": "user"}]
+
+    # testing clear history for defined agent
+    with mock.patch.object(builtins, "input", lambda _: "clear history bob. How you doing?"):
+        agent1.initiate_chat(group_chat_manager, message="hello")
+    agent1_history = list(agent1._oai_messages.values())[0]
+    agent2_history = list(agent2._oai_messages.values())[0]
+    assert agent1_history == [
+        {"content": "hello", "role": "assistant"},
+        {"content": "This is bob speaking.", "name": "bob", "role": "user"},
+        {"content": "How you doing?", "name": "sam", "role": "user"},
+    ]
+    assert agent2_history == [{"content": "How you doing?", "name": "sam", "role": "user"}]
+
+    # testing clear history with defined nr of messages to preserve
+    with mock.patch.object(builtins, "input", lambda _: "clear history 1. How you doing?"):
+        agent1.initiate_chat(group_chat_manager, message="hello")
+    agent1_history = list(agent1._oai_messages.values())[0]
+    agent2_history = list(agent2._oai_messages.values())[0]
+    assert agent1_history == [
+        {"content": "This is bob speaking.", "name": "bob", "role": "user"},
+        {"content": "How you doing?", "name": "sam", "role": "user"},
+    ]
+    assert agent2_history == [
+        {"content": "This is bob speaking.", "role": "assistant"},
+        {"content": "How you doing?", "name": "sam", "role": "user"},
+    ]
+
+    # testing clear history with defined agent and nr of messages to preserve
+    with mock.patch.object(builtins, "input", lambda _: "clear history bob 1. How you doing?"):
+        agent1.initiate_chat(group_chat_manager, message="hello")
+    agent1_history = list(agent1._oai_messages.values())[0]
+    agent2_history = list(agent2._oai_messages.values())[0]
+    assert agent1_history == [
+        {"content": "hello", "role": "assistant"},
+        {"content": "This is bob speaking.", "name": "bob", "role": "user"},
+        {"content": "How you doing?", "name": "sam", "role": "user"},
+    ]
+    assert agent2_history == [
+        {"content": "This is bob speaking.", "role": "assistant"},
+        {"content": "How you doing?", "name": "sam", "role": "user"},
+    ]
+
+
 if __name__ == "__main__":
     # test_func_call_groupchat()
     # test_broadcast()
@@ -515,3 +589,4 @@ def test_selection_helpers():
     # test_termination()
     # test_next_agent()
     test_invalid_allow_repeat_speaker()
+    test_clear_agents_history()