diff --git a/examples/features/blocked_domains.py b/examples/features/blocked_domains.py index 3de5e3046..a6f9e0738 100644 --- a/examples/features/blocked_domains.py +++ b/examples/features/blocked_domains.py @@ -17,13 +17,13 @@ llm = ChatOpenAI(model='gpt-4o-mini') task = 'Navigate to example.com, then try to go to x.com, then facebook.com, and finally visit google.com. Tell me which sites you were able to access.' prohibited_domains = [ - 'x.com', # Block X (formerly Twitter) - "locked the f in" - 'twitter.com', # Block Twitter (redirects to x.com anyway) - 'facebook.com', # Lock the F in Facebook too - '*.meta.com', # Block all Meta properties - '*.adult-site.com', # Block all subdomains of adult sites + 'x.com', # Block X (formerly Twitter) - "locked the f in" + 'twitter.com', # Block Twitter (redirects to x.com anyway) + 'facebook.com', # Lock the F in Facebook too + '*.meta.com', # Block all Meta properties + '*.adult-site.com', # Block all subdomains of adult sites 'https://explicit-content.org', # Block specific protocol/domain - 'gambling-site.net', # Block gambling sites + 'gambling-site.net', # Block gambling sites ] browser_session = BrowserSession( @@ -43,7 +43,7 @@ agent = Agent( async def main(): print('Demo: Blocked Domains Feature - "Lock the F in" Edition') - print('We\'re literally locking the F in Facebook and X!') + print("We're literally locking the F in Facebook and X!") print(f'Prohibited domains: {prohibited_domains}') print('The agent will try to visit various sites, but blocked domains will be prevented.') print() diff --git a/tests/agent_tasks/sandeep_test.py b/tests/agent_tasks/sandeep_test.py new file mode 100644 index 000000000..767da5c74 --- /dev/null +++ b/tests/agent_tasks/sandeep_test.py @@ -0,0 +1,114 @@ +""" +Simple test runner for agent tasks using OCI Raw API model. +Just runs the tasks from YAML files without complex judging. +""" + +import asyncio +import glob +import logging +import os + +import aiofiles +import yaml + +from browser_use import Agent +from browser_use.logging_config import setup_logging +from examples.models.oci_models import meta_llm + +# Enable debug logging +setup_logging(log_level=logging.INFO) + + +async def run_task_from_yaml(yaml_file): + """Load and run a single task from a YAML file""" + print(f'\nšŸš€ Running task: {os.path.basename(yaml_file)}') + print('=' * 50) + + # Load task from YAML + async with aiofiles.open(yaml_file, 'r') as f: + content = await f.read() + task_data = yaml.safe_load(content) + + task_name = task_data.get('name', 'Unnamed Task') + task_description = task_data['task'] + max_steps = task_data.get('max_steps', 15) + + print(f'Task Name: {task_name}') + print(f'Description: {task_description}') + print(f'Max Steps: {max_steps}') + print('-' * 50) + + # Create OCI model + model = meta_llm + + # Create and run agent + agent = Agent(task=task_description, llm=model, use_vision=True) + + try: + history = await agent.run(max_steps=max_steps) + + print('\nāœ… Task completed!') + print(f'Steps taken: {len(history.history) if hasattr(history, "history") else "Unknown"}') + + final_result = history.final_result() + if final_result: + print(f'Final result: {final_result}') + else: + print('No final result returned') + + return True + + except Exception as e: + print(f'\nāŒ Task failed with error: {str(e)}') + return False + + +async def run_all_tasks(): + """Find and run all YAML tasks in the current directory""" + current_dir = os.path.dirname(__file__) + yaml_files = glob.glob(os.path.join(current_dir, '*.yaml')) + + if not yaml_files: + print('No YAML task files found in the current directory!') + return + + print(f'Found {len(yaml_files)} task files:') + for yaml_file in yaml_files: + print(f' - {os.path.basename(yaml_file)}') + + successful = 0 + total = len(yaml_files) + + for yaml_file in yaml_files: + try: + success = await run_task_from_yaml(yaml_file) + if success: + successful += 1 + except KeyboardInterrupt: + print('\nšŸ›‘ Interrupted by user') + break + except Exception as e: + print(f'āŒ Failed to run {os.path.basename(yaml_file)}: {str(e)}') + + print(f'\nšŸ Summary: {successful}/{total} tasks completed successfully') + + +if __name__ == '__main__': + import sys + + if len(sys.argv) > 1: + # Run specific task file + yaml_file = sys.argv[1] + if not yaml_file.endswith('.yaml'): + yaml_file += '.yaml' + + if not os.path.isabs(yaml_file): + yaml_file = os.path.join(os.path.dirname(__file__), yaml_file) + + if os.path.exists(yaml_file): + asyncio.run(run_task_from_yaml(yaml_file)) + else: + print(f'Task file not found: {yaml_file}') + else: + # Run all tasks + asyncio.run(run_all_tasks())