BaseSwarm - swarms

Overview

The BaseSwarm class is an abstract base class (ABC) that provides the foundation for all multi-agent systems in the Swarms framework. It defines the core interface and common functionality for orchestrating multiple agents to work together on complex tasks.

Import

from swarms.structs import BaseSwarm

Key Features

Agent Management: Add, remove, and query agents in the swarm
Task Distribution: Assign tasks to specific agents or broadcast to all
Lifecycle Management: Pause, resume, stop, and restart agents
Scaling: Dynamic agent scaling (scale up/down/to specific size)
State Management: Save and load swarm state
Batch Operations: Run tasks across multiple agents concurrently
Async Support: Full async/await support for concurrent execution
Conversation Management: Built-in conversation tracking

Initialization

str

Name of the swarm for identification and logging

str

Description of the swarm’s purpose and capabilities

List[Union[Agent, Callable]]

List of Agent instances or callables to include in the swarm

List[Any]

List of language models to use across agents

int

default:"200"

Maximum number of execution loops for the swarm

Sequence[callable]

Callback functions to execute during swarm operations

bool

default:"false"

Automatically save swarm state to file

bool

default:"false"

Enable detailed logging for debugging

bool

default:"false"

Include metadata in return values

str

default:"multiagent_structure_metadata.json"

Filename for saving swarm metadata

Callable

Function that determines when the swarm should stop execution

str

default:"stop"

Condition string that triggers swarm termination

Dict

Arguments passed to the stopping condition function

Callable

Function to select which agent should speak next in multi-agent conversations

str

Rules that govern swarm behavior and agent interactions

Any

default:"false"

Shared memory system accessible by all agents

bool

default:"false"

Enable AgentOps tracking for all agents in the swarm

BaseModel

Pydantic model defining the expected output structure

Core Methods

run

