ai-mirow-projects
/
swarms
mirror of https://github.com/kyegomez/swarms
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[FEAT][BatchedGridWorkflow] [Docs] [Example][Geo-Guesser Agent]

Browse Source
pull/1072/head
Kye Gomez 1 month ago
parent 13df6f6198
commit fdf037b965
13 changed files with 1274 additions and 0 deletions
Show Stats Download Patch File Download Diff File

28
batched_grid_simple_example.py
Unescape View File

@ -0,0 +1,28 @@
from swarms import Agent
from swarms.structs.batched_grid_workflow import BatchedGridWorkflow
# Initialize the ETF-focused agent
agent = Agent(
agent_name="ETF-Research-Agent",
agent_description="Specialized agent for researching, analyzing, and recommending Exchange-Traded Funds (ETFs) across various sectors and markets.",
model_name="claude-sonnet-4-20250514",
dynamic_temperature_enabled=True,
max_loops=1,
dynamic_context_window=True,
)
# Create workflow with default settings
workflow = BatchedGridWorkflow(agents=[agent, agent])
# Define simple tasks
tasks = [
"What are the best GOLD ETFs?",
"What are the best american energy ETFs?",
]
# Run the workflow
result = workflow.run(tasks)
print(result)

4
docs/mkdocs.yml
Unescape View File

@ -315,6 +315,7 @@ nav:
- ConcurrentWorkflow: "swarms/structs/concurrentworkflow.md"
- SequentialWorkflow: "swarms/structs/sequential_workflow.md"
- GraphWorkflow: "swarms/structs/graph_workflow.md"
- BatchedGridWorkflow: "swarms/structs/batched_grid_workflow.md"
- Communication Structure: "swarms/structs/conversation.md"
@ -422,6 +423,9 @@ nav:
- Agents as Tools: "swarms/examples/agents_as_tools.md"
- Aggregate Multi-Agent Responses: "swarms/examples/aggregate.md"
- Interactive GroupChat Example: "swarms/examples/igc_example.md"
- BatchedGridWorkflow Examples:
- Simple BatchedGridWorkflow: "swarms/examples/batched_grid_simple_example.md"
- Advanced BatchedGridWorkflow: "swarms/examples/batched_grid_advanced_example.md"
- Applications:
- Swarms DAO: "swarms/examples/swarms_dao.md"
- Swarms of Browser Agents: "swarms/examples/swarms_of_browser_agents.md"

256
docs/swarms/examples/batched_grid_advanced_example.md
Unescape View File

