Heavy Swarm

The HeavySwarm implements a sophisticated 5-phase workflow inspired by X.AI’s Grok heavy implementation. It uses intelligent question generation and specialized agents (Research, Analysis, Alternatives, Verification) to provide comprehensive task analysis through multiple perspectives.

When to Use

Complex research tasks: In-depth investigation and analysis
Financial analysis: Investment decisions requiring multiple viewpoints
Strategic planning: Comprehensive evaluation of options
Due diligence: Thorough verification and risk assessment
Multi-faceted problems: Issues requiring research, analysis, and synthesis

Key Features

Intelligent question generation for specialized roles
4 specialized expert agents
True parallel execution
Synthesis agent for integration
Optional real-time dashboard
Multi-loop iterative refinement
Configurable agent loops
Tool integration support

Architecture Phases

1. Question Generation
   - Analyze task
   - Generate specialized questions for each agent role
   
2. Parallel Expert Execution
   - Research Agent: Gather comprehensive information
   - Analysis Agent: Analyze patterns and data
   - Alternatives Agent: Explore different approaches
   - Verification Agent: Validate and assess feasibility
   
3. Synthesis
   - Synthesis Agent: Integrate all expert outputs
   - Generate comprehensive final report

Basic Example

from swarms import HeavySwarm

# Create HeavySwarm
swarm = HeavySwarm(
    name="Investment-Research-Team",
    description="Comprehensive investment analysis",
    question_agent_model_name="gpt-4o-mini",
    worker_model_name="gpt-4o-mini",
    show_dashboard=True,
    max_loops=1,
)

# Execute comprehensive analysis
result = swarm.run(
    "Should we invest in NVIDIA stock? Provide detailed analysis."
)
print(result)

With Tools Integration

from swarms import HeavySwarm
from swarms_tools import exa_search  # Web search tool

swarm = HeavySwarm(
    name="Market-Research-Team",
    description="Research team with web search capabilities",
    question_agent_model_name="gpt-4o-mini",
    worker_model_name="claude-sonnet-4-20250514",
    worker_tools=[exa_search],  # Add tools to all workers
    show_dashboard=True,
    max_loops=2,  # Multiple refinement loops
)

result = swarm.run(
    "Find the best 3 gold ETFs with current data from the web"
)

Specialized Agents

Research Agent

Expert in:

Comprehensive information gathering
Data collection and validation
Source verification
Literature review
Statistical data interpretation

Analysis Agent

Expert in:

Pattern recognition
Statistical analysis
Predictive modeling
Data interpretation
Performance metrics

Alternatives Agent

Expert in:

Strategic thinking
Creative problem-solving
Option generation
Trade-off evaluation
Scenario planning

Verification Agent

Expert in:

Fact-checking
Feasibility assessment
Risk analysis
Compliance verification
Quality assurance

Synthesis Agent

Expert in:

Multi-perspective integration
Comprehensive analysis
Executive summary creation
Strategic alignment
Actionable recommendations

Key Parameters

name

str

default:"HeavySwarm"

Name for the swarm instance

description

str

Description of the swarm’s purpose

question_agent_model_name

str

default:"gpt-4o-mini"

Model for question generation agent

worker_model_name

str

default:"gpt-4o-mini"

Model for all specialized worker agents

loops_per_agent

int

default:1

Number of loops each worker agent executes

show_dashboard

bool

default:false

Enable real-time progress dashboard

agent_prints_on

bool

default:false

Enable individual agent output printing

worker_tools

List[Callable]

Tools available to worker agents

max_loops

int

default:1

Maximum swarm execution loops for iterative refinement

random_loops_per_agent

bool

default:false

Randomize loops per agent (1-10 range)

output_type

str

default:"dict-all-except-first"

Output format type

Methods

run()

Execute the complete 5-phase workflow.

result = swarm.run(
    task="Analyze cryptocurrency market trends",
    img=None,  # Optional image input
)

Question Generation

The question agent generates specialized questions:

