Adding Py best pratices flakes and pyupgrade

csvagent_example.py

@@ -1,7 +1,10 @@
 # Example usage
 from pathlib import Path
-from swarms.structs.csv_to_agent import AgentLoader, AgentValidationError
 
+from swarms.structs.csv_to_agent import (
+    AgentLoader,
+    AgentValidationError,
+)
 
 if __name__ == "__main__":
     # Example agent configurations

example.py

@@ -1,11 +1,12 @@
 import os
 
+from dotenv import load_dotenv
 from swarm_models import OpenAIChat
+
 from swarms import Agent
 from swarms.prompts.finance_agent_sys_prompt import (
     FINANCIAL_AGENT_SYS_PROMPT,
 )
-from dotenv import load_dotenv
 
 load_dotenv()
 

graph_swarm_example.py

@@ -1,8 +1,8 @@
 from loguru import logger
+
 from swarms.structs.agent import Agent
 from swarms.structs.graph_swarm import GraphSwarm
 
-
 if __name__ == "__main__":
     try:
         # Create agents

multi_agent_router_example.py

@@ -39,4 +39,4 @@ if __name__ == "__main__":
         print(result_execute)
 
     except Exception as e:
-        print(f"Error occurred: {str(e)}")
+        print(f"Error occurred: {e!s}")

new_features_examples/agent_showcase_example.py

@@ -1,8 +1,8 @@
 import os
 
-from swarms import Agent
-
 from swarm_models import OpenAIChat
+
+from swarms import Agent
 from swarms.structs.agents_available import showcase_available_agents
 
 # Get the OpenAI API key from the environment variable

new_features_examples/agent_with_fluidapi.py

@@ -1,10 +1,10 @@
 import os
 
-from swarm_models import OpenAIChat
-from swarms import Agent
-from fluid_api_agent.main import fluid_api_request
 from dotenv import load_dotenv
+from fluid_api_agent.main import fluid_api_request
+from swarm_models import OpenAIChat
 
+from swarms import Agent
 
 load_dotenv()
 

new_features_examples/async_agent.py

@@ -1,8 +1,9 @@
+from swarm_models import OpenAIChat
+
 from swarms import Agent
 from swarms.prompts.finance_agent_sys_prompt import (
     FINANCIAL_AGENT_SYS_PROMPT,
 )
-from swarm_models import OpenAIChat
 
 model = OpenAIChat(model_name="gpt-4o")
 

new_features_examples/async_agents.py

@@ -3,11 +3,11 @@ import os
 from dotenv import load_dotenv
 from swarm_models import OpenAIChat
 
+from new_features_examples.async_executor import HighSpeedExecutor
 from swarms import Agent
 from swarms.prompts.finance_agent_sys_prompt import (
     FINANCIAL_AGENT_SYS_PROMPT,
 )
-from new_features_examples.async_executor import HighSpeedExecutor
 
 load_dotenv()
 

new_features_examples/async_executor.py

@@ -2,11 +2,11 @@ import asyncio
 import multiprocessing as mp
 import time
 from functools import partial
-from typing import Any, Dict, Union
+from typing import Any, Optional, Union
 
 
 class HighSpeedExecutor:
-    def __init__(self, num_processes: int = None):
+    def __init__(self, num_processes: Optional[int] = None):
         """
         Initialize the executor with configurable number of processes.
         If num_processes is None, it uses CPU count.
@@ -45,7 +45,7 @@ class HighSpeedExecutor:
         num_executions: int,
         *args: Any,
         **kwargs: Any,
-    ) -> Dict[str, Union[int, float]]:
+    ) -> dict[str, Union[int, float]]:
         """
         Execute the given function multiple times concurrently.
 

new_features_examples/async_workflow_example.py

@@ -1,21 +1,20 @@
 import asyncio
-from typing import List
 
 from swarm_models import OpenAIChat
 
+from swarms.prompts.finance_agent_sys_prompt import (
+    FINANCIAL_AGENT_SYS_PROMPT,
+)
+from swarms.structs.agent import Agent
 from swarms.structs.async_workflow import (
     SpeakerConfig,
     SpeakerRole,
     create_default_workflow,
     run_workflow_with_retry,
 )
-from swarms.prompts.finance_agent_sys_prompt import (
-    FINANCIAL_AGENT_SYS_PROMPT,
-)
-from swarms.structs.agent import Agent
 
 
-async def create_specialized_agents() -> List[Agent]:
+async def create_specialized_agents() -> list[Agent]:
     """Create a set of specialized agents for financial analysis"""
 
     # Base model configuration
@@ -165,7 +164,7 @@ async def main():
                     print(value)
 
     except Exception as e:
-        print(f"Workflow failed: {str(e)}")
+        print(f"Workflow failed: {e!s}")
 
     finally:
         await workflow.cleanup()

new_features_examples/auto_agent.py

@@ -1,7 +1,7 @@
 import json
 import os
 from contextlib import suppress
-from typing import Any, Callable, Dict, Optional, Type, Union
+from typing import Any, Callable, Optional, Union
 
 from dotenv import load_dotenv
 from pydantic import BaseModel, Field, ValidationError, create_model
@@ -10,7 +10,7 @@ from swarm_models.openai_function_caller import OpenAIFunctionCaller
 
 class DynamicParser:
     @staticmethod
-    def extract_fields(model: Type[BaseModel]) -> Dict[str, Any]:
+    def extract_fields(model: type[BaseModel]) -> dict[str, Any]:
         return {
             field_name: (
                 field.annotation,
@@ -21,8 +21,8 @@ class DynamicParser:
 
     @staticmethod
     def create_partial_model(
-        model: Type[BaseModel], data: Dict[str, Any]
-    ) -> Type[BaseModel]:
+        model: type[BaseModel], data: dict[str, Any]
+    ) -> type[BaseModel]:
         fields = {
             field_name: (
                 field.annotation,
@@ -35,7 +35,7 @@ class DynamicParser:
 
     @classmethod
     def parse(
-        cls, data: Union[str, Dict[str, Any]], model: Type[BaseModel]
+        cls, data: Union[str, dict[str, Any]], model: type[BaseModel]
     ) -> Optional[BaseModel]:
         if isinstance(data, str):
             try:
@@ -88,7 +88,7 @@ class Command(BaseModel):
         ...,
         description="Command name to execute from the provided list of commands.",
     )
-    args: Dict[str, Any] = Field(
+    args: dict[str, Any] = Field(
         ..., description="Arguments required to execute the command."
     )
 
@@ -134,9 +134,9 @@ def task_complete_command(reason: str):
 
 
 # Dynamic command execution
-def execute_command(name: str, args: Dict[str, Any]):
+def execute_command(name: str, args: dict[str, Any]):
     """Dynamically execute a command based on its name and arguments."""
-    command_map: Dict[str, Callable] = {
+    command_map: dict[str, Callable] = {
         "fluid_api": lambda **kwargs: fluid_api_command(
             task=kwargs.get("task")
         ),
@@ -157,8 +157,8 @@ def execute_command(name: str, args: Dict[str, Any]):
 
 
 def parse_and_execute_command(
-    response: Union[str, Dict[str, Any]],
-    base_model: Type[BaseModel] = AgentResponse,
+    response: Union[str, dict[str, Any]],
+    base_model: type[BaseModel] = AgentResponse,
 ) -> Any:
     """Enhanced command parser with flexible input handling"""
     parsed = DynamicParser.parse(response, base_model)

new_features_examples/auto_swarm_router.py

@@ -1,7 +1,9 @@
 import os
+
 from dotenv import load_dotenv
-from swarms import Agent
 from swarm_models import OpenAIChat
+
+from swarms import Agent
 from swarms.structs.swarm_router import SwarmRouter
 
 load_dotenv()

new_features_examples/crypto/swarms_coin_agent.py

@@ -1,4 +1,5 @@
 import requests
+
 from swarms import Agent
 
 # Define the system prompt specialized for $Swarms

new_features_examples/crypto/swarms_coin_multimarket.py

@@ -1,8 +1,9 @@
 import asyncio
-import aiohttp
-from typing import Dict, List, Optional
 from datetime import datetime
 from statistics import mean, median
+from typing import Optional
+
+import aiohttp
 
 from swarms.structs.agent import Agent
 
@@ -121,7 +122,7 @@ class MultiExchangeDataFetcher:
             "birdeye": "https://public-api.birdeye.so/public",  # Using Birdeye instead of Jupiter
         }
 
-    async def fetch_data(self, url: str) -> Optional[Dict]:
+    async def fetch_data(self, url: str) -> Optional[dict]:
         """Generic async function to fetch data from APIs with error handling"""
         async with aiohttp.ClientSession() as session:
             try:
@@ -136,10 +137,10 @@ class MultiExchangeDataFetcher:
                 print(f"Timeout while fetching from {url}")
                 return None
             except Exception as e:
-                print(f"Error fetching from {url}: {str(e)}")
+                print(f"Error fetching from {url}: {e!s}")
                 return None
 
-    async def get_coingecko_data(self) -> Optional[Dict]:
+    async def get_coingecko_data(self) -> Optional[dict]:
         """Fetch $Swarms data from CoinGecko"""
         try:
             url = f"{self.base_urls['coingecko']}/simple/price"
@@ -160,10 +161,10 @@ class MultiExchangeDataFetcher:
                 }
             return None
         except Exception as e:
-            print(f"Error processing CoinGecko data: {str(e)}")
+            print(f"Error processing CoinGecko data: {e!s}")
             return None
 
-    async def get_dexscreener_data(self) -> Optional[Dict]:
+    async def get_dexscreener_data(self) -> Optional[dict]:
         """Fetch $Swarms data from DexScreener"""
         try:
             url = (
@@ -179,10 +180,10 @@ class MultiExchangeDataFetcher:
                 }
             return None
         except Exception as e:
-            print(f"Error processing DexScreener data: {str(e)}")
+            print(f"Error processing DexScreener data: {e!s}")
             return None
 
-    async def get_birdeye_data(self) -> Optional[Dict]:
+    async def get_birdeye_data(self) -> Optional[dict]:
         """Fetch $Swarms data from Birdeye"""
         try:
             # Example Birdeye endpoint - replace ADDRESS with actual Swarms token address
@@ -201,12 +202,12 @@ class MultiExchangeDataFetcher:
                 }
             return None
         except Exception as e:
-            print(f"Error processing Birdeye data: {str(e)}")
+            print(f"Error processing Birdeye data: {e!s}")
             return None
 
     def aggregate_data(
-        self, data_points: List[Optional[Dict]]
-    ) -> Dict:
+        self, data_points: list[Optional[dict]]
+    ) -> dict:
         """Aggregate data from multiple sources with null checking"""
         prices = []
         volumes = []
@@ -297,9 +298,9 @@ async def answer_swarms_query(query: str) -> str:
         )
         return swarms_agent.run(full_query)
     except Exception as e:
-        print(f"Error in answer_swarms_query: {str(e)}")
+        print(f"Error in answer_swarms_query: {e!s}")
         return (
-            f"An error occurred while processing your query: {str(e)}"
+            f"An error occurred while processing your query: {e!s}"
         )
 
 

new_features_examples/dict_to_table.py

@@ -1,5 +1,6 @@
-import pandas as pd
 import json
+
+import pandas as pd
 from loguru import logger
 
 

new_features_examples/ethchain_agent.py

@@ -1,15 +1,17 @@
-import os
-from swarms import Agent
-from swarm_models import OpenAIChat
-from web3 import Web3
-from typing import Dict, Optional, Any
-from datetime import datetime
 import asyncio
-from loguru import logger
-from dotenv import load_dotenv
 import csv
-import requests
+import os
 import time
+from datetime import datetime
+from typing import Any, Optional
+
+import requests
+from dotenv import load_dotenv
+from loguru import logger
+from swarm_models import OpenAIChat
+from web3 import Web3
+
+from swarms import Agent
 
 BLOCKCHAIN_AGENT_PROMPT = """
 You are an expert blockchain and cryptocurrency analyst with deep knowledge of Ethereum markets and DeFi ecosystems.
@@ -110,7 +112,7 @@ class EthereumAnalyzer:
             )
             return float(response.json()["ethereum"]["usd"])
         except Exception as e:
-            logger.error(f"Error fetching ETH price: {str(e)}")
+            logger.error(f"Error fetching ETH price: {e!s}")
             return 0.0
 
     def update_eth_price(self):
@@ -143,7 +145,7 @@ class EthereumAnalyzer:
 
     async def analyze_transaction(
         self, tx_hash: str
-    ) -> Optional[Dict[str, Any]]:
+    ) -> Optional[dict[str, Any]]:
         """Analyze a single transaction."""
         try:
             tx = self.w3.eth.get_transaction(tx_hash)
@@ -191,11 +193,11 @@ class EthereumAnalyzer:
 
         except Exception as e:
             logger.error(
-                f"Error analyzing transaction {tx_hash}: {str(e)}"
+                f"Error analyzing transaction {tx_hash}: {e!s}"
             )
             return None
 
-    def prepare_analysis_prompt(self, tx_data: Dict[str, Any]) -> str:
+    def prepare_analysis_prompt(self, tx_data: dict[str, Any]) -> str:
         """Prepare detailed analysis prompt including price context."""
         value_usd = tx_data["value_usd"]
         eth_price = tx_data["eth_price"]
@@ -220,7 +222,7 @@ Analyze market impact, patterns, risks, and strategic implications."""
 
         return prompt
 