@ -0,0 +1,256 @@
# Advanced BatchedGridWorkflow Examples
This example demonstrates advanced usage patterns and configurations of the `BatchedGridWorkflow` for complex multi-agent scenarios.
## Custom Conversation Configuration
```python
from swarms import Agent
from swarms.structs.batched_grid_workflow import BatchedGridWorkflow
# Create agents with specific roles
researcher = Agent(
model="gpt-4",
system_prompt="You are a research specialist who conducts thorough investigations."
)
writer = Agent(
model="gpt-4",
system_prompt="You are a technical writer who creates clear, comprehensive documentation."
)
reviewer = Agent(
model="gpt-4",
system_prompt="You are a quality reviewer who ensures accuracy and completeness."
)
# Create workflow with custom conversation settings
workflow = BatchedGridWorkflow(
id="custom-research-workflow",
name="Custom Research Pipeline",
description="Research, writing, and review pipeline with custom conversation tracking",
agents=[researcher, writer, reviewer],
conversation_args={
"message_id_on": True,
"conversation_id": "research-pipeline-001"
},
max_loops=2
)
# Research and documentation tasks
tasks = [
"Research the latest developments in artificial intelligence safety",
"Write comprehensive documentation for a new API endpoint",
"Review and validate the technical specifications document"
]
# Execute with custom configuration
result = workflow.run(tasks)
```
## Iterative Refinement Workflow
```python
# Create refinement agents
initial_creator = Agent(
model="gpt-4",
system_prompt="You are a creative content creator who generates initial ideas and drafts."
)
detail_enhancer = Agent(
model="gpt-4",
system_prompt="You are a detail specialist who adds depth and specificity to content."
)
polish_expert = Agent(
model="gpt-4",
system_prompt="You are a polish expert who refines content for maximum impact and clarity."
)
# Create workflow with multiple refinement loops
workflow = BatchedGridWorkflow(
name="Iterative Content Refinement",
description="Multi-stage content creation with iterative improvement",
agents=[initial_creator, detail_enhancer, polish_expert],
max_loops=4
)
# Content creation tasks
tasks = [
"Create an initial draft for a product launch announcement",
"Add detailed specifications and technical details to the content",
"Polish and refine the content for maximum engagement"
]
# Execute iterative refinement
result = workflow.run(tasks)
```
## Specialized Domain Workflow
```python
# Create domain-specific agents
medical_expert = Agent(
model="gpt-4",
system_prompt="You are a medical expert specializing in diagnostic procedures and treatment protocols."
)
legal_advisor = Agent(
model="gpt-4",
system_prompt="You are a legal advisor specializing in healthcare regulations and compliance."
)
technology_architect = Agent(
model="gpt-4",
system_prompt="You are a technology architect specializing in healthcare IT systems and security."
)
# Create specialized workflow
workflow = BatchedGridWorkflow(
name="Healthcare Technology Assessment",
description="Multi-domain assessment of healthcare technology solutions",
agents=[medical_expert, legal_advisor, technology_architect],
max_loops=1
)
# Domain-specific assessment tasks
tasks = [
"Evaluate the medical efficacy and safety of a new diagnostic AI system",
"Assess the legal compliance and regulatory requirements for the system",
"Analyze the technical architecture and security implications of implementation"
]
# Execute specialized assessment
result = workflow.run(tasks)
```
## Parallel Analysis Workflow
```python
# Create analysis agents
market_analyst = Agent(
model="gpt-4",
system_prompt="You are a market analyst who evaluates business opportunities and market trends."
)
financial_analyst = Agent(
model="gpt-4",
system_prompt="You are a financial analyst who assesses investment potential and financial viability."
)
risk_assessor = Agent(
model="gpt-4",
system_prompt="You are a risk assessor who identifies potential threats and mitigation strategies."
)
# Create parallel analysis workflow
workflow = BatchedGridWorkflow(
name="Comprehensive Business Analysis",
description="Parallel analysis of market, financial, and risk factors",
agents=[market_analyst, financial_analyst, risk_assessor],
max_loops=2
)
# Analysis tasks
tasks = [
"Analyze the market opportunity for a new fintech product",
"Evaluate the financial projections and investment requirements",
"Assess the regulatory and operational risks associated with the venture"
]
# Execute parallel analysis
result = workflow.run(tasks)
```
## Creative Collaboration Workflow
```python
# Create creative agents
visual_artist = Agent(
model="gpt-4",
system_prompt="You are a visual artist who creates compelling imagery and visual concepts."
)
music_composer = Agent(
model="gpt-4",
system_prompt="You are a music composer who creates original compositions and soundscapes."
)
story_writer = Agent(
model="gpt-4",
system_prompt="You are a story writer who crafts engaging narratives and character development."
)
# Create creative collaboration workflow
workflow = BatchedGridWorkflow(
name="Multi-Media Creative Project",
description="Collaborative creation across visual, audio, and narrative elements",
agents=[visual_artist, music_composer, story_writer],
max_loops=3
)
# Creative tasks
tasks = [
"Design visual concepts for a fantasy adventure game",
"Compose background music that enhances the game's atmosphere",
"Write character backstories and dialogue for the main protagonists"
]
# Execute creative collaboration
result = workflow.run(tasks)
```
## Quality Assurance Workflow
```python
# Create QA agents
functional_tester = Agent(
model="gpt-4",
system_prompt="You are a functional tester who validates system behavior and user workflows."
)
security_tester = Agent(
model="gpt-4",
system_prompt="You are a security tester who identifies vulnerabilities and security issues."
)
performance_tester = Agent(
model="gpt-4",
system_prompt="You are a performance tester who evaluates system speed and resource usage."
)
# Create QA workflow
workflow = BatchedGridWorkflow(
name="Comprehensive Quality Assurance",
description="Multi-faceted testing across functional, security, and performance domains",
agents=[functional_tester, security_tester, performance_tester],
max_loops=1
)
# QA tasks
tasks = [
"Test the user registration and authentication workflows",
"Conduct security analysis of the payment processing system",
"Evaluate the performance characteristics under high load conditions"
]
# Execute QA workflow
result = workflow.run(tasks)
```
## Advanced Features Demonstrated
- **Custom Conversation Tracking**: Advanced conversation management with custom IDs
- **Iterative Refinement**: Multiple loops for content improvement
- **Domain Specialization**: Agents with specific expertise areas
- **Parallel Analysis**: Simultaneous evaluation from multiple perspectives
- **Creative Collaboration**: Multi-modal creative content generation
- **Quality Assurance**: Comprehensive testing across multiple domains
## Best Practices
1. **Agent Specialization**: Create agents with specific roles and expertise
2. **Task Alignment**: Ensure tasks match agent capabilities
3. **Loop Configuration**: Use multiple loops for iterative processes
4. **Error Monitoring**: Monitor logs for execution issues
5. **Resource Management**: Consider computational requirements for multiple agents