# Internal schema for question generation
{
    "thinking": "Reasoning for how to break down task",
    "research_question": "Question for Research Agent",
    "analysis_question": "Question for Analysis Agent",
    "alternatives_question": "Question for Alternatives Agent",
    "verification_question": "Question for Verification Agent"
}

Example generated questions for “Invest in NVIDIA?”:

{
    "thinking": "Need to evaluate financials, market position, alternatives, and risks",
    "research_question": "Research NVIDIA's financial performance, product pipeline, and market position in AI chips",
    "analysis_question": "Analyze NVIDIA's revenue growth, profit margins, and competitive advantages in the AI accelerator market",
    "alternatives_question": "What are alternative semiconductor investments with similar growth potential?",
    "verification_question": "Verify NVIDIA's claimed AI leadership and assess risks to their market dominance"
}

With max_loops > 1, the swarm iterates for deeper analysis:

swarm = HeavySwarm(
    name="Deep-Analysis-Team",
    worker_model_name="claude-sonnet-4-20250514",
    max_loops=3,  # Three rounds of analysis
)

result = swarm.run("Complex strategic decision")

Each loop:

Loop 1: Initial analysis with original task
Loop 2: Refine based on Loop 1 results + original task
Loop 3: Final refinement with complete context

Dashboard Features

When show_dashboard=True: Configuration Panel:

Swarm name and description
Model configurations
Timeout and loop settings
Worker count

Reliability Check Phase:

Animated progress bars
Validation status for each component
System ready confirmation

Question Generation Phase:

Real-time generation progress
Success confirmation

Agent Execution Phase:

Individual progress bars per agent
Status updates (INITIALIZING, PROCESSING, GENERATING, COMPLETE)
Animated indicators
Error detection

Synthesis Phase:

Integration progress
Final report generation
Completion confirmation

Use Cases

Investment Analysis

swarm = HeavySwarm(
    name="Investment-Committee",
    description="Comprehensive investment evaluation team",
    worker_model_name="claude-sonnet-4-20250514",
    show_dashboard=True,
)

analysis = swarm.run(
    "Evaluate Tesla as a long-term investment. Consider financials, "
    "market position, competition, risks, and alternatives."
)

Market Research

from swarms_tools import exa_search

swarm = HeavySwarm(
    name="Market-Intel-Team",
    worker_model_name="gpt-4o-mini",
    worker_tools=[exa_search],
    loops_per_agent=2,
)

research = swarm.run(
    "Research the enterprise AI market: size, growth, key players, "
    "trends, and opportunities. Use current web data."
)

Strategic Planning

swarm = HeavySwarm(
    name="Strategy-Team",
    description="Strategic planning and analysis",
    worker_model_name="claude-sonnet-4-20250514",
    max_loops=2,
    show_dashboard=True,
)

strategy = swarm.run(
    "Should our company enter the AI agent services market? "
    "Analyze market, competition, requirements, risks, and alternatives."
)

Due Diligence

swarm = HeavySwarm(
    name="Due-Diligence-Team",
    worker_model_name="gpt-4o-mini",
    loops_per_agent=3,  # Thorough investigation
)

report = swarm.run(
    "Conduct due diligence on acquiring Company X: "
    "financials, legal, operations, technology, culture fit."
)

Performance Optimization

Parallel Execution

All 4 expert agents execute simultaneously:

# Uses ThreadPoolExecutor for true parallel execution
with concurrent.futures.ThreadPoolExecutor(
    max_workers=max_workers
) as executor:
    futures = [executor.submit(execute_agent, task) for task in agent_tasks]

Agent Loop Configuration

# Fixed loops per agent
swarm = HeavySwarm(loops_per_agent=3)

# Random loops (1-10) per agent
swarm = HeavySwarm(random_loops_per_agent=True)

Advanced Configuration

Custom Aggregation

swarm = HeavySwarm(
    aggregation_strategy="synthesis",  # Default
    # Future: "voting", "consensus", "weighted"
)

Timeout Management

