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

[FIX][Board of Directors] [CLEANUP __INITS__] [New Examples for livestream]

Browse Source
pull/1043/head
Kye Gomez 2 weeks ago
parent 3d4aa7a932
commit 582edf427f
12 changed files with 941 additions and 899 deletions
Show Stats Download Patch File Download Diff File

31
example.py
Unescape View File

@ -4,40 +4,11 @@ from swarms import Agent
agent = Agent(
agent_name="Quantitative-Trading-Agent",
agent_description="Advanced quantitative trading and algorithmic analysis agent",
system_prompt="""You are an expert quantitative trading agent with deep expertise in:
- Algorithmic trading strategies and implementation
- Statistical arbitrage and market making
- Risk management and portfolio optimization
- High-frequency trading systems
- Market microstructure analysis
- Quantitative research methodologies
- Financial mathematics and stochastic processes
- Machine learning applications in trading
Your core responsibilities include:
1. Developing and backtesting trading strategies
2. Analyzing market data and identifying alpha opportunities
3. Implementing risk management frameworks
4. Optimizing portfolio allocations
5. Conducting quantitative research
6. Monitoring market microstructure
7. Evaluating trading system performance
You maintain strict adherence to:
- Mathematical rigor in all analyses
- Statistical significance in strategy development
- Risk-adjusted return optimization
- Market impact minimization
- Regulatory compliance
- Transaction cost analysis
- Performance attribution
You communicate in precise, technical terms while maintaining clarity for stakeholders.""",
model_name="claude-sonnet-4-20250514",
dynamic_temperature_enabled=True,
output_type="str-all-except-first",
max_loops=1,
dynamic_context_window=True,
streaming_on=True,
)
out = agent.run(

83
examples/multi_agent/board_of_directors/board_of_directors_example.py
Unescape View File

@ -10,23 +10,8 @@ To run this example:
2. Run: python examples/multi_agent/board_of_directors/board_of_directors_example.py
"""
import os
import sys
from typing import List
# Add the root directory to the Python path if running from examples directory
current_dir = os.path.dirname(os.path.abspath(__file__))
if "examples" in current_dir:
root_dir = current_dir
while os.path.basename(
root_dir
) != "examples" and root_dir != os.path.dirname(root_dir):
root_dir = os.path.dirname(root_dir)
if os.path.basename(root_dir) == "examples":
root_dir = os.path.dirname(root_dir)
if root_dir not in sys.path:
sys.path.insert(0, root_dir)
from swarms.structs.board_of_directors_swarm import (
BoardOfDirectorsSwarm,
BoardMember,
@ -37,7 +22,6 @@ from swarms.structs.agent import Agent
def create_board_members() -> List[BoardMember]:
"""Create board members with specific roles."""
chairman = Agent(
agent_name="Chairman",
agent_description="Executive Chairman with strategic vision",
@ -86,7 +70,6 @@ def create_board_members() -> List[BoardMember]:
def create_worker_agents() -> List[Agent]:
"""Create worker agents for the swarm."""
researcher = Agent(
agent_name="Researcher",
agent_description="Research analyst for data analysis",
@ -114,9 +97,8 @@ def create_worker_agents() -> List[Agent]:
return [researcher, developer, marketer]
def run_board_example() -> None:
def run_board_example() -> str:
"""Run a Board of Directors example."""
# Create board members and worker agents
board_members = create_board_members()
worker_agents = create_worker_agents()
@ -127,7 +109,7 @@ def run_board_example() -> None:
board_members=board_members,
agents=worker_agents,
max_loops=2,
verbose=True,
verbose=False,
decision_threshold=0.6,
)
@ -137,66 +119,17 @@ def run_board_example() -> None:
Include market research, technical planning, marketing strategy, and financial projections.
"""
# Execute the task
result = board_swarm.run(task=task)
print("Task completed successfully!")
print(f"Result: {result}")
def run_simple_example() -> None:
"""Run a simple Board of Directors example."""
# Create simple agents
analyst = Agent(
agent_name="Analyst",
agent_description="Data analyst",
model_name="gpt-4o-mini",
max_loops=1,
)
writer = Agent(
agent_name="Writer",
agent_description="Content writer",
model_name="gpt-4o-mini",
max_loops=1,
)
# Create swarm with default settings
board_swarm = BoardOfDirectorsSwarm(
name="Simple_Board",
agents=[analyst, writer],
verbose=True,
)
# Execute simple task
task = (
"Analyze current market trends and create a summary report."
)
result = board_swarm.run(task=task)
print("Simple example completed!")
print(f"Result: {result}")
# Execute the task and return result
return board_swarm.run(task=task)
def main() -> None:
"""Main function to run the examples."""
if not os.getenv("OPENAI_API_KEY"):
print(
"Warning: OPENAI_API_KEY not set. Example may not work."
)
return
try:
print("Running simple Board of Directors example...")
run_simple_example()
print("\nRunning comprehensive Board of Directors example...")
run_board_example()
except Exception as e:
print(f"Error: {e}")
result = run_board_example()
return result
except Exception:
pass
if __name__ == "__main__":

51
examples/multi_agent/board_of_directors/minimal_board_example.py
Unescape View File

@ -0,0 +1,51 @@
"""
Minimal Board of Directors Example
This example demonstrates the most basic Board of Directors swarm setup
with minimal configuration and agents.
To run this example:
1. Make sure you're in the root directory of the swarms project
2. Run: python examples/multi_agent/board_of_directors/minimal_board_example.py
"""
from swarms.structs.board_of_directors_swarm import (
BoardOfDirectorsSwarm,
)
from swarms.structs.agent import Agent
def run_minimal_example() -> str:
"""Run a minimal Board of Directors example."""
# Create a single agent
agent = Agent(
agent_name="General_Agent",
agent_description="General purpose agent",
model_name="gpt-4o-mini",
max_loops=1,
)
# Create minimal swarm
board_swarm = BoardOfDirectorsSwarm(
name="Minimal_Board",
agents=[agent],
verbose=False,
)
# Execute minimal task
task = "Provide a brief overview of artificial intelligence."
return board_swarm.run(task=task)
def main() -> None:
"""Main function to run the minimal example."""
try:
result = run_minimal_example()
return result
except Exception:
pass
if __name__ == "__main__":
main()

35
examples/multi_agent/board_of_directors/simple_board_example.py
Unescape View File

@ -0,0 +1,35 @@
from swarms.structs.board_of_directors_swarm import (
BoardOfDirectorsSwarm,
)
from swarms.structs.agent import Agent
# Create simple agents for basic tasks
analyst = Agent(
agent_name="Analyst",
agent_description="Data analyst",
model_name="gpt-4o-mini",
max_loops=1,
)
writer = Agent(
agent_name="Writer",
agent_description="Content writer",
model_name="gpt-4o-mini",
max_loops=1,
)
agents = [analyst, writer]
# Create swarm with default settings
board_swarm = BoardOfDirectorsSwarm(
name="Simple_Board",
agents=agents,
verbose=False,
)
# Execute simple task
task = "Analyze current market trends and create a summary report."
result = board_swarm.run(task=task)
print(result)

89
examples/multi_agent/hscf/single_file_hierarchical_framework_example.py
Unescape View File

@ -9,10 +9,11 @@ All components are now in one file: hierarchical_structured_communication_framew
import os
import sys
from typing import Dict, Any
# Add the project root to the Python path
project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
project_root = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "..")
)
sys.path.insert(0, project_root)
from dotenv import load_dotenv
@ -20,11 +21,6 @@ from dotenv import load_dotenv
# Import everything from the single file
from swarms.structs.hierarchical_structured_communication_framework import (
HierarchicalStructuredCommunicationFramework,
HierarchicalStructuredCommunicationGenerator,
HierarchicalStructuredCommunicationEvaluator,
HierarchicalStructuredCommunicationRefiner,
HierarchicalStructuredCommunicationSupervisor,
# Convenience aliases
TalkHierarchicalGenerator,
TalkHierarchicalEvaluator,
TalkHierarchicalRefiner,
@ -45,9 +41,7 @@ def example_basic_usage():
# Create framework with default configuration
framework = HierarchicalStructuredCommunicationFramework(
name="BasicFramework",
max_loops=2,
verbose=True
name="BasicFramework", max_loops=2, verbose=True
)
# Run a simple task
@ -64,7 +58,9 @@ def example_basic_usage():
print(result["final_result"])
print(f"\nTotal loops: {result['total_loops']}")
print(f"Conversation history entries: {len(result['conversation_history'])}")
print(
f"Conversation history entries: {len(result['conversation_history'])}"
)
print(f"Evaluation results: {len(result['evaluation_results'])}")
@ -80,33 +76,33 @@ def example_custom_agents():
generator = TalkHierarchicalGenerator(
agent_name="ContentCreator",
model_name="gpt-4o-mini",
verbose=True
verbose=True,
)
evaluator1 = TalkHierarchicalEvaluator(
agent_name="AccuracyChecker",
evaluation_criteria=["accuracy", "technical_correctness"],
model_name="gpt-4o-mini",
verbose=True
verbose=True,
)
evaluator2 = TalkHierarchicalEvaluator(
agent_name="ClarityChecker",
evaluation_criteria=["clarity", "readability", "coherence"],
model_name="gpt-4o-mini",
verbose=True
verbose=True,
)
refiner = TalkHierarchicalRefiner(
agent_name="ContentImprover",
model_name="gpt-4o-mini",
verbose=True
verbose=True,
)
supervisor = TalkHierarchicalSupervisor(
agent_name="WorkflowManager",
model_name="gpt-4o-mini",
verbose=True
verbose=True,
)
# Create framework with custom agents
@ -117,7 +113,7 @@ def example_custom_agents():
evaluators=[evaluator1, evaluator2],
refiners=[refiner],
max_loops=3,
verbose=True
verbose=True,
)
# Run a complex task
@ -134,7 +130,9 @@ def example_custom_agents():
print(result["final_result"])
print(f"\nTotal loops: {result['total_loops']}")
print(f"Conversation history entries: {len(result['conversation_history'])}")
print(
f"Conversation history entries: {len(result['conversation_history'])}"
)
print(f"Evaluation results: {len(result['evaluation_results'])}")
@ -154,7 +152,7 @@ def example_ollama_integration():
model_name="llama3:latest",
use_ollama=True,
ollama_base_url="http://localhost:11434/v1",
ollama_api_key="ollama"
ollama_api_key="ollama",
)
# Run a task with local model
@ -172,8 +170,12 @@ def example_ollama_integration():
print(result["final_result"])
print(f"\nTotal loops: {result['total_loops']}")
print(f"Conversation history entries: {len(result['conversation_history'])}")
print(f"Evaluation results: {len(result['evaluation_results'])}")
print(
f"Conversation history entries: {len(result['conversation_history'])}"
)
print(
f"Evaluation results: {len(result['evaluation_results'])}"
)
except Exception as e:
print(f"Error with Ollama: {e}")
@ -190,8 +192,7 @@ def example_structured_communication():
# Create framework
framework = HierarchicalStructuredCommunicationFramework(
name="CommunicationDemo",
verbose=True
name="CommunicationDemo", verbose=True
)
# Demonstrate structured message sending
@ -202,13 +203,17 @@ def example_structured_communication():
recipient="Generator",
message="Create a technical documentation outline",
background="For a Python library focused on data processing",
intermediate_output="Previous research on similar libraries"
intermediate_output="Previous research on similar libraries",
)
print(f"Message sent: {structured_msg.message}")
print(f"Background: {structured_msg.background}")
print(f"Intermediate output: {structured_msg.intermediate_output}")
print(f"From: {structured_msg.sender} -> To: {structured_msg.recipient}")
print(
f"Intermediate output: {structured_msg.intermediate_output}"
)
print(
f"From: {structured_msg.sender} -> To: {structured_msg.recipient}"
)
def example_agent_interaction():
@ -221,19 +226,17 @@ def example_agent_interaction():
# Create agents
generator = TalkHierarchicalGenerator(
agent_name="ContentGenerator",
verbose=True
agent_name="ContentGenerator", verbose=True
)
evaluator = TalkHierarchicalEvaluator(
agent_name="QualityEvaluator",
evaluation_criteria=["accuracy", "clarity"],
verbose=True
verbose=True,
)
refiner = TalkHierarchicalRefiner(
agent_name="ContentRefiner",
verbose=True
agent_name="ContentRefiner", verbose=True
)
# Generate content
@ -241,7 +244,7 @@ def example_agent_interaction():
gen_result = generator.generate_with_structure(
message="Create a brief explanation of machine learning",
background="For beginners with no technical background",
intermediate_output=""
intermediate_output="",
)
print(f"Generated content: {gen_result.content[:200]}...")
@ -249,8 +252,7 @@ def example_agent_interaction():
# Evaluate content
print("\n2. Evaluating content...")
eval_result = evaluator.evaluate_with_criterion(
content=gen_result.content,
criterion="clarity"
content=gen_result.content, criterion="clarity"
)
print(f"Evaluation score: {eval_result.score}/10")
@ -260,10 +262,12 @@ def example_agent_interaction():
print("\n3. Refining content...")
refine_result = refiner.refine_with_feedback(
original_content=gen_result.content,
evaluation_results=[eval_result]
evaluation_results=[eval_result],
)
print(f"Refined content: {refine_result.refined_content[:200]}...")
print(
f"Refined content: {refine_result.refined_content[:200]}..."
)
print(f"Changes made: {refine_result.changes_made}")
@ -271,9 +275,13 @@ def main():
"""
Main function to run all examples
"""
print("SINGLE-FILE HIERARCHICAL STRUCTURED COMMUNICATION FRAMEWORK")
print(
"SINGLE-FILE HIERARCHICAL STRUCTURED COMMUNICATION FRAMEWORK"
)
print("=" * 80)
print("This demonstrates the consolidated single-file implementation")
print(
"This demonstrates the consolidated single-file implementation"
)
print("based on the research paper: arXiv:2502.11098")
print("=" * 80)
@ -290,7 +298,9 @@ def main():
print("=" * 80)
print("Framework Features Demonstrated:")
print("✓ Single-file implementation")
print("✓ Structured Communication Protocol (M_ij, B_ij, I_ij)")
print(
"✓ Structured Communication Protocol (M_ij, B_ij, I_ij)"
)
print("✓ Hierarchical Evaluation System")
print("✓ Iterative Refinement Process")
print("✓ Flexible Model Configuration (OpenAI/Ollama)")
@ -303,6 +313,7 @@ def main():
except Exception as e:
print(f"Error during execution: {e}")
import traceback
traceback.print_exc()