115
docs/swarms/examples/batched_grid_simple_example.md
Unescape View File

@ -0,0 +1,115 @@
# Simple BatchedGridWorkflow Example
This example demonstrates the basic usage of `BatchedGridWorkflow` with minimal configuration for easy understanding.
## Basic Example
```python
from swarms import Agent
from swarms.structs.batched_grid_workflow import BatchedGridWorkflow
# Create two basic agents
agent1 = Agent(model="gpt-4")
agent2 = Agent(model="gpt-4")
# Create workflow with default settings
workflow = BatchedGridWorkflow(
agents=[agent1, agent2]
)
# Define simple tasks
tasks = [
"What is the capital of France?",
"Explain photosynthesis in simple terms"
]
# Run the workflow
result = workflow.run(tasks)
```
## Named Workflow Example
```python
# Create agents
writer = Agent(model="gpt-4")
analyst = Agent(model="gpt-4")
# Create named workflow
workflow = BatchedGridWorkflow(
name="Content Analysis Workflow",
description="Analyze and write content in parallel",
agents=[writer, analyst]
)
# Content tasks
tasks = [
"Write a short paragraph about renewable energy",
"Analyze the benefits of solar power"
]
# Execute workflow
result = workflow.run(tasks)
```
## Multi-Loop Example
```python
# Create agents
agent1 = Agent(model="gpt-4")
agent2 = Agent(model="gpt-4")
# Create workflow with multiple loops
workflow = BatchedGridWorkflow(
agents=[agent1, agent2],
max_loops=3
)
# Tasks for iterative processing
tasks = [
"Generate ideas for a mobile app",
"Evaluate the feasibility of each idea"
]
# Run with multiple loops
result = workflow.run(tasks)
```
## Three Agent Example
```python
# Create three agents
researcher = Agent(model="gpt-4")
writer = Agent(model="gpt-4")
editor = Agent(model="gpt-4")
# Create workflow
workflow = BatchedGridWorkflow(
name="Research and Writing Pipeline",
agents=[researcher, writer, editor]
)
# Three different tasks
tasks = [
"Research the history of artificial intelligence",
"Write a summary of the research findings",
"Review and edit the summary for clarity"
]
# Execute workflow
result = workflow.run(tasks)
```
## Key Points
- **Simple Setup**: Minimal configuration required for basic usage
- **Parallel Execution**: Tasks run simultaneously across agents
- **Flexible Configuration**: Easy to customize names, descriptions, and loop counts
- **Error Handling**: Built-in error handling and logging
- **Scalable**: Works with any number of agents and tasks
## Use Cases
- **Content Creation**: Multiple writers working on different topics
- **Research Tasks**: Different researchers investigating various aspects
- **Analysis Work**: Multiple analysts processing different datasets
- **Educational Content**: Different instructors creating materials for various subjects

174
docs/swarms/structs/batched_grid_workflow.md
Unescape View File

