Update vis_python_exe.py

vis_python_exe.py

@@ -1,439 +1,783 @@
+import sys
 import os
-import io
-import regex
-import pickle
-import traceback
-import copy
-import datetime
-import dateutil.relativedelta
-import multiprocess
-from multiprocess import Pool
-from typing import Any, Dict, Optional, Tuple, List, Union
-from pebble import ProcessPool
-from tqdm import tqdm
-from concurrent.futures import TimeoutError
-from functools import partial
-from timeout_decorator import timeout
-from contextlib import redirect_stdout
+import re
+import json
 import base64
 from io import BytesIO
 from PIL import Image
-import pdb
+import argparse
+from inference_engine.safe_persis_shared_vis_python_exe import PythonExecutor, ImageRuntime
+from openai import OpenAI
+import anthropic
 
-def encode_image(image_path):
-    with open(image_path, "rb") as image_file:
-        return base64.b64encode(image_file.read()).decode('utf-8')
-
-def base64_to_image(
-    base64_str: str, 
-    remove_prefix: bool = True, 
-    convert_mode: Optional[str] = "RGB"
-) -> Union[Image.Image, None]:
+def encode_image(image):
     """
-    将Base64编码的图片字符串转换为PIL Image对象
+    Convert a PIL.Image object or image file path to base64-encoded string, and get resolution info.
     
     Args:
-        base64_str: Base64编码的图片字符串（可带data:前缀）
-        remove_prefix: 是否自动去除"data:image/..."前缀（默认True）
-        convert_mode: 转换为指定模式（如"RGB"/"RGBA"，None表示不转换）
+        image: Can be a PIL.Image object or image file path.
+    Returns:
+        dict with keys:
+        - 'base64': base64-encoded string
+        - 'width': width in pixels
+        - 'height': height in pixels
+        - 'resolution': string "widthxheight"
+    """
+    img_obj = None
+    
+    if isinstance(image, str):
+        # Handle file path
+        img_obj = Image.open(image)
+        with open(image, "rb") as image_file:
+            base64_str = base64.b64encode(image_file.read()).decode('utf-8')
+    else:
+        # Handle PIL.Image object
+        img_obj = image
+        buffered = BytesIO()
+        image.save(buffered, format='PNG')
+        base64_str = base64.b64encode(buffered.getvalue()).decode('utf-8')
+    
+    width, height = img_obj.size
+    
+    return {
+        'base64': base64_str,
+        'width': width,
+        'height': height
+    }
+
+def encode_image_with_resize(image):
+    """
+    Convert a PIL.Image object or image file path to base64-encoded string, get resolution info.
+    If resolution > 1024x1024, resize to half.
     
+    Args:
+        image: Can be a PIL.Image object or image file path
     Returns:
-        PIL.Image.Image 对象，解码失败时返回None
-        
-    Examples:
-        >>> img = base64_to_image("data:image/png;base64,iVBORw0KGg...")
-        >>> img = base64_to_image("iVBORw0KGg...", remove_prefix=False)
+        dict with keys:
+        - 'base64': base64-encoded string
+        - 'width': width in pixels
+        - 'height': height in pixels
+        - 'resolution': string "widthxheight"
     """
