Swarm Strategies

[cobycloud]

discussion regarding swarm (homogenous) strategies

[3rdsonx]

Here is my proposed structure for the swarm

1. Inside swarmauri_core, let's have a swarm folder that we will have our Iswarm.py and Iagent.py

• The Iswarm.py will contain the Base Swarm Interface
• The Iagent.py will contain the base Agent Interface/Abstract Class

---

• The ISwarm contains a LIST of Iagent instances
• This allows for more flexible agent management
• Uses generic type Iagent to ensure type consistency
• Validates that only Iagent subclasses can be used
• Provides methods to dynamically add/remove agents

---

2. Now in the swarmauri we can have a

• basic_swamrm.py for homogenous tasks and more simple tasks
• advanced_swamrm.py for heterogenous tasks

Ths will allow the swarm to

• Be customized by overriding methods
• Support for different processing and aggregation strategies

We can also have some mixins like

• CommunicationMixin: Adds messaging capabilities
• LearningMixin: Provides knowledge base and learning method
• MonitoringMixin: Tracks performance metrics (etc)

[3rdsonx]

from abc import ABC, abstractmethod
from typing import List, Any, Dict, Optional
from uuid import uuid4

class Iagent(ABC):
    """
    Abstract base class for all agents in the SwarmAuri ecosystem.
    Provides core functionality and interface for agent behavior.
    """
    def __init__(
        self, 
        agent_id: Optional[str] = None, 
        config: Optional[Dict[str, Any]] = None
    ):
        # Unique identifier for the agent
        self.id = agent_id or str(uuid4())
        
        # Configuration for the agent
        self.config = config or {}
        
        # Core agent state
        self._state = {}
        
        # Initialization hook
        self.initialize()
    
    @abstractmethod
    def initialize(self):
        """
        Hook for agent-specific initialization.
        Each concrete agent implements its own initialization logic.
        """
        pass
    
    @abstractmethod
    def process(self, input_data: Any) -> Any:
        """
        Core processing method that defines agent's primary behavior.
        
        :param input_data: Input data to be processed
        :return: Processed output
        """
        pass
    
    def update_state(self, key: str, value: Any):
        """
        Update agent's internal state.
        
        :param key: State key
        :param value: State value
        """
        self._state[key] = value
    
    def get_state(self, key: str, default: Any = None):
        """
        Retrieve agent's state.
        
        :param key: State key
        :param default: Default value if key doesn't exist
        :return: State value
        """
        return self._state.get(key, default)

class ISwarm:
    """
    Base class for implementing homogeneous agent swarms.
    Manages a collection of identical agents with coordinated behavior.
    """
    def __init__(
        self, 
        agent_class: type[Iagent], 
        num_agents: int,
        swarm_config: Optional[Dict[str, Any]] = None
    ):
        # Configuration for the entire swarm
        self.config = swarm_config or {}
        
        # Create homogeneous agents
        self.agents: List[Iagent] = [
            agent_class(config=self.config) for _ in range(num_agents)
        ]
    
    def distribute_task(self, task_data: Any):
        """
        Distribute a task across all agents in the swarm.
        
        :param task_data: Data to be processed
        :return: Aggregated results from all agents
        """
        results = []
        for agent in self.agents:
            result = agent.process(task_data)
            results.append(result)
        return results
    
    def aggregate_results(self, results: List[Any]):
        """
        Aggregate results from distributed processing.
        
        :param results: List of results from agents
        :return: Aggregated/consolidated result
        """
        # Default implementation: simple averaging
        return sum(results) / len(results) if results else None

[cobycloud]

@3rd-Son

Now in the swarmauri we can have a
basic_swamrm.py for homogenous tasks and more simple tasks
advanced_swamrm.py for heterogenous tasks

^ Swarms are composed of homogenous agents, meaning all agents in the swarm are instances of the same kind. They all have the same methods and capabilities as every other agent in the set.

We do not distinguish between "homogeneous tasks" and "heterogenous tasks". Swarms are distinguished from MAFs based on whether or not the agents themselves are homogenous or heterogenous. We shall only refer to whether or not a set of agents is homogenous (ie: a swarm) or heterogenous (ie: a maf), we will not refer to the tasks themselves as homo/hetero.

