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

removed unnecessary test files

Browse Source
pull/1203/head
Steve-Dusty 1 month ago
parent ec3c64c621
commit 903bb2a8c8
2 changed files with 0 additions and 411 deletions
Show Stats Download Patch File Download Diff File

268
test.py
Unescape View File

@ -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

143
test_majority_voting_complete.py
Unescape View File

@ -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()
Write Preview
Loading…
Cancel
Save