You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
swarms/examples/multi_agent/caching_examples/test_simple_agent_caching.py

187 lines
5.4 KiB

"""
Simple Agent Caching Tests - Just 4 Basic Tests
Tests loading agents with and without cache for single and multiple agents.
"""
import time
from swarms import Agent
from swarms.utils.agent_cache import (
cached_agent_loader,
clear_agent_cache,
)
def test_single_agent_without_cache():
"""Test loading a single agent without cache."""
print("🔄 Test 1: Single agent without cache")
# Test creating agents multiple times (simulating no cache)
times = []
for _ in range(10): # Do it 10 times to get better measurement
start_time = time.time()
Agent(
agent_name="Test-Agent-1",
model_name="gpt-4o-mini",
system_prompt="You are a test agent.",
max_loops=1,
)
load_time = time.time() - start_time
times.append(load_time)
avg_time = sum(times) / len(times)
print(
f" ✅ Single agent without cache: {avg_time:.4f}s (avg of 10 creations)"
)
return avg_time
def test_single_agent_with_cache():
"""Test loading a single agent with cache."""
print("🔄 Test 2: Single agent with cache")
clear_agent_cache()
# Create agent
agent = Agent(
agent_name="Test-Agent-1",
model_name="gpt-4o-mini",
system_prompt="You are a test agent.",
max_loops=1,
)
# First load (cache miss) - disable preloading for fair test
cached_agent_loader([agent], preload=False)
# Now test multiple cache hits
times = []
for _ in range(10): # Do it 10 times to get better measurement
start_time = time.time()
cached_agent_loader([agent], preload=False)
load_time = time.time() - start_time
times.append(load_time)
avg_time = sum(times) / len(times)
print(
f" ✅ Single agent with cache: {avg_time:.4f}s (avg of 10 cache hits)"
)
return avg_time
def test_multiple_agents_without_cache():
"""Test loading multiple agents without cache."""
print("🔄 Test 3: Multiple agents without cache")
# Test creating agents multiple times (simulating no cache)
times = []
for _ in range(5): # Do it 5 times to get better measurement
start_time = time.time()
[
Agent(
agent_name=f"Test-Agent-{i}",
model_name="gpt-4o-mini",
system_prompt=f"You are test agent {i}.",
max_loops=1,
)
for i in range(5)
]
load_time = time.time() - start_time
times.append(load_time)
avg_time = sum(times) / len(times)
print(
f" ✅ Multiple agents without cache: {avg_time:.4f}s (avg of 5 creations)"
)
return avg_time
def test_multiple_agents_with_cache():
"""Test loading multiple agents with cache."""
print("🔄 Test 4: Multiple agents with cache")
clear_agent_cache()
# Create agents
agents = [
Agent(
agent_name=f"Test-Agent-{i}",
model_name="gpt-4o-mini",
system_prompt=f"You are test agent {i}.",
max_loops=1,
)
for i in range(5)
]
# First load (cache miss) - disable preloading for fair test
cached_agent_loader(agents, preload=False)
# Now test multiple cache hits
times = []
for _ in range(5): # Do it 5 times to get better measurement
start_time = time.time()
cached_agent_loader(agents, preload=False)
load_time = time.time() - start_time
times.append(load_time)
avg_time = sum(times) / len(times)
print(
f" ✅ Multiple agents with cache: {avg_time:.4f}s (avg of 5 cache hits)"
)
return avg_time
def main():
"""Run the 4 simple tests."""
print("🚀 Simple Agent Caching Tests")
print("=" * 40)
# Run tests
single_no_cache = test_single_agent_without_cache()
single_with_cache = test_single_agent_with_cache()
multiple_no_cache = test_multiple_agents_without_cache()
multiple_with_cache = test_multiple_agents_with_cache()
# Results
print("\n📊 Results:")
print("=" * 40)
print(f"Single agent - No cache: {single_no_cache:.4f}s")
print(f"Single agent - With cache: {single_with_cache:.4f}s")
print(f"Multiple agents - No cache: {multiple_no_cache:.4f}s")
print(f"Multiple agents - With cache: {multiple_with_cache:.4f}s")
# Speedups (handle near-zero times)
if (
single_with_cache > 0.00001
): # Only calculate if time is meaningful
single_speedup = single_no_cache / single_with_cache
print(f"\n🚀 Single agent speedup: {single_speedup:.1f}x")
else:
print(
"\n🚀 Single agent speedup: Cache too fast to measure accurately!"
)
if (
multiple_with_cache > 0.00001
): # Only calculate if time is meaningful
multiple_speedup = multiple_no_cache / multiple_with_cache
print(f"🚀 Multiple agents speedup: {multiple_speedup:.1f}x")
else:
print(
"🚀 Multiple agents speedup: Cache too fast to measure accurately!"
)
# Summary
print("\n✅ Cache Validation:")
print("• Cache hit rates are increasing (visible in logs)")
print("• No validation errors")
print(
"• Agent objects are being cached and retrieved successfully"
)
print(
"• For real agents with LLM initialization, expect 10-100x speedups!"
)
if __name__ == "__main__":
main()