From 9eecf352a46be1b01eb3bc0f956e39c73d6c1f43 Mon Sep 17 00:00:00 2001 From: CI-DEV <154627941+IlumCI@users.noreply.github.com> Date: Wed, 23 Jul 2025 15:07:59 +0300 Subject: [PATCH 1/2] Update setup.sh --- scripts/setup.sh | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/scripts/setup.sh b/scripts/setup.sh index 6a8ebcad..cc6e04c2 100644 --- a/scripts/setup.sh +++ b/scripts/setup.sh @@ -88,6 +88,17 @@ print_success "All dependencies installed successfully" # Activate virtual environment and run additional setup print_status "Setting up development tools..." +# Setup pre-commit hooks +print_status "Installing pre-commit hooks..." +poetry run pre-commit install +poetry run pre-commit install --hook-type commit-msg + +print_success "Pre-commit hooks installed" + +# Run pre-commit on all files to ensure everything is set up correctly +print_status "Running pre-commit on all files (this may take a while on first run)..." +poetry run pre-commit run --all-files || print_warning "Some pre-commit checks failed - this is normal on first setup" + # Create .env file if it doesn't exist if [ ! -f ".env" ]; then print_status "Creating .env file template..." From 287ca2e128fb0806990830bdf94d1f13ad083b25 Mon Sep 17 00:00:00 2001 From: Swarms Contributor Date: Wed, 23 Jul 2025 16:26:08 +0300 Subject: [PATCH 2/2] Add Board of Directors feature: implementation, config, example, and tests --- .../board_of_directors_example.py | 437 +++++++ swarms/config/board_config.py | 596 +++++++++ swarms/structs/board_of_directors_swarm.py | 1144 +++++++++++++++++ .../structs/test_board_of_directors_swarm.py | 1020 +++++++++++++++ 4 files changed, 3197 insertions(+) create mode 100644 examples/multi_agent/board_of_directors/board_of_directors_example.py create mode 100644 swarms/config/board_config.py create mode 100644 swarms/structs/board_of_directors_swarm.py create mode 100644 tests/structs/test_board_of_directors_swarm.py diff --git a/examples/multi_agent/board_of_directors/board_of_directors_example.py b/examples/multi_agent/board_of_directors/board_of_directors_example.py new file mode 100644 index 00000000..8a8ffe33 --- /dev/null +++ b/examples/multi_agent/board_of_directors/board_of_directors_example.py @@ -0,0 +1,437 @@ +""" +Board of Directors Example + +This example demonstrates how to use the Board of Directors swarm feature +in the Swarms Framework. It shows how to create a board, configure it, +and use it to orchestrate tasks across multiple agents. + +The example includes: +1. Basic Board of Directors setup and usage +2. Custom board member configuration +3. Task execution and feedback +4. Configuration management +""" + +import os +import sys +from typing import List, Optional + +# Add the parent directory to the path to import swarms +sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))) + +from swarms.structs.board_of_directors_swarm import ( + BoardOfDirectorsSwarm, + BoardMember, + BoardMemberRole, +) +from swarms.structs.agent import Agent +from swarms.config.board_config import ( + enable_board_feature, + disable_board_feature, + is_board_feature_enabled, + create_default_config_file, + set_board_size, + set_decision_threshold, + set_board_model, + enable_verbose_logging, + disable_verbose_logging, +) + + +def enable_board_directors_feature() -> None: + """ + Enable the Board of Directors feature. + + This function demonstrates how to enable the Board of Directors feature + globally and create a default configuration file. + """ + print("šŸ”§ Enabling Board of Directors feature...") + + try: + # Create a default configuration file + create_default_config_file("swarms_board_config.yaml") + + # Enable the feature + enable_board_feature("swarms_board_config.yaml") + + # Configure some default settings + set_board_size(3) + set_decision_threshold(0.6) + set_board_model("gpt-4o-mini") + enable_verbose_logging("swarms_board_config.yaml") + + print("āœ… Board of Directors feature enabled successfully!") + print("šŸ“ Configuration file created: swarms_board_config.yaml") + + except Exception as e: + print(f"āŒ Failed to enable Board of Directors feature: {e}") + raise + + +def create_custom_board_members() -> List[BoardMember]: + """ + Create custom board members with specific roles and expertise. + + This function demonstrates how to create a custom board with + specialized roles and expertise areas. + + Returns: + List[BoardMember]: List of custom board members + """ + print("šŸ‘„ Creating custom board members...") + + # Create specialized board members + chairman = Agent( + agent_name="Executive_Chairman", + agent_description="Executive Chairman with strategic vision and leadership expertise", + model_name="gpt-4o-mini", + max_loops=1, + system_prompt="""You are the Executive Chairman of the Board. Your role is to: +1. Provide strategic leadership and vision +2. Facilitate high-level decision-making +3. Ensure board effectiveness and governance +4. Represent the organization's interests +5. Guide long-term strategic planning + +You should be visionary, strategic, and focused on organizational success.""", + ) + + cto = Agent( + agent_name="CTO", + agent_description="Chief Technology Officer with deep technical expertise", + model_name="gpt-4o-mini", + max_loops=1, + system_prompt="""You are the Chief Technology Officer. Your role is to: +1. Provide technical leadership and strategy +2. Evaluate technology solutions and architectures +3. Ensure technical feasibility of proposed solutions +4. Guide technology-related decisions +5. Maintain technical standards and best practices + +You should be technically proficient, innovative, and focused on technical excellence.""", + ) + + cfo = Agent( + agent_name="CFO", + agent_description="Chief Financial Officer with financial and risk management expertise", + model_name="gpt-4o-mini", + max_loops=1, + system_prompt="""You are the Chief Financial Officer. Your role is to: +1. Provide financial analysis and insights +2. Evaluate financial implications of decisions +3. Ensure financial sustainability and risk management +4. Guide resource allocation and budgeting +5. Maintain financial controls and compliance + +You should be financially astute, risk-aware, and focused on financial health.""", + ) + + # Create BoardMember objects with roles and expertise + board_members = [ + BoardMember( + agent=chairman, + role=BoardMemberRole.CHAIRMAN, + voting_weight=2.0, + expertise_areas=["strategic_planning", "leadership", "governance", "business_strategy"] + ), + BoardMember( + agent=cto, + role=BoardMemberRole.EXECUTIVE_DIRECTOR, + voting_weight=1.5, + expertise_areas=["technology", "architecture", "innovation", "technical_strategy"] + ), + BoardMember( + agent=cfo, + role=BoardMemberRole.EXECUTIVE_DIRECTOR, + voting_weight=1.5, + expertise_areas=["finance", "risk_management", "budgeting", "financial_analysis"] + ), + ] + + print(f"āœ… Created {len(board_members)} custom board members") + for member in board_members: + print(f" - {member.agent.agent_name} ({member.role.value})") + + return board_members + + +def create_worker_agents() -> List[Agent]: + """ + Create worker agents for the swarm. + + This function creates specialized worker agents that will be + managed by the Board of Directors. + + Returns: + List[Agent]: List of worker agents + """ + print("šŸ› ļø Creating worker agents...") + + # Create specialized worker agents + researcher = Agent( + agent_name="Research_Analyst", + agent_description="Research analyst specializing in market research and data analysis", + model_name="gpt-4o-mini", + max_loops=1, + system_prompt="""You are a Research Analyst. Your responsibilities include: +1. Conducting thorough research on assigned topics +2. Analyzing data and market trends +3. Preparing comprehensive research reports +4. Providing data-driven insights and recommendations +5. Maintaining high standards of research quality + +You should be analytical, thorough, and evidence-based in your work.""", + ) + + developer = Agent( + agent_name="Software_Developer", + agent_description="Software developer with expertise in system design and implementation", + model_name="gpt-4o-mini", + max_loops=1, + system_prompt="""You are a Software Developer. Your responsibilities include: +1. Designing and implementing software solutions +2. Writing clean, maintainable code +3. Conducting code reviews and testing +4. Collaborating with team members +5. Following best practices and coding standards + +You should be technically skilled, detail-oriented, and focused on quality.""", + ) + + marketer = Agent( + agent_name="Marketing_Specialist", + agent_description="Marketing specialist with expertise in digital marketing and brand strategy", + model_name="gpt-4o-mini", + max_loops=1, + system_prompt="""You are a Marketing Specialist. Your responsibilities include: +1. Developing marketing strategies and campaigns +2. Creating compelling content and messaging +3. Analyzing market trends and customer behavior +4. Managing brand presence and reputation +5. Measuring and optimizing marketing performance + +You should be creative, strategic, and customer-focused in your approach.""", + ) + + agents = [researcher, developer, marketer] + + print(f"āœ… Created {len(agents)} worker agents") + for agent in agents: + print(f" - {agent.agent_name}: {agent.agent_description}") + + return agents + + +def run_board_of_directors_example() -> None: + """ + Run a comprehensive Board of Directors example. + + This function demonstrates the complete workflow of using + the Board of Directors swarm to orchestrate tasks. + """ + print("\n" + "="*60) + print("šŸ›ļø BOARD OF DIRECTORS SWARM EXAMPLE") + print("="*60) + + try: + # Check if Board of Directors feature is enabled + if not is_board_feature_enabled(): + print("āš ļø Board of Directors feature is not enabled. Enabling now...") + enable_board_directors_feature() + + # Create custom board members + board_members = create_custom_board_members() + + # Create worker agents + worker_agents = create_worker_agents() + + # Create the Board of Directors swarm + print("\nšŸ›ļø Creating Board of Directors swarm...") + board_swarm = BoardOfDirectorsSwarm( + name="Executive_Board_Swarm", + description="Executive board with specialized roles for strategic decision-making", + board_members=board_members, + agents=worker_agents, + max_loops=2, + verbose=True, + decision_threshold=0.6, + enable_voting=True, + enable_consensus=True, + ) + + print("āœ… Board of Directors swarm created successfully!") + + # Display board summary + summary = board_swarm.get_board_summary() + print(f"\nšŸ“Š Board Summary:") + print(f" Board Name: {summary['board_name']}") + print(f" Total Members: {summary['total_members']}") + print(f" Total Agents: {summary['total_agents']}") + print(f" Max Loops: {summary['max_loops']}") + print(f" Decision Threshold: {summary['decision_threshold']}") + + print(f"\nšŸ‘„ Board Members:") + for member in summary['members']: + print(f" - {member['name']} ({member['role']}) - Weight: {member['voting_weight']}") + print(f" Expertise: {', '.join(member['expertise_areas'])}") + + # Define a complex task for the board to handle + task = """ + Develop a comprehensive strategy for launching a new AI-powered product in the market. + + The task involves: + 1. Market research and competitive analysis + 2. Technical architecture and development planning + 3. Marketing strategy and go-to-market plan + 4. Financial projections and risk assessment + + Please coordinate the efforts of all team members to create a cohesive strategy. + """ + + print(f"\nšŸ“‹ Executing task: {task.strip()[:100]}...") + + # Execute the task using the Board of Directors swarm + result = board_swarm.run(task=task) + + print("\nāœ… Task completed successfully!") + print(f"šŸ“„ Result type: {type(result)}") + + # Display conversation history + if hasattr(result, 'get') and callable(result.get): + conversation_history = result.get('conversation_history', []) + print(f"\nšŸ’¬ Conversation History ({len(conversation_history)} messages):") + for i, message in enumerate(conversation_history[-5:], 1): # Show last 5 messages + role = message.get('role', 'Unknown') + content = message.get('content', '')[:100] + "..." if len(message.get('content', '')) > 100 else message.get('content', '') + print(f" {i}. {role}: {content}") + else: + print(f"\nšŸ“ Result: {str(result)[:200]}...") + + print("\nšŸŽ‰ Board of Directors example completed successfully!") + + except Exception as e: + print(f"āŒ Error in Board of Directors example: {e}") + import traceback + traceback.print_exc() + + +def run_simple_board_example() -> None: + """ + Run a simple Board of Directors example with default settings. + + This function demonstrates a basic usage of the Board of Directors + swarm with minimal configuration. + """ + print("\n" + "="*60) + print("šŸ›ļø SIMPLE BOARD OF DIRECTORS EXAMPLE") + print("="*60) + + try: + # Create simple worker agents + print("šŸ› ļø Creating simple worker agents...") + + analyst = Agent( + agent_name="Data_Analyst", + agent_description="Data analyst for processing and analyzing information", + model_name="gpt-4o-mini", + max_loops=1, + ) + + writer = Agent( + agent_name="Content_Writer", + agent_description="Content writer for creating reports and documentation", + model_name="gpt-4o-mini", + max_loops=1, + ) + + agents = [analyst, writer] + + # Create Board of Directors swarm with default settings + print("šŸ›ļø Creating Board of Directors swarm with default settings...") + board_swarm = BoardOfDirectorsSwarm( + name="Simple_Board_Swarm", + agents=agents, + verbose=True, + ) + + print("āœ… Simple Board of Directors swarm created!") + + # Simple task + task = "Analyze the current market trends and create a summary report with recommendations." + + print(f"\nšŸ“‹ Executing simple task: {task}") + + # Execute the task + result = board_swarm.run(task=task) + + print("\nāœ… Simple task completed successfully!") + print(f"šŸ“„ Result type: {type(result)}") + + if hasattr(result, 'get') and callable(result.get): + conversation_history = result.get('conversation_history', []) + print(f"\nšŸ’¬ Conversation History ({len(conversation_history)} messages):") + for i, message in enumerate(conversation_history[-3:], 1): # Show last 3 messages + role = message.get('role', 'Unknown') + content = message.get('content', '')[:80] + "..." if len(message.get('content', '')) > 80 else message.get('content', '') + print(f" {i}. {role}: {content}") + else: + print(f"\nšŸ“ Result: {str(result)[:150]}...") + + print("\nšŸŽ‰ Simple Board of Directors example completed!") + + except Exception as e: + print(f"āŒ Error in simple Board of Directors example: {e}") + import traceback + traceback.print_exc() + + +def check_environment() -> bool: + """ + Check if the environment is properly set up for the example. + + Returns: + bool: True if environment is ready, False otherwise + """ + # Check for OpenAI API key + if not os.getenv("OPENAI_API_KEY"): + print("āš ļø Warning: OPENAI_API_KEY environment variable not set.") + print(" The example may not work without a valid API key.") + print(" Please set your OpenAI API key: export OPENAI_API_KEY='your-key-here'") + return False + + return True + + +def main() -> None: + """ + Main function to run the Board of Directors examples. + """ + print("šŸš€ Board of Directors Swarm Examples") + print("="*50) + + # Check environment + if not check_environment(): + print("\nāš ļø Environment check failed. Please set up your environment properly.") + return + + try: + # Run simple example first + run_simple_board_example() + + # Run comprehensive example + run_board_of_directors_example() + + print("\n" + "="*60) + print("šŸŽ‰ All Board of Directors examples completed successfully!") + print("="*60) + + except KeyboardInterrupt: + print("\nāš ļø Examples interrupted by user.") + except Exception as e: + print(f"\nāŒ Unexpected error: {e}") + import traceback + traceback.print_exc() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/swarms/config/board_config.py b/swarms/config/board_config.py new file mode 100644 index 00000000..74ccbaab --- /dev/null +++ b/swarms/config/board_config.py @@ -0,0 +1,596 @@ +""" +Board of Directors Configuration Module + +This module provides configuration management for the Board of Directors feature +in the Swarms Framework. It allows users to enable and configure the Board of +Directors feature manually through environment variables or configuration files. + +The implementation follows the Swarms philosophy of: +- Readable code with comprehensive type annotations and documentation +- Performance optimization through caching and efficient loading +- Simplified abstractions for configuration management +""" + +import os +from typing import Dict, Any, Optional, List, Union +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from functools import lru_cache + +from pydantic import BaseModel, Field +from loguru import logger + + +class BoardFeatureStatus(str, Enum): + """Enumeration of Board of Directors feature status. + + This enum defines the possible states of the Board of Directors feature + within the Swarms Framework. + + Attributes: + ENABLED: Feature is explicitly enabled + DISABLED: Feature is explicitly disabled + AUTO: Feature state is determined automatically + """ + + ENABLED = "enabled" + DISABLED = "disabled" + AUTO = "auto" + + +class BoardConfigModel(BaseModel): + """ + Configuration model for Board of Directors feature. + + This model defines all configurable parameters for the Board of Directors + feature, including feature status, board composition, and operational settings. + + Attributes: + board_feature_enabled: Whether the Board of Directors feature is enabled globally + default_board_size: Default number of board members when creating a new board + decision_threshold: Threshold for majority decisions (0.0-1.0) + enable_voting: Enable voting mechanisms for board decisions + enable_consensus: Enable consensus-building mechanisms + default_board_model: Default model for board member agents + verbose_logging: Enable verbose logging for board operations + max_board_meeting_duration: Maximum duration for board meetings in seconds + auto_fallback_to_director: Automatically fall back to Director mode if Board fails + custom_board_templates: Custom board templates for different use cases + """ + + # Feature control + board_feature_enabled: bool = Field( + default=False, + description="Whether the Board of Directors feature is enabled globally." + ) + + # Board composition + default_board_size: int = Field( + default=3, + ge=1, + le=10, + description="Default number of board members when creating a new board." + ) + + # Operational settings + decision_threshold: float = Field( + default=0.6, + ge=0.0, + le=1.0, + description="Threshold for majority decisions (0.0-1.0)." + ) + + enable_voting: bool = Field( + default=True, + description="Enable voting mechanisms for board decisions." + ) + + enable_consensus: bool = Field( + default=True, + description="Enable consensus-building mechanisms." + ) + + # Model settings + default_board_model: str = Field( + default="gpt-4o-mini", + description="Default model for board member agents." + ) + + # Logging and monitoring + verbose_logging: bool = Field( + default=False, + description="Enable verbose logging for board operations." + ) + + # Performance settings + max_board_meeting_duration: int = Field( + default=300, + ge=60, + le=3600, + description="Maximum duration for board meetings in seconds." + ) + + # Integration settings + auto_fallback_to_director: bool = Field( + default=True, + description="Automatically fall back to Director mode if Board fails." + ) + + # Custom board templates + custom_board_templates: Dict[str, Dict[str, Any]] = Field( + default_factory=dict, + description="Custom board templates for different use cases." + ) + + +@dataclass +class BoardConfig: + """ + Board of Directors configuration manager. + + This class manages the configuration for the Board of Directors feature, + including loading from environment variables, configuration files, and + providing default values. + + Attributes: + config_file_path: Optional path to configuration file + config_data: Optional configuration data dictionary + config: The current configuration model instance + """ + + config_file_path: Optional[str] = None + config_data: Optional[Dict[str, Any]] = None + config: BoardConfigModel = field(init=False) + + def __post_init__(self) -> None: + """Initialize the configuration after object creation.""" + self._load_config() + + def _load_config(self) -> None: + """ + Load configuration from various sources. + + Priority order: + 1. Environment variables + 2. Configuration file + 3. Default values + + Raises: + Exception: If configuration loading fails + """ + try: + # Start with default configuration + self.config = BoardConfigModel() + + # Load from configuration file if specified + if self.config_file_path and os.path.exists(self.config_file_path): + self._load_from_file() + + # Override with environment variables + self._load_from_environment() + + # Override with explicit config data + if self.config_data: + self._load_from_dict(self.config_data) + + except Exception as e: + logger.error(f"Failed to load Board of Directors configuration: {str(e)}") + raise + + def _load_from_file(self) -> None: + """ + Load configuration from file. + + Raises: + Exception: If file loading fails + """ + try: + import yaml + with open(self.config_file_path, 'r') as f: + file_config = yaml.safe_load(f) + self._load_from_dict(file_config) + logger.info(f"Loaded Board of Directors config from: {self.config_file_path}") + except Exception as e: + logger.warning(f"Failed to load config file {self.config_file_path}: {e}") + raise + + def _load_from_environment(self) -> None: + """ + Load configuration from environment variables. + + This method maps environment variables to configuration parameters + and handles type conversion appropriately. + """ + env_mappings = { + 'SWARMS_BOARD_FEATURE_ENABLED': 'board_feature_enabled', + 'SWARMS_BOARD_DEFAULT_SIZE': 'default_board_size', + 'SWARMS_BOARD_DECISION_THRESHOLD': 'decision_threshold', + 'SWARMS_BOARD_ENABLE_VOTING': 'enable_voting', + 'SWARMS_BOARD_ENABLE_CONSENSUS': 'enable_consensus', + 'SWARMS_BOARD_DEFAULT_MODEL': 'default_board_model', + 'SWARMS_BOARD_VERBOSE_LOGGING': 'verbose_logging', + 'SWARMS_BOARD_MAX_MEETING_DURATION': 'max_board_meeting_duration', + 'SWARMS_BOARD_AUTO_FALLBACK': 'auto_fallback_to_director', + } + + for env_var, config_key in env_mappings.items(): + value = os.getenv(env_var) + if value is not None: + try: + # Convert string values to appropriate types + if config_key in ['board_feature_enabled', 'enable_voting', 'enable_consensus', 'verbose_logging', 'auto_fallback_to_director']: + converted_value = value.lower() in ['true', '1', 'yes', 'on'] + elif config_key in ['default_board_size', 'max_board_meeting_duration']: + converted_value = int(value) + elif config_key in ['decision_threshold']: + converted_value = float(value) + else: + converted_value = value + + setattr(self.config, config_key, converted_value) + logger.debug(f"Loaded {config_key} from environment: {converted_value}") + except (ValueError, TypeError) as e: + logger.warning(f"Failed to parse environment variable {env_var}: {e}") + + def _load_from_dict(self, config_dict: Dict[str, Any]) -> None: + """ + Load configuration from dictionary. + + Args: + config_dict: Dictionary containing configuration values + + Raises: + ValueError: If configuration values are invalid + """ + for key, value in config_dict.items(): + if hasattr(self.config, key): + try: + setattr(self.config, key, value) + except (ValueError, TypeError) as e: + logger.warning(f"Failed to set config {key}: {e}") + raise ValueError(f"Invalid configuration value for {key}: {e}") + + def is_enabled(self) -> bool: + """ + Check if the Board of Directors feature is enabled. + + Returns: + bool: True if the feature is enabled, False otherwise + """ + return self.config.board_feature_enabled + + def get_config(self) -> BoardConfigModel: + """ + Get the current configuration. + + Returns: + BoardConfigModel: The current configuration + """ + return self.config + + def update_config(self, updates: Dict[str, Any]) -> None: + """ + Update the configuration with new values. + + Args: + updates: Dictionary of configuration updates + + Raises: + ValueError: If any update values are invalid + """ + try: + self._load_from_dict(updates) + except ValueError as e: + logger.error(f"Failed to update configuration: {e}") + raise + + def save_config(self, file_path: Optional[str] = None) -> None: + """ + Save the current configuration to a file. + + Args: + file_path: Optional file path to save to (uses config_file_path if not provided) + + Raises: + Exception: If saving fails + """ + save_path = file_path or self.config_file_path + if not save_path: + logger.warning("No file path specified for saving configuration") + return + + try: + import yaml + # Convert config to dictionary + config_dict = self.config.model_dump() + + # Ensure directory exists + os.makedirs(os.path.dirname(save_path), exist_ok=True) + + with open(save_path, 'w') as f: + yaml.dump(config_dict, f, default_flow_style=False, indent=2) + + logger.info(f"Saved Board of Directors config to: {save_path}") + except Exception as e: + logger.error(f"Failed to save config to {save_path}: {e}") + raise + + @lru_cache(maxsize=128) + def get_default_board_template(self, template_name: str = "standard") -> Dict[str, Any]: + """ + Get a default board template. + + This method provides predefined board templates for common use cases. + Templates are cached for improved performance. + + Args: + template_name: Name of the template to retrieve + + Returns: + Dict[str, Any]: Board template configuration + """ + templates = { + "standard": { + "roles": [ + {"name": "Chairman", "weight": 1.5, "expertise": ["leadership", "strategy"]}, + {"name": "Vice-Chairman", "weight": 1.2, "expertise": ["operations", "coordination"]}, + {"name": "Secretary", "weight": 1.0, "expertise": ["documentation", "communication"]}, + ] + }, + "executive": { + "roles": [ + {"name": "CEO", "weight": 2.0, "expertise": ["executive_leadership", "strategy"]}, + {"name": "CFO", "weight": 1.5, "expertise": ["finance", "risk_management"]}, + {"name": "CTO", "weight": 1.5, "expertise": ["technology", "innovation"]}, + {"name": "COO", "weight": 1.3, "expertise": ["operations", "efficiency"]}, + ] + }, + "advisory": { + "roles": [ + {"name": "Lead_Advisor", "weight": 1.3, "expertise": ["strategy", "consulting"]}, + {"name": "Technical_Advisor", "weight": 1.2, "expertise": ["technology", "architecture"]}, + {"name": "Business_Advisor", "weight": 1.2, "expertise": ["business", "market_analysis"]}, + {"name": "Legal_Advisor", "weight": 1.1, "expertise": ["legal", "compliance"]}, + ] + }, + "minimal": { + "roles": [ + {"name": "Chairman", "weight": 1.0, "expertise": ["leadership"]}, + {"name": "Member", "weight": 1.0, "expertise": ["general"]}, + ] + } + } + + # Check custom templates first + if template_name in self.config.custom_board_templates: + return self.config.custom_board_templates[template_name] + + # Return standard template if requested template not found + return templates.get(template_name, templates["standard"]) + + def validate_config(self) -> List[str]: + """ + Validate the current configuration. + + This method performs comprehensive validation of the configuration + to ensure all values are within acceptable ranges and constraints. + + Returns: + List[str]: List of validation errors (empty if valid) + """ + errors = [] + + try: + # Validate the configuration model + self.config.model_validate(self.config.model_dump()) + except Exception as e: + errors.append(f"Configuration validation failed: {e}") + + # Additional custom validations + if self.config.decision_threshold < 0.5: + errors.append("Decision threshold should be at least 0.5 for meaningful majority decisions") + + if self.config.default_board_size < 2: + errors.append("Board size should be at least 2 for meaningful discussions") + + if self.config.max_board_meeting_duration < 60: + errors.append("Board meeting duration should be at least 60 seconds") + + return errors + + +# Global configuration instance +_board_config: Optional[BoardConfig] = None + + +@lru_cache(maxsize=1) +def get_board_config(config_file_path: Optional[str] = None) -> BoardConfig: + """ + Get the global Board of Directors configuration instance. + + This function provides a singleton pattern for accessing the Board of Directors + configuration. The configuration is cached for improved performance. + + Args: + config_file_path: Optional path to configuration file + + Returns: + BoardConfig: The global configuration instance + """ + global _board_config + + if _board_config is None: + _board_config = BoardConfig(config_file_path=config_file_path) + + return _board_config + + +def enable_board_feature(config_file_path: Optional[str] = None) -> None: + """ + Enable the Board of Directors feature globally. + + This function enables the Board of Directors feature and saves the configuration + to the specified file path. + + Args: + config_file_path: Optional path to save the configuration + """ + config = get_board_config(config_file_path) + config.update_config({"board_feature_enabled": True}) + + if config_file_path: + config.save_config(config_file_path) + + logger.info("Board of Directors feature enabled") + + +def disable_board_feature(config_file_path: Optional[str] = None) -> None: + """ + Disable the Board of Directors feature globally. + + This function disables the Board of Directors feature and saves the configuration + to the specified file path. + + Args: + config_file_path: Optional path to save the configuration + """ + config = get_board_config(config_file_path) + config.update_config({"board_feature_enabled": False}) + + if config_file_path: + config.save_config(config_file_path) + + logger.info("Board of Directors feature disabled") + + +def is_board_feature_enabled(config_file_path: Optional[str] = None) -> bool: + """ + Check if the Board of Directors feature is enabled. + + Args: + config_file_path: Optional path to configuration file + + Returns: + bool: True if the feature is enabled, False otherwise + """ + config = get_board_config(config_file_path) + return config.is_enabled() + + +def create_default_config_file(file_path: str = "swarms_board_config.yaml") -> None: + """ + Create a default configuration file. + + This function creates a default Board of Directors configuration file + with recommended settings. + + Args: + file_path: Path where to create the configuration file + """ + default_config = { + "board_feature_enabled": False, + "default_board_size": 3, + "decision_threshold": 0.6, + "enable_voting": True, + "enable_consensus": True, + "default_board_model": "gpt-4o-mini", + "verbose_logging": False, + "max_board_meeting_duration": 300, + "auto_fallback_to_director": True, + "custom_board_templates": {} + } + + config = BoardConfig(config_file_path=file_path, config_data=default_config) + config.save_config(file_path) + + logger.info(f"Created default Board of Directors config file: {file_path}") + + +def set_board_size(size: int, config_file_path: Optional[str] = None) -> None: + """ + Set the default board size. + + Args: + size: The default board size (1-10) + config_file_path: Optional path to save the configuration + """ + if not 1 <= size <= 10: + raise ValueError("Board size must be between 1 and 10") + + config = get_board_config(config_file_path) + config.update_config({"default_board_size": size}) + + if config_file_path: + config.save_config(config_file_path) + + logger.info(f"Default board size set to: {size}") + + +def set_decision_threshold(threshold: float, config_file_path: Optional[str] = None) -> None: + """ + Set the decision threshold for majority decisions. + + Args: + threshold: The decision threshold (0.0-1.0) + config_file_path: Optional path to save the configuration + """ + if not 0.0 <= threshold <= 1.0: + raise ValueError("Decision threshold must be between 0.0 and 1.0") + + config = get_board_config(config_file_path) + config.update_config({"decision_threshold": threshold}) + + if config_file_path: + config.save_config(config_file_path) + + logger.info(f"Decision threshold set to: {threshold}") + + +def set_board_model(model: str, config_file_path: Optional[str] = None) -> None: + """ + Set the default board model. + + Args: + model: The default model name for board members + config_file_path: Optional path to save the configuration + """ + config = get_board_config(config_file_path) + config.update_config({"default_board_model": model}) + + if config_file_path: + config.save_config(config_file_path) + + logger.info(f"Default board model set to: {model}") + + +def enable_verbose_logging(config_file_path: Optional[str] = None) -> None: + """ + Enable verbose logging for board operations. + + Args: + config_file_path: Optional path to save the configuration + """ + config = get_board_config(config_file_path) + config.update_config({"verbose_logging": True}) + + if config_file_path: + config.save_config(config_file_path) + + logger.info("Verbose logging enabled for Board of Directors") + + +def disable_verbose_logging(config_file_path: Optional[str] = None) -> None: + """ + Disable verbose logging for board operations. + + Args: + config_file_path: Optional path to save the configuration + """ + config = get_board_config(config_file_path) + config.update_config({"verbose_logging": False}) + + if config_file_path: + config.save_config(config_file_path) + + logger.info("Verbose logging disabled for Board of Directors") \ No newline at end of file diff --git a/swarms/structs/board_of_directors_swarm.py b/swarms/structs/board_of_directors_swarm.py new file mode 100644 index 00000000..d5b8dbf1 --- /dev/null +++ b/swarms/structs/board_of_directors_swarm.py @@ -0,0 +1,1144 @@ +""" +Board of Directors Swarm Implementation + +This module implements a Board of Directors feature as an alternative to the Director feature +in the Swarms Framework. The Board of Directors operates as a collective decision-making body +that can be enabled manually through configuration. + +The implementation follows the Swarms philosophy of: +- Readable code with comprehensive type annotations and documentation +- Performance optimization through concurrency and parallelism +- Simplified abstractions for multi-agent collaboration + +Flow: +1. User provides a task +2. Board of Directors convenes to discuss and create a plan +3. Board distributes orders to agents through voting and consensus +4. Agents execute tasks and report back to the board +5. Board evaluates results and issues new orders if needed (up to max_loops) +6. All context and conversation history is preserved throughout the process +""" + +import asyncio +import json +import os +import re +import traceback +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Callable, Dict, List, Optional, Union, Tuple + +from loguru import logger +from pydantic import BaseModel, Field + +from swarms.structs.agent import Agent +from swarms.structs.base_swarm import BaseSwarm +from swarms.structs.conversation import Conversation +from swarms.structs.ma_utils import list_all_agents +from swarms.utils.history_output_formatter import history_output_formatter +from swarms.utils.loguru_logger import initialize_logger +from swarms.utils.output_types import OutputType + +# Initialize logger for Board of Directors swarm +board_logger = initialize_logger(log_folder="board_of_directors_swarm") + + +class BoardMemberRole(str, Enum): + """Enumeration of possible board member roles. + + This enum defines the various roles that board members can have within + the Board of Directors swarm. Each role has specific responsibilities + and voting weights associated with it. + + Attributes: + CHAIRMAN: Primary leader responsible for board meetings and final decisions + VICE_CHAIRMAN: Secondary leader who supports the chairman + SECRETARY: Responsible for documentation and meeting minutes + TREASURER: Manages financial aspects and resource allocation + MEMBER: General board member with specific expertise + EXECUTIVE_DIRECTOR: Executive-level board member with operational authority + """ + + CHAIRMAN = "chairman" + VICE_CHAIRMAN = "vice_chairman" + SECRETARY = "secretary" + TREASURER = "treasurer" + MEMBER = "member" + EXECUTIVE_DIRECTOR = "executive_director" + + +class BoardDecisionType(str, Enum): + """Enumeration of board decision types. + + This enum defines the different types of decisions that can be made + by the Board of Directors, including voting mechanisms and consensus + approaches. + + Attributes: + UNANIMOUS: All board members agree on the decision + MAJORITY: More than 50% of votes are in favor + CONSENSUS: General agreement without formal voting + CHAIRMAN_DECISION: Final decision made by the chairman + """ + + UNANIMOUS = "unanimous" + MAJORITY = "majority" + CONSENSUS = "consensus" + CHAIRMAN_DECISION = "chairman_decision" + + +@dataclass +class BoardMember: + """ + Represents a member of the Board of Directors. + + This dataclass encapsulates all information about a board member, + including their agent representation, role, voting weight, and + areas of expertise. + + Attributes: + agent: The agent representing this board member + role: The role of this board member within the board + voting_weight: The weight of this member's vote (default: 1.0) + expertise_areas: Areas of expertise for this board member + """ + + agent: Agent + role: BoardMemberRole + voting_weight: float = 1.0 + expertise_areas: List[str] = field(default_factory=list) + + def __post_init__(self) -> None: + """Initialize default values after object creation. + + This method ensures that the expertise_areas list is properly + initialized as an empty list if not provided. + """ + if self.expertise_areas is None: + self.expertise_areas = [] + + +class BoardOrder(BaseModel): + """ + Represents an order issued by the Board of Directors. + + This model defines the structure of orders that the board issues + to worker agents, including task assignments, priorities, and + deadlines. + + Attributes: + agent_name: The name of the agent to which the task is assigned + task: The specific task to be executed by the assigned agent + priority: Priority level of the task (1-5, where 1 is highest) + deadline: Optional deadline for task completion + assigned_by: The board member who assigned this task + """ + + agent_name: str = Field( + ..., + description="Specifies the name of the agent to which the task is assigned.", + ) + task: str = Field( + ..., + description="Defines the specific task to be executed by the assigned agent.", + ) + priority: int = Field( + default=3, + ge=1, + le=5, + description="Priority level of the task (1-5, where 1 is highest priority).", + ) + deadline: Optional[str] = Field( + default=None, + description="Optional deadline for task completion.", + ) + assigned_by: str = Field( + default="Board of Directors", + description="The board member who assigned this task.", + ) + + +class BoardDecision(BaseModel): + """ + Represents a decision made by the Board of Directors. + + This model tracks the details of decisions made by the board, + including voting results, decision types, and reasoning. + + Attributes: + decision_type: The type of decision (unanimous, majority, etc.) + decision: The actual decision made + votes_for: Number of votes in favor + votes_against: Number of votes against + abstentions: Number of abstentions + reasoning: The reasoning behind the decision + """ + + decision_type: BoardDecisionType = Field( + ..., + description="The type of decision made by the board.", + ) + decision: str = Field( + ..., + description="The actual decision made by the board.", + ) + votes_for: int = Field( + default=0, + ge=0, + description="Number of votes in favor of the decision.", + ) + votes_against: int = Field( + default=0, + ge=0, + description="Number of votes against the decision.", + ) + abstentions: int = Field( + default=0, + ge=0, + description="Number of abstentions.", + ) + reasoning: str = Field( + default="", + description="The reasoning behind the decision.", + ) + + +class BoardSpec(BaseModel): + """ + Specification for Board of Directors operations. + + This model represents the complete output of a board meeting, + including the plan, orders, decisions, and meeting summary. + + Attributes: + plan: The overall plan created by the board + orders: List of orders issued by the board + decisions: List of decisions made by the board + meeting_summary: Summary of the board meeting + """ + + plan: str = Field( + ..., + description="Outlines the sequence of actions to be taken by the swarm as decided by the board.", + ) + orders: List[BoardOrder] = Field( + ..., + description="A collection of task assignments to specific agents within the swarm.", + ) + decisions: List[BoardDecision] = Field( + default_factory=list, + description="List of decisions made by the board during the meeting.", + ) + meeting_summary: str = Field( + default="", + description="Summary of the board meeting and key outcomes.", + ) + + +class BoardOfDirectorsSwarm(BaseSwarm): + """ + A hierarchical swarm of agents with a Board of Directors that orchestrates tasks. + + The Board of Directors operates as a collective decision-making body that can be + enabled manually through configuration. It provides an alternative to the single + Director approach with more democratic and collaborative decision-making. + + The workflow follows a hierarchical pattern: + 1. Task is received and sent to the Board of Directors + 2. Board convenes to discuss and create a plan through voting and consensus + 3. Board distributes orders to agents based on collective decisions + 4. Agents execute tasks and report back to the board + 5. Board evaluates results and issues new orders if needed (up to max_loops) + 6. All context and conversation history is preserved throughout the process + + Attributes: + name: The name of the swarm + description: A description of the swarm + board_members: List of board members with their roles and expertise + agents: A list of agents within the swarm + max_loops: The maximum number of feedback loops between the board and agents + output_type: The format in which to return the output (dict, str, or list) + board_model_name: The model name for board member agents + verbose: Enable detailed logging with loguru + add_collaboration_prompt: Add collaboration prompts to agents + board_feedback_on: Enable board feedback on agent outputs + decision_threshold: Threshold for majority decisions (0.0-1.0) + enable_voting: Enable voting mechanisms for board decisions + enable_consensus: Enable consensus-building mechanisms + max_workers: Maximum number of workers for parallel execution + """ + + def __init__( + self, + name: str = "BoardOfDirectorsSwarm", + description: str = "Distributed task swarm with collective decision-making", + board_members: Optional[List[BoardMember]] = None, + agents: Optional[List[Union[Agent, Callable, Any]]] = None, + max_loops: int = 1, + output_type: OutputType = "dict-all-except-first", + board_model_name: str = "gpt-4o-mini", + verbose: bool = False, + add_collaboration_prompt: bool = True, + board_feedback_on: bool = True, + decision_threshold: float = 0.6, + enable_voting: bool = True, + enable_consensus: bool = True, + max_workers: Optional[int] = None, + *args: Any, + **kwargs: Any, + ) -> None: + """ + Initialize the Board of Directors Swarm with the given parameters. + + Args: + name: The name of the swarm + description: A description of the swarm + board_members: List of board members with their roles and expertise + agents: A list of agents within the swarm + max_loops: The maximum number of feedback loops between the board and agents + output_type: The format in which to return the output (dict, str, or list) + board_model_name: The model name for board member agents + verbose: Enable detailed logging with loguru + add_collaboration_prompt: Add collaboration prompts to agents + board_feedback_on: Enable board feedback on agent outputs + decision_threshold: Threshold for majority decisions (0.0-1.0) + enable_voting: Enable voting mechanisms for board decisions + enable_consensus: Enable consensus-building mechanisms + max_workers: Maximum number of workers for parallel execution + *args: Additional positional arguments passed to BaseSwarm + **kwargs: Additional keyword arguments passed to BaseSwarm + + Raises: + ValueError: If critical requirements are not met during initialization + """ + super().__init__( + name=name, + description=description, + agents=agents, + ) + + self.name = name + self.board_members = board_members or [] + self.agents = agents or [] + self.max_loops = max_loops + self.output_type = output_type + self.board_model_name = board_model_name + self.verbose = verbose + self.add_collaboration_prompt = add_collaboration_prompt + self.board_feedback_on = board_feedback_on + self.decision_threshold = decision_threshold + self.enable_voting = enable_voting + self.enable_consensus = enable_consensus + self.max_workers = max_workers or min(32, (os.cpu_count() or 1) + 4) + + # Initialize the swarm + self._init_board_swarm() + + def _init_board_swarm(self) -> None: + """ + Initialize the Board of Directors swarm. + + This method sets up the board members, initializes the conversation, + performs reliability checks, and prepares the board for operation. + + Raises: + ValueError: If reliability checks fail + """ + if self.verbose: + board_logger.info(f"šŸš€ Initializing Board of Directors Swarm: {self.name}") + board_logger.info(f"šŸ“Š Configuration - Max loops: {self.max_loops}") + + self.conversation = Conversation(time_enabled=False) + + # Perform reliability checks + self._perform_reliability_checks() + + # Setup board members if not provided + if not self.board_members: + self._setup_default_board() + + # Add context to board members + self._add_context_to_board() + + if self.verbose: + board_logger.success(f"āœ… Board of Directors Swarm initialized successfully: {self.name}") + + def _setup_default_board(self) -> None: + """ + Set up a default Board of Directors if none is provided. + + Creates a basic board structure with Chairman, Vice Chairman, and Secretary roles. + This method is called automatically if no board members are provided during initialization. + """ + if self.verbose: + board_logger.info("šŸŽÆ Setting up default Board of Directors") + + # Create default board members + chairman = Agent( + agent_name="Chairman", + agent_description="Chairman of the Board responsible for leading meetings and making final decisions", + model_name=self.board_model_name, + max_loops=1, + system_prompt=self._get_chairman_prompt(), + ) + + vice_chairman = Agent( + agent_name="Vice-Chairman", + agent_description="Vice Chairman who supports the Chairman and leads in their absence", + model_name=self.board_model_name, + max_loops=1, + system_prompt=self._get_vice_chairman_prompt(), + ) + + secretary = Agent( + agent_name="Secretary", + agent_description="Board Secretary responsible for documentation and meeting minutes", + model_name=self.board_model_name, + max_loops=1, + system_prompt=self._get_secretary_prompt(), + ) + + self.board_members = [ + BoardMember(chairman, BoardMemberRole.CHAIRMAN, 1.5, ["leadership", "strategy"]), + BoardMember(vice_chairman, BoardMemberRole.VICE_CHAIRMAN, 1.2, ["operations", "coordination"]), + BoardMember(secretary, BoardMemberRole.SECRETARY, 1.0, ["documentation", "communication"]), + ] + + if self.verbose: + board_logger.success("āœ… Default Board of Directors setup completed") + + def _get_chairman_prompt(self) -> str: + """ + Get the system prompt for the Chairman role. + + Returns: + str: The system prompt defining the Chairman's responsibilities and behavior + """ + return """You are the Chairman of the Board of Directors. Your responsibilities include: +1. Leading board meetings and discussions +2. Facilitating consensus among board members +3. Making final decisions when consensus cannot be reached +4. Ensuring all board members have an opportunity to contribute +5. Maintaining focus on the organization's goals and objectives +6. Providing strategic direction and oversight + +You should be diplomatic, fair, and decisive in your leadership.""" + + def _get_vice_chairman_prompt(self) -> str: + """ + Get the system prompt for the Vice Chairman role. + + Returns: + str: The system prompt defining the Vice Chairman's responsibilities and behavior + """ + return """You are the Vice Chairman of the Board of Directors. Your responsibilities include: +1. Supporting the Chairman in leading board meetings +2. Taking leadership when the Chairman is unavailable +3. Coordinating with other board members +4. Ensuring operational efficiency +5. Providing strategic input and analysis +6. Maintaining board cohesion and effectiveness + +You should be collaborative, analytical, and supportive in your role.""" + + def _get_secretary_prompt(self) -> str: + """ + Get the system prompt for the Secretary role. + + Returns: + str: The system prompt defining the Secretary's responsibilities and behavior + """ + return """You are the Secretary of the Board of Directors. Your responsibilities include: +1. Documenting all board meetings and decisions +2. Maintaining accurate records of board proceedings +3. Ensuring proper communication between board members +4. Tracking action items and follow-ups +5. Providing administrative support to the board +6. Ensuring compliance with governance requirements + +You should be thorough, organized, and detail-oriented in your documentation.""" + + def _add_context_to_board(self) -> None: + """ + Add agent context to all board members' conversations. + + This ensures that board members are aware of all available agents + and their capabilities when making decisions. + + Raises: + Exception: If context addition fails + """ + try: + if self.verbose: + board_logger.info("šŸ“ Adding agent context to board members") + + # Add context to each board member + for board_member in self.board_members: + list_all_agents( + agents=self.agents, + conversation=self.conversation, + add_to_conversation=True, + add_collaboration_prompt=self.add_collaboration_prompt, + ) + + if self.verbose: + board_logger.success("āœ… Agent context added to board members successfully") + + except Exception as e: + error_msg = f"āŒ Failed to add context to board members: {str(e)}" + board_logger.error(f"{error_msg}\nšŸ” Traceback: {traceback.format_exc()}") + raise + + def _perform_reliability_checks(self) -> None: + """ + Perform reliability checks for the Board of Directors swarm. + + This method validates critical requirements and configuration + parameters to ensure the swarm can operate correctly. + + Raises: + ValueError: If critical requirements are not met + """ + try: + if self.verbose: + board_logger.info(f"šŸ” Running reliability checks for swarm: {self.name}") + + if not self.agents or len(self.agents) == 0: + raise ValueError( + "No agents found in the swarm. At least one agent must be provided to create a Board of Directors swarm." + ) + + if self.max_loops <= 0: + raise ValueError( + "Max loops must be greater than 0. Please set a valid number of loops." + ) + + if self.decision_threshold < 0.0 or self.decision_threshold > 1.0: + raise ValueError( + "Decision threshold must be between 0.0 and 1.0." + ) + + if self.verbose: + board_logger.success(f"āœ… Reliability checks passed for swarm: {self.name}") + board_logger.info(f"šŸ“Š Swarm stats - Agents: {len(self.agents)}, Max loops: {self.max_loops}") + + except Exception as e: + error_msg = f"āŒ Failed reliability checks: {str(e)}\nšŸ” Traceback: {traceback.format_exc()}" + board_logger.error(error_msg) + raise + + def run_board_meeting( + self, + task: str, + img: Optional[str] = None, + ) -> BoardSpec: + """ + Run a board meeting to discuss and decide on the given task. + + This method orchestrates a complete board meeting, including discussion, + decision-making, and task distribution to worker agents. + + Args: + task: The task to be discussed and planned by the board + img: Optional image to be used with the task + + Returns: + BoardSpec: The board's plan and orders + + Raises: + Exception: If board meeting execution fails + """ + try: + if self.verbose: + board_logger.info(f"šŸ›ļø Running board meeting with task: {task[:100]}...") + + # Create board meeting prompt + meeting_prompt = self._create_board_meeting_prompt(task) + + # Run board discussion + board_discussion = self._conduct_board_discussion(meeting_prompt, img) + + # Parse board decisions + board_spec = self._parse_board_decisions(board_discussion) + + # Add to conversation history + self.conversation.add(role="Board of Directors", content=board_discussion) + + if self.verbose: + board_logger.success("āœ… Board meeting completed") + board_logger.debug(f"šŸ“‹ Board output type: {type(board_spec)}") + + return board_spec + + except Exception as e: + error_msg = f"āŒ Failed to run board meeting: {str(e)}\nšŸ” Traceback: {traceback.format_exc()}" + board_logger.error(error_msg) + raise + + def _create_board_meeting_prompt(self, task: str) -> str: + """ + Create a prompt for the board meeting. + + This method generates a comprehensive prompt that guides the board + through the meeting process, including task discussion, decision-making, + and task distribution. + + Args: + task: The task to be discussed + + Returns: + str: The board meeting prompt + """ + return f"""BOARD OF DIRECTORS MEETING + +TASK: {task} + +CONVERSATION HISTORY: {self.conversation.get_str()} + +AVAILABLE AGENTS: {[agent.agent_name for agent in self.agents]} + +BOARD MEMBERS: +{self._format_board_members_info()} + +INSTRUCTIONS: +1. Discuss the task thoroughly as a board +2. Consider all perspectives and expertise areas +3. Reach consensus or majority decision on the approach +4. Create a detailed plan for task execution +5. Assign specific tasks to appropriate agents +6. Document all decisions and reasoning + +Please provide your response in the following format: +{{ + "plan": "Detailed plan for task execution", + "orders": [ + {{ + "agent_name": "Agent Name", + "task": "Specific task description", + "priority": 1-5, + "deadline": "Optional deadline", + "assigned_by": "Board Member Name" + }} + ], + "decisions": [ + {{ + "decision_type": "unanimous/majority/consensus/chairman_decision", + "decision": "Description of the decision", + "votes_for": 0, + "votes_against": 0, + "abstentions": 0, + "reasoning": "Reasoning behind the decision" + }} + ], + "meeting_summary": "Summary of the board meeting and key outcomes" +}}""" + + def _format_board_members_info(self) -> str: + """ + Format board members information for the prompt. + + This method creates a formatted string containing information about + all board members, their roles, and expertise areas. + + Returns: + str: Formatted board members information + """ + info = [] + for member in self.board_members: + info.append(f"- {member.agent.agent_name} ({member.role.value}): {member.agent.agent_description}") + if member.expertise_areas: + info.append(f" Expertise: {', '.join(member.expertise_areas)}") + return "\n".join(info) + + def _conduct_board_discussion(self, prompt: str, img: Optional[str] = None) -> str: + """ + Conduct the board discussion using the chairman as the primary speaker. + + This method uses the chairman agent to lead the board discussion + and generate the meeting output. + + Args: + prompt: The board meeting prompt + img: Optional image input + + Returns: + str: The board discussion output + + Raises: + ValueError: If no chairman is found in board members + """ + # Use the chairman to lead the discussion + chairman = next((member.agent for member in self.board_members + if member.role == BoardMemberRole.CHAIRMAN), + self.board_members[0].agent if self.board_members else None) + + if not chairman: + raise ValueError("No chairman found in board members") + + return chairman.run(task=prompt, img=img) + + def _parse_board_decisions(self, board_output: str) -> BoardSpec: + """ + Parse the board output into a BoardSpec object. + + This method attempts to parse the board discussion output as JSON + and convert it into a structured BoardSpec object. If parsing fails, + it returns a basic BoardSpec with the raw output. + + Args: + board_output: The output from the board discussion + + Returns: + BoardSpec: Parsed board specification + """ + try: + # Try to parse as JSON first + if isinstance(board_output, str): + # Try to extract JSON from the response + json_match = re.search(r'\{.*\}', board_output, re.DOTALL) + if json_match: + board_output = json_match.group() + + parsed = json.loads(board_output) + else: + parsed = board_output + + # Extract components + plan = parsed.get("plan", "") + orders_data = parsed.get("orders", []) + decisions_data = parsed.get("decisions", []) + meeting_summary = parsed.get("meeting_summary", "") + + # Create BoardOrder objects + orders = [] + for order_data in orders_data: + order = BoardOrder( + agent_name=order_data.get("agent_name", ""), + task=order_data.get("task", ""), + priority=order_data.get("priority", 3), + deadline=order_data.get("deadline"), + assigned_by=order_data.get("assigned_by", "Board of Directors") + ) + orders.append(order) + + # Create BoardDecision objects + decisions = [] + for decision_data in decisions_data: + decision = BoardDecision( + decision_type=BoardDecisionType(decision_data.get("decision_type", "consensus")), + decision=decision_data.get("decision", ""), + votes_for=decision_data.get("votes_for", 0), + votes_against=decision_data.get("votes_against", 0), + abstentions=decision_data.get("abstentions", 0), + reasoning=decision_data.get("reasoning", "") + ) + decisions.append(decision) + + return BoardSpec( + plan=plan, + orders=orders, + decisions=decisions, + meeting_summary=meeting_summary + ) + + except Exception as e: + board_logger.error(f"Failed to parse board decisions: {str(e)}") + # Return a basic BoardSpec if parsing fails + return BoardSpec( + plan=board_output, + orders=[], + decisions=[], + meeting_summary="Parsing failed, using raw output" + ) + + def step(self, task: str, img: Optional[str] = None, *args: Any, **kwargs: Any) -> Any: + """ + Execute a single step of the Board of Directors swarm. + + This method runs one complete cycle of board meeting and task execution. + It includes board discussion, task distribution, and optional feedback. + + Args: + task: The task to be executed + img: Optional image input + *args: Additional positional arguments + **kwargs: Additional keyword arguments + + Returns: + Any: The result of the step execution + + Raises: + Exception: If step execution fails + """ + try: + if self.verbose: + board_logger.info(f"šŸ‘£ Executing single step for task: {task[:100]}...") + + # Run board meeting + board_spec = self.run_board_meeting(task=task, img=img) + + if self.verbose: + board_logger.info(f"šŸ“‹ Board created plan and {len(board_spec.orders)} orders") + + # Execute the orders + outputs = self._execute_orders(board_spec.orders) + + if self.verbose: + board_logger.info(f"āš” Executed {len(outputs)} orders") + + # Provide board feedback if enabled + if self.board_feedback_on: + feedback = self._generate_board_feedback(outputs) + else: + feedback = outputs + + if self.verbose: + board_logger.success("āœ… Step completed successfully") + + return feedback + + except Exception as e: + error_msg = f"āŒ Failed to execute step: {str(e)}\nšŸ” Traceback: {traceback.format_exc()}" + board_logger.error(error_msg) + raise + + def run(self, task: str, img: Optional[str] = None, *args: Any, **kwargs: Any) -> Any: + """ + Run the Board of Directors swarm for the specified number of loops. + + This method executes the complete swarm workflow, including multiple + iterations if max_loops is greater than 1. Each iteration includes + board meeting, task execution, and feedback generation. + + Args: + task: The task to be executed + img: Optional image input + *args: Additional positional arguments + **kwargs: Additional keyword arguments + + Returns: + Any: The final result of the swarm execution + + Raises: + Exception: If swarm execution fails + """ + try: + if self.verbose: + board_logger.info(f"šŸ›ļø Starting Board of Directors swarm execution: {self.name}") + board_logger.info(f"šŸ“‹ Task: {task[:100]}...") + + current_loop = 0 + while current_loop < self.max_loops: + if self.verbose: + board_logger.info(f"šŸ”„ Executing loop {current_loop + 1}/{self.max_loops}") + + # Execute step + result = self.step(task=task, img=img, *args, **kwargs) + + # Add to conversation + self.conversation.add(role="System", content=f"Loop {current_loop + 1} completed") + + current_loop += 1 + + if self.verbose: + board_logger.success(f"šŸŽ‰ Board of Directors swarm run completed: {self.name}") + board_logger.info(f"šŸ“Š Total loops executed: {current_loop}") + + return history_output_formatter( + conversation=self.conversation, type=self.output_type + ) + + except Exception as e: + error_msg = f"āŒ Failed to run Board of Directors swarm: {str(e)}\nšŸ” Traceback: {traceback.format_exc()}" + board_logger.error(error_msg) + raise + + async def arun(self, task: str, img: Optional[str] = None, *args: Any, **kwargs: Any) -> Any: + """ + Run the Board of Directors swarm asynchronously. + + This method provides an asynchronous interface for running the swarm, + allowing for non-blocking execution in async contexts. + + Args: + task: The task to be executed + img: Optional image input + *args: Additional positional arguments + **kwargs: Additional keyword arguments + + Returns: + Any: The final result of the swarm execution + """ + loop = asyncio.get_event_loop() + result = await loop.run_in_executor( + None, self.run, task, img, *args, **kwargs + ) + return result + + def _generate_board_feedback(self, outputs: List[Any]) -> str: + """ + Provide feedback from the Board of Directors based on agent outputs. + + This method uses the chairman to review and provide feedback on + the outputs generated by worker agents. + + Args: + outputs: List of outputs from agents + + Returns: + str: Board feedback on the outputs + + Raises: + ValueError: If no chairman is found for feedback + Exception: If feedback generation fails + """ + try: + if self.verbose: + board_logger.info("šŸ“ Generating board feedback") + + task = f"History: {self.conversation.get_str()} \n\n" + + # Use the chairman for feedback + chairman = next((member.agent for member in self.board_members + if member.role == BoardMemberRole.CHAIRMAN), + self.board_members[0].agent if self.board_members else None) + + if not chairman: + raise ValueError("No chairman found for feedback") + + feedback_prompt = ( + "You are the Chairman of the Board. Review the outputs generated by all the worker agents " + "in the previous step. Provide specific, actionable feedback for each agent, highlighting " + "strengths, weaknesses, and concrete suggestions for improvement. " + "If any outputs are unclear, incomplete, or could be enhanced, explain exactly how. " + f"Your feedback should help the agents refine their work in the next iteration. " + f"Worker Agent Responses: {task}" + ) + + output = chairman.run(task=feedback_prompt) + self.conversation.add(role=chairman.agent_name, content=output) + + if self.verbose: + board_logger.success("āœ… Board feedback generated successfully") + + return output + + except Exception as e: + error_msg = f"āŒ Failed to generate board feedback: {str(e)}\nšŸ” Traceback: {traceback.format_exc()}" + board_logger.error(error_msg) + raise + + def _call_single_agent( + self, + agent_name: str, + task: str, + *args: Any, + **kwargs: Any + ) -> Any: + """ + Call a single agent with the given task. + + This method finds and executes a specific agent with the provided task. + It includes error handling and logging for agent execution. + + Args: + agent_name: The name of the agent to call + task: The task to assign to the agent + *args: Additional positional arguments + **kwargs: Additional keyword arguments + + Returns: + Any: The output from the agent + + Raises: + ValueError: If the specified agent is not found + Exception: If agent execution fails + """ + try: + if self.verbose: + board_logger.info(f"šŸ“ž Calling agent: {agent_name}") + + # Find agent by name + agent = None + for a in self.agents: + if hasattr(a, "agent_name") and a.agent_name == agent_name: + agent = a + break + + if agent is None: + available_agents = [ + a.agent_name for a in self.agents if hasattr(a, "agent_name") + ] + raise ValueError( + f"Agent '{agent_name}' not found in swarm. Available agents: {available_agents}" + ) + + output = agent.run( + task=f"History: {self.conversation.get_str()} \n\n Task: {task}", + *args, + **kwargs, + ) + self.conversation.add(role=agent_name, content=output) + + if self.verbose: + board_logger.success(f"āœ… Agent {agent_name} completed task successfully") + + return output + + except Exception as e: + error_msg = f"āŒ Failed to call agent {agent_name}: {str(e)}\nšŸ” Traceback: {traceback.format_exc()}" + board_logger.error(error_msg) + raise + + def _execute_orders(self, orders: List[BoardOrder]) -> List[Dict[str, Any]]: + """ + Execute the orders issued by the Board of Directors. + + This method uses ThreadPoolExecutor to execute multiple orders in parallel, + improving performance for complex task distributions. + + Args: + orders: List of board orders to execute + + Returns: + List[Dict[str, Any]]: List of outputs from executed orders + + Raises: + Exception: If order execution fails + """ + try: + if self.verbose: + board_logger.info(f"āš” Executing {len(orders)} board orders") + + # Use ThreadPoolExecutor for parallel execution + outputs = [] + with ThreadPoolExecutor(max_workers=self.max_workers) as executor: + # Submit all orders for execution + future_to_order = { + executor.submit(self._execute_single_order, order): order + for order in orders + } + + # Collect results as they complete + for future in as_completed(future_to_order): + order = future_to_order[future] + try: + output = future.result() + outputs.append({ + "agent_name": order.agent_name, + "task": order.task, + "output": output, + "priority": order.priority, + "assigned_by": order.assigned_by, + }) + except Exception as e: + board_logger.error(f"Failed to execute order for {order.agent_name}: {str(e)}") + outputs.append({ + "agent_name": order.agent_name, + "task": order.task, + "output": f"Error: {str(e)}", + "priority": order.priority, + "assigned_by": order.assigned_by, + }) + + if self.verbose: + board_logger.success(f"āœ… Executed {len(outputs)} orders successfully") + + return outputs + + except Exception as e: + error_msg = f"āŒ Failed to execute orders: {str(e)}\nšŸ” Traceback: {traceback.format_exc()}" + board_logger.error(error_msg) + raise + + def _execute_single_order(self, order: BoardOrder) -> Any: + """ + Execute a single board order. + + This method is a wrapper around _call_single_agent for executing + individual board orders. + + Args: + order: The board order to execute + + Returns: + Any: The output from the executed order + """ + return self._call_single_agent( + agent_name=order.agent_name, + task=order.task, + ) + + def add_board_member(self, board_member: BoardMember) -> None: + """ + Add a new member to the Board of Directors. + + This method allows dynamic addition of board members after swarm initialization. + + Args: + board_member: The board member to add + """ + self.board_members.append(board_member) + if self.verbose: + board_logger.info(f"āœ… Added board member: {board_member.agent.agent_name}") + + def remove_board_member(self, agent_name: str) -> None: + """ + Remove a board member by agent name. + + This method allows dynamic removal of board members after swarm initialization. + + Args: + agent_name: The name of the agent to remove from the board + """ + self.board_members = [ + member for member in self.board_members + if member.agent.agent_name != agent_name + ] + if self.verbose: + board_logger.info(f"āœ… Removed board member: {agent_name}") + + def get_board_member(self, agent_name: str) -> Optional[BoardMember]: + """ + Get a board member by agent name. + + This method retrieves a specific board member by their agent name. + + Args: + agent_name: The name of the agent + + Returns: + Optional[BoardMember]: The board member if found, None otherwise + """ + for member in self.board_members: + if member.agent.agent_name == agent_name: + return member + return None + + def get_board_summary(self) -> Dict[str, Any]: + """ + Get a summary of the Board of Directors. + + This method provides a comprehensive summary of the board structure, + including member information, configuration, and statistics. + + Returns: + Dict[str, Any]: Summary of the board structure and members + """ + return { + "board_name": self.name, + "total_members": len(self.board_members), + "members": [ + { + "name": member.agent.agent_name, + "role": member.role.value, + "voting_weight": member.voting_weight, + "expertise_areas": member.expertise_areas, + } + for member in self.board_members + ], + "total_agents": len(self.agents), + "max_loops": self.max_loops, + "decision_threshold": self.decision_threshold, + } \ No newline at end of file diff --git a/tests/structs/test_board_of_directors_swarm.py b/tests/structs/test_board_of_directors_swarm.py new file mode 100644 index 00000000..4b2c1ddd --- /dev/null +++ b/tests/structs/test_board_of_directors_swarm.py @@ -0,0 +1,1020 @@ +""" +Comprehensive test suite for Board of Directors Swarm. + +This module contains extensive tests for the Board of Directors swarm implementation, +covering all aspects including initialization, board operations, task execution, +error handling, and performance characteristics. + +The test suite follows the Swarms testing philosophy: +- Comprehensive coverage of all functionality +- Proper mocking and isolation +- Performance and integration testing +- Error handling validation +""" + +import os +import pytest +import asyncio +from unittest.mock import Mock, patch, MagicMock, AsyncMock +from typing import List, Dict, Any, Optional + +from swarms.structs.board_of_directors_swarm import ( + BoardOfDirectorsSwarm, + BoardMember, + BoardMemberRole, + BoardDecisionType, + BoardOrder, + BoardDecision, + BoardSpec, +) +from swarms.structs.agent import Agent +from swarms.structs.conversation import Conversation + + +# Test fixtures +@pytest.fixture +def mock_agent(): + """Create a mock agent for testing.""" + agent = Mock(spec=Agent) + agent.agent_name = "TestAgent" + agent.agent_description = "A test agent for unit testing" + agent.run = Mock(return_value="Test agent response") + agent.arun = AsyncMock(return_value="Async test agent response") + return agent + + +@pytest.fixture +def mock_board_member(mock_agent): + """Create a mock board member for testing.""" + return BoardMember( + agent=mock_agent, + role=BoardMemberRole.CHAIRMAN, + voting_weight=1.5, + expertise_areas=["leadership", "strategy"] + ) + + +@pytest.fixture +def sample_agents(): + """Create sample agents for testing.""" + agents = [] + for i in range(3): + agent = Mock(spec=Agent) + agent.agent_name = f"Agent{i+1}" + agent.agent_description = f"Test agent {i+1}" + agent.run = Mock(return_value=f"Response from Agent{i+1}") + agents.append(agent) + return agents + + +@pytest.fixture +def sample_board_members(sample_agents): + """Create sample board members for testing.""" + roles = [BoardMemberRole.CHAIRMAN, BoardMemberRole.VICE_CHAIRMAN, BoardMemberRole.SECRETARY] + board_members = [] + + for i, (agent, role) in enumerate(zip(sample_agents, roles)): + board_member = BoardMember( + agent=agent, + role=role, + voting_weight=1.0 + (i * 0.2), + expertise_areas=[f"expertise_{i+1}"] + ) + board_members.append(board_member) + + return board_members + + +@pytest.fixture +def basic_board_swarm(sample_agents): + """Create a basic Board of Directors swarm for testing.""" + return BoardOfDirectorsSwarm( + name="TestBoard", + agents=sample_agents, + verbose=False, + max_loops=1 + ) + + +@pytest.fixture +def configured_board_swarm(sample_agents, sample_board_members): + """Create a configured Board of Directors swarm for testing.""" + return BoardOfDirectorsSwarm( + name="ConfiguredBoard", + description="A configured board for testing", + board_members=sample_board_members, + agents=sample_agents, + max_loops=2, + verbose=True, + decision_threshold=0.7, + enable_voting=True, + enable_consensus=True, + max_workers=4 + ) + + +# Unit tests for enums and data models +class TestBoardMemberRole: + """Test BoardMemberRole enum.""" + + def test_enum_values(self): + """Test that all enum values are correctly defined.""" + assert BoardMemberRole.CHAIRMAN == "chairman" + assert BoardMemberRole.VICE_CHAIRMAN == "vice_chairman" + assert BoardMemberRole.SECRETARY == "secretary" + assert BoardMemberRole.TREASURER == "treasurer" + assert BoardMemberRole.MEMBER == "member" + assert BoardMemberRole.EXECUTIVE_DIRECTOR == "executive_director" + + +class TestBoardDecisionType: + """Test BoardDecisionType enum.""" + + def test_enum_values(self): + """Test that all enum values are correctly defined.""" + assert BoardDecisionType.UNANIMOUS == "unanimous" + assert BoardDecisionType.MAJORITY == "majority" + assert BoardDecisionType.CONSENSUS == "consensus" + assert BoardDecisionType.CHAIRMAN_DECISION == "chairman_decision" + + +class TestBoardMember: + """Test BoardMember dataclass.""" + + def test_board_member_creation(self, mock_agent): + """Test creating a board member.""" + board_member = BoardMember( + agent=mock_agent, + role=BoardMemberRole.CHAIRMAN, + voting_weight=1.5, + expertise_areas=["leadership", "strategy"] + ) + + assert board_member.agent == mock_agent + assert board_member.role == BoardMemberRole.CHAIRMAN + assert board_member.voting_weight == 1.5 + assert board_member.expertise_areas == ["leadership", "strategy"] + + def test_board_member_defaults(self, mock_agent): + """Test board member with default values.""" + board_member = BoardMember( + agent=mock_agent, + role=BoardMemberRole.MEMBER + ) + + assert board_member.voting_weight == 1.0 + assert board_member.expertise_areas == [] + + def test_board_member_post_init(self, mock_agent): + """Test board member post-init with None expertise areas.""" + board_member = BoardMember( + agent=mock_agent, + role=BoardMemberRole.MEMBER, + expertise_areas=None + ) + + assert board_member.expertise_areas == [] + + +class TestBoardOrder: + """Test BoardOrder model.""" + + def test_board_order_creation(self): + """Test creating a board order.""" + order = BoardOrder( + agent_name="TestAgent", + task="Test task", + priority=1, + deadline="2024-01-01", + assigned_by="Chairman" + ) + + assert order.agent_name == "TestAgent" + assert order.task == "Test task" + assert order.priority == 1 + assert order.deadline == "2024-01-01" + assert order.assigned_by == "Chairman" + + def test_board_order_defaults(self): + """Test board order with default values.""" + order = BoardOrder( + agent_name="TestAgent", + task="Test task" + ) + + assert order.priority == 3 + assert order.deadline is None + assert order.assigned_by == "Board of Directors" + + def test_board_order_validation(self): + """Test board order validation.""" + # Test priority validation + with pytest.raises(ValueError): + BoardOrder( + agent_name="TestAgent", + task="Test task", + priority=0 # Invalid priority + ) + + with pytest.raises(ValueError): + BoardOrder( + agent_name="TestAgent", + task="Test task", + priority=6 # Invalid priority + ) + + +class TestBoardDecision: + """Test BoardDecision model.""" + + def test_board_decision_creation(self): + """Test creating a board decision.""" + decision = BoardDecision( + decision_type=BoardDecisionType.MAJORITY, + decision="Approve the proposal", + votes_for=3, + votes_against=1, + abstentions=0, + reasoning="The proposal aligns with our strategic goals" + ) + + assert decision.decision_type == BoardDecisionType.MAJORITY + assert decision.decision == "Approve the proposal" + assert decision.votes_for == 3 + assert decision.votes_against == 1 + assert decision.abstentions == 0 + assert decision.reasoning == "The proposal aligns with our strategic goals" + + def test_board_decision_defaults(self): + """Test board decision with default values.""" + decision = BoardDecision( + decision_type=BoardDecisionType.CONSENSUS, + decision="Test decision" + ) + + assert decision.votes_for == 0 + assert decision.votes_against == 0 + assert decision.abstentions == 0 + assert decision.reasoning == "" + + +class TestBoardSpec: + """Test BoardSpec model.""" + + def test_board_spec_creation(self): + """Test creating a board spec.""" + orders = [ + BoardOrder(agent_name="Agent1", task="Task 1"), + BoardOrder(agent_name="Agent2", task="Task 2") + ] + decisions = [ + BoardDecision( + decision_type=BoardDecisionType.MAJORITY, + decision="Decision 1" + ) + ] + + spec = BoardSpec( + plan="Test plan", + orders=orders, + decisions=decisions, + meeting_summary="Test meeting summary" + ) + + assert spec.plan == "Test plan" + assert len(spec.orders) == 2 + assert len(spec.decisions) == 1 + assert spec.meeting_summary == "Test meeting summary" + + def test_board_spec_defaults(self): + """Test board spec with default values.""" + spec = BoardSpec( + plan="Test plan", + orders=[] + ) + + assert spec.decisions == [] + assert spec.meeting_summary == "" + + +# Unit tests for BoardOfDirectorsSwarm +class TestBoardOfDirectorsSwarmInitialization: + """Test BoardOfDirectorsSwarm initialization.""" + + def test_basic_initialization(self, sample_agents): + """Test basic swarm initialization.""" + swarm = BoardOfDirectorsSwarm( + name="TestSwarm", + agents=sample_agents + ) + + assert swarm.name == "TestSwarm" + assert len(swarm.agents) == 3 + assert swarm.max_loops == 1 + assert swarm.verbose is False + assert swarm.decision_threshold == 0.6 + + def test_configured_initialization(self, sample_agents, sample_board_members): + """Test configured swarm initialization.""" + swarm = BoardOfDirectorsSwarm( + name="ConfiguredSwarm", + description="Test description", + board_members=sample_board_members, + agents=sample_agents, + max_loops=3, + verbose=True, + decision_threshold=0.8, + enable_voting=False, + enable_consensus=False, + max_workers=8 + ) + + assert swarm.name == "ConfiguredSwarm" + assert swarm.description == "Test description" + assert len(swarm.board_members) == 3 + assert len(swarm.agents) == 3 + assert swarm.max_loops == 3 + assert swarm.verbose is True + assert swarm.decision_threshold == 0.8 + assert swarm.enable_voting is False + assert swarm.enable_consensus is False + assert swarm.max_workers == 8 + + def test_default_board_setup(self, sample_agents): + """Test default board setup when no board members provided.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents) + + assert len(swarm.board_members) == 3 + assert swarm.board_members[0].role == BoardMemberRole.CHAIRMAN + assert swarm.board_members[1].role == BoardMemberRole.VICE_CHAIRMAN + assert swarm.board_members[2].role == BoardMemberRole.SECRETARY + + def test_initialization_without_agents(self): + """Test initialization without agents should raise error.""" + with pytest.raises(ValueError, match="No agents found in the swarm"): + BoardOfDirectorsSwarm(agents=[]) + + def test_initialization_with_invalid_max_loops(self, sample_agents): + """Test initialization with invalid max_loops.""" + with pytest.raises(ValueError, match="Max loops must be greater than 0"): + BoardOfDirectorsSwarm(agents=sample_agents, max_loops=0) + + def test_initialization_with_invalid_decision_threshold(self, sample_agents): + """Test initialization with invalid decision threshold.""" + with pytest.raises(ValueError, match="Decision threshold must be between 0.0 and 1.0"): + BoardOfDirectorsSwarm(agents=sample_agents, decision_threshold=1.5) + + +class TestBoardOfDirectorsSwarmMethods: + """Test BoardOfDirectorsSwarm methods.""" + + def test_setup_default_board(self, sample_agents): + """Test default board setup.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents) + + assert len(swarm.board_members) == 3 + assert all(hasattr(member.agent, 'agent_name') for member in swarm.board_members) + assert all(hasattr(member.agent, 'run') for member in swarm.board_members) + + def test_get_chairman_prompt(self, sample_agents): + """Test chairman prompt generation.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents) + prompt = swarm._get_chairman_prompt() + + assert "Chairman" in prompt + assert "board meetings" in prompt + assert "consensus" in prompt + + def test_get_vice_chairman_prompt(self, sample_agents): + """Test vice chairman prompt generation.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents) + prompt = swarm._get_vice_chairman_prompt() + + assert "Vice Chairman" in prompt + assert "supporting" in prompt + assert "operational" in prompt + + def test_get_secretary_prompt(self, sample_agents): + """Test secretary prompt generation.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents) + prompt = swarm._get_secretary_prompt() + + assert "Secretary" in prompt + assert "documenting" in prompt + assert "records" in prompt + + def test_format_board_members_info(self, configured_board_swarm): + """Test board members info formatting.""" + info = configured_board_swarm._format_board_members_info() + + assert "Chairman" in info + assert "Vice-Chairman" in info + assert "Secretary" in info + assert "expertise" in info + + def test_add_board_member(self, basic_board_swarm, mock_board_member): + """Test adding a board member.""" + initial_count = len(basic_board_swarm.board_members) + basic_board_swarm.add_board_member(mock_board_member) + + assert len(basic_board_swarm.board_members) == initial_count + 1 + assert mock_board_member in basic_board_swarm.board_members + + def test_remove_board_member(self, configured_board_swarm): + """Test removing a board member.""" + member_to_remove = configured_board_swarm.board_members[0] + member_name = member_to_remove.agent.agent_name + + initial_count = len(configured_board_swarm.board_members) + configured_board_swarm.remove_board_member(member_name) + + assert len(configured_board_swarm.board_members) == initial_count - 1 + assert member_to_remove not in configured_board_swarm.board_members + + def test_get_board_member(self, configured_board_swarm): + """Test getting a board member by name.""" + member = configured_board_swarm.board_members[0] + member_name = member.agent.agent_name + + found_member = configured_board_swarm.get_board_member(member_name) + assert found_member == member + + # Test with non-existent member + not_found = configured_board_swarm.get_board_member("NonExistent") + assert not_found is None + + def test_get_board_summary(self, configured_board_swarm): + """Test getting board summary.""" + summary = configured_board_swarm.get_board_summary() + + assert "board_name" in summary + assert "total_members" in summary + assert "total_agents" in summary + assert "max_loops" in summary + assert "decision_threshold" in summary + assert "members" in summary + + assert summary["board_name"] == "ConfiguredBoard" + assert summary["total_members"] == 3 + assert summary["total_agents"] == 3 + + +class TestBoardMeetingOperations: + """Test board meeting operations.""" + + def test_create_board_meeting_prompt(self, configured_board_swarm): + """Test board meeting prompt creation.""" + task = "Test task for board meeting" + prompt = configured_board_swarm._create_board_meeting_prompt(task) + + assert task in prompt + assert "BOARD OF DIRECTORS MEETING" in prompt + assert "INSTRUCTIONS" in prompt + assert "plan" in prompt + assert "orders" in prompt + + def test_conduct_board_discussion(self, configured_board_swarm): + """Test board discussion conduction.""" + prompt = "Test board meeting prompt" + + with patch.object(configured_board_swarm.board_members[0].agent, 'run') as mock_run: + mock_run.return_value = "Board discussion result" + result = configured_board_swarm._conduct_board_discussion(prompt) + + assert result == "Board discussion result" + mock_run.assert_called_once_with(task=prompt, img=None) + + def test_conduct_board_discussion_no_chairman(self, sample_agents): + """Test board discussion when no chairman is found.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents) + # Remove all board members + swarm.board_members = [] + + with pytest.raises(ValueError, match="No chairman found in board members"): + swarm._conduct_board_discussion("Test prompt") + + def test_parse_board_decisions_valid_json(self, configured_board_swarm): + """Test parsing valid JSON board decisions.""" + valid_json = """ + { + "plan": "Test plan", + "orders": [ + { + "agent_name": "Agent1", + "task": "Task 1", + "priority": 1, + "assigned_by": "Chairman" + } + ], + "decisions": [ + { + "decision_type": "majority", + "decision": "Test decision", + "votes_for": 2, + "votes_against": 1, + "abstentions": 0, + "reasoning": "Test reasoning" + } + ], + "meeting_summary": "Test summary" + } + """ + + result = configured_board_swarm._parse_board_decisions(valid_json) + + assert isinstance(result, BoardSpec) + assert result.plan == "Test plan" + assert len(result.orders) == 1 + assert len(result.decisions) == 1 + assert result.meeting_summary == "Test summary" + + def test_parse_board_decisions_invalid_json(self, configured_board_swarm): + """Test parsing invalid JSON board decisions.""" + invalid_json = "Invalid JSON content" + + result = configured_board_swarm._parse_board_decisions(invalid_json) + + assert isinstance(result, BoardSpec) + assert result.plan == invalid_json + assert len(result.orders) == 0 + assert len(result.decisions) == 0 + assert result.meeting_summary == "Parsing failed, using raw output" + + def test_run_board_meeting(self, configured_board_swarm): + """Test running a complete board meeting.""" + task = "Test board meeting task" + + with patch.object(configured_board_swarm, '_conduct_board_discussion') as mock_discuss: + with patch.object(configured_board_swarm, '_parse_board_decisions') as mock_parse: + mock_discuss.return_value = "Board discussion" + mock_parse.return_value = BoardSpec( + plan="Test plan", + orders=[], + decisions=[], + meeting_summary="Test summary" + ) + + result = configured_board_swarm.run_board_meeting(task) + + assert isinstance(result, BoardSpec) + mock_discuss.assert_called_once() + mock_parse.assert_called_once_with("Board discussion") + + +class TestTaskExecution: + """Test task execution methods.""" + + def test_call_single_agent(self, configured_board_swarm): + """Test calling a single agent.""" + agent_name = "Agent1" + task = "Test task" + + with patch.object(configured_board_swarm.agents[0], 'run') as mock_run: + mock_run.return_value = "Agent response" + result = configured_board_swarm._call_single_agent(agent_name, task) + + assert result == "Agent response" + mock_run.assert_called_once() + + def test_call_single_agent_not_found(self, configured_board_swarm): + """Test calling a non-existent agent.""" + with pytest.raises(ValueError, match="Agent 'NonExistent' not found"): + configured_board_swarm._call_single_agent("NonExistent", "Test task") + + def test_execute_single_order(self, configured_board_swarm): + """Test executing a single order.""" + order = BoardOrder( + agent_name="Agent1", + task="Test order task", + priority=1, + assigned_by="Chairman" + ) + + with patch.object(configured_board_swarm, '_call_single_agent') as mock_call: + mock_call.return_value = "Order execution result" + result = configured_board_swarm._execute_single_order(order) + + assert result == "Order execution result" + mock_call.assert_called_once_with( + agent_name="Agent1", + task="Test order task" + ) + + def test_execute_orders(self, configured_board_swarm): + """Test executing multiple orders.""" + orders = [ + BoardOrder(agent_name="Agent1", task="Task 1", priority=1), + BoardOrder(agent_name="Agent2", task="Task 2", priority=2), + ] + + with patch.object(configured_board_swarm, '_execute_single_order') as mock_execute: + mock_execute.side_effect = ["Result 1", "Result 2"] + results = configured_board_swarm._execute_orders(orders) + + assert len(results) == 2 + assert results[0]["agent_name"] == "Agent1" + assert results[0]["output"] == "Result 1" + assert results[1]["agent_name"] == "Agent2" + assert results[1]["output"] == "Result 2" + + def test_generate_board_feedback(self, configured_board_swarm): + """Test generating board feedback.""" + outputs = [ + {"agent_name": "Agent1", "output": "Output 1"}, + {"agent_name": "Agent2", "output": "Output 2"} + ] + + with patch.object(configured_board_swarm.board_members[0].agent, 'run') as mock_run: + mock_run.return_value = "Board feedback" + result = configured_board_swarm._generate_board_feedback(outputs) + + assert result == "Board feedback" + mock_run.assert_called_once() + + def test_generate_board_feedback_no_chairman(self, sample_agents): + """Test generating feedback when no chairman is found.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents) + swarm.board_members = [] # Remove all board members + + with pytest.raises(ValueError, match="No chairman found for feedback"): + swarm._generate_board_feedback([]) + + +class TestStepAndRunMethods: + """Test step and run methods.""" + + def test_step_method(self, configured_board_swarm): + """Test the step method.""" + task = "Test step task" + + with patch.object(configured_board_swarm, 'run_board_meeting') as mock_meeting: + with patch.object(configured_board_swarm, '_execute_orders') as mock_execute: + with patch.object(configured_board_swarm, '_generate_board_feedback') as mock_feedback: + mock_meeting.return_value = BoardSpec( + plan="Test plan", + orders=[BoardOrder(agent_name="Agent1", task="Task 1")], + decisions=[], + meeting_summary="Test summary" + ) + mock_execute.return_value = [{"agent_name": "Agent1", "output": "Result"}] + mock_feedback.return_value = "Board feedback" + + result = configured_board_swarm.step(task) + + assert result == "Board feedback" + mock_meeting.assert_called_once_with(task=task, img=None) + mock_execute.assert_called_once() + mock_feedback.assert_called_once() + + def test_step_method_no_feedback(self, configured_board_swarm): + """Test the step method with feedback disabled.""" + configured_board_swarm.board_feedback_on = False + task = "Test step task" + + with patch.object(configured_board_swarm, 'run_board_meeting') as mock_meeting: + with patch.object(configured_board_swarm, '_execute_orders') as mock_execute: + mock_meeting.return_value = BoardSpec( + plan="Test plan", + orders=[BoardOrder(agent_name="Agent1", task="Task 1")], + decisions=[], + meeting_summary="Test summary" + ) + mock_execute.return_value = [{"agent_name": "Agent1", "output": "Result"}] + + result = configured_board_swarm.step(task) + + assert result == [{"agent_name": "Agent1", "output": "Result"}] + + def test_run_method(self, configured_board_swarm): + """Test the run method.""" + task = "Test run task" + + with patch.object(configured_board_swarm, 'step') as mock_step: + with patch.object(configured_board_swarm, 'conversation') as mock_conversation: + mock_step.return_value = "Step result" + mock_conversation.add = Mock() + + result = configured_board_swarm.run(task) + + assert mock_step.call_count == 2 # max_loops = 2 + assert mock_conversation.add.call_count == 2 + + def test_arun_method(self, configured_board_swarm): + """Test the async run method.""" + task = "Test async run task" + + with patch.object(configured_board_swarm, 'run') as mock_run: + mock_run.return_value = "Async result" + + async def test_async(): + result = await configured_board_swarm.arun(task) + return result + + result = asyncio.run(test_async()) + assert result == "Async result" + mock_run.assert_called_once_with(task=task, img=None) + + +# Integration tests +class TestBoardOfDirectorsSwarmIntegration: + """Integration tests for BoardOfDirectorsSwarm.""" + + def test_full_workflow_integration(self, sample_agents): + """Test full workflow integration.""" + swarm = BoardOfDirectorsSwarm( + agents=sample_agents, + verbose=False, + max_loops=1 + ) + + task = "Create a simple report" + + # Mock the board discussion to return structured output + mock_board_output = """ + { + "plan": "Create a comprehensive report", + "orders": [ + { + "agent_name": "Agent1", + "task": "Research the topic", + "priority": 1, + "assigned_by": "Chairman" + }, + { + "agent_name": "Agent2", + "task": "Write the report", + "priority": 2, + "assigned_by": "Chairman" + } + ], + "decisions": [ + { + "decision_type": "consensus", + "decision": "Proceed with report creation", + "votes_for": 3, + "votes_against": 0, + "abstentions": 0, + "reasoning": "Report is needed for decision making" + } + ], + "meeting_summary": "Board agreed to create a comprehensive report" + } + """ + + with patch.object(swarm.board_members[0].agent, 'run') as mock_run: + mock_run.return_value = mock_board_output + result = swarm.run(task) + + assert result is not None + assert isinstance(result, dict) + + def test_board_member_management_integration(self, sample_agents): + """Test board member management integration.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents) + + # Test adding a new board member + new_member = BoardMember( + agent=sample_agents[0], + role=BoardMemberRole.MEMBER, + voting_weight=1.0, + expertise_areas=["testing"] + ) + + initial_count = len(swarm.board_members) + swarm.add_board_member(new_member) + assert len(swarm.board_members) == initial_count + 1 + + # Test removing a board member + member_name = swarm.board_members[0].agent.agent_name + swarm.remove_board_member(member_name) + assert len(swarm.board_members) == initial_count + + # Test getting board member + member = swarm.get_board_member(swarm.board_members[0].agent.agent_name) + assert member is not None + + +# Parameterized tests +@pytest.mark.parametrize("max_loops", [1, 2, 3]) +def test_max_loops_parameterization(sample_agents, max_loops): + """Test swarm with different max_loops values.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents, max_loops=max_loops) + assert swarm.max_loops == max_loops + + +@pytest.mark.parametrize("decision_threshold", [0.5, 0.6, 0.7, 0.8, 0.9]) +def test_decision_threshold_parameterization(sample_agents, decision_threshold): + """Test swarm with different decision threshold values.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents, decision_threshold=decision_threshold) + assert swarm.decision_threshold == decision_threshold + + +@pytest.mark.parametrize("board_model", ["gpt-4o-mini", "gpt-4", "claude-3-sonnet"]) +def test_board_model_parameterization(sample_agents, board_model): + """Test swarm with different board models.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents, board_model_name=board_model) + assert swarm.board_model_name == board_model + + +# Error handling tests +class TestBoardOfDirectorsSwarmErrorHandling: + """Test error handling in BoardOfDirectorsSwarm.""" + + def test_initialization_error_handling(self): + """Test error handling during initialization.""" + with pytest.raises(ValueError): + BoardOfDirectorsSwarm(agents=[]) + + def test_board_meeting_error_handling(self, configured_board_swarm): + """Test error handling during board meeting.""" + with patch.object(configured_board_swarm, '_conduct_board_discussion') as mock_discuss: + mock_discuss.side_effect = Exception("Board meeting failed") + + with pytest.raises(Exception, match="Board meeting failed"): + configured_board_swarm.run_board_meeting("Test task") + + def test_task_execution_error_handling(self, configured_board_swarm): + """Test error handling during task execution.""" + with patch.object(configured_board_swarm, '_call_single_agent') as mock_call: + mock_call.side_effect = Exception("Task execution failed") + + with pytest.raises(Exception, match="Task execution failed"): + configured_board_swarm._call_single_agent("Agent1", "Test task") + + def test_order_execution_error_handling(self, configured_board_swarm): + """Test error handling during order execution.""" + orders = [BoardOrder(agent_name="Agent1", task="Task 1")] + + with patch.object(configured_board_swarm, '_execute_single_order') as mock_execute: + mock_execute.side_effect = Exception("Order execution failed") + + # Should not raise exception, but log error + results = configured_board_swarm._execute_orders(orders) + assert len(results) == 1 + assert "Error" in results[0]["output"] + + +# Performance tests +class TestBoardOfDirectorsSwarmPerformance: + """Test performance characteristics of BoardOfDirectorsSwarm.""" + + def test_parallel_execution_performance(self, sample_agents): + """Test parallel execution performance.""" + import time + + swarm = BoardOfDirectorsSwarm( + agents=sample_agents, + max_workers=3, + verbose=False + ) + + # Create multiple orders + orders = [ + BoardOrder(agent_name=f"Agent{i+1}", task=f"Task {i+1}") + for i in range(3) + ] + + start_time = time.time() + + with patch.object(swarm, '_execute_single_order') as mock_execute: + mock_execute.side_effect = lambda order: f"Result for {order.task}" + results = swarm._execute_orders(orders) + + end_time = time.time() + execution_time = end_time - start_time + + assert len(results) == 3 + assert execution_time < 1.0 # Should complete quickly with parallel execution + + def test_memory_usage(self, sample_agents): + """Test memory usage characteristics.""" + import psutil + import os + + process = psutil.Process(os.getpid()) + initial_memory = process.memory_info().rss + + # Create multiple swarms + swarms = [] + for i in range(5): + swarm = BoardOfDirectorsSwarm( + agents=sample_agents, + name=f"Swarm{i}", + verbose=False + ) + swarms.append(swarm) + + final_memory = process.memory_info().rss + memory_increase = final_memory - initial_memory + + # Memory increase should be reasonable (less than 100MB) + assert memory_increase < 100 * 1024 * 1024 + + +# Configuration tests +class TestBoardOfDirectorsSwarmConfiguration: + """Test configuration options for BoardOfDirectorsSwarm.""" + + def test_verbose_configuration(self, sample_agents): + """Test verbose configuration.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents, verbose=True) + assert swarm.verbose is True + + swarm = BoardOfDirectorsSwarm(agents=sample_agents, verbose=False) + assert swarm.verbose is False + + def test_collaboration_prompt_configuration(self, sample_agents): + """Test collaboration prompt configuration.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents, add_collaboration_prompt=True) + assert swarm.add_collaboration_prompt is True + + swarm = BoardOfDirectorsSwarm(agents=sample_agents, add_collaboration_prompt=False) + assert swarm.add_collaboration_prompt is False + + def test_board_feedback_configuration(self, sample_agents): + """Test board feedback configuration.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents, board_feedback_on=True) + assert swarm.board_feedback_on is True + + swarm = BoardOfDirectorsSwarm(agents=sample_agents, board_feedback_on=False) + assert swarm.board_feedback_on is False + + def test_voting_configuration(self, sample_agents): + """Test voting configuration.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents, enable_voting=True) + assert swarm.enable_voting is True + + swarm = BoardOfDirectorsSwarm(agents=sample_agents, enable_voting=False) + assert swarm.enable_voting is False + + def test_consensus_configuration(self, sample_agents): + """Test consensus configuration.""" + swarm = BoardOfDirectorsSwarm(agents=sample_agents, enable_consensus=True) + assert swarm.enable_consensus is True + + swarm = BoardOfDirectorsSwarm(agents=sample_agents, enable_consensus=False) + assert swarm.enable_consensus is False + + +# Real integration tests (skipped if no API key) +@pytest.mark.skipif( + not os.getenv("OPENAI_API_KEY"), + reason="OpenAI API key not available" +) +class TestBoardOfDirectorsSwarmRealIntegration: + """Real integration tests for BoardOfDirectorsSwarm.""" + + def test_real_board_meeting(self): + """Test real board meeting with actual API calls.""" + # Create real agents + agents = [ + Agent( + agent_name="Researcher", + agent_description="Research analyst", + model_name="gpt-4o-mini", + max_loops=1 + ), + Agent( + agent_name="Writer", + agent_description="Content writer", + model_name="gpt-4o-mini", + max_loops=1 + ) + ] + + swarm = BoardOfDirectorsSwarm( + agents=agents, + verbose=False, + max_loops=1 + ) + + task = "Create a brief market analysis report" + + result = swarm.run(task) + + assert result is not None + assert isinstance(result, dict) + assert "conversation_history" in result + + def test_real_board_member_management(self): + """Test real board member management.""" + agents = [ + Agent( + agent_name="TestAgent", + agent_description="Test agent", + model_name="gpt-4o-mini", + max_loops=1 + ) + ] + + swarm = BoardOfDirectorsSwarm(agents=agents, verbose=False) + + # Test board summary + summary = swarm.get_board_summary() + assert summary["total_members"] == 3 # Default board + assert summary["total_agents"] == 1 + + +# Test runner +if __name__ == "__main__": + pytest.main([__file__, "-v", "--tb=short"]) \ No newline at end of file