Alibaba-NLP/Tongyi-DeepResearch
63
1import re2from typing import Dict, List, Optional, Union3import json54from qwen_agent.tools.base import BaseToolWithFileAccess, register_tool5from qwen_agent.utils.utils import extract_code6from sandbox_fusion import run_code, RunCodeRequest, RunStatus7from requests.exceptions import Timeout8import os9import random10import time11from concurrent.futures import ThreadPoolExecutor, as_completed12 13SANDBOX_URL=os.getenv('SANDBOX_URL','')14SANDBOX_FUSION_ENDPOINTS = [15 SANDBOX_URL16]17 18# Fallback to single endpoint if environment variable exists19if 'SANDBOX_FUSION_ENDPOINT' in os.environ:20 SANDBOX_FUSION_ENDPOINTS = os.environ['SANDBOX_FUSION_ENDPOINT'].split(',')21 22 23@register_tool('PythonInterpreter', allow_overwrite=True)24class PythonInterpreter(BaseToolWithFileAccess):25 name = "PythonInterpreter"26 description = 'Execute Python code in a sandboxed environment. Use this to run Python code and get the execution results.\n**Make sure to use print() for any output you want to see in the results.**\nFor code parameters, use placeholders first, and then put the code within <code></code> XML tags, such as:\n<tool_call>\n{"purpose": <detailed-purpose-of-this-tool-call>, "name": <tool-name>, "arguments": {"code": ""}}\n<code>\nHere is the code.\n</code>\n</tool_call>\n'27 28 parameters = {29 "type": "object",30 "properties": {31 "code": {32 "type": "string",33 "description": "The Python code to execute. Must be provided within <code></code> XML tags. Remember to use print() statements for any output you want to see.",34 }35 },36 "required": ["code"],37 }38 39 def __init__(self, cfg: Optional[Dict] = None):40 super().__init__(cfg)41 # self.summary_mapping = SummaryMapping()42 43 @property44 def args_format(self) -> str:45 fmt = self.cfg.get('args_format')46 if fmt is None:47 if has_chinese_chars([self.name_for_human, self.name, self.description, self.parameters]):48 fmt = 'The input for this tool should be a Markdown code block.'49 50 else:51 fmt = 'Enclose the code within triple backticks (`) at the beginning and end of the code.'52 return fmt53 54 def observation(self, tool: dict, tool_dict: dict, tool_results, empty_mode: bool=False, readpage: bool=False, max_observation_length: int=None, tokenizer=None):55 print('test')56 assert isinstance(tool_results, str), f"result of python code should be str, instead of {type(tool_results)}. {tool_results}"57 return tool_results58 59 @property60 def function(self) -> dict: 61 return {62 'name': self.name,63 'description': self.description,64 'parameters': self.parameters,65 }66 67 def call(self, params, files= None, timeout = 50, **kwargs) -> str:68 try:69 # super().call(params=params, files=files) # copy remote files to work_dir70 try:71 code=params.split('<code>')[1].split('</code')[0]72 # print(params)73 # if type(params) is str:74 # params = json5.loads(params)75 # code = params.get('code', '')76 # if not code:77 # code = params.get('raw', '')78 # triple_match = re.search(r'```[^\n]*\n(.+?)```', code, re.DOTALL)79 # if triple_match:80 # code = triple_match.group(1)81 except Exception:82 return '[Python Interpreter Error]: format error.'83 84 if not code.strip():85 return '[Python Interpreter Error]: Empty code.'86 87 # Retry mechanism with randomly sampled endpoints88 # code=params89 last_error = None90 for attempt in range(8):91 try:92 # Randomly sample an endpoint for each attempt93 endpoint = random.choice(SANDBOX_FUSION_ENDPOINTS)94 print(f"Attempt {attempt + 1}/5 using endpoint: {endpoint}")95 96 code_result = run_code(RunCodeRequest(code=code, language='python', run_timeout=timeout), max_attempts=1, client_timeout=timeout, endpoint=endpoint)97 print("[Python] Code Result", code_result)98 result = []99 if code_result.run_result.stdout:100 result.append(f"stdout:\n{code_result.run_result.stdout}")101 if code_result.run_result.stderr:102 result.append(f"stderr:\n{code_result.run_result.stderr}")103 if code_result.run_result.execution_time >= timeout-1:104 result.append(f"[PythonInterpreter Error] TimeoutError: Execution timed out.")105 result = '\n'.join(result)106 print('SUCCESS RUNNING TOOL')107 return result if result.strip() else 'Finished execution.'108 109 except Timeout as e:110 last_error = f'[Python Interpreter Error] TimeoutError: Execution timed out on endpoint {endpoint}.'111 print(f"Timeout on attempt {attempt + 1}: {last_error}")112 if attempt == 4: # Last attempt113 return last_error114 continue115 116 except Exception as e:117 last_error = f'[Python Interpreter Error]: {str(e)} on endpoint {endpoint}'118 print(f"Error on attempt {attempt + 1}: {last_error}")119 if attempt == 4: # Last attempt120 return last_error121 continue122 123 return last_error if last_error else '[Python Interpreter Error]: All attempts failed.'124 125 except Exception as e:126 return f"[Python Interpreter Error]: {str(e)}"127 128 def call_specific_endpoint(self, params: Union[str, dict], endpoint: str, timeout: Optional[int] = 30, **kwargs) -> tuple:129 """Test a specific endpoint directly"""130 try:131 if type(params) is str:132 params = json5.loads(params)133 code = params.get('code', '')134 if not code:135 code = params.get('raw', '')136 triple_match = re.search(r'```[^\n]*\n(.+?)```', code, re.DOTALL)137 if triple_match:138 code = triple_match.group(1)139 except Exception:140 code = extract_code(params)141 142 if not code.strip():143 return False, '[Python Interpreter Error]: Empty code.'144 145 try:146 start_time = time.time()147 code_result = run_code(RunCodeRequest(code=code, language='python', run_timeout=timeout), 148 max_attempts=1, client_timeout=timeout, endpoint=endpoint)149 end_time = time.time()150 151 result = []152 if code_result.run_result.stdout:153 result.append(f"stdout:\n{code_result.run_result.stdout}")154 if code_result.run_result.stderr:155 result.append(f"stderr:\n{code_result.run_result.stderr}")156 157 result = '\n'.join(result)158 execution_time = end_time - start_time159 return True, result if result.strip() else 'Finished execution.', execution_time160 161 except Timeout as e:162 return False, f'[Python Interpreter Error] TimeoutError: Execution timed out.', None163 except Exception as e:164 return False, f'[Python Interpreter Error]: {str(e)}', None165 166 167 168def test_single_endpoint(endpoint: str, test_cases: List[dict], timeout: int = 30) -> dict:169 """Test a single endpoint with multiple test cases"""170 executor = PythonInterpreter()171 results = {172 'endpoint': endpoint,173 'status': 'unknown',174 'passed_tests': 0,175 'total_tests': len(test_cases),176 'test_results': [],177 'avg_execution_time': 0,178 'errors': []179 }180 181 execution_times = []182 183 print(f"\n🧪 Testing endpoint: {endpoint}")184 185 for i, test_case in enumerate(test_cases):186 test_name = test_case['name']187 test_code = test_case['code']188 expected_output = test_case.get('expected_output')189 190 print(f" ├─ Running test {i+1}/{len(test_cases)}: {test_name}")191 192 try:193 success, result, exec_time = executor.call_specific_endpoint(194 {"code": test_code}, endpoint, timeout195 )196 197 test_result = {198 'name': test_name,199 'success': success,200 'result': result,201 'execution_time': exec_time,202 'expected_match': False203 }204 205 if success and exec_time is not None:206 execution_times.append(exec_time)207 208 # Check if output matches expected (if provided)209 if expected_output:210 # Clean up output for comparison211 actual_output = result.replace('stdout:\n', '').strip()212 if expected_output.strip() in actual_output:213 test_result['expected_match'] = True214 results['passed_tests'] += 1215 print(f" │ ✅ PASSED ({exec_time:.2f}s)")216 else:217 print(f" │ ❌ OUTPUT MISMATCH ({exec_time:.2f}s)")218 print(f" │ Expected: {expected_output.strip()}")219 print(f" │ Got: {actual_output}")220 else:221 # No expected output specified, just check if it ran successfully222 test_result['expected_match'] = True223 results['passed_tests'] += 1224 print(f" │ ✅ PASSED ({exec_time:.2f}s)")225 else:226 print(f" │ ❌ FAILED: {result}")227 results['errors'].append(f"{test_name}: {result}")228 229 results['test_results'].append(test_result)230 231 except Exception as e:232 print(f" │ 💥 EXCEPTION: {str(e)}")233 results['errors'].append(f"{test_name}: Exception - {str(e)}")234 results['test_results'].append({235 'name': test_name,236 'success': False,237 'result': f"Exception: {str(e)}",238 'execution_time': None,239 'expected_match': False240 })241 242 # Calculate statistics243 if execution_times:244 results['avg_execution_time'] = sum(execution_times) / len(execution_times)245 246 # Determine overall status247 if results['passed_tests'] == results['total_tests']:248 results['status'] = 'healthy'249 print(f" └─ ✅ ALL TESTS PASSED ({results['passed_tests']}/{results['total_tests']})")250 elif results['passed_tests'] > 0:251 results['status'] = 'partial'252 print(f" └─ ⚠️ PARTIAL SUCCESS ({results['passed_tests']}/{results['total_tests']})")253 else:254 results['status'] = 'failed'255 print(f" └─ ❌ ALL TESTS FAILED ({results['passed_tests']}/{results['total_tests']})")256 257 return results258 259 260def test_all_endpoints_comprehensive():261 """Comprehensive test suite for all sandbox fusion endpoints"""262 263 # Define comprehensive test cases264 test_cases = [265 {266 'name': 'Basic Math',267 'code': 'print(2 + 2)',268 'expected_output': '4'269 },270 {271 'name': 'String Operations',272 'code': 'print("Hello, " + "World!")',273 'expected_output': 'Hello, World!'274 },275 {276 'name': 'List Operations',277 'code': '''278numbers = [1, 2, 3, 4, 5]279print(f"Sum: {sum(numbers)}")280print(f"Length: {len(numbers)}")281''',282 'expected_output': 'Sum: 15'283 },284 {285 'name': 'Dictionary Operations',286 'code': '''287data = {"name": "Test", "value": 42}288print(f"Name: {data['name']}, Value: {data['value']}")289''',290 'expected_output': 'Name: Test, Value: 42'291 },292 {293 'name': 'Loop and Conditionals',294 'code': '''295result = []296for i in range(5):297 if i % 2 == 0:298 result.append(i)299print(f"Even numbers: {result}")300''',301 'expected_output': 'Even numbers: [0, 2, 4]'302 },303 {304 'name': 'Function Definition',305 'code': '''306def factorial(n):307 if n <= 1:308 return 1309 return n * factorial(n-1)310 311print(f"5! = {factorial(5)}")312''',313 'expected_output': '5! = 120'314 },315 {316 'name': 'Exception Handling',317 'code': '''318try:319 result = 10 / 2320 print(f"Division result: {result}")321except ZeroDivisionError:322 print("Cannot divide by zero")323''',324 'expected_output': 'Division result: 5.0'325 },326 {327 'name': 'Import Standard Library',328 'code': '''329import math330print(f"Pi: {math.pi:.2f}")331print(f"Square root of 16: {math.sqrt(16)}")332''',333 'expected_output': 'Pi: 3.14'334 },335 {336 'name': 'Complex Calculation',337 'code': '''338import math339 340# Calculate area of a circle341radius = 5342area = math.pi * radius ** 2343print(f"Area of circle with radius {radius}: {area:.2f}")344 345# Fibonacci sequence346def fibonacci(n):347 if n <= 1:348 return n349 return fibonacci(n-1) + fibonacci(n-2)350 351fib_10 = fibonacci(10)352print(f"10th Fibonacci number: {fib_10}")353''',354 'expected_output': 'Area of circle with radius 5: 78.54'355 },356 {357 'name': 'Error Handling Test',358 'code': '''359try:360 undefined_variable361except NameError as e:362 print("Caught NameError as expected")363 print("Test passed")364''',365 'expected_output': 'Test passed'366 }367 ]368 369 print("🚀 Starting comprehensive endpoint testing...")370 print(f"📊 Testing {len(SANDBOX_FUSION_ENDPOINTS)} endpoints with {len(test_cases)} test cases each")371 print("=" * 80)372 373 # Test all endpoints concurrently374 all_results = []375 376 # Use ThreadPoolExecutor for concurrent testing377 with ThreadPoolExecutor(max_workers=min(len(SANDBOX_FUSION_ENDPOINTS), 8)) as executor:378 future_to_endpoint = {379 executor.submit(test_single_endpoint, endpoint, test_cases): endpoint 380 for endpoint in SANDBOX_FUSION_ENDPOINTS381 }382 383 for future in as_completed(future_to_endpoint):384 endpoint = future_to_endpoint[future]385 try:386 result = future.result()387 all_results.append(result)388 except Exception as exc:389 print(f'❌ Endpoint {endpoint} generated an exception: {exc}')390 all_results.append({391 'endpoint': endpoint,392 'status': 'failed',393 'passed_tests': 0,394 'total_tests': len(test_cases),395 'test_results': [],396 'avg_execution_time': 0,397 'errors': [f'Exception during testing: {exc}']398 })399 400 # Print comprehensive summary401 print("\n" + "=" * 80)402 print("📈 COMPREHENSIVE TEST RESULTS SUMMARY")403 print("=" * 80)404 405 healthy_endpoints = [r for r in all_results if r['status'] == 'healthy']406 partial_endpoints = [r for r in all_results if r['status'] == 'partial']407 failed_endpoints = [r for r in all_results if r['status'] == 'failed']408 409 print(f"✅ Healthy endpoints: {len(healthy_endpoints)}/{len(SANDBOX_FUSION_ENDPOINTS)}")410 print(f"⚠️ Partial endpoints: {len(partial_endpoints)}/{len(SANDBOX_FUSION_ENDPOINTS)}")411 print(f"❌ Failed endpoints: {len(failed_endpoints)}/{len(SANDBOX_FUSION_ENDPOINTS)}")412 413 if healthy_endpoints:414 print(f"\n🎉 HEALTHY ENDPOINTS ({len(healthy_endpoints)}):")415 for result in healthy_endpoints:416 print(f" ✅ {result['endpoint']} (avg: {result['avg_execution_time']:.2f}s)")417 418 if partial_endpoints:419 print(f"\n⚠️ PARTIAL ENDPOINTS ({len(partial_endpoints)}):")420 for result in partial_endpoints:421 print(f" ⚠️ {result['endpoint']} ({result['passed_tests']}/{result['total_tests']} tests passed)")422 423 if failed_endpoints:424 print(f"\n❌ FAILED ENDPOINTS ({len(failed_endpoints)}):")425 for result in failed_endpoints:426 print(f" ❌ {result['endpoint']} (0/{result['total_tests']} tests passed)")427 if result['errors']:428 for error in result['errors'][:3]: # Show first 3 errors429 print(f" └─ {error}")430 431 # Performance statistics432 healthy_times = [r['avg_execution_time'] for r in healthy_endpoints if r['avg_execution_time'] > 0]433 if healthy_times:434 print(f"\n⚡ PERFORMANCE STATISTICS:")435 print(f" Fastest endpoint: {min(healthy_times):.2f}s")436 print(f" Slowest endpoint: {max(healthy_times):.2f}s")437 print(f" Average response time: {sum(healthy_times)/len(healthy_times):.2f}s")438 439 # Overall health score440 total_tests = len(SANDBOX_FUSION_ENDPOINTS) * len(test_cases)441 passed_tests = sum(r['passed_tests'] for r in all_results)442 health_score = (passed_tests / total_tests) * 100443 444 print(f"\n🏥 OVERALL SYSTEM HEALTH: {health_score:.1f}%")445 print(f" ({passed_tests}/{total_tests} total tests passed)")446 447 if health_score >= 90:448 print(" 🟢 Excellent - System is performing well")449 elif health_score >= 70:450 print(" 🟡 Good - Some endpoints may need attention")451 elif health_score >= 50:452 print(" 🟠 Fair - Multiple endpoints having issues")453 else:454 print(" 🔴 Poor - System requires immediate attention")455 456 return all_results457 458 459def _test():460 """Original test function"""461 # test_code = """import sympy as sp\nX=sp.symbols('X')\npoly_factor = (X**2 - (sp.sqrt(34)+sp.sqrt(14))*X + 2*sp.sqrt(119))*(X**2 - 2*(sp.sqrt(11)+sp.sqrt(6))*X + 4*sp.sqrt(66))\npoly_original = X**4 - sp.sqrt(34)*X**3 - sp.sqrt(14)*X**3 - 2*sp.sqrt(11)*X**3 - 2*sp.sqrt(6)*X**3 + 2*sp.sqrt(374)*X**2 + 2*sp.sqrt(154)*X**2 + 2*sp.sqrt(119)*X**2 + 4*sp.sqrt(66)*X**2 + 4*sp.sqrt(51)*X**2 + 4*sp.sqrt(21)*X**2 - 4*sp.sqrt(1309)*X - 4*sp.sqrt(714)*X - 8*sp.sqrt(561)*X - 8*sp.sqrt(231)*X + 8*sp.sqrt(7854)\nprint('expanded factor matches?', sp.simplify(poly_factor - poly_original) == 0)\nprint('Difference simplified:', sp.simplify(poly_factor - poly_original))\n"""462 # params = {"code": test_code}463 # params = {"code": "x = 1.25\nprint(x*x)"}464 params = "print(1234)"465 # ''466 executor = PythonInterpreter()467 out = executor.call(params)468 from pprint import pprint469 pprint(out)470 471 472if __name__ == '__main__':473 # Run comprehensive endpoint testing474 test_all_endpoints_comprehensive()475 476 print("\n" + "=" * 80)477 print("🔧 Running original test...")478 print("=" * 80)479 480 # Also run the original test481 _test()