You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
558 lines
17 KiB
558 lines
17 KiB
# HeavySwarm
|
|
|
|
HeavySwarm is a sophisticated multi-agent orchestration system that decomposes complex tasks into specialized questions and executes them using four specialized agents: Research, Analysis, Alternatives, and Verification. The results are then synthesized into a comprehensive response.
|
|
|
|
Inspired by X.AI's Grok 4 heavy implementation, HeavySwarm provides robust task analysis through intelligent question generation, parallel execution, and comprehensive synthesis with real-time progress monitoring.
|
|
|
|
## Installation
|
|
|
|
```bash
|
|
pip install swarms
|
|
```
|
|
|
|
## Quick Start
|
|
|
|
```python
|
|
from swarms.structs.heavy_swarm import HeavySwarm
|
|
|
|
# Initialize HeavySwarm
|
|
swarm = HeavySwarm(
|
|
name="Research Team",
|
|
description="Multi-agent analysis system",
|
|
worker_model_name="gpt-4o-mini",
|
|
show_dashboard=True
|
|
)
|
|
|
|
# Run analysis
|
|
result = swarm.run("Analyze the impact of AI on healthcare")
|
|
print(result)
|
|
```
|
|
|
|
## Architecture
|
|
|
|
The HeavySwarm follows a structured 5-phase workflow:
|
|
|
|
1. **Task Decomposition**: Complex tasks are broken down into specialized questions
|
|
2. **Question Generation**: AI-powered generation of role-specific questions
|
|
3. **Parallel Execution**: Four specialized agents work concurrently
|
|
4. **Result Collection**: Outputs are gathered and validated
|
|
5. **Synthesis**: Integration into a comprehensive final response
|
|
|
|
### Agent Specialization
|
|
|
|
| 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 |
|
|
|
|
## API Reference
|
|
|
|
### HeavySwarm Class
|
|
|
|
A sophisticated multi-agent orchestration system for complex task analysis.
|
|
|
|
#### Constructor
|
|
|
|
```python
|
|
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-4o-mini",
|
|
worker_model_name: str = "gpt-4o-mini",
|
|
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,
|
|
)
|
|
```
|
|
|
|
##### 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-4o-mini"` | Language model for question generation |
|
|
| `worker_model_name` | `str` | `"gpt-4o-mini"` | 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 loops when using random_loops_per_agent |
|
|
|
|
##### 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
|
|
|
|
```python
|
|
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:**
|
|
|
|
```python
|
|
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:
|
|
- `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
|
|
- `error` (str): Error message if question generation fails (on error only)
|
|
|
|
**Example:**
|
|
|
|
```python
|
|
questions = swarm.get_questions_only("Analyze market trends for EVs")
|
|
print(questions['research_question'])
|
|
print(questions['analysis_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 4 specialized 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
|
|
- Single-item list containing error message (on error)
|
|
|
|
**Example:**
|
|
|
|
```python
|
|
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:**
|
|
|
|
```python
|
|
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:**
|
|
|
|
```python
|
|
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 the 4 specialized agents with detailed role-specific prompts.
|
|
|
|
**Parameters:**
|
|
|
|
- None
|
|
|
|
**Returns:**
|
|
|
|
| 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 |
|
|
|
|
**Example:**
|
|
|
|
```python
|
|
agents = swarm.create_agents()
|
|
research_agent = agents['research']
|
|
print(f"Research agent name: {research_agent.agent_name}")
|
|
```
|
|
|
|
## Examples
|
|
|
|
### Basic Usage
|
|
|
|
```python
|
|
from swarms.structs.heavy_swarm import HeavySwarm
|
|
|
|
# Initialize HeavySwarm
|
|
swarm = HeavySwarm(
|
|
name="Analysis Team",
|
|
description="Multi-agent analysis system",
|
|
worker_model_name="gpt-4o-mini",
|
|
show_dashboard=True
|
|
)
|
|
|
|
# Run analysis
|
|
result = swarm.run("Analyze the impact of AI on healthcare")
|
|
print(result)
|
|
```
|
|
|
|
### Financial Research with Tools
|
|
|
|
```python
|
|
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-4o-mini",
|
|
question_agent_model_name="gpt-4o-mini",
|
|
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
|
|
|
|
```python
|
|
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
|
|
|
|
```python
|
|
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-4o-mini",
|
|
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
|
|
|
|
```python
|
|
from swarms.structs.heavy_swarm import HeavySwarm
|
|
|
|
# Configure for batch processing
|
|
swarm = HeavySwarm(
|
|
name="BatchProcessor",
|
|
worker_model_name="gpt-4o-mini",
|
|
question_agent_model_name="gpt-4o-mini",
|
|
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
|
|
|
|
```python
|
|
from swarms.structs.heavy_swarm import HeavySwarm
|
|
|
|
# Test configuration validation
|
|
try:
|
|
swarm = HeavySwarm(
|
|
name="TestSwarm",
|
|
worker_model_name="gpt-4o-mini",
|
|
question_agent_model_name="gpt-4o-mini"
|
|
)
|
|
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-4o-mini",
|
|
question_agent_model_name="gpt-4o-mini"
|
|
)
|
|
|
|
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
|
|
|
|
```python
|
|
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
|
|
|
|
```python
|
|
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
|
|
|
|
```python
|
|
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-4o-mini",
|
|
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 (`gpt-4o`) provide better analysis but increase costs |
|
|
| **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 |
|