@ -0,0 +1,174 @@
# BatchedGridWorkflow
The `BatchedGridWorkflow` is a multi-agent orchestration pattern that executes tasks in a batched grid format, where each agent processes a different task simultaneously. This workflow is particularly useful for parallel processing scenarios where you have multiple agents and multiple tasks that can be distributed across them.
## Overview
The BatchedGridWorkflow provides a structured approach to:
- Execute multiple tasks across multiple agents in parallel
- Manage conversation state across execution loops
- Handle error scenarios gracefully
- Control the number of execution iterations
## Architecture
```mermaid
graph TD
A[Input Tasks] --> B[BatchedGridWorkflow]
B --> C[Initialize Agents]
C --> D[Create Conversation]
D --> E[Start Execution Loop]
E --> F[Distribute Tasks to Agents]
F --> G[Agent 1: Task 1]
F --> H[Agent 2: Task 2]
F --> I[Agent N: Task N]
G --> J[Collect Results]
H --> J
I --> J
J --> K[Update Conversation]
K --> L{More Loops?}
L -->|Yes| E
L -->|No| M[Return Final Results]
M --> N[Output]
```
## Key Features
| Feature | Description |
|--------------------------|-----------------------------------------------------------------------------------------------|
| **Parallel Execution** | Multiple agents work on different tasks simultaneously |
| **Conversation Management** | Maintains conversation state across execution loops |
| **Error Handling** | Comprehensive error logging and exception handling |
| **Configurable Loops** | Control the number of execution iterations |
| **Agent Flexibility** | Supports any agent type that implements the `AgentType` interface |
## Class Definition
```python
class BatchedGridWorkflow:
def __init__(
self,
id: str = swarm_id(),
name: str = "BatchedGridWorkflow",
description: str = "For every agent, run the task on a different task",
agents: List[AgentType] = None,
conversation_args: dict = None,
max_loops: int = 1,
):
```
## Parameters
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `id` | str | `swarm_id()` | Unique identifier for the workflow |
| `name` | str | `"BatchedGridWorkflow"` | Name of the workflow |
| `description` | str | `"For every agent, run the task on a different task"` | Description of what the workflow does |
| `agents` | List[AgentType] | `None` | List of agents to execute tasks |
| `conversation_args` | dict | `None` | Arguments for the conversation |
| `max_loops` | int | `1` | Maximum number of execution loops to run (must be >= 1) |
## Methods
### `step(tasks: List[str])`
Execute one step of the batched grid workflow.
**Parameters:**
- `tasks` (List[str]): List of tasks to execute
**Returns:**
- Output from the batched grid agent execution
### `run(tasks: List[str])`
Run the batched grid workflow with the given tasks. This is the main entry point that includes error handling.
**Parameters:**
- `tasks` (List[str]): List of tasks to execute
**Returns:**
- str: The final conversation string after all loops
### `run_(tasks: List[str])`
Internal method that runs the workflow without error handling.
**Parameters:**
- `tasks` (List[str]): List of tasks to execute
**Returns:**
- str: The final conversation string after all loops
## Usage Patterns
### Basic Usage
```python
from swarms.structs.batched_grid_workflow import BatchedGridWorkflow
from swarms import Agent
# Create agents
agent1 = Agent(model="gpt-4", system_prompt="You are a helpful assistant")
agent2 = Agent(model="gpt-4", system_prompt="You are a creative writer")
# Create workflow
workflow = BatchedGridWorkflow(
agents=[agent1, agent2],
max_loops=1
)
# Execute tasks
tasks = ["Write a poem about nature", "Analyze market trends"]
result = workflow.run(tasks)
```
### Multi-Loop Execution
```python
# Create workflow with multiple loops
workflow = BatchedGridWorkflow(
agents=[agent1, agent2, agent3],
max_loops=3,
conversation_args={"message_id_on": True}
)
# Execute tasks with multiple iterations
tasks = ["Task 1", "Task 2", "Task 3"]
result = workflow.run(tasks)
```
## Error Handling
The workflow includes comprehensive error handling:
- **Validation**: Ensures `max_loops` is a positive integer
- **Execution Errors**: Catches and logs exceptions during execution
- **Detailed Logging**: Provides detailed error information including traceback
## Best Practices
| Best Practice | Description |
|----------------------------|-------------------------------------------------------------------------------------------------------|
| **Agent Selection** | Choose agents with complementary capabilities for diverse task processing |
| **Task Distribution** | Ensure tasks are well-distributed and can be processed independently |
| **Loop Configuration** | Use multiple loops when iterative refinement is needed |
| **Error Monitoring** | Monitor logs for execution errors and adjust agent configurations accordingly |
| **Resource Management** | Consider computational resources when setting up multiple agents |
## Use Cases
| Use Case | Description |
|------------------------|-----------------------------------------------------------------------------------------------|
| **Content Generation** | Multiple writers working on different topics |
| **Data Analysis** | Different analysts processing various datasets |
| **Research Tasks** | Multiple researchers investigating different aspects of a problem |
| **Parallel Processing**| Any scenario requiring simultaneous task execution across multiple agents |

44
examples/guides/geo_guesser_agent/README.md
Unescape View File

@ -0,0 +1,44 @@
# Geo Guesser Agent - Quick Setup Guide
This example demonstrates how to create an AI agent that can analyze images and predict their geographical location using visual cues.
## 3-Step Setup Guide
### Step 1: Install Dependencies
```bash
pip install swarms
```
### Step 2: Prepare Your Image
```bash
GEMINI_API_KEY=""
```
### Step 3: Run the Agent
```bash
python geo_guesser_agent.py
```
## What It Does
The agent analyzes visual elements in your image such as:
- Architecture and building styles
- Landscape and terrain features
- Vegetation and plant life
- Weather patterns
- Cultural elements and signs
- Any other geographical indicators
## Expected Output
The agent will provide:
- Most likely location prediction
- Detailed reasoning for the prediction
- Confidence level assessment
## Customization
- Change the `model_name` to use different AI models
- Modify the `SYSTEM_PROMPT` to adjust the agent's behavior
- Adjust `max_loops` for more or fewer analysis iterations