Execute the swarm’s main task loop. In BaseSwarm this is an unimplemented placeholder (def run(self): ..., no parameters) — subclasses are expected to override it. __call__ and the various *_run/*batch* helpers all invoke self.run(task, *args, **kwargs), so overrides should accept at least a task argument.

def run(
    task: str = None,
    *args,
    **kwargs
) -> Any

str

The task for the swarm to execute

Any

Result of swarm execution (implementation-specific)

call

Alternative syntax for running the swarm (calls run internally).

def __call__(
    task: str,
    *args,
    **kwargs
) -> Any

Agent Management

add_agent

Add a single agent to the swarm.

def add_agent(
    agent: AgentType
) -> None

AgentType

The agent instance to add

add_agents

Add multiple agents to the swarm.

def add_agents(
    agents: List[AgentType]
) -> None

List[AgentType]

List of agent instances to add

add_agent_by_id

Look up an agent by ID via get_agent_by_id and add it to the swarm.

def add_agent_by_id(
    agent_id: str
) -> None

remove_agent

Remove an agent from the swarm.

def remove_agent(
    agent: AgentType
) -> None

get_agent_by_name

Retrieve an agent by its name.

def get_agent_by_name(
    name: str
) -> AgentType

str

Name of the agent to retrieve

AgentType

The agent instance with matching name, or None if not found

get_agent_by_id

Retrieve an agent by its ID.

def get_agent_by_id(
    id: str
) -> AgentType

str

ID of the agent to retrieve

AgentType

The agent instance with matching ID, or None if not found

self_find_agent_by_name / self_find_agent_by_id

Find an agent within self.agents by name or ID. Thin wrappers around the module-level find_agent_by_name / find_agent_by_id helpers.

def self_find_agent_by_name(name: str) -> Agent
def self_find_agent_by_id(id: uuid.UUID) -> Agent

Task Management

assign_task

Assign a specific task to an agent.

def assign_task(
    agent: AgentType,
    task: Any
) -> Dict

AgentType

The agent to assign the task to

Any

The task to assign

Dict

Task execution result

task_assignment_by_name

Assign a task to an agent by name.

def task_assignment_by_name(
    task: str,
    agent_name: str,
    *args,
    **kwargs
) -> Any

task_assignment_by_id

Assign a task to an agent by ID (looked up via select_agent, not get_agent_by_id).

def task_assignment_by_id(
    task: str,
    agent_id: str,
    *args,
    **kwargs
) -> Any

get_all_tasks / get_finished_tasks / get_pending_tasks

Placeholder stubs for task-tracking (unimplemented in BaseSwarm).

def get_all_tasks(agent: AgentType, task: Any) -> None
def get_finished_tasks() -> List[Dict]
def get_pending_tasks() -> List[Dict]

broadcast

Broadcast a message to all agents in the swarm.

def broadcast(
    message: str,
    sender: Optional[AgentType] = None
) -> None

str

Message to broadcast

AgentType

Optional sender agent

direct_message

Send a direct message from one agent to another.

def direct_message(
    message: str,
    sender: AgentType,
    recipient: AgentType
) -> None

Batch Operations

batched_run

Run multiple tasks in batch mode.

def batched_run(
    tasks: List[Any],
    *args,
    **kwargs
) -> List[Any]

List[Any]

List of tasks to execute

List[Any]

List of results for each task

run_batch

Alias for batched_run.

def run_batch(
    tasks: List[str],
    *args,
    **kwargs
) -> List[Any]

concurrent_run

Run a task concurrently across all agents.

def concurrent_run(
    task: str
) -> List[str]

str

Task to run on all agents

List[str]

List of responses from each agent

concurrent_run appends to self.task_history, which BaseSwarm.__init__ does not initialize. Calling it on a fresh instance without something else having set self.task_history first will raise AttributeError.

run_all

Run a task on all agents sequentially.

def run_all(
    task: str = None,
    *args,
    **kwargs
) -> List[Any]

run_on_all_agents

Run a task on all agents using ThreadPoolExecutor.

def run_on_all_agents(
    task: str = None,
    *args,
    **kwargs
) -> List[Any]

Async Operations

arun

Run the swarm asynchronously.

async def arun(
    task: Optional[str] = None,
    *args,
    **kwargs
) -> Any

abatch_run

Run multiple tasks asynchronously in batch.

async def abatch_run(
    tasks: List[str],
    *args,
    **kwargs
) -> List[Any]

run_async

Run the swarm asynchronously (synchronous wrapper).

def run_async(
    task: Optional[str] = None,
    *args,
    **kwargs
) -> Any

run_batch_async

Run batch tasks asynchronously (synchronous wrapper).

def run_batch_async(
    tasks: List[str],
    *args,
    **kwargs
) -> Any

Lifecycle Management

pause_agent

Pause an agent’s execution.

def pause_agent(
    agent: AgentType,
    agent_id: str
) -> None

resume_agent

Resume a paused agent.

def resume_agent(
    agent: AgentType,
    agent_id: str
) -> None

stop_agent

Stop an agent’s execution.

def stop_agent(
    agent: AgentType,
    agent_id: str
) -> None

restart_agent

Restart an agent.

def restart_agent(
    agent: AgentType
) -> None

reset_all_agents

Reset the state of all agents.

def reset_all_agents() -> None

Scaling

scale_up, scale_down, and scale_to are unimplemented placeholders in BaseSwarm (docstring only, no body) — subclasses must override them to actually add/remove agents.

scale_up

Increase the number of agents.

def scale_up(
    num_agent: int
) -> None

int

Number of agents to add

scale_down

Decrease the number of agents.

def scale_down(
    num_agent: int
) -> None

int

Number of agents to remove

scale_to

Scale to a specific number of agents.

def scale_to(
    num_agent: int
) -> None

int

Target number of agents

get_all_agents / get_swarm_size / get_swarm_status / save_swarm_state

Placeholder stubs (unimplemented in BaseSwarm) intended for subclasses to override.

def get_all_agents() -> List[AgentType]
def get_swarm_size() -> int
def get_swarm_status() -> Dict
def save_swarm_state() -> None

add_llm / remove_llm

Add or remove a callable/agent from self.agents (aliases predating the add_agent/remove_agent naming).

def add_llm(agent: Callable) -> None
def remove_llm(agent: Callable) -> None

loop / aloop

Call run(task, *args, **kwargs) repeatedly, max_loops times. aloop runs loop in an executor.

def loop(task: Optional[str] = None, *args, **kwargs) -> None
async def aloop(task: Optional[str] = None, *args, **kwargs) -> Any

State Management

save_to_json

Save swarm state to JSON file.

def save_to_json(
    filename: str
) -> None

str

Path to save JSON file

load_from_json

Load swarm state from JSON file.

def load_from_json(
    filename: str
) -> None

str

Path to JSON file to load

save_to_yaml

Save swarm state to YAML file.

def save_to_yaml(
    filename: str
) -> None

load_from_yaml

Load swarm state from YAML file.

def load_from_yaml(
    filename: str
) -> None

metadata

Get swarm metadata.

def metadata() -> dict

dict

Dictionary containing swarm metadata (agents, callbacks, autosave, logging, conversation)

Examples

Basic Swarm Implementation

from swarms import Agent, BaseSwarm

class SimpleSwarm(BaseSwarm):
    def run(self, task: str, *args, **kwargs):
        """Run task on all agents and aggregate results."""
        results = []
        for agent in self.agents:
            result = agent.run(task)
            results.append(result)
        return results

# Create agents
agent1 = Agent(agent_name="Agent-1", model_name="claude-sonnet-4-6")
agent2 = Agent(agent_name="Agent-2", model_name="claude-sonnet-4-6")

# Create swarm
swarm = SimpleSwarm(
    name="Research-Swarm",
    description="A swarm for research tasks",
    agents=[agent1, agent2]
)

# Run task
results = swarm.run("Analyze market trends")

Dynamic Agent Management

from swarms import Agent, BaseSwarm

class DynamicSwarm(BaseSwarm):
    def run(self, task: str, *args, **kwargs):
        return self.run_all(task)

swarm = DynamicSwarm(
    name="Dynamic-Swarm",
    agents=[]
)

# Add agents dynamically
for i in range(5):
    agent = Agent(
        agent_name=f"Agent-{i}",
        model_name="claude-sonnet-4-6"
    )
    swarm.add_agent(agent)

# Check if agent exists (container protocol / get_agent_by_name)
if swarm.get_agent_by_name("Agent-0") is not None:
    print("Agent-0 is in the swarm")

# Get agent by name
agent = swarm.get_agent_by_name("Agent-2")

# Scale the swarm
swarm.scale_up(3)  # Add 3 more agents
swarm.scale_down(2)  # Remove 2 agents

Batch Task Processing

class BatchSwarm(BaseSwarm):
    def run(self, task: str, *args, **kwargs):
        # Implementation
        pass

swarm = BatchSwarm(
    name="Batch-Processor",
    agents=[agent1, agent2, agent3]
)

# Run multiple tasks
tasks = [
    "Task 1: Analyze data",
    "Task 2: Generate report",
    "Task 3: Send summary"
]

results = swarm.batched_run(tasks)

Async Swarm Execution

import asyncio

class AsyncSwarm(BaseSwarm):
    async def run(self, task: str, *args, **kwargs):
        # Async implementation
        pass

swarm = AsyncSwarm(
    name="Async-Swarm",
    agents=[agent1, agent2]
)

# Run asynchronously
result = await swarm.arun("Process async task")

# Run batch async
tasks = ["Task 1", "Task 2", "Task 3"]
results = await swarm.abatch_run(tasks)

Swarm with State Persistence

class PersistentSwarm(BaseSwarm):
    def run(self, task: str, *args, **kwargs):
        results = self.run_all(task)
        
        # Auto-save after execution
        if self.autosave:
            self.save_to_json(self.metadata_filename)
        
        return results

swarm = PersistentSwarm(
    name="Persistent-Swarm",
    agents=[agent1, agent2],
    autosave=True,
    metadata_filename="swarm_state.json"
)

result = swarm.run("Execute task")

# Load saved state later
swarm.load_from_json("swarm_state.json")

Properties

BaseSwarm provides several built-in properties:

agents_dict: Dictionary mapping agent names to agent instances
conversation: Conversation object for tracking agent interactions

Sequence/Container Protocol

BaseSwarm implements the standard container dunder methods over self.agents, so a swarm instance can be used like a list of agents:

len(swarm)              # __len__  -> number of agents
swarm[0]                # __getitem__
swarm[0] = new_agent    # __setitem__
del swarm[0]            # __delitem__
for agent in swarm: ... # __iter__
reversed(swarm)         # __reversed__
agent in swarm          # __contains__
repr(swarm) / str(swarm) # __repr__ / __str__

Registry & Miscellaneous Stubs

The following are unimplemented placeholder methods (docstring only, no body) intended for subclasses that maintain a swarm registry:

def add_swarm_entry(swarm) -> None
def add_agent_entry(agent: Agent) -> None
def retrieve_swarm_information(swarm_id: str)
def retrieve_joined_agents(agent_id: str) -> List[Agent]
def join_swarm(from_entity, to_entity: Agent)
def list_agents() -> None
def agents_to_dataframe() -> None
def agent_error_handling_check() -> None  # raises ValueError if self.agents is None

Best Practices

Always validate agents: Ensure agents list is not empty and all agents are valid
Implement abstract methods: Override run(), communicate(), and other abstract methods
Use async for I/O-bound tasks: Leverage async methods for better performance
Enable autosave for long-running swarms: Prevent state loss
Set appropriate max_loops: Prevent infinite loops
Use callbacks for monitoring: Track swarm execution progress
Implement proper error handling: Handle agent failures gracefully
Scale dynamically based on load: Use scaling methods to optimize resources

Agent - Core agent implementation
BaseStructure - Base class for structures
Swarm Architectures - Different swarm patterns

​Overview

​Import

​Key Features

​Initialization

​Core Methods

​run

​call

​Agent Management

​add_agent

​add_agents

​add_agent_by_id

​remove_agent

​get_agent_by_name

​get_agent_by_id

​self_find_agent_by_name / self_find_agent_by_id

​Task Management

​assign_task

​task_assignment_by_name

​task_assignment_by_id

​get_all_tasks / get_finished_tasks / get_pending_tasks

​broadcast

​direct_message

​Batch Operations

​batched_run

​run_batch

​concurrent_run

​run_all

​run_on_all_agents

​Async Operations

​arun

​abatch_run

​run_async

​run_batch_async

​Lifecycle Management

​pause_agent

​resume_agent

​stop_agent

​restart_agent

​reset_all_agents

​Scaling

​scale_up

​scale_down

​scale_to

​get_all_agents / get_swarm_size / get_swarm_status / save_swarm_state

​add_llm / remove_llm

​loop / aloop

​State Management

​save_to_json

​load_from_json

​save_to_yaml

​load_from_yaml

​metadata

​Examples

​Basic Swarm Implementation

​Dynamic Agent Management

​Batch Task Processing

​Async Swarm Execution

​Swarm with State Persistence

​Properties

​Sequence/Container Protocol

​Registry & Miscellaneous Stubs

​Best Practices

​Related

Overview

Import

Key Features

Initialization

Core Methods

run

call

Agent Management

add_agent

add_agents

add_agent_by_id

remove_agent

get_agent_by_name

get_agent_by_id

self_find_agent_by_name / self_find_agent_by_id

Task Management

assign_task

task_assignment_by_name

task_assignment_by_id

get_all_tasks / get_finished_tasks / get_pending_tasks

broadcast

direct_message

Batch Operations

batched_run

run_batch

concurrent_run

run_all

run_on_all_agents

Async Operations

arun

abatch_run

run_async

run_batch_async

Lifecycle Management

pause_agent

resume_agent

stop_agent

restart_agent

reset_all_agents

Scaling

scale_up

scale_down

scale_to

get_all_agents / get_swarm_size / get_swarm_status / save_swarm_state

add_llm / remove_llm

loop / aloop

State Management

save_to_json

load_from_json

save_to_yaml

load_from_yaml

metadata

Examples

Basic Swarm Implementation

Dynamic Agent Management

Batch Task Processing

Async Swarm Execution

Swarm with State Persistence

Properties

Sequence/Container Protocol

Registry & Miscellaneous Stubs

Best Practices

Related