9
examples/utils/agent_loader/claude_code_compatible.py
Unescape View File

@ -0,0 +1,9 @@
from swarms import load_agents_from_markdown
agents = load_agents_from_markdown(["finance_advisor.md"])
# Use the agent
response = agents[0].run(
"I have $100k to invest. I want to hedge my bets on the energy companies that will benefit from the AI revoltion"
"What are the top 4 stocks to invest in?"
)

2
examples/utils/agent_loader/finance_advisor.md
Unescape View File

@ -1,7 +1,7 @@
---
name: FinanceAdvisor
description: Expert financial advisor for investment and budgeting guidance
model_name: gpt-4o
model_name: claude-sonnet-4-20250514
temperature: 0.7
max_loops: 1
---

1
requirements.txt
Unescape View File

@ -28,4 +28,3 @@ mcp
numpy
openai
schedule
colorama

25
swarms/cli/main.py
Unescape View File

@ -22,6 +22,9 @@ from swarms.cli.onboarding_process import OnboardingProcess
from swarms.structs.agent import Agent
from swarms.utils.agent_loader import AgentLoader
from swarms.utils.formatter import formatter
from dotenv import load_dotenv
load_dotenv()
# Initialize console with custom styling
console = Console()
@ -397,7 +400,14 @@ def check_python_version() -> tuple[bool, str, str]:
def check_api_keys() -> tuple[bool, str, str]:
"""Check if common API keys are set."""
"""
Check if at least one common API key is set in the environment variables.
Returns:
tuple: (True, "✓", message) if at least one API key is set,
(False, "✗", message) otherwise.
"""
api_keys = {
"OPENAI_API_KEY": os.getenv("OPENAI_API_KEY"),
"ANTHROPIC_API_KEY": os.getenv("ANTHROPIC_API_KEY"),
@ -405,9 +415,16 @@ def check_api_keys() -> tuple[bool, str, str]:
"COHERE_API_KEY": os.getenv("COHERE_API_KEY"),
}
set_keys = [key for key, value in api_keys.items() if value]
if set_keys:
return True, "✓", f"API keys found: {', '.join(set_keys)}"
# At least one key must be present and non-empty
if any(value for value in api_keys.values()):
present_keys = [
key for key, value in api_keys.items() if value
]
return (
True,
"✓",
f"At least one API key found: {', '.join(present_keys)}",
)
else:
return (
False,

48
swarms/structs/__init__.py
Unescape View File

@ -1,12 +1,10 @@
from swarms.structs.agent import Agent
from swarms.structs.agent_builder import AgentsBuilder
from swarms.structs.agent_rearrange import AgentRearrange, rearrange
from swarms.structs.auto_swarm_builder import AutoSwarmBuilder
from swarms.structs.base_structure import BaseStructure
from swarms.structs.base_swarm import BaseSwarm
from swarms.structs.batch_agent_execution import batch_agent_execution
from swarms.structs.board_of_directors_swarm import (
BoardOfDirectorsSwarm,
)
from swarms.structs.concurrent_workflow import ConcurrentWorkflow
from swarms.structs.conversation import Conversation
from swarms.structs.council_as_judge import CouncilAsAJudge
@ -24,8 +22,8 @@ from swarms.structs.groupchat import (
expertise_based,
)
from swarms.structs.heavy_swarm import HeavySwarm
from swarms.structs.hierarchical_swarm import HierarchicalSwarm
from swarms.structs.hybrid_hierarchical_peer_swarm import (
from swarms.structs.hiearchical_swarm import HierarchicalSwarm
from swarms.structs.hybrid_hiearchical_peer_swarm import (
HybridHierarchicalClusterSwarm,
)
from swarms.structs.interactive_groupchat import (
@ -66,7 +64,6 @@ from swarms.structs.multi_agent_exec import (
run_single_agent,
)
from swarms.structs.multi_agent_router import MultiAgentRouter
from swarms.structs.rearrange import AgentRearrange, rearrange
from swarms.structs.round_robin import RoundRobinSwarm
from swarms.structs.sequential_workflow import SequentialWorkflow
from swarms.structs.spreadsheet_swarm import SpreadSheetSwarm
@ -82,7 +79,7 @@ from swarms.structs.stopping_conditions import (
check_stopped,
check_success,
)
from swarms.structs.swarm_arange import SwarmRearrange
from swarms.structs.swarm_rearrange import SwarmRearrange
from swarms.structs.swarm_router import (
SwarmRouter,
SwarmType,
@ -107,32 +104,11 @@ from swarms.structs.swarming_architectures import (
staircase_swarm,
star_swarm,
)
from swarms.structs.hierarchical_structured_communication_framework import (
HierarchicalStructuredCommunicationFramework,
HierarchicalStructuredCommunicationGenerator,
HierarchicalStructuredCommunicationEvaluator,
HierarchicalStructuredCommunicationRefiner,
HierarchicalStructuredCommunicationSupervisor,
StructuredMessage,
HierarchicalOrder,
EvaluationResult,
StructuredMessageSchema,
EvaluationResultSchema,
GeneratorResponseSchema,
EvaluatorResponseSchema,
RefinerResponseSchema,
CommunicationType,
AgentRole,
)
# Convenience alias(fixes old code if any was left out in the wild)
HierarchicalStructuredCommunicationSwarm = HierarchicalStructuredCommunicationFramework
__all__ = [
"Agent",
"BaseStructure",
"BaseSwarm",
"BoardOfDirectorsSwarm",
"ConcurrentWorkflow",
"Conversation",
"GroupChat",
@ -206,22 +182,6 @@ __all__ = [
"HierarchicalSwarm",
"HeavySwarm",
"CronJob",
"HierarchicalStructuredCommunicationSwarm",
"HierarchicalStructuredCommunicationGenerator",
"HierarchicalStructuredCommunicationEvaluator",
"HierarchicalStructuredCommunicationRefiner",
"HierarchicalStructuredCommunicationSupervisor",
"StructuredMessage",
"HierarchicalOrder",
"EvaluationResult",
"StructuredMessageSchema",
"EvaluationResultSchema",
"GeneratorResponseSchema",
"EvaluatorResponseSchema",
"RefinerResponseSchema",
"CommunicationType",
"AgentRole",
# Stopping conditions
"check_done",
"check_finished",
"check_complete",

198
swarms/structs/board_of_directors_swarm.py
Unescape View File

@ -19,7 +19,6 @@ Flow:
6. All context and conversation history is preserved throughout the process
"""
import asyncio
import json
import os
import re
@ -34,7 +33,6 @@ 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 (
@ -54,21 +52,6 @@ board_logger = initialize_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):
@ -91,12 +74,6 @@ class BoardConfigModel(BaseModel):
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,
@ -201,9 +178,6 @@ class BoardConfig:
):
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)
@ -236,62 +210,6 @@ class BoardConfig:
)
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.
@ -312,15 +230,6 @@ class BoardConfig:
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.
@ -562,63 +471,6 @@ def get_board_config(
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",
@ -953,7 +805,7 @@ class BoardSpec(BaseModel):
)
class BoardOfDirectorsSwarm(BaseSwarm):
class BoardOfDirectorsSwarm:
"""
A hierarchical swarm of agents with a Board of Directors that orchestrates tasks.
@ -1029,13 +881,8 @@ class BoardOfDirectorsSwarm(BaseSwarm):
Raises:
ValueError: If critical requirements are not met during initialization
"""
super().__init__(
name=name,
description=description,
agents=agents,
)
self.name = name
self.description = description
self.board_members = board_members or []
self.agents = agents or []
self.max_loops = max_loops
@ -1047,9 +894,8 @@ class BoardOfDirectorsSwarm(BaseSwarm):
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
)
self.max_workers = max_workers
self.max_workers = os.cpu_count()
# Initialize the swarm
self._init_board_swarm()
@ -1258,14 +1104,6 @@ You should be thorough, organized, and detail-oriented in your documentation."""
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."
@ -1687,34 +1525,6 @@ Please provide your response in the following format:
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.

750
swarms/structs/hierarchical_structured_communication_framework.py
View File

File diff suppressed because it is too large Load Diff
Write Preview
Loading…
Cancel
Save