27
examples/guides/geo_guesser_agent/geo_guesser_agent.py
Unescape View File

@ -0,0 +1,27 @@
from swarms import Agent
SYSTEM_PROMPT = (
"You are an expert in image geolocalization. Given an image, provide the most likely location it was taken. "
"Analyze visual cues such as architecture, landscape, vegetation, weather patterns, cultural elements, "
"and any other geographical indicators to determine the precise location. Provide your reasoning and "
"confidence level for your prediction."
)
# Agent for image geolocalization
agent = Agent(
agent_name="Geo-Guesser-Agent",
agent_description="Expert agent specialized in image geolocalization, capable of identifying geographical locations from visual cues in images.",
model_name="gemini/gemini-2.5-flash-image-preview",
dynamic_temperature_enabled=True,
max_loops=1,
dynamic_context_window=True,
retry_interval=1,
)
out = agent.run(
task=f"{SYSTEM_PROMPT}",
img="miami.jpg",
)
print(out)

BIN
examples/guides/geo_guesser_agent/miami.jpg
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 151 KiB

56
examples/multi_agent/batched_grid_workflow/README.md
Unescape View File

@ -0,0 +1,56 @@
# BatchedGridWorkflow Examples
This directory contains examples demonstrating various usage patterns of the `BatchedGridWorkflow` for parallel multi-agent task execution.
## Examples
### 1. `batched_grid_simple_example.py`
Basic usage examples with minimal configuration:
- Simple two-agent setup
- Named workflows with custom descriptions
- Multi-loop execution
- Three-agent workflows
- Error handling demonstration
### 2. `batched_grid_workflow_example.py`
Comprehensive examples covering different use cases:
- Content creation workflows
- Research and analysis tasks
- Problem-solving scenarios
- Educational content generation
- Multi-domain assessments
### 3. `batched_grid_advanced_example.py`
Advanced configuration examples:
- Custom conversation settings
- Iterative refinement workflows
- Specialized domain expertise
- Parallel analysis scenarios
- Creative collaboration
- Quality assurance workflows
## Usage
Each example can be run independently:
```bash
python batched_grid_simple_example.py
python batched_grid_workflow_example.py
python batched_grid_advanced_example.py
```
## Key Features Demonstrated
- **Parallel Execution**: Multiple agents working on different tasks simultaneously
- **Conversation Management**: Maintaining state across execution loops
- **Error Handling**: Built-in error logging and exception handling
- **Flexible Configuration**: Custom names, descriptions, and loop counts
- **Agent Specialization**: Different agents with specific roles and expertise
- **Iterative Refinement**: Multiple loops for content improvement
## Documentation
For complete documentation, see:
- [BatchedGridWorkflow Reference](../../../docs/swarms/structs/batched_grid_workflow.md)
- [Simple Examples Documentation](../../../docs/swarms/examples/batched_grid_simple_example.md)
- [Advanced Examples Documentation](../../../docs/swarms/examples/batched_grid_advanced_example.md)

267
examples/multi_agent/batched_grid_workflow/batched_grid_advanced_example.py
Unescape View File

