BaseStructure - swarms

Overview

The BaseStructure class provides the foundational infrastructure for all swarm structures and agent systems in the Swarms framework. It offers essential utilities for state management, file I/O, metadata tracking, error logging, and resource monitoring.

Import

from swarms.structs import BaseStructure

Key Features

File Operations: Save and load data in JSON, YAML, and TOML formats
State Management: Automatic metadata and artifact tracking
Error Logging: Centralized error logging to files
Event Tracking: Log events with timestamps and severity levels
Async Support: Full async/await support for I/O operations
Thread Support: Run operations in separate threads
Batch Processing: Process multiple tasks concurrently
Data Compression: Built-in gzip compression/decompression
Resource Monitoring: Track CPU and memory usage
Serialization: Convert structures to dict/JSON/YAML/TOML

Initialization

str

Name of the structure for identification and file naming

str

Description of the structure’s purpose

bool

default:"true"

Enable automatic metadata saving

str

default:"./artifacts"

Directory path for saving artifacts

str

default:"./metadata"

Directory path for saving metadata

str

default:"./errors"

Directory path for saving error logs

Core Methods

run

Execute the structure’s main operation (abstract method to be implemented by subclasses).

def run(
    *args,
    **kwargs
) -> Any

Any

Implementation-specific return value

File Operations

save_to_file

Save data to a JSON file.

def save_to_file(
    data: Any,
    file_path: str
) -> None

Any

Data to save (must be JSON serializable)

str

Path to save the file

load_from_file

Load data from a JSON file.

def load_from_file(
    file_path: str
) -> Any

str

Path to the file to load

Any

Loaded data from the JSON file

asave_to_file

Asynchronously save data to a file.

async def asave_to_file(
    data: Any,
    file: str,
    *args,
    **kwargs
) -> None

aload_from_file

Asynchronously load data from a file.

async def aload_from_file(
    file: str
) -> Any

Metadata Management

save_metadata

Save metadata to a JSON file.

def save_metadata(
    metadata: Dict[str, Any]
) -> None

Dict[str, Any]

Metadata dictionary to save

load_metadata

Load metadata from a JSON file.

def load_metadata() -> Dict[str, Any]

Dict[str, Any]

Loaded metadata dictionary

save_metadata_async

Asynchronously save metadata.

async def save_metadata_async(
    metadata: Dict[str, Any]
) -> None

load_metadata_async

Asynchronously load metadata.

async def load_metadata_async() -> Dict[str, Any]

Artifact Management

save_artifact

Save an artifact to a JSON file.

def save_artifact(
    artifact: Any,
    artifact_name: str
) -> None

Any

Artifact data to save

str

Name for the artifact file (without extension)

load_artifact

Load an artifact from a JSON file.

def load_artifact(
    artifact_name: str
) -> Any

str

Name of the artifact to load

Any

Loaded artifact data

save_artifact_async

Asynchronously save an artifact.

async def save_artifact_async(
    artifact: Any,
    artifact_name: str
) -> None

load_artifact_async

Asynchronously load an artifact.

async def load_artifact_async(
    artifact_name: str
) -> Any

Error and Event Logging

log_error

Log an error message to file.

def log_error(
    error_message: str
) -> None

str

Error message to log

log_error_async

Asynchronously log an error.

async def log_error_async(
    error_message: str
) -> None

log_event

Log an event with timestamp and type.

def log_event(
    event: str,
    event_type: str = "INFO"
) -> None

str

Event description

str

default:"INFO"

Event type (INFO, WARNING, ERROR, etc.)

log_event_async

Asynchronously log an event.

async def log_event_async(
    event: str,
    event_type: str = "INFO"
) -> None

Async Operations

run_async

Run the structure asynchronously.

async def run_async(
    *args,
    **kwargs
) -> Any

run_concurrent

Run the structure concurrently using asyncio.

def run_concurrent(
    *args,
    **kwargs
) -> Any

Thread Operations

run_in_thread

Run the structure in a separate thread.

def run_in_thread(
    *args,
    **kwargs
) -> concurrent.futures.Future

concurrent.futures.Future

Future object representing the thread execution

save_metadata_in_thread

Save metadata in a separate thread.

def save_metadata_in_thread(
    metadata: Dict[str, Any]
) -> concurrent.futures.Future

Batch Processing

run_batched

Run multiple tasks in batches using ThreadPoolExecutor.

