From 2d558061ba49bd6ae8187170c21b4d1fa95adcaf Mon Sep 17 00:00:00 2001 From: Kye Gomez Date: Sat, 12 Jul 2025 14:35:22 -0700 Subject: [PATCH] docs fix --- docs/examples/paper_implementations.md | 7 +- scripts/examples.py | 6 +- swarms/structs/hiearchical_swarm.py | 21 +- tests/comprehensive_test.py | 726 ++++++++++++++++++------- 4 files changed, 536 insertions(+), 224 deletions(-) diff --git a/docs/examples/paper_implementations.md b/docs/examples/paper_implementations.md index 63e9b691..e9211a7d 100644 --- a/docs/examples/paper_implementations.md +++ b/docs/examples/paper_implementations.md @@ -47,10 +47,9 @@ We invite you to explore these implementations, contribute to our research effor | Paper Name | Description | Original Paper | Implementation | Status | Key Features | |------------|-------------|----------------|----------------|--------|--------------| | **MALT (Multi-Agent Learning Task)** | A sophisticated orchestration framework that coordinates multiple specialized AI agents to tackle complex tasks through structured conversations. | [arXiv:2412.01928](https://arxiv.org/pdf/2412.01928) | [`swarms.structs.malt`](https://docs.swarms.world/en/latest/swarms/structs/malt/) | ✅ Complete | Creator-Verifier-Refiner architecture, structured conversations, reliability guarantees | -| **MAI-DxO (MAI Diagnostic Orchestrator)** | An open-source implementation of Microsoft Research's "Sequential Diagnosis with Language Models" paper, simulating a virtual panel of physician-agents for iterative medical diagnosis. | Microsoft Research Paper | [GitHub Repository](https://github.com/The-Swarm-Corporation/Open-MAI-Dx-Orchestrator) | ✅ Complete | Cost-effective medical diagnosis, physician-agent panel, iterative refinement | -| **AI-CoScientist** | A multi-agent AI framework for collaborative scientific research, implementing the "Towards an AI Co-Scientist" methodology with tournament-based hypothesis evolution. | "Towards an AI Co-Scientist" Paper | [GitHub Repository](https://github.com/The-Swarm-Corporation/AI-CoScientist) | ✅ Complete | Tournament-based selection, peer review systems, hypothesis evolution, Elo rating system | -| **Mixture of Agents (MoA)** | A sophisticated multi-agent architecture that implements parallel processing with iterative refinement, combining diverse expert agents for comprehensive analysis. | Multi-agent collaboration concepts | [`swarms.structs.moa`](https://docs.swarms.world/en/latest/swarms/structs/moa/) | ✅ Complete | Parallel processing, expert agent combination, iterative refinement, state-of-the-art performance | -| **Open Scientist** | A multi-agent system for scientific research exploration using specialized agents for hypothesis generation, peer review, ranking, evolution, and meta-analysis. | Scientific research methodology | [`examples/demos/open_scientist.py`](https://github.com/kyegomez/swarms/blob/main/examples/demos/open_scientist.py) | ✅ Complete | Hypothesis generation, peer review, ranking, evolution, meta-analysis, proximity control | +| **[MAI-DxO (MAI Diagnostic Orchestrator)](https://arxiv.org/abs/2506.22405)** | An open-source implementation of Microsoft Research's "[Sequential Diagnosis with Language Models](https://arxiv.org/abs/2506.22405)" paper, simulating a virtual panel of physician-agents for iterative medical diagnosis. | Microsoft Research Paper | [GitHub Repository](https://github.com/The-Swarm-Corporation/Open-MAI-Dx-Orchestrator) | ✅ Complete | Cost-effective medical diagnosis, physician-agent panel, iterative refinement | +| **[AI-CoScientist](https://storage.googleapis.com/coscientist_paper/ai_coscientist.pdf)** | A multi-agent AI framework for collaborative scientific research, implementing the "Towards an AI Co-Scientist" methodology with tournament-based hypothesis evolution. | "Towards an AI Co-Scientist" Paper | [GitHub Repository](https://github.com/The-Swarm-Corporation/AI-CoScientist) | ✅ Complete | Tournament-based selection, peer review systems, hypothesis evolution, Elo rating system | +| **[Mixture of Agents (MoA)](https://arxiv.org/abs/2406.04692)** | A sophisticated multi-agent architecture that implements parallel processing with iterative refinement, combining diverse expert agents for comprehensive analysis. | Multi-agent collaboration concepts | [`swarms.structs.moa`](https://docs.swarms.world/en/latest/swarms/structs/moa/) | ✅ Complete | Parallel processing, expert agent combination, iterative refinement, state-of-the-art performance | | **Deep Research Swarm** | A production-grade research system that conducts comprehensive analysis across multiple domains using parallel processing and advanced AI agents. | Research methodology | [`swarms.structs.deep_research_swarm`](https://docs.swarms.world/en/latest/swarms/structs/deep_research_swarm/) | ✅ Complete | Parallel search processing, multi-agent coordination, information synthesis, concurrent execution | | **Agent-as-a-Judge** | An evaluation framework that uses agents to evaluate other agents, implementing the "Agent-as-a-Judge: Evaluate Agents with Agents" methodology. | [arXiv:2410.10934](https://arxiv.org/abs/2410.10934) | [`swarms.agents.agent_judge`](https://docs.swarms.world/en/latest/swarms/agents/agent_judge/) | ✅ Complete | Agent evaluation, quality assessment, automated judging, performance metrics | diff --git a/scripts/examples.py b/scripts/examples.py index 698bc296..b355bf71 100644 --- a/scripts/examples.py +++ b/scripts/examples.py @@ -1,5 +1,6 @@ import requests + def get_example_py_urls(): owner = "kyegomez" repo = "swarms" @@ -7,7 +8,9 @@ def get_example_py_urls(): examples_path = "examples" api_url = f"https://api.github.com/repos/{owner}/{repo}/git/trees/{branch}?recursive=1" - raw_base = f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/" + raw_base = ( + f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/" + ) response = requests.get(api_url) response.raise_for_status() @@ -25,6 +28,7 @@ def get_example_py_urls(): return example_files + if __name__ == "__main__": urls = get_example_py_urls() for url in urls: diff --git a/swarms/structs/hiearchical_swarm.py b/swarms/structs/hiearchical_swarm.py index 84662839..2ff30a06 100644 --- a/swarms/structs/hiearchical_swarm.py +++ b/swarms/structs/hiearchical_swarm.py @@ -11,7 +11,7 @@ Flow: """ import traceback -from typing import Any, List, Literal, Optional, Union, Callable +from typing import Any, Callable, List, Literal, Optional, Union from pydantic import BaseModel, Field @@ -21,15 +21,13 @@ from swarms.prompts.hiearchical_system_prompt import ( 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.tools.base_tool import BaseTool from swarms.utils.history_output_formatter import ( history_output_formatter, ) -from swarms.utils.output_types import OutputType - from swarms.utils.loguru_logger import initialize_logger -from swarms.tools.base_tool import BaseTool -from swarms.structs.ma_utils import list_all_agents - +from swarms.utils.output_types import OutputType logger = initialize_logger(log_folder="hierarchical_swarm") @@ -118,6 +116,7 @@ class HierarchicalSwarm(BaseSwarm): planning_director_agent: Optional[ Union[Agent, Callable, Any] ] = None, + director_feedback_on: bool = True, *args, **kwargs, ): @@ -150,6 +149,7 @@ class HierarchicalSwarm(BaseSwarm): self.director_model_name = director_model_name self.add_collaboration_prompt = add_collaboration_prompt self.planning_director_agent = planning_director_agent + self.director_feedback_on = director_feedback_on self.init_swarm() @@ -354,7 +354,10 @@ class HierarchicalSwarm(BaseSwarm): if self.verbose: logger.info(f"⚡ Executed {len(outputs)} orders") - feedback = self.feedback_director(outputs) + if self.director_feedback_on is True: + feedback = self.feedback_director(outputs) + else: + feedback = outputs if self.verbose: logger.success("✅ Step completed successfully") @@ -469,7 +472,9 @@ class HierarchicalSwarm(BaseSwarm): f"Worker Agent Responses: {task}" ) ) - self.conversation.add(role="Director", content=output) + self.conversation.add( + role=self.director.agent_name, content=output + ) if self.verbose: logger.success( diff --git a/tests/comprehensive_test.py b/tests/comprehensive_test.py index d4925236..01a74954 100644 --- a/tests/comprehensive_test.py +++ b/tests/comprehensive_test.py @@ -1,16 +1,14 @@ import os import json -import time from datetime import datetime from typing import List, Dict, Any, Callable -import requests from dotenv import load_dotenv # Basic Imports for Swarms from swarms.structs import ( - Agent, - SequentialWorkflow, + Agent, + SequentialWorkflow, ConcurrentWorkflow, AgentRearrange, MixtureOfAgents, @@ -20,16 +18,19 @@ from swarms.structs import ( MajorityVoting, SwarmRouter, RoundRobinSwarm, - InteractiveGroupChat + InteractiveGroupChat, ) + # Import swarms not in __init__.py directly from swarms.structs.hiearchical_swarm import HierarchicalSwarm from swarms.structs.tree_swarm import ForestSwarm, Tree, TreeAgent -from swarms.tools.base_tool import BaseTool # Setup Logging from loguru import logger -logger.add("test_runs/test_failures.log", rotation="10 MB", level="ERROR") + +logger.add( + "test_runs/test_failures.log", rotation="10 MB", level="ERROR" +) # Load environment variables load_dotenv() @@ -49,18 +50,24 @@ API_KEY = os.getenv("OPENAI_API_KEY") # --- Helper Functions --- + def generate_timestamp() -> str: """Generate a timestamp string for filenames""" return datetime.now().strftime("%Y%m%d_%H%M%S") -def write_markdown_report(results: List[Dict[str, Any]], filename: str): + +def write_markdown_report( + results: List[Dict[str, Any]], filename: str +): """Write test results to a markdown file""" if not os.path.exists("test_runs"): os.makedirs("test_runs") with open(f"test_runs/{filename}.md", "w") as f: f.write("# Swarms Comprehensive Test Report\n\n") - f.write(f"Test Run: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n") + f.write( + f"Test Run: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n" + ) total = len(results) passed = sum(1 for r in results if r["status"] == "passed") @@ -80,16 +87,23 @@ def write_markdown_report(results: List[Dict[str, Any]], filename: str): f.write("Response:\n```json\n") response_str = result["response"] try: - response_json = json.loads(response_str) if isinstance(response_str, str) else response_str + response_json = ( + json.loads(response_str) + if isinstance(response_str, str) + else response_str + ) f.write(json.dumps(response_json, indent=2)) except (json.JSONDecodeError, TypeError): f.write(str(response_str)) f.write("\n```\n\n") if result.get("error"): - f.write(f"**Error:**\n```\n{result['error']}\n```\n\n") + f.write( + f"**Error:**\n```\n{result['error']}\n```\n\n" + ) f.write("---\n\n") + # def create_github_issue(test_result: Dict[str, Any]) -> Dict[str, Any]: # """Create a GitHub issue for a failed test""" # if not all([GITHUB_TOKEN, GITHUB_REPO_OWNER, GITHUB_REPO_NAME]): @@ -100,7 +114,7 @@ def write_markdown_report(results: List[Dict[str, Any]], filename: str): # return None # issue_title = f"Automated Test Failure: {test_result['test_name']}" - + # issue_body = f""" # ## Test Failure Report @@ -141,43 +155,66 @@ def write_markdown_report(results: List[Dict[str, Any]], filename: str): # logger.error(f"Failed to create GitHub issue: {e.response.text if e.response else str(e)}") # return None -def create_test_agent(name: str, system_prompt: str = None, model_name: str = "gpt-4o-mini", tools: List[Callable] = None, **kwargs) -> Agent: + +def create_test_agent( + name: str, + system_prompt: str = None, + model_name: str = "gpt-4o-mini", + tools: List[Callable] = None, + **kwargs, +) -> Agent: """Create a properly configured test agent with error handling""" try: return Agent( agent_name=name, - system_prompt=system_prompt or f"You are {name}, a helpful AI assistant.", + system_prompt=system_prompt + or f"You are {name}, a helpful AI assistant.", model_name=model_name, # Use mini model for faster/cheaper testing max_loops=1, max_tokens=200, tools=tools, - **kwargs + **kwargs, ) except Exception as e: logger.error(f"Failed to create agent {name}: {e}") raise + # --- Basic Agent Tests --- + def test_basic_agent_functionality(): """Test basic agent creation and execution""" agent = create_test_agent("BasicAgent") response = agent.run("Say hello and explain what you are.") - + assert isinstance(response, str) and len(response) > 0 - return {"test_name": "test_basic_agent_functionality", "status": "passed", "response": "Agent created and responded successfully"} + return { + "test_name": "test_basic_agent_functionality", + "status": "passed", + "response": "Agent created and responded successfully", + } + def test_agent_with_custom_prompt(): """Test agent with custom system prompt""" custom_prompt = "You are a mathematician who only responds with numbers and mathematical expressions." - agent = create_test_agent("MathAgent", system_prompt=custom_prompt) + agent = create_test_agent( + "MathAgent", system_prompt=custom_prompt + ) response = agent.run("What is 2+2?") - + assert isinstance(response, str) and len(response) > 0 - return {"test_name": "test_agent_with_custom_prompt", "status": "passed", "response": response[:100]} + return { + "test_name": "test_agent_with_custom_prompt", + "status": "passed", + "response": response[:100], + } + def test_tool_execution_with_agent(): """Test agent's ability to use tools""" + def simple_calculator(a: int, b: int) -> int: """Add two numbers together""" return a + b @@ -187,147 +224,262 @@ def test_tool_execution_with_agent(): return f"The weather in {location} is sunny and 75°F" agent = create_test_agent( - "ToolAgent", + "ToolAgent", system_prompt="You are a helpful assistant that can use tools to help users.", - tools=[simple_calculator, get_weather] + tools=[simple_calculator, get_weather], + ) + response = agent.run( + "What's 5 + 7 and what's the weather like in New York?" ) - response = agent.run("What's 5 + 7 and what's the weather like in New York?") - + assert isinstance(response, str) and len(response) > 0 - return {"test_name": "test_tool_execution_with_agent", "status": "passed", "response": "Tool execution completed"} + return { + "test_name": "test_tool_execution_with_agent", + "status": "passed", + "response": "Tool execution completed", + } + # --- Multi-Modal Tests --- + def test_multimodal_execution(): """Test agent's ability to process images""" - agent = create_test_agent("VisionAgent", model_name="gpt-4o", multi_modal=True) - + agent = create_test_agent( + "VisionAgent", model_name="gpt-4o", multi_modal=True + ) + try: # Check if test images exist, if not skip the test if os.path.exists("tests/test_data/image1.jpg"): - response = agent.run("Describe this image.", img="tests/test_data/image1.jpg") + response = agent.run( + "Describe this image.", + img="tests/test_data/image1.jpg", + ) assert isinstance(response, str) and len(response) > 0 else: - logger.warning("Test image not found, skipping multimodal test") + logger.warning( + "Test image not found, skipping multimodal test" + ) response = "Test skipped - no test image available" - - return {"test_name": "test_multimodal_execution", "status": "passed", "response": "Multimodal response received"} + + return { + "test_name": "test_multimodal_execution", + "status": "passed", + "response": "Multimodal response received", + } except Exception as e: logger.warning(f"Multimodal test failed: {e}") - return {"test_name": "test_multimodal_execution", "status": "passed", "response": "Multimodal test skipped due to missing dependencies"} + return { + "test_name": "test_multimodal_execution", + "status": "passed", + "response": "Multimodal test skipped due to missing dependencies", + } + # --- Workflow Tests --- + def test_sequential_workflow(): """Test SequentialWorkflow with multiple agents""" agents = [ - create_test_agent("ResearchAgent", "You are a research specialist who gathers information."), - create_test_agent("AnalysisAgent", "You are an analyst who analyzes information and provides insights."), - create_test_agent("WriterAgent", "You are a writer who creates clear, concise summaries.") + create_test_agent( + "ResearchAgent", + "You are a research specialist who gathers information.", + ), + create_test_agent( + "AnalysisAgent", + "You are an analyst who analyzes information and provides insights.", + ), + create_test_agent( + "WriterAgent", + "You are a writer who creates clear, concise summaries.", + ), ] workflow = SequentialWorkflow( - name="research-analysis-workflow", - agents=agents, - max_loops=1 + name="research-analysis-workflow", agents=agents, max_loops=1 ) - + try: - response = workflow.run("Research and analyze the benefits of renewable energy, then write a brief summary.") - logger.info(f"SequentialWorkflow response type: {type(response)}") - + response = workflow.run( + "Research and analyze the benefits of renewable energy, then write a brief summary." + ) + logger.info( + f"SequentialWorkflow response type: {type(response)}" + ) + # SequentialWorkflow returns conversation history assert response is not None - return {"test_name": "test_sequential_workflow", "status": "passed", "response": "Sequential workflow completed"} + return { + "test_name": "test_sequential_workflow", + "status": "passed", + "response": "Sequential workflow completed", + } except Exception as e: - logger.error(f"SequentialWorkflow test failed with exception: {e}") - return {"test_name": "test_sequential_workflow", "status": "failed", "error": str(e)} + logger.error( + f"SequentialWorkflow test failed with exception: {e}" + ) + return { + "test_name": "test_sequential_workflow", + "status": "failed", + "error": str(e), + } + def test_concurrent_workflow(): """Test ConcurrentWorkflow with multiple agents""" agents = [ - create_test_agent("TechAnalyst", "You are a technology analyst who focuses on tech trends."), - create_test_agent("MarketAnalyst", "You are a market analyst who focuses on market conditions.") + create_test_agent( + "TechAnalyst", + "You are a technology analyst who focuses on tech trends.", + ), + create_test_agent( + "MarketAnalyst", + "You are a market analyst who focuses on market conditions.", + ), ] workflow = ConcurrentWorkflow( - name="concurrent-analysis", - agents=agents, - max_loops=1 + name="concurrent-analysis", agents=agents, max_loops=1 ) - + try: - response = workflow.run("Analyze the current state of AI technology and its market impact.") - logger.info(f"ConcurrentWorkflow response type: {type(response)}") - + response = workflow.run( + "Analyze the current state of AI technology and its market impact." + ) + logger.info( + f"ConcurrentWorkflow response type: {type(response)}" + ) + assert response is not None - return {"test_name": "test_concurrent_workflow", "status": "passed", "response": "Concurrent workflow completed"} + return { + "test_name": "test_concurrent_workflow", + "status": "passed", + "response": "Concurrent workflow completed", + } except Exception as e: - logger.error(f"ConcurrentWorkflow test failed with exception: {e}") - return {"test_name": "test_concurrent_workflow", "status": "failed", "error": str(e)} + logger.error( + f"ConcurrentWorkflow test failed with exception: {e}" + ) + return { + "test_name": "test_concurrent_workflow", + "status": "failed", + "error": str(e), + } + # --- Advanced Swarm Tests --- + def test_agent_rearrange(): """Test AgentRearrange dynamic workflow""" agents = [ - create_test_agent("Researcher", "You are a researcher who gathers information."), - create_test_agent("Analyst", "You are an analyst who analyzes information."), - create_test_agent("Writer", "You are a writer who creates final reports.") + create_test_agent( + "Researcher", + "You are a researcher who gathers information.", + ), + create_test_agent( + "Analyst", "You are an analyst who analyzes information." + ), + create_test_agent( + "Writer", "You are a writer who creates final reports." + ), ] - + flow = "Researcher -> Analyst -> Writer" swarm = AgentRearrange(agents=agents, flow=flow, max_loops=1) - - response = swarm.run("Research renewable energy, analyze the benefits, and write a summary.") - + + response = swarm.run( + "Research renewable energy, analyze the benefits, and write a summary." + ) + assert response is not None - return {"test_name": "test_agent_rearrange", "status": "passed", "response": "AgentRearrange completed"} + return { + "test_name": "test_agent_rearrange", + "status": "passed", + "response": "AgentRearrange completed", + } + def test_mixture_of_agents(): """Test MixtureOfAgents collaboration""" agents = [ - create_test_agent("TechExpert", "You are a technology expert."), - create_test_agent("BusinessAnalyst", "You are a business analyst."), - create_test_agent("Strategist", "You are a strategic planner.") + create_test_agent( + "TechExpert", "You are a technology expert." + ), + create_test_agent( + "BusinessAnalyst", "You are a business analyst." + ), + create_test_agent( + "Strategist", "You are a strategic planner." + ), ] - + swarm = MixtureOfAgents(agents=agents, max_loops=1) - response = swarm.run("Analyze the impact of AI on modern businesses.") - + response = swarm.run( + "Analyze the impact of AI on modern businesses." + ) + assert response is not None - return {"test_name": "test_mixture_of_agents", "status": "passed", "response": "MixtureOfAgents completed"} + return { + "test_name": "test_mixture_of_agents", + "status": "passed", + "response": "MixtureOfAgents completed", + } + def test_spreadsheet_swarm(): """Test SpreadSheetSwarm for data processing""" agents = [ - create_test_agent("DataProcessor1", "You process and analyze numerical data."), - create_test_agent("DataProcessor2", "You perform calculations and provide insights.") + create_test_agent( + "DataProcessor1", + "You process and analyze numerical data.", + ), + create_test_agent( + "DataProcessor2", + "You perform calculations and provide insights.", + ), ] - + swarm = SpreadSheetSwarm( name="data-processing-swarm", description="A swarm for processing data", - agents=agents, - max_loops=1, - autosave_on=False + agents=agents, + max_loops=1, + autosave_on=False, + ) + + response = swarm.run( + "Calculate the sum of 25 + 75 and provide analysis." ) - - response = swarm.run("Calculate the sum of 25 + 75 and provide analysis.") - + assert response is not None - return {"test_name": "test_spreadsheet_swarm", "status": "passed", "response": "SpreadSheetSwarm completed"} + return { + "test_name": "test_spreadsheet_swarm", + "status": "passed", + "response": "SpreadSheetSwarm completed", + } + def test_hierarchical_swarm(): """Test HierarchicalSwarm structure""" try: - from swarms.utils.function_caller_model import OpenAIFunctionCaller + from swarms.utils.function_caller_model import ( + OpenAIFunctionCaller, + ) from swarms.structs.hiearchical_swarm import SwarmSpec - + # Create worker agents workers = [ - create_test_agent("Worker1", "You are Worker1 who handles research tasks and data gathering."), - create_test_agent("Worker2", "You are Worker2 who handles analysis tasks and reporting.") + create_test_agent( + "Worker1", + "You are Worker1 who handles research tasks and data gathering.", + ), + create_test_agent( + "Worker2", + "You are Worker2 who handles analysis tasks and reporting.", + ), ] - + # Create director agent with explicit knowledge of available agents director = OpenAIFunctionCaller( base_model=SwarmSpec, @@ -344,129 +496,219 @@ def test_hierarchical_swarm(): "4. Provide actionable task descriptions" ), temperature=0.1, - max_tokens=1000 + max_tokens=1000, ) - + swarm = HierarchicalSwarm( description="A test hierarchical swarm for task delegation", director=director, agents=workers, - max_loops=1 + max_loops=1, ) - - response = swarm.run("Research current team meeting best practices and analyze them to create recommendations.") - + + response = swarm.run( + "Research current team meeting best practices and analyze them to create recommendations." + ) + assert response is not None - return {"test_name": "test_hierarchical_swarm", "status": "passed", "response": "HierarchicalSwarm completed"} + return { + "test_name": "test_hierarchical_swarm", + "status": "passed", + "response": "HierarchicalSwarm completed", + } except ImportError as e: - logger.warning(f"HierarchicalSwarm test skipped due to missing dependencies: {e}") - return {"test_name": "test_hierarchical_swarm", "status": "passed", "response": "Test skipped due to missing dependencies"} + logger.warning( + f"HierarchicalSwarm test skipped due to missing dependencies: {e}" + ) + return { + "test_name": "test_hierarchical_swarm", + "status": "passed", + "response": "Test skipped due to missing dependencies", + } + def test_majority_voting(): """Test MajorityVoting consensus mechanism""" agents = [ - create_test_agent("Judge1", "You are a judge who evaluates options carefully."), - create_test_agent("Judge2", "You are a judge who provides thorough analysis."), - create_test_agent("Judge3", "You are a judge who considers all perspectives.") + create_test_agent( + "Judge1", + "You are a judge who evaluates options carefully.", + ), + create_test_agent( + "Judge2", + "You are a judge who provides thorough analysis.", + ), + create_test_agent( + "Judge3", + "You are a judge who considers all perspectives.", + ), ] - + swarm = MajorityVoting(agents=agents) - response = swarm.run("Should companies invest more in renewable energy? Provide YES or NO with reasoning.") - + response = swarm.run( + "Should companies invest more in renewable energy? Provide YES or NO with reasoning." + ) + assert response is not None - return {"test_name": "test_majority_voting", "status": "passed", "response": "MajorityVoting completed"} + return { + "test_name": "test_majority_voting", + "status": "passed", + "response": "MajorityVoting completed", + } + def test_round_robin_swarm(): """Test RoundRobinSwarm task distribution""" agents = [ create_test_agent("Agent1", "You handle counting tasks."), - create_test_agent("Agent2", "You handle color-related tasks."), - create_test_agent("Agent3", "You handle animal-related tasks.") + create_test_agent( + "Agent2", "You handle color-related tasks." + ), + create_test_agent( + "Agent3", "You handle animal-related tasks." + ), ] - + swarm = RoundRobinSwarm(agents=agents) tasks = [ - "Count from 1 to 5", - "Name 3 primary colors", - "List 3 common pets" + "Count from 1 to 5", + "Name 3 primary colors", + "List 3 common pets", ] - + response = swarm.run(tasks) - + assert response is not None - return {"test_name": "test_round_robin_swarm", "status": "passed", "response": "RoundRobinSwarm completed"} + return { + "test_name": "test_round_robin_swarm", + "status": "passed", + "response": "RoundRobinSwarm completed", + } + def test_swarm_router(): """Test SwarmRouter dynamic routing""" agents = [ - create_test_agent("DataAnalyst", "You specialize in data analysis and statistics."), - create_test_agent("ReportWriter", "You specialize in writing clear, professional reports.") + create_test_agent( + "DataAnalyst", + "You specialize in data analysis and statistics.", + ), + create_test_agent( + "ReportWriter", + "You specialize in writing clear, professional reports.", + ), ] - + router = SwarmRouter( name="analysis-router", description="Routes analysis and reporting tasks to appropriate agents", agents=agents, swarm_type="SequentialWorkflow", - max_loops=1 + max_loops=1, ) - - response = router.run("Analyze customer satisfaction data and write a summary report.") - + + response = router.run( + "Analyze customer satisfaction data and write a summary report." + ) + assert response is not None - return {"test_name": "test_swarm_router", "status": "passed", "response": "SwarmRouter completed"} + return { + "test_name": "test_swarm_router", + "status": "passed", + "response": "SwarmRouter completed", + } + def test_groupchat(): """Test GroupChat functionality""" agents = [ - create_test_agent("Moderator", "You are a discussion moderator who guides conversations."), - create_test_agent("Expert1", "You are a subject matter expert who provides insights."), - create_test_agent("Expert2", "You are another expert who offers different perspectives.") + create_test_agent( + "Moderator", + "You are a discussion moderator who guides conversations.", + ), + create_test_agent( + "Expert1", + "You are a subject matter expert who provides insights.", + ), + create_test_agent( + "Expert2", + "You are another expert who offers different perspectives.", + ), ] - - groupchat = GroupChat( - agents=agents, - messages=[], - max_round=2 - ) - + + groupchat = GroupChat(agents=agents, messages=[], max_round=2) + # GroupChat requires a different interface than other swarms - response = groupchat.run("Discuss the benefits and challenges of remote work.") - + response = groupchat.run( + "Discuss the benefits and challenges of remote work." + ) + assert response is not None - return {"test_name": "test_groupchat", "status": "passed", "response": "GroupChat completed"} + return { + "test_name": "test_groupchat", + "status": "passed", + "response": "GroupChat completed", + } + def test_multi_agent_router(): """Test MultiAgentRouter functionality""" agents = [ - create_test_agent("TechAgent", "You handle technology-related queries."), - create_test_agent("BusinessAgent", "You handle business-related queries."), - create_test_agent("GeneralAgent", "You handle general queries.") + create_test_agent( + "TechAgent", "You handle technology-related queries." + ), + create_test_agent( + "BusinessAgent", "You handle business-related queries." + ), + create_test_agent( + "GeneralAgent", "You handle general queries." + ), ] - + router = MultiAgentRouter(agents=agents) - response = router.run("What are the latest trends in business technology?") - + response = router.run( + "What are the latest trends in business technology?" + ) + assert response is not None - return {"test_name": "test_multi_agent_router", "status": "passed", "response": "MultiAgentRouter completed"} + return { + "test_name": "test_multi_agent_router", + "status": "passed", + "response": "MultiAgentRouter completed", + } + def test_interactive_groupchat(): """Test InteractiveGroupChat functionality""" agents = [ - create_test_agent("Facilitator", "You facilitate group discussions."), - create_test_agent("Participant1", "You are an active discussion participant."), - create_test_agent("Participant2", "You provide thoughtful contributions to discussions.") + create_test_agent( + "Facilitator", "You facilitate group discussions." + ), + create_test_agent( + "Participant1", + "You are an active discussion participant.", + ), + create_test_agent( + "Participant2", + "You provide thoughtful contributions to discussions.", + ), ] - + interactive_chat = InteractiveGroupChat( - agents=agents, - max_loops=2 + agents=agents, max_loops=2 + ) + + response = interactive_chat.run( + "Let's discuss the future of artificial intelligence." ) - - response = interactive_chat.run("Let's discuss the future of artificial intelligence.") - + assert response is not None - return {"test_name": "test_interactive_groupchat", "status": "passed", "response": "InteractiveGroupChat completed"} + return { + "test_name": "test_interactive_groupchat", + "status": "passed", + "response": "InteractiveGroupChat completed", + } + def test_forest_swarm(): """Test ForestSwarm tree-based structure""" @@ -475,117 +717,175 @@ def test_forest_swarm(): tree1_agents = [ TreeAgent( system_prompt="You analyze market trends", - agent_name="Market-Analyst" + agent_name="Market-Analyst", ), TreeAgent( - system_prompt="You provide financial insights", - agent_name="Financial-Advisor" - ) + system_prompt="You provide financial insights", + agent_name="Financial-Advisor", + ), ] - + tree2_agents = [ TreeAgent( system_prompt="You assess investment risks", - agent_name="Risk-Assessor" + agent_name="Risk-Assessor", ), TreeAgent( system_prompt="You create investment strategies", - agent_name="Strategy-Planner" - ) + agent_name="Strategy-Planner", + ), ] - + # Create trees tree1 = Tree(tree_name="Analysis-Tree", agents=tree1_agents) tree2 = Tree(tree_name="Strategy-Tree", agents=tree2_agents) - + # Create ForestSwarm forest = ForestSwarm(trees=[tree1, tree2]) - - response = forest.run("Analyze the current market and develop an investment strategy.") - + + response = forest.run( + "Analyze the current market and develop an investment strategy." + ) + assert response is not None - return {"test_name": "test_forest_swarm", "status": "passed", "response": "ForestSwarm completed"} + return { + "test_name": "test_forest_swarm", + "status": "passed", + "response": "ForestSwarm completed", + } except Exception as e: logger.error(f"ForestSwarm test failed: {e}") - return {"test_name": "test_forest_swarm", "status": "failed", "error": str(e)} + return { + "test_name": "test_forest_swarm", + "status": "failed", + "error": str(e), + } + # --- Performance & Features Tests --- + def test_streaming_mode(): """Test streaming response generation""" agent = create_test_agent("StreamingAgent", streaming_on=True) - response = agent.run("Tell me a very short story about technology.") - + response = agent.run( + "Tell me a very short story about technology." + ) + assert response is not None - return {"test_name": "test_streaming_mode", "status": "passed", "response": "Streaming mode tested"} + return { + "test_name": "test_streaming_mode", + "status": "passed", + "response": "Streaming mode tested", + } + def test_agent_memory_persistence(): """Test agent memory functionality""" - agent = create_test_agent("MemoryAgent", - system_prompt="You remember information from previous conversations.", - return_history=True) - + agent = create_test_agent( + "MemoryAgent", + system_prompt="You remember information from previous conversations.", + return_history=True, + ) + # First interaction response1 = agent.run("My name is Alice. Please remember this.") - # Second interaction + # Second interaction response2 = agent.run("What is my name?") - + assert response1 is not None and response2 is not None - return {"test_name": "test_agent_memory_persistence", "status": "passed", "response": "Memory persistence tested"} + return { + "test_name": "test_agent_memory_persistence", + "status": "passed", + "response": "Memory persistence tested", + } + def test_error_handling(): """Test agent error handling with various inputs""" agent = create_test_agent("ErrorTestAgent") - + try: # Test with empty task response = agent.run("") assert response is not None or response == "" - + # Test with very simple task response = agent.run("Hi") assert response is not None - - return {"test_name": "test_error_handling", "status": "passed", "response": "Error handling tests passed"} + + return { + "test_name": "test_error_handling", + "status": "passed", + "response": "Error handling tests passed", + } except Exception as e: - return {"test_name": "test_error_handling", "status": "failed", "error": str(e)} + return { + "test_name": "test_error_handling", + "status": "failed", + "error": str(e), + } + # --- Integration Tests --- + def test_complex_workflow_integration(): """Test complex multi-agent workflow integration""" try: # Create specialized agents - researcher = create_test_agent("Researcher", "You research topics thoroughly and gather information.") - analyst = create_test_agent("Analyst", "You analyze research data and provide insights.") - writer = create_test_agent("Writer", "You write clear, comprehensive summaries.") - + researcher = create_test_agent( + "Researcher", + "You research topics thoroughly and gather information.", + ) + analyst = create_test_agent( + "Analyst", + "You analyze research data and provide insights.", + ) + writer = create_test_agent( + "Writer", "You write clear, comprehensive summaries." + ) + # Test SequentialWorkflow sequential = SequentialWorkflow( name="research-workflow", - agents=[researcher, analyst, writer], - max_loops=1 + agents=[researcher, analyst, writer], + max_loops=1, + ) + + seq_response = sequential.run( + "Research AI trends, analyze them, and write a summary." ) - - seq_response = sequential.run("Research AI trends, analyze them, and write a summary.") - - # Test ConcurrentWorkflow + + # Test ConcurrentWorkflow concurrent = ConcurrentWorkflow( name="parallel-analysis", - agents=[researcher, analyst], - max_loops=1 + agents=[researcher, analyst], + max_loops=1, ) - - conc_response = concurrent.run("What are the benefits and challenges of AI?") - + + conc_response = concurrent.run( + "What are the benefits and challenges of AI?" + ) + assert seq_response is not None and conc_response is not None - return {"test_name": "test_complex_workflow_integration", "status": "passed", "response": "Complex workflow integration completed"} + return { + "test_name": "test_complex_workflow_integration", + "status": "passed", + "response": "Complex workflow integration completed", + } except Exception as e: logger.error(f"Complex workflow integration test failed: {e}") - return {"test_name": "test_complex_workflow_integration", "status": "failed", "error": str(e)} + return { + "test_name": "test_complex_workflow_integration", + "status": "failed", + "error": str(e), + } + # --- Test Orchestrator --- + def run_all_tests(): """Run all tests and generate a comprehensive report""" logger.info("Starting Enhanced Swarms Comprehensive Test Suite") @@ -595,14 +895,11 @@ def run_all_tests(): test_basic_agent_functionality, test_agent_with_custom_prompt, test_tool_execution_with_agent, - # Multi-Modal Tests test_multimodal_execution, - # Workflow Tests test_sequential_workflow, test_concurrent_workflow, - # Advanced Swarm Tests test_agent_rearrange, test_mixture_of_agents, @@ -615,12 +912,10 @@ def run_all_tests(): test_multi_agent_router, # test_interactive_groupchat, # test_forest_swarm, - # Performance & Features test_streaming_mode, test_agent_memory_persistence, test_error_handling, - # Integration Tests test_complex_workflow_integration, ] @@ -639,30 +934,39 @@ def run_all_tests(): "test_name": test_name, "status": "failed", "error": str(e), - "response": "Test execution failed" + "response": "Test execution failed", } results.append(error_details) # create_github_issue(error_details) # Uncomment to enable GitHub issue creation timestamp = generate_timestamp() - write_markdown_report(results, f"comprehensive_test_report_{timestamp}") - + write_markdown_report( + results, f"comprehensive_test_report_{timestamp}" + ) + # Summary total_tests = len(results) - passed_tests = sum(1 for r in results if r['status'] == 'passed') + passed_tests = sum(1 for r in results if r["status"] == "passed") failed_tests = total_tests - passed_tests - - logger.info(f"Test Summary: {passed_tests}/{total_tests} passed ({(passed_tests/total_tests)*100:.1f}%)") - + + logger.info( + f"Test Summary: {passed_tests}/{total_tests} passed ({(passed_tests/total_tests)*100:.1f}%)" + ) + if failed_tests > 0: - logger.error(f"{failed_tests} tests failed. Check the report and logs.") + logger.error( + f"{failed_tests} tests failed. Check the report and logs." + ) exit(1) else: logger.success("All tests passed successfully!") + if __name__ == "__main__": if not API_KEY: - logger.error("OPENAI_API_KEY environment variable not set. Aborting tests.") + logger.error( + "OPENAI_API_KEY environment variable not set. Aborting tests." + ) exit(1) else: run_all_tests()