first version of loop detection

Useful for smaller models

ae/core/autogen_wrapper.py

@@ -11,6 +11,8 @@
 import nest_asyncio  # type: ignore
 import openai
 
+from collections import defaultdict
+
 #from autogen import Cache
 from ae.config import SOURCE_LOG_FOLDER_PATH
 from ae.core.agents.browser_nav_agent import BrowserNavAgent
@@ -19,6 +21,7 @@
 from ae.core.prompts import LLM_PROMPTS
 from ae.core.skills.get_url import geturl
 from ae.utils.autogen_sequential_function_call import UserProxyAgent_SequentialFunctionExecution
+from ae.utils.detect_llm_loops import is_agent_stuck_in_loop
 from ae.utils.logger import logger
 from ae.utils.response_parser import parse_response
 from ae.utils.ui_messagetype import MessageType
@@ -119,7 +122,7 @@ def my_custom_summary_method(sender: autogen.ConversableAgent,recipient: autogen
             self.__save_chat_log(list(messages_str_keys.values())[0]) # type: ignore
             last_message=recipient.last_message(sender)["content"] # type: ignore
             if not last_message or last_message.strip() == "": # type: ignore
-                return "I received an empty message. Try a different approach."
+                return "I received an empty message. This is not an empty and is recoverable. Try to reformulate the task..."
             elif "##TERMINATE TASK##" in last_message:
                 last_message=last_message.replace("##TERMINATE TASK##", "") # type: ignore
                 last_message=last_message+" "+  get_url() # type: ignore
@@ -280,10 +283,17 @@ def __create_browser_nav_executor_agent(self):
 
         """
         def is_browser_executor_termination_message(x: dict[str, str])->bool: # type: ignore
+
              tools_call:Any = x.get("tool_calls", "")
              if tools_call :
-                return False
+                chat_messages=self.agents_map["browser_nav_executor"].chat_messages #type: ignore
+                # Get the only key from the dictionary
+                agent_key = next(iter(chat_messages)) # type: ignore
+                # Get the chat messages corresponding to the only key
+                messages = chat_messages[agent_key] # type: ignore
+                return is_agent_stuck_in_loop(messages) # type: ignore
              else:
+                print("Terminating browser executor")
                 return True
 
         browser_nav_executor_agent = UserProxyAgent_SequentialFunctionExecution(

ae/utils/autogen_sequential_function_call.py

@@ -10,7 +10,8 @@
 class UserProxyAgent_SequentialFunctionExecution(UserProxyAgent):
     def __init__(self, *args, **kwargs): # type: ignore
         super().__init__(*args, **kwargs) # type: ignore
-        self.register_reply(Agent, UserProxyAgent_SequentialFunctionExecution.sequential_generate_tool_calls_reply) # type: ignore
+        #position = 2 allows termination check to be called earlier, this helps detect loops.
+        self.register_reply(Agent, UserProxyAgent_SequentialFunctionExecution.sequential_generate_tool_calls_reply, position=2) # type: ignore
 
 
     def sequential_generate_tool_calls_reply( # type: ignore

ae/utils/detect_llm_loops.py

@@ -0,0 +1,44 @@
+from typing import Any, Dict, List
+from ae.utils.logger import logger
+
+
+def is_agent_stuck_in_loop(messages: List[Dict[str, Any]]) -> bool:
+    """
+    Detects loops in the agent's responses by iterating over the last N responses.
+
+    Parameters
+    ----------
+    messages : List[Dict[str, Any]]
+        A list of dictionaries representing the agent's messages.
+
+    Returns
+    -------
+    bool
+        True if a loop is detected, False otherwise.
+    """
+    number_of_turns_to_check_for_loops = 6
+    # Detect any loops by checking the last 3 tool responses and their corresponding tool calls
+    if len(messages) > number_of_turns_to_check_for_loops:
+        last_six_items = messages[-number_of_turns_to_check_for_loops:]
+        logger.debug(f"More than {number_of_turns_to_check_for_loops} messages in the conversation. Checking for loops..")
+        # Filter items by role
+        tool_calls = [item for item in last_six_items if item.get("role") == "assistant"]
+
+        # Check if function attributes are the same for tool items
+        if tool_calls:
+            tool_functions = [item.get("tool_calls", [{}])[0].get("function") for item in tool_calls]
+            logger.debug(f"Last 3 tool calls: {tool_functions}")
+            if all(func == tool_functions[0] for func in tool_functions):
+                logger.debug("Last 3 tool calls are identical. Checking Tool responses..")
+                # Check if content attributes are the same for assistant items
+                tool_responses = [item for item in last_six_items if item.get("role") == "tool"]
+
+                if tool_responses:
+                    assistant_contents = [item.get("content") for item in tool_responses]
+                    logger.debug(f"Last N tool responses: {assistant_contents}")
+                    if all(content == assistant_contents[0] for content in assistant_contents):
+                        logger.debug("Last 3 tool responses are identical. Terminating")
+                        logger.info("Terminating browser executor since a loop was detected...")
+                        return True
+
+    return False