|
import sys |
|
import asyncio |
|
from pathlib import 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_enhanced_features(): |
|
"""Test the enhanced UI and coordination features""" |
|
print("=== Enhanced Features Test ===") |
|
print() |
|
|
|
|
|
user_query = "What are the benefits of meditation for stress management?" |
|
user_id = "test_user" |
|
|
|
print(f"User Query: {user_query}") |
|
print() |
|
|
|
|
|
print("1. Testing Enhanced Coordination with Detailed Tracking:") |
|
try: |
|
print(" Starting enhanced 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}:") |
|
print(f" Type: {response_chunk['type']}") |
|
print(f" Content: {response_chunk['content'][:100]}...") |
|
if 'details' in response_chunk: |
|
print(f" Details: {response_chunk['details']}") |
|
print() |
|
|
|
|
|
if response_count >= 8: |
|
print(" ... (truncated for brevity)") |
|
break |
|
|
|
print(" β
Enhanced Coordination Test Passed") |
|
except Exception as e: |
|
print(f" β Enhanced Coordination Test Failed: {e}") |
|
|
|
print() |
|
|
|
|
|
print("2. 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() |
|
print("π Enhanced Features Test Completed!") |
|
|
|
if __name__ == "__main__": |
|
asyncio.run(test_enhanced_features()) |
|
|