diff --git a/graph_swarm.py b/graph_swarm.py index 31842225..82cef523 100644 --- a/graph_swarm.py +++ b/graph_swarm.py @@ -202,7 +202,6 @@ class GraphSwarm: self.memory_collection = memory_collection self.memory = SwarmMemory(collection_name=memory_collection) - if agents: self.initialize_agents(agents) diff --git a/swarms/utils/any_to_str.py b/swarms/utils/any_to_str.py index 125e233e..2b0e3809 100644 --- a/swarms/utils/any_to_str.py +++ b/swarms/utils/any_to_str.py @@ -63,40 +63,40 @@ def any_to_str(data: Union[str, Dict, List, Tuple, Any]) -> str: return f"Error converting data: {str(e)}" -def main(): - # Example 1: Dictionary - print("Dictionary:") - print( - any_to_str( - { - "name": "John", - "age": 30, - "hobbies": ["reading", "hiking"], - } - ) - ) - - print("\nNested Dictionary:") - print( - any_to_str( - { - "user": { - "id": 123, - "details": {"city": "New York", "active": True}, - }, - "data": [1, 2, 3], - } - ) - ) - - print("\nList and Tuple:") - print(any_to_str([1, "text", None, (1, 2)])) - print(any_to_str((True, False, None))) - - print("\nEmpty Collections:") - print(any_to_str([])) - print(any_to_str({})) - - -if __name__ == "__main__": - main() +# def main(): +# # Example 1: Dictionary +# print("Dictionary:") +# print( +# any_to_str( +# { +# "name": "John", +# "age": 30, +# "hobbies": ["reading", "hiking"], +# } +# ) +# ) + +# print("\nNested Dictionary:") +# print( +# any_to_str( +# { +# "user": { +# "id": 123, +# "details": {"city": "New York", "active": True}, +# }, +# "data": [1, 2, 3], +# } +# ) +# ) + +# print("\nList and Tuple:") +# print(any_to_str([1, "text", None, (1, 2)])) +# print(any_to_str((True, False, None))) + +# print("\nEmpty Collections:") +# print(any_to_str([])) +# print(any_to_str({})) + + +# if __name__ == "__main__": +# main() diff --git a/swarms/utils/calculate_func_metrics.py b/swarms/utils/calculate_func_metrics.py index bfb8a528..795e7bb2 100644 --- a/swarms/utils/calculate_func_metrics.py +++ b/swarms/utils/calculate_func_metrics.py @@ -4,7 +4,6 @@ from functools import wraps from typing import Any, Callable import psutil -from loguru import logger from pydantic import BaseModel from swarms.utils.loguru_logger import initialize_logger diff --git a/swarms/utils/callable_name.py b/swarms/utils/callable_name.py deleted file mode 100644 index 9a0b037f..00000000 --- a/swarms/utils/callable_name.py +++ /dev/null @@ -1,203 +0,0 @@ -from typing import Any -import inspect -from functools import partial -import logging - - -class NameResolver: - """Utility class for resolving names of various objects""" - - @staticmethod - def get_name(obj: Any, default: str = "unnamed_callable") -> str: - """ - Get the name of any object with multiple fallback strategies. - - Args: - obj: The object to get the name from - default: Default name if all strategies fail - - Returns: - str: The resolved name - """ - strategies = [ - # Try getting __name__ attribute - lambda x: getattr(x, "__name__", None), - # Try getting class name - lambda x: ( - x.__class__.__name__ - if hasattr(x, "__class__") - else None - ), - # Try getting function name if it's a partial - lambda x: ( - x.func.__name__ if isinstance(x, partial) else None - ), - # Try getting the name from the class's type - lambda x: type(x).__name__, - # Try getting qualname - lambda x: getattr(x, "__qualname__", None), - # Try getting the module and class name - lambda x: ( - f"{x.__module__}.{x.__class__.__name__}" - if hasattr(x, "__module__") - else None - ), - # For async functions - lambda x: ( - x.__name__ if inspect.iscoroutinefunction(x) else None - ), - # For classes with custom __str__ - lambda x: ( - str(x) - if hasattr(x, "__str__") - and x.__str__ != object.__str__ - else None - ), - # For wrapped functions - lambda x: ( - getattr(x, "__wrapped__", None).__name__ - if hasattr(x, "__wrapped__") - else None - ), - ] - - # Try each strategy - for strategy in strategies: - try: - name = strategy(obj) - if name and isinstance(name, str): - return name.replace(" ", "_").replace("-", "_") - except Exception: - continue - - # Return default if all strategies fail - return default - - @staticmethod - def get_callable_details(obj: Any) -> dict: - """ - Get detailed information about a callable object. - - Returns: - dict: Dictionary containing: - - name: The resolved name - - type: The type of callable - - signature: The signature if available - - module: The module name if available - - doc: The docstring if available - """ - details = { - "name": NameResolver.get_name(obj), - "type": "unknown", - "signature": None, - "module": getattr(obj, "__module__", "unknown"), - "doc": inspect.getdoc(obj) - or "No documentation available", - } - - # Determine the type - if inspect.isclass(obj): - details["type"] = "class" - elif inspect.iscoroutinefunction(obj): - details["type"] = "async_function" - elif inspect.isfunction(obj): - details["type"] = "function" - elif isinstance(obj, partial): - details["type"] = "partial" - elif callable(obj): - details["type"] = "callable" - - # Try to get signature - try: - details["signature"] = str(inspect.signature(obj)) - except (ValueError, TypeError): - details["signature"] = "Unknown signature" - - return details - - @classmethod - def get_safe_name(cls, obj: Any, max_retries: int = 3) -> str: - """ - Safely get a name with retries and validation. - - Args: - obj: Object to get name from - max_retries: Maximum number of retry attempts - - Returns: - str: A valid name string - """ - retries = 0 - last_error = None - - while retries < max_retries: - try: - name = cls.get_name(obj) - - # Validate and clean the name - if name: - # Remove invalid characters - clean_name = "".join( - c - for c in name - if c.isalnum() or c in ["_", "."] - ) - - # Ensure it starts with a letter or underscore - if ( - not clean_name[0].isalpha() - and clean_name[0] != "_" - ): - clean_name = f"_{clean_name}" - - return clean_name - - except Exception as e: - last_error = e - retries += 1 - - # If all retries failed, generate a unique fallback name - import uuid - - fallback = f"callable_{uuid.uuid4().hex[:8]}" - logging.warning( - f"Failed to get name after {max_retries} retries. Using fallback: {fallback}. " - f"Last error: {str(last_error)}" - ) - return fallback - - -# # Example usage -# if __name__ == "__main__": -# def test_resolver(): -# # Test cases -# class TestClass: -# def method(self): -# pass - -# async def async_func(): -# pass - -# test_cases = [ -# TestClass, # Class -# TestClass(), # Instance -# async_func, # Async function -# lambda x: x, # Lambda -# partial(print, end=""), # Partial -# TestClass.method, # Method -# print, # Built-in function -# str, # Built-in class -# ] - -# resolver = NameResolver() - -# print("\nName Resolution Results:") -# print("-" * 50) -# for obj in test_cases: -# details = resolver.get_callable_details(obj) -# safe_name = resolver.get_safe_name(obj) -# print(f"\nObject: {obj}") -# print(f"Safe Name: {safe_name}") -# print(f"Details: {details}") - -# test_resolver() diff --git a/swarms/utils/dict_to_table.py b/swarms/utils/dict_to_table.py deleted file mode 100644 index e69de29b..00000000 diff --git a/swarms/utils/pdf_to_text.py b/swarms/utils/pdf_to_text.py index 90711691..8df8e065 100644 --- a/swarms/utils/pdf_to_text.py +++ b/swarms/utils/pdf_to_text.py @@ -1,14 +1,12 @@ -import sys from swarms.utils.try_except_wrapper import try_except_wrapper try: import pypdf except ImportError: - print( - "pypdf not installed. Please install it using: pip install" - " pypdf" - ) - sys.exit(1) + import subprocess + + subprocess.check_call(["python", "-m", "pip", "install", "pypdf"]) + import pypdf @try_except_wrapper diff --git a/swarms/utils/remove_json_whitespace.py b/swarms/utils/remove_json_whitespace.py deleted file mode 100644 index 0a043e7c..00000000 --- a/swarms/utils/remove_json_whitespace.py +++ /dev/null @@ -1,51 +0,0 @@ -import json - -import yaml - - -def remove_whitespace_from_json(json_string: str) -> str: - """ - Removes unnecessary whitespace from a JSON string. - - This function parses the JSON string into a Python object and then - serializes it back into a JSON string without unnecessary whitespace. - - Args: - json_string (str): The JSON string. - - Returns: - str: The JSON string with whitespace removed. - """ - parsed = json.loads(json_string) - return json.dumps(parsed, separators=(",", ":")) - - -# # Example usage for JSON -# json_string = '{"field1": 123, "field2": "example text"}' -# print(remove_whitespace_from_json(json_string)) - - -def remove_whitespace_from_yaml(yaml_string: str) -> str: - """ - Removes unnecessary whitespace from a YAML string. - - This function parses the YAML string into a Python object and then - serializes it back into a YAML string with minimized whitespace. - Note: This might change the representation style of YAML data. - - Args: - yaml_string (str): The YAML string. - - Returns: - str: The YAML string with whitespace reduced. - """ - parsed = yaml.safe_load(yaml_string) - return yaml.dump(parsed, default_flow_style=True) - - -# # Example usage for YAML -# yaml_string = """ -# field1: 123 -# field2: example text -# """ -# print(remove_whitespace_from_yaml(yaml_string))