swarm = HeavySwarm(
    timeout=300,  # 5 minutes per agent
)

Verbose Logging

swarm = HeavySwarm(
    verbose=True,  # Detailed execution logs
    agent_prints_on=True,  # Individual agent outputs
)

Output Structure

The swarm returns formatted conversation history:

# With output_type="dict-all-except-first"
{
    "Question Generator Agent": "<generated questions>",
    "Research-Agent": "<research findings>",
    "Analysis-Agent": "<analysis insights>",
    "Alternatives-Agent": "<alternative options>",
    "Verification-Agent": "<verification results>",
    "Synthesis-Agent": "<comprehensive synthesis>"
}

Best Practices

Model Selection: Use stronger models (Claude Sonnet, GPT-4) for worker agents on complex tasks

Question Quality: Stronger model for question generation = better agent coordination
Worker Models: Balance cost vs quality for worker agents
Loop Count: Start with 1, add more only for complex tasks
Dashboard: Enable for demos and debugging, disable for production
Tools: Provide tools when real-time data is needed

HeavySwarm is resource-intensive: 4 agents + synthesis + question generation per loop

Reliability Checks

The swarm validates configuration:

# Automatic validation:
# - loops_per_agent > 0
# - worker_model_name is set
# - question_agent_model_name is set

try:
    swarm = HeavySwarm(loops_per_agent=0)
except ValueError as e:
    print(e)
    # "loops_per_agent must be greater than 0"

Conversation History

Access complete analysis history:

result = swarm.run("Task")

# Access conversation
history = swarm.conversation.conversation_history

# Format:
# [{"role": "User", "content": "Task"},
#  {"role": "Question Generator Agent", "content": {...}},
#  {"role": "Research-Agent", "content": "..."},
#  {"role": "Analysis-Agent", "content": "..."},
#  {"role": "Alternatives-Agent", "content": "..."},
#  {"role": "Verification-Agent", "content": "..."},
#  {"role": "Synthesis-Agent", "content": "..."}]

Error Handling

try:
    result = swarm.run("Task")
except TimeoutError:
    print("Agent execution exceeded timeout")
except Exception as e:
    print(f"Swarm execution failed: {e}")
    # Individual agent errors are logged but don't stop other agents

Mixture of Agents - Similar parallel expert pattern
Hierarchical Swarm - Director-worker coordination
Concurrent Workflow - Simple parallel execution
Agent Rearrange - Custom flow patterns

Getting Started

Core Concepts

Agent Development

Multi-Agent Architectures

Deployment & Operations

Integrations

CLI Reference

​Heavy Swarm

​When to Use

​Key Features

​Architecture Phases

​Basic Example

​With Tools Integration

​Specialized Agents

​Research Agent

​Analysis Agent

​Alternatives Agent

​Verification Agent

​Synthesis Agent

​Key Parameters

​Methods

​run()

​Question Generation

​Multi-Loop Refinement

​Dashboard Features

​Use Cases

​Investment Analysis

​Market Research

​Strategic Planning

​Due Diligence

​Performance Optimization

​Parallel Execution

​Agent Loop Configuration

​Advanced Configuration

​Custom Aggregation

​Timeout Management

​Verbose Logging

​Output Structure

​Best Practices

​Reliability Checks

​Conversation History

​Error Handling

​Related Architectures

Build docs developers (and LLMs) love

Heavy Swarm

When to Use

Key Features

Architecture Phases

Basic Example

With Tools Integration

Specialized Agents

Research Agent

Analysis Agent

Alternatives Agent

Verification Agent

Synthesis Agent

Key Parameters

Methods

run()

Question Generation

Multi-Loop Refinement

Dashboard Features

Use Cases

Investment Analysis

Market Research

Strategic Planning

Due Diligence

Performance Optimization

Parallel Execution

Agent Loop Configuration

Advanced Configuration

Custom Aggregation

Timeout Management

Verbose Logging

Output Structure

Best Practices

Reliability Checks

Conversation History

Error Handling

Related Architectures