If a goal requires a single task, this is likely a swarm (homogenous agents) or a pipeline (special purpose sequence). If a goal requires multiple tasks, this means that either every single agent in the swarm can perform any and all of the tasks OR this is a MAF, in which each agent would likely correspond to a task.

[3rdsonx]

@cobycloud
Thanks for the clarification.

I understand that the distinction is based on whether the agents are homogeneous (a swarm) or heterogeneous (a MAF), not the tasks themselves.

For single-task goals, a swarm or a pipeline works depending on the setup. For multiple tasks, it’s either a swarm (if all agents can do all tasks) or a MAF (if agents specialize in specific tasks).

This makes the framework agent-focused, not task-focused.

Am I correct?

[cobycloud]

Correct!

[3rdsonx]

Thanks for the clarifications

[MichaelDecent]

For the Swarm Agent Interface i propose the name ISwarmAgent.py in the folder swarmauri_core/swarm_agent.py with this class

ISwarmAgent(ABC):

   @abstractmethod
   async def process(self, task: Any) -> Any:
       """
       Core processing method that each agent implements.
       """
       pass 

[MichaelDecent]

am looking at an async implementation. as it will be handling multiple agent simultaneously.

[3rdsonx]

Yes, very important

[MichaelDecent]

for the Base Class Swarm Agent

class AgentStatus(Enum):
        IDLE = auto()
        WORKING = auto()
        COMPLETED = auto()
        FAILED = auto()

class SwarmBaseAgent(ISwarmAgent, ComponentBase):
    	"""
    	Abstract base class for homogeneous swarm agents.
    	"""
        status: AgentStatus = AgentStatus.IDLE
        message_queue: asyncio.Queue = asyncio.Queue()
	results: List[Any] = []
	resource: ResourceTypes =  Field(default=ResourceTypes.SWARM_BASE_AGENT.value)
	type: Literal['SwarmBaseAgent'] = 'SwarmBaseAgent'


    	async def send_message(self, message: Message, swarm: 'AgentSwarm'):
        	"""
        	Send a message to another agent in the swarm.
        	"""
        	await swarm.broadcast(message)

    	async def receive_message(self, message: Message):
        	"""
        	Handle incoming messages.
        	"""
        	await self.message_queue.put(message)

[MichaelDecent]

yeah. i totally agree with leveraging our Conversation class for messages and results, also the use of the message brokers (celery and RabbitMQ) makes a lot of sense.

[MichaelDecent]

We will utilize Agents and Models in our Swarms, however I agree that creating a unique class to introduce additional attributes and methods may be necessary.

In particular, the ability to share status and send/receive messages both seem like viable additions.

Regarding message queue and results - I imagine that we would be able to leverage an agent's conversation to determine the results of the conversation.

In terms of the message queue itself, typically message brokers handle the queue itself, then workers (such as a our swarm agents) approach the message broker when they are free and they ask for the next task. For example, with celery and rabbitmq.

For a completely built-in solution, I anticipated that we have a Swarm that has a queue and SwarmAgent that only work on one task at a time, though there may be scenarios where a SwarmAgent takes a slice of an index or several items from a queue and works on the batch of tasks before requesting more tasks.

In my opinion, we'll want to target the simple use case first (ie: one task), but perhaps this can be done by leveraging queue max size:

queue = asyncio.Queue(maxsize=10)  # Creates a queue with a maximum size of 10

On another note, we might also want to consider the potential uses of LifoQueue and PriorityQueue.

Am looking at extending the AgentSwarm i created below to inherit from the existing AgentBase, then rewrite AgentSwarm to utilize llms models, Conversation to handle message , and other necessary implementation the LifoQueue and PriorityQueue you stated.

[MichaelDecent]

