HeavySwarm¶

HeavySwarm is a sophisticated multi-agent orchestration system that decomposes complex tasks into specialized questions and executes them using specialized agents in parallel. The results are then synthesized into a comprehensive response.

HeavySwarm supports two agent architectures:

Default mode — 5 task-phase agents (Research, Analysis, Alternatives, Verification, Synthesis)
Grok mode (use_grok_agents=True) — 4 thinking-style agents inspired by the Grok 4.20 Heavy architecture (Captain Swarm, Harper, Benjamin, Lucas)

Installation¶

pip install swarms

Quick Start¶

from swarms.structs.heavy_swarm import HeavySwarm

# Initialize HeavySwarm
swarm = HeavySwarm(
    name="Research Team",
    description="Multi-agent analysis system",
    worker_model_name="gpt-5.4",
    show_dashboard=True
)

# Run analysis
result = swarm.run("Analyze the impact of AI on healthcare")
print(result)

Architecture¶

The HeavySwarm follows a structured workflow:

Task Decomposition: Complex tasks are broken down into specialized questions
Question Generation: AI-powered generation of role-specific questions
Parallel Execution: Specialized agents work concurrently
Result Collection: Outputs are gathered and validated
Synthesis: Integration into a comprehensive final response

Default Mode — Task-Phase Agents¶

The default mode decomposes work by task phase — each agent handles a different stage of analysis.

Agent	Role/Function
Research Agent	Comprehensive information gathering and synthesis
Analysis Agent	Pattern recognition and statistical analysis
Alternatives Agent	Creative problem-solving and strategic options
Verification Agent	Validation, feasibility assessment, and quality assurance
Synthesis Agent	Multi-perspective integration and executive reporting

Flow: LiteLLM question generator → 4 workers in parallel → Synthesis Agent

Grok Mode — Thinking-Style Agents (`use_grok_agents=True`)¶

The Grok mode is inspired by the Grok 4.20 Heavy architecture and decomposes work by thinking style — each agent applies a fundamentally different cognitive approach.

Agent	Specialization	Thinking Style
Captain Swarm	Orchestration, task decomposition, conflict resolution, final synthesis	Structured, managerial
Harper	Research and facts, evidence gathering, source verification, data grounding	Empirical, evidence-based
Benjamin	Logic, math, and code — rigorous reasoning, numerical verification, stress-testing	Rigorous, analytical
Lucas	Creative and divergent thinking, contrarian analysis, blind-spot detection, bias identification	Divergent, lateral

Flow: Captain Swarm decomposes → Harper, Benjamin, Lucas work in parallel → Captain Swarm mediates conflicts and synthesizes

Key Differences from Default Mode¶

Aspect	Default Mode	Grok Mode
Decomposition	By task phase (research → analyze → verify)	By thinking style (facts vs logic vs creativity)
Agent count	5 (4 workers + 1 synthesizer)	4 (3 specialists + 1 leader)
Synthesis	Generic integration agent	Captain Swarm mediates conflicts between agents
Contrarian check	Not present	Lucas specifically challenges assumptions and detects biases
Question schema	4 questions (research, analysis, alternatives, verification)	3 questions (harper, benjamin, lucas)

API Reference¶

HeavySwarm Class¶

A sophisticated multi-agent orchestration system for complex task analysis.

Constructor¶

HeavySwarm(
    name: str = "HeavySwarm",
    description: str = "A swarm of agents that can analyze a task and generate specialized questions for each agent role",
    timeout: int = 300,
    aggregation_strategy: str = "synthesis",
    loops_per_agent: int = 1,
    question_agent_model_name: str = "gpt-5.4",
    worker_model_name: str = "gpt-5.4",
    verbose: bool = False,
    max_workers: int = int(os.cpu_count() * 0.9),
    show_dashboard: bool = False,
    agent_prints_on: bool = False,
    output_type: str = "dict-all-except-first",
    worker_tools: Optional[tool_type] = None,
    random_loops_per_agent: bool = False,
    max_loops: int = 1,
    use_grok_agents: bool = False,
)

