MixtureOfAgents

Overview

The MixtureOfAgents class manages and executes multiple agents in parallel layers, aggregating their responses through a specialized aggregator agent. This architecture enables sophisticated multi-perspective analysis by running agents concurrently and synthesizing their outputs.

Class Definition

from swarms.structs.mixture_of_agents import MixtureOfAgents

Parameters

str

default:"uuid.uuid4()"

Unique identifier for the mixture of agents instance

str

default:"MixtureOfAgents"

The name of the mixture of agents

str

A description of the mixture of agents purpose and functionality

List[Agent]

default:"None"

A list of reference agents to be used in the mixture. These agents will be executed in parallel layers

Agent

default:"None"

The aggregator agent to be used for synthesizing responses from all agents. If None, a default aggregator agent will be created

str

default:"AGGREGATOR_SYSTEM_PROMPT_MAIN"

The system prompt for the aggregator agent that guides how responses are synthesized

int

default:"3"

The number of processing layers to execute. Each layer runs all agents and passes context forward

int

default:"1"

Maximum number of execution loops for the aggregator agent

OutputType

default:"final"

Output format type. Options: “final”, “all”, “list”, etc.

str

default:"claude-sonnet-4-20250514"

The model name for the aggregator agent

Methods

`reliability_check()`

def reliability_check(self) -> None

Performs a reliability check on the Mixture of Agents class configuration. Raises:

ValueError: If no agents are provided
ValueError: If no aggregator system prompt is provided
ValueError: If no layers are specified

`step()`

def step(
    self,
    task: str,
    img: Optional[str] = None,
) -> Dict[str, str]

Executes a single step by running all agents concurrently with the given task. Parameters:

str

required

The task to be executed by all agents

str

Optional image input for the task

Returns:

Dict[str, str]

Dictionary mapping agent names to their output responses

`run()`

def run(
    self,
    task: str,
    img: Optional[str] = None,
) -> str

Executes the complete mixture of agents workflow with multiple layers and aggregation. Parameters:

str

required

The task to be executed by the mixture of agents

str

Optional image input for the task

Returns:

str

The aggregated response from all agents after processing through all layers

Workflow:

Adds initial task to conversation history
For each layer:
- Runs all agents concurrently with full context
- Adds each agent’s output to conversation
- Updates context with latest conversation history
Runs aggregator agent on complete conversation
Returns formatted output based on output_type

`run_batched()`

def run_batched(self, tasks: List[str]) -> List[str]

Runs the mixture of agents for a batch of tasks sequentially. Parameters:

List[str]

required

A list of tasks to be executed by the mixture of agents

Returns:

List[str]

A list of aggregated responses from the mixture of agents, one for each task

`run_concurrently()`

def run_concurrently(self, tasks: List[str]) -> List[str]

Runs the mixture of agents for a batch of tasks concurrently using ThreadPoolExecutor. Parameters:

List[str]

required

A list of tasks to be executed concurrently

Returns:

List[str]

A list of aggregated responses from the mixture of agents

Usage Example

from swarms.structs.agent import Agent
from swarms.structs.mixture_of_agents import MixtureOfAgents

# Create specialized agents
research_agent = Agent(
    agent_name="Research-Agent",
    system_prompt="You are a research expert...",
    model_name="claude-sonnet-4-6"
)

analysis_agent = Agent(
    agent_name="Analysis-Agent",
    system_prompt="You are an analysis expert...",
    model_name="claude-sonnet-4-6"
)

writing_agent = Agent(
    agent_name="Writing-Agent",
    system_prompt="You are a writing expert...",
    model_name="claude-sonnet-4-6"
)

# Create mixture of agents
mixture = MixtureOfAgents(
    name="Research-Analysis-Writing-Team",
    description="A team that researches, analyzes, and writes reports",
    agents=[research_agent, analysis_agent, writing_agent],
    layers=3,
    output_type="final"
)

# Run a task
result = mixture.run(
    task="Analyze the impact of AI on healthcare and write a comprehensive report"
)

print(result)

Architecture

The MixtureOfAgents architecture works as follows:

Layer Processing: Each layer runs all agents concurrently with the full conversation context
Context Accumulation: Agent outputs are added to the conversation history after each layer
Iterative Refinement: Subsequent layers build upon previous outputs, enabling deeper analysis
Final Aggregation: An aggregator agent synthesizes all responses into a coherent final output

Source Code

View the source code on GitHub

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

MixtureOfAgents

Overview

Class Definition

Parameters

Methods

`reliability_check()`

`step()`

`run()`

`run_batched()`

`run_concurrently()`

Usage Example

Architecture

Source Code

​Overview

​Class Definition

​Parameters

​Methods

​reliability_check()

​step()

​run()

​run_batched()

​run_concurrently()

​Usage Example

​Architecture

​Source Code

Overview

Class Definition

Parameters

Methods

`reliability_check()`

`step()`

`run()`

`run_batched()`

`run_concurrently()`

Usage Example

Architecture

Source Code