@ -0,0 +1,267 @@
"""
Advanced BatchedGridWorkflow Examples
This module demonstrates advanced usage patterns and configurations
of the BatchedGridWorkflow for complex multi-agent scenarios.
"""
from swarms import Agent
from swarms.structs.batched_grid_workflow import BatchedGridWorkflow
def custom_conversation_workflow():
"""Example with custom conversation configuration."""
# Create agents with specific roles
researcher = Agent(
model="gpt-4",
system_prompt="You are a research specialist who conducts thorough investigations.",
)
writer = Agent(
model="gpt-4",
system_prompt="You are a technical writer who creates clear, comprehensive documentation.",
)
reviewer = Agent(
model="gpt-4",
system_prompt="You are a quality reviewer who ensures accuracy and completeness.",
)
# Create workflow with custom conversation settings
workflow = BatchedGridWorkflow(
id="custom-research-workflow",
name="Custom Research Pipeline",
description="Research, writing, and review pipeline with custom conversation tracking",
agents=[researcher, writer, reviewer],
conversation_args={
"message_id_on": True,
"conversation_id": "research-pipeline-001",
},
max_loops=2,
)
# Research and documentation tasks
tasks = [
"Research the latest developments in artificial intelligence safety",
"Write comprehensive documentation for a new API endpoint",
"Review and validate the technical specifications document",
]
# Execute with custom configuration
result = workflow.run(tasks)
return result
def iterative_refinement_workflow():
"""Example showing iterative refinement across multiple loops."""
# Create refinement agents
initial_creator = Agent(
model="gpt-4",
system_prompt="You are a creative content creator who generates initial ideas and drafts.",
)
detail_enhancer = Agent(
model="gpt-4",
system_prompt="You are a detail specialist who adds depth and specificity to content.",
)
polish_expert = Agent(
model="gpt-4",
system_prompt="You are a polish expert who refines content for maximum impact and clarity.",
)
# Create workflow with multiple refinement loops
workflow = BatchedGridWorkflow(
name="Iterative Content Refinement",
description="Multi-stage content creation with iterative improvement",
agents=[initial_creator, detail_enhancer, polish_expert],
max_loops=4,
)
# Content creation tasks
tasks = [
"Create an initial draft for a product launch announcement",
"Add detailed specifications and technical details to the content",
"Polish and refine the content for maximum engagement",
]
# Execute iterative refinement
result = workflow.run(tasks)
return result
def specialized_domain_workflow():
"""Example with agents specialized in different domains."""
# Create domain-specific agents
medical_expert = Agent(
model="gpt-4",
system_prompt="You are a medical expert specializing in diagnostic procedures and treatment protocols.",
)
legal_advisor = Agent(
model="gpt-4",
system_prompt="You are a legal advisor specializing in healthcare regulations and compliance.",
)
technology_architect = Agent(
model="gpt-4",
system_prompt="You are a technology architect specializing in healthcare IT systems and security.",
)
# Create specialized workflow
workflow = BatchedGridWorkflow(
name="Healthcare Technology Assessment",
description="Multi-domain assessment of healthcare technology solutions",
agents=[medical_expert, legal_advisor, technology_architect],
max_loops=1,
)
# Domain-specific assessment tasks
tasks = [
"Evaluate the medical efficacy and safety of a new diagnostic AI system",
"Assess the legal compliance and regulatory requirements for the system",
"Analyze the technical architecture and security implications of implementation",
]
# Execute specialized assessment
result = workflow.run(tasks)
return result
def parallel_analysis_workflow():
"""Example for parallel analysis of different aspects of a problem."""
# Create analysis agents
market_analyst = Agent(
model="gpt-4",
system_prompt="You are a market analyst who evaluates business opportunities and market trends.",
)
financial_analyst = Agent(
model="gpt-4",
system_prompt="You are a financial analyst who assesses investment potential and financial viability.",
)
risk_assessor = Agent(
model="gpt-4",
system_prompt="You are a risk assessor who identifies potential threats and mitigation strategies.",
)
# Create parallel analysis workflow
workflow = BatchedGridWorkflow(
name="Comprehensive Business Analysis",
description="Parallel analysis of market, financial, and risk factors",
agents=[market_analyst, financial_analyst, risk_assessor],
max_loops=2,
)
# Analysis tasks
tasks = [
"Analyze the market opportunity for a new fintech product",
"Evaluate the financial projections and investment requirements",
"Assess the regulatory and operational risks associated with the venture",
]
# Execute parallel analysis
result = workflow.run(tasks)
return result
def creative_collaboration_workflow():
"""Example for creative collaboration across different artistic domains."""
# Create creative agents
visual_artist = Agent(
model="gpt-4",
system_prompt="You are a visual artist who creates compelling imagery and visual concepts.",
)
music_composer = Agent(
model="gpt-4",
system_prompt="You are a music composer who creates original compositions and soundscapes.",
)
story_writer = Agent(
model="gpt-4",
system_prompt="You are a story writer who crafts engaging narratives and character development.",
)
# Create creative collaboration workflow
workflow = BatchedGridWorkflow(
name="Multi-Media Creative Project",
description="Collaborative creation across visual, audio, and narrative elements",
agents=[visual_artist, music_composer, story_writer],
max_loops=3,
)
# Creative tasks
tasks = [
"Design visual concepts for a fantasy adventure game",
"Compose background music that enhances the game's atmosphere",
"Write character backstories and dialogue for the main protagonists",
]
# Execute creative collaboration
result = workflow.run(tasks)
return result
def quality_assurance_workflow():
"""Example for quality assurance across different testing domains."""
# Create QA agents
functional_tester = Agent(
model="gpt-4",
system_prompt="You are a functional tester who validates system behavior and user workflows.",
)
security_tester = Agent(
model="gpt-4",
system_prompt="You are a security tester who identifies vulnerabilities and security issues.",
)
performance_tester = Agent(
model="gpt-4",
system_prompt="You are a performance tester who evaluates system speed and resource usage.",
)
# Create QA workflow
workflow = BatchedGridWorkflow(
name="Comprehensive Quality Assurance",
description="Multi-faceted testing across functional, security, and performance domains",
agents=[
functional_tester,
security_tester,
performance_tester,
],
max_loops=1,
)
# QA tasks
tasks = [
"Test the user registration and authentication workflows",
"Conduct security analysis of the payment processing system",
"Evaluate the performance characteristics under high load conditions",
]
# Execute QA workflow
result = workflow.run(tasks)
return result
# Advanced configuration examples
if __name__ == "__main__":
# Run custom conversation example
custom_result = custom_conversation_workflow()
# Run iterative refinement example
refinement_result = iterative_refinement_workflow()
# Run specialized domain example
domain_result = specialized_domain_workflow()
# Run parallel analysis example
analysis_result = parallel_analysis_workflow()
# Run creative collaboration example
creative_result = creative_collaboration_workflow()
# Run quality assurance example
qa_result = quality_assurance_workflow()