Parameters¶

Parameter	Type	Default	Description
`name`	`str`	`"HeavySwarm"`	Identifier name for the swarm instance
`description`	`str`	`"A swarm of agents that can analyze a task and generate specialized questions for each agent role"`	Description of the swarm's purpose and capabilities
`timeout`	`int`	`300`	Maximum execution time per agent in seconds
`aggregation_strategy`	`str`	`"synthesis"`	Strategy for result aggregation (currently only 'synthesis' is supported)
`loops_per_agent`	`int`	`1`	Number of execution loops each agent should perform
`question_agent_model_name`	`str`	`"gpt-5.4"`	Language model for question generation
`worker_model_name`	`str`	`"gpt-5.4"`	Language model for specialized worker agents
`verbose`	`bool`	`False`	Enable detailed logging and debug output
`max_workers`	`int`	`int(os.cpu_count() * 0.9)`	Maximum concurrent workers for parallel execution
`show_dashboard`	`bool`	`False`	Enable rich dashboard with progress visualization
`agent_prints_on`	`bool`	`False`	Enable individual agent output printing
`output_type`	`str`	`"dict-all-except-first"`	Output formatting type for conversation history
`worker_tools`	`Optional[tool_type]`	`None`	Tools available to worker agents for enhanced functionality
`random_loops_per_agent`	`bool`	`False`	Enable random number of loops per agent (1-10 range)
`max_loops`	`int`	`1`	Maximum number of execution loops for iterative refinement
`use_grok_agents`	`bool`	`False`	Enable Grok 4.20 Heavy architecture with Captain Swarm, Harper, Benjamin, and Lucas agents instead of the default Research/Analysis/Alternatives/Verification agents

Raises¶

ValueError: If loops_per_agent is 0 or negative
ValueError: If worker_model_name is None
ValueError: If question_agent_model_name is None

Example¶

swarm = HeavySwarm(
    name="AdvancedAnalysisSwarm",
    description="Comprehensive multi-agent analysis system",
    timeout=600,
    loops_per_agent=2,
    question_agent_model_name="gpt-4.1",
    worker_model_name="gpt-4.1",
    verbose=True,
    max_workers=8,
    show_dashboard=True,
    agent_prints_on=True,
    output_type="dict-all-except-first",
    worker_tools=None,
    random_loops_per_agent=False
)

Methods¶

`run(task: str, img: Optional[str] = None) -> str`¶

Execute the complete HeavySwarm orchestration flow.

Parameters:

Parameter	Type	Required	Default	Description
`task`	`str`	Yes	-	The main task to analyze and decompose
`img`	`Optional[str]`	No	`None`	Image input for visual analysis tasks

Returns:

str: Comprehensive final analysis from synthesis agent

Example:

result = swarm.run("Develop a go-to-market strategy for a new SaaS product")
print(result)

`get_questions_only(task: str) -> Dict[str, str]`¶

Generate and extract only the specialized questions without metadata or execution.

Parameters:

Parameter	Type	Required	Default	Description
`task`	`str`	Yes	-	The main task or query to analyze and decompose into specialized questions

Returns:

Dict[str, str]: Clean dictionary containing only the questions.

Default mode keys: - research_question (str): Question for comprehensive information gathering - analysis_question (str): Question for pattern analysis and insights - alternatives_question (str): Question for exploring creative solutions - verification_question (str): Question for validation and feasibility

Grok mode keys (use_grok_agents=True): - harper_question (str): Question for evidence-based research and fact verification - benjamin_question (str): Question for logical reasoning and mathematical verification - lucas_question (str): Question for creative/contrarian analysis and blind-spot detection

error (str): Error message if question generation fails (on error only)

Example:

# Default mode
questions = swarm.get_questions_only("Analyze market trends for EVs")
print(questions['research_question'])
print(questions['analysis_question'])

