> ## Documentation Index
> Fetch the complete documentation index at: https://docs.swarms.world/llms.txt
> Use this file to discover all available pages before exploring further.

# BaseStructure

> Foundation class providing core functionality for swarm structures and agent systems

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

```python theme={null}
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

<ParamField path="name" type="str">
  Name of the structure for identification and file naming
</ParamField>

<ParamField path="description" type="str">
  Description of the structure's purpose
</ParamField>

<ParamField path="save_metadata_on" type="bool" default="true">
  Enable automatic metadata saving
</ParamField>

<ParamField path="save_artifact_path" type="str" default="./artifacts">
  Directory path for saving artifacts
</ParamField>

<ParamField path="save_metadata_path" type="str" default="./metadata">
  Directory path for saving metadata
</ParamField>

<ParamField path="save_error_path" type="str" default="./errors">
  Directory path for saving error logs
</ParamField>

## Core Methods

### run

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

```python theme={null}
def run(
    *args,
    **kwargs
) -> Any
```

<ResponseField name="return" type="Any">
  Implementation-specific return value
</ResponseField>

## File Operations

### save\_to\_file

Save data to a JSON file.

```python theme={null}
def save_to_file(
    data: Any,
    file_path: str
) -> None
```

<ParamField path="data" type="Any">
  Data to save (must be JSON serializable)
</ParamField>

<ParamField path="file_path" type="str">
  Path to save the file
</ParamField>

### load\_from\_file

Load data from a JSON file.

```python theme={null}
def load_from_file(
    file_path: str
) -> Any
```

<ParamField path="file_path" type="str">
  Path to the file to load
</ParamField>

<ResponseField name="return" type="Any">
  Loaded data from the JSON file
</ResponseField>

### asave\_to\_file

Asynchronously save data to a file.

```python theme={null}
async def asave_to_file(
    data: Any,
    file: str,
    *args,
    **kwargs
) -> None
```

### aload\_from\_file

Asynchronously load data from a file.

```python theme={null}
async def aload_from_file(
    file: str
) -> Any
```

## Metadata Management

### save\_metadata

Save metadata to a JSON file.

```python theme={null}
def save_metadata(
    metadata: Dict[str, Any]
) -> None
```

<ParamField path="metadata" type="Dict[str, Any]">
  Metadata dictionary to save
</ParamField>

### load\_metadata

Load metadata from a JSON file.

```python theme={null}
def load_metadata() -> Dict[str, Any]
```

<ResponseField name="return" type="Dict[str, Any]">
  Loaded metadata dictionary
</ResponseField>

### save\_metadata\_async

Asynchronously save metadata.

```python theme={null}
async def save_metadata_async(
    metadata: Dict[str, Any]
) -> None
```

### load\_metadata\_async

Asynchronously load metadata.

```python theme={null}
async def load_metadata_async() -> Dict[str, Any]
```

## Artifact Management

### save\_artifact

Save an artifact to a JSON file.

```python theme={null}
def save_artifact(
    artifact: Any,
    artifact_name: str
) -> None
```

<ParamField path="artifact" type="Any">
  Artifact data to save
</ParamField>

<ParamField path="artifact_name" type="str">
  Name for the artifact file (without extension)
</ParamField>

### load\_artifact

Load an artifact from a JSON file.

```python theme={null}
def load_artifact(
    artifact_name: str
) -> Any
```

<ParamField path="artifact_name" type="str">
  Name of the artifact to load
</ParamField>

<ResponseField name="return" type="Any">
  Loaded artifact data
</ResponseField>

### save\_artifact\_async

Asynchronously save an artifact.

```python theme={null}
async def save_artifact_async(
    artifact: Any,
    artifact_name: str
) -> None
```

### load\_artifact\_async

Asynchronously load an artifact.

```python theme={null}
async def load_artifact_async(
    artifact_name: str
) -> Any
```

## Error and Event Logging

### log\_error

Log an error message to file.

```python theme={null}
def log_error(
    error_message: str
) -> None
```

<ParamField path="error_message" type="str">
  Error message to log
</ParamField>

### log\_error\_async

Asynchronously log an error.

```python theme={null}
async def log_error_async(
    error_message: str
) -> None
```

### log\_event

Log an event with timestamp and type.

```python theme={null}
def log_event(
    event: str,
    event_type: str = "INFO"
) -> None
```

<ParamField path="event" type="str">
  Event description
</ParamField>

<ParamField path="event_type" type="str" default="INFO">
  Event type (INFO, WARNING, ERROR, etc.)
</ParamField>

### log\_event\_async

Asynchronously log an event.

```python theme={null}
async def log_event_async(
    event: str,
    event_type: str = "INFO"
) -> None
```

## Async Operations

### run\_async

Run the structure asynchronously.

```python theme={null}
async def run_async(
    *args,
    **kwargs
) -> Any
```

### run\_concurrent

Run the structure concurrently using asyncio.

```python theme={null}
def run_concurrent(
    *args,
    **kwargs
) -> Any
```

## Thread Operations

### run\_in\_thread

Run the structure in a separate thread.

```python theme={null}
def run_in_thread(
    *args,
    **kwargs
) -> concurrent.futures.Future
```

<ResponseField name="return" type="concurrent.futures.Future">
  Future object representing the thread execution
</ResponseField>

### save\_metadata\_in\_thread

Save metadata in a separate thread.

```python theme={null}
def save_metadata_in_thread(
    metadata: Dict[str, Any]
) -> concurrent.futures.Future
```

## Batch Processing

### run\_batched

Run multiple tasks in batches using ThreadPoolExecutor.

```python theme={null}
def run_batched(
    batched_data: List[Any],
    batch_size: int = 10,
    *args,
    **kwargs
) -> List[Any]
```

<ParamField path="batched_data" type="List[Any]">
  List of data items to process
</ParamField>

<ParamField path="batch_size" type="int" default="10">
  Number of concurrent workers
</ParamField>

<ResponseField name="return" type="List[Any]">
  List of results for each batch item
</ResponseField>

### run\_with\_resources\_batched

Run batched data with resource monitoring.

```python theme={null}
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.

