functionality of manual history cleaning by user proxy added (#1230)

* functionality of manual history cleaning by admin added

* formatting improved

* formatting improved 2

* formatting improved 3

* test function added

* test code formatting

* test code formatting 2

* more advanced logging. Now user can see nr of messages to preserve as confirmation

* test_invalid_allow_repeat_speaker uncommented

* warning when providing recepient agent and nr messages to preserve added, changed variables names

* code formatting

* code formatting

* code formatting

* added 'enable_clear_history' variable to GroupChat

* 'enable_clear_history' added, better descripted

* clearing groupchat history added

* clearing groupchat history added

* two ifs merged into one, formatting improved

* two ifs merged into one, formatting improved

* two ifs merged into one, formatting improved

* formatting

* formatting

---------

Co-authored-by: Davor Runje <[email protected]>
Co-authored-by: Chi Wang <[email protected]>

Author: 3 people

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

@@ -31,6 +31,9 @@ class GroupChat:
         - "random": the next speaker is selected randomly.
         - "round_robin": the next speaker is selected in a round robin fashion, i.e., iterating in the same order as provided in `agents`.
     - allow_repeat_speaker: whether to allow the same speaker to speak consecutively. Default is True, in which case all speakers are allowed to speak consecutively. If allow_repeat_speaker is a list of Agents, then only those listed agents are allowed to repeat. If set to False, then no speakers are allowed to repeat.
+    - enable_clear_history: enable possibility to clear history of messages for agents manually by providing
+        "clear history" phrase in user prompt. This is experimental feature.
+        See description of GroupChatManager.clear_agents_history function for more info.
     """
 
     agents: List[Agent]
@@ -40,6 +43,7 @@ class GroupChat:
     func_call_filter: Optional[bool] = True
     speaker_selection_method: Optional[str] = "auto"
     allow_repeat_speaker: Optional[Union[bool, List[Agent]]] = True
+    enable_clear_history: Optional[bool] = False
 
     _VALID_SPEAKER_SELECTION_METHODS = ["auto", "manual", "random", "round_robin"]
 
@@ -388,6 +392,14 @@ def run_chat(
                     raise
             if reply is None:
                 break
+
+            # check for "clear history" phrase in reply and activate clear history function if found
+            if (
+                groupchat.enable_clear_history
+                and isinstance(reply, dict)
+                and "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 +476,70 @@ 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.
+        That function is called when user manually provide "clear history" phrase in his reply.
+        When "clear history" is provided, the history of messages for all agents is cleared.
+        When "clear history <agent_name>" is provided, the history of messages for selected agent is cleared.
+        When "clear history <nr_of_messages_to_preserve>" is provided, the history of messages for all agents is cleared
+        except last <nr_of_messages_to_preserve> messages.
+        When "clear history <agent_name> <nr_of_messages_to_preserve>" is provided, the history of messages for selected
+        agent is cleared except last <nr_of_messages_to_preserve> messages.
+        Phrase "clear history" and optional arguments are cut out from the reply before it passed to the chat.
+
+        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.")
+                # clearing history for groupchat here
+                temp = groupchat.messages[-nr_messages_to_preserve:]
+                groupchat.messages.clear()
+                groupchat.messages.extend(temp)
+            else:
+                print("Clearing history for all agents.")
+                # clearing history for groupchat here
+                groupchat.messages.clear()
+            # clearing history for 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,95 @@ 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, enable_clear_history=True)
+    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"}]
+    assert groupchat.messages == [{"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"}]
+    assert groupchat.messages == [
+        {"content": "hello", "role": "user", "name": "alice"},
+        {"content": "This is bob speaking.", "name": "bob", "role": "user"},
+        {"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"},
+    ]
+    assert groupchat.messages == [
+        {"content": "This is bob speaking.", "role": "user", "name": "bob"},
+        {"content": "How you doing?", "role": "user", "name": "sam"},
+    ]
+
+    # 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"},
+    ]
+    assert groupchat.messages == [
+        {"content": "hello", "name": "alice", "role": "user"},
+        {"content": "This is bob speaking.", "name": "bob", "role": "user"},
+        {"content": "How you doing?", "name": "sam", "role": "user"},
+    ]
+
+
 if __name__ == "__main__":
     # test_func_call_groupchat()
     # test_broadcast()
@@ -514,4 +603,5 @@ def test_selection_helpers():
     # test_agent_mentions()
     # test_termination()
     # test_next_agent()
-    test_invalid_allow_repeat_speaker()
+    # test_invalid_allow_repeat_speaker()
+    test_clear_agents_history()