# Grok mode
grok_swarm = HeavySwarm(use_grok_agents=True, worker_model_name="gpt-4.1", question_agent_model_name="gpt-4.1")
questions = grok_swarm.get_questions_only("Analyze market trends for EVs")
print(questions['harper_question'])
print(questions['benjamin_question'])
print(questions['lucas_question'])

`get_questions_as_list(task: str) -> List[str]`¶

Generate specialized questions and return them as an ordered list.

Parameters:

Parameter	Type	Required	Default	Description
`task`	`str`	Yes	-	The main task or query to decompose into specialized questions

Returns:

List[str]: Ordered list of specialized questions.

Default mode (4 questions): - [0] Research question for comprehensive information gathering - [1] Analysis question for pattern analysis and insights - [2] Alternatives question for exploring creative solutions - [3] Verification question for validation and feasibility assessment

Grok mode (3 questions): - [0] Harper question for evidence-based research and fact verification - [1] Benjamin question for logical reasoning and mathematical verification - [2] Lucas question for creative/contrarian analysis and blind-spot detection

Single-item list containing error message (on error)

Example:

questions = swarm.get_questions_as_list("Optimize supply chain efficiency")
for i, question in enumerate(questions):
    print(f"Agent {i+1}: {question}")

`show_swarm_info() -> None`¶

Display comprehensive swarm configuration information in a rich dashboard format.

Parameters:

None

Returns:

None: This method only displays output and has no return value.

Example:

swarm.show_swarm_info()  # Displays configuration dashboard

`reliability_check() -> None`¶

Perform comprehensive reliability and configuration validation checks.

Parameters:

None

Returns:

None: This method only performs validation and has no return value.

Raises:

Exception	Condition
`ValueError`	If `loops_per_agent` is 0 or negative (agents won't execute)
`ValueError`	If `worker_model_name` is None (agents can't be created)
`ValueError`	If `question_agent_model_name` is None (questions can't be generated)

Example:

try:
    swarm.reliability_check()  # Automatically called in __init__
    print("All checks passed!")
except ValueError as e:
    print(f"Configuration error: {e}")

`create_agents() -> Dict[str, Agent]`¶

Create and cache specialized agents with detailed role-specific prompts. The agents created depend on the use_grok_agents setting.

Parameters:

None

Returns (default mode):

Key	Agent Instance	Description
`'research'`	Research Agent	Research Agent instance
`'analysis'`	Analysis Agent	Analysis Agent instance
`'alternatives'`	Alternatives Agent	Alternatives Agent instance
`'verification'`	Verification Agent	Verification Agent instance
`'synthesis'`	Synthesis Agent	Synthesis Agent instance

Returns (Grok mode, use_grok_agents=True):

Key	Agent Instance	Description
`'captain'`	Captain Swarm	Leader and orchestrator — handles decomposition and synthesis
`'harper'`	Harper	Research and facts specialist
`'benjamin'`	Benjamin	Logic, math, and code specialist
`'lucas'`	Lucas	Creative and divergent thinking specialist

Example:

# Default mode
agents = swarm.create_agents()
research_agent = agents['research']
print(f"Research agent name: {research_agent.agent_name}")

# Grok mode
grok_swarm = HeavySwarm(use_grok_agents=True, worker_model_name="gpt-4.1", question_agent_model_name="gpt-4.1")
agents = grok_swarm.create_agents()
print(f"Captain: {agents['captain'].agent_name}")
print(f"Harper: {agents['harper'].agent_name}")

Examples¶

Grok Mode — Quick Start¶

from swarms.structs.heavy_swarm import HeavySwarm

# Enable Grok 4.20 Heavy architecture
swarm = HeavySwarm(
    name="Grok Analysis Team",
    description="Multi-agent analysis with Grok-style agents",
    worker_model_name="gpt-4.1",
    question_agent_model_name="gpt-4.1",
    use_grok_agents=True,
    show_dashboard=True,
)

# Captain Swarm decomposes the task, Harper/Benjamin/Lucas
# work in parallel, then Captain synthesizes with conflict resolution
result = swarm.run("Evaluate whether Tesla should expand into commercial trucking")
print(result)

Grok Mode — Medical Research¶

from swarms.structs.heavy_swarm import HeavySwarm

swarm = HeavySwarm(
    name="Medical Research Team",
    worker_model_name="gpt-4.1",
    question_agent_model_name="gpt-4.1",
    use_grok_agents=True,
    loops_per_agent=2,
    show_dashboard=True,
)

# Harper gathers clinical evidence, Benjamin verifies
# statistical claims, Lucas challenges treatment assumptions
result = swarm.run(
    "Analyze the latest research on GLP-1 receptor agonists "
    "for treating obesity: efficacy, long-term safety, cost-effectiveness, "
    "and potential off-label applications"
)
print(result)

Grok Mode — Strategic Analysis with Tools¶

from swarms.structs.heavy_swarm import HeavySwarm

def web_search(query: str) -> str:
    """Search the web for information."""
    # Your search implementation
    ...

swarm = HeavySwarm(
    name="Strategic Intel Team",
    worker_model_name="gpt-4.1",
    question_agent_model_name="gpt-4.1",
    use_grok_agents=True,
    worker_tools=[web_search],
    show_dashboard=True,
    timeout=600,
)

# Harper uses tools for real-time fact-gathering,
# Benjamin stress-tests the strategy mathematically,
# Lucas identifies blind spots and contrarian opportunities
result = swarm.run(
    "Should our company enter the Southeast Asian market in 2026? "
    "Consider regulatory environment, competitive landscape, "
    "logistics costs, and consumer behavior patterns."
)
print(result)

Grok Mode via SwarmRouter¶

from swarms.structs.swarm_router import SwarmRouter

router = SwarmRouter(
    name="GrokRouter",
    description="Router with Grok Heavy agents",
    swarm_type="HeavySwarm",
    heavy_swarm_worker_model_name="gpt-4.1",
    heavy_swarm_question_agent_model_name="gpt-4.1",
    heavy_swarm_use_grok_agents=True,
)

result = router.run("Analyze the future of autonomous vehicles")
print(result)

Grok Mode — Question Preview¶

from swarms.structs.heavy_swarm import HeavySwarm

swarm = HeavySwarm(
    worker_model_name="gpt-4.1",
    question_agent_model_name="gpt-4.1",
    use_grok_agents=True,
)

# Preview what Captain Swarm will ask each specialist
questions = swarm.get_questions_only("Should we migrate from PostgreSQL to CockroachDB?")
print(f"Harper (facts): {questions['harper_question']}")
print(f"Benjamin (logic): {questions['benjamin_question']}")
print(f"Lucas (creative): {questions['lucas_question']}")

# Or as a list
question_list = swarm.get_questions_as_list("Should we migrate from PostgreSQL to CockroachDB?")
for i, q in enumerate(question_list):
    print(f"Agent {i+1}: {q}")

Basic Usage (Default Mode)¶

from swarms.structs.heavy_swarm import HeavySwarm

# Initialize HeavySwarm
swarm = HeavySwarm(
    name="Analysis Team",
    description="Multi-agent analysis system",
    worker_model_name="gpt-5.4",
    show_dashboard=True
)

# Run analysis
result = swarm.run("Analyze the impact of AI on healthcare")
print(result)

Financial Research with Tools¶

from swarms.structs.heavy_swarm import HeavySwarm
from swarms_tools import exa_search

# Initialize with tools
swarm = HeavySwarm(
    name="Gold ETF Research Team",
    description="Financial research with web scraping",
    worker_model_name="gpt-5.4",
    question_agent_model_name="gpt-5.4",
    show_dashboard=True,
    worker_tools=[exa_search],
    timeout=300
)

# Research prompt
prompt = """
Find the best 3 gold ETFs. For each ETF, provide:
- Ticker symbol and full name
- Current price and expense ratio
- Assets under management
- Why it's considered among the best
"""

result = swarm.run(prompt)
print(result)

Advanced Market Analysis¶

from swarms.structs.heavy_swarm import HeavySwarm

# Advanced configuration
swarm = HeavySwarm(
    name="AdvancedMarketAnalysis",
    description="Deep market analysis with multiple iterations",
    worker_model_name="gpt-4.1",
    question_agent_model_name="gpt-4.1",
    show_dashboard=True,
    loops_per_agent=3,  # Multiple iterations for depth
    timeout=600,
    max_workers=8,
    verbose=True
)

# Complex analysis prompt
complex_prompt = """
Conduct comprehensive analysis of renewable energy sector:
1. Market size and growth projections
2. Technology landscape and adoption rates
3. Investment opportunities and risks
4. Regulatory environment
5. Competitive analysis
"""

result = swarm.run(complex_prompt)
print(result)

Question Generation Only¶

from swarms.structs.heavy_swarm import HeavySwarm

# Configure for question generation
swarm = HeavySwarm(
    name="QuestionGenerator",
    question_agent_model_name="gpt-4.1",
    worker_model_name="gpt-5.4",
    show_dashboard=False
)

# Generate questions without execution
task = "Develop digital transformation strategy for manufacturing"
questions = swarm.get_questions_only(task)

print("Generated Questions:")
for key, question in questions.items():
    if key != 'error':
        print(f"\n{key.upper()}:")
        print(question)

Batch Processing¶

from swarms.structs.heavy_swarm import HeavySwarm

# Configure for batch processing
swarm = HeavySwarm(
    name="BatchProcessor",
    worker_model_name="gpt-5.4",
    question_agent_model_name="gpt-5.4",
    show_dashboard=False,  # Disable for batch efficiency
    loops_per_agent=1,
    timeout=300,
    verbose=False
)

# Batch tasks
tasks = [
    "Analyze remote work impact on commercial real estate",
    "Evaluate electric vehicles in taxi industry",
    "Assess IoT cybersecurity risks",
    "Explore blockchain in supply chain"
]

# Process batch
results = {}
for task in tasks:
    try:
        result = swarm.run(task)
        results[task] = result
        print(f"✅ Completed: {task[:50]}...")
    except Exception as e:
        results[task] = f"Error: {str(e)}"
        print(f"❌ Failed: {task[:50]}...")

print(f"\nBatch complete: {len(results)} tasks processed")

Error Handling¶

from swarms.structs.heavy_swarm import HeavySwarm

# Test configuration validation
try:
    swarm = HeavySwarm(
        name="TestSwarm",
        worker_model_name="gpt-5.4",
        question_agent_model_name="gpt-5.4"
    )
    print("✅ Configuration valid")
except ValueError as e:
    print(f"❌ Configuration error: {e}")

# Test question generation with error handling
swarm = HeavySwarm(
    name="ErrorTest",
    worker_model_name="gpt-5.4",
    question_agent_model_name="gpt-5.4"
)

try:
    questions = swarm.get_questions_only("Test task")
    if 'error' in questions:
        print(f"❌ Question generation failed: {questions['error']}")
    else:
        print("✅ Questions generated successfully")
except Exception as e:
    print(f"❌ Unexpected error: {e}")

Custom Tools Integration¶

from swarms.structs.heavy_swarm import HeavySwarm
from swarms_tools import exa_search, calculator

# Initialize with custom tools
swarm = HeavySwarm(
    name="CustomToolsSwarm",
    description="HeavySwarm with enhanced tool capabilities",
    worker_model_name="gpt-4.1",
    worker_tools=[exa_search, calculator],
    show_dashboard=True,
    timeout=600
)

# Task that benefits from tools
analysis_prompt = """
Conduct competitive intelligence analysis:
1. Research competitor strategies using web search
2. Calculate market share percentages
3. Analyze financial metrics
4. Provide strategic recommendations
"""

result = swarm.run(analysis_prompt)
print(result)

Enterprise Configuration¶

from swarms.structs.heavy_swarm import HeavySwarm

# Enterprise-grade configuration
swarm = HeavySwarm(
    name="EnterpriseAnalysis",
    description="High-performance enterprise analysis",
    worker_model_name="gpt-4.1",  # Highest quality
    question_agent_model_name="gpt-4.1",
    show_dashboard=True,
    loops_per_agent=5,  # Maximum depth
    timeout=1800,  # 30 minutes
    max_workers=16,  # Maximum parallelization
    verbose=True
)

# Enterprise-level analysis
enterprise_prompt = """
Conduct enterprise strategic analysis:
- Executive summary and positioning
- Financial analysis and projections
- Operational assessment
- Market intelligence
- Strategic recommendations with implementation roadmap
"""

result = swarm.run(enterprise_prompt)
print(result)

Random Loops for Diverse Perspectives¶

from swarms.structs.heavy_swarm import HeavySwarm

# Enable random loops for varied analysis depth
swarm = HeavySwarm(
    name="DiverseAnalysis",
    description="Analysis with random iteration depth",
    worker_model_name="gpt-4.1",
    question_agent_model_name="gpt-5.4",
    show_dashboard=True,
    loops_per_agent=1,  # Base loops
    random_loops_per_agent=True,  # Enable randomization (1-10 loops)
    timeout=900
)

# Analysis that benefits from multiple perspectives
prompt = """
Analyze healthcare policy impacts on:
1. Access to care and underserved populations
2. Cost implications and spending trends
3. Quality of care metrics
4. Healthcare workforce dynamics
5. Technology integration
"""

result = swarm.run(prompt)
print(result)

Notes¶

Aspect	Recommendation/Description
Performance	Use `max_workers` based on your CPU cores for optimal parallel execution
Cost	Higher model versions provide better analysis but increase costs. Grok mode uses 4 agents vs 5, slightly reducing cost per run
Timeouts	Complex tasks may require longer `timeout` values
Tools	Integrate domain-specific tools for enhanced analysis capabilities
Dashboard	Enable `show_dashboard=True` for visual progress tracking
Batch Processing	Disable dashboard and verbose logging for efficient batch operations
When to use Grok mode	Best for tasks requiring debate-style analysis where facts, logic, and creative thinking need to be weighed against each other (e.g., strategic decisions, investment analysis, policy evaluation). Captain Swarm's conflict mediation is especially valuable when different perspectives might disagree.
When to use Default mode	Best for systematic, phased analysis where each step builds on the previous one (e.g., research reports, market studies, compliance audits). The dedicated Verification Agent provides stronger validation coverage.

HeavySwarm¶

Installation¶

Quick Start¶

Architecture¶

Default Mode — Task-Phase Agents¶

Grok Mode — Thinking-Style Agents (use_grok_agents=True)¶

Key Differences from Default Mode¶

API Reference¶

HeavySwarm Class¶

Constructor¶

Parameters¶

Raises¶

Example¶

Methods¶

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

get_questions_only(task: str) -> Dict[str, str]¶

get_questions_as_list(task: str) -> List[str]¶

show_swarm_info() -> None¶

reliability_check() -> None¶

create_agents() -> Dict[str, Agent]¶

Examples¶

Grok Mode — Quick Start¶

Grok Mode — Medical Research¶

Grok Mode — Strategic Analysis with Tools¶

Grok Mode via SwarmRouter¶

Grok Mode — Question Preview¶

Basic Usage (Default Mode)¶

Financial Research with Tools¶

Advanced Market Analysis¶

Question Generation Only¶

Batch Processing¶

Error Handling¶

Custom Tools Integration¶

Enterprise Configuration¶

Random Loops for Diverse Perspectives¶

Notes¶

Grok Mode — Thinking-Style Agents (`use_grok_agents=True`)¶

`run(task: str, img: Optional[str] = None) -> str`¶

`get_questions_only(task: str) -> Dict[str, str]`¶

`get_questions_as_list(task: str) -> List[str]`¶

`show_swarm_info() -> None`¶

`reliability_check() -> None`¶

`create_agents() -> Dict[str, Agent]`¶