@ -1,33 +1,51 @@
import math import math
from typing import , Union , Dict from typing import Dict, List, Union
from loguru import logger from loguru import logger
from swarms . structs . agent import Agent from swarms . structs . agent import Agent
from swarms . structs . omni_agent_types import AgentListType
from swarms . structs . conversation import Conversation from swarms . structs . conversation import Conversation
from swarms . structs . omni_agent_types import AgentListType
from swarms . utils . history_output_formatter import (
history_output_formatter ,
)
# Base Swarm class that all other swarm types will inherit from # Base Swarm class that all other swarm types will inherit from
class BaseSwarm : class BaseSwarm :
def __init__ ( self , agents : AgentListType ) :
def __init__ (
self ,
agents : AgentListType ,
name : str = " BaseSwarm " ,
description : str = " A base swarm implementation " ,
output_type : str = " dict " ,
) :
"""
Initialize the BaseSwarm with agents , name , description , and output type .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
# Ensure agents is a flat list of Agent objects
# Ensure agents is a flat list of Agent objects
self . agents = (
self . agents = (
[ agent for sublist in agents for agent in sublist ]
[ agent for sublist in agents for agent in sublist ]
if isinstance ( agents [ 0 ] , list )
if isinstance ( agents [ 0 ] , list )
else agents
else agents
)
)
self . name = name
self . description = description
self . output_type = output_type
self . conversation = Conversation ( )
self . conversation = Conversation ( )
def run (
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
self , tasks : List [ str ] , return_type : str = " dict "
) - > Union [ Dict , List , str ] :
"""
"""
Run the swarm with the given tasks
Run the swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -42,19 +60,10 @@ class BaseSwarm:
" This method should be implemented by child classes "
" This method should be implemented by child classes "
)
)
def _format_return (
def _format_return ( self ) - > Union [ Dict , List , str ] :
self , return_type : str
""" Format the return value based on the output_type using history_output_formatter """
) - > Union [ Dict , List , str ] :
return history_output_formatter (
""" Format the return value based on the return_type """
self . conversation , self . output_type
if return_type . lower ( ) == " dict " :
return self . conversation . return_messages_as_dictionary ( )
elif return_type . lower ( ) == " list " :
return self . conversation . return_messages_as_list ( )
elif return_type . lower ( ) == " string " :
return self . conversation . return_history_as_string ( )
else :
raise ValueError (
" return_type must be one of ' dict ' , ' list ' , or ' string ' "
)
)
@ -63,15 +72,30 @@ class CircularSwarm(BaseSwarm):
Implements a circular swarm where agents pass tasks in a circular manner .
Implements a circular swarm where agents pass tasks in a circular manner .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " CircularSwarm " ,
description : str = " A circular swarm where agents pass tasks in a circular manner " ,
output_type : str = " dict " ,
) :
"""
Initialize the CircularSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the circular swarm with the given tasks
Run the circular swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -92,7 +116,7 @@ class CircularSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class LinearSwarm ( BaseSwarm ) : class LinearSwarm ( BaseSwarm ) :
@ -100,15 +124,30 @@ class LinearSwarm(BaseSwarm):
Implements a linear swarm where agents process tasks sequentially .
Implements a linear swarm where agents process tasks sequentially .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " LinearSwarm " ,
description : str = " A linear swarm where agents process tasks sequentially " ,
output_type : str = " dict " ,
) :
"""
Initialize the LinearSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the linear swarm with the given tasks
Run the linear swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -131,7 +170,7 @@ class LinearSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class StarSwarm ( BaseSwarm ) : class StarSwarm ( BaseSwarm ) :
@ -139,15 +178,30 @@ class StarSwarm(BaseSwarm):
Implements a star swarm where a central agent processes all tasks , followed by others .
Implements a star swarm where a central agent processes all tasks , followed by others .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " StarSwarm " ,
description : str = " A star swarm where a central agent processes all tasks, followed by others " ,
output_type : str = " dict " ,
) :
"""
Initialize the StarSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the star swarm with the given tasks
Run the star swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -178,7 +232,7 @@ class StarSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class MeshSwarm ( BaseSwarm ) : class MeshSwarm ( BaseSwarm ) :
@ -186,15 +240,30 @@ class MeshSwarm(BaseSwarm):
Implements a mesh swarm where agents work on tasks randomly from a task queue .
Implements a mesh swarm where agents work on tasks randomly from a task queue .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " MeshSwarm " ,
description : str = " A mesh swarm where agents work on tasks randomly from a task queue " ,
output_type : str = " dict " ,
) :
"""
Initialize the MeshSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the mesh swarm with the given tasks
Run the mesh swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -218,7 +287,7 @@ class MeshSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class PyramidSwarm ( BaseSwarm ) : class PyramidSwarm ( BaseSwarm ) :
@ -226,15 +295,30 @@ class PyramidSwarm(BaseSwarm):
Implements a pyramid swarm where agents are arranged in a pyramid structure .
Implements a pyramid swarm where agents are arranged in a pyramid structure .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " PyramidSwarm " ,
description : str = " A pyramid swarm where agents are arranged in a pyramid structure " ,
output_type : str = " dict " ,
) :
"""
Initialize the PyramidSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the pyramid swarm with the given tasks
Run the pyramid swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -264,7 +348,7 @@ class PyramidSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class FibonacciSwarm ( BaseSwarm ) : class FibonacciSwarm ( BaseSwarm ) :
@ -272,15 +356,30 @@ class FibonacciSwarm(BaseSwarm):
Implements a Fibonacci swarm where agents are arranged according to the Fibonacci sequence .
Implements a Fibonacci swarm where agents are arranged according to the Fibonacci sequence .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " FibonacciSwarm " ,
description : str = " A Fibonacci swarm where agents are arranged according to the Fibonacci sequence " ,
output_type : str = " dict " ,
) :
"""
Initialize the FibonacciSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the Fibonacci swarm with the given tasks
Run the Fibonacci swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -309,7 +408,7 @@ class FibonacciSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class PrimeSwarm ( BaseSwarm ) : class PrimeSwarm ( BaseSwarm ) :
@ -317,15 +416,30 @@ class PrimeSwarm(BaseSwarm):
Implements a Prime swarm where agents at prime indices process tasks .
Implements a Prime swarm where agents at prime indices process tasks .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " PrimeSwarm " ,
description : str = " A Prime swarm where agents at prime indices process tasks " ,
output_type : str = " dict " ,
) :
"""
Initialize the PrimeSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the Prime swarm with the given tasks
Run the Prime swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -376,7 +490,7 @@ class PrimeSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class PowerSwarm ( BaseSwarm ) : class PowerSwarm ( BaseSwarm ) :
@ -384,15 +498,30 @@ class PowerSwarm(BaseSwarm):
Implements a Power swarm where agents at power - of - 2 indices process tasks .
Implements a Power swarm where agents at power - of - 2 indices process tasks .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " PowerSwarm " ,
description : str = " A Power swarm where agents at power-of-2 indices process tasks " ,
output_type : str = " dict " ,
) :
"""
Initialize the PowerSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the Power swarm with the given tasks
Run the Power swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -417,7 +546,7 @@ class PowerSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class LogSwarm ( BaseSwarm ) : class LogSwarm ( BaseSwarm ) :
@ -425,15 +554,30 @@ class LogSwarm(BaseSwarm):
Implements a Log swarm where agents at logarithmic indices process tasks .
Implements a Log swarm where agents at logarithmic indices process tasks .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " LogSwarm " ,
description : str = " A Log swarm where agents at logarithmic indices process tasks " ,
output_type : str = " dict " ,
) :
"""
Initialize the LogSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the Log swarm with the given tasks
Run the Log swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -457,7 +601,7 @@ class LogSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class ExponentialSwarm ( BaseSwarm ) : class ExponentialSwarm ( BaseSwarm ) :
@ -465,15 +609,30 @@ class ExponentialSwarm(BaseSwarm):
Implements an Exponential swarm where agents at exponential indices process tasks .
Implements an Exponential swarm where agents at exponential indices process tasks .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " ExponentialSwarm " ,
description : str = " An Exponential swarm where agents at exponential indices process tasks " ,
output_type : str = " dict " ,
) :
"""
Initialize the ExponentialSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the Exponential swarm with the given tasks
Run the Exponential swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -497,7 +656,7 @@ class ExponentialSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class GeometricSwarm ( BaseSwarm ) : class GeometricSwarm ( BaseSwarm ) :
@ -505,15 +664,30 @@ class GeometricSwarm(BaseSwarm):
Implements a Geometric swarm where agents at geometrically increasing indices process tasks .
Implements a Geometric swarm where agents at geometrically increasing indices process tasks .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " GeometricSwarm " ,
description : str = " A Geometric swarm where agents at geometrically increasing indices process tasks " ,
output_type : str = " dict " ,
) :
"""
Initialize the GeometricSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the Geometric swarm with the given tasks
Run the Geometric swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -538,7 +712,7 @@ class GeometricSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class HarmonicSwarm ( BaseSwarm ) : class HarmonicSwarm ( BaseSwarm ) :
@ -546,15 +720,30 @@ class HarmonicSwarm(BaseSwarm):
Implements a Harmonic swarm where agents at harmonically spaced indices process tasks .
Implements a Harmonic swarm where agents at harmonically spaced indices process tasks .
"""
"""
def run (
def __init__ (
self , tasks : List [ str ] , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " HarmonicSwarm " ,
description : str = " A Harmonic swarm where agents at harmonically spaced indices process tasks " ,
output_type : str = " dict " ,
) :
"""
Initialize the HarmonicSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , tasks : List [ str ] ) - > Union [ Dict , List , str ] :
"""
"""
Run the Harmonic swarm with the given tasks
Run the Harmonic swarm with the given tasks
Args :
Args :
tasks : List of tasks to be processed
tasks : List of tasks to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -580,7 +769,7 @@ class HarmonicSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class StaircaseSwarm ( BaseSwarm ) : class StaircaseSwarm ( BaseSwarm ) :
@ -588,15 +777,30 @@ class StaircaseSwarm(BaseSwarm):
Implements a Staircase swarm where agents at staircase - patterned indices process a task .
Implements a Staircase swarm where agents at staircase - patterned indices process a task .
"""
"""
def run (
def __init__ (
self , task : str , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " StaircaseSwarm " ,
description : str = " A Staircase swarm where agents at staircase-patterned indices process a task " ,
output_type : str = " dict " ,
) :
"""
Initialize the StaircaseSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , task : str ) - > Union [ Dict , List , str ] :
"""
"""
Run the Staircase swarm with the given task
Run the Staircase swarm with the given task
Args :
Args :
task : Task to be processed
task : Task to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -617,7 +821,7 @@ class StaircaseSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class SigmoidSwarm ( BaseSwarm ) : class SigmoidSwarm ( BaseSwarm ) :
@ -625,15 +829,30 @@ class SigmoidSwarm(BaseSwarm):
Implements a Sigmoid swarm where agents at sigmoid - distributed indices process a task .
Implements a Sigmoid swarm where agents at sigmoid - distributed indices process a task .
"""
"""
def run (
def __init__ (
self , task : str , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " SigmoidSwarm " ,
description : str = " A Sigmoid swarm where agents at sigmoid-distributed indices process a task " ,
output_type : str = " dict " ,
) :
"""
Initialize the SigmoidSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , task : str ) - > Union [ Dict , List , str ] :
"""
"""
Run the Sigmoid swarm with the given task
Run the Sigmoid swarm with the given task
Args :
Args :
task : Task to be processed
task : Task to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -653,7 +872,7 @@ class SigmoidSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
class SinusoidalSwarm ( BaseSwarm ) : class SinusoidalSwarm ( BaseSwarm ) :
@ -661,15 +880,30 @@ class SinusoidalSwarm(BaseSwarm):
Implements a Sinusoidal swarm where agents at sinusoidally - distributed indices process a task .
Implements a Sinusoidal swarm where agents at sinusoidally - distributed indices process a task .
"""
"""
def run (
def __init__ (
self , task : str , return_type : str = " dict "
self ,
) - > Union [ Dict , List , str ] :
agents : AgentListType ,
name : str = " SinusoidalSwarm " ,
description : str = " A Sinusoidal swarm where agents at sinusoidally-distributed indices process a task " ,
output_type : str = " dict " ,
) :
"""
Initialize the SinusoidalSwarm .
Args :
agents : List of Agent objects or nested list of Agent objects
name : Name of the swarm
description : Description of the swarm ' s purpose
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
super ( ) . __init__ ( agents , name , description , output_type )
def run ( self , task : str ) - > Union [ Dict , List , str ] :
"""
"""
Run the Sinusoidal swarm with the given task
Run the Sinusoidal swarm with the given task
Args :
Args :
task : Task to be processed
task : Task to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -689,7 +923,7 @@ class SinusoidalSwarm(BaseSwarm):
)
)
responses . append ( response )
responses . append ( response )
return self . _format_return ( return_type )
return self . _format_return ( )
# Communication classes # Communication classes
@ -698,13 +932,27 @@ class OneToOne:
Facilitates one - to - one communication between two agents .
Facilitates one - to - one communication between two agents .
"""
"""
def __init__ ( self , sender : Agent , receiver : Agent ) :
def __init__ (
self ,
sender : Agent ,
receiver : Agent ,
output_type : str = " dict " ,
) :
"""
Initialize the OneToOne communication .
Args :
sender : The sender agent
receiver : The receiver agent
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
self . sender = sender
self . sender = sender
self . receiver = receiver
self . receiver = receiver
self . output_type = output_type
self . conversation = Conversation ( )
self . conversation = Conversation ( )
def run (
def run (
self , task : str , max_loops : int = 1 , return_type : str = " dict "
self , task : str , max_loops : int = 1
) - > Union [ Dict , List , str ] :
) - > Union [ Dict , List , str ] :
"""
"""
Run the one - to - one communication with the given task
Run the one - to - one communication with the given task
@ -712,7 +960,6 @@ class OneToOne:
Args :
Args :
task : Task to be processed
task : Task to be processed
max_loops : Number of exchange iterations
max_loops : Number of exchange iterations
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -752,15 +999,8 @@ class OneToOne:
)
)
raise error
raise error
if return_type . lower ( ) == " dict " :
return history_output_formatter (
return self . conversation . return_messages_as_dictionary ( )
self . conversation , self . output_type
elif return_type . lower ( ) == " list " :
return self . conversation . return_messages_as_list ( )
elif return_type . lower ( ) == " string " :
return self . conversation . return_history_as_string ( )
else :
raise ValueError (
" return_type must be one of ' dict ' , ' list ' , or ' string ' "
)
)
@ -769,24 +1009,35 @@ class Broadcast:
Facilitates broadcasting from one agent to many agents .
Facilitates broadcasting from one agent to many agents .
"""
"""
def __init__ ( self , sender : Agent , receivers : AgentListType ) :
def __init__ (
self ,
sender : Agent ,
receivers : AgentListType ,
output_type : str = " dict " ,
) :
"""
Initialize the Broadcast communication .
Args :
sender : The sender agent
receivers : List of receiver agents
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
self . sender = sender
self . sender = sender
self . receivers = (
self . receivers = (
[ agent for sublist in receivers for agent in sublist ]
[ agent for sublist in receivers for agent in sublist ]
if isinstance ( receivers [ 0 ] , list )
if isinstance ( receivers [ 0 ] , list )
else receivers
else receivers
)
)
self . output_type = output_type
self . conversation = Conversation ( )
self . conversation = Conversation ( )
def run (
def run ( self , task : str ) - > Union [ Dict , List , str ] :
self , task : str , return_type : str = " dict "
) - > Union [ Dict , List , str ] :
"""
"""
Run the broadcast communication with the given task
Run the broadcast communication with the given task
Args :
Args :
task : Task to be processed
task : Task to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -812,17 +1063,8 @@ class Broadcast:
content = response ,
content = response ,
)
)
if return_type . lower ( ) == " dict " :
return history_output_formatter (
return (
self . conversation , self . output_type
self . conversation . return_messages_as_dictionary ( )
)
elif return_type . lower ( ) == " list " :
return self . conversation . return_messages_as_list ( )
elif return_type . lower ( ) == " string " :
return self . conversation . return_history_as_string ( )
else :
raise ValueError (
" return_type must be one of ' dict ' , ' list ' , or ' string ' "
)
)
except Exception as error :
except Exception as error :
@ -835,7 +1077,20 @@ class OneToThree:
Facilitates one - to - three communication from one agent to exactly three agents .
Facilitates one - to - three communication from one agent to exactly three agents .
"""
"""
def __init__ ( self , sender : Agent , receivers : AgentListType ) :
def __init__ (
self ,
sender : Agent ,
receivers : AgentListType ,
output_type : str = " dict " ,
) :
"""
Initialize the OneToThree communication .
Args :
sender : The sender agent
receivers : List of exactly three receiver agents
output_type : Type of output format , one of ' dict ' , ' list ' , ' string ' , ' json ' , ' yaml ' , ' xml ' , etc .
"""
if len ( receivers ) != 3 :
if len ( receivers ) != 3 :
raise ValueError (
raise ValueError (
" The number of receivers must be exactly 3. "
" The number of receivers must be exactly 3. "
@ -843,17 +1098,15 @@ class OneToThree:
self . sender = sender
self . sender = sender
self . receivers = receivers
self . receivers = receivers
self . output_type = output_type
self . conversation = Conversation ( )
self . conversation = Conversation ( )
def run (
def run ( self , task : str ) - > Union [ Dict , List , str ] :
self , task : str , return_type : str = " dict "
) - > Union [ Dict , List , str ] :
"""
"""
Run the one - to - three communication with the given task
Run the one - to - three communication with the given task
Args :
Args :
task : Task to be processed
task : Task to be processed
return_type : Type of return value , one of ' dict ' , ' list ' , or ' string '
Returns :
Returns :
Union [ Dict , List , str ] : The conversation history in the requested format
Union [ Dict , List , str ] : The conversation history in the requested format
@ -877,17 +1130,8 @@ class OneToThree:
content = response ,
content = response ,
)
)
if return_type . lower ( ) == " dict " :
return history_output_formatter (
return (
self . conversation , self . output_type
self . conversation . return_messages_as_dictionary ( )
)
elif return_type . lower ( ) == " list " :
return self . conversation . return_messages_as_list ( )
elif return_type . lower ( ) == " string " :
return self . conversation . return_history_as_string ( )
else :
raise ValueError (
" return_type must be one of ' dict ' , ' list ' , or ' string ' "
)
)
except Exception as error :
except Exception as error :
Kye Gomez
3 days ago