refine utils code

metagpt/utils/common.py

@@ -301,9 +301,6 @@ class NoMoneyException(Exception):
 def print_members(module, indent=0):
     """
     https://stackoverflow.com/questions/1796180/how-can-i-get-a-list-of-all-classes-within-current-module-in-python
-    :param module:
-    :param indent:
-    :return:
     """
     prefix = " " * indent
     for name, obj in inspect.getmembers(module):
@@ -321,6 +318,7 @@ def print_members(module, indent=0):
 
 
 def parse_recipient(text):
+    # FIXME: use ActionNode instead.
     pattern = r"## Send To:\s*([A-Za-z]+)\s*?"  # hard code for now
     recipient = re.search(pattern, text)
     if recipient:
@@ -337,18 +335,12 @@ def get_class_name(cls) -> str:
     return f"{cls.__module__}.{cls.__name__}"
 
 
-def get_object_name(obj) -> str:
-    """Return class name of the object"""
-    cls = type(obj)
-    return f"{cls.__module__}.{cls.__name__}"
-
-
-def any_to_str(val) -> str:
+def any_to_str(val: str | typing.Callable) -> str:
     """Return the class name or the class name of the object, or 'val' if it's a string type."""
     if isinstance(val, str):
         return val
     if not callable(val):
-        return get_object_name(val)
+        return get_class_name(type(val))
 
     return get_class_name(val)
 
@@ -356,32 +348,57 @@ def any_to_str(val) -> str:
 def any_to_str_set(val) -> set:
     """Convert any type to string set."""
     res = set()
-    if isinstance(val, dict) or isinstance(val, list) or isinstance(val, set) or isinstance(val, tuple):
+
+    # Check if the value is iterable, but not a string (since strings are technically iterable)
+    if isinstance(val, (dict, list, set, tuple)):
+        # Special handling for dictionaries to iterate over values
+        if isinstance(val, dict):
+            val = val.values()
+
         for i in val:
             res.add(any_to_str(i))
     else:
         res.add(any_to_str(val))
+
     return res
 
 
-def is_subscribed(message, tags):
+def is_subscribed(message: "Message", tags: set):
     """Return whether it's consumer"""
     if MESSAGE_ROUTE_TO_ALL in message.send_to:
         return True
 
-    for t in tags:
-        if t in message.send_to:
+    for i in tags:
+        if i in message.send_to:
             return True
     return False
 
 
-def general_after_log(logger: "loguru.Logger", sec_format: str = "%0.3f") -> typing.Callable[["RetryCallState"], None]:
+def general_after_log(i: "loguru.Logger", sec_format: str = "%0.3f") -> typing.Callable[["RetryCallState"], None]:
+    """
+    Generates a logging function to be used after a call is retried.
+
+    This generated function logs an error message with the outcome of the retried function call. It includes
+    the name of the function, the time taken for the call in seconds (formatted according to `sec_format`),
+    the number of attempts made, and the exception raised, if any.
+
+    :param i: A Logger instance from the loguru library used to log the error message.
+    :param sec_format: A string format specifier for how to format the number of seconds since the start of the call.
+                       Defaults to three decimal places.
+    :return: A callable that accepts a RetryCallState object and returns None. This callable logs the details
+             of the retried call.
+    """
+
     def log_it(retry_state: "RetryCallState") -> None:
+        # If the function name is not known, default to "<unknown>"
         if retry_state.fn is None:
             fn_name = "<unknown>"
         else:
+            # Retrieve the callable's name using a utility function
             fn_name = _utils.get_callback_name(retry_state.fn)
-        logger.error(
+
+        # Log an error message with the function name, time since start, attempt number, and the exception
+        i.error(
             f"Finished call to '{fn_name}' after {sec_format % retry_state.seconds_since_start}(s), "
             f"this was the {_utils.to_ordinal(retry_state.attempt_number)} time calling it. "
             f"exp: {retry_state.outcome.exception()}"