216
examples/multi_agent/batched_grid_workflow/batched_grid_workflow_example.py
Unescape View File

@ -0,0 +1,216 @@
"""
BatchedGridWorkflow Examples
This module demonstrates various usage patterns of the BatchedGridWorkflow
for parallel task execution across multiple agents.
"""
from swarms import Agent
from swarms.structs.batched_grid_workflow import BatchedGridWorkflow
def basic_batched_workflow():
"""Basic example of BatchedGridWorkflow with two agents."""
# Create specialized agents
writer_agent = Agent(
model="gpt-4",
system_prompt="You are a creative writer who specializes in storytelling and creative content.",
)
analyst_agent = Agent(
model="gpt-4",
system_prompt="You are a data analyst who specializes in research and analysis.",
)
# Create the workflow
workflow = BatchedGridWorkflow(
name="Content Creation Workflow",
description="Parallel content creation and analysis",
agents=[writer_agent, analyst_agent],
max_loops=1,
)
# Define tasks for each agent
tasks = [
"Write a short story about a robot learning to paint",
"Analyze the impact of AI on creative industries",
]
# Execute the workflow
result = workflow.run(tasks)
return result
def multi_agent_research_workflow():
"""Example with multiple agents for research tasks."""
# Create research agents with different specializations
tech_agent = Agent(
model="gpt-4",
system_prompt="You are a technology researcher focused on emerging tech trends.",
)
business_agent = Agent(
model="gpt-4",
system_prompt="You are a business analyst specializing in market research and strategy.",
)
science_agent = Agent(
model="gpt-4",
system_prompt="You are a scientific researcher focused on breakthrough discoveries.",
)
# Create workflow with conversation tracking
workflow = BatchedGridWorkflow(
name="Multi-Domain Research",
description="Parallel research across technology, business, and science",
agents=[tech_agent, business_agent, science_agent],
conversation_args={"message_id_on": True},
max_loops=2,
)
# Research tasks
tasks = [
"Research the latest developments in quantum computing",
"Analyze market trends in renewable energy sector",
"Investigate recent breakthroughs in gene therapy",
]
# Execute research workflow
result = workflow.run(tasks)
return result
def content_creation_workflow():
"""Example for content creation with iterative refinement."""
# Create content creation agents
blog_writer = Agent(
model="gpt-4",
system_prompt="You are a professional blog writer who creates engaging, informative content.",
)
social_media_manager = Agent(
model="gpt-4",
system_prompt="You are a social media expert who creates viral content and captions.",
)
editor_agent = Agent(
model="gpt-4",
system_prompt="You are an editor who reviews and improves content for clarity and impact.",
)
# Create workflow with multiple loops for refinement
workflow = BatchedGridWorkflow(
name="Content Creation Pipeline",
description="Multi-stage content creation with review and refinement",
agents=[blog_writer, social_media_manager, editor_agent],
max_loops=3,
)
# Content creation tasks
tasks = [
"Write a comprehensive blog post about sustainable living",
"Create engaging social media posts for a tech startup",
"Review and improve the following content for maximum impact",
]
# Execute content creation workflow
result = workflow.run(tasks)
return result
def problem_solving_workflow():
"""Example for collaborative problem solving."""
# Create problem-solving agents
technical_expert = Agent(
model="gpt-4",
system_prompt="You are a technical expert who analyzes problems from an engineering perspective.",
)
user_experience_designer = Agent(
model="gpt-4",
system_prompt="You are a UX designer who focuses on user needs and experience.",
)
business_strategist = Agent(
model="gpt-4",
system_prompt="You are a business strategist who considers market viability and business impact.",
)
# Create collaborative workflow
workflow = BatchedGridWorkflow(
name="Collaborative Problem Solving",
description="Multi-perspective problem analysis and solution development",
agents=[
technical_expert,
user_experience_designer,
business_strategist,
],
max_loops=1,
)
# Problem-solving tasks
tasks = [
"Analyze the technical feasibility of implementing a new feature",
"Evaluate the user experience impact of the proposed changes",
"Assess the business value and market potential of the solution",
]
# Execute problem-solving workflow
result = workflow.run(tasks)
return result
def educational_workflow():
"""Example for educational content creation."""
# Create educational agents
math_tutor = Agent(
model="gpt-4",
system_prompt="You are a mathematics tutor who explains concepts clearly and provides examples.",
)
science_teacher = Agent(
model="gpt-4",
system_prompt="You are a science teacher who makes complex topics accessible and engaging.",
)
language_instructor = Agent(
model="gpt-4",
system_prompt="You are a language instructor who helps students improve their communication skills.",
)
# Create educational workflow
workflow = BatchedGridWorkflow(
name="Educational Content Creation",
description="Parallel creation of educational materials across subjects",
agents=[math_tutor, science_teacher, language_instructor],
max_loops=1,
)
# Educational tasks
tasks = [
"Create a lesson plan for teaching calculus to high school students",
"Develop an experiment to demonstrate photosynthesis",
"Design a writing exercise to improve essay composition skills",
]
# Execute educational workflow
result = workflow.run(tasks)
return result
# Example usage patterns
if __name__ == "__main__":
# Run basic example
basic_result = basic_batched_workflow()
# Run research example
research_result = multi_agent_research_workflow()
# Run content creation example
content_result = content_creation_workflow()
# Run problem solving example
problem_result = problem_solving_workflow()
# Run educational example
education_result = educational_workflow()