We will utilize Agents and Models in our Swarms, however I agree that creating a unique class to introduce additional attributes and methods may be necessary.
In particular, the ability to share status and send/receive messages both seem like viable additions.
Regarding message queue and results - I imagine that we would be able to leverage an agent's conversation to determine the results of the conversation.
In terms of the message queue itself, typically message brokers handle the queue itself, then workers (such as a our swarm agents) approach the message broker when they are free and they ask for the next task. For example, with celery and rabbitmq.
For a completely built-in solution, I anticipated that we have a Swarm that has a queue and SwarmAgent that only work on one task at a time, though there may be scenarios where a SwarmAgent takes a slice of an index or several items from a queue and works on the batch of tasks before requesting more tasks.
In my opinion, we'll want to target the simple use case first (ie: one task), but perhaps this can be done by leveraging queue max size:

queue = asyncio.Queue(maxsize=10)  # Creates a queue with a maximum size of 10

On another note, we might also want to consider the potential uses of LifoQueue and PriorityQueue.

Am looking at extending the AgentSwarm i created below to inherit from the existing AgentBase, then rewrite AgentSwarm to utilize llms models, Conversation to handle message , and other necessary implementation the LifoQueue and PriorityQueue you stated.

from abc import ABC, abstractmethod
import asyncio
from enum import Enum, auto
import queue
from typing import Any, Optional, Type, Union, Dict, List, Literal
from pydantic import Field, ConfigDict, model_validator, root_validator
from swarmauri_core.ComponentBase import ComponentBase, ResourceTypes
from swarmauri_core.messages import IMessage
from swarmauri.llms.base.LLMBase import LLMBase
from swarmauri.agents.base.AgentBase import AgentBase
from swarmauri.agents.base.AgentConversationMixin import AgentConversationMixin
from swarmauri.messages.concrete import HumanMessage, AgentMessage
from swarmauri.conversations.base.ConversationBase import ConversationBase
from swarmauri_core.typing import SubclassUnion


class ISwarmAgent(ABC):

    @abstractmethod
    async def process(self, task: Any) -> Any:
        """
        Core processing method that each agent implements.
        """
        pass


class AgentStatus(Enum):
    IDLE = auto()
    WORKING = auto()
    COMPLETED = auto()
    FAILED = auto()


class SwarmBaseAgent(ISwarmAgent, ComponentBase):
    """
    Abstract base class for homogeneous swarm agents.
    """

    llm: SubclassUnion[LLMBase]
    conversation: SubclassUnion[ConversationBase]
    status: AgentStatus = AgentStatus.IDLE
    resource: ResourceTypes = Field(default=ResourceTypes.SWARM_BASE_AGENT.value)
    type: Literal["SwarmBaseAgent"] = "SwarmBaseAgent"

    async def process(self, task: Any) -> Any:
        raise NotImplementedError("Subclasses must implement the `process` method.")


