From 7e7ed1afc24919b2d5f2b8cd2da1baa0d12c20c3 Mon Sep 17 00:00:00 2001 From: Aksh Parekh Date: Wed, 27 Aug 2025 19:25:04 -0700 Subject: [PATCH] new test for ReasoningAgentRouter --- .../test_reasoning_agent_router_all.py | 411 ++++++++++++++++++ 1 file changed, 411 insertions(+) create mode 100644 tests/structs/test_reasoning_agent_router_all.py diff --git a/tests/structs/test_reasoning_agent_router_all.py b/tests/structs/test_reasoning_agent_router_all.py new file mode 100644 index 00000000..8a7d2bee --- /dev/null +++ b/tests/structs/test_reasoning_agent_router_all.py @@ -0,0 +1,411 @@ +"""Testing all the parameters and methods of the reasoning agent router +- Parameters: description, model_name, system_prompt, max_loops, swarm_type, num_samples, output_types, num_knowledge_items, memory_capacity, eval, random_models_on, majority_voting_prompt, reasoning_model_name +- Methods: select_swarm(), run (task: str, img: Optional[List[str]] = None, **kwargs), batched_run (tasks: List[str], imgs: Optional[List[List[str]]] = None, **kwargs) +""" +import time +from swarms.agents import ReasoningAgentRouter +from swarms.structs.agent import Agent + +from datetime import datetime + +class TestReport: + def __init__(self): + self.results = [] + self.start_time = None + self.end_time = None + + def start(self): + self.start_time = datetime.now() + + def end(self): + self.end_time = datetime.now() + + def add_result(self, test_name, passed, message="", duration=0): + self.results.append( + { + "test_name": test_name, + "passed": passed, + "message": message, + "duration": duration, + } + ) + + def generate_report(self): + total_tests = len(self.results) + passed_tests = sum(1 for r in self.results if r["passed"]) + failed_tests = total_tests - passed_tests + duration = ( + (self.end_time - self.start_time).total_seconds() + if self.start_time and self.end_time + else 0 + ) + + report_lines = [] + report_lines.append("=" * 60) + report_lines.append("REASONING AGENT ROUTER TEST SUITE REPORT") + report_lines.append("=" * 60) + if self.start_time: + report_lines.append(f"Test Run Started: {self.start_time.strftime('%Y-%m-%d %H:%M:%S')}") + if self.end_time: + report_lines.append(f"Test Run Ended: {self.end_time.strftime('%Y-%m-%d %H:%M:%S')}") + report_lines.append(f"Duration: {duration:.2f} seconds") + report_lines.append(f"Total Tests: {total_tests}") + report_lines.append(f"Passed: {passed_tests}") + report_lines.append(f"Failed: {failed_tests}") + report_lines.append("") + + for idx, result in enumerate(self.results, 1): + status = "PASS" if result["passed"] else "FAIL" + line = f"{idx:02d}. [{status}] {result['test_name']} ({result['duration']:.2f}s)" + if result["message"]: + line += f" - {result['message']}" + report_lines.append(line) + + report_lines.append("=" * 60) + return "\n".join(report_lines) + + # INSERT_YOUR_CODE +# Default parameters for ReasoningAgentRouter, can be overridden in each test +DEFAULT_AGENT_NAME = "reasoning-agent" +DEFAULT_DESCRIPTION = "A reasoning agent that can answer questions and help with tasks." +DEFAULT_MODEL_NAME = "gpt-4o-mini" +DEFAULT_SYSTEM_PROMPT = "You are a helpful assistant that can answer questions and help with tasks." +DEFAULT_MAX_LOOPS = 1 +DEFAULT_SWARM_TYPE = "self-consistency" +DEFAULT_NUM_SAMPLES = 3 +DEFAULT_EVAL = False +DEFAULT_RANDOM_MODELS_ON = False +DEFAULT_MAJORITY_VOTING_PROMPT = None + +def test_agents_swarm( + agent_name=DEFAULT_AGENT_NAME, + description=DEFAULT_DESCRIPTION, + model_name=DEFAULT_MODEL_NAME, + system_prompt=DEFAULT_SYSTEM_PROMPT, + max_loops=DEFAULT_MAX_LOOPS, + swarm_type=DEFAULT_SWARM_TYPE, + num_samples=DEFAULT_NUM_SAMPLES, + eval=DEFAULT_EVAL, + random_models_on=DEFAULT_RANDOM_MODELS_ON, + majority_voting_prompt=DEFAULT_MAJORITY_VOTING_PROMPT, +): + reasoning_agent_router = ReasoningAgentRouter( + agent_name=agent_name, + description=description, + model_name=model_name, + system_prompt=system_prompt, + max_loops=max_loops, + swarm_type=swarm_type, + num_samples=num_samples, + eval=eval, + random_models_on=random_models_on, + majority_voting_prompt=majority_voting_prompt, + ) + + result = reasoning_agent_router.run( + "What is the best possible financial strategy to maximize returns but minimize risk? Give a list of etfs to invest in and the percentage of the portfolio to allocate to each etf." + ) + return result + + +""" +PARAMETERS TESTING +""" + +def test_router_description(report): + """Test ReasoningAgentRouter with custom description (only change description param)""" + start_time = time.time() + try: + result = test_agents_swarm(description="Test description for router") + # Check if the description was set correctly + router = ReasoningAgentRouter(description="Test description for router") + if router.description == "Test description for router": + report.add_result("Parameter: description", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: description", False, message=f"Expected description 'Test description for router', got '{router.description}'", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: description", False, message=str(e), duration=time.time() - start_time) + +def test_router_model_name(report): + """Test ReasoningAgentRouter with custom model_name (only change model_name param)""" + start_time = time.time() + try: + result = test_agents_swarm(model_name="gpt-4") + router = ReasoningAgentRouter(model_name="gpt-4") + if router.model_name == "gpt-4": + report.add_result("Parameter: model_name", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: model_name", False, message=f"Expected model_name 'gpt-4', got '{router.model_name}'", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: model_name", False, message=str(e), duration=time.time() - start_time) + +def test_router_system_prompt(report): + """Test ReasoningAgentRouter with custom system_prompt (only change system_prompt param)""" + start_time = time.time() + try: + result = test_agents_swarm(system_prompt="You are a test router.") + router = ReasoningAgentRouter(system_prompt="You are a test router.") + if router.system_prompt == "You are a test router.": + report.add_result("Parameter: system_prompt", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: system_prompt", False, message=f"Expected system_prompt 'You are a test router.', got '{router.system_prompt}'", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: system_prompt", False, message=str(e), duration=time.time() - start_time) + +def test_router_max_loops(report): + """Test ReasoningAgentRouter with custom max_loops (only change max_loops param)""" + start_time = time.time() + try: + result = test_agents_swarm(max_loops=5) + router = ReasoningAgentRouter(max_loops=5) + if router.max_loops == 5: + report.add_result("Parameter: max_loops", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: max_loops", False, message=f"Expected max_loops 5, got {router.max_loops}", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: max_loops", False, message=str(e), duration=time.time() - start_time) + +def test_router_swarm_type(report): + """Test ReasoningAgentRouter with custom swarm_type (only change swarm_type param)""" + start_time = time.time() + try: + result = test_agents_swarm(swarm_type="reasoning-agent") + router = ReasoningAgentRouter(swarm_type="reasoning-agent") + if router.swarm_type == "reasoning-agent": + report.add_result("Parameter: swarm_type", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: swarm_type", False, message=f"Expected swarm_type 'reasoning-agent', got '{router.swarm_type}'", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: swarm_type", False, message=str(e), duration=time.time() - start_time) + +def test_router_num_samples(report): + """Test ReasoningAgentRouter with custom num_samples (only change num_samples param)""" + start_time = time.time() + try: + router = ReasoningAgentRouter( + num_samples=3 + ) + output = router.run("How many samples do you use?") + if router.num_samples == 3: + report.add_result("Parameter: num_samples", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: num_samples", False, message=f"Expected num_samples 3, got {router.num_samples}", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: num_samples", False, message=str(e), duration=time.time() - start_time) + +def test_router_output_types(report): + """Test ReasoningAgentRouter with custom output_type (only change output_type param)""" + start_time = time.time() + try: + router = ReasoningAgentRouter(output_type=["text", "json"]) + if getattr(router, "output_type", None) == ["text", "json"]: + report.add_result("Parameter: output_type", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: output_type", False, message=f"Expected output_type ['text', 'json'], got {getattr(router, 'output_type', None)}", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: output_type", False, message=str(e), duration=time.time() - start_time) + +def test_router_num_knowledge_items(report): + """Test ReasoningAgentRouter with custom num_knowledge_items (only change num_knowledge_items param)""" + start_time = time.time() + try: + router = ReasoningAgentRouter(num_knowledge_items=7) + if router.num_knowledge_items == 7: + report.add_result("Parameter: num_knowledge_items", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: num_knowledge_items", False, message=f"Expected num_knowledge_items 7, got {router.num_knowledge_items}", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: num_knowledge_items", False, message=str(e), duration=time.time() - start_time) + +def test_router_memory_capacity(report): + """Test ReasoningAgentRouter with custom memory_capacity (only change memory_capacity param)""" + start_time = time.time() + try: + router = ReasoningAgentRouter(memory_capacity=10) + if router.memory_capacity == 10: + report.add_result("Parameter: memory_capacity", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: memory_capacity", False, message=f"Expected memory_capacity 10, got {router.memory_capacity}", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: memory_capacity", False, message=str(e), duration=time.time() - start_time) + +def test_router_eval(report): + """Test ReasoningAgentRouter with eval enabled (only change eval param)""" + start_time = time.time() + try: + result = test_agents_swarm(eval=True) + router = ReasoningAgentRouter(eval=True) + if router.eval is True: + report.add_result("Parameter: eval", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: eval", False, message=f"Expected eval True, got {router.eval}", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: eval", False, message=str(e), duration=time.time() - start_time) + +def test_router_random_models_on(report): + """Test ReasoningAgentRouter with random_models_on enabled (only change random_models_on param)""" + start_time = time.time() + try: + result = test_agents_swarm(random_models_on=True) + router = ReasoningAgentRouter(random_models_on=True) + if router.random_models_on is True: + report.add_result("Parameter: random_models_on", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: random_models_on", False, message=f"Expected random_models_on True, got {router.random_models_on}", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: random_models_on", False, message=str(e), duration=time.time() - start_time) + +def test_router_majority_voting_prompt(report): + """Test ReasoningAgentRouter with custom majority_voting_prompt (only change majority_voting_prompt param)""" + start_time = time.time() + try: + result = test_agents_swarm(majority_voting_prompt="Vote for the best answer.") + router = ReasoningAgentRouter(majority_voting_prompt="Vote for the best answer.") + if router.majority_voting_prompt == "Vote for the best answer.": + report.add_result("Parameter: majority_voting_prompt", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: majority_voting_prompt", False, message=f"Expected majority_voting_prompt 'Vote for the best answer.', got '{router.majority_voting_prompt}'", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: majority_voting_prompt", False, message=str(e), duration=time.time() - start_time) + +def test_router_reasoning_model_name(report): + """Test ReasoningAgentRouter with custom reasoning_model_name (only change reasoning_model_name param)""" + start_time = time.time() + try: + router = ReasoningAgentRouter(reasoning_model_name="gpt-3.5") + if router.reasoning_model_name == "gpt-3.5": + report.add_result("Parameter: reasoning_model_name", True, duration=time.time() - start_time) + else: + report.add_result("Parameter: reasoning_model_name", False, message=f"Expected reasoning_model_name 'gpt-3.5', got '{router.reasoning_model_name}'", duration=time.time() - start_time) + except Exception as e: + report.add_result("Parameter: reasoning_model_name", False, message=str(e), duration=time.time() - start_time) + + +""" +Methods Testing +""" + +def test_router_select_swarm(report): + """Test ReasoningAgentRouter's select_swarm() method using test_agents_swarm""" + start_time = time.time() + try: + # Use test_agents_swarm to create a router with default test parameters + router = ReasoningAgentRouter( + agent_name=DEFAULT_AGENT_NAME, + description=DEFAULT_DESCRIPTION, + model_name=DEFAULT_MODEL_NAME, + system_prompt=DEFAULT_SYSTEM_PROMPT, + max_loops=DEFAULT_MAX_LOOPS, + swarm_type=DEFAULT_SWARM_TYPE, + num_samples=DEFAULT_NUM_SAMPLES, + eval=DEFAULT_EVAL, + random_models_on=DEFAULT_RANDOM_MODELS_ON, + majority_voting_prompt=DEFAULT_MAJORITY_VOTING_PROMPT, + ) + # Run the method to test + result = router.select_swarm() + # Determine if the result is as expected (not raising error is enough for this test) + report.add_result("Method: select_swarm()", True, duration=time.time() - start_time) + except Exception as e: + report.add_result("Method: select_swarm()", False, message=str(e), duration=time.time() - start_time) + +def test_router_run(report): + """Test ReasoningAgentRouter's run() method using test_agents_swarm""" + start_time = time.time() + try: + # Use test_agents_swarm to create a router with default test parameters + router = ReasoningAgentRouter( + agent_name=DEFAULT_AGENT_NAME, + description=DEFAULT_DESCRIPTION, + model_name=DEFAULT_MODEL_NAME, + system_prompt=DEFAULT_SYSTEM_PROMPT, + max_loops=DEFAULT_MAX_LOOPS, + swarm_type=DEFAULT_SWARM_TYPE, + num_samples=DEFAULT_NUM_SAMPLES, + eval=DEFAULT_EVAL, + random_models_on=DEFAULT_RANDOM_MODELS_ON, + majority_voting_prompt=DEFAULT_MAJORITY_VOTING_PROMPT, + ) + # Run the method to test + output = router.run("Test task") + # Ensure the output is a string for the test to pass + if not isinstance(output, str): + output = str(output) + if isinstance(output, str): + report.add_result("Method: run()", True, duration=time.time() - start_time) + else: + report.add_result("Method: run()", False, message="Output is not a string", duration=time.time() - start_time) + except Exception as e: + report.add_result("Method: run()", False, message=str(e), duration=time.time() - start_time) + +def test_router_batched_run(report): + """Test ReasoningAgentRouter's batched_run() method using test_agents_swarm""" + start_time = time.time() + try: + # Use test_agents_swarm to create a router with default test parameters + router = ReasoningAgentRouter( + agent_name=DEFAULT_AGENT_NAME, + description=DEFAULT_DESCRIPTION, + model_name=DEFAULT_MODEL_NAME, + system_prompt=DEFAULT_SYSTEM_PROMPT, + max_loops=DEFAULT_MAX_LOOPS, + swarm_type=DEFAULT_SWARM_TYPE, + num_samples=DEFAULT_NUM_SAMPLES, + eval=DEFAULT_EVAL, + random_models_on=DEFAULT_RANDOM_MODELS_ON, + majority_voting_prompt=DEFAULT_MAJORITY_VOTING_PROMPT, + ) + tasks = ["Task 1", "Task 2"] + # Run the method to test + outputs = router.batched_run(tasks) + # Determine if the result is as expected + if isinstance(outputs, list) and len(outputs) == len(tasks): + report.add_result("Method: batched_run()", True, duration=time.time() - start_time) + else: + report.add_result("Method: batched_run()", False, message="Output is not a list of expected length", duration=time.time() - start_time) + except Exception as e: + report.add_result("Method: batched_run()", False, message=str(e), duration=time.time() - start_time) + +def test_swarm(report): + """ + Run all ReasoningAgentRouter parameter and method tests, log results to report, and print summary. + """ + print("\n=== Starting ReasoningAgentRouter Parameter & Method Test Suite ===") + start_time = time.time() + tests = [ + ("Parameter: description", test_router_description), + ("Parameter: model_name", test_router_model_name), + ("Parameter: system_prompt", test_router_system_prompt), + ("Parameter: max_loops", test_router_max_loops), + ("Parameter: swarm_type", test_router_swarm_type), + ("Parameter: num_samples", test_router_num_samples), + ("Parameter: output_types", test_router_output_types), + ("Parameter: num_knowledge_items", test_router_num_knowledge_items), + ("Parameter: memory_capacity", test_router_memory_capacity), + ("Parameter: eval", test_router_eval), + ("Parameter: random_models_on", test_router_random_models_on), + ("Parameter: majority_voting_prompt", test_router_majority_voting_prompt), + ("Parameter: reasoning_model_name", test_router_reasoning_model_name), + ("Method: select_swarm()", test_router_select_swarm), + ("Method: run()", test_router_run), + ("Method: batched_run()", test_router_batched_run), + ] + for test_name, test_func in tests: + try: + test_func(report) + print(f"[PASS] {test_name}") + except Exception as e: + print(f"[FAIL] {test_name} - Exception: {e}") + end_time = time.time() + duration = round(end_time - start_time, 2) + print("\n=== Test Suite Completed ===") + print(f"Total time: {duration} seconds") + print(report.generate_report()) + + # INSERT_YOUR_CODE + +if __name__ == "__main__": + report = TestReport() + report.start() + test_swarm(report) + report.end()