Merge 8f6c0edb64 into a762c79d6e
CI-DEV
1 week ago
committed by
GitHub
commit
07bd21c26e
@ -0,0 +1,888 @@
|
||||
# ElectionSwarm
|
||||
|
||||
The `ElectionSwarm` is a sophisticated multi-agent orchestrator selection system that implements AGENTSNET communication protocols for democratic decision-making. It enables agent voters to elect the best orchestrator candidate through various coordination algorithms, providing a structured approach to leadership selection in multi-agent workflows.
|
||||
|
||||
## Imports!
|
||||
|
||||
```python
|
||||
from swarms.structs import ElectionSwarm
|
||||
# or
|
||||
from swarms.structs.election_swarm import ElectionSwarm
|
||||
```
|
||||
|
||||
## Overview
|
||||
|
||||
The ElectionSwarm follows a democratic election workflow pattern:
|
||||
|
||||
1. **Task Reception**: User provides an election task to the swarm
|
||||
2. **Voter Registration**: Agent voters are registered with their preferences and expertise
|
||||
3. **Candidate Presentation**: Orchestrator candidates present their platforms and capabilities
|
||||
4. **Message Passing**: Voters communicate with neighbors using AGENTSNET protocols
|
||||
5. **Voting Process**: Voters cast structured votes with detailed reasoning
|
||||
6. **Result Determination**: Election algorithms determine the winning orchestrator
|
||||
7. **Context Preservation**: All conversation history and voting records are maintained
|
||||
|
||||
## Architecture
|
||||
|
||||
```mermaid
|
||||
graph TD
|
||||
A[User Task] --> B[ElectionSwarm]
|
||||
B --> C[Voter Registration]
|
||||
B --> D[Candidate Registration]
|
||||
C --> E[Agent Voters]
|
||||
D --> F[Orchestrator Candidates]
|
||||
E --> G[Message Passing Protocol]
|
||||
F --> G
|
||||
G --> H[AGENTSNET Communication]
|
||||
H --> I[Voting Process]
|
||||
I --> J[Election Algorithms]
|
||||
J --> K[Consensus Building]
|
||||
J --> L[Leader Election]
|
||||
J --> M[Matching Algorithm]
|
||||
J --> N[Coloring Algorithm]
|
||||
J --> O[Vertex Cover]
|
||||
K --> P[Election Result]
|
||||
L --> P
|
||||
M --> P
|
||||
N --> P
|
||||
O --> P
|
||||
P --> Q[Selected Orchestrator]
|
||||
```
|
||||
|
||||
## Key Features
|
||||
|
||||
| Feature | Description |
|
||||
|---------|-------------|
|
||||
| **AGENTSNET Protocols** | Implements distributed computing algorithms for coordination |
|
||||
| **Message Passing** | Synchronous neighbor-to-neighbor communication in rounds |
|
||||
| **Structured Voting** | Voters provide detailed reasoning, confidence, and candidate rankings |
|
||||
| **Multiple Algorithms** | Consensus, Leader Election, Matching, Coloring, Vertex Cover |
|
||||
| **Cost Tracking** | Budget management and token usage monitoring |
|
||||
| **Caching System** | Response caching to avoid redundant LLM calls |
|
||||
| **Lazy Loading** | Agents instantiated only when needed |
|
||||
| **Parallel Execution** | Concurrent agent operations for efficiency |
|
||||
| **Comprehensive Logging** | Detailed logging for debugging and monitoring |
|
||||
| **Flexible Configuration** | Configurable election parameters and algorithms |
|
||||
| **Voter Tool Calls** | Optional tool calls for detailed voting explanations |
|
||||
|
||||
## ElectionSwarm Constructor
|
||||
|
||||
| Parameter | Type | Default | Description |
|
||||
|-----------|------|---------|-------------|
|
||||
| `name` | `str` | `"ElectionSwarm"` | Name of the election swarm |
|
||||
| `description` | `str` | `"Orchestrator selection with AGENTSNET algorithms"` | Description of the swarm purpose |
|
||||
| `voters` | `List[VoterProfile]` | `None` | List of agent voters (defaults created if None) |
|
||||
| `candidates` | `List[CandidateProfile]` | `None` | List of orchestrator candidates (defaults created if None) |
|
||||
| `election_config` | `ElectionConfig` | `None` | Election configuration settings |
|
||||
| `max_loops` | `int` | `1` | Maximum number of election loops |
|
||||
| `output_type` | `OutputType` | `"dict-all-except-first"` | Output format type |
|
||||
| `verbose` | `bool` | `False` | Enable verbose logging |
|
||||
| `enable_lazy_loading` | `bool` | `True` | Enable lazy agent loading |
|
||||
| `enable_caching` | `bool` | `True` | Enable response caching |
|
||||
| `batch_size` | `int` | `25` | Batch size for parallel operations |
|
||||
| `budget_limit` | `float` | `200.0` | Budget limit for cost tracking |
|
||||
| `*args` | `Any` | - | Additional positional arguments |
|
||||
| `**kwargs` | `Any` | - | Additional keyword arguments |
|
||||
|
||||
## Core Components
|
||||
|
||||
### VoterProfile
|
||||
|
||||
Represents an agent voter in the orchestrator selection system.
|
||||
|
||||
```python
|
||||
@dataclass
|
||||
class VoterProfile:
|
||||
voter_id: str
|
||||
name: str
|
||||
voter_type: VoterType
|
||||
preferences: Dict[str, Any] = field(default_factory=dict)
|
||||
expertise_areas: List[str] = field(default_factory=list)
|
||||
voting_weight: float = 1.0
|
||||
agent: Optional[Agent] = None
|
||||
is_loaded: bool = False
|
||||
demographics: Dict[str, Any] = field(default_factory=dict)
|
||||
past_voting_history: List[Dict[str, Any]] = field(default_factory=list)
|
||||
neighbors: List[str] = field(default_factory=list)
|
||||
coordination_style: str = "collaborative"
|
||||
leadership_preferences: Dict[str, float] = field(default_factory=dict)
|
||||
```
|
||||
|
||||
### VoterType
|
||||
|
||||
Enumeration of voter types in the election system.
|
||||
|
||||
```python
|
||||
class VoterType(str, Enum):
|
||||
INDIVIDUAL = "individual"
|
||||
GROUP = "group"
|
||||
EXPERT = "expert"
|
||||
DELEGATE = "delegate"
|
||||
```
|
||||
|
||||
### CandidateProfile
|
||||
|
||||
Represents an orchestrator candidate in the selection system.
|
||||
|
||||
```python
|
||||
@dataclass
|
||||
class CandidateProfile:
|
||||
candidate_id: str
|
||||
name: str
|
||||
candidate_type: CandidateType
|
||||
party_affiliation: Optional[str] = None
|
||||
policy_positions: Dict[str, Any] = field(default_factory=dict)
|
||||
campaign_promises: List[str] = field(default_factory=list)
|
||||
experience: List[str] = field(default_factory=list)
|
||||
agent: Optional[Agent] = None
|
||||
is_loaded: bool = False
|
||||
support_base: Dict[str, float] = field(default_factory=dict)
|
||||
campaign_strategy: Dict[str, Any] = field(default_factory=dict)
|
||||
leadership_style: str = "collaborative"
|
||||
coordination_approach: Dict[str, Any] = field(default_factory=dict)
|
||||
technical_expertise: List[str] = field(default_factory=list)
|
||||
```
|
||||
|
||||
### CandidateType
|
||||
|
||||
Enumeration of candidate types in the election system.
|
||||
|
||||
```python
|
||||
class CandidateType(str, Enum):
|
||||
INDIVIDUAL = "individual"
|
||||
COALITION = "coalition"
|
||||
PARTY = "party"
|
||||
MOVEMENT = "movement"
|
||||
```
|
||||
|
||||
### ElectionAlgorithm
|
||||
|
||||
AGENTSNET algorithms for coordination.
|
||||
|
||||
```python
|
||||
class ElectionAlgorithm(str, Enum):
|
||||
CONSENSUS = "consensus"
|
||||
LEADER_ELECTION = "leader_election"
|
||||
MATCHING = "matching"
|
||||
COLORING = "coloring"
|
||||
VERTEX_COVER = "vertex_cover"
|
||||
```
|
||||
|
||||
### VoterDecision
|
||||
|
||||
Structured output from voter agents.
|
||||
|
||||
```python
|
||||
@dataclass
|
||||
class VoterDecision:
|
||||
voter_id: str
|
||||
rationality: str
|
||||
vote: VoteResult
|
||||
confidence: float = 0.0
|
||||
reasoning_factors: List[str] = field(default_factory=list)
|
||||
candidate_rankings: Dict[str, int] = field(default_factory=dict)
|
||||
tool_call_explanation: Optional[str] = None
|
||||
timestamp: datetime = field(default_factory=datetime.now)
|
||||
```
|
||||
|
||||
### VoteResult
|
||||
|
||||
Possible vote results.
|
||||
|
||||
```python
|
||||
class VoteResult(str, Enum):
|
||||
FOR = "for"
|
||||
AGAINST = "against"
|
||||
ABSTAIN = "abstain"
|
||||
INVALID = "invalid"
|
||||
```
|
||||
|
||||
### ElectionResult
|
||||
|
||||
Results of an election.
|
||||
|
||||
```python
|
||||
@dataclass
|
||||
class ElectionResult:
|
||||
election_id: str
|
||||
algorithm_used: ElectionAlgorithm
|
||||
total_voters: int
|
||||
total_candidates: int
|
||||
votes_cast: int
|
||||
winner: Optional[str] = None
|
||||
vote_distribution: Dict[str, int] = field(default_factory=dict)
|
||||
voter_decisions: List[VoterDecision] = field(default_factory=list)
|
||||
consensus_reached: bool = False
|
||||
rounds_to_consensus: int = 0
|
||||
timestamp: datetime = field(default_factory=datetime.now)
|
||||
```
|
||||
|
||||
### MessagePassingProtocol
|
||||
|
||||
AGENTSNET-inspired message-passing protocol.
|
||||
|
||||
```python
|
||||
class MessagePassingProtocol:
|
||||
def __init__(self, rounds: int = 5, synchronous: bool = True):
|
||||
self.rounds = rounds
|
||||
self.synchronous = synchronous
|
||||
self.current_round = 0
|
||||
self.message_history: List[Dict[str, Any]] = []
|
||||
```
|
||||
|
||||
### CostTracker
|
||||
|
||||
Track costs and usage for budget management.
|
||||
|
||||
```python
|
||||
@dataclass
|
||||
class CostTracker:
|
||||
total_tokens_used: int = 0
|
||||
total_cost_estimate: float = 0.0
|
||||
budget_limit: float = DEFAULT_BUDGET_LIMIT
|
||||
token_cost_per_1m: float = 0.15
|
||||
requests_made: int = 0
|
||||
cache_hits: int = 0
|
||||
```
|
||||
|
||||
## Voter Tool Calls
|
||||
|
||||
The ElectionSwarm supports optional tool calls that allow voter agents to provide detailed explanations of their voting decisions. When enabled, each voter agent will make a tool call to `explain_voting_decision` before casting their vote.
|
||||
|
||||
### Tool Call Features
|
||||
|
||||
- **Detailed Explanations**: Voters provide comprehensive reasoning for their choices
|
||||
- **Structured Data**: Tool calls include voter name, ID, chosen candidate, and reasoning
|
||||
- **Confidence Levels**: Voters specify their confidence in their decision
|
||||
- **Alternative Considerations**: Voters explain why other candidates were not chosen
|
||||
- **Key Factors**: Voters list the most important factors influencing their decision
|
||||
|
||||
### Tool Call Structure
|
||||
|
||||
```python
|
||||
{
|
||||
"voter_name": "Alice Johnson",
|
||||
"voter_id": "voter_001",
|
||||
"chosen_candidate": "John Progressive",
|
||||
"voting_reasoning": "Detailed explanation of why this candidate was chosen",
|
||||
"key_factors": ["healthcare policy", "environmental stance", "experience"],
|
||||
"confidence_level": 0.85,
|
||||
"alternative_considerations": "Considered Sarah Conservative but disagreed on healthcare"
|
||||
}
|
||||
```
|
||||
|
||||
### Enabling Tool Calls
|
||||
|
||||
```python
|
||||
# Enable voter tool calls in configuration
|
||||
config = ElectionConfig(
|
||||
config_data={
|
||||
"enable_voter_tool_calls": True,
|
||||
# ... other config options
|
||||
}
|
||||
)
|
||||
|
||||
election = ElectionSwarm(
|
||||
election_config=config,
|
||||
# ... other parameters
|
||||
)
|
||||
```
|
||||
|
||||
## AGENTSNET Communication Protocols
|
||||
|
||||
### Message Passing Protocol
|
||||
|
||||
The ElectionSwarm implements AGENTSNET-inspired message-passing for distributed coordination:
|
||||
|
||||
- **Synchronous Communication**: Agents communicate in discrete rounds
|
||||
- **Neighbor-Only Messaging**: Agents can only communicate with immediate neighbors
|
||||
- **JSON Message Format**: Structured message exchange
|
||||
- **Multi-Round Process**: Multiple rounds for information propagation
|
||||
- **Final Decision**: Agents make final decisions after communication rounds
|
||||
|
||||
### Communication Flow
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant V1 as Voter 1
|
||||
participant V2 as Voter 2
|
||||
participant V3 as Voter 3
|
||||
participant ES as ElectionSwarm
|
||||
|
||||
Note over ES: Round 1: Initial Communication
|
||||
V1->>V2: Share candidate preferences
|
||||
V2->>V1: Share candidate preferences
|
||||
V2->>V3: Share candidate preferences
|
||||
V3->>V2: Share candidate preferences
|
||||
|
||||
Note over ES: Round 2: Information Exchange
|
||||
V1->>V2: Discuss reasoning factors
|
||||
V2->>V1: Discuss reasoning factors
|
||||
V2->>V3: Discuss reasoning factors
|
||||
V3->>V2: Discuss reasoning factors
|
||||
|
||||
Note over ES: Round 3: Final Consensus
|
||||
V1->>ES: Final vote with reasoning
|
||||
V2->>ES: Final vote with reasoning
|
||||
V3->>ES: Final vote with reasoning
|
||||
```
|
||||
|
||||
## Election Algorithms
|
||||
|
||||
The ElectionSwarm implements five AGENTSNET-inspired algorithms for orchestrator selection. Each algorithm is designed for different coordination scenarios and communication patterns.
|
||||
|
||||
### 1. Consensus Algorithm
|
||||
|
||||
All agents must agree on a single orchestrator candidate through multiple rounds of communication.
|
||||
|
||||
- **Purpose**: Achieve unanimous agreement on orchestrator selection
|
||||
- **Process**: Multiple rounds until consensus threshold is reached
|
||||
- **Threshold**: Configurable consensus threshold (default: 60%)
|
||||
- **Fallback**: Majority vote if consensus not reached
|
||||
- **Implementation**: `_conduct_consensus_election()` method
|
||||
- **Use Case**: When unanimous agreement is required
|
||||
|
||||
### 2. Leader Election Algorithm
|
||||
|
||||
Select a single leader orchestrator through distributed voting.
|
||||
|
||||
- **Purpose**: Elect one orchestrator to lead the multi-agent workflow
|
||||
- **Process**: Direct voting with winner-takes-all mechanism
|
||||
- **Complexity**: O(D) rounds where D is network diameter
|
||||
- **Implementation**: `_conduct_leader_election()` method
|
||||
- **Use Case**: When clear leadership hierarchy is needed
|
||||
|
||||
### 3. Matching Algorithm
|
||||
|
||||
Pair voters with compatible orchestrator candidates using maximal matching.
|
||||
|
||||
- **Purpose**: Create optimal voter-candidate pairings
|
||||
- **Process**: Maximal matching to avoid conflicts
|
||||
- **Complexity**: O(log* n) rounds
|
||||
- **Implementation**: `_conduct_matching_election()` method
|
||||
- **Use Case**: When specialized orchestrator-voter relationships are important
|
||||
|
||||
### 4. Coloring Algorithm
|
||||
|
||||
Group voters and candidates into compatible categories using graph coloring.
|
||||
|
||||
- **Purpose**: Assign roles such that neighbors have different roles
|
||||
- **Process**: (Δ+1)-coloring where Δ is maximum node degree
|
||||
- **Complexity**: O(log* n) rounds
|
||||
- **Implementation**: `_conduct_coloring_election()` method
|
||||
- **Use Case**: Role assignment with conflict avoidance
|
||||
|
||||
### 5. Vertex Cover Algorithm
|
||||
|
||||
Select minimal set of coordinator agents to cover all voters.
|
||||
|
||||
- **Purpose**: Choose minimal set of orchestrators to cover all voters
|
||||
- **Process**: Minimal vertex cover selection
|
||||
- **Complexity**: O(log* n) rounds
|
||||
- **Implementation**: `_conduct_vertex_cover_election()` method
|
||||
- **Use Case**: When resource-efficient coordination is needed
|
||||
|
||||
### Algorithm Selection
|
||||
|
||||
Choose the appropriate algorithm based on your coordination needs:
|
||||
|
||||
```python
|
||||
# For unanimous decisions
|
||||
ElectionAlgorithm.CONSENSUS
|
||||
|
||||
# For clear leadership
|
||||
ElectionAlgorithm.LEADER_ELECTION
|
||||
|
||||
# For optimal pairings
|
||||
ElectionAlgorithm.MATCHING
|
||||
|
||||
# For role assignment
|
||||
ElectionAlgorithm.COLORING
|
||||
|
||||
# For minimal coordination
|
||||
ElectionAlgorithm.VERTEX_COVER
|
||||
```
|
||||
|
||||
## Usage Examples
|
||||
|
||||
### Basic Election Setup
|
||||
|
||||
```python
|
||||
from swarms.structs.election_swarm import (
|
||||
ElectionSwarm,
|
||||
VoterProfile,
|
||||
CandidateProfile,
|
||||
VoterType,
|
||||
CandidateType,
|
||||
ElectionAlgorithm
|
||||
)
|
||||
|
||||
# Create agent voters
|
||||
voters = [
|
||||
VoterProfile(
|
||||
voter_id="agent_001",
|
||||
name="Data Analysis Agent",
|
||||
voter_type=VoterType.EXPERT,
|
||||
preferences={
|
||||
"technical_leadership": 0.9,
|
||||
"data_processing": 0.8,
|
||||
"workflow_efficiency": 0.6
|
||||
},
|
||||
expertise_areas=["data_science", "machine_learning"],
|
||||
neighbors=["Research Agent", "Backend Agent"]
|
||||
),
|
||||
# ... more voters
|
||||
]
|
||||
|
||||
# Create orchestrator candidates
|
||||
candidates = [
|
||||
CandidateProfile(
|
||||
candidate_id="orchestrator_001",
|
||||
name="Technical Lead Orchestrator",
|
||||
candidate_type=CandidateType.INDIVIDUAL,
|
||||
policy_positions={
|
||||
"technical_leadership": "Focus on code quality and architecture",
|
||||
"team_coordination": "Agile methodologies with clear planning"
|
||||
},
|
||||
campaign_promises=[
|
||||
"Implement comprehensive code review processes",
|
||||
"Establish automated testing pipelines"
|
||||
],
|
||||
leadership_style="transformational",
|
||||
technical_expertise=["software_architecture", "devops"]
|
||||
),
|
||||
# ... more candidates
|
||||
]
|
||||
|
||||
# Create election swarm
|
||||
election = ElectionSwarm(
|
||||
name="Multi-Agent Orchestrator Election",
|
||||
voters=voters,
|
||||
candidates=candidates,
|
||||
verbose=True
|
||||
)
|
||||
```
|
||||
|
||||
### Running Elections
|
||||
|
||||
```python
|
||||
# Run consensus election
|
||||
result = election.run(
|
||||
task="Select the best orchestrator for our multi-agent workflow",
|
||||
election_type=ElectionAlgorithm.CONSENSUS,
|
||||
max_rounds=5
|
||||
)
|
||||
|
||||
# Run leader election
|
||||
result = election.run(
|
||||
task="Elect a leader to coordinate our development team",
|
||||
election_type=ElectionAlgorithm.LEADER_ELECTION
|
||||
)
|
||||
|
||||
# Run matching election
|
||||
result = election.run(
|
||||
task="Match voters with compatible orchestrator candidates",
|
||||
election_type=ElectionAlgorithm.MATCHING
|
||||
)
|
||||
|
||||
# Run election session with comprehensive analysis
|
||||
session_result = election.run_election_session(
|
||||
election_type=ElectionAlgorithm.CONSENSUS,
|
||||
max_rounds=5
|
||||
)
|
||||
|
||||
# Access detailed results
|
||||
election_result = session_result["election_result"]
|
||||
analysis = session_result["analysis"]
|
||||
cost_stats = session_result["cost_statistics"]
|
||||
```
|
||||
|
||||
### Board CEO Election Example
|
||||
|
||||
```python
|
||||
# Create board of directors as voters
|
||||
board_members = [
|
||||
VoterProfile(
|
||||
voter_id="director_001",
|
||||
name="Sarah Chen - Technology Director",
|
||||
voter_type=VoterType.EXPERT,
|
||||
preferences={
|
||||
"innovation": 0.9,
|
||||
"technical_excellence": 0.8,
|
||||
"digital_transformation": 0.7
|
||||
},
|
||||
expertise_areas=["technology", "innovation"],
|
||||
coordination_style="data_driven",
|
||||
leadership_preferences={
|
||||
"technical_leadership": 0.9,
|
||||
"innovation": 0.8
|
||||
}
|
||||
),
|
||||
# ... more board members
|
||||
]
|
||||
|
||||
# Create CEO candidates with diverse approaches
|
||||
ceo_candidates = [
|
||||
CandidateProfile(
|
||||
candidate_id="ceo_001",
|
||||
name="Alexandra Martinez - Innovation CEO",
|
||||
candidate_type=CandidateType.INDIVIDUAL,
|
||||
policy_positions={
|
||||
"innovation_leadership": "Aggressive digital transformation",
|
||||
"growth_strategy": "Rapid expansion with high-risk investments"
|
||||
},
|
||||
leadership_style="transformational",
|
||||
coordination_approach={"innovation": 0.9, "risk_taking": 0.8}
|
||||
),
|
||||
# ... more CEO candidates
|
||||
]
|
||||
|
||||
# Create board CEO election
|
||||
ceo_election = ElectionSwarm(
|
||||
name="Board of Directors CEO Election",
|
||||
voters=board_members,
|
||||
candidates=ceo_candidates,
|
||||
verbose=True
|
||||
)
|
||||
|
||||
# Run the election
|
||||
result = ceo_election.run(
|
||||
task="Elect the best CEO to lead our company",
|
||||
election_type=ElectionAlgorithm.CONSENSUS
|
||||
)
|
||||
|
||||
# Get comprehensive statistics
|
||||
stats = ceo_election.get_election_statistics()
|
||||
print(f"Election statistics: {stats}")
|
||||
|
||||
# Get swarm status
|
||||
status = ceo_election.get_swarm_status()
|
||||
print(f"Swarm status: {status}")
|
||||
```
|
||||
|
||||
## Advanced Configuration
|
||||
|
||||
### Election Configuration
|
||||
|
||||
The ElectionSwarm uses a comprehensive configuration system with Pydantic validation.
|
||||
|
||||
```python
|
||||
from swarms.structs.election_swarm import ElectionConfig, ElectionConfigModel
|
||||
|
||||
# Create configuration with validation
|
||||
config = ElectionConfig(
|
||||
config_data={
|
||||
"election_type": "orchestrator_selection",
|
||||
"max_candidates": 5,
|
||||
"max_voters": 50,
|
||||
"enable_consensus": True,
|
||||
"enable_leader_election": True,
|
||||
"enable_matching": True,
|
||||
"enable_coloring": True,
|
||||
"enable_vertex_cover": True,
|
||||
"enable_caching": True,
|
||||
"enable_voter_tool_calls": True,
|
||||
"batch_size": 10,
|
||||
"max_workers": 5,
|
||||
"budget_limit": 100.0,
|
||||
"default_model": "gpt-4o-mini",
|
||||
"verbose_logging": True
|
||||
}
|
||||
)
|
||||
|
||||
# Validate configuration
|
||||
errors = config.validate_config()
|
||||
if errors:
|
||||
print(f"Configuration errors: {errors}")
|
||||
|
||||
# Save configuration to file
|
||||
config.save_config("election_config.yaml")
|
||||
|
||||
election = ElectionSwarm(
|
||||
name="Advanced Election",
|
||||
election_config=config,
|
||||
voters=voters,
|
||||
candidates=candidates
|
||||
)
|
||||
```
|
||||
|
||||
### ElectionConfigModel
|
||||
|
||||
The configuration model with validation constraints:
|
||||
|
||||
```python
|
||||
class ElectionConfigModel(BaseModel):
|
||||
election_type: str = Field(default="orchestrator_selection")
|
||||
max_candidates: int = Field(default=5, ge=1, le=20)
|
||||
max_voters: int = Field(default=100, ge=1, le=1000)
|
||||
enable_consensus: bool = Field(default=True)
|
||||
enable_leader_election: bool = Field(default=True)
|
||||
enable_matching: bool = Field(default=True)
|
||||
enable_coloring: bool = Field(default=True)
|
||||
enable_vertex_cover: bool = Field(default=True)
|
||||
enable_caching: bool = Field(default=True)
|
||||
enable_voter_tool_calls: bool = Field(default=True)
|
||||
batch_size: int = Field(default=25, ge=1, le=100)
|
||||
max_workers: int = Field(default=10, ge=1, le=50)
|
||||
budget_limit: float = Field(default=200.0, ge=0.0)
|
||||
default_model: str = Field(default="gpt-4o-mini")
|
||||
verbose_logging: bool = Field(default=False)
|
||||
```
|
||||
|
||||
### Cost Tracking and Budget Management
|
||||
|
||||
```python
|
||||
# Monitor election costs
|
||||
election = ElectionSwarm(
|
||||
budget_limit=50.0,
|
||||
enable_caching=True
|
||||
)
|
||||
|
||||
# Run election with cost monitoring
|
||||
result = election.run(task="Select orchestrator")
|
||||
|
||||
# Get cost statistics
|
||||
stats = election.get_election_statistics()
|
||||
print(f"Total cost: ${stats['cost_statistics']['total_cost']:.2f}")
|
||||
print(f"Cache hit rate: {stats['cost_statistics']['cache_hit_rate']:.1%}")
|
||||
```
|
||||
|
||||
## Core Methods
|
||||
|
||||
### Step Method
|
||||
|
||||
Execute a single step of the ElectionSwarm.
|
||||
|
||||
```python
|
||||
# Execute a single election step
|
||||
result = election.step(
|
||||
task="Select the best orchestrator for our team",
|
||||
election_type=ElectionAlgorithm.CONSENSUS,
|
||||
max_rounds=5
|
||||
)
|
||||
```
|
||||
|
||||
### Run Method
|
||||
|
||||
Run the ElectionSwarm for the specified number of loops.
|
||||
|
||||
```python
|
||||
# Run multiple election loops
|
||||
result = election.run(
|
||||
task="Elect a leader for our development team",
|
||||
election_type=ElectionAlgorithm.LEADER_ELECTION,
|
||||
max_rounds=3
|
||||
)
|
||||
```
|
||||
|
||||
### Conduct Election
|
||||
|
||||
Conduct an election using the specified algorithm.
|
||||
|
||||
```python
|
||||
# Conduct election directly
|
||||
result = election.conduct_election(
|
||||
election_type=ElectionAlgorithm.CONSENSUS,
|
||||
max_rounds=5
|
||||
)
|
||||
```
|
||||
|
||||
### Run Election Session
|
||||
|
||||
Run a complete election session with comprehensive analysis.
|
||||
|
||||
```python
|
||||
# Run election session with detailed analysis
|
||||
session_result = election.run_election_session(
|
||||
election_type=ElectionAlgorithm.CONSENSUS,
|
||||
max_rounds=5
|
||||
)
|
||||
|
||||
# Access results
|
||||
election_result = session_result["election_result"]
|
||||
analysis = session_result["analysis"]
|
||||
cost_statistics = session_result["cost_statistics"]
|
||||
```
|
||||
|
||||
## Swarm Management Methods
|
||||
|
||||
### Adding and Removing Participants
|
||||
|
||||
```python
|
||||
# Add new voter
|
||||
new_voter = VoterProfile(
|
||||
voter_id="agent_007",
|
||||
name="New Agent",
|
||||
voter_type=VoterType.EXPERT
|
||||
)
|
||||
election.add_voter(new_voter)
|
||||
|
||||
# Add new candidate
|
||||
new_candidate = CandidateProfile(
|
||||
candidate_id="orchestrator_004",
|
||||
name="New Orchestrator",
|
||||
candidate_type=CandidateType.INDIVIDUAL
|
||||
)
|
||||
election.add_candidate(new_candidate)
|
||||
|
||||
# Remove participants
|
||||
election.remove_voter("agent_007")
|
||||
election.remove_candidate("orchestrator_004")
|
||||
```
|
||||
|
||||
### Swarm Status and Statistics
|
||||
|
||||
```python
|
||||
# Get swarm information
|
||||
status = election.get_swarm_status()
|
||||
print(f"Total participants: {status['total_participants']}")
|
||||
print(f"Voters: {status['voters']}")
|
||||
print(f"Candidates: {status['candidates']}")
|
||||
print(f"Budget remaining: ${status['budget_remaining']:.2f}")
|
||||
|
||||
# Get comprehensive statistics
|
||||
stats = election.get_election_statistics()
|
||||
print(f"Swarm name: {stats['swarm_info']['name']}")
|
||||
print(f"Cache size: {stats['cache_stats']['cache_size']}")
|
||||
|
||||
# Get swarm size
|
||||
swarm_size = election.get_swarm_size()
|
||||
print(f"Total swarm size: {swarm_size}")
|
||||
|
||||
# Get specific voter or candidate
|
||||
voter = election.get_voter("voter_001")
|
||||
candidate = election.get_candidate("candidate_001")
|
||||
```
|
||||
|
||||
### Reset and Cleanup
|
||||
|
||||
```python
|
||||
# Reset election swarm to initial state
|
||||
election.reset()
|
||||
|
||||
# This clears:
|
||||
# - Conversation history
|
||||
# - Cost tracking
|
||||
# - Message protocol
|
||||
# - Cache
|
||||
```
|
||||
|
||||
## Best Practices
|
||||
|
||||
### 1. Voter Design
|
||||
|
||||
- **Diverse Expertise**: Include voters with different specializations
|
||||
- **Balanced Preferences**: Ensure varied preference profiles
|
||||
- **Neighbor Connections**: Create meaningful neighbor relationships
|
||||
- **Realistic Demographics**: Use realistic agent characteristics
|
||||
|
||||
### 2. Candidate Design
|
||||
|
||||
- **Clear Platforms**: Define distinct policy positions
|
||||
- **Compelling Promises**: Create realistic campaign promises
|
||||
- **Relevant Experience**: Include pertinent background experience
|
||||
- **Leadership Styles**: Vary leadership and coordination approaches
|
||||
|
||||
### 3. Election Configuration
|
||||
|
||||
- **Appropriate Algorithms**: Choose algorithms matching your use case
|
||||
- **Budget Management**: Set realistic budget limits
|
||||
- **Caching Strategy**: Enable caching for repeated elections
|
||||
- **Logging Level**: Use verbose logging for debugging
|
||||
|
||||
### 4. Performance Optimization
|
||||
|
||||
- **Lazy Loading**: Enable lazy loading for large swarms
|
||||
- **Batch Processing**: Use appropriate batch sizes
|
||||
- **Parallel Execution**: Leverage concurrent operations
|
||||
- **Cost Monitoring**: Track and optimize token usage
|
||||
|
||||
## Error Handling
|
||||
|
||||
The ElectionSwarm includes comprehensive error handling:
|
||||
|
||||
```python
|
||||
try:
|
||||
result = election.run(
|
||||
task="Select orchestrator",
|
||||
election_type=ElectionAlgorithm.CONSENSUS
|
||||
)
|
||||
except ValueError as e:
|
||||
print(f"Configuration error: {e}")
|
||||
except Exception as e:
|
||||
print(f"Election failed: {e}")
|
||||
```
|
||||
|
||||
Common error scenarios:
|
||||
|
||||
- **No voters or candidates**: Ensure at least one participant
|
||||
- **Invalid algorithm**: Use supported ElectionAlgorithm values
|
||||
- **Budget exceeded**: Monitor and adjust budget limits
|
||||
- **Agent loading failures**: Check agent configurations
|
||||
|
||||
## Integration with Other Swarms
|
||||
|
||||
The ElectionSwarm can be integrated with other Swarms architectures:
|
||||
|
||||
```python
|
||||
# Use with BoardOfDirectorsSwarm
|
||||
from swarms.structs.board_of_directors_swarm import BoardOfDirectorsSwarm
|
||||
|
||||
# Create board election
|
||||
board_election = ElectionSwarm(
|
||||
name="Board CEO Election",
|
||||
voters=board_members,
|
||||
candidates=ceo_candidates
|
||||
)
|
||||
|
||||
# Elect CEO
|
||||
ceo_result = board_election.run(
|
||||
task="Elect CEO for board governance",
|
||||
election_type=ElectionAlgorithm.CONSENSUS
|
||||
)
|
||||
|
||||
# Use elected CEO with BoardOfDirectorsSwarm
|
||||
board = BoardOfDirectorsSwarm(
|
||||
chairman=ceo_result.winner,
|
||||
# ... other board configuration
|
||||
)
|
||||
```
|
||||
|
||||
## Utility Functions
|
||||
|
||||
### Create Default Configuration
|
||||
|
||||
Create a default election configuration file.
|
||||
|
||||
```python
|
||||
from swarms.structs.election_swarm import create_default_election_config
|
||||
|
||||
# Create default configuration file
|
||||
create_default_election_config("my_election_config.yaml")
|
||||
```
|
||||
|
||||
### Default Constants
|
||||
|
||||
The ElectionSwarm uses several default constants:
|
||||
|
||||
```python
|
||||
DEFAULT_BUDGET_LIMIT = 200.0
|
||||
DEFAULT_CONSENSUS_THRESHOLD = 0.6
|
||||
DEFAULT_MAX_WORKERS = 10
|
||||
DEFAULT_MAX_ROUNDS = 5
|
||||
DEFAULT_BATCH_SIZE = 25
|
||||
```
|
||||
|
||||
## Conclusion
|
||||
|
||||
The ElectionSwarm provides a robust, scalable solution for orchestrator selection in multi-agent systems. By implementing AGENTSNET communication protocols and multiple election algorithms, it enables sophisticated democratic decision-making processes that can adapt to various coordination requirements and communication complexities.
|
||||
|
||||
Key advantages:
|
||||
|
||||
- **Distributed Coordination**: Implements proven distributed computing algorithms
|
||||
- **Flexible Architecture**: Supports various election types and configurations
|
||||
- **Production Ready**: Includes cost tracking, caching, and comprehensive error handling
|
||||
- **Swarms Integration**: Seamlessly integrates with other Swarms architectures
|
||||
- **Real-world Applications**: Suitable for corporate governance, team leadership, and workflow orchestration
|
||||
|
||||
The ElectionSwarm represents a significant advancement in multi-agent coordination, providing the tools necessary for sophisticated democratic decision-making in complex agent-based systems.
|
||||
@ -0,0 +1,17 @@
|
||||
"""
|
||||
Basic ElectionSwarm Usage Example
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', '..'))
|
||||
|
||||
from swarms.structs.election_swarm import ElectionSwarm, ElectionAlgorithm
|
||||
|
||||
# Create election
|
||||
election = ElectionSwarm(verbose=True)
|
||||
|
||||
# Run election
|
||||
task = "Vote on the best candidate for mayor based on their policies and experience"
|
||||
result = election.run(task=task, election_type=ElectionAlgorithm.CONSENSUS)
|
||||
@ -0,0 +1,434 @@
|
||||
"""
|
||||
Real-World Board CEO Election Example
|
||||
|
||||
This example demonstrates ElectionSwarm integrated with BoardOfDirectorsSwarm
|
||||
to elect a CEO. The board of directors will select from CEO candidates with
|
||||
diverse leadership approaches: strong views, negative views, and compromises.
|
||||
It includes:
|
||||
- Board of directors as voters with different expertise areas
|
||||
- CEO candidates with varying leadership styles and strategic approaches
|
||||
- AGENTSNET communication protocols for consensus building
|
||||
- Cost tracking and budget management
|
||||
- Multiple election algorithms to select the best CEO
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
from datetime import datetime
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', '..'))
|
||||
|
||||
from swarms.structs.election_swarm import (
|
||||
ElectionSwarm,
|
||||
ElectionAlgorithm,
|
||||
VoterProfile,
|
||||
CandidateProfile,
|
||||
VoteCounterProfile,
|
||||
VoterType,
|
||||
CandidateType,
|
||||
ElectionConfig
|
||||
)
|
||||
from swarms.structs.board_of_directors_swarm import BoardOfDirectorsSwarm
|
||||
|
||||
|
||||
def create_board_ceo_election() -> ElectionSwarm:
|
||||
"""Create a CEO election with board of directors as voters."""
|
||||
|
||||
# Create board of directors as voters
|
||||
voters = [
|
||||
VoterProfile(
|
||||
voter_id="director_001",
|
||||
name="Sarah Chen - Technology Director",
|
||||
voter_type=VoterType.EXPERT,
|
||||
preferences={
|
||||
"innovation": 0.9,
|
||||
"technical_excellence": 0.8,
|
||||
"digital_transformation": 0.7,
|
||||
"scalability": 0.6,
|
||||
"team_leadership": 0.8
|
||||
},
|
||||
expertise_areas=["technology", "innovation", "digital_strategy"],
|
||||
demographics={"specialization": "cto", "experience": "15_years", "domain": "tech"},
|
||||
neighbors=["Finance Director", "Operations Director"],
|
||||
coordination_style="data_driven",
|
||||
leadership_preferences={"technical_leadership": 0.9, "innovation": 0.8, "growth": 0.7}
|
||||
),
|
||||
VoterProfile(
|
||||
voter_id="director_002",
|
||||
name="Michael Rodriguez - Finance Director",
|
||||
voter_type=VoterType.EXPERT,
|
||||
preferences={
|
||||
"financial_stability": 0.9,
|
||||
"cost_optimization": 0.8,
|
||||
"risk_management": 0.7,
|
||||
"profitability": 0.6,
|
||||
"strategic_planning": 0.8
|
||||
},
|
||||
expertise_areas=["finance", "accounting", "risk_management"],
|
||||
demographics={"specialization": "cfo", "experience": "20_years", "domain": "finance"},
|
||||
neighbors=["Technology Director", "Operations Director"],
|
||||
coordination_style="conservative",
|
||||
leadership_preferences={"financial_discipline": 0.9, "risk_management": 0.8, "growth": 0.6}
|
||||
),
|
||||
VoterProfile(
|
||||
voter_id="director_003",
|
||||
name="Lisa Johnson - Operations Director",
|
||||
voter_type=VoterType.EXPERT,
|
||||
preferences={
|
||||
"operational_efficiency": 0.9,
|
||||
"process_optimization": 0.8,
|
||||
"supply_chain": 0.7,
|
||||
"quality_control": 0.6,
|
||||
"team_management": 0.8
|
||||
},
|
||||
expertise_areas=["operations", "supply_chain", "process_improvement"],
|
||||
demographics={"specialization": "coo", "experience": "18_years", "domain": "operations"},
|
||||
neighbors=["Technology Director", "Finance Director"],
|
||||
coordination_style="systematic",
|
||||
leadership_preferences={"operational_excellence": 0.9, "efficiency": 0.8, "quality": 0.7}
|
||||
),
|
||||
VoterProfile(
|
||||
voter_id="director_004",
|
||||
name="David Kim - Marketing Director",
|
||||
voter_type=VoterType.EXPERT,
|
||||
preferences={
|
||||
"brand_building": 0.9,
|
||||
"customer_acquisition": 0.8,
|
||||
"market_expansion": 0.7,
|
||||
"digital_marketing": 0.6,
|
||||
"creative_leadership": 0.8
|
||||
},
|
||||
expertise_areas=["marketing", "brand_management", "customer_relations"],
|
||||
demographics={"specialization": "cmo", "experience": "12_years", "domain": "marketing"},
|
||||
neighbors=["HR Director", "Legal Director"],
|
||||
coordination_style="creative",
|
||||
leadership_preferences={"brand_leadership": 0.9, "innovation": 0.8, "growth": 0.8}
|
||||
),
|
||||
VoterProfile(
|
||||
voter_id="director_005",
|
||||
name="Amanda Foster - HR Director",
|
||||
voter_type=VoterType.EXPERT,
|
||||
preferences={
|
||||
"talent_development": 0.9,
|
||||
"culture_building": 0.8,
|
||||
"diversity_inclusion": 0.7,
|
||||
"employee_engagement": 0.6,
|
||||
"organizational_development": 0.8
|
||||
},
|
||||
expertise_areas=["human_resources", "talent_management", "organizational_psychology"],
|
||||
demographics={"specialization": "chro", "experience": "14_years", "domain": "hr"},
|
||||
neighbors=["Marketing Director", "Legal Director"],
|
||||
coordination_style="collaborative",
|
||||
leadership_preferences={"people_leadership": 0.9, "culture": 0.8, "inclusion": 0.8}
|
||||
),
|
||||
VoterProfile(
|
||||
voter_id="director_006",
|
||||
name="Robert Thompson - Legal Director",
|
||||
voter_type=VoterType.EXPERT,
|
||||
preferences={
|
||||
"compliance": 0.9,
|
||||
"risk_mitigation": 0.8,
|
||||
"governance": 0.7,
|
||||
"ethical_leadership": 0.6,
|
||||
"strategic_advice": 0.8
|
||||
},
|
||||
expertise_areas=["corporate_law", "compliance", "governance"],
|
||||
demographics={"specialization": "general_counsel", "experience": "16_years", "domain": "legal"},
|
||||
neighbors=["Marketing Director", "HR Director"],
|
||||
coordination_style="analytical",
|
||||
leadership_preferences={"ethical_leadership": 0.9, "compliance": 0.8, "governance": 0.8}
|
||||
)
|
||||
]
|
||||
|
||||
# Create CEO candidate profiles with diverse approaches
|
||||
candidates = [
|
||||
CandidateProfile(
|
||||
candidate_id="ceo_001",
|
||||
name="Alexandra Martinez - Innovation CEO",
|
||||
candidate_type=CandidateType.INDIVIDUAL,
|
||||
party_affiliation="Innovation First",
|
||||
policy_positions={
|
||||
"innovation_leadership": "Aggressive digital transformation and cutting-edge technology adoption",
|
||||
"growth_strategy": "Rapid expansion into new markets with high-risk, high-reward investments",
|
||||
"culture_change": "Complete organizational restructuring to embrace startup mentality",
|
||||
"talent_acquisition": "Hire top-tier talent at premium costs to drive innovation",
|
||||
"risk_taking": "Embrace calculated risks and move fast, break things mentality"
|
||||
},
|
||||
campaign_promises=[
|
||||
"Double R&D investment within 18 months",
|
||||
"Launch 5 new product lines in 2 years",
|
||||
"Acquire 3 innovative startups",
|
||||
"Transform company culture to be more agile and innovative",
|
||||
"Achieve 50% revenue growth through innovation"
|
||||
],
|
||||
experience=[
|
||||
"Former Tech Startup CEO (6 years)",
|
||||
"VP of Innovation at Fortune 500 (4 years)",
|
||||
"Serial Entrepreneur",
|
||||
"Venture Capital Advisor"
|
||||
],
|
||||
support_base={
|
||||
"technology_team": 0.9,
|
||||
"marketing_team": 0.8,
|
||||
"young_professionals": 0.9,
|
||||
"innovation_advocates": 0.9
|
||||
},
|
||||
leadership_style="transformational",
|
||||
coordination_approach={"innovation": 0.9, "risk_taking": 0.8, "growth": 0.9},
|
||||
technical_expertise=["digital_transformation", "product_development", "market_expansion"]
|
||||
),
|
||||
CandidateProfile(
|
||||
candidate_id="ceo_002",
|
||||
name="James Anderson - Conservative CEO",
|
||||
candidate_type=CandidateType.INDIVIDUAL,
|
||||
party_affiliation="Stability First",
|
||||
policy_positions={
|
||||
"financial_discipline": "Maintain strict cost controls and conservative financial management",
|
||||
"risk_management": "Avoid high-risk investments and focus on proven business models",
|
||||
"operational_efficiency": "Streamline existing operations rather than major changes",
|
||||
"compliance_focus": "Strengthen governance and regulatory compliance",
|
||||
"incremental_growth": "Steady, sustainable growth through optimization"
|
||||
},
|
||||
campaign_promises=[
|
||||
"Reduce operational costs by 15% through efficiency improvements",
|
||||
"Strengthen compliance and governance frameworks",
|
||||
"Focus on core business optimization",
|
||||
"Maintain stable dividend payments to shareholders",
|
||||
"Avoid major acquisitions or risky expansions"
|
||||
],
|
||||
experience=[
|
||||
"CFO at Fortune 500 (8 years)",
|
||||
"Investment Banking (10 years)",
|
||||
"Audit Partner at Big 4",
|
||||
"Board Member at 3 public companies"
|
||||
],
|
||||
support_base={
|
||||
"finance_team": 0.9,
|
||||
"operations_team": 0.8,
|
||||
"legal_team": 0.9,
|
||||
"risk_management": 0.9
|
||||
},
|
||||
leadership_style="conservative",
|
||||
coordination_approach={"financial_discipline": 0.9, "risk_management": 0.9, "stability": 0.8},
|
||||
technical_expertise=["financial_management", "risk_assessment", "compliance"]
|
||||
),
|
||||
CandidateProfile(
|
||||
candidate_id="ceo_003",
|
||||
name="Dr. Maria Santos - Balanced CEO",
|
||||
candidate_type=CandidateType.INDIVIDUAL,
|
||||
party_affiliation="Collaborative Leadership",
|
||||
policy_positions={
|
||||
"balanced_approach": "Combine innovation with financial discipline and operational excellence",
|
||||
"stakeholder_engagement": "Involve all stakeholders in decision-making processes",
|
||||
"sustainable_growth": "Moderate growth with focus on long-term sustainability",
|
||||
"culture_building": "Strengthen company culture and employee engagement",
|
||||
"strategic_partnerships": "Build strategic alliances rather than aggressive expansion"
|
||||
},
|
||||
campaign_promises=[
|
||||
"Increase employee engagement scores by 25%",
|
||||
"Achieve 10-15% annual growth through balanced approach",
|
||||
"Strengthen strategic partnerships and alliances",
|
||||
"Improve diversity and inclusion initiatives",
|
||||
"Balance innovation investment with financial returns"
|
||||
],
|
||||
experience=[
|
||||
"COO at Mid-Cap Company (5 years)",
|
||||
"Management Consultant (8 years)",
|
||||
"Academic Leadership (6 years)",
|
||||
"Non-profit Board Leadership"
|
||||
],
|
||||
support_base={
|
||||
"hr_team": 0.9,
|
||||
"operations_team": 0.8,
|
||||
"marketing_team": 0.7,
|
||||
"middle_management": 0.8
|
||||
},
|
||||
leadership_style="collaborative",
|
||||
coordination_approach={"balance": 0.9, "collaboration": 0.8, "sustainability": 0.8},
|
||||
technical_expertise=["organizational_development", "strategic_planning", "stakeholder_management"]
|
||||
)
|
||||
]
|
||||
|
||||
# Create vote counter for the election
|
||||
vote_counter = VoteCounterProfile(
|
||||
counter_id="board_counter_001",
|
||||
name="Dr. Patricia Williams - Election Commissioner",
|
||||
role="Board Election Vote Counter",
|
||||
credentials=[
|
||||
"Certified Corporate Election Official",
|
||||
"Board Governance Specialist",
|
||||
"Transparency and Compliance Expert"
|
||||
],
|
||||
counting_methodology="transparent",
|
||||
reporting_style="comprehensive",
|
||||
counting_experience=[
|
||||
"Corporate board elections (15 years)",
|
||||
"Public company governance oversight",
|
||||
"Vote counting and verification",
|
||||
"Result documentation and reporting",
|
||||
"Regulatory compliance auditing"
|
||||
],
|
||||
verification_protocols=[
|
||||
"Double-count verification",
|
||||
"Cross-reference validation",
|
||||
"Audit trail documentation",
|
||||
"Regulatory compliance check",
|
||||
"Board member verification"
|
||||
],
|
||||
documentation_standards=[
|
||||
"Detailed vote breakdown by director",
|
||||
"Verification documentation",
|
||||
"Transparent reporting",
|
||||
"Regulatory compliance documentation",
|
||||
"Board meeting minutes integration"
|
||||
],
|
||||
result_presentation_style="detailed"
|
||||
)
|
||||
|
||||
# Create election configuration
|
||||
config = ElectionConfig(
|
||||
config_data={
|
||||
"election_type": "ceo_selection",
|
||||
"max_candidates": 5,
|
||||
"max_voters": 50,
|
||||
"enable_consensus": True,
|
||||
"enable_leader_election": True,
|
||||
"enable_matching": True,
|
||||
"enable_coloring": True,
|
||||
"enable_vertex_cover": True,
|
||||
"enable_caching": True,
|
||||
"batch_size": 10,
|
||||
"max_workers": 5,
|
||||
"budget_limit": 100.0,
|
||||
"default_model": "gpt-4o-mini",
|
||||
"verbose_logging": True
|
||||
}
|
||||
)
|
||||
|
||||
# Create the election swarm
|
||||
election = ElectionSwarm(
|
||||
name="Board of Directors CEO Election",
|
||||
description="Election to select the best CEO candidate to lead the company with diverse leadership approaches",
|
||||
voters=voters,
|
||||
candidates=candidates,
|
||||
vote_counter=vote_counter,
|
||||
election_config=config,
|
||||
max_loops=1,
|
||||
output_type="dict-all-except-first",
|
||||
verbose=True,
|
||||
enable_lazy_loading=True,
|
||||
enable_caching=True,
|
||||
batch_size=10,
|
||||
budget_limit=100.0
|
||||
)
|
||||
|
||||
return election
|
||||
|
||||
|
||||
def run_board_ceo_election():
|
||||
"""Run the board CEO election to select the best company leader."""
|
||||
|
||||
# Create the election
|
||||
election = create_board_ceo_election()
|
||||
|
||||
# Run different election algorithms
|
||||
algorithms = [
|
||||
(ElectionAlgorithm.LEADER_ELECTION, "Leader Election"),
|
||||
(ElectionAlgorithm.CONSENSUS, "Consensus Building"),
|
||||
(ElectionAlgorithm.MATCHING, "Director-CEO Matching")
|
||||
]
|
||||
|
||||
results = {}
|
||||
|
||||
for algorithm, description in algorithms:
|
||||
print(f"🔄 Running {description}...")
|
||||
|
||||
task = f"Vote for the best CEO candidate to lead our company. Consider their leadership style, strategic vision, alignment with your department's priorities, and ability to work with the board. The elected CEO will need to balance innovation, financial discipline, operational excellence, and stakeholder management."
|
||||
|
||||
try:
|
||||
# Run the election using the run method to get conversation results
|
||||
result = election.run(
|
||||
task=task,
|
||||
election_type=algorithm,
|
||||
max_rounds=3
|
||||
)
|
||||
|
||||
results[algorithm.value] = result
|
||||
|
||||
except Exception as e:
|
||||
pass
|
||||
|
||||
# Set the elected leader as orchestrator/CEO
|
||||
set_elected_ceo(election, results)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def set_elected_ceo(election: ElectionSwarm, results: dict) -> None:
|
||||
"""Set the elected leader as the orchestrator/CEO."""
|
||||
|
||||
# Find the winner from the results
|
||||
winner = None
|
||||
winning_algorithm = None
|
||||
|
||||
# Look for winner in results
|
||||
for algorithm_name, result in results.items():
|
||||
if isinstance(result, list):
|
||||
for item in result:
|
||||
if isinstance(item, dict) and 'election_result' in item:
|
||||
election_result = item['election_result']
|
||||
if election_result.get('winner'):
|
||||
winner = election_result.get('winner')
|
||||
winning_algorithm = algorithm_name
|
||||
break
|
||||
elif isinstance(result, dict) and 'election_result' in result:
|
||||
election_result = result['election_result']
|
||||
if election_result.get('winner'):
|
||||
winner = election_result.get('winner')
|
||||
winning_algorithm = algorithm_name
|
||||
break
|
||||
|
||||
# Fallback: use first candidate if no winner found
|
||||
if not winner and election.candidates:
|
||||
winner = election.candidates[0].name
|
||||
winning_algorithm = "Fallback Selection"
|
||||
|
||||
if winner:
|
||||
# Find the candidate profile
|
||||
elected_candidate = None
|
||||
for candidate in election.candidates:
|
||||
if candidate.name == winner or winner in candidate.name:
|
||||
elected_candidate = candidate
|
||||
break
|
||||
|
||||
if elected_candidate:
|
||||
# Execute a task with the elected CEO
|
||||
execute_ceo_task(election, elected_candidate)
|
||||
|
||||
|
||||
def execute_ceo_task(election: ElectionSwarm, elected_candidate: CandidateProfile) -> None:
|
||||
"""Execute a task with the elected CEO and their swarm."""
|
||||
|
||||
# Create a real business task for the elected CEO
|
||||
task = "Analyze market trends in the renewable energy sector and develop a comprehensive expansion strategy for entering the European market. Consider regulatory requirements, competitive landscape, partnership opportunities, and financial projections for the next 3 years."
|
||||
|
||||
# Simulate task execution based on CEO's leadership style
|
||||
if elected_candidate.leadership_style == "transformational":
|
||||
# Transformational approach: Inspiring vision, empowering teams, driving innovation
|
||||
# Focus on breakthrough technologies and market disruption
|
||||
pass
|
||||
elif elected_candidate.leadership_style == "conservative":
|
||||
# Conservative approach: Risk-managed growth, cost optimization, efficiency focus
|
||||
# Focus on proven markets and gradual expansion
|
||||
pass
|
||||
else: # collaborative
|
||||
# Collaborative approach: Cross-functional engagement, balanced growth, team development
|
||||
# Focus on stakeholder partnerships and sustainable growth
|
||||
pass
|
||||
|
||||
# CEO completes the task and continues leading the swarm
|
||||
# No new elections - the elected CEO remains in charge
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_board_ceo_election()
|
||||
@ -0,0 +1,19 @@
|
||||
"""
|
||||
ElectionSwarm Specialized Session Example
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', '..'))
|
||||
|
||||
from swarms.structs.election_swarm import ElectionSwarm, ElectionAlgorithm
|
||||
|
||||
# Create election
|
||||
election = ElectionSwarm(verbose=True)
|
||||
|
||||
# Run specialized session
|
||||
session_result = election.run_election_session(
|
||||
election_type=ElectionAlgorithm.CONSENSUS,
|
||||
max_rounds=3
|
||||
)
|
||||
@ -0,0 +1,17 @@
|
||||
"""
|
||||
ElectionSwarm Step Interface Example
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', '..'))
|
||||
|
||||
from swarms.structs.election_swarm import ElectionSwarm, ElectionAlgorithm
|
||||
|
||||
# Create election
|
||||
election = ElectionSwarm(verbose=True)
|
||||
|
||||
# Run single step
|
||||
task = "Vote on the best candidate for mayor based on their policies and experience"
|
||||
step_result = election.step(task=task, election_type=ElectionAlgorithm.LEADER_ELECTION)
|
||||
@ -0,0 +1,16 @@
|
||||
"""
|
||||
ElectionSwarm Statistics Example
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', '..'))
|
||||
|
||||
from swarms.structs.election_swarm import ElectionSwarm
|
||||
|
||||
# Create election
|
||||
election = ElectionSwarm(verbose=True)
|
||||
|
||||
# Get statistics
|
||||
stats = election.get_election_statistics()
|
||||
File diff suppressed because it is too large
Load Diff
Loading…
Reference in new issue