Spaces:
Running
Running
File size: 5,234 Bytes
92d2175 |
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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 |
"""
Test file để kiểm tra toàn bộ utils system
"""
import os
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from utils import (
process_question_with_tools,
get_agent_state,
reset_agent_state,
analyze_question_type,
reverse_text_if_needed,
get_system_prompt
)
def test_text_processing():
"""Test text processing tool"""
print("=== Testing Text Processing ===")
# Test reverse text
reversed_question = ".rewsna eht sa \"tfel\" drow eht fo etisoppo eht etirw ,ecnetnes siht dnatsrednu uoy fI"
processed = reverse_text_if_needed(reversed_question)
print(f"Original: {reversed_question}")
print(f"Processed: {processed}")
print()
def test_question_analysis():
"""Test question type analysis"""
print("=== Testing Question Analysis ===")
test_questions = [
"Who was Marie Curie?",
"What is in this YouTube video? https://www.youtube.com/watch?v=dQw4w9WgXcQ",
"Can you read this image for me?",
"What does this audio file say?",
"Calculate 2+2*3",
".rewsna eht sa \"tfel\" drow eht fo etisoppo eht etirw"
]
for question in test_questions:
q_type = analyze_question_type(question)
print(f"Question: {question[:50]}...")
print(f"Type: {q_type}")
print()
def test_state_management():
"""Test state management"""
print("=== Testing State Management ===")
# Reset state
reset_agent_state()
state = get_agent_state()
# Start task
task = state.start_new_task("test_001", "Who is Einstein?")
print(f"Task created: {task.task_id}")
print(f"Question: {task.question}")
# Update context
state.update_task_context(question_type="wiki", processed_question="Who is Einstein?")
print(f"Updated type: {task.question_type}")
# Generate summary
summary = state.generate_context_summary()
print(f"Context summary:\n{summary}")
print()
def test_prompts():
"""Test prompt system"""
print("=== Testing Prompts ===")
# System prompt
system_prompt = get_system_prompt("main_agent")
print(f"System prompt length: {len(system_prompt)} characters")
print(f"First 100 chars: {system_prompt[:100]}...")
print()
def test_orchestration():
"""Test tool orchestration"""
print("=== Testing Tool Orchestration ===")
# Test with Wikipedia question (no external dependencies)
question = "Who was Albert Einstein?"
print(f"Processing question: {question}")
try:
result = process_question_with_tools(question)
print(f"Question type detected: {result['question_type']}")
print(f"Tools used: {result['tools_used']}")
print(f"Successful tools: {result['successful_tools']}")
print(f"Failed tools: {result['failed_tools']}")
# Print tool results summary
if result['tool_results']:
print("\nTool Results Summary:")
for tool_result in result['tool_results']:
status = "✅" if tool_result.success else "❌"
print(f" {status} {tool_result.tool_name}: {tool_result.execution_time:.2f}s")
if not tool_result.success and tool_result.error_message:
print(f" Error: {tool_result.error_message}")
print("\nCached data keys:", list(result['cached_data'].keys()))
except Exception as e:
print(f"Error in orchestration: {e}")
import traceback
traceback.print_exc()
print()
def test_youtube_detection():
"""Test YouTube URL detection"""
print("=== Testing YouTube Detection ===")
youtube_question = "What is this video about? https://www.youtube.com/watch?v=dQw4w9WgXcQ"
print(f"Processing YouTube question: {youtube_question}")
try:
result = process_question_with_tools(youtube_question)
print(f"Question type: {result['question_type']}")
print(f"Tools to run: {result['tools_used']}")
# Check if YouTube tool was attempted
youtube_results = [r for r in result['tool_results'] if r.tool_name == "youtube_tool"]
if youtube_results:
youtube_result = youtube_results[0]
print(f"YouTube tool status: {'Success' if youtube_result.success else 'Failed'}")
if youtube_result.success:
print(f"YouTube content available: {bool(youtube_result.result)}")
else:
print(f"YouTube error: {youtube_result.error_message}")
except Exception as e:
print(f"Error testing YouTube: {e}")
print()
def run_all_tests():
"""Run all tests"""
print("🧪 Starting Utils System Tests\n")
test_text_processing()
test_question_analysis()
test_state_management()
test_prompts()
test_orchestration()
test_youtube_detection()
print("✅ All tests completed!")
if __name__ == "__main__":
# Set environment variables if needed
os.environ.setdefault("GROQ_API_KEY", "your_groq_key_here")
os.environ.setdefault("HF_TOKEN", "your_hf_token_here")
run_all_tests() |