-    try:
-        # 1. 处理Base64前缀
-        if remove_prefix and "," in base64_str:
-            base64_str = base64_str.split(",")[1]
+    img_obj = None
+    
+    if isinstance(image, str):
+        img_obj = Image.open(image)
+    else:
+        img_obj = image
+
+    # Resize if larger than 1024x1024
+    width, height = img_obj.size
+    if width > 1024 or height > 1024:
+        new_size = (width // 2, height // 2)
+        img_obj = img_obj.resize(new_size, Image.LANCZOS)
+        width, height = img_obj.size
+
+    buffered = BytesIO()
+    img_obj.save(buffered, format='PNG')
+    base64_str = base64.b64encode(buffered.getvalue()).decode('utf-8')
+
+    return {
+        'base64': base64_str,
+        'width': width,
+        'height': height,
+        'resolution': f"{width}x{height}"
+    }
+
+def check(evaluator, pred_ans, real_ans):
+    if len(pred_ans) == 0:
+        return []
+    correctness = evaluator.score(pred_ans, real_ans)
+    return correctness
+
+def execute_codes(codes, messages, executor: PythonExecutor):
+    no_code_idx = []
+    codes_use = []
+    for i, code in enumerate(codes):
+        if code == "":
+            no_code_idx.append(i)
+        else:
+            codes_use.append(code)
+    batch_results = executor.batch_apply(codes_use, messages)
+    return batch_results, no_code_idx
 
-        # 2. 解码Base64
-        image_data = base64.b64decode(base64_str)
-        
-        # 3. 转换为PIL Image
-        image = Image.open(BytesIO(image_data))
-        
-        # 4. 可选模式转换
-        if convert_mode:
-            image = image.convert(convert_mode)
+def process_prompt_init(question, image_path_list, prompt_template, prompt_type, api_name):
+    with open(prompt_template, "r") as fin:
+        sys = json.load(fin)
+    prompt_prefix = sys[prompt_type]
+
+    image_path = image_path_list[0]
+
+    if "<IMAGE_PLACE_HOLDER_0>" in question:
+        if "no_tool" in prompt_type:
+
+            if "claude" in api_name:
+                img_result = encode_image_with_resize(image_path)
+            else:
+                img_result = encode_image(image_path)
+            image_base64 = img_result['base64']
+            question_with_options = question
+            question = prompt_prefix.format(query=question_with_options)
+
+            parts = question.split("<IMAGE_PLACE_HOLDER_0>")
+            content = []
             
-        return image
-    
-    except (base64.binascii.Error, OSError, Exception) as e:
-        print(f"Base64解码失败: {str(e)}")
-        return None
+            # Add text before image (if any)
+            if parts[0].strip():
+                content.append({"type": "text", "text": parts[0].strip()})
+            # Add image
+            content.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}})
+            
+            # Add text after image (if any)
+            if len(parts) > 1 and parts[1].strip():
+                content.append({"type": "text", "text": parts[1].strip()})
+
+            messages = [
+                {
+                    "role": "user",
+                    "content": content
+                }
+            ]
+
+            return messages
+
+        else:
+            if "claude" in api_name:
+                img_result = encode_image_with_resize(image_path)
+            else:
+                img_result = encode_image(image_path)
+            image_base64 = img_result['base64']
+            width = img_result['width']
+            height = img_result['height']
+            question_with_options = question
+            question = prompt_prefix.format(query=question_with_options, width=str(width), height=str(height))
+
+            # Split question into parts
+            parts = question.split("<IMAGE_PLACE_HOLDER_0>")
+            # Build message with image_clue tags
+            content = []
+            
+            # Add text before image (if any)
+            if parts[0].strip():
+                content.append({"type": "text", "text": parts[0].strip()})
+            
+            # Add image with tags
+            content.append({"type": "text", "text": "<image_clue_0>"})
+            content.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}})
+            content.append({"type": "text", "text": "</image_clue_0>\n\n"})
+            
+            # Add text after image (if any)
+            if len(parts) > 1 and parts[1].strip():
+                content.append({"type": "text", "text": parts[1].strip()})
 
+            messages = [
+                {
+                    "role": "user",
+                    "content": content
+                }
+            ]
 
-class GenericRuntime:
-    GLOBAL_DICT = {}
-    LOCAL_DICT = None
-    HEADERS = []
+            return messages
 
-    def __init__(self):
-        self._global_vars = copy.copy(self.GLOBAL_DICT)
-        self._local_vars = copy.copy(self.LOCAL_DICT) if self.LOCAL_DICT else None
-        self._captured_figures = []
+    else:
+        if "no_tool" in prompt_type:
+
+            if "claude" in api_name:
+                img_result = encode_image_with_resize(image_path)
+            else:
+                img_result = encode_image(image_path)
+            image_base64 = img_result['base64']
+            question_with_options = question
 
-        for c in self.HEADERS:
-            self.exec_code(c)
+            messages = [
+                {
+                    "role": "user",
+                    "content": [{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}}] + [{"type": "text", "text": prompt_prefix.format(query=question_with_options)}]
+                }
+            ]
 
-    def exec_code(self, code_piece: str) -> None:
-        if regex.search(r"(\s|^)?input\(", code_piece) or regex.search(
-            r"(\s|^)?os.system\(", code_piece
-        ):
-            raise RuntimeError("Forbidden function calls detected")
+            return messages
 
+        else:
+            if "claude" in api_name:
+                img_result = encode_image_with_resize(image_path)
+            else:
+                img_result = encode_image(image_path)
+            image_base64 = img_result['base64']
+            width = img_result['width']
+            height = img_result['height']
+            question_with_options = question
+
+            messages = [
+                {
+                    "role": "user",
+                    "content": [{"type": "text", "text": "<image_clue_0>"}] + [{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}}] + [{"type": "text", "text": "</image_clue_0>\n\n"}] + [{"type": "text", "text": prompt_prefix.format(query=question_with_options, width=str(width), height=str(height))}]
+                }
+            ]
+
+            return messages
+
+def process_prompt_init_multi_images(question, image_path_list, prompt_template, prompt_type, api_name):
+    with open(prompt_template, "r") as fin:
+        sys = json.load(fin)
+    prompt_prefix = sys[prompt_type]
+    
+    # Prepare image data
+    image_data = []
+    image_information = ""
+    
+    for i, image_path in enumerate(image_path_list):
+        if "claude" in api_name:
+            img_result = encode_image_with_resize(image_path)
+        else:
+            img_result = encode_image(image_path)
+        image_base64 = img_result['base64']
+        width = img_result['width']
+        height = img_result['height']
         