2
swarms/structs/__init__.py
Unescape View File

@ -100,6 +100,7 @@ from swarms.structs.swarming_architectures import (
staircase_swarm,
star_swarm,
)
from swarms.structs.batched_grid_workflow import BatchedGridWorkflow
__all__ = [
"Agent",
@ -184,4 +185,5 @@ __all__ = [
"check_exit",
"check_end",
"AgentLoader",
"BatchedGridWorkflow",
]

85
swarms/structs/batched_grid_workflow.py
Unescape View File

@ -0,0 +1,85 @@
import traceback
from typing import List
from loguru import logger
from swarms.structs.multi_agent_exec import (
batched_grid_agent_execution,
)
from swarms.structs.omni_agent_types import AgentType
from swarms.structs.swarm_id import swarm_id
class BatchedGridWorkflow:
def __init__(
self,
id: str = swarm_id(),
name: str = "BatchedGridWorkflow",
description: str = "For every agent, run the task on a different task",
agents: List[AgentType] = None,
max_loops: int = 1,
):
"""
Initialize a BatchedGridWorkflow instance.
Args:
id: Unique identifier for the workflow
name: Name of the workflow
description: Description of what the workflow does
agents: List of agents to execute tasks
max_loops: Maximum number of execution loops to run (must be >= 1)
"""
self.id = id
self.name = name
self.description = description
self.agents = agents
self.max_loops = max_loops
# Validate max_loops parameter
if not isinstance(max_loops, int) or max_loops < 1:
raise ValueError("max_loops must be a positive integer")
def step(self, tasks: List[str]):
"""
Execute one step of the batched grid workflow.
Args:
tasks: List of tasks to execute
Returns:
Output from the batched grid agent execution
"""
return batched_grid_agent_execution(self.agents, tasks)
def run_(self, tasks: List[str]):
"""
Run the batched grid workflow with the given tasks.
Args:
tasks: List of tasks to execute
Returns:
List: Results from all execution loops
"""
results = []
current_loop = 0
while current_loop < self.max_loops:
# Run the step with the original tasks
output = self.step(tasks)
results.append(output)
current_loop += 1
return results
def run(self, tasks: List[str]):
"""
Run the batched grid workflow with the given tasks.
"""
try:
return self.run_(tasks)
except Exception as e:
logger.error(
f"BatchedGridWorkflow Error: {self.name}\n\nId: {self.id}\n\nAn error occurred while running the batched grid workflow: {e}\nTraceback:\n{traceback.format_exc()}"
)
raise e
Write Preview
Loading…
Cancel
Save