class AgentSwarm(AgentConversationMixin, AgentBase):
    """
    Manages a swarm of agents, handling distributed tasks and conversation management.
    """

    llm: SubclassUnion[LLMBase]
    conversation: SubclassUnion[ConversationBase]
    agent_class: SubclassUnion[SwarmBaseAgent]

    num_agents: int = Field(
        default=5, gt=0, le=100, description="Number of agents in the swarm"
    )
    agent_timeout: float = Field(
        default=1.0, gt=0, description="Timeout for each agent"
    )
    retry_attempts: int = Field(
        default=3, ge=0, description="Number of retry attempts per agent"
    )
    agents: List[SwarmBaseAgent] = Field(default_factory=list)
    type: Literal["AgentSwarm"] = "AgentSwarm"
    queue_type: Literal["Queue", "LifoQueue", "PriorityQueue"] = "Queue"
    max_size: int = 10

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._task_queue = self._initialize_task_queue()
        self._initialize_agents()

    def _initialize_task_queue(self):
        if self.queue_type == "Queue":
            return asyncio.Queue(maxsize=self.max_size)
        elif self.queue_type == "LifoQueue":
            return queue.LifoQueue(maxsize=self.max_size)
        elif self.queue_type == "PriorityQueue":
            return queue.PriorityQueue(maxsize=self.max_size)
        else:
            raise ValueError(f"Unsupported queue type: {self.queue_type}")

    def _initialize_agents(self):
        for _ in range(self.num_agents):
            agent = self.agent_class(llm=self.llm, conversation=self.conversation)
            self.agents.append(agent)

    async def _distribute_tasks(self, conversation: ConversationBase):
        for message in conversation.history:
            await self._task_queue.put(message)

    async def _agent_worker(
        self, agent: SwarmBaseAgent, llm_kwargs: Optional[Dict] = {}
    ):
        try:
            task = await asyncio.wait_for(
                self._task_queue.get(), timeout=self.agent_timeout
            )
            agent.status = AgentStatus.WORKING
            result = await agent.process(
                task, llm_kwargs
            )  # Ensure `process` is asynchronous
            agent_message = AgentMessage(content=result)
            self.conversation.add_message(agent_message)
            agent.status = AgentStatus.COMPLETED
        except asyncio.TimeoutError:
            agent.status = AgentStatus.IDLE
        except Exception as e:
            agent.status = AgentStatus.FAILED
            print(f"Error in agent worker: {e}")

    def get_swarm_status(self) -> Dict[int, AgentStatus]:
        """
        Get the current status of all agents in the swarm.
        """
        return {i: agent.status for i, agent in enumerate(self.agents)}

    async def exec(
        self,
        input_data: Optional[Union[List[str], IMessage]] = [],
        llm_kwargs: Optional[Dict] = {},
    ) -> Any:
        if isinstance(input_data, list):
            for data in input_data:
                if not isinstance(data, str):
                    raise TypeError("All items in the input data list must be strings.")
                human_message = HumanMessage(content=data)
                self.conversation.add_message(human_message)
        elif isinstance(input_data, IMessage):
            self.conversation.add_message(input_data)
        else:
            raise TypeError(
                "Input data must be a list of strings or an instance of IMessage."
            )

        await self._distribute_tasks(self.conversation)

        agent_tasks = [self._agent_worker(agent, llm_kwargs) for agent in self.agents]
        await asyncio.gather(*agent_tasks)
        return [agent.conversation.get_last().content for agent in self.agents]

[MichaelDecent]

Here is a simple use case with a custom swarm agent implementation of text summarization swarm

class TextSummaryAgent(SwarmBaseAgent):
    async def process(self, task: str, llm_kwargs: dict = {}) -> str:
        """
        Process a chunk of text and generate a summary
        """
        # Simulate LLM-based summarization
        prompt = f"Provide a concise summary of the following text: {task}"
        summary = await self.llm.apredict(prompt, **llm_kwargs)
        return summary

# Example use case
async def main(document: str, chunk_size: int = 1000) -> List[str]:
    """
    Break a long document into chunks and use a swarm of agents to summarize
    """
    long_document = """
    This is a very long document about artificial intelligence... 
    (imagine a multi-page text about AI technologies, 
    machine learning, and computational approaches)
    """
    chunk_size = 1000

    # Split document into chunks
    chunks = [document[i:i+chunk_size] for i in range(0, len(long_document), chunk_size)]
    
    # Configure swarm
    swarm = AgentSwarm(
        agent_class=TextSummaryAgent,
        llm=OpenAIModel(api_key=os.get_env(API_KEY)),
        conversation=Conversation(),
        num_agents=3,  # Use 3 agents to process chunks in parallel
        agent_timeout=30.0  # 30-second timeout per agent
    )
    
    # Execute swarm summarization
    summaries = await swarm.exec(input_data=chunks)

if __name__ == "__main__":
    asyncio.run(main())

[MichaelDecent]

@cobycloud @3rd-Son please review, your feedback is highly needed

[MichaelDecent]

AgentSwarm class manages the homogenous swarm agent

class AgentStatus(Enum):
    IDLE = auto()
    WORKING = auto()
    COMPLETED = auto()
    FAILED = auto()

