pull/652/head
Your Name 1 month ago
parent 7a66cbd705
commit f82a57d70e

@ -202,7 +202,6 @@ class GraphSwarm:
self.memory_collection = memory_collection
self.memory = SwarmMemory(collection_name=memory_collection)
if agents:
self.initialize_agents(agents)

@ -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()

@ -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

@ -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()

@ -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

@ -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))
Loading…
Cancel
Save