-    def save_to_csv(self, tx_data: Dict[str, Any], ai_analysis: str):
+    def save_to_csv(self, tx_data: dict[str, Any], ai_analysis: str):
         """Save transaction data and analysis to CSV."""
         row = [
             tx_data["timestamp"],
@@ -288,7 +290,7 @@ Analyze market impact, patterns, risks, and strategic implications."""
                 await asyncio.sleep(1)  # Wait for next block
 
             except Exception as e:
-                logger.error(f"Error in monitoring loop: {str(e)}")
+                logger.error(f"Error in monitoring loop: {e!s}")
                 await asyncio.sleep(1)
 
 

new_features_examples/example_async_vs_multithread.py

@@ -1,14 +1,15 @@
-import os
 import asyncio
-from swarms import Agent
-from swarm_models import OpenAIChat
+import os
 import time
+
 import psutil
+from dotenv import load_dotenv
+from swarm_models import OpenAIChat
 
+from swarms import Agent
 from swarms.prompts.finance_agent_sys_prompt import (
     FINANCIAL_AGENT_SYS_PROMPT,
 )
-from dotenv import load_dotenv
 
 load_dotenv()
 

new_features_examples/forest_swarm_examples/tree_swarm_test.py

@@ -1,6 +1,5 @@
 from swarms.structs.tree_swarm import ForestSwarm, Tree, TreeAgent
 
-
 agents_tree1 = [
     TreeAgent(
         system_prompt="Stock Analysis Agent",

new_features_examples/full_agent_rag_example.py

@@ -84,7 +84,7 @@ class LlamaIndexDB:
                 f"Successfully indexed documents from {self.data_dir}"
             )
         except Exception as e:
-            logger.error(f"Error indexing documents: {str(e)}")
+            logger.error(f"Error indexing documents: {e!s}")
             raise
 
     def query(self, query: str, **kwargs) -> str:
@@ -115,7 +115,7 @@ class LlamaIndexDB:
             logger.info(f"Successfully queried: {query}")
             return str(response)
         except Exception as e:
-            logger.error(f"Error during query: {str(e)}")
+            logger.error(f"Error during query: {e!s}")
             raise
 
 

new_features_examples/gemini_model.py

@@ -1,4 +1,5 @@
 import os
+
 import google.generativeai as genai
 from loguru import logger
 

new_features_examples/groupchat_examples/crypto_tax.py

@@ -1,6 +1,8 @@
 import os
+
 from dotenv import load_dotenv
 from swarm_models import OpenAIChat
+
 from swarms import Agent, GroupChat, expertise_based
 
 if __name__ == "__main__":

new_features_examples/groupchat_examples/crypto_tax_swarm 2.py

@@ -1,6 +1,8 @@
 import os
+
 from dotenv import load_dotenv
 from swarm_models import OpenAIChat
+
 from swarms import Agent, GroupChat
 
 if __name__ == "__main__":

new_features_examples/groupchat_examples/crypto_tax_swarm.py

@@ -1,6 +1,8 @@
 import os
+
 from dotenv import load_dotenv
 from swarm_models import OpenAIChat
+
 from swarms import Agent, GroupChat
 
 if __name__ == "__main__":

new_features_examples/groupchat_examples/group_chat_example.py

@@ -1,8 +1,9 @@
 import os
+
 from dotenv import load_dotenv
 from swarm_models import OpenAIChat
-from swarms import Agent, GroupChat
 
+from swarms import Agent, GroupChat
 
 if __name__ == "__main__":
 

new_features_examples/insurance_agent.py

@@ -1,6 +1,5 @@
 from swarms import Agent
 
-
 # Claims Processing Agent system prompt
 CLAIMS_PROCESSING_AGENT_SYS_PROMPT = """
 Here's an extended and detailed system prompt for the **Claims Processing Agent**, incorporating reasoning steps, output format, and examples for structured responses:

new_features_examples/insurance_swarm.py

@@ -1,7 +1,7 @@
 import asyncio
 from dataclasses import dataclass
 from enum import Enum
-from typing import List, Optional
+from typing import Optional
 
 from swarms import Agent
 
@@ -22,11 +22,11 @@ class InsuranceProduct:
     name: str
     type: InsuranceType
     description: str
-    coverage: List[str]
+    coverage: list[str]
     price_range: str
     min_coverage: float
     max_coverage: float
-    payment_options: List[str]
+    payment_options: list[str]
     waiting_period: str
     available: bool
 
@@ -290,7 +290,7 @@ Explique brevemente cada uno y solicite información sobre sus necesidades espec
         Estado: {'Disponible' if product.available else 'No disponible'}
         """
 
-    def handle_main_menu(self) -> List[str]:
+    def handle_main_menu(self) -> list[str]:
         """Return main menu options"""
         return [
             "1. Consultar productos de seguro",

new_features_examples/main.py

@@ -1,12 +1,13 @@
-from typing import List, Dict
+import asyncio
+import json
 from dataclasses import dataclass
 from datetime import datetime
-import asyncio
+from pathlib import Path
+
 import aiohttp
 from loguru import logger
+
 from swarms import Agent
-from pathlib import Path
-import json
 
 
 @dataclass
@@ -39,7 +40,7 @@ class DataFetcher:
 
     async def get_market_data(
         self, limit: int = 20
-    ) -> List[CryptoData]:
+    ) -> list[CryptoData]:
         """Fetch market data for top cryptocurrencies"""
         await self._init_session()
 
@@ -91,7 +92,7 @@ class DataFetcher:
                         )
                     except (ValueError, TypeError) as e:
                         logger.error(
-                            f"Error processing coin data: {str(e)}"
+                            f"Error processing coin data: {e!s}"
                         )
                         continue
 
@@ -101,7 +102,7 @@ class DataFetcher:
                 return crypto_data
 
         except Exception as e:
-            logger.error(f"Exception in get_market_data: {str(e)}")
+            logger.error(f"Exception in get_market_data: {e!s}")
             return []
 
 
@@ -111,7 +112,7 @@ class CryptoSwarmSystem:
         self.data_fetcher = DataFetcher()
         logger.info("Crypto Swarm System initialized")
 
-    def _initialize_agents(self) -> Dict[str, Agent]:
+    def _initialize_agents(self) -> dict[str, Agent]:
         """Initialize different specialized agents"""
         base_config = {
             "max_loops": 1,
@@ -160,7 +161,7 @@ class CryptoSwarmSystem:
         }
         return agents
 
-    async def analyze_market(self) -> Dict:
+    async def analyze_market(self) -> dict:
         """Run real-time market analysis using all agents"""
         try:
             # Fetch market data
@@ -198,14 +199,14 @@ class CryptoSwarmSystem:
             }
 
         except Exception as e:
-            logger.error(f"Error in market analysis: {str(e)}")
+            logger.error(f"Error in market analysis: {e!s}")
             return {
                 "error": str(e),
                 "timestamp": datetime.now().isoformat(),
             }
 
     def _run_agent_analysis(
-        self, agent: Agent, crypto_data: List[CryptoData]
+        self, agent: Agent, crypto_data: list[CryptoData]
     ) -> str:
         """Run analysis for a single agent"""
         try:
@@ -230,8 +231,8 @@ class CryptoSwarmSystem:
             return agent.run(prompt)
 
         except Exception as e:
-            logger.error(f"Error in {agent.agent_name}: {str(e)}")
-            return f"Error: {str(e)}"
+            logger.error(f"Error in {agent.agent_name}: {e!s}")
+            return f"Error: {e!s}"
 
 
 async def main():
@@ -261,7 +262,7 @@ async def main():
             await asyncio.sleep(300)  # 5 minutes
 
         except Exception as e:
-            logger.error(f"Error in main loop: {str(e)}")
+            logger.error(f"Error in main loop: {e!s}")
             await asyncio.sleep(60)  # Wait 1 minute before retrying
         finally:
             if swarm.data_fetcher.session:

new_features_examples/medical_analysis/health_privacy_swarm 2.py

@@ -1,7 +1,9 @@
 import os
-from swarms import Agent, AgentRearrange
+
 from swarm_models import OpenAIChat
 
+from swarms import Agent, AgentRearrange
+
 # Get the OpenAI API key from the environment variable
 api_key = os.getenv("OPENAI_API_KEY")
 

new_features_examples/medical_analysis/health_privacy_swarm.py

@@ -1,7 +1,9 @@
 import os
-from swarms import Agent, AgentRearrange
+
 from swarm_models import OpenAIChat
 
+from swarms import Agent, AgentRearrange
+
 # Get the OpenAI API key from the environment variable
 api_key = os.getenv("OPENAI_API_KEY")
 

new_features_examples/medical_analysis/health_privacy_swarm_two 2.py

@@ -1,7 +1,9 @@
 import os
-from swarms import Agent, AgentRearrange
+
 from swarm_models import OpenAIChat
 
+from swarms import Agent, AgentRearrange
+
 # Initialize OpenAI model
 api_key = os.getenv(
     "OPENAI_API_KEY"

new_features_examples/medical_analysis/health_privacy_swarm_two.py

@@ -1,7 +1,9 @@
 import os
-from swarms import Agent, AgentRearrange
+
 from swarm_models import OpenAIChat
 
+from swarms import Agent, AgentRearrange
+
 # Initialize OpenAI model
 api_key = os.getenv(
     "OPENAI_API_KEY"

new_features_examples/medical_analysis/rearrange_video_examples/term_sheet_swarm.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+
 from swarms import Agent, AgentRearrange, create_file_in_folder
 
 # Lead Investment Analyst

new_features_examples/microstructure.py

@@ -5,7 +5,7 @@ from collections import deque
 from dataclasses import dataclass
 from datetime import datetime
 from queue import Queue
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Optional
 
 import ccxt
 import numpy as np
@@ -25,9 +25,9 @@ class MarketSignal:
     timestamp: datetime
     signal_type: str
     source: str
-    data: Dict[str, Any]
+    data: dict[str, Any]
     confidence: float
-    metadata: Dict[str, Any]
+    metadata: dict[str, Any]
 
 
 class MarketDataBuffer:
@@ -40,7 +40,7 @@ class MarketDataBuffer:
         with self.lock:
             self.data.append(item)
 
-    def get_latest(self, n: int = None) -> List[Any]:
+    def get_latest(self, n: Optional[int] = None) -> list[Any]:
         with self.lock:
             if n is None:
                 return list(self.data)
@@ -208,7 +208,7 @@ class ExchangeManager:
             raise RuntimeError("No exchanges available")
         return next(iter(self.active_exchanges.values()))
 
-    def get_all_active_exchanges(self) -> Dict[str, ccxt.Exchange]:
+    def get_all_active_exchanges(self) -> dict[str, ccxt.Exchange]:
         """Get all active exchanges"""
         return self.active_exchanges
 
@@ -269,8 +269,8 @@ class OrderBookAgent(BaseMarketAgent):
         self.vwap_window = 20
 
     def calculate_order_book_metrics(
-        self, order_book: Dict
-    ) -> Dict[str, float]:
+        self, order_book: dict
+    ) -> dict[str, float]:
         bids = np.array(order_book["bids"])
         asks = np.array(order_book["asks"])
 
@@ -321,7 +321,7 @@ class OrderBookAgent(BaseMarketAgent):
         }
 
     def detect_large_orders(
-        self, metrics: Dict[str, float], threshold: float = 2.0
+        self, metrics: dict[str, float], threshold: float = 2.0
     ) -> bool:
         historical_books = self.order_book_buffer.get_latest(20)
         if not historical_books:
@@ -402,7 +402,7 @@ class OrderBookAgent(BaseMarketAgent):
                 },
             )
         except Exception as e:
-            logger.error(f"Error in order book analysis: {str(e)}")
+            logger.error(f"Error in order book analysis: {e!s}")
             return None
 
 
@@ -418,42 +418,6 @@ class TickDataAgent(BaseMarketAgent):
         exchange_manager = ExchangeManager()
         self.exchange = exchange_manager.get_primary_exchange()
 
-    def calculate_tick_metrics(
-        self, ticks: List[Dict]
-    ) -> Dict[str, float]:
-        df = pd.DataFrame(ticks)
-        df["price"] = pd.to_numeric(df["price"])
-        df["volume"] = pd.to_numeric(df["amount"])
-
-        # Calculate key metrics
-        metrics = {}
-
-        # Volume-weighted average price (VWAP)
-        metrics["vwap"] = (df["price"] * df["volume"]).sum() / df[
-            "volume"
-        ].sum()
-
-        # Price momentum
-        metrics["price_momentum"] = df["price"].diff().mean()
-
-        # Volume profile
-        metrics["volume_mean"] = df["volume"].mean()
-        metrics["volume_std"] = df["volume"].std()
-
-        # Trade intensity
-        time_diff = (
-            df["timestamp"].max() - df["timestamp"].min()
-        ) / 1000  # Convert to seconds
-        metrics["trade_intensity"] = (
-            len(df) / time_diff if time_diff > 0 else 0
-        )
-
-        # Microstructure indicators
-        metrics["kyle_lambda"] = self.calculate_kyle_lambda(df)
-        metrics["roll_spread"] = self.calculate_roll_spread(df)
-
-        return metrics
-
     def calculate_kyle_lambda(self, df: pd.DataFrame) -> float:
         """Calculate Kyle's Lambda (price impact coefficient)"""
         try:
@@ -483,8 +447,8 @@ class TickDataAgent(BaseMarketAgent):
         return 0.0
 
     def calculate_tick_metrics(
-        self, ticks: List[Dict]
-    ) -> Dict[str, float]:
+        self, ticks: list[dict]
+    ) -> dict[str, float]:
         try:
             # Debug the incoming data structure
             logger.info(
@@ -557,7 +521,7 @@ class TickDataAgent(BaseMarketAgent):
 
         except Exception as e:
             logger.error(
-                f"Error in calculate_tick_metrics: {str(e)}",
+                f"Error in calculate_tick_metrics: {e!s}",
                 exc_info=True,
             )
             # Return default metrics on error
@@ -629,7 +593,7 @@ class TickDataAgent(BaseMarketAgent):
 
         except Exception as e:
             logger.error(
-                f"Error in tick analysis: {str(e)}", exc_info=True
+                f"Error in tick analysis: {e!s}", exc_info=True
             )
             return None
 
@@ -659,8 +623,8 @@ class LatencyArbitrageAgent(BaseMarketAgent):
         }
 
     def calculate_effective_prices(
-        self, ticker: Dict, venue: str
-    ) -> Tuple[float, float]:
+        self, ticker: dict, venue: str
+    ) -> tuple[float, float]:
         """Calculate effective prices including fees"""
         fee = self.fee_structure[venue]
         return (
@@ -669,8 +633,8 @@ class LatencyArbitrageAgent(BaseMarketAgent):
         )
 
     def calculate_arbitrage_metrics(
-        self, prices: Dict[str, Dict]
-    ) -> Dict:
+        self, prices: dict[str, dict]
+    ) -> dict:
         opportunities = []
 
         for venue1 in prices:
@@ -818,7 +782,7 @@ class LatencyArbitrageAgent(BaseMarketAgent):
             )
 
         except Exception as e:
-            logger.error(f"Error in arbitrage analysis: {str(e)}")
+            logger.error(f"Error in arbitrage analysis: {e!s}")
             return None
 
 
@@ -841,7 +805,7 @@ class SwarmCoordinator:
         with self.lock:
             self.signal_processors.append(processor)
 
-    def process_signals(self, signals: List[MarketSignal]):
+    def process_signals(self, signals: list[MarketSignal]):
         """Process signals through all registered processors"""
         if not signals:
             return
@@ -855,8 +819,8 @@ class SwarmCoordinator:
             logger.error(f"Error in signal processing: {e}")
 
     def aggregate_signals(
-        self, signals: List[MarketSignal]
-    ) -> Dict[str, Any]:
+        self, signals: list[MarketSignal]
+    ) -> dict[str, Any]:
         """Aggregate multiple signals into a combined market view"""
         if not signals:
             return {}
@@ -925,7 +889,7 @@ class SwarmCoordinator:
 
         return aggregated
 
-    def start(self, symbols: List[str], interval: float = 1.0):
+    def start(self, symbols: list[str], interval: float = 1.0):
         """Start the swarm monitoring system"""
         if self.running:
             logger.warning("Swarm is already running")
@@ -1010,7 +974,7 @@ class SwarmCoordinator:
         logger.info("Swarm stopped")
 
 
-def market_making_processor(signals: List[MarketSignal]):
+def market_making_processor(signals: list[MarketSignal]):
     """Enhanced signal processor with LLM analysis integration"""
     for signal in signals:
         if signal.confidence > 0.8:

new_features_examples/multi_tool_usage_agent.py

@@ -1,25 +1,32 @@
+import inspect
+import json
 import os
-from typing import List, Dict, Any, Optional, Callable, get_type_hints
+import typing
 from dataclasses import dataclass, field
-import json
 from datetime import datetime
-import inspect
-import typing
-from typing import Union
-from swarms import Agent
+from typing import (
+    Any,
+    Callable,
+    Optional,
+    Union,
+    get_type_hints,
+)
+
 from swarm_models import OpenAIChat
 
+from swarms import Agent
+
 
 @dataclass
 class ToolDefinition:
     name: str
     description: str
-    parameters: Dict[str, Any]
-    required_params: List[str]
+    parameters: dict[str, Any]
+    required_params: list[str]
     callable: Optional[Callable] = None
 
 
-def extract_type_hints(func: Callable) -> Dict[str, Any]:
+def extract_type_hints(func: Callable) -> dict[str, Any]:
     """Extract parameter types from function type hints."""
     return typing.get_type_hints(func)
 
@@ -81,7 +88,7 @@ class FunctionSpec:
 
     name: str
     description: str
-    parameters: Dict[
+    parameters: dict[
         str, dict
     ]  # Contains type and description for each parameter
     return_type: str
@@ -94,7 +101,7 @@ class ExecutionStep:
 
     step_id: int
     function_name: str
-    parameters: Dict[str, Any]
+    parameters: dict[str, Any]
     expected_output: str
     completed: bool = False
     result: Any = None
@@ -105,10 +112,10 @@ class ExecutionContext:
     """Maintains state during execution."""
 
     task: str
-    steps: List[ExecutionStep] = field(default_factory=list)
-    results: Dict[int, Any] = field(default_factory=dict)
+    steps: list[ExecutionStep] = field(default_factory=list)
+    results: dict[int, Any] = field(default_factory=dict)
     current_step: int = 0
-    history: List[Dict[str, Any]] = field(default_factory=list)
+    history: list[dict[str, Any]] = field(default_factory=list)
 
 
 def func():
@@ -121,7 +128,7 @@ hints = get_type_hints(func)
 class ToolAgent:
     def __init__(
         self,
-        functions: List[Callable],
+        functions: list[Callable],
         openai_api_key: str,
         model_name: str = "gpt-4",
         temperature: float = 0.1,
@@ -145,8 +152,8 @@ class ToolAgent:
         )
 
     def _analyze_functions(
-        self, functions: List[Callable]
-    ) -> Dict[str, FunctionSpec]:
+        self, functions: list[Callable]
+    ) -> dict[str, FunctionSpec]:
         """Analyze functions to create detailed specifications."""
         specs = {}
         for func in functions:
@@ -254,7 +261,7 @@ Always:
 """
 
     def _execute_function(
-        self, spec: FunctionSpec, parameters: Dict[str, Any]
+        self, spec: FunctionSpec, parameters: dict[str, Any]
     ) -> Any:
         """Execute a function with type checking."""
         converted_params = {}
@@ -269,12 +276,12 @@ Always:
                     converted_params[name] = value
             except (ValueError, TypeError) as e:
                 raise ValueError(
-                    f"Parameter '{name}' conversion failed: {str(e)}"
+                    f"Parameter '{name}' conversion failed: {e!s}"
                 )
 
         return self.functions[spec.name](**converted_params)
 
-    def run(self, task: str) -> Dict[str, Any]:
+    def run(self, task: str) -> dict[str, Any]:
         """Execute task with planning and step-by-step execution."""
         context = ExecutionContext(task=task)
         execution_log = {
@@ -363,7 +370,7 @@ Always:
                     context.current_step += 1
 
                 except Exception as e:
-                    print(f"Error in step {step.step_id}: {str(e)}")
+                    print(f"Error in step {step.step_id}: {e!s}")
                     execution_log["steps"].append(
                         {
                             "step_id": step.step_id,

new_features_examples/new_spreadsheet_swarm_examples/crypto_tax_swarm/crypto_tax_spreadsheet.py

@@ -31,4 +31,4 @@ try:
     print(results)
 
 except Exception as e:
-    print(f"An error occurred: {str(e)}")
+    print(f"An error occurred: {e!s}")

new_features_examples/new_spreadsheet_swarm_examples/financial_analysis/swarm_csv.py

@@ -31,4 +31,4 @@ try:
     print(results)
 
 except Exception as e:
-    print(f"An error occurred: {str(e)}")
+    print(f"An error occurred: {e!s}")

new_features_examples/ollama_demo.py

@@ -22,10 +22,10 @@
 
 from datetime import datetime
 
-from swarms import Agent, AgentRearrange, create_file_in_folder
-
 from swarm_models import OllamaModel
 
+from swarms import Agent, AgentRearrange, create_file_in_folder
+
 model = OllamaModel(model_name="llama3.2")
 
 chief_medical_officer = Agent(

new_features_examples/openai_assistant_wrapper.py

@@ -1,7 +1,7 @@
+from swarms.agents.openai_assistant import OpenAIAssistant
 from swarms.prompts.finance_agent_sys_prompt import (
     FINANCIAL_AGENT_SYS_PROMPT,
 )
-from swarms.agents.openai_assistant import OpenAIAssistant
 
 agent = OpenAIAssistant(
     name="test", instructions=FINANCIAL_AGENT_SYS_PROMPT

new_features_examples/persistent_legal_agent.py

@@ -1,7 +1,9 @@
 import os
-from swarms import Agent
-from swarm_models import OpenAIChat
+
 from dotenv import load_dotenv
+from swarm_models import OpenAIChat
+
+from swarms import Agent
 
 # Custom system prompt for VC legal document generation
 VC_LEGAL_AGENT_PROMPT = """You are a specialized legal document assistant focusing on venture capital documentation.

new_features_examples/privacy_building.py

@@ -1,7 +1,9 @@
 import os
-from swarms import Agent, AgentRearrange
+
 from swarm_models import OpenAIChat
 
+from swarms import Agent, AgentRearrange
+
 # Get the OpenAI API key from the environment variable
 api_key = os.getenv("OPENAI_API_KEY")
 

new_features_examples/qdrant_agent.py

@@ -1,7 +1,7 @@
 import os
 import uuid
 from datetime import datetime
-from typing import Dict, List, Optional
+from typing import Optional
 
 from qdrant_client import QdrantClient
 from qdrant_client.http import models
@@ -59,8 +59,8 @@ class QdrantMemory:
     def add(
         self,
         text: str,
-        embedding: List[float],
-        metadata: Optional[Dict] = None,
+        embedding: list[float],
+        metadata: Optional[dict] = None,
     ) -> str:
         """Add a memory to the store.
 
@@ -95,10 +95,10 @@ class QdrantMemory:
 
     def query(
         self,
-        query_embedding: List[float],
+        query_embedding: list[float],
         limit: int = 5,
         score_threshold: float = 0.7,
-    ) -> List[Dict]:
+    ) -> list[dict]:
         """Query memories based on vector similarity.
 
         Args:

new_features_examples/real_estate_agent.py

@@ -3,17 +3,19 @@ Zoe - Real Estate Agent
 
 """
 
-from typing import Optional, Dict, Any, List
+import json
+import os
 from dataclasses import dataclass
 from datetime import datetime
-import os
-import json
+from enum import Enum
+from typing import Any, Optional
+
 import requests
+from dotenv import load_dotenv
 from loguru import logger
-from swarms import Agent
 from swarm_models import OpenAIChat
-from dotenv import load_dotenv
-from enum import Enum
+
+from swarms import Agent
 
 # Configure loguru logger
 logger.add(
@@ -52,8 +54,8 @@ class PropertyListing:
     lat: float
     lng: float
     description: Optional[str] = None
-    features: Optional[List[str]] = None
-    images: Optional[List[str]] = None
+    features: Optional[list[str]] = None
+    images: Optional[list[str]] = None
 
 
 class PropertyRadarAPI:
@@ -78,13 +80,13 @@ class PropertyRadarAPI:
     def search_properties(
         self,
         max_price: float = 10_000_000,
-        property_types: List[PropertyType] = None,
-        location: Dict[str, Any] = None,
+        property_types: Optional[list[PropertyType]] = None,
+        location: Optional[dict[str, Any]] = None,
         min_sqft: Optional[float] = None,
         max_sqft: Optional[float] = None,
         page: int = 1,
         limit: int = 20,
-    ) -> List[PropertyListing]:
+    ) -> list[PropertyListing]:
         """
         Search for commercial properties using PropertyRadar API
 
@@ -163,7 +165,7 @@ class PropertyRadarAPI:
             ]
 
         except requests.RequestException as e:
-            logger.error(f"Error fetching properties: {str(e)}")
+            logger.error(f"Error fetching properties: {e!s}")
             raise
 
 
@@ -234,11 +236,11 @@ class CommercialRealEstateAgent:
     def search_properties(
         self,
         max_price: float = 10_000_000,
-        property_types: List[PropertyType] = None,
-        location: Dict[str, Any] = None,
+        property_types: Optional[list[PropertyType]] = None,
+        location: Optional[dict[str, Any]] = None,
         min_sqft: Optional[float] = None,
         max_sqft: Optional[float] = None,
-    ) -> List[Dict[str, Any]]:
+    ) -> list[dict[str, Any]]:
         """
         Search for properties and provide analysis
 
@@ -286,7 +288,7 @@ class CommercialRealEstateAgent:
 
         except Exception as e:
             logger.error(
-                f"Error in property search and analysis: {str(e)}"
+                f"Error in property search and analysis: {e!s}"
             )
             raise
 

new_features_examples/sequential_workflow/sequential_worflow_test 2.py

@@ -1,8 +1,10 @@
 import os
+
 from dotenv import load_dotenv
-from swarms import Agent, SequentialWorkflow
 from swarm_models import OpenAIChat
 
+from swarms import Agent, SequentialWorkflow
+
 load_dotenv()
 
 # Get the OpenAI API key from the environment variable

new_features_examples/sequential_workflow/sequential_worflow_test.py

@@ -1,8 +1,10 @@
 import os
+
 from dotenv import load_dotenv
-from swarms import Agent, SequentialWorkflow
 from swarm_models import OpenAIChat
 
+from swarms import Agent, SequentialWorkflow
+
 load_dotenv()
 
 # Get the OpenAI API key from the environment variable

new_features_examples/sequential_workflow/sequential_workflow 2.py

@@ -1,8 +1,10 @@
 import os
+
 from dotenv import load_dotenv
-from swarms import Agent, SequentialWorkflow
 from swarm_models import OpenAIChat
 
+from swarms import Agent, SequentialWorkflow
+
 load_dotenv()
 
 # Get the OpenAI API key from the environment variable

new_features_examples/sequential_workflow/sequential_workflow.py

@@ -1,8 +1,10 @@
 import os
+
 from dotenv import load_dotenv
-from swarms import Agent, SequentialWorkflow
 from swarm_models import OpenAIChat
 
+from swarms import Agent, SequentialWorkflow
+
 load_dotenv()
 
 # Get the OpenAI API key from the environment variable

new_features_examples/solana_agent.py

@@ -1,21 +1,22 @@
-from dataclasses import dataclass
-from typing import List, Optional, Dict, Any
-from datetime import datetime
 import asyncio
-from loguru import logger
 import json
-import base58
+from dataclasses import dataclass
+from datetime import datetime
 from decimal import Decimal
+from typing import Any, Optional
 
-# Swarms imports
-from swarms import Agent
+import aiohttp
+import base58
+from anchorpy import Provider, Wallet
+from loguru import logger
+from solders.keypair import Keypair
 
 # Solana imports
 from solders.rpc.responses import GetTransactionResp
 from solders.transaction import Transaction
-from anchorpy import Provider, Wallet
-from solders.keypair import Keypair
-import aiohttp
+
+# Swarms imports
+from swarms import Agent
 
 # Specialized Solana Analysis System Prompt
 SOLANA_ANALYSIS_PROMPT = """You are a specialized Solana blockchain analyst agent. Your role is to:
@@ -74,14 +75,14 @@ class TransactionData:
     to_address: str
     program_id: str
     instruction_data: Optional[str] = None
-    program_logs: List[str] = None
+    program_logs: list[str] = None
 
     @property
     def sol_amount(self) -> Decimal:
         """Convert lamports to SOL"""
         return Decimal(self.lamports) / Decimal(1e9)
 
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self) -> dict[str, Any]:
         """Convert transaction data to dictionary for agent analysis"""
         return {
             "signature": self.signature,
@@ -132,7 +133,7 @@ class SolanaSwarmAgent:
 
     async def analyze_transaction(
         self, tx_data: TransactionData
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """Analyze a transaction using the specialized agent"""
         try:
             # Update recent transactions for pattern analysis
@@ -192,7 +193,7 @@ class SolanaSwarmAgent:
             return json.loads(analysis)
 
         except Exception as e:
-            logger.error(f"Error in agent analysis: {str(e)}")
+            logger.error(f"Error in agent analysis: {e!s}")
             return {
                 "analysis_type": "error",
                 "severity": "low",
@@ -258,7 +259,7 @@ class SolanaTransactionMonitor:
                 ),
             )
         except Exception as e:
-            logger.error(f"Failed to parse transaction: {str(e)}")
+            logger.error(f"Failed to parse transaction: {e!s}")
             return None
 
     async def start_monitoring(self):
@@ -323,7 +324,7 @@ class SolanaTransactionMonitor:
 
                         except Exception as e:
                             logger.error(
-                                f"Error processing message: {str(e)}"
+                                f"Error processing message: {e!s}"
                             )
                             continue
 

new_features_examples/solana_tool/solana_tool.py

@@ -1,10 +1,10 @@
-from datetime import datetime
 import json
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
 import requests
 from loguru import logger
-from dataclasses import dataclass
-from datetime import timezone
-import time
 from requests.adapters import HTTPAdapter
 from urllib3.util.retry import Retry
 
@@ -86,7 +86,7 @@ def get_working_endpoint(session: requests.Session) -> str:
                 return endpoint
         except Exception as e:
             logger.warning(
-                f"Endpoint {endpoint} failed health check: {str(e)}"
+                f"Endpoint {endpoint} failed health check: {e!s}"
             )
             continue
 
@@ -189,7 +189,7 @@ def fetch_wallet_transactions(wallet_address: str) -> str:
 
             except Exception as e:
                 logger.error(
-                    f"Error during transaction fetch: {str(e)}"
+                    f"Error during transaction fetch: {e!s}"
                 )
                 # Try to get a new endpoint if the current one fails
                 api_endpoint = get_working_endpoint(session)
@@ -217,7 +217,7 @@ def fetch_wallet_transactions(wallet_address: str) -> str:
                 )
                 tx_data = response.json()
 