def run_batched(
    batched_data: List[Any],
    batch_size: int = 10,
    *args,
    **kwargs
) -> List[Any]

List[Any]

List of data items to process

int

default:"10"

Number of concurrent workers

List[Any]

List of results for each batch item

run_with_resources_batched

Run batched data with resource monitoring.

def run_with_resources_batched(
    batched_data: List[Any],
    batch_size: int = 10,
    *args,
    **kwargs
) -> List[Any]

Data Compression

compress_data

Compress data using gzip.

def compress_data(
    data: Any
) -> bytes

Any

Data to compress (must be JSON serializable)

bytes

Compressed data as bytes

decompres_data

Decompress gzip data.

def decompres_data(
    data: bytes
) -> Any

bytes

Compressed data to decompress

Any

Decompressed data

Resource Monitoring

monitor_resources

Monitor CPU and memory usage.

def monitor_resources() -> None

Logs current CPU and memory usage to the event log.

run_with_resources

Run the structure with resource monitoring.

def run_with_resources(
    *args,
    **kwargs
) -> Any

Configuration

load_config

Load configuration from a file.

def load_config(
    config: str = None,
    *args,
    **kwargs
) -> Dict[str, Any]

str

Path to configuration file

Dict[str, Any]

Configuration dictionary

backup_data

Backup data with timestamp.

def backup_data(
    data: Any,
    backup_path: str = None,
    *args,
    **kwargs
) -> None

Any

Data to backup

str

Path to backup directory

Serialization

to_dict

Convert structure to dictionary.

def to_dict() -> Dict[str, Any]

Dict[str, Any]

Dictionary representation of the structure

to_json

Convert structure to JSON string.

def to_json(
    indent: int = 4,
    *args,
    **kwargs
) -> str

int

default:"4"

Number of spaces for indentation

str

JSON string representation

to_yaml

Convert structure to YAML string.

def to_yaml(
    indent: int = 4,
    *args,
    **kwargs
) -> str

int

default:"4"

Number of spaces for indentation

str

YAML string representation

to_toml

Convert structure to TOML string.

def to_toml(
    *args,
    **kwargs
) -> str

str

TOML string representation

Examples

Basic Structure Implementation

from swarms.structs import BaseStructure

class MyStructure(BaseStructure):
    def __init__(self, name: str):
        super().__init__(
            name=name,
            description="Custom structure",
            save_metadata_on=True,
            save_artifact_path="./my_artifacts",
            save_metadata_path="./my_metadata"
        )
    
    def run(self, task: str):
        # Custom logic
        result = f"Processed: {task}"
        
        # Log event
        self.log_event(f"Executed task: {task}")
        
        # Save artifact
        self.save_artifact(result, "task_result")
        
        return result

# Use the structure
struct = MyStructure(name="processor")
result = struct.run("analyze data")

Async Structure

import asyncio
from swarms.structs import BaseStructure

class AsyncStructure(BaseStructure):
    async def run(self, task: str):
        # Async processing
        await asyncio.sleep(1)
        result = f"Async result: {task}"
        
        # Save async
        await self.save_artifact_async(result, "async_result")
        await self.log_event_async("Async task completed")
        
        return result

struct = AsyncStructure(name="async_processor")
result = await struct.run_async("process async")

Batch Processing Structure

class BatchStructure(BaseStructure):
    def run(self, data: str):
        # Process single item
        return f"Processed: {data}"

struct = BatchStructure(name="batch_processor")

# Process multiple items
data_items = ["item1", "item2", "item3", "item4", "item5"]
results = struct.run_batched(
    batched_data=data_items,
    batch_size=3  # Process 3 items concurrently
)

print(results)
# ['Processed: item1', 'Processed: item2', ...]

Structure with Resource Monitoring

class MonitoredStructure(BaseStructure):
    def run(self, task: str):
        # Heavy processing
        result = self.process_heavy_task(task)
        return result

struct = MonitoredStructure(name="monitored")

# Run with automatic resource monitoring
result = struct.run_with_resources("heavy task")
# Logs CPU and memory usage automatically

Structure with Error Handling

class RobustStructure(BaseStructure):
    def run(self, task: str):
        try:
            # Processing logic
            result = self.process(task)
            self.log_event("Task completed successfully")
            return result
        except Exception as e:
            # Log error
            self.log_error(f"Error processing task: {str(e)}")
            raise