+        image_data.append({
+            "index": i,
+            "base64": image_base64,
+            "width": width,
+            "height": height,
+            "placeholder": f"<IMAGE_PLACE_HOLDER_{i}>"
+        })
         
-        # 检测并修改plt.show()调用
-        if "plt.show()" in code_piece:
-            modified_code = code_piece.replace("plt.show()", """
-# 捕获当前图像
-buf = io.BytesIO()
-plt.savefig(buf, format='png')
-buf.seek(0)
-_captured_image = base64.b64encode(buf.read()).decode('utf-8')
-_captured_figures.append(_captured_image)
-plt.close()
-""")
-            # 确保_captured_figures变量存在
-            if "_captured_figures" not in self._global_vars:
-                self._global_vars["_captured_figures"] = []
+        image_information += f"width of image_clue_{i}: {width}, height of image_clue_{i}: {height}\n"
+    
+    # Format question
+    formatted_question = prompt_prefix.format(query=question, image_information=image_information)
+    
+    # Check if placeholder exists
+    has_placeholders = any(f"<IMAGE_PLACE_HOLDER_{i}>" in formatted_question for i in range(len(image_path_list)))
+    
+    if has_placeholders:
+        # Insert images at placeholder positions
+        if "no_tool" in prompt_type:
+            content = []
+            remaining_text = formatted_question
+            
+            for img_data in image_data:
+                placeholder = img_data["placeholder"]
+                if placeholder in remaining_text:
+                    parts = remaining_text.split(placeholder, 1)
+                    
+                    if parts[0]:
+                        content.append({"type": "text", "text": parts[0]})
+                    
+                    content.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_data['base64']}"}})
+                    
+                    remaining_text = parts[1]
+            
+            if remaining_text:
+                content.append({"type": "text", "text": remaining_text})
             
-            exec(modified_code, self._global_vars)
+            messages = [{"role": "user", "content": content}]
+            return messages
         else:
-            print("###################################### I am excuting code. ##############################################")
-            exec(code_piece, self._global_vars)
+            content = []
+            remaining_text = formatted_question
+            
+            for img_data in image_data:
+                placeholder = img_data["placeholder"]
+                if placeholder in remaining_text:
+                    parts = remaining_text.split(placeholder, 1)
+                    
+                    if parts[0]:
+                        content.append({"type": "text", "text": parts[0]})
+                    
+                    i = img_data["index"]
+                    content.append({"type": "text", "text": f"<image_clue_{i}>"})
+                    content.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_data['base64']}"}})
+                    content.append({"type": "text", "text": f"</image_clue_{i}>\n\n"})
+                    
+                    remaining_text = parts[1]
+            
+            if remaining_text:
+                content.append({"type": "text", "text": remaining_text})
+            
+            messages = [{"role": "user", "content": content}]
+            return messages
+    else:
+        # Handle as usual if no placeholder
+        if "no_tool" in prompt_type:
+            content = []
+            
+            for i, img_data in enumerate(image_data):
+                content.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_data['base64']}"}})
+            
+            content.append({"type": "text", "text": formatted_question})
+            
+            messages = [{"role": "user", "content": content}]
+            return messages
+        else:
+            content = []
+            
+            for i, img_data in enumerate(image_data):
+                content.append({"type": "text", "text": f"<image_clue_{i}>"})
+                content.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_data['base64']}"}})
+                content.append({"type": "text", "text": f"</image_clue_{i}>\n\n"})
+            
+            content.append({"type": "text", "text": formatted_question})
+            
+            messages = [{"role": "user", "content": content}]
+            return messages
+
+
+def update_messages_with_execute_content(image_nums_in_input, messages, images_result, text_result, error_result, image_clue_idx):
+    if error_result is None:
+        new_messages = []
+        image_content = []
+        for message_item in messages[:-1]:
+            new_messages.append(message_item)
+
+        assistant_message_item = messages[-1]['content']
+        interpreter_message_text_prefix = [{"type": "text", "text": f"<interpreter>\nText Result:\n{text_result}\nImage Result:\n"}]
+        if images_result is not None:
+            print(f"#### image_clue_index: {image_clue_idx},Image_nums_in_input: {image_nums_in_input}, len of images_result: {len(images_result)}")
+            # for image_base64_item in images_result[image_clue_idx-image_nums_in_input:]:
+            for image_base64_item in images_result:
+                interpreter_message_images = [{"type": "text", "text": f"<image_clue_{image_clue_idx}>"}] + [{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64_item}"}}] + [{"type": "text", "text": f"</image_clue_{image_clue_idx}>"}]
+                image_content += interpreter_message_images
+                image_clue_idx += 1
+        else:
+            image_content = [{"type": "text", "text": "None"}]
+        interpreter_message_text_profill = [{"type": "text", "text": "</interpreter>\n"}]
 
-    def eval_code(self, expr: str) -> Any:
-        return eval(expr, self._global_vars)
+        interpreter_message_item = interpreter_message_text_prefix + image_content + interpreter_message_text_profill
+        new_messages.append({"role": "assistant", "content": assistant_message_item})
+        new_messages.append({"role": "user", "content": interpreter_message_item})
+    else:
+        new_messages = []
+        for message_item in messages[:-1]:
+            new_messages.append(message_item)
+    
+        assistant_message_item = messages[-1]['content']
+        interpreter_message_text_prefix = [{"type": "text", "text": f"<interpreter>{error_result}"}]
+        interpreter_message_text_profill = [{"type": "text", "text": "</interpreter>\n"}]
+    
+        interpreter_message_item = interpreter_message_text_prefix + interpreter_message_text_profill
+        new_messages.append({"role": "assistant", "content": assistant_message_item})
+        new_messages.append({"role": "user", "content": interpreter_message_item})
 
-    def inject(self, var_dict: Dict[str, Any]) -> None:
-        for k, v in var_dict.items():
-            self._global_vars[k] = v
+    return new_messages, image_clue_idx
 
-    @property
-    def answer(self):
-        return self._global_vars.get("answer", None)
-    
-    @property
-    def captured_figures(self):
-        return self._global_vars.get("_captured_figures", [])
+def update_messages_with_code(messages, generated_content):
+    message_item = {
+        "role": "assistant",
+        "content": [{"type": "text", "text": f"{generated_content}</code>\n"}]
+    }
+
+    messages.append(message_item)
+    return messages
 
+def update_messages_with_text(messages, generated_content):
+    message_item = {
+        "role": "assistant",
+        "content": [{"type": "text", "text": f"{generated_content}"}]
+    }
 
-class ImageRuntime(GenericRuntime):
-    """支持图像处理的运行时环境"""
-    GLOBAL_DICT = {}  # 不预加载模块，避免序列化问题
-    LOCAL_DICT = None
+    messages.append(message_item)
+    return messages
+
+def call_chatgpt_api(args, messages, client, max_tokens=10000, stop=None, temperature=0.6):
+    """Call ChatGPT API with the given messages"""
+    try:
+        client_type = args.client_type
+        api_name = args.api_name
+    except:
+        client_type = args['client_type']
+        api_name = args['api_name']
     
-    HEADERS = [
-        "import matplotlib",
-        "matplotlib.use('Agg')",  # 使用非交互式后端
-        "import matplotlib.pyplot as plt",
-        "from PIL import Image",
-        "import io",
-        "import base64",
-        "import numpy as np",
-        "_captured_figures = []",  # 初始化图像捕获列表
-    ]
-
-    def __init__(self, messages):
-        super().__init__()
-
-        image_var_dict = {}
-        image_var_idx = 0
-        for message_item in messages:
-            content = message_item['content']  # {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}}
-            for item in content:
-                item_type = item['type']
-                if item_type == "image_url":
-                    item_image_url = item['image_url']['url']
-                    image = base64_to_image(item_image_url)
-                    image_var_dict[f"image_clue_{image_var_idx}"] = image
-                    image_var_idx += 1
-
-        self.inject(image_var_dict)
-        print("##################### Initialized ImageRuntime. ##########################")
-                    
+    if client_type == "openai" or client_type == "azure":
+        response = client.chat.completions.create(
+            model=api_name,
+            messages=messages,
+            max_tokens=max_tokens,
+            temperature=temperature,
+            top_p=1.0,
+            stop=stop,
+            timeout=300
+        )
+        response_text = response.choices[0].message.content
+    elif client_type == "anthropic":
+        message = client.messages.create(
+            model=api_name,
+            max_tokens=max_tokens,
+            messages=messages,
+            temperature=temperature,
+            top_p=1.0,
+            stop_sequences=stop
+        )  
+        response_text = message.content[0].text if isinstance(message.content, list) else message.content
+    elif client_type == "vllm":
+        response = client.chat.completions.create(
+            model=api_name,
+            messages=messages,
+            max_tokens=max_tokens,
+            temperature=temperature,
+            top_p=1.0,
+            stop=stop
+        )
+        response_text = response.choices[0].message.content
+    else:
+        print("Your args.client_type must be one of openai, azure, anthropic and vllm.")
+        return None, None
+    
+    # Check if stop sequence is encountered
+    stop_reason = None
+    if stop and any(s in response_text for s in stop):
+        for s in stop:
+            if s in response_text:
+                stop_reason = s
+                break
+    else:
+        if client_type in ["openai", "azure", "vllm"]:
+            stop_reason = response.choices[0].finish_reason
+        else:
+            stop_reason = "stop"
+
+    if "<code>" in response_text:
+        stop_reason = "</code>"
+    
+    return response_text, stop_reason
 
-class DateRuntime(GenericRuntime):
-    GLOBAL_DICT = {}
-    HEADERS = [
-        "import datetime",
-        "from dateutil.relativedelta import relativedelta",
-        "timedelta = relativedelta"
-    ]
-
-
-class CustomDict(dict):
-    def __iter__(self):
-        return list(super().__iter__()).__iter__()
-
-
-class ColorObjectRuntime(GenericRuntime):
-    GLOBAL_DICT = {"dict": CustomDict}
-
-
-class PythonExecutor:
-    def __init__(
-        self,
-        runtime_class=None,
-        get_answer_symbol: Optional[str] = None,
-        get_answer_expr: Optional[str] = None,
-        get_answer_from_stdout: bool = True,
-        timeout_length: int = 20,
-    ) -> None:
-        print(f"#################### When Init PythonExcutor, RunTime typel:, TimeOut Length: {timeout_length} #############################")
-        self.runtime_class = runtime_class if runtime_class else ImageRuntime
-        print(self.runtime_class)
-        self.answer_symbol = get_answer_symbol
-        self.answer_expr = get_answer_expr
-        self.get_answer_from_stdout = get_answer_from_stdout
-        self.pool = Pool(multiprocess.cpu_count())
-        self.timeout_length = timeout_length
-
-    def process_generation_to_code(self, gens: str):
-        return [g.split("\n") for g in gens]
-
-    @staticmethod
-    def execute(
-        code,
-        messages,
-        get_answer_from_stdout=True,
-        runtime_class=None,
-        answer_symbol=None,
-        answer_expr=None,
-        timeout_length=20,
-    ) -> Tuple[Union[str, Dict[str, Any]], str]:
-        # print("dome")
-        try:
-            # 在每个进程中创建新的运行时实例
-            runtime = runtime_class(messages)
+def evaluate_single_data(args, data, client, executor):
+    try:
+        prompt_template = args.prompt_template
+        prompt = args.prompt
+        exe_code = args.exe_code
+        max_tokens = args.max_tokens
+        temperature = args.temperature
+        api_name = args.api_name
+    except:
+        prompt_template = args['prompt_template']
+        prompt = args['prompt']
+        exe_code = args['exe_code']
+        max_tokens = args['max_tokens']
+        temperature = args['temperature']
+        api_name = args['api_name']
+
+    image_path_list = data['image_path_list']
+
+    if "no_tool" in prompt:
+        if len(image_path_list) == 1:
+            messages = process_prompt_init(data["question"], image_path_list, prompt_template, prompt, api_name)
+        elif len(image_path_list) >= 2:
+            messages = process_prompt_init_multi_images(data["question"], image_path_list, prompt_template, prompt, api_name)
+    else:
+        if len(image_path_list) == 1:
+            prompt = "vistool_with_img_info_v2"
+            messages = process_prompt_init(data["question"], image_path_list, prompt_template, prompt, api_name)
+        elif len(image_path_list) >= 2:
+            prompt = "vistool_with_img_info_multi_image"
+            messages = process_prompt_init_multi_images(data["question"], image_path_list, prompt_template, prompt, api_name)
+    
+    # Generate initial response
+    response_text, pred_stop_reason = call_chatgpt_api(
+        args,
+        messages, 
+        client,
+        max_tokens=max_tokens,
+        stop=["</code>"] if exe_code else None,
+        temperature=temperature
+    )
+    
+    # Handle response
+    final_response = response_text
+    code_execution_count = 0
+    image_clue_idx = len(image_path_list)
+    
+    while True:
+        # Check if code execution is needed
+        if exe_code and pred_stop_reason == "</code>":
+            # Extract code to execute
+            messages = update_messages_with_code(messages, response_text)
+            code_to_execute = response_text.split("```python")[-1].split("```")[0].strip()
             
-            if get_answer_from_stdout:
-                program_io = io.StringIO()
-                with redirect_stdout(program_io):
-                    timeout(timeout_length)(runtime.exec_code)("\n".join(code))
-                program_io.seek(0)
-                result = program_io.read()
-            elif answer_symbol:
-                timeout(timeout_length)(runtime.exec_code)("\n".join(code))
-                result = runtime._global_vars.get(answer_symbol, "")
-            elif answer_expr:
-                timeout(timeout_length)(runtime.exec_code)("\n".join(code))
-                result = timeout(timeout_length)(runtime.eval_code)(answer_expr)
+            # Execute code
+            exe_result = execute_codes([code_to_execute], messages, executor)[0][0]
+            if exe_result is None:
+                text_result = "None"
+                images_result = None
             else:
-                if len(code) > 1:
-                    timeout(timeout_length)(runtime.exec_code)("\n".join(code[:-1]))
-                    result = timeout(timeout_length)(runtime.eval_code)(code[-1])
+                output, report = exe_result
+                if report == "Done":
+                    error_result = None
+                    try:
+                        text_result = exe_result[0]['text']
+                    except:
+                        text_result = None
+                        print("text result is none.")
+                    try:
+                        images_result = exe_result[0]['images']
+                    except:
+                        images_result = None
+                        print("image result is none.")
                 else:
-                    timeout(timeout_length)(runtime.exec_code)("\n".join(code))
-                    result = ""
+                    error_result = report
+                    text_result = None
+                    images_result = None
+
+            messages, new_image_clue_idx = update_messages_with_execute_content(len(image_path_list), messages, images_result, text_result, error_result, image_clue_idx)
+            image_clue_idx = new_image_clue_idx
+            
+            code_execution_count += 1
+            
+            # Generate next response part
+            response_text, pred_stop_reason = call_chatgpt_api(
+                args,
+                messages, 
+                client,
+                max_tokens=max_tokens,
+                stop=["</code>"] if exe_code else None,
+                temperature=temperature
+            )
+
+        else:
+            final_response = response_text
+            messages = update_messages_with_text(messages, response_text)
+            break
+       
+    return messages, final_response
+
+
+def evaluate_single_data_multi_images(args, data, client, executor):
+    try:
+        prompt_template = args.prompt_template
+        prompt = args.prompt
+        exe_code = args.exe_code
+        max_tokens = args.max_tokens
+    except:
+        prompt_template = args['prompt_template']
+        prompt = args['prompt']
+        exe_code = args['exe_code']
+        max_tokens = args['max_tokens']
+
+    messages = process_prompt_init_multi_images(data["question"], data['image_path_list'], prompt_template, prompt)
+    
+    # Generate initial response
+    response_text, pred_stop_reason = call_chatgpt_api(
+        args,
+        messages, 
+        client,
+        max_tokens=max_tokens,
+        stop=["</code>"] if exe_code else None
+    )
+    
+    # Handle response
+    final_response = response_text
+    code_execution_count = 0
+    image_clue_idx = data['image_nums_in_input']
+    
+    while True:
+        # Check if code execution is needed
+        if exe_code and pred_stop_reason == "</code>":
+            # Extract code to execute
+            messages = update_messages_with_code(messages, response_text)
+            code_to_execute = response_text.split("```python")[-1].split("```")[0].strip()
             
-            # 检查是否有捕获的图像
-            captured_figures = runtime._global_vars.get("_captured_figures", [])
-            if captured_figures:
-                # 如果有文本输出和图像，将它们组合
-                if result:
-                    result = {
-                        'text': result,
-                        'images': captured_figures
-                    }
+            # Execute code
+            exe_result = execute_codes([code_to_execute], messages, executor)[0][0]
+            if exe_result is None:
+                text_result = "None"
+                images_result = None
+            else:
+                output, report = exe_result
+                if report == "Done":
+                    error_result = None
+                    try:
+                        text_result = exe_result[0]['text']
+                    except:
+                        text_result = None
+                        print("text result is none.")
+                    try:
+                        images_result = exe_result[0]['images']
+                    except:
+                        images_result = None
+                        print("image result is none.")
                 else:
-                    result = {'images': captured_figures}
+                    error_result = report
+                    text_result = None
+                    images_result = None
+
+            messages, new_image_clue_idx = update_messages_with_execute_content(data['image_nums_in_input'], messages, images_result, text_result, error_result, image_clue_idx)
+            image_clue_idx = new_image_clue_idx
             
-            report = "Done"
-        except Exception as e:
-            result = ""
-            report = f"Error: {str(e)}\n{traceback.format_exc()}"
-        
-        # 确保结果可序列化
-        try:
-            pickle.dumps(result)
-        except Exception as e:
-            result = f"Result serialization error: {str(e)}"
-            report = f"Serialization Error: {str(e)}"
+            code_execution_count += 1
             
-        return result, report
-
-    def apply(self, code, messages):
-        return self.batch_apply([code], messages)[0]
-
-    @staticmethod
-    def truncate(s, max_length=400):
-        if isinstance(s, dict):
-            # 如果是字典（包含图像），只截断文本部分
-            if 'text' in s:
-                half = max_length // 2
-                if len(s['text']) > max_length:
-                    s['text'] = s['text'][:half] + "..." + s['text'][-half:]
-            return s
-        else:
-            half = max_length // 2
-            if isinstance(s, str) and len(s) > max_length:
-                s = s[:half] + "..." + s[-half:]
-            return s
-
-    def batch_apply(self, batch_code, messages):
-        all_code_snippets = self.process_generation_to_code(batch_code)
-
-        timeout_cnt = 0
-        all_exec_results = []
-        print(f"################################### num of cpu: {os.cpu_count()} ; len of code: {len(all_code_snippets)} ######################################")
-        with ProcessPool(
-            max_workers=min(len(all_code_snippets), os.cpu_count())
-        ) as pool:
-            executor = partial(
-                self.execute,
-                get_answer_from_stdout=self.get_answer_from_stdout,
-                runtime_class=self.runtime_class,
-                answer_symbol=self.answer_symbol,
-                answer_expr=self.answer_expr,
-                timeout_length=self.timeout_length,
+            # Generate next response part
+            response_text, pred_stop_reason = call_chatgpt_api(
+                args,
+                messages, 
+                client,
+                max_tokens=max_tokens,
+                stop=["</code>"] if exe_code else None
             )
-            future = pool.map(executor, all_code_snippets, [messages], timeout=self.timeout_length)
-            iterator = future.result()
 
-            if len(all_code_snippets) > 100:
-                progress_bar = tqdm(total=len(all_code_snippets), desc="Execute")
-            else:
-                progress_bar = None
-
-            while True:
-                try:
-                    result = next(iterator)
-                    all_exec_results.append(result)
-                except StopIteration:
-                    break
-                except TimeoutError as error:
-                    print(error)
-                    all_exec_results.append(("", "Timeout Error"))
-                    timeout_cnt += 1
-                except Exception as error:
-                    print(f"Error in batch_apply: {error}")
-                    all_exec_results.append(("", f"Error: {str(error)}"))
-                if progress_bar is not None:
-                    progress_bar.update(1)
-
-            if progress_bar is not None:
-                progress_bar.close()
-
-        batch_results = []
-        for code, (res, report) in zip(all_code_snippets, all_exec_results):
-            # 处理结果
-            if isinstance(res, dict):
-                # 如果结果包含图像，特殊处理
-                if 'text' in res:
-                    res['text'] = str(res['text']).strip()
-                    res['text'] = self.truncate(res['text'])
-                report = str(report).strip()
-                report = self.truncate(report)
-            else:
-                # 普通文本结果
-                res = str(res).strip()
-                res = self.truncate(res)
-                report = str(report).strip()
-                report = self.truncate(report)
-            batch_results.append((res, report))
-        return batch_results
-
-
-def _test():
-    image_path = "/mnt/petrelfs/zhaoshitian/vis_tool_inference_engine/test_data/0.JPG"
-    image_base64 = encode_image(image_path)
-    messages = [
-        {
-            "role": "user",
-            "content": [{"type": "text", "text": "From the information on that advertising board, what is the type of this shop?"}]
-        },
-        {
-            "role": "user",
-            "content": [{"type": "text", "text": "image_clue_0"}] + [{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}}]
-        }
-    ]
-    # 测试普通计算
-    math_code ="""
-a = 1
-b = 2
-c = a + b
-print(c)
-"""
-
-    batch_code = [math_code]
-
-    executor = PythonExecutor()
-    predictions = executor.apply(batch_code[0], messages)
-    print("数学计算结果:", predictions)
+        else:
+            final_response = response_text
+            messages = update_messages_with_text(messages, response_text)
+            break
+       
+    return messages, final_response
+
+def evaluate_single_data_video(args, data, client, executor):
+    try:
+        prompt_template = args.prompt_template
+        prompt = args.prompt
+        exe_code = args.exe_code
+        max_tokens = args.max_tokens
+    except:
+        prompt_template = args['prompt_template']
+        prompt = args['prompt']
+        exe_code = args['exe_code']
+        max_tokens = args['max_tokens']
+
+    messages = process_prompt_init_multi_images(data["question"], data['image_path_list'], prompt_template, prompt)
     
-    # 测试图像显示
-    image_code = """
-import matplotlib.pyplot as plt
-import numpy as np
-from PIL import Image
-import io
-
-# 创建一个简单的图像
-x = np.linspace(0, 10, 100)
-y = np.sin(x)
-
-plt.figure(figsize=(8, 6))
-plt.plot(x, y, 'r-', linewidth=2)
-plt.title('Sine Wave')
-plt.grid(True)
-plt.show()
-
-# 也可以显示一个简单的图像
-# 创建一个彩色渐变图像
-arr = np.zeros((100, 100, 3), dtype=np.uint8)
-for i in range(100):
-    for j in range(100):
-        arr[i, j, 0] = i  # 红色通道
-        arr[i, j, 1] = j  # 绿色通道
-        arr[i, j, 2] = 100  # 蓝色通道
-
-img = Image.fromarray(arr)
-plt.figure()
-plt.imshow(img)
-plt.title('Gradient Image')
-plt.show()
-
-print("图像生成完成")
-    """
+    # Generate initial response
+    response_text, pred_stop_reason = call_chatgpt_api(
+        args,
+        messages, 
+        client,
+        max_tokens=max_tokens,
+        stop=["</code>"] if exe_code else None
+    )
+    
+    # Handle response
+    final_response = response_text
+    code_execution_count = 0
+    image_clue_idx = data['image_nums_in_input']
+    
+    while True:
+        # Check if code execution is needed
+        if exe_code and pred_stop_reason == "</code>":
+            # Extract code to execute
+            messages = update_messages_with_code(messages, response_text)
+            code_to_execute = response_text.split("```python")[-1].split("```")[0].strip()
+            
+            # Execute code
+            exe_result = execute_codes([code_to_execute], messages, executor)[0][0]
+            if exe_result is None:
+                text_result = "None"
+                images_result = None
+            else:
+                output, report = exe_result
+                if report == "Done":
+                    error_result = None
+                    try:
+                        text_result = exe_result[0]['text']
+                    except:
+                        text_result = None
+                        print("text result is none.")
+                    try:
+                        images_result = exe_result[0]['images']
+                    except:
+                        images_result = None
+                        print("image result is none.")
+                else:
+                    error_result = report
+                    text_result = None
+                    images_result = None
 
-    image_code = """
-import matplotlib.pyplot as plt
-import numpy as np
-from PIL import Image
-import io
+            messages, new_image_clue_idx = update_messages_with_execute_content(data['image_nums_in_input'], messages, images_result, text_result, error_result, image_clue_idx)
+            image_clue_idx = new_image_clue_idx
+            
+            code_execution_count += 1
+            
+            # Generate next response part
+            response_text, pred_stop_reason = call_chatgpt_api(
+                args,
+                messages, 
+                client,
+                max_tokens=max_tokens,
+                stop=["</code>"] if exe_code else None
+            )
 
-plt.imshow(image_clue_0)
-plt.title("Original Image - Locate Advertising Board")
-plt.show()
-    """
+        else:
+            final_response = response_text
+            messages = update_messages_with_text(messages, response_text)
+            break
+       
+    return messages, final_response
+
+
+# New wrapper functions for safe execution with cleanup
+def evaluate_batch_with_cleanup(args, data_list, client):
+    """Wrapper function to ensure proper cleanup of resources when processing multiple items"""
+    # Initialize executor with process isolation
+    executor = PythonExecutor(use_process_isolation=True)
     
-    image_result = executor.apply(image_code, messages)
-    print("\n图像结果类型:", type(image_result[0]))
-    if isinstance(image_result[0], dict) and 'images' in image_result[0]:
-        print(f"捕获到 {len(image_result[0]['images'])} 个图像")
-        print("第一个图像的base64编码前20个字符:", image_result[0]['images'][0][:20])
+    try:
+        results = []
+        for data in data_list:
+            try:
+                result = evaluate_single_data(args, data, client, executor)
+                results.append(result)
+            except Exception as e:
+                print(f"Error processing data item: {str(e)}")
+                results.append((None, f"Error: {str(e)}"))
+                # Reset the executor for the next item
+                executor.reset()
         
-        # 可选：保存图像到文件
-        for i, img_data in enumerate(image_result[0]['images']):
-            img_bytes = base64.b64decode(img_data)
-            with open(f"captured_image_{i}.png", "wb") as f:
-                f.write(img_bytes)
-            print(f"图像已保存为 captured_image_{i}.png")
-            
-        if 'text' in image_result[0]:
-            print("文本输出:", image_result[0]['text'])
-    else:
-        print("未捕获到图像")
-        print("结果:", image_result[0])
-    
-    print("\n执行状态:", image_result[1])
+        return results
+    finally:
+        # Ensure cleanup of persistent worker
+        del executor
 
+def evaluate_single_with_cleanup(args, data, client):
+    """Wrapper function for evaluating a single item with proper cleanup"""
+    # Initialize executor with process isolation
+    executor = PythonExecutor(use_process_isolation=True)
 
-if __name__ == "__main__":
-    _test()
+    try:
+        result = evaluate_single_data(args, data, client, executor)
+        return result
+    finally:
+        # Ensure cleanup of persistent worker
+        del executor
+
+def evaluate_multi_images_with_cleanup(args, data_list, client):
+    """Wrapper function for multi-image evaluation with proper cleanup"""
+    # Initialize executor with process isolation
+    executor = PythonExecutor(use_process_isolation=True)
+    
+    try:
+        results = []
+        for data in data_list:
+            try:
+                result = evaluate_single_data_multi_images(args, data, client, executor)
+                results.append(result)
+            except Exception as e:
+                print(f"Error processing multi-image data: {str(e)}")
+                results.append((None, f"Error: {str(e)}"))
+                # Reset the executor for the next item
+                executor.reset()
+        
+        return results
+    finally:
+        # Ensure cleanup of persistent worker
+        del executor
+
+def evaluate_video_with_cleanup(args, data_list, client):
+    """Wrapper function for video evaluation with proper cleanup"""
+    # Initialize executor with process isolation
+    executor = PythonExecutor(use_process_isolation=True)
+    
+    try:
+        results = []
+        for data in data_list:
+            try:
+                result = evaluate_single_data_video(args, data, client, executor)
+                results.append(result)
+            except Exception as e:
+                print(f"Error processing video data: {str(e)}")
+                results.append((None, f"Error: {str(e)}"))
+                # Reset the executor for the next item
+                executor.reset()
+        
+        return results
+    finally:
+        # Ensure cleanup of persistent worker
+        del executor