From 6d70c32cb51a60dbe745f77ebbae972e685762f1 Mon Sep 17 00:00:00 2001 From: CI-DEV <154627941+IlumCI@users.noreply.github.com> Date: Tue, 29 Jul 2025 00:49:28 +0300 Subject: [PATCH] Update board_of_directors_swarm.py --- swarms/structs/board_of_directors_swarm.py | 594 ++++++++++++++++++++- 1 file changed, 593 insertions(+), 1 deletion(-) diff --git a/swarms/structs/board_of_directors_swarm.py b/swarms/structs/board_of_directors_swarm.py index a6d24794..f80fb4a4 100644 --- a/swarms/structs/board_of_directors_swarm.py +++ b/swarms/structs/board_of_directors_swarm.py @@ -27,6 +27,8 @@ import traceback from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from enum import Enum +from functools import lru_cache +from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Union, Tuple from loguru import logger @@ -44,6 +46,588 @@ from swarms.utils.output_types import OutputType board_logger = initialize_logger(log_folder="board_of_directors_swarm") +# ============================================================================ +# BOARD OF DIRECTORS CONFIGURATION +# ============================================================================ + +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") + + +# ============================================================================ +# BOARD OF DIRECTORS IMPLEMENTATION +# ============================================================================ + class BoardMemberRole(str, Enum): """Enumeration of possible board member roles. @@ -504,6 +1088,14 @@ You should be thorough, organized, and detail-oriented in your documentation.""" if self.verbose: board_logger.info(f"🔍 Running reliability checks for swarm: {self.name}") + # Check if Board of Directors feature is enabled + board_config = get_board_config() + if not board_config.is_enabled(): + raise ValueError( + "Board of Directors feature is not enabled. Please enable it using " + "enable_board_feature() or set SWARMS_BOARD_FEATURE_ENABLED=true environment variable." + ) + 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." @@ -1141,4 +1733,4 @@ Please provide your response in the following format: "total_agents": len(self.agents), "max_loops": self.max_loops, "decision_threshold": self.decision_threshold, - } \ No newline at end of file + }