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. The aggregator reads every worker answer from every layer as a separate, labelled chat turn. With more than one layer, each label carries the layer, for example Analyst (layer 2/3), so the aggregator can tell a refinement from a first pass.

Class Definition

from swarms import MixtureOfAgents

Parameters

str

default:"auto-generated"

Unique identifier for the mixture of agents instance. Auto-generated via generate_id("mixture-of-agents") if not provided, producing mixture-of-agents-<32 hex chars>.

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. An empty list raises ValueError

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"

Stored for compatibility with other swarm classes. It does not change how many times the workers or the aggregator run; use layers for more worker rounds

OutputType

default:"final"

Output format type. "final" returns the aggregator’s answer; "all", "list", "dict" and the other OutputType values return the whole conversation in that format

str

default:"claude-sonnet-4-20250514"

The model name for the aggregator agent

int

Cap on concurrent worker agents per layer. When omitted, the pool uses 95% of the available CPU cores.

Dict[str, Any]

Extra keyword arguments forwarded to the aggregator Agent when one is built for you.

The parameter is spelled aggegrator_args in the code — the letters are transposed. That misspelling is the name you must type; aggregator_args raises TypeError.

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. It does not write to the conversation; run() records the results. 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 final answers. Each value is the agent’s last message, not its whole transcript

`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

run() catches exceptions internally and returns the string "Error: {e}" instead of raising, so callers should check the returned value rather than wrapping the call in a try/except.

Workflow:

Starts a fresh conversation, so a reused instance does not carry the previous task into the next one, and adds the task
For each layer:
- Runs all agents concurrently. The first layer receives the task; later layers receive the original task plus the previous layer’s answers, each prefixed with the agent’s name, not the full transcript
- Adds each agent’s answer to the conversation. When layers > 1 the speaker is labelled with the layer, for example Research-Agent (layer 1/3)
Runs the aggregator on the conversation, passed as chat turns: the task, then every labelled worker answer. The team roster is left out
Adds the aggregator’s answer to the conversation and returns the conversation formatted by 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, in task order

run_concurrently() runs every task on this one instance. Each run() replaces self.conversation and reuses the same worker agents, so concurrent tasks can overwrite each other’s state. Use run_batched(), or one MixtureOfAgents per task, when results must stay isolated.

Usage Example

from swarms import Agent, 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. The first layer sees only the task; each later layer sees the task plus the previous layer’s answers
Context Accumulation: Agent answers are added to the conversation after each layer, labelled with the agent and, when layers > 1, the layer
Iterative Refinement: Subsequent layers build upon previous outputs, enabling deeper analysis
Final Aggregation: The aggregator reads every labelled answer from every layer as chat turns and synthesizes them into one 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