struct = RobustStructure(name="robust")
try:
    result = struct.run("task")
except Exception as e:
    # Error is logged to ./errors/robust_errors.log
    pass

Data Compression Example

class CompressedStructure(BaseStructure):
    def run(self, large_data: dict):
        # Compress large data
        compressed = self.compress_data(large_data)
        
        # Save compressed data
        with open("compressed_data.gz", "wb") as f:
            f.write(compressed)
        
        # Later, decompress
        with open("compressed_data.gz", "rb") as f:
            compressed = f.read()
        
        decompressed = self.decompres_data(compressed)
        return decompressed

struct = CompressedStructure(name="compressor")
large_data = {"key": "value" * 1000}
result = struct.run(large_data)

Serialization Example

class ConfigStructure(BaseStructure):
    def __init__(self, name: str, setting: str):
        super().__init__(name=name)
        self.setting = setting

struct = ConfigStructure(
    name="config",
    setting="production"
)

# Export to different formats
json_str = struct.to_json()
yaml_str = struct.to_yaml()
toml_str = struct.to_toml()
dict_repr = struct.to_dict()

print(json_str)
# {
#   "name": "config",
#   "setting": "production",
#   ...
# }

Properties

BaseStructure provides the following properties:

workspace_dir: Path to the workspace directory (from environment variable)
name: Name of the structure
description: Description of the structure
save_metadata_on: Whether metadata saving is enabled
save_artifact_path: Path for artifacts
save_metadata_path: Path for metadata
save_error_path: Path for error logs

Best Practices

Always call super().init(): When extending BaseStructure, call parent constructor
Implement the run method: Override the abstract run() method with your logic
Use logging methods: Leverage log_event and log_error for tracking
Save artifacts: Use save_artifact for important outputs
Handle errors gracefully: Wrap operations in try/except and use log_error
Use async for I/O: Leverage async methods for file operations
Monitor resources: Use run_with_resources for resource-intensive tasks
Batch when possible: Use run_batched for processing multiple items
Compress large data: Use compress_data for large datasets
Enable metadata: Set save_metadata_on=True for production systems

Agent - Core agent implementation built on BaseStructure
BaseSwarm - Multi-agent swarm base class
State Management - Managing agent and swarm state

​Overview

​Import

​Key Features

​Initialization

​Core Methods

​run

​File Operations

​save_to_file

​load_from_file

​asave_to_file

​aload_from_file

​Metadata Management

​save_metadata

​load_metadata

​save_metadata_async

​load_metadata_async

​Artifact Management

​save_artifact

​load_artifact

​save_artifact_async

​load_artifact_async

​Error and Event Logging

​log_error

​log_error_async

​log_event

​log_event_async

​Async Operations

​run_async

​run_concurrent

​Thread Operations

​run_in_thread

​save_metadata_in_thread

​Batch Processing

​run_batched

​run_with_resources_batched

​Data Compression

​compress_data

​decompres_data

​Resource Monitoring

​monitor_resources

​run_with_resources

​Configuration

​load_config

​backup_data

​Serialization

​to_dict

​to_json

​to_yaml

​to_toml

​Examples

​Basic Structure Implementation

​Async Structure

​Batch Processing Structure

​Structure with Resource Monitoring

​Structure with Error Handling

​Data Compression Example

​Serialization Example

​Properties

​Best Practices

​Related

Overview

Import

Key Features

Initialization

Core Methods

run

File Operations

save_to_file

load_from_file

asave_to_file

aload_from_file

Metadata Management

save_metadata

load_metadata

save_metadata_async

load_metadata_async

Artifact Management

save_artifact

load_artifact

save_artifact_async

load_artifact_async

Error and Event Logging

log_error

log_error_async

log_event

log_event_async

Async Operations

run_async

run_concurrent

Thread Operations

run_in_thread

save_metadata_in_thread

Batch Processing

run_batched

run_with_resources_batched

Data Compression

compress_data

decompres_data

Resource Monitoring

monitor_resources

run_with_resources

Configuration

load_config

backup_data

Serialization

to_dict

to_json

to_yaml

to_toml

Examples

Basic Structure Implementation

Async Structure

Batch Processing Structure

Structure with Resource Monitoring

Structure with Error Handling

Data Compression Example

Serialization Example

Properties

Best Practices

Related