class AgentSwarm(BaseModel):
    """
    Manages a homogeneous swarm of agents with Pydantic configuration.
    """

    num_agents: int = Field(default=5, gt=0, le=100, description="Number of agents in the swarm")
    agent_timeout: float = Field(default=1.0, gt=0, description="Timeout for each agent")
    retry_attempts: int = Field(default=3, ge=0, description="Number of retry attempts per agent")
    agent_class: Type[SwarmBaseAgent] = Field(default=SwarmBaseAgent, description="Class of the agents")
    task_queue: asyncio.Queue = asyncio.Queue()
    results_queue: asyncio.Queue = asyncio.Queue()
    agents: List[SwarmBaseAgent] = []

    @root_validator(pre=True)
    def initialize_agents(cls, values):
        num_agents = values.get("num_agents", 5)
        agent_class = values.get("agent_class", SwarmBaseAgent)
        values["agents"] = [agent_class() for _ in range(num_agents)]
        return values

    async def broadcast(self, message: Message):
        """
        Broadcast a message to all agents.
        """
        for agent in self.agents:
            await agent.receive_message(message)

    async def distribute_task(self, task: Any):
        """
        Distribute a task to the task queue.
        """
        await self.task_queue.put(task)

    async def run(self, tasks: List[Any]):
        """
        Run the swarm on a list of tasks.
        """
        # Distribute tasks to agents
        for task in tasks:
            await self.distribute_task(task)

        # Create tasks for agent processing
        agent_tasks = [asyncio.create_task(self._agent_worker(agent)) for agent in self.agents]

        # Wait for all agent tasks to complete
        await asyncio.gather(*agent_tasks)
       
        results = []
        while not self.results_queue.empty():
                results.append(await self.results_queue.get())
       return results

    async def _agent_worker(self, agent: SwarmBaseAgent):
        """
        Worker method for individual agents with retry logic.
        """
        retry_count = 0
        while retry_count < self.retry_attempts:
            try:
                # Try to get a task with configurable timeout
                task = await asyncio.wait_for(self.task_queue.get(), timeout=self.agent_timeout)

                try:
                    agent.status = AgentStatus.WORKING
                    result = await agent.process(task)
                    agent.results.append(result)
                    await self.results_queue.put(result)
                    agent.status = AgentStatus.COMPLETED
                    break  # Successfully processed task
                except Exception as e:
                    agent.status = AgentStatus.FAILED
                    retry_count += 1
                    print(f"Error processing task: {e}")

                self.task_queue.task_done()

            except asyncio.TimeoutError:
                # No more tasks or timeout occurred
                print(f"Agent {agent.type} timed out.")
                break```

[MichaelDecent]

this is a Stateless agent design with async implementation

[MichaelDecent]

use case

class TextProcessingAgent(BaseAgent):
    """
    Example implementation of a text processing agent.
    """
    async def process(self, text: str) -> str:
        """
        Process text by converting to uppercase.
        """
        await asyncio.sleep(0.1)  # Simulate processing time
        return text.upper()

# Usage Example
async def main():
    swarm = AgentSwarm(num_agents=3, agent_class=TextProcessingAgent)

    # Sample tasks
    tasks = [
        "hello world", 
        "python is awesome", 
        "agent swarms are cool"
    ]

    # Run the swarm
    results = await swarm.run(tasks)
    print("Processed Results:", results)

[MichaelDecent]

Homogeneous Agents: Intra-System Communication Patterns

1. Broadcasting

• Description: A single message is sent to all agents within the system. Each agent decides individually whether the message is relevant and what action to take based on its context.
• Key Features:
  • All agents receive the same message simultaneously.
  • No guarantee of message acknowledgment or processing.
    Example Use Case: In a swarm of robots, a leader robot broadcasts a message about an obstacle. All robots adjust their path as needed.

2. Point-to-Point Messaging

• Description: Messages are sent directly between specific agents. Communication occurs in a one-to-one manner, ensuring targeted information exchange.
• Key Features:
  • Explicit sender and receiver.
  • Suitable for private or task-specific communication.
  • Often more efficient than broadcasting in systems with high agent density.
    Example Use Case: In a distributed computing system, a task scheduler assigns a job to a specific worker node by sending a direct message.

3. Peer-to-Peer Communication

• Description: Agents communicate symmetrically without any central authority or hierarchy. Each agent can act as both a sender and receiver of information.
• Key Features:
  • Decentralized: No master-agent or central control.
  • Agents dynamically establish connections based on the task or need.
  • Scalable for large systems as it avoids bottlenecks at a central node.
    Example Use Case: In a peer-to-peer file-sharing network, agents share files directly without relying on a central server.

[MichaelDecent]

Homogeneous Agents: Inter-System Communication Patterns

1. Federated Communication

• Description: Independent systems collaborate by sharing data and resources while maintaining autonomy. A common governance protocol ensures uniformity.
• Key Features:
  • Systems remain loosely coupled and independent.
  • Communication occurs through standard protocols.
    Example Use Case: Federated learning in AI, where multiple devices train a model locally and share updates without sharing raw data.

2. Replication

• Description: Data or state is duplicated across systems to ensure consistency and fault tolerance.
• Key Features:
  • Ensures all systems have synchronized data.
  • Commonly used for fault tolerance and high availability.
    Example Use Case: In distributed databases, replicas of the same data are maintained across nodes for consistency and availability.

3. Synchronous Communication

• Description: Systems operate in lock-step communication cycles. Agents exchange messages at defined intervals or based on predefined triggers.
• Key Features:
  • Guarantees coordinated operation across systems.
  • Requires tight timing synchronization.

Example Use Case: In a parallel computing setup, worker nodes synchronize after completing each phase of computation.

[MichaelDecent]

Common communication patterns and the mechanisms that can implement them, categorized for both homogeneous agents

1. Request-Response Pattern

   • Description: One agent requests information or service, and another responds.
   • Mechanisms:
     • REST APIs: Platform-agnostic HTTP-based mechanism.
     • gRPC: High-performance RPC framework for efficient binary communication.
     • SOAP (Simple Object Access Protocol): XML-based messaging for stricter type enforcement.

2. Publish-Subscribe Pattern

   • Description: Messages are published to a topic; subscribers to the topic receive relevant messages.
   • Mechanisms:
     • Message Brokers: Kafka, RabbitMQ, or ActiveMQ provide robust pub-sub messaging.
     • Event Streaming Platforms: Apache Pulsar and Amazon Kinesis for real-time data distribution.
     • WebSockets: Enables real-time communication for low-latency scenarios.

3. Broadcast Pattern

   • Description: A single agent sends a message to all agents in a group.
   • Mechanisms:
     • UDP Multicast: Efficient for broadcasting messages to multiple agents in a local network.
     • Message Queues: Brokers like Redis Streams for reliable broadcast in larger systems.
     • Middleware Frameworks: JADE for multi-agent broadcast in heterogeneous systems.

4. Peer-to-Peer Pattern

   • Description: Direct communication between two agents without intermediaries.
   • Mechanisms:
     • P2P Protocols: WebRTC or Libp2p for decentralized, real-time communication.
     • Direct Socket Programming: Enables low-level TCP/UDP-based peer interaction.
     • Blockchain Networks: Distributed P2P ledger for transactional communication.

[MichaelDecent]

5. Federated Communication Pattern

• Description: Coordinated message exchange across agents in distinct but homogeneous systems.
• Mechanisms:
  • Federated APIs: REST or gRPC exposed by multiple homogeneous systems.
  • Distributed Coordination Tools: Apache Zookeeper for managing shared states and tasks.
  • Cross-Platform SDKs: Tools like Google Cloud Pub/Sub ensure synchronization across platforms.

6. Gateway Communication Pattern

• Description: Communication is routed through a mediator to bridge different systems or protocols.
• Mechanisms:
  • API Gateways: Tools like Kong and AWS API Gateway handle routing and protocol bridging.
  • Service Mesh: Istio or Linkerd for managing inter-system communications.
  • Middleware: FIPA-compliant systems for heterogeneous multi-agent communication.

7. Event-Driven Communication

• Description: Agents react to specific events, enabling asynchronous workflows.
• Mechanisms:
  • Event Brokers: Kafka and Amazon EventBridge for event-based communication.
  • Reactive Programming Frameworks: RxJS or Akka Streams for asynchronous event handling.
  • Webhook Services: Trigger external workflows upon events.