From 903bb2a8c8bccae8f36d9292d6e14ed048a1243b Mon Sep 17 00:00:00 2001 From: Steve-Dusty Date: Thu, 13 Nov 2025 15:03:58 -0800 Subject: [PATCH] removed unnecessary test files --- test.py | 268 ------------------------------- test_majority_voting_complete.py | 143 ----------------- 2 files changed, 411 deletions(-) delete mode 100644 test.py delete mode 100644 test_majority_voting_complete.py diff --git a/test.py b/test.py deleted file mode 100644 index 86cef9b3..00000000 --- a/test.py +++ /dev/null @@ -1,268 +0,0 @@ -from swarms.structs.agent import Agent -from swarms.structs.majority_voting import MajorityVoting - - -def test_majority_voting_basic_execution(): - """Test basic MajorityVoting execution with multiple agents""" - # Create specialized agents with different perspectives - geographer = Agent( - agent_name="Geography-Expert", - agent_description="Expert in geography and world capitals", - model_name="gpt-4o", - max_loops=1, - ) - - historian = Agent( - agent_name="History-Scholar", - agent_description="Historical and cultural context specialist", - model_name="gpt-4o", - max_loops=1, - ) - - political_analyst = Agent( - agent_name="Political-Analyst", - agent_description="Political and administrative specialist", - model_name="gpt-4o", - max_loops=1, - ) - - # Create majority voting system - mv = MajorityVoting( - name="Geography-Consensus-System", - description="Majority voting system for geographical questions", - agents=[geographer, historian, political_analyst], - max_loops=1, - verbose=True, - ) - - # Test execution - result = mv.run("What is the capital city of France?") - assert result is not None - - -def test_majority_voting_multiple_loops(): - """Test MajorityVoting with multiple loops for consensus refinement""" - # Create agents with different knowledge bases - trivia_expert = Agent( - agent_name="Trivia-Expert", - agent_description="General knowledge and trivia specialist", - model_name="gpt-4o", - max_loops=1, - ) - - research_analyst = Agent( - agent_name="Research-Analyst", - agent_description="Research and fact-checking specialist", - model_name="gpt-4o", - max_loops=1, - ) - - subject_matter_expert = Agent( - agent_name="Subject-Matter-Expert", - agent_description="Deep subject matter expertise specialist", - model_name="gpt-4o", - max_loops=1, - ) - - # Create majority voting with multiple loops for iterative refinement - mv = MajorityVoting( - name="Multi-Loop-Consensus-System", - description="Majority voting with iterative consensus refinement", - agents=[ - trivia_expert, - research_analyst, - subject_matter_expert, - ], - max_loops=3, # Allow multiple iterations - verbose=True, - ) - - # Test multi-loop execution - result = mv.run( - "What are the main causes of climate change and what can be done to mitigate them?" - ) - assert result is not None - - -def test_majority_voting_business_scenario(): - """Test MajorityVoting in a realistic business scenario""" - # Create agents representing different business perspectives - market_strategist = Agent( - agent_name="Market-Strategist", - agent_description="Market strategy and competitive analysis specialist", - model_name="gpt-4o", - max_loops=1, - ) - - financial_analyst = Agent( - agent_name="Financial-Analyst", - agent_description="Financial modeling and ROI analysis specialist", - model_name="gpt-4o", - max_loops=1, - ) - - technical_architect = Agent( - agent_name="Technical-Architect", - agent_description="Technical feasibility and implementation specialist", - model_name="gpt-4o", - max_loops=1, - ) - - risk_manager = Agent( - agent_name="Risk-Manager", - agent_description="Risk assessment and compliance specialist", - model_name="gpt-4o", - max_loops=1, - ) - - operations_expert = Agent( - agent_name="Operations-Expert", - agent_description="Operations and implementation specialist", - model_name="gpt-4o", - max_loops=1, - ) - - # Create majority voting for business decisions - mv = MajorityVoting( - name="Business-Decision-Consensus", - description="Majority voting system for business strategic decisions", - agents=[ - market_strategist, - financial_analyst, - technical_architect, - risk_manager, - operations_expert, - ], - max_loops=2, - verbose=True, - ) - - # Test with complex business decision - result = mv.run( - "Should our company invest in developing an AI-powered customer service platform? " - "Consider market demand, financial implications, technical feasibility, risk factors, " - "and operational requirements." - ) - - assert result is not None - - -def test_majority_voting_error_handling(): - """Test MajorityVoting error handling and validation""" - # Test with empty agents list - try: - MajorityVoting(agents=[]) - assert ( - False - ), "Should have raised ValueError for empty agents list" - except ValueError as e: - assert "agents" in str(e).lower() or "empty" in str(e).lower() - - # Test with invalid max_loops - analyst = Agent( - agent_name="Test-Analyst", - agent_description="Test analyst", - model_name="gpt-4o", - max_loops=1, - ) - - try: - MajorityVoting(agents=[analyst], max_loops=0) - assert ( - False - ), "Should have raised ValueError for invalid max_loops" - except ValueError as e: - assert "max_loops" in str(e).lower() or "0" in str(e) - - -def test_majority_voting_different_output_types(): - """Test MajorityVoting with different output types""" - # Create agents for technical analysis - Agent( - agent_name="Security-Expert", - agent_description="Cybersecurity and data protection specialist", - model_name="gpt-4o", - max_loops=1, - ) - - Agent( - agent_name="Compliance-Officer", - agent_description="Regulatory compliance and legal specialist", - model_name="gpt-4o", - max_loops=1, - ) - - Agent( - agent_name="Privacy-Advocate", - agent_description="Privacy protection and data rights specialist", - model_name="gpt-4o", - max_loops=1, - ) - - # Assert majority vote is correct - assert majority_vote is not None - - -def test_streaming_majority_voting(): - """ - Test the streaming_majority_voting with logging/try-except and assertion. - """ - logs = [] - - def streaming_callback( - agent_name: str, chunk: str, is_final: bool - ): - # Chunk buffer static per call (reset each session) - if not hasattr(streaming_callback, "_buffer"): - streaming_callback._buffer = "" - streaming_callback._buffer_size = 0 - - min_chunk_size = 512 # or any large chunk size you want - - if chunk: - streaming_callback._buffer += chunk - streaming_callback._buffer_size += len(chunk) - if ( - streaming_callback._buffer_size >= min_chunk_size - or is_final - ): - if streaming_callback._buffer: - print(streaming_callback._buffer, end="", flush=True) - logs.append(streaming_callback._buffer) - streaming_callback._buffer = "" - streaming_callback._buffer_size = 0 - if is_final: - print() - - try: - # Initialize the agent - agent = Agent( - agent_name="Financial-Analysis-Agent", - agent_description="Personal finance advisor agent", - system_prompt="You are a financial analysis agent.", # replaced missing const - max_loops=1, - model_name="gpt-4.1", - dynamic_temperature_enabled=True, - user_name="swarms_corp", - retry_attempts=3, - context_length=8192, - return_step_meta=False, - output_type="str", # "json", "dict", "csv" OR "string" "yaml" and - auto_generate_prompt=False, # Auto generate prompt for the agent based on name, description, and system prompt, task - max_tokens=4000, # max output tokens - saved_state_path="agent_00.json", - interactive=False, - streaming_on=True, # if concurrent agents want to be streamed - ) - - swarm = MajorityVoting(agents=[agent, agent, agent]) - - result = swarm.run( - "Create a table of super high growth opportunities for AI. I have $40k to invest in ETFs, index funds, and more. Please create a table in markdown.", - streaming_callback=streaming_callback, - ) - assert result is not None - except Exception as e: - print("Error in test_streaming_majority_voting:", e) - print("Logs so far:", logs) - raise diff --git a/test_majority_voting_complete.py b/test_majority_voting_complete.py deleted file mode 100644 index 563a7bb0..00000000 --- a/test_majority_voting_complete.py +++ /dev/null @@ -1,143 +0,0 @@ -""" -Complete test to verify MajorityVoting works correctly after the fix. -Tests that all features work the same from API perspective. -""" -from swarms.structs.agent import Agent -from swarms.structs.majority_voting import MajorityVoting - - -def test_complete_functionality(): - """Test that all MajorityVoting features work correctly""" - - print("=" * 70) - print("COMPLETE MAJORITY VOTING FUNCTIONALITY TEST") - print("=" * 70) - - # Create test agents (simulating what the API would create) - print("\n1. Creating worker agents...") - agent1 = Agent( - agent_name="Financial-Analyst", - agent_description="Analyzes financial aspects", - system_prompt="You are a financial analyst.", - model_name="gpt-4o-mini", - max_loops=1, - ) - - agent2 = Agent( - agent_name="Tech-Expert", - agent_description="Understands tech industry", - system_prompt="You are a tech industry expert.", - model_name="gpt-4o-mini", - max_loops=1, - ) - - agent3 = Agent( - agent_name="Risk-Assessor", - agent_description="Evaluates risks", - system_prompt="You are a risk assessor.", - model_name="gpt-4o-mini", - max_loops=1, - ) - print(" ✓ Created 3 worker agents") - - # Test 1: Create MajorityVoting (as API would) - print("\n2. Creating MajorityVoting swarm...") - try: - mv = MajorityVoting( - name="Investment-Analysis-Swarm", - description="A swarm for investment analysis", - agents=[agent1, agent2, agent3], - max_loops=1, - verbose=False, - ) - print(" ✓ MajorityVoting created successfully") - except Exception as e: - print(f" ✗ Failed to create MajorityVoting: {e}") - raise - - # Test 2: Verify internal consensus agent was created - print("\n3. Verifying internal consensus agent...") - try: - assert mv.consensus_agent is not None, "Consensus agent should exist" - assert mv.consensus_agent.agent_name == "Consensus-Agent", \ - f"Expected 'Consensus-Agent', got '{mv.consensus_agent.agent_name}'" - print(f" ✓ Consensus agent created: {mv.consensus_agent.agent_name}") - except Exception as e: - print(f" ✗ Consensus agent verification failed: {e}") - raise - - # Test 3: Verify conversation object - print("\n4. Verifying conversation object...") - try: - assert mv.conversation is not None, "Conversation should exist" - print(" ✓ Conversation object created successfully") - except Exception as e: - print(f" ✗ Conversation verification failed: {e}") - raise - - # Test 4: Verify all worker agents are registered - print("\n5. Verifying worker agents...") - try: - assert len(mv.agents) == 3, f"Expected 3 agents, got {len(mv.agents)}" - agent_names = [a.agent_name for a in mv.agents] - print(f" ✓ All 3 worker agents registered: {agent_names}") - except Exception as e: - print(f" ✗ Worker agents verification failed: {e}") - raise - - # Test 5: Test with custom consensus agent configuration - print("\n6. Testing custom consensus agent configuration...") - try: - mv_custom = MajorityVoting( - name="Custom-Consensus-Swarm", - description="Swarm with custom consensus agent", - agents=[agent1, agent2], - consensus_agent_name="Custom-Consensus", - consensus_agent_model_name="gpt-4o-mini", - consensus_agent_prompt="You are a custom consensus agent.", - max_loops=1, - ) - assert mv_custom.consensus_agent.agent_name == "Custom-Consensus" - print(f" ✓ Custom consensus agent: {mv_custom.consensus_agent.agent_name}") - except Exception as e: - print(f" ✗ Custom consensus configuration failed: {e}") - raise - - # Test 6: Verify backward compatibility (consensus_agent param should be ignored) - print("\n7. Testing backward compatibility with unused consensus_agent param...") - try: - dummy_agent = Agent( - agent_name="Dummy", - system_prompt="Dummy", - model_name="gpt-4o-mini", - max_loops=1, - ) - mv_compat = MajorityVoting( - name="Backward-Compat-Swarm", - description="Testing backward compatibility", - agents=[agent1, agent2], - consensus_agent=dummy_agent, # This should be ignored - max_loops=1, - ) - # The consensus agent should still be the default one, not the dummy - assert mv_compat.consensus_agent.agent_name == "Consensus-Agent" - print(" ✓ Unused consensus_agent parameter properly ignored") - except Exception as e: - print(f" ✗ Backward compatibility test failed: {e}") - raise - - print("\n" + "=" * 70) - print("✅ ALL TESTS PASSED!") - print("=" * 70) - print("\nConclusion:") - print("- All MajorityVoting features work correctly") - print("- Consensus agent is properly created internally") - print("- Worker agents are properly registered") - print("- Custom consensus configuration works") - print("- Backward compatibility maintained") - print("- API will work without errors") - print("=" * 70) - - -if __name__ == "__main__": - test_complete_functionality()