File size: 3,102 Bytes
a20d863 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
import sys
import asyncio
from pathlib import Path
# Add project root to path
project_root = Path(__file__).parent
sys.path.append(str(project_root))
from core.coordinator import coordinator
from core.session import session_manager
async def test_hierarchical_coordination():
"""Test the hierarchical coordination system"""
print("=== Hierarchical Coordination System Test ===")
print()
# Test user query
user_query = "What are the key principles of effective time management?"
user_id = "test_user"
print(f"User Query: {user_query}")
print()
# Test coordination status
print("1. Testing Coordination Status:")
try:
coord_status = coordinator.get_coordination_status()
print(f" Tavily Available: {coord_status.get('tavily_available', False)}")
print(f" Weather Available: {coord_status.get('weather_available', False)}")
print(f" Web Search Enabled: {coord_status.get('web_search_enabled', False)}")
print(" ✅ Coordination Status Check Passed")
except Exception as e:
print(f" ❌ Coordination Status Check Failed: {e}")
print()
# Test hierarchical conversation coordination
print("2. Testing Hierarchical Conversation Coordination:")
try:
print(" Starting hierarchical coordination...")
response_count = 0
async for response_chunk in coordinator.coordinate_hierarchical_conversation(user_id, user_query):
response_count += 1
print(f" Chunk {response_count}: {response_chunk['type']} - {response_chunk['content'][:50]}...")
# Limit output for readability
if response_count >= 5:
print(" ... (truncated for brevity)")
break
print(" ✅ Hierarchical Coordination Test Passed")
except Exception as e:
print(f" ❌ Hierarchical Coordination Test Failed: {e}")
print()
# Test hierarchical session tracking
print("3. Testing Hierarchical Session Tracking:")
try:
# Update with test coordination data
test_data = {
'hf_engaged': True,
'ollama_responded': True,
'success': True
}
update_result = session_manager.update_hierarchical_coordination(user_id, test_data)
print(f" Update Result: {'✅ Success' if update_result else '❌ Failed'}")
# Get hierarchical stats
stats = session_manager.get_hierarchical_stats(user_id)
print(f" Total Conversations: {stats.get('total_conversations', 0)}")
print(f" HF Engagements: {stats.get('hf_engagements', 0)}")
print(f" Ollama Responses: {stats.get('ollama_responses', 0)}")
print(" ✅ Hierarchical Session Tracking Passed")
except Exception as e:
print(f" ❌ Hierarchical Session Tracking Failed: {e}")
print()
print("🎉 Hierarchical Coordination System Test Completed!")
if __name__ == "__main__":
asyncio.run(test_hierarchical_coordination())
|