```python theme={null}
def compress_data(
    data: Any
) -> bytes
```

<ParamField path="data" type="Any">
  Data to compress (must be JSON serializable)
</ParamField>

<ResponseField name="return" type="bytes">
  Compressed data as bytes
</ResponseField>

### decompres\_data

Decompress gzip data.

```python theme={null}
def decompres_data(
    data: bytes
) -> Any
```

<ParamField path="data" type="bytes">
  Compressed data to decompress
</ParamField>

<ResponseField name="return" type="Any">
  Decompressed data
</ResponseField>

## Resource Monitoring

### monitor\_resources

Monitor CPU and memory usage.

```python theme={null}
def monitor_resources() -> None
```

Logs current CPU and memory usage to the event log.

### run\_with\_resources

Run the structure with resource monitoring.

```python theme={null}
def run_with_resources(
    *args,
    **kwargs
) -> Any
```

## Configuration

### load\_config

Load configuration from a file.

```python theme={null}
def load_config(
    config: str = None,
    *args,
    **kwargs
) -> Dict[str, Any]
```

<ParamField path="config" type="str">
  Path to configuration file
</ParamField>

<ResponseField name="return" type="Dict[str, Any]">
  Configuration dictionary
</ResponseField>

### backup\_data

Backup data with timestamp.

```python theme={null}
def backup_data(
    data: Any,
    backup_path: str = None,
    *args,
    **kwargs
) -> None
```

<ParamField path="data" type="Any">
  Data to backup
</ParamField>

<ParamField path="backup_path" type="str">
  Path to backup directory
</ParamField>

## Serialization

### to\_dict

Convert structure to dictionary.

```python theme={null}
def to_dict() -> Dict[str, Any]
```

<ResponseField name="return" type="Dict[str, Any]">
  Dictionary representation of the structure
</ResponseField>

### to\_json

Convert structure to JSON string.

```python theme={null}
def to_json(
    indent: int = 4,
    *args,
    **kwargs
) -> str
```

<ParamField path="indent" type="int" default="4">
  Number of spaces for indentation
</ParamField>

<ResponseField name="return" type="str">
  JSON string representation
</ResponseField>

### to\_yaml

Convert structure to YAML string.

```python theme={null}
def to_yaml(
    indent: int = 4,
    *args,
    **kwargs
) -> str
```

<ParamField path="indent" type="int" default="4">
  Number of spaces for indentation
</ParamField>

<ResponseField name="return" type="str">
  YAML string representation
</ResponseField>

### to\_toml

Convert structure to TOML string.

```python theme={null}
def to_toml(
    *args,
    **kwargs
) -> str
```

<ResponseField name="return" type="str">
  TOML string representation
</ResponseField>

## Examples

### Basic Structure Implementation

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

1. **Always call super().**init**()**: When extending BaseStructure, call parent constructor
2. **Implement the run method**: Override the abstract run() method with your logic
3. **Use logging methods**: Leverage log\_event and log\_error for tracking
4. **Save artifacts**: Use save\_artifact for important outputs
5. **Handle errors gracefully**: Wrap operations in try/except and use log\_error
6. **Use async for I/O**: Leverage async methods for file operations
7. **Monitor resources**: Use run\_with\_resources for resource-intensive tasks
8. **Batch when possible**: Use run\_batched for processing multiple items
9. **Compress large data**: Use compress\_data for large datasets
10. **Enable metadata**: Set save\_metadata\_on=True for production systems

## Related

* [Agent](/api/agent) - Core agent implementation built on BaseStructure
* [BaseSwarm](/api/base-swarm) - Multi-agent swarm base class
* [State Management](/concepts/agents) - Managing agent and swarm state