-                if "result" in tx_data and tx_data["result"]:
+                if tx_data.get("result"):
                     enriched_transactions.append(
                         {
                             "signature": tx["signature"],
@@ -240,7 +240,7 @@ def fetch_wallet_transactions(wallet_address: str) -> str:
 
             except Exception as e:
                 logger.warning(
-                    f"Failed to fetch details for transaction {tx['signature']}: {str(e)}"
+                    f"Failed to fetch details for transaction {tx['signature']}: {e!s}"
                 )
                 continue
 
@@ -272,7 +272,7 @@ def fetch_wallet_transactions(wallet_address: str) -> str:
     except Exception as e:
         error = TransactionError(
             error_type="UNKNOWN_ERROR",
-            message=f"An unexpected error occurred: {str(e)}",
+            message=f"An unexpected error occurred: {e!s}",
         )
         logger.error(f"Unexpected error: {error.message}")
         return json.dumps(
@@ -292,4 +292,4 @@ if __name__ == "__main__":
         result = fetch_wallet_transactions(wallet)
         print(json.dumps(json.loads(result), indent=2))
     except Exception as e:
-        logger.error(f"Failed to fetch transactions: {str(e)}")
+        logger.error(f"Failed to fetch transactions: {e!s}")

new_features_examples/solana_tool/solana_tool_test.py

@@ -1,12 +1,11 @@
-from typing import List
-from datetime import datetime
 import json
+import random
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
 import requests
 from loguru import logger
-from dataclasses import dataclass
-from datetime import timezone
-import time
-import random
 
 # Configure loguru logger
 logger.add(
@@ -43,7 +42,7 @@ class SolanaAPIException(Exception):
 class RPCEndpointManager:
     """Manages RPC endpoints and handles switching between them"""
 
-    def __init__(self, endpoints: List[str]):
+    def __init__(self, endpoints: list[str]):
         self.endpoints = endpoints.copy()
         self.current_endpoint = self.endpoints[0]
         self.last_request_time = 0
@@ -132,20 +131,20 @@ def make_request(
             requests.exceptions.ConnectionError,
         ) as e:
             logger.warning(
-                f"Connection error with {endpoint}: {str(e)}"
+                f"Connection error with {endpoint}: {e!s}"
             )
             endpoint_manager.switch_endpoint()
             continue
 
         except Exception as e:
             last_error = e
-            logger.warning(f"Request failed: {str(e)}")
+            logger.warning(f"Request failed: {e!s}")
             endpoint_manager.switch_endpoint()
             time.sleep(1)
             continue
 
     raise SolanaAPIException(
-        f"All retry attempts failed. Last error: {str(last_error)}"
+        f"All retry attempts failed. Last error: {last_error!s}"
     )
 
 
@@ -231,7 +230,7 @@ def fetch_wallet_transactions(
 
                 tx_data = make_request(endpoint_manager, tx_payload)
 
-                if "result" in tx_data and tx_data["result"]:
+                if tx_data.get("result"):
                     result = tx_data["result"]
                     enriched_tx = {
                         "signature": tx["signature"],
@@ -257,7 +256,7 @@ def fetch_wallet_transactions(
 
             except Exception as e:
                 logger.warning(
-                    f"Failed to process transaction {tx['signature']}: {str(e)}"
+                    f"Failed to process transaction {tx['signature']}: {e!s}"
                 )
                 continue
 
@@ -299,4 +298,4 @@ if __name__ == "__main__":
         result = fetch_wallet_transactions(wallet)
         print(result)
     except Exception as e:
-        logger.error(f"Failed to fetch transactions: {str(e)}")
+        logger.error(f"Failed to fetch transactions: {e!s}")

new_features_examples/spike/agent_rearrange_test.py

@@ -7,11 +7,12 @@ Todo
 """
 
 import os
+
 from dotenv import load_dotenv
-from swarms import Agent, AgentRearrange
-from swarm_models import OpenAIChat, OpenAIFunctionCaller
 from pydantic import BaseModel
-from typing import List
+from swarm_models import OpenAIChat, OpenAIFunctionCaller
+
+from swarms import Agent, AgentRearrange
 
 
 class CollegeLog(BaseModel):
@@ -21,7 +22,7 @@ class CollegeLog(BaseModel):
 
 
 class CollegesRecommendation(BaseModel):
-    colleges: List[CollegeLog]
+    colleges: list[CollegeLog]
     reasoning: str
 
 

new_features_examples/spike/function_caller_example.py

@@ -7,10 +7,10 @@ Todo
 """
 
 import os
+
 from dotenv import load_dotenv
-from swarm_models import OpenAIChat, OpenAIFunctionCaller
 from pydantic import BaseModel
-from typing import List
+from swarm_models import OpenAIChat, OpenAIFunctionCaller
 
 
 class CollegeLog(BaseModel):
@@ -20,7 +20,7 @@ class CollegeLog(BaseModel):
 
 
 class CollegesRecommendation(BaseModel):
-    colleges: List[CollegeLog]
+    colleges: list[CollegeLog]
     reasoning: str
 
 

new_features_examples/spike/memory.py

@@ -1,7 +1,8 @@
-from typing import Optional
 from pathlib import Path
+from typing import Optional
+
+from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
 from loguru import logger
-from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
 
 
 class LlamaIndexDB:
@@ -65,7 +66,7 @@ class LlamaIndexDB:
                 f"Successfully indexed documents from {self.data_dir}"
             )
         except Exception as e:
-            logger.error(f"Error indexing documents: {str(e)}")
+            logger.error(f"Error indexing documents: {e!s}")
             raise
 
     def query(self, query: str, **kwargs) -> str:
@@ -96,7 +97,7 @@ class LlamaIndexDB:
             logger.info(f"Successfully queried: {query}")
             return str(response)
         except Exception as e:
-            logger.error(f"Error during query: {str(e)}")
+            logger.error(f"Error during query: {e!s}")
             raise
 
 

new_features_examples/spike/test.py

@@ -7,11 +7,12 @@ Todo
 """
 
 import os
+
 from dotenv import load_dotenv
-from swarms import Agent, SequentialWorkflow
-from swarm_models import OpenAIChat, OpenAIFunctionCaller
 from pydantic import BaseModel
-from typing import List
+from swarm_models import OpenAIChat, OpenAIFunctionCaller
+
+from swarms import Agent, SequentialWorkflow
 
 
 class CollegeLog(BaseModel):
@@ -21,7 +22,7 @@ class CollegeLog(BaseModel):
 
 
 class CollegesRecommendation(BaseModel):
-    colleges: List[CollegeLog]
+    colleges: list[CollegeLog]
     reasoning: str
 
 

new_features_examples/swarmarrange/rearrange_test.py

@@ -1,9 +1,9 @@
 import os
 
-from swarms import Agent, AgentRearrange
-
 from swarm_models import OpenAIChat
 
+from swarms import Agent, AgentRearrange
+
 # Get the OpenAI API key from the environment variable
 api_key = os.getenv("OPENAI_API_KEY")
 

new_features_examples/voice.py

@@ -4,8 +4,9 @@ import asyncio
 import base64
 import io
 import threading
+from collections.abc import Awaitable
 from os import getenv
-from typing import Any, Awaitable, Callable, cast
+from typing import Any, Callable, cast
 
 import numpy as np
 
@@ -81,7 +82,7 @@ class AudioPlayerAsync:
         self.playing = False
         self._frame_count = 0
 
-    def callback(self, outdata, frames, time, status):  # noqa
+    def callback(self, outdata, frames, time, status):
         with self.lock:
             data = np.empty(0, dtype=np.int16)
 
@@ -204,7 +205,7 @@ class RealtimeApp:
         - Sends text prompts to the GPT-4 Realtime API.
     """
 
-    def __init__(self, system_prompt: str = None) -> None:
+    def __init__(self, system_prompt: str | None = None) -> None:
         self.connection: AsyncRealtimeConnection | None = None
         self.session: Session | None = None
         self.client = AsyncOpenAI(api_key=getenv("OPENAI_API_KEY"))

pyproject.toml

@@ -121,6 +121,17 @@ pytest = "^8.1.1"
 
 [tool.ruff]
 line-length = 70
+fix = true
+format.preview = true
+lint.select = [
+  # pyflakes
+  "F",
+  # isort
+  "I",
+  # pyupgrade
+  "UP",
+  "W"
+]
 
 [tool.black]
 target-version = ["py38"]

run_all_tests.py

@@ -1,7 +1,8 @@
+import logging
 import os
 import subprocess
-import logging
 import time
+
 import psutil
 
 # Configure logging

scripts/auto_tests_docs/auto_docs.py

@@ -4,9 +4,9 @@ import os
 import threading
 
 from dotenv import load_dotenv
+from swarm_models import OpenAIChat
 
 from scripts.auto_tests_docs.docs import DOCUMENTATION_WRITER_SOP
-from swarm_models import OpenAIChat
 from swarms.structs.majority_voting import MajorityVoting
 from swarms.structs.stackoverflow_swarm import StackOverflowSwarm
 from swarms.structs.task_queue_base import TaskQueueBase

scripts/auto_tests_docs/auto_docs_functions.py

@@ -4,9 +4,9 @@ import sys
 import threading
 
 from dotenv import load_dotenv
+from swarm_models import OpenAIChat
 
 from scripts.auto_tests_docs.docs import DOCUMENTATION_WRITER_SOP
-from swarm_models import OpenAIChat
 
 load_dotenv()
 

scripts/auto_tests_docs/auto_docs_omni.py

@@ -3,9 +3,9 @@ import os
 import threading
 
 from dotenv import load_dotenv
+from swarm_models import OpenAIChat
 
 from scripts.auto_tests_docs.docs import DOCUMENTATION_WRITER_SOP
-from swarm_models import OpenAIChat
 
 ###########
 

scripts/auto_tests_docs/auto_tests.py

@@ -4,10 +4,10 @@ import re
 import threading
 
 from dotenv import load_dotenv
+from swarm_models import OpenAIChat
 from swarms_memory import DictInternalMemory, DictSharedMemory
 
 from scripts.auto_tests_docs.docs import TEST_WRITER_SOP_PROMPT
-from swarm_models import OpenAIChat
 
 load_dotenv()
 

scripts/auto_tests_docs/auto_tests_functions.py

@@ -4,9 +4,9 @@ import sys
 import threading
 
 from dotenv import load_dotenv
+from swarm_models import OpenAIChat
 
 from scripts.auto_tests_docs.docs import TEST_WRITER_SOP_PROMPT
-from swarm_models import OpenAIChat
 from swarms.utils.parse_code import extract_code_from_markdown
 
 load_dotenv()

swarms/__init__.py

@@ -2,15 +2,15 @@ from dotenv import load_dotenv
 
 load_dotenv()
 
-from swarms.telemetry.bootup import bootup  # noqa: E402, F403
+from swarms.telemetry.bootup import bootup
 
 bootup()
 
-from swarms.agents import *  # noqa: E402, F403
-from swarms.artifacts import *  # noqa: E402, F403
-from swarms.prompts import *  # noqa: E402, F403
-from swarms.schemas import *  # noqa: E402, F403
-from swarms.structs import *  # noqa: E402, F403
-from swarms.telemetry import *  # noqa: E402, F403
-from swarms.tools import *  # noqa: E402, F403
-from swarms.utils import *  # noqa: E402, F403
+from swarms.agents import *  # noqa: F403
+from swarms.artifacts import *  # noqa: F403
+from swarms.prompts import *  # noqa: F403
+from swarms.schemas import *  # noqa: F403
+from swarms.structs import *  # noqa: F403
+from swarms.telemetry import *  # noqa: F403
+from swarms.tools import *  # noqa: F403
+from swarms.utils import *  # noqa: F403

swarms/agents/__init__.py

@@ -1,3 +1,7 @@
+from swarms.agents.create_agents_from_yaml import (
+    create_agents_from_yaml,
+)
+from swarms.agents.tool_agent import ToolAgent
 from swarms.structs.stopping_conditions import (
     check_cancelled,
     check_complete,
@@ -10,10 +14,6 @@ from swarms.structs.stopping_conditions import (
     check_stopped,
     check_success,
 )
-from swarms.agents.tool_agent import ToolAgent
-from swarms.agents.create_agents_from_yaml import (
-    create_agents_from_yaml,
-)
 
 __all__ = [
     "ToolAgent",

swarms/agents/ape_agent.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, Optional
 
 from tenacity import retry, stop_after_attempt, wait_exponential
 
@@ -18,7 +18,7 @@ logger = initialize_logger(log_folder="ape_agent")
     wait=wait_exponential(multiplier=1, min=4, max=10),
 )
 def auto_generate_prompt(
-    task: str = None,
+    task: Optional[str] = None,
     model: Any = None,
     max_tokens: int = 4000,
     use_second_sys_prompt: bool = True,
@@ -49,5 +49,5 @@ def auto_generate_prompt(
         print(output)
         return output
     except Exception as e:
-        logger.error(f"Error generating prompt: {str(e)}")
+        logger.error(f"Error generating prompt: {e!s}")
         raise

swarms/agents/auto_generate_swarm_config.py

@@ -245,7 +245,7 @@ def generate_swarm_config(
 
         except Exception as e:
             formatter.print_panel(
-                f"Error generating swarm configuration: {str(e)}",
+                f"Error generating swarm configuration: {e!s}",
                 "Error",
             )
             raise

swarms/agents/create_agents_from_yaml.py

@@ -1,22 +1,23 @@
 import os
-from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+from typing import Any, Callable, Optional, Union
 
 import yaml
-from tenacity import (
-    retry,
-    stop_after_attempt,
-    wait_exponential,
-    retry_if_exception_type,
-)
 from pydantic import (
     BaseModel,
     Field,
     field_validator,
 )
-from swarms.utils.loguru_logger import initialize_logger
+from tenacity import (
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_exponential,
+)
+
 from swarms.structs.agent import Agent
 from swarms.structs.swarm_router import SwarmRouter
 from swarms.utils.litellm_wrapper import LiteLLM
+from swarms.utils.loguru_logger import initialize_logger
 
 logger = initialize_logger(log_folder="create_agents_from_yaml")
 
@@ -57,7 +58,7 @@ class SwarmConfig(BaseModel):
     max_loops: int = Field(default=1, ge=1)
     swarm_type: str
     task: Optional[str] = None
-    flow: Optional[Dict] = None
+    flow: Optional[dict] = None
     autosave: bool = True
     return_json: bool = False
     rules: str = ""
@@ -80,7 +81,7 @@ class SwarmConfig(BaseModel):
 
 
 class YAMLConfig(BaseModel):
-    agents: List[AgentConfig] = Field(..., min_length=1)
+    agents: list[AgentConfig] = Field(..., min_length=1)
     swarm_architecture: Optional[SwarmConfig] = None
 
     model_config = {
@@ -89,8 +90,8 @@ class YAMLConfig(BaseModel):
 
 
 def load_yaml_safely(
-    yaml_file: str = None, yaml_string: str = None
-) -> Dict:
+    yaml_file: Optional[str] = None, yaml_string: Optional[str] = None
+) -> dict:
     """Safely load and validate YAML configuration using Pydantic."""
     try:
         if yaml_string:
@@ -100,7 +101,7 @@ def load_yaml_safely(
                 raise FileNotFoundError(
                     f"YAML file {yaml_file} not found."
                 )
-            with open(yaml_file, "r") as file:
+            with open(yaml_file) as file:
                 config_dict = yaml.safe_load(file)
         else:
             raise ValueError(
@@ -111,9 +112,9 @@ def load_yaml_safely(
         YAMLConfig(**config_dict)
         return config_dict
     except yaml.YAMLError as e:
-        raise ValueError(f"Error parsing YAML: {str(e)}")
+        raise ValueError(f"Error parsing YAML: {e!s}")
     except Exception as e:
-        raise ValueError(f"Error validating configuration: {str(e)}")
+        raise ValueError(f"Error validating configuration: {e!s}")
 
 
 @retry(
@@ -125,7 +126,7 @@ def load_yaml_safely(
     ),
 )
 def create_agent_with_retry(
-    agent_config: Dict, model: LiteLLM
+    agent_config: dict, model: LiteLLM
 ) -> Agent:
     """Create an agent with retry logic for handling transient failures."""
     try:
@@ -153,22 +154,22 @@ def create_agent_with_retry(
         return agent
     except Exception as e:
         logger.error(
-            f"Error creating agent {agent_config.get('agent_name', 'unknown')}: {str(e)}"
+            f"Error creating agent {agent_config.get('agent_name', 'unknown')}: {e!s}"
         )
         raise
 
 
 def create_agents_from_yaml(
-    model: Callable = None,
+    model: Optional[Callable] = None,
     yaml_file: str = "agents.yaml",
-    yaml_string: str = None,
+    yaml_string: Optional[str] = None,
     return_type: str = "auto",
 ) -> Union[
     SwarmRouter,
     Agent,
-    List[Agent],
-    Tuple[Union[SwarmRouter, Agent], List[Agent]],
-    List[Dict[str, Any]],
+    list[Agent],
+    tuple[Union[SwarmRouter, Agent], list[Agent]],
+    list[dict[str, Any]],
 ]:
     """
     Create agents and/or SwarmRouter based on configurations defined in a YAML file or string.
@@ -225,9 +226,9 @@ def create_agents_from_yaml(
                     f"SwarmRouter '{swarm_config.name}' created successfully."
                 )
             except Exception as e:
-                logger.error(f"Error creating SwarmRouter: {str(e)}")
+                logger.error(f"Error creating SwarmRouter: {e!s}")
                 raise ValueError(
-                    f"Failed to create SwarmRouter: {str(e)}"
+                    f"Failed to create SwarmRouter: {e!s}"
                 )
 
         # Handle return types with improved error checking
@@ -254,7 +255,7 @@ def create_agents_from_yaml(
                     config["swarm_architecture"]["task"]
                 )
             except Exception as e:
-                logger.error(f"Error running SwarmRouter: {str(e)}")
+                logger.error(f"Error running SwarmRouter: {e!s}")
                 raise
 
         # Return appropriate type based on configuration
@@ -276,13 +277,15 @@ def create_agents_from_yaml(
             return (
                 swarm_router
                 if swarm_router
-                else agents[0] if len(agents) == 1 else agents
+                else agents[0]
+                if len(agents) == 1
+                else agents
             ), agents
         elif return_type == "tasks":
             return task_results
 
     except Exception as e:
         logger.error(
-            f"Critical error in create_agents_from_yaml: {str(e)}"
+            f"Critical error in create_agents_from_yaml: {e!s}"
         )
         raise

swarms/agents/openai_assistant.py

@@ -3,7 +3,7 @@ import os
 import subprocess
 import sys
 import time
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Callable, Optional
 
 from loguru import logger
 
@@ -58,10 +58,10 @@ class OpenAIAssistant(Agent):
         description: str = "Standard openai assistant wrapper",
         instructions: Optional[str] = None,
         model: str = "gpt-4o",
-        tools: Optional[List[Dict[str, Any]]] = None,
-        file_ids: Optional[List[str]] = None,
-        metadata: Optional[Dict[str, Any]] = None,
-        functions: Optional[List[Dict[str, Any]]] = None,
+        tools: Optional[list[dict[str, Any]]] = None,
+        file_ids: Optional[list[str]] = None,
+        metadata: Optional[dict[str, Any]] = None,
+        functions: Optional[list[dict[str, Any]]] = None,
         *args,
         **kwargs,
     ):
@@ -110,13 +110,13 @@ class OpenAIAssistant(Agent):
         )
 
         # Store available functions
-        self.available_functions: Dict[str, Callable] = {}
+        self.available_functions: dict[str, Callable] = {}
 
     def add_function(
         self,
         func: Callable,
         description: str,
-        parameters: Dict[str, Any],
+        parameters: dict[str, Any],
     ) -> None:
         """Add a function that the assistant can call.
 
@@ -246,7 +246,7 @@ class OpenAIAssistant(Agent):
         self.thread = self.client.beta.threads.create()
 
     def add_message(
-        self, content: str, file_ids: Optional[List[str]] = None
+        self, content: str, file_ids: Optional[list[str]] = None
     ) -> None:
         """Add a message to the thread.
 

swarms/agents/tool_agent.py

@@ -1,7 +1,8 @@
-from typing import Any, Optional, Callable
+from typing import Any, Callable, Optional
+
 from swarms.tools.json_former import Jsonformer
-from swarms.utils.loguru_logger import initialize_logger
 from swarms.utils.lazy_loader import lazy_import_decorator
+from swarms.utils.loguru_logger import initialize_logger
 
 logger = initialize_logger(log_folder="tool_agent")
 

swarms/artifacts/main_artifact.py

@@ -3,7 +3,7 @@ import os
 import subprocess
 import time
 from datetime import datetime
-from typing import Any, Dict, List, Union
+from typing import Any, Union
 
 from pydantic import BaseModel, Field
 from pydantic.v1 import validator
@@ -60,7 +60,7 @@ class Artifact(BaseModel):
     contents: str = Field(
         ..., description="The contents of the file in string format"
     )
-    versions: List[FileVersion] = Field(default_factory=list)
+    versions: list[FileVersion] = Field(default_factory=list)
     edit_count: int = Field(
         ...,
         description="The number of times the file has been edited",
@@ -157,7 +157,7 @@ class Artifact(BaseModel):
         """
         Loads the file contents from the specified file path into the artifact.
         """
-        with open(self.file_path, "r") as f:
+        with open(self.file_path) as f:
             self.contents = f.read()
         self.create(self.contents)
 
@@ -207,7 +207,7 @@ class Artifact(BaseModel):
         Returns:
             Artifact: The imported artifact instance.
         """
-        with open(file_path, "r") as json_file:
+        with open(file_path) as json_file:
             data = json.load(json_file)
         # Convert timestamp strings back to datetime objects
         for version in data["versions"]:
@@ -232,14 +232,14 @@ class Artifact(BaseModel):
         )
         return metrics
 
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self) -> dict[str, Any]:
         """
         Converts the artifact instance to a dictionary representation.
         """
         return self.dict()
 
     @classmethod
-    def from_dict(cls, data: Dict[str, Any]) -> "Artifact":
+    def from_dict(cls, data: dict[str, Any]) -> "Artifact":
         """
         Creates an artifact instance from a dictionary representation.
         """

swarms/cli/create_agent.py

@@ -39,5 +39,5 @@ def run_agent_by_name(
 
         return output
     except Exception as e:
-        print(f"An error occurred: {str(e)}")
+        print(f"An error occurred: {e!s}")
         return None

swarms/cli/main.py

@@ -2,6 +2,7 @@ import argparse
 import os
 import time
 import webbrowser
+from typing import Optional
 
 from rich.console import Console
 from rich.panel import Panel
@@ -114,7 +115,7 @@ def show_help():
     )
 
 
-def show_error(message: str, help_text: str = None):
+def show_error(message: str, help_text: Optional[str] = None):
     """Display error message in a formatted panel"""
     error_panel = Panel(
         f"[bold red]{message}[/bold red]",
@@ -153,7 +154,7 @@ def check_login():
     cache_file = "cache.txt"
 
     if os.path.exists(cache_file):
-        with open(cache_file, "r") as f:
+        with open(cache_file) as f:
             if f.read() == "logged_in":
                 console.print(
                     f"[{COLORS['success']}]✓ Authentication verified[/{COLORS['success']}]"
@@ -218,7 +219,7 @@ def run_autoswarm(task: str, model: str):
             )
         else:
             show_error(
-                f"Error during autoswarm execution: {str(e)}",
+                f"Error during autoswarm execution: {e!s}",
                 "For debugging, try:\n"
                 + "1. Check your API keys are set correctly\n"
                 + "2. Verify your network connection\n"
@@ -293,7 +294,7 @@ def main():
                 run_autoswarm(args.task, args.model)
         except Exception as e:
             console.print(
-                f"[{COLORS['error']}]Error: {str(e)}[/{COLORS['error']}]"
+                f"[{COLORS['error']}]Error: {e!s}[/{COLORS['error']}]"
             )
             return
     except Exception as error:

swarms/cli/onboarding_process.py

@@ -1,15 +1,12 @@
 import json
 import os
 import time
-from typing import Dict
-
-from swarms.utils.loguru_logger import initialize_logger
-
 
 from swarms.telemetry.capture_sys_data import (
     capture_system_data,
     log_agent_data,
 )
+from swarms.utils.loguru_logger import initialize_logger
 
 logger = initialize_logger(log_folder="onboarding_process")
 
@@ -33,8 +30,8 @@ class OnboardingProcess:
             auto_save_path (str): The path where user data is automatically saved.
             cache_save_path (str): The path where user data is cached for reliability.
         """
-        self.user_data: Dict[str, str] = {}
-        self.system_data: Dict[str, str] = capture_system_data()
+        self.user_data: dict[str, str] = {}
+        self.system_data: dict[str, str] = capture_system_data()
         self.auto_save_path = auto_save_path
         self.cache_save_path = cache_save_path
         self.load_existing_data()
@@ -45,7 +42,7 @@ class OnboardingProcess:
         """
         if os.path.exists(self.auto_save_path):
             try:
-                with open(self.auto_save_path, "r") as f:
+                with open(self.auto_save_path) as f:
                     self.user_data = json.load(f)
                     logger.info(
                         "Existing user data loaded from {}",
@@ -60,7 +57,7 @@ class OnboardingProcess:
         # Fallback to cache if main file fails
         if os.path.exists(self.cache_save_path):
             try:
-                with open(self.cache_save_path, "r") as f:
+                with open(self.cache_save_path) as f:
                     self.user_data = json.load(f)
                     logger.info(
                         "User data loaded from cache: {}",

swarms/prompts/agent_prompt.py

@@ -1,5 +1,4 @@
 import json
-from typing import List
 
 
 class PromptGenerator:
@@ -7,10 +6,10 @@ class PromptGenerator:
 
     def __init__(self) -> None:
         """Initialize the PromptGenerator object."""
-        self.constraints: List[str] = []
-        self.commands: List[str] = []
-        self.resources: List[str] = []
-        self.performance_evaluation: List[str] = []
+        self.constraints: list[str] = []
+        self.commands: list[str] = []
+        self.resources: list[str] = []
+        self.performance_evaluation: list[str] = []
         self.response_format = {
             "thoughts": {
                 "text": "thought",

swarms/prompts/agent_system_prompts.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from swarms.prompts.tools import (
     DYNAMIC_STOP_PROMPT,
     DYNAMICAL_TOOL_USAGE,
@@ -24,7 +26,7 @@ You are programmed to follow these rules:
 def autonomous_agent_prompt_v2(
     tools_prompt: str = DYNAMICAL_TOOL_USAGE,
     dynamic_stop_prompt: str = DYNAMIC_STOP_PROMPT,
-    agent_name: str = None,
+    agent_name: Optional[str] = None,
 ):
     return f"""
     You are {agent_name}, an elite autonomous agent operating within a sophisticated autonomous loop structure.
@@ -79,7 +81,7 @@ to aid in these complex tasks. Your responses should be coherent, contextually r
 def autonomous_agent_prompt(
     tools_prompt: str = DYNAMICAL_TOOL_USAGE,
     dynamic_stop_prompt: str = DYNAMIC_STOP_PROMPT,
-    agent_name: str = None,
+    agent_name: Optional[str] = None,
 ):
     """Autonomous agent prompt"""
     return f"""

swarms/prompts/chat_prompt.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import abstractmethod
-from typing import Sequence
+from collections.abc import Sequence
 
 
 class Message:
@@ -11,7 +11,7 @@ class Message:
     """
 
     def __init__(
-        self, content: str, role: str, additional_kwargs: dict = None
+        self, content: str, role: str, additional_kwargs: dict | None = None
     ):
         self.content = content
         self.role = role
@@ -33,7 +33,7 @@ class HumanMessage(Message):
         self,
         content: str,
         role: str = "Human",
-        additional_kwargs: dict = None,
+        additional_kwargs: dict | None = None,
         example: bool = False,
     ):
         super().__init__(content, role, additional_kwargs)
@@ -52,7 +52,7 @@ class AIMessage(Message):
         self,
         content: str,
         role: str = "AI",
-        additional_kwargs: dict = None,
+        additional_kwargs: dict | None = None,
         example: bool = False,
     ):
         super().__init__(content, role, additional_kwargs)
@@ -72,7 +72,7 @@ class SystemMessage(Message):
         self,
         content: str,
         role: str = "System",
-        additional_kwargs: dict = None,
+        additional_kwargs: dict | None = None,
     ):
         super().__init__(content, role, additional_kwargs)
 
@@ -89,8 +89,8 @@ class FunctionMessage(Message):
         self,
         content: str,
         role: str = "Function",
-        name: str = None,
-        additional_kwargs: dict = None,
+        name: str | None = None,
+        additional_kwargs: dict | None = None,
     ):
         super().__init__(content, role, additional_kwargs)
         self.name = name
@@ -105,7 +105,7 @@ class ChatMessage(Message):
     """
 
     def __init__(
-        self, content: str, role: str, additional_kwargs: dict = None
+        self, content: str, role: str, additional_kwargs: dict | None = None
     ):
         super().__init__(content, role, additional_kwargs)
 

swarms/prompts/prompt.py

@@ -2,7 +2,7 @@ import json
 import os
 import time
 import uuid
-from typing import Any, Callable, List
+from typing import Any, Callable
 
 from pydantic import (
     BaseModel,
@@ -64,7 +64,7 @@ class Prompt(BaseModel):
         default=0,
         description="The number of times the prompt has been edited",
     )
-    edit_history: List[str] = Field(
+    edit_history: list[str] = Field(
         default_factory=list,
         description="The history of edits, storing all prompt versions",
     )
@@ -227,7 +227,7 @@ class Prompt(BaseModel):
             "Persistent storage integration is required."
         )
 
-    def add_tools(self, tools: List[Callable]) -> str:
+    def add_tools(self, tools: list[Callable]) -> str:
         tools_prompt = BaseTool(
             tools=tools, tool_system_prompt=None
         ).convert_tool_into_openai_schema()

swarms/prompts/prompt_generator_optimizer.py

@@ -1,6 +1,5 @@
 from swarms.prompts.prompt import Prompt
 
-
 OPENAI_PROMPT_GENERATOR_SYS_PROMPT = """
 
 Given a task description or existing prompt, produce a detailed system prompt to guide a language model in completing the task effectively.

swarms/prompts/react.py

@@ -1,4 +1,7 @@
-def react_prompt(task: str = None):
+from typing import Optional
+
+
+def react_prompt(task: Optional[str] = None):
     PROMPT = f"""
         Task Description:
         Accomplish the following {task} using the reasoning guidelines below.

swarms/prompts/worker_prompt.py

@@ -1,5 +1,4 @@
 import datetime
-from typing import List
 
 from pydantic import BaseModel, Field
 
@@ -113,7 +112,7 @@ browser_and_terminal_tool_two = """
 
 
 # Function to parse tools and get their documentation
-def parse_tools(tools: List[BaseTool] = []):
+def parse_tools(tools: list[BaseTool] = []):
     tool_docs = []
     for tool in tools:
         tool_doc = scrape_tool_func_docs(tool)
@@ -123,7 +122,7 @@ def parse_tools(tools: List[BaseTool] = []):
 
 # Function to generate the worker prompt
 def tool_usage_worker_prompt(
-    current_time=time, tools: List[callable] = []
+    current_time=time, tools: list[callable] = []
 ):
     tool_docs = BaseTool(verbose=True, functions=tools)
 

swarms/schemas/__init__.py

@@ -1,7 +1,5 @@
-from swarms.schemas.agent_step_schemas import Step, ManySteps
-
 from swarms.schemas.agent_input_schema import AgentSchema
-
+from swarms.schemas.agent_step_schemas import ManySteps, Step
 
 __all__ = [
     "Step",

swarms/schemas/agent_input_schema.py

@@ -1,4 +1,5 @@
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Callable, Optional
+
 from pydantic import BaseModel, Field
 from pydantic.v1 import validator
 
@@ -28,10 +29,10 @@ class AgentSchema(BaseModel):
     interactive: Optional[bool] = Field(default=False)
     dashboard: Optional[bool] = Field(default=False)
     agent_description: Optional[str] = Field(default=None)
-    tools: Optional[List[Callable]] = Field(default=None)
+    tools: Optional[list[Callable]] = Field(default=None)
     dynamic_temperature_enabled: Optional[bool] = Field(default=False)
     sop: Optional[str] = Field(default=None)
-    sop_list: Optional[List[str]] = Field(default=None)
+    sop_list: Optional[list[str]] = Field(default=None)
     saved_state_path: Optional[str] = Field(default=None)
     autosave: Optional[bool] = Field(default=False)
     self_healing_enabled: Optional[bool] = Field(default=False)
@@ -45,14 +46,14 @@ class AgentSchema(BaseModel):
     traceback: Optional[Any] = Field(default=None)
     traceback_handlers: Optional[Any] = Field(default=None)
     streaming_on: Optional[bool] = Field(default=False)
-    docs: Optional[List[str]] = Field(default=None)
+    docs: Optional[list[str]] = Field(default=None)
     docs_folder: Optional[str] = Field(default=None)
     verbose: Optional[bool] = Field(default=False)
     parser: Optional[Callable] = Field(default=None)
     best_of_n: Optional[int] = Field(default=None)
     callback: Optional[Callable] = Field(default=None)
-    metadata: Optional[Dict[str, Any]] = Field(default=None)
-    callbacks: Optional[List[Callable]] = Field(default=None)
+    metadata: Optional[dict[str, Any]] = Field(default=None)
+    callbacks: Optional[list[Callable]] = Field(default=None)
     logger_handler: Optional[Any] = Field(default=None)
     search_algorithm: Optional[Callable] = Field(default=None)
     logs_to_filename: Optional[str] = Field(default=None)
@@ -72,7 +73,7 @@ class AgentSchema(BaseModel):
     function_calling_format_type: Optional[str] = Field(
         default="OpenAI"
     )
-    list_base_models: Optional[List[Any]] = Field(default=None)
+    list_base_models: Optional[list[Any]] = Field(default=None)
     metadata_output_type: Optional[str] = Field(default="json")
     state_save_file_type: Optional[str] = Field(default="json")
     chain_of_thoughts: Optional[bool] = Field(default=False)

swarms/schemas/agent_step_schemas.py

@@ -2,13 +2,13 @@ from __future__ import annotations
 
 import time
 import uuid
-from typing import List, Optional
 from typing import Any
+
 from pydantic import BaseModel, Field
+
 from swarms.schemas.base_schemas import (
     AgentChatCompletionResponse,
 )
-from typing import Union
 
 
 def get_current_time():
@@ -19,49 +19,49 @@ uuid_hex = uuid.uuid4().hex
 
 
 class Step(BaseModel):
-    step_id: Optional[str] = Field(
+    step_id: str | None = Field(
         default_factory=lambda: uuid.uuid4().hex,
         description="The ID of the task step.",
         examples=["6bb1801a-fd80-45e8-899a-4dd723cc602e"],
     )
-    time: Optional[float] = Field(
+    time: float | None = Field(
         default_factory=get_current_time,
         description="The time taken to complete the task step.",
     )
-    response: Optional[AgentChatCompletionResponse]
+    response: AgentChatCompletionResponse | None
 
 
 class ManySteps(BaseModel):
-    agent_id: Optional[str] = Field(
+    agent_id: str | None = Field(
         ...,
         description="The ID of the agent.",
         examples=["financial-agent-1"],
     )
-    agent_name: Optional[str] = Field(
+    agent_name: str | None = Field(
         ...,
         description="The ID of the agent.",
         examples=["financial-agent-1"],
     )
-    task: Optional[str] = Field(
+    task: str | None = Field(
         ...,
         description="The name of the task.",
         examples=["Write to file"],
     )
-    max_loops: Optional[Any] = Field(
+    max_loops: Any | None = Field(
         ...,
         description="The number of steps in the task.",
         examples=[3],
     )
-    run_id: Optional[str] = Field(
+    run_id: str | None = Field(
         uuid.uuid4().hex,
         description="The ID of the task this step belongs to.",
         examples=["50da533e-3904-4401-8a07-c49adf88b5eb"],
     )
-    steps: Optional[List[Union[Step, Any]]] = Field(
+    steps: list[Step | Any] | None = Field(
         [],
         description="The steps of the task.",
     )
-    full_history: Optional[str] = Field(
+    full_history: str | None = Field(
         ...,
         description="The full history of the task.",
         examples=[
@@ -70,21 +70,21 @@ class ManySteps(BaseModel):
             " <write_to_file('output.txt', 'Washington')"
         ],
     )
-    total_tokens: Optional[int] = Field(
+    total_tokens: int | None = Field(
         ...,
         description="The total number of tokens generated.",
         examples=[7894],
     )
-    stopping_token: Optional[str] = Field(
+    stopping_token: str | None = Field(
         ...,
         description="The token at which the task stopped.",
     )
-    interactive: Optional[bool] = Field(
+    interactive: bool | None = Field(
         ...,
         description="The interactive status of the task.",
         examples=[True],
     )
-    dynamic_temperature_enabled: Optional[bool] = Field(
+    dynamic_temperature_enabled: bool | None = Field(
         ...,
         description="The dynamic temperature status of the task.",
         examples=[True],

swarms/schemas/base_schemas.py

@@ -1,6 +1,6 @@
-import uuid
 import time
-from typing import List, Literal, Optional, Union
+import uuid
+from typing import Literal, Optional, Union
 
 from pydantic import BaseModel, Field
 
@@ -22,7 +22,7 @@ class ModelCard(BaseModel):
 
 class ModelList(BaseModel):
     object: str = "list"
-    data: List[ModelCard] = []
+    data: list[ModelCard] = []
 
 
 class ImageUrl(BaseModel):
@@ -47,7 +47,7 @@ class ChatMessageInput(BaseModel):
         ...,
         description="The role of the message sender. Could be 'user', 'assistant', or 'system'.",
     )
-    content: Union[str, List[ContentItem]]
+    content: Union[str, list[ContentItem]]
 
 
 class ChatMessageResponse(BaseModel):
@@ -65,7 +65,7 @@ class DeltaMessage(BaseModel):
 
 class ChatCompletionRequest(BaseModel):
     model: str = "gpt-4o"
-    messages: List[ChatMessageInput]
+    messages: list[ChatMessageInput]
     temperature: Optional[float] = 0.8
     top_p: Optional[float] = 0.8
     max_tokens: Optional[int] = 4000
@@ -96,7 +96,7 @@ class UsageInfo(BaseModel):
 class ChatCompletionResponse(BaseModel):
     model: str
     object: Literal["chat.completion", "chat.completion.chunk"]
-    choices: List[
+    choices: list[
         Union[
             ChatCompletionResponseChoice,
             ChatCompletionResponseStreamChoice,

swarms/structs/__init__.py

@@ -1,5 +1,6 @@
 from swarms.structs.agent import Agent
 from swarms.structs.agents_available import showcase_available_agents
+from swarms.structs.async_workflow import AsyncWorkflow
 from swarms.structs.auto_swarm import AutoSwarm, AutoSwarmRouter
 from swarms.structs.base_structure import BaseStructure
 from swarms.structs.base_swarm import BaseSwarm
@@ -13,10 +14,10 @@ from swarms.structs.graph_workflow import (
     NodeType,
 )
 from swarms.structs.groupchat import (
-    GroupChat,
+    AgentResponse,
     ChatHistory,
     ChatTurn,
-    AgentResponse,
+    GroupChat,
     expertise_based,
 )
 from swarms.structs.majority_voting import (
@@ -38,6 +39,7 @@ from swarms.structs.multi_agent_exec import (
     run_agents_with_tasks_concurrently,
     run_single_agent,
 )
+from swarms.structs.multi_agent_orchestrator import MultiAgentRouter
 from swarms.structs.queue_swarm import TaskQueueSwarm
 from swarms.structs.rearrange import AgentRearrange, rearrange
 from swarms.structs.round_robin import RoundRobinSwarm
@@ -79,8 +81,6 @@ from swarms.structs.utils import (
     find_token_in_text,
     parse_tasks,
 )
-from swarms.structs.async_workflow import AsyncWorkflow
-from swarms.structs.multi_agent_orchestrator import MultiAgentRouter
 
 __all__ = [
     "Agent",

swarms/structs/agent.py

@@ -11,11 +11,8 @@ from datetime import datetime
 from typing import (
     Any,
     Callable,
-    Dict,
-    List,
     Literal,
     Optional,
-    Tuple,
     Union,
 )
 
@@ -44,6 +41,7 @@ from swarms.structs.safe_loading import (
     SafeLoaderUtils,
     SafeStateManager,
 )
+from swarms.telemetry.capture_sys_data import log_agent_data
 from swarms.tools.base_tool import BaseTool
 from swarms.tools.tool_parse_exec import parse_and_execute_json
 from swarms.utils.data_to_text import data_to_text
@@ -53,7 +51,6 @@ from swarms.utils.pdf_to_text import pdf_to_text
 from swarms.utils.wrapper_clusterop import (
     exec_callable_with_clusterops,
 )
-from swarms.telemetry.capture_sys_data import log_agent_data
 
 
 # Utils
@@ -84,7 +81,7 @@ def exists(val):
 agent_output_type = Literal[
     "string", "str", "list", "json", "dict", "yaml", "json_schema"
 ]
-ToolUsageType = Union[BaseModel, Dict[str, Any]]
+ToolUsageType = Union[BaseModel, dict[str, Any]]
 
 
 # [FEAT][AGENT]
@@ -243,10 +240,10 @@ class Agent:
         agent_description: Optional[str] = None,
         system_prompt: Optional[str] = AGENT_SYSTEM_PROMPT_3,
         # TODO: Change to callable, then parse the callable to a string
-        tools: List[Callable] = None,
+        tools: Optional[list[Callable]] = None,
         dynamic_temperature_enabled: Optional[bool] = False,
         sop: Optional[str] = None,
-        sop_list: Optional[List[str]] = None,
+        sop_list: Optional[list[str]] = None,
         saved_state_path: Optional[str] = None,
         autosave: Optional[bool] = False,
         context_length: Optional[int] = 8192,
@@ -262,14 +259,14 @@ class Agent:
         traceback: Optional[Any] = None,
         traceback_handlers: Optional[Any] = None,
         streaming_on: Optional[bool] = False,
-        docs: List[str] = None,
+        docs: Optional[list[str]] = None,
         docs_folder: Optional[str] = None,
         verbose: Optional[bool] = False,
         parser: Optional[Callable] = None,
         best_of_n: Optional[int] = None,
         callback: Optional[Callable] = None,
-        metadata: Optional[Dict[str, Any]] = None,
-        callbacks: Optional[List[Callable]] = None,
+        metadata: Optional[dict[str, Any]] = None,
+        callbacks: Optional[list[Callable]] = None,
         search_algorithm: Optional[Callable] = None,
         logs_to_filename: Optional[str] = None,
         evaluator: Optional[Callable] = None,  # Custom LLM or agent
@@ -288,20 +285,20 @@ class Agent:
         function_calling_type: str = "json",
         output_cleaner: Optional[Callable] = None,
         function_calling_format_type: Optional[str] = "OpenAI",
-        list_base_models: Optional[List[BaseModel]] = None,
+        list_base_models: Optional[list[BaseModel]] = None,
         metadata_output_type: str = "json",
         state_save_file_type: str = "json",
         chain_of_thoughts: bool = False,
         algorithm_of_thoughts: bool = False,
         tree_of_thoughts: bool = False,
         tool_choice: str = "auto",
-        rules: str = None,  # type: ignore
+        rules: Optional[str] = None,  # type: ignore
         planning: Optional[str] = False,
         planning_prompt: Optional[str] = None,
-        custom_planning_prompt: str = None,
+        custom_planning_prompt: Optional[str] = None,
         memory_chunk_size: int = 2000,
         agent_ops_on: bool = False,
-        log_directory: str = None,
+        log_directory: Optional[str] = None,
         tool_system_prompt: str = tool_sop_prompt(),
         max_tokens: int = 4096,
         frequency_penalty: float = 0.0,
@@ -312,9 +309,9 @@ class Agent:
         # short_memory: Optional[str] = None,
         created_at: float = time.time(),
         return_step_meta: Optional[bool] = False,
-        tags: Optional[List[str]] = None,
-        use_cases: Optional[List[Dict[str, str]]] = None,
-        step_pool: List[Step] = [],
+        tags: Optional[list[str]] = None,
+        use_cases: Optional[list[dict[str, str]]] = None,
+        step_pool: list[Step] = [],
         print_every_step: Optional[bool] = False,
         time_created: Optional[str] = time.strftime(
             "%Y-%m-%d %H:%M:%S", time.localtime()
@@ -322,22 +319,22 @@ class Agent:
         agent_output: ManySteps = None,
         executor_workers: int = os.cpu_count(),
         data_memory: Optional[Callable] = None,
-        load_yaml_path: str = None,
+        load_yaml_path: Optional[str] = None,
         auto_generate_prompt: bool = False,
         rag_every_loop: bool = False,
         plan_enabled: bool = False,
         artifacts_on: bool = False,
-        artifacts_output_path: str = None,
-        artifacts_file_extension: str = None,
+        artifacts_output_path: Optional[str] = None,
+        artifacts_file_extension: Optional[str] = None,
         device: str = "cpu",
         all_cores: bool = True,
         device_id: int = 0,
         scheduled_run_date: Optional[datetime] = None,
         do_not_use_cluster_ops: bool = True,
         all_gpus: bool = False,
-        model_name: str = None,
-        llm_args: dict = None,
-        load_state_path: str = None,
+        model_name: Optional[str] = None,
+        llm_args: Optional[dict] = None,
+        load_state_path: Optional[str] = None,
         *args,
         **kwargs,
     ):
@@ -591,7 +588,7 @@ class Agent:
 
         return llm
 
-    def check_if_no_prompt_then_autogenerate(self, task: str = None):
+    def check_if_no_prompt_then_autogenerate(self, task: Optional[str] = None):
         """
         Checks if auto_generate_prompt is enabled and generates a prompt by combining agent name, description and system prompt if available.
         Falls back to task if all other fields are missing.
@@ -1294,7 +1291,7 @@ class Agent:
                 f"Error running agent: {error} while running concurrently"
             )
 
-    def run_concurrent_tasks(self, tasks: List[str], *args, **kwargs):
+    def run_concurrent_tasks(self, tasks: list[str], *args, **kwargs):
         """
         Run multiple tasks concurrently.
 
@@ -1315,7 +1312,7 @@ class Agent:
         except Exception as error:
             logger.error(f"Error running concurrent tasks: {error}")
 
-    def bulk_run(self, inputs: List[Dict[str, Any]]) -> List[str]:
+    def bulk_run(self, inputs: list[dict[str, Any]]) -> list[str]:
         """
         Generate responses for multiple input sets.
 
@@ -1336,7 +1333,7 @@ class Agent:
 
     async def arun_batched(
         self,
-        tasks: List[str],
+        tasks: list[str],
         *args,
         **kwargs,
     ):
@@ -1354,7 +1351,7 @@ class Agent:
             logger.error(f"Error running batched tasks: {error}")
             raise
 
-    def save(self, file_path: str = None) -> None:
+    def save(self, file_path: Optional[str] = None) -> None:
         """
         Save the agent state to a file using SafeStateManager with atomic writing
         and backup functionality. Automatically handles complex objects and class instances.
@@ -1523,7 +1520,7 @@ class Agent:
         except Exception as e:
             logger.error(f"Error during cleanup: {e}")
 
-    def load(self, file_path: str = None) -> None:
+    def load(self, file_path: Optional[str] = None) -> None:
         """
         Load agent state from a file using SafeStateManager.
         Automatically preserves class instances and complex objects.
@@ -1654,7 +1651,7 @@ class Agent:
         except Exception as e:
             logger.error(f"Error logging state info: {e}")
 
-    def get_saveable_state(self) -> Dict[str, Any]:
+    def get_saveable_state(self) -> dict[str, Any]:
         """
         Get a dictionary of all saveable state values.
         Useful for debugging or manual state inspection.
@@ -1664,7 +1661,7 @@ class Agent:
         """
         return SafeLoaderUtils.create_state_dict(self)
 
-    def get_preserved_instances(self) -> Dict[str, Any]:
+    def get_preserved_instances(self) -> dict[str, Any]:
         """
         Get a dictionary of all preserved class instances.
         Useful for debugging or manual state inspection.
@@ -1689,7 +1686,7 @@ class Agent:
                 feedback_counts[feedback] = 1
         print(f"Feedback counts: {feedback_counts}")
 
-    def undo_last(self) -> Tuple[str, str]:
+    def undo_last(self) -> tuple[str, str]:
         """
         Response the last response and return the previous state
 
@@ -1790,7 +1787,7 @@ class Agent:
         """Reset the agent"""
         self.short_memory = None
 
-    def ingest_docs(self, docs: List[str], *args, **kwargs):
+    def ingest_docs(self, docs: list[str], *args, **kwargs):
         """Ingest the docs into the memory
 
         Args:
@@ -1857,7 +1854,7 @@ class Agent:
         logger.info(f"Adding tool: {tool.__name__}")
         return self.tools.append(tool)
 
-    def add_tools(self, tools: List[Callable]):
+    def add_tools(self, tools: list[Callable]):
         """Add multiple tools to the agent's tools list.
 
         Args:
@@ -1881,7 +1878,7 @@ class Agent:
         logger.info(f"Removing tool: {tool.__name__}")
         return self.tools.remove(tool)
 
-    def remove_tools(self, tools: List[Callable]):
+    def remove_tools(self, tools: list[Callable]):
         """Remove multiple tools from the agent's tools list.
 
         Args:
@@ -1930,7 +1927,7 @@ class Agent:
                     "Could not import agentops, try installing agentops: $ pip3 install agentops"
                 )
 
-    def memory_query(self, task: str = None, *args, **kwargs) -> None:
+    def memory_query(self, task: Optional[str] = None, *args, **kwargs) -> None:
         try:
             # Query the long term memory
             if self.long_term_memory is not None:
@@ -1941,7 +1938,7 @@ class Agent:
                 )
 
                 memory_retrieval = (
-                    f"Documents Available: {str(memory_retrieval)}"
+                    f"Documents Available: {memory_retrieval!s}"
                 )
 
                 # # Count the tokens
@@ -1964,7 +1961,7 @@ class Agent:
             logger.error(f"An error occurred: {e}")
             raise e
 
-    def sentiment_analysis_handler(self, response: str = None):
+    def sentiment_analysis_handler(self, response: Optional[str] = None):
         """
         Performs sentiment analysis on the given response and stores the result in the short-term memory.
 
@@ -2264,7 +2261,7 @@ class Agent:
 
     def _serialize_callable(
         self, attr_value: Callable
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         Serializes callable attributes by extracting their name and docstring.
 
@@ -2307,7 +2304,7 @@ class Agent:
         except (TypeError, ValueError):
             return f"<Non-serializable: {type(attr_value).__name__}>"
 
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self) -> dict[str, Any]:
         """
         Converts all attributes of the class, including callables, into a dictionary.
         Handles non-serializable attributes by converting them or skipping them.
@@ -2562,15 +2559,15 @@ class Agent:
 
         except ValueError as e:
             logger.error(
-                f"Invalid input values for artifact: {str(e)}"
+                f"Invalid input values for artifact: {e!s}"
             )
             raise
-        except IOError as e:
-            logger.error(f"Error saving artifact to file: {str(e)}")
+        except OSError as e:
+            logger.error(f"Error saving artifact to file: {e!s}")
             raise
         except Exception as e:
             logger.error(
-                f"Unexpected error handling artifact: {str(e)}"
+                f"Unexpected error handling artifact: {e!s}"
             )
             raise
 

swarms/structs/agent_memory_manager.py

@@ -3,7 +3,7 @@ import logging
 import time
 import uuid
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Optional
 
 import yaml
 from pydantic import BaseModel
@@ -64,8 +64,8 @@ class MemoryManager:
         self.long_term_memory = long_term_memory
 
         # Initialize memories
-        self.short_term_memory: List[MemoryEntry] = []
-        self.system_messages: List[MemoryEntry] = []
+        self.short_term_memory: list[MemoryEntry] = []
+        self.system_messages: list[MemoryEntry] = []
 
         # Memory statistics
         self.total_tokens_processed: int = 0
@@ -253,7 +253,7 @@ class MemoryManager:
             else ""
         )
 
-    def get_memory_stats(self) -> Dict[str, Any]:
+    def get_memory_stats(self) -> dict[str, Any]:
         """Get detailed memory statistics"""
         return {
             "short_term_messages": len(self.short_term_memory),
@@ -306,7 +306,7 @@ class MemoryManager:
     def load_memory_snapshot(self, file_path: str) -> None:
         """Load memory state from file"""
         try:
-            with open(file_path, "r") as f:
+            with open(file_path) as f:
                 if file_path.endswith(".yaml"):
                     data = yaml.safe_load(f)
                 else:
@@ -330,7 +330,7 @@ class MemoryManager:
 
     def search_memories(
         self, query: str, memory_type: str = "all"
-    ) -> List[MemoryEntry]:
+    ) -> list[MemoryEntry]:
         """
         Search through memories of specified type
 
@@ -390,7 +390,7 @@ class MemoryManager:
 
     def get_memory_by_timeframe(
         self, start_time: float, end_time: float
-    ) -> List[MemoryEntry]:
+    ) -> list[MemoryEntry]:
         """Get memories within a specific timeframe"""
         return [
             entry

swarms/structs/agent_registry.py

@@ -1,7 +1,7 @@
 import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from threading import Lock
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Callable, Optional
 
 from pydantic import BaseModel, Field, ValidationError
 
@@ -20,13 +20,13 @@ class AgentConfigSchema(BaseModel):
         time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()),
         description="Time when the agent was added to the registry.",
     )
-    config: Dict[Any, Any] = None
+    config: dict[Any, Any] = None
 
 
 class AgentRegistrySchema(BaseModel):
     name: str
     description: str
-    agents: List[AgentConfigSchema]
+    agents: list[AgentConfigSchema]
     time_registry_creatd: str = Field(
         time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()),
         description="Time when the registry was created.",
@@ -55,7 +55,7 @@ class AgentRegistry:
         self,
         name: str = "Agent Registry",
         description: str = "A registry for managing agents.",
-        agents: Optional[List[Agent]] = None,
+        agents: Optional[list[Agent]] = None,
         return_json: bool = True,
         auto_save: bool = False,
         *args,
@@ -75,7 +75,7 @@ class AgentRegistry:
         self.description = description
         self.return_json = return_json
         self.auto_save = auto_save
-        self.agents: Dict[str, Agent] = {}
+        self.agents: dict[str, Agent] = {}
         self.lock = Lock()
 
         # Initialize the agent registry
@@ -119,7 +119,7 @@ class AgentRegistry:
                 logger.error(f"Validation error: {e}")
                 raise
 
-    def add_many(self, agents: List[Agent]) -> None:
+    def add_many(self, agents: list[Agent]) -> None:
         """
         Adds multiple agents to the registry.
 
@@ -215,7 +215,7 @@ class AgentRegistry:
                 logger.error(f"Error: {e}")
                 raise
 
-    def list_agents(self) -> List[str]:
+    def list_agents(self) -> list[str]:
         """
         Lists all agent names in the registry.
 
@@ -231,7 +231,7 @@ class AgentRegistry:
             logger.error(f"Error: {e}")
             raise e
 
-    def return_all_agents(self) -> List[Agent]:
+    def return_all_agents(self) -> list[Agent]:
         """
         Returns all agents from the registry.
 
@@ -249,7 +249,7 @@ class AgentRegistry:
 
     def query(
         self, condition: Optional[Callable[[Agent], bool]] = None
-    ) -> List[Agent]:
+    ) -> list[Agent]:
         """
         Queries agents based on a condition.
 

swarms/structs/agent_router.py

@@ -1,14 +1,13 @@
-from typing import List, Optional
+from typing import Any, Callable, Optional, Union
 
 from tenacity import retry, stop_after_attempt, wait_exponential
-from typing import Union, Callable, Any
+
 from swarms import Agent
-from swarms.utils.loguru_logger import initialize_logger
-from swarms.utils.lazy_loader import lazy_import_decorator
 from swarms.utils.auto_download_check_packages import (
     auto_check_and_download_package,
 )
-
+from swarms.utils.lazy_loader import lazy_import_decorator
+from swarms.utils.loguru_logger import initialize_logger
 
 logger = initialize_logger(log_folder="agent_router")
 
@@ -49,7 +48,7 @@ class AgentRouter:
         self.collection = self.client.create_collection(
             collection_name
         )
-        self.agents: List[Agent] = []
+        self.agents: list[Agent] = []
 
     @retry(
         stop=stop_after_attempt(3),
@@ -78,12 +77,12 @@ class AgentRouter:
             )
         except Exception as e:
             logger.error(
-                f"Error adding agent {agent.name} to the vector database: {str(e)}"
+                f"Error adding agent {agent.name} to the vector database: {e!s}"
             )
             raise
 
     def add_agents(
-        self, agents: List[Union[Agent, Callable, Any]]
+        self, agents: list[Union[Agent, Callable, Any]]
     ) -> None:
         """
         Add multiple agents to the vector database.
@@ -177,7 +176,7 @@ class AgentRouter:
 
             return None
         except Exception as e:
-            logger.error(f"Error finding best agent: {str(e)}")
+            logger.error(f"Error finding best agent: {e!s}")
             raise
 
 

swarms/structs/agents_available.py

@@ -1,11 +1,12 @@
+from typing import Optional
+
 from swarms.structs.agent import Agent
-from typing import List
 
 
 def showcase_available_agents(
-    agents: List[Agent],
-    name: str = None,
-    description: str = None,
+    agents: list[Agent],
+    name: Optional[str] = None,
+    description: Optional[str] = None,
     format: str = "XML",
 ) -> str:
     """

swarms/structs/async_workflow.py

@@ -9,7 +9,7 @@ from dataclasses import asdict, dataclass
 from datetime import datetime
 from enum import Enum
 from logging.handlers import RotatingFileHandler
-from typing import Any, Dict, List, Optional
+from typing import Any, Optional
 
 from pydantic import BaseModel, Field
 
@@ -42,8 +42,8 @@ class WorkflowOutput(BaseModel):
     total_agents: int
     successful_tasks: int
     failed_tasks: int
-    agent_outputs: List[AgentOutput]
-    metadata: Dict[str, Any] = Field(default_factory=dict)
+    agent_outputs: list[AgentOutput]
+    metadata: dict[str, Any] = Field(default_factory=dict)
 
 
 class SpeakerRole(str, Enum):
@@ -59,7 +59,7 @@ class SpeakerMessage(BaseModel):
     content: Any
     timestamp: datetime
     agent_name: str
-    metadata: Dict[str, Any] = Field(default_factory=dict)
+    metadata: dict[str, Any] = Field(default_factory=dict)
 
 
 class GroupChatConfig(BaseModel):
@@ -76,7 +76,7 @@ class SharedMemoryItem:
     value: Any
     timestamp: datetime
     author: str
-    metadata: Dict[str, Any] = None
+    metadata: dict[str, Any] = None
 
 
 @dataclass
@@ -103,7 +103,7 @@ class SharedMemory:
         key: str,
         value: Any,
         author: str,
-        metadata: Dict[str, Any] = None,
+        metadata: Optional[dict[str, Any]] = None,
     ) -> None:
         with self._lock:
             item = SharedMemoryItem(
@@ -138,7 +138,7 @@ class SharedMemory:
         if self._persistence_path and os.path.exists(
             self._persistence_path
         ):
-            with open(self._persistence_path, "r") as f:
+            with open(self._persistence_path) as f:
                 data = json.load(f)
                 self._memory = {
                     k: SharedMemoryItem(**v) for k, v in data.items()
@@ -149,8 +149,8 @@ class SpeakerSystem:
     """Manages speaker interactions and group chat functionality"""
 
     def __init__(self, default_timeout: float = 30.0):
-        self.speakers: Dict[SpeakerRole, SpeakerConfig] = {}
-        self.message_history: List[SpeakerMessage] = []
+        self.speakers: dict[SpeakerRole, SpeakerConfig] = {}
+        self.message_history: list[SpeakerMessage] = []
         self.default_timeout = default_timeout
         self._lock = threading.Lock()
 
@@ -166,7 +166,7 @@ class SpeakerSystem:
         self,
         config: SpeakerConfig,
         input_data: Any,
-        context: Dict[str, Any] = None,
+        context: Optional[dict[str, Any]] = None,
     ) -> SpeakerMessage:
         try:
             result = await asyncio.wait_for(
@@ -204,7 +204,7 @@ class AsyncWorkflow(BaseWorkflow):
     def __init__(
         self,
         name: str = "AsyncWorkflow",
-        agents: List[Agent] = None,
+        agents: Optional[list[Agent]] = None,
         max_workers: int = 5,
         dashboard: bool = False,
         autosave: bool = False,
@@ -265,8 +265,8 @@ class AsyncWorkflow(BaseWorkflow):
             self.speaker_system.add_speaker(config)
 
     async def run_concurrent_speakers(
-        self, task: str, context: Dict[str, Any] = None
-    ) -> List[SpeakerMessage]:
+        self, task: str, context: Optional[dict[str, Any]] = None
+    ) -> list[SpeakerMessage]:
         """Run all concurrent speakers in parallel"""
         concurrent_tasks = [
             self.speaker_system._execute_speaker(
@@ -282,8 +282,8 @@ class AsyncWorkflow(BaseWorkflow):
         return [r for r in results if isinstance(r, SpeakerMessage)]
 
     async def run_sequential_speakers(
-        self, task: str, context: Dict[str, Any] = None
-    ) -> List[SpeakerMessage]:
+        self, task: str, context: Optional[dict[str, Any]] = None
+    ) -> list[SpeakerMessage]:
         """Run non-concurrent speakers in sequence"""
         results = []
         for config in sorted(
@@ -298,15 +298,15 @@ class AsyncWorkflow(BaseWorkflow):
         return results
 
     async def run_group_chat(
-        self, initial_message: str, context: Dict[str, Any] = None
-    ) -> List[SpeakerMessage]:
+        self, initial_message: str, context: Optional[dict[str, Any]] = None
+    ) -> list[SpeakerMessage]:
         """Run a group chat discussion among speakers"""
         if not self.enable_group_chat:
             raise ValueError(
                 "Group chat is not enabled for this workflow"
             )
 
-        messages: List[SpeakerMessage] = []
+        messages: list[SpeakerMessage] = []
         current_turn = 0
 
         while current_turn < self.group_chat_config.max_loops:
@@ -349,7 +349,7 @@ class AsyncWorkflow(BaseWorkflow):
         return messages
 
     def _should_end_group_chat(
-        self, messages: List[SpeakerMessage]
+        self, messages: list[SpeakerMessage]
     ) -> bool:
         """Determine if group chat should end based on messages"""
         if not messages:
@@ -412,7 +412,7 @@ class AsyncWorkflow(BaseWorkflow):
         except Exception as e:
             end_time = datetime.utcnow()
             self.logger.error(
-                f"Error in agent {agent.agent_name} task {task_id}: {str(e)}",
+                f"Error in agent {agent.agent_name} task {task_id}: {e!s}",
                 exc_info=True,
             )
 
@@ -511,7 +511,7 @@ class AsyncWorkflow(BaseWorkflow):
 
             except Exception as e:
                 self.logger.error(
-                    f"Critical workflow error: {str(e)}",
+                    f"Critical workflow error: {e!s}",
                     exc_info=True,
                 )
                 raise
@@ -561,7 +561,7 @@ class AsyncWorkflow(BaseWorkflow):
             self.logger.info(f"Workflow results saved to {filename}")
         except Exception as e:
             self.logger.error(
-                f"Error saving workflow results: {str(e)}"
+                f"Error saving workflow results: {e!s}"
             )
 
     def _validate_config(self) -> None:
@@ -607,13 +607,13 @@ class AsyncWorkflow(BaseWorkflow):
             self.shared_memory._memory.clear()
 
         except Exception as e:
-            self.logger.error(f"Error during cleanup: {str(e)}")
+            self.logger.error(f"Error during cleanup: {e!s}")
             raise
 
 
 # Utility functions for the workflow
 def create_default_workflow(
-    agents: List[Agent],
+    agents: list[Agent],
     name: str = "DefaultWorkflow",
     enable_group_chat: bool = False,
 ) -> AsyncWorkflow:
@@ -651,7 +651,7 @@ async def run_workflow_with_retry(
             if attempt == max_retries - 1:
                 raise
             workflow.logger.warning(
-                f"Attempt {attempt + 1} failed, retrying in {retry_delay} seconds: {str(e)}"
+                f"Attempt {attempt + 1} failed, retrying in {retry_delay} seconds: {e!s}"
             )
             await asyncio.sleep(retry_delay)
             retry_delay *= 2  # Exponential backoff

swarms/structs/auto_swarm.py

@@ -1,4 +1,5 @@
-from typing import Any, Callable, Dict, Optional, Sequence
+from collections.abc import Sequence
+from typing import Any, Callable, Optional
 
 from swarms.structs.base_swarm import BaseSwarm
 from swarms.utils.loguru_logger import logger
@@ -33,8 +34,8 @@ class AutoSwarmRouter(BaseSwarm):
         name: Optional[str] = None,
         description: Optional[str] = None,
         verbose: bool = False,
-        custom_params: Optional[Dict[str, Any]] = None,
-        swarms: Sequence[BaseSwarm] = None,
+        custom_params: Optional[dict[str, Any]] = None,
+        swarms: Optional[Sequence[BaseSwarm]] = None,
         custom_preprocess: Optional[Callable] = None,
         custom_postprocess: Optional[Callable] = None,
         custom_router: Optional[Callable] = None,
@@ -60,7 +61,7 @@ class AutoSwarmRouter(BaseSwarm):
             f"AutoSwarmRouter has been initialized with {self.len_of_swarms()} swarms."
         )
 
-    def run(self, task: str = None, *args, **kwargs):
+    def run(self, task: Optional[str] = None, *args, **kwargs):
         try:
             """Run the swarm simulation and route the task to the appropriate swarm."""
 
@@ -137,7 +138,7 @@ class AutoSwarm(BaseSwarm):
         name: Optional[str] = None,
         description: Optional[str] = None,
         verbose: bool = False,
-        custom_params: Optional[Dict[str, Any]] = None,
+        custom_params: Optional[dict[str, Any]] = None,
         custom_preprocess: Optional[Callable] = None,
         custom_postprocess: Optional[Callable] = None,
         custom_router: Optional[Callable] = None,
@@ -179,7 +180,7 @@ class AutoSwarm(BaseSwarm):
 
     # def name_swarm_check(self, name: str = None):
 
-    def run(self, task: str = None, *args, **kwargs):
+    def run(self, task: Optional[str] = None, *args, **kwargs):
         """Run the swarm simulation."""
         try:
             loop = 0

swarms/structs/auto_swarm_builder.py

@@ -1,13 +1,13 @@
 import os
-from typing import List
+from typing import Optional
 
 from pydantic import BaseModel, Field
-from swarm_models import OpenAIFunctionCaller, OpenAIChat
+from swarm_models import OpenAIChat, OpenAIFunctionCaller
 
 from swarms.structs.agent import Agent
+from swarms.structs.agents_available import showcase_available_agents
 from swarms.structs.swarm_router import SwarmRouter
 from swarms.utils.loguru_logger import initialize_logger
-from swarms.structs.agents_available import showcase_available_agents
 
 logger = initialize_logger(log_folder="auto_swarm_builder")
 
@@ -43,7 +43,7 @@ class SwarmConfig(BaseModel):
         description="The description of the swarm's purpose and capabilities",
         example="A swarm of agents that work together to research topics and write articles",
     )
-    agents: List[AgentConfig] = Field(
+    agents: list[AgentConfig] = Field(
         description="The list of agents that make up the swarm",
         example=[
             AgentConfig(
@@ -127,8 +127,8 @@ class AutoSwarmBuilder:
 
     def __init__(
         self,
-        name: str = None,
-        description: str = None,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
         verbose: bool = True,
         max_loops: int = 1,
     ):
@@ -142,7 +142,7 @@ class AutoSwarmBuilder:
         )
 
     # @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
-    def run(self, task: str, image_url: str = None, *args, **kwargs):
+    def run(self, task: str, image_url: Optional[str] = None, *args, **kwargs):
         """Run the swarm on a given task.
 
         Args:
@@ -269,9 +269,9 @@ class AutoSwarmBuilder:
     # @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
     def swarm_router(
         self,
-        agents: List[Agent],
+        agents: list[Agent],
         task: str,
-        image_url: str = None,
+        image_url: Optional[str] = None,
         *args,
         **kwargs,
     ):

swarms/structs/base_structure.py

@@ -1,15 +1,14 @@
-import toml
-import yaml
 import asyncio
 import concurrent.futures
 import json
 import os
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Callable
-
+from typing import Any, Callable, Optional
 
 import psutil
+import toml
+import yaml
 
 try:
     import gzip
@@ -111,7 +110,7 @@ class BaseStructure:
         with open(file_path) as file:
             return json.load(file)
 
-    def save_metadata(self, metadata: Dict[str, Any]):
+    def save_metadata(self, metadata: dict[str, Any]):
         """Save metadata to file.
 
         Args:
@@ -123,7 +122,7 @@ class BaseStructure:
             )
             self.save_to_file(metadata, file_path)
 
-    def load_metadata(self) -> Dict[str, Any]:
+    def load_metadata(self) -> dict[str, Any]:
         """Load metadata from file.
 
         Returns:
@@ -206,7 +205,7 @@ class BaseStructure:
             None, self.run, *args, **kwargs
         )
 
-    async def save_metadata_async(self, metadata: Dict[str, Any]):
+    async def save_metadata_async(self, metadata: dict[str, Any]):
         """Save metadata to file asynchronously.
 
         Args:
@@ -217,7 +216,7 @@ class BaseStructure:
             None, self.save_metadata, metadata
         )
 
-    async def load_metadata_async(self) -> Dict[str, Any]:
+    async def load_metadata_async(self) -> dict[str, Any]:
         """Load metadata from file asynchronously.
 
         Returns:
@@ -316,7 +315,7 @@ class BaseStructure:
         with concurrent.futures.ThreadPoolExecutor() as executor:
             return executor.submit(self.run, *args, **kwargs)
 
-    def save_metadata_in_thread(self, metadata: Dict[str, Any]):
+    def save_metadata_in_thread(self, metadata: dict[str, Any]):
         """Save metadata to file in a thread.
 
         Args:
@@ -356,7 +355,7 @@ class BaseStructure:
 
     def run_batched(
         self,
-        batched_data: List[Any],
+        batched_data: list[Any],
         batch_size: int = 10,
         *args,
         **kwargs,
@@ -378,8 +377,8 @@ class BaseStructure:
             return [future.result() for future in futures]
 
     def load_config(
-        self, config: str = None, *args, **kwargs
-    ) -> Dict[str, Any]:
+        self, config: Optional[str] = None, *args, **kwargs
+    ) -> dict[str, Any]:
         """Load config from file.
 
         Args:
@@ -391,7 +390,7 @@ class BaseStructure:
         return self.load_from_file(config)
 
     def backup_data(
-        self, data: Any, backup_path: str = None, *args, **kwargs
+        self, data: Any, backup_path: Optional[str] = None, *args, **kwargs
     ):
         """Backup data to file.
 
@@ -418,7 +417,7 @@ class BaseStructure:
 
     def run_with_resources_batched(
         self,
-        batched_data: List[Any],
+        batched_data: list[Any],
         batch_size: int = 10,
         *args,
         **kwargs,
@@ -439,7 +438,7 @@ class BaseStructure:
 
     def _serialize_callable(
         self, attr_value: Callable
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         Serializes callable attributes by extracting their name and docstring.
 
@@ -482,7 +481,7 @@ class BaseStructure:
         except (TypeError, ValueError):
             return f"<Non-serializable: {type(attr_value).__name__}>"
 
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self) -> dict[str, Any]:
         """
         Converts all attributes of the class, including callables, into a dictionary.
         Handles non-serializable attributes by converting them or skipping them.

swarms/structs/base_swarm.py

@@ -1,31 +1,29 @@
-import os
 import asyncio
 import json
+import os
 import uuid
-from swarms.utils.file_processing import create_file_in_folder
 from abc import ABC
+from collections.abc import Sequence
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import (
     Any,
     Callable,
-    Dict,
-    List,
     Optional,
-    Sequence,
 )
 
 import yaml
+from pydantic import BaseModel
 
 from swarms.structs.agent import Agent
 from swarms.structs.conversation import Conversation
 from swarms.structs.omni_agent_types import AgentType
-from pydantic import BaseModel
+from swarms.utils.file_processing import create_file_in_folder
+from swarms.utils.loguru_logger import initialize_logger
 from swarms.utils.pandas_utils import (
     dict_to_dataframe,
     display_agents_info,
     pydantic_model_to_dataframe,
 )
-from swarms.utils.loguru_logger import initialize_logger
 
 logger = initialize_logger(log_folder="base_swarm")
 
@@ -81,8 +79,8 @@ class BaseSwarm(ABC):
         self,
         name: Optional[str] = None,
         description: Optional[str] = None,
-        agents: Optional[List[Agent]] = None,
-        models: Optional[List[Any]] = None,
+        agents: Optional[list[Agent]] = None,
+        models: Optional[list[Any]] = None,
         max_loops: Optional[int] = 200,
         callbacks: Optional[Sequence[callable]] = None,
         autosave: Optional[bool] = False,
@@ -93,7 +91,7 @@ class BaseSwarm(ABC):
         ] = "multiagent_structure_metadata.json",
         stopping_function: Optional[Callable] = None,
         stopping_condition: Optional[str] = "stop",
-        stopping_condition_args: Optional[Dict] = None,
+        stopping_condition_args: Optional[dict] = None,
         agentops_on: Optional[bool] = False,
         speaker_selection_func: Optional[Callable] = None,
         rules: Optional[str] = None,
@@ -230,7 +228,7 @@ class BaseSwarm(ABC):
         """Add a agent to the swarm"""
         self.agents.append(agent)
 
-    def add_agents(self, agents: List[AgentType]):
+    def add_agents(self, agents: list[AgentType]):
         """Add a list of agents to the swarm"""
         self.agents.extend(agents)
 
@@ -315,22 +313,22 @@ class BaseSwarm(ABC):
     ):
         """Send a direct message to a agent"""
 
-    def autoscaler(self, num_agents: int, agent: List[AgentType]):
+    def autoscaler(self, num_agents: int, agent: list[AgentType]):
         """Autoscaler that acts like kubernetes for autonomous agents"""
 
     def get_agent_by_id(self, id: str) -> AgentType:
         """Locate a agent by id"""
 
-    def assign_task(self, agent: AgentType, task: Any) -> Dict:
+    def assign_task(self, agent: AgentType, task: Any) -> dict:
         """Assign a task to a agent"""
 
     def get_all_tasks(self, agent: AgentType, task: Any):
         """Get all tasks"""
 
-    def get_finished_tasks(self) -> List[Dict]:
+    def get_finished_tasks(self) -> list[dict]:
         """Get all finished tasks"""
 
-    def get_pending_tasks(self) -> List[Dict]:
+    def get_pending_tasks(self) -> list[dict]:
         """Get all pending tasks"""
 
     def pause_agent(self, agent: AgentType, agent_id: str):
@@ -354,21 +352,21 @@ class BaseSwarm(ABC):
     def scale_to(self, num_agent: int):
         """Scale to a specific number of agents"""
 
-    def get_all_agents(self) -> List[AgentType]:
+    def get_all_agents(self) -> list[AgentType]:
         """Get all agents"""
 
     def get_swarm_size(self) -> int:
         """Get the size of the swarm"""
 
     # #@abstractmethod
-    def get_swarm_status(self) -> Dict:
+    def get_swarm_status(self) -> dict:
         """Get the status of the swarm"""
 
     # #@abstractmethod
     def save_swarm_state(self):
         """Save the swarm state"""
 
-    def batched_run(self, tasks: List[Any], *args, **kwargs):
+    def batched_run(self, tasks: list[Any], *args, **kwargs):
         """_summary_
 
         Args:
@@ -377,7 +375,7 @@ class BaseSwarm(ABC):
         # Implement batched run
         return [self.run(task, *args, **kwargs) for task in tasks]
 
-    async def abatch_run(self, tasks: List[str], *args, **kwargs):
+    async def abatch_run(self, tasks: list[str], *args, **kwargs):
         """Asynchronous batch run with language model
 
         Args:
@@ -447,7 +445,7 @@ class BaseSwarm(ABC):
         )
         return result
 
-    def run_batch_async(self, tasks: List[str], *args, **kwargs):
+    def run_batch_async(self, tasks: list[str], *args, **kwargs):
         """Run the swarm asynchronously
 
         Args:
@@ -459,7 +457,7 @@ class BaseSwarm(ABC):
         )
         return result
 
-    def run_batch(self, tasks: List[str], *args, **kwargs):
+    def run_batch(self, tasks: list[str], *args, **kwargs):
         """Run the swarm asynchronously
 
         Args:
@@ -496,7 +494,7 @@ class BaseSwarm(ABC):
         agent = self.select_agent_by_name(agent_name)
         return agent.run(task, *args, **kwargs)
 
-    def concurrent_run(self, task: str) -> List[str]:
+    def concurrent_run(self, task: str) -> list[str]:
         """Synchronously run the task on all llms and collect responses"""
         with ThreadPoolExecutor() as executor:
             future_to_llm = {
@@ -524,7 +522,7 @@ class BaseSwarm(ABC):
         """Remove an llm from the god mode"""
         self.agents.remove(agent)
 
-    def run_all(self, task: str = None, *args, **kwargs):
+    def run_all(self, task: Optional[str] = None, *args, **kwargs):
         """Run all agents
 
         Args:
@@ -538,7 +536,7 @@ class BaseSwarm(ABC):
             responses.append(agent(task, *args, **kwargs))
         return responses
 
-    def run_on_all_agents(self, task: str = None, *args, **kwargs):
+    def run_on_all_agents(self, task: Optional[str] = None, *args, **kwargs):
         """Run on all agents
 
         Args:
@@ -587,7 +585,7 @@ class BaseSwarm(ABC):
             SwarmManagerBase: Instance of SwarmManagerBase representing the retrieved Swarm, or None if not found.
         """
 
-    def retrieve_joined_agents(self, agent_id: str) -> List[Agent]:
+    def retrieve_joined_agents(self, agent_id: str) -> list[Agent]:
         """
         Retrieve the information the Agents which have joined the registry.
 

swarms/structs/base_workflow.py

@@ -1,10 +1,10 @@
 import json
-from typing import Any, Dict, List, Optional
+from typing import Any, Optional
 
-from swarms.utils.formatter import formatter
 from swarms.structs.agent import Agent
 from swarms.structs.base_structure import BaseStructure
 from swarms.structs.task import Task
+from swarms.utils.formatter import formatter
 from swarms.utils.loguru_logger import initialize_logger
 
 logger = initialize_logger("base-workflow")
@@ -42,9 +42,9 @@ class BaseWorkflow(BaseStructure):
 
     def __init__(
         self,
-        agents: List[Agent] = None,
-        task_pool: List[Task] = None,
-        models: List[Any] = None,
+        agents: Optional[list[Agent]] = None,
+        task_pool: Optional[list[Task]] = None,
+        models: Optional[list[Any]] = None,
         *args,
         **kwargs,
     ):
@@ -70,7 +70,7 @@ class BaseWorkflow(BaseStructure):
     def add_task(
         self,
         task: Task = None,
-        tasks: List[Task] = None,
+        tasks: Optional[list[Task]] = None,
         *args,
         **kwargs,
     ):
@@ -136,7 +136,7 @@ class BaseWorkflow(BaseStructure):
             )
             raise error
 
-    def get_task_results(self) -> Dict[str, Any]:
+    def get_task_results(self) -> dict[str, Any]:
         """
         Returns the results of each task in the workflow.
 
@@ -307,7 +307,7 @@ class BaseWorkflow(BaseStructure):
             raise error
 
     def load_workflow_state(
-        self, filepath: str = None, **kwargs
+        self, filepath: Optional[str] = None, **kwargs
     ) -> None:
         """
         Loads the workflow state from a json file and restores the workflow state.

swarms/structs/company.py

@@ -1,11 +1,10 @@
 from dataclasses import dataclass, field
-from typing import Dict, List, Optional, Union
+from typing import Optional, Union
 
 from swarms.structs.agent import Agent
 from swarms.structs.base_swarm import BaseSwarm
 from swarms.utils.loguru_logger import initialize_logger
 
-
 logger = initialize_logger("company-swarm")
 
 
@@ -15,11 +14,11 @@ class Company(BaseSwarm):
     Represents a company with a hierarchical organizational structure.
     """
 
-    org_chart: List[List[Agent]]
+    org_chart: list[list[Agent]]
     shared_instructions: str = None
     ceo: Optional[Agent] = None
-    agents: List[Agent] = field(default_factory=list)
-    agent_interactions: Dict[str, List[str]] = field(
+    agents: list[Agent] = field(default_factory=list)
+    agent_interactions: dict[str, list[str]] = field(
         default_factory=dict
     )
 
@@ -96,7 +95,7 @@ class Company(BaseSwarm):
             raise error
 
     def _parse_org_chart(
-        self, org_chart: Union[List[Agent], List[List[Agent]]]
+        self, org_chart: Union[list[Agent], list[list[Agent]]]
     ) -> None:
         """
         Parses the organization chart and adds agents to the company.

swarms/structs/concat.py

@@ -1,7 +1,6 @@
-from typing import List
 
 
-def concat_strings(string_list: List[str]) -> str:
+def concat_strings(string_list: list[str]) -> str:
     """
     Concatenates a list of strings into a single string.
 

swarms/structs/concurrent_workflow.py

@@ -1,25 +1,25 @@
 import asyncio
+import concurrent
 import os
 import uuid
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Optional, Union
 
+from clusterops import (
+    execute_on_gpu,
+    execute_on_multiple_gpus,
+    execute_with_cpu_cores,
+    list_available_gpus,
+)
 from pydantic import BaseModel, Field
 from tenacity import retry, stop_after_attempt, wait_exponential
 
 from swarms.structs.agent import Agent
 from swarms.structs.base_swarm import BaseSwarm
+from swarms.structs.swarm_id_generator import generate_swarm_id
 from swarms.utils.file_processing import create_file_in_folder
-import concurrent
-from clusterops import (
-    execute_on_gpu,
-    execute_with_cpu_cores,
-    execute_on_multiple_gpus,
-    list_available_gpus,
-)
 from swarms.utils.loguru_logger import initialize_logger
-from swarms.structs.swarm_id_generator import generate_swarm_id
 
 logger = initialize_logger(log_folder="concurrent_workflow")
 
@@ -60,7 +60,7 @@ class MetadataSchema(BaseModel):
         "Concurrent execution of multiple agents",
         description="Description of the workflow",
     )
-    agents: Optional[List[AgentOutputSchema]] = Field(
+    agents: Optional[list[AgentOutputSchema]] = Field(
         ..., description="List of agent outputs and metadata"
     )
     timestamp: Optional[datetime] = Field(
@@ -107,7 +107,7 @@ class ConcurrentWorkflow(BaseSwarm):
         self,
         name: str = "ConcurrentWorkflow",
         description: str = "Execution of multiple agents concurrently",
-        agents: List[Agent] = [],
+        agents: list[Agent] = [],
         metadata_output_path: str = "agent_metadata.json",
         auto_save: bool = True,
         output_schema: BaseModel = MetadataSchema,
@@ -115,7 +115,7 @@ class ConcurrentWorkflow(BaseSwarm):
         return_str_on: bool = False,
         agent_responses: list = [],
         auto_generate_prompts: bool = False,
-        max_workers: int = None,
+        max_workers: Optional[int] = None,
         *args,
         **kwargs,
     ):
@@ -324,7 +324,7 @@ class ConcurrentWorkflow(BaseSwarm):
 
     def _run(
         self, task: str, img: str, *args, **kwargs
-    ) -> Union[Dict[str, Any], str]:
+    ) -> Union[dict[str, Any], str]:
         """
         Runs the workflow for the given task, executes agents concurrently, and saves metadata in a production-grade manner.
 
@@ -442,8 +442,8 @@ class ConcurrentWorkflow(BaseSwarm):
             raise e
 
     def run_batched(
-        self, tasks: List[str]
-    ) -> List[Union[Dict[str, Any], str]]:
+        self, tasks: list[str]
+    ) -> list[Union[dict[str, Any], str]]:
         """
         Runs the workflow for a batch of tasks, executes agents concurrently for each task, and saves metadata in a production-grade manner.
 
@@ -486,8 +486,8 @@ class ConcurrentWorkflow(BaseSwarm):
         return asyncio.ensure_future(self.run(task))
 
     def run_batched_async(
-        self, tasks: List[str]
-    ) -> List[asyncio.Future]:
+        self, tasks: list[str]
+    ) -> list[asyncio.Future]:
         """
         Runs the workflow asynchronously for a batch of tasks, executes agents concurrently for each task, and saves metadata in a production-grade manner.
 
@@ -510,8 +510,8 @@ class ConcurrentWorkflow(BaseSwarm):
         return futures
 
     def run_parallel(
-        self, tasks: List[str]
-    ) -> List[Union[Dict[str, Any], str]]:
+        self, tasks: list[str]
+    ) -> list[Union[dict[str, Any], str]]:
         """
         Runs the workflow in parallel for a batch of tasks, executes agents concurrently for each task, and saves metadata in a production-grade manner.
 
@@ -540,8 +540,8 @@ class ConcurrentWorkflow(BaseSwarm):
         return results
 
     def run_parallel_async(
-        self, tasks: List[str]
-    ) -> List[asyncio.Future]:
+        self, tasks: list[str]
+    ) -> list[asyncio.Future]:
         """
         Runs the workflow in parallel asynchronously for a batch of tasks, executes agents concurrently for each task, and saves metadata in a production-grade manner.
 

swarms/structs/conversation.py

@@ -1,16 +1,16 @@
 import datetime
 import json
-from typing import Any, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 import yaml
+
 from swarms.structs.base_structure import BaseStructure
-from typing import TYPE_CHECKING
 from swarms.utils.formatter import formatter
 
 if TYPE_CHECKING:
     from swarms.structs.agent import (
         Agent,
-    )  # Only imported during type checking
+    )
 
 
 class Conversation(BaseStructure):
@@ -69,11 +69,11 @@ class Conversation(BaseStructure):
         system_prompt: Optional[str] = None,
         time_enabled: bool = False,
         autosave: bool = False,
-        save_filepath: str = None,
+        save_filepath: Optional[str] = None,
         tokenizer: Any = None,
         context_length: int = 8192,
-        rules: str = None,
-        custom_rules_prompt: str = None,
+        rules: Optional[str] = None,
+        custom_rules_prompt: Optional[str] = None,
         user: str = "User:",
         auto_save: bool = True,
         save_as_yaml: bool = True,
@@ -244,7 +244,7 @@ class Conversation(BaseStructure):
     def get_str(self):
         return self.return_history_as_string()
 
-    def save_as_json(self, filename: str = None):
+    def save_as_json(self, filename: Optional[str] = None):
         """Save the conversation history as a JSON file
 
         Args: