diff --git a/dataflow/operators/core_vision/generate/batch_vqa_generator.py b/dataflow/operators/core_vision/generate/batch_vqa_generator.py
index 3194879..398bb1c 100644
--- a/dataflow/operators/core_vision/generate/batch_vqa_generator.py
+++ b/dataflow/operators/core_vision/generate/batch_vqa_generator.py
@@ -1,8 +1,18 @@
+import pandas as pd
+from typing import List
+
 from dataflow.utils.registry import OPERATOR_REGISTRY
-from dataflow.utils.storage import DataFlowStorage
+from dataflow.utils.storage import FileStorage, DataFlowStorage
 from dataflow.core import OperatorABC, LLMServingABC
 from dataflow import get_logger
-from qwen_vl_utils import process_vision_info
+
+from dataflow.serving.local_model_vlm_serving import LocalModelVLMServing_vllm
+from dataflow.serving.api_vlm_serving_openai import APIVLMServing_openai
+
+
+# 提取判断是否为 API Serving 的辅助函数
+def is_api_serving(serving):
+    return isinstance(serving, APIVLMServing_openai)
 
 
 @OPERATOR_REGISTRY.register()
@@ -30,7 +40,8 @@ def get_desc(lang: str = "zh"):
                 "  - output_key: 生成的答案列表列 (List[str])\n"
                 "功能特点：\n"
                 "  - 自动进行广播 (Broadcasting)，将单图映射到多个问题\n"
-                "  - 适用于由粗到细 (Coarse-to-Fine) 的密集描述生成场景\n"
+                "  - 统一支持 API 和本地 Local 模型部署模式\n"
+                "  - 支持全局批处理加速推理\n"
             )
         else:
             return (
@@ -43,52 +54,119 @@ def get_desc(lang: str = "zh"):
                 "  - output_key: Column storing the list of generated answers\n"
                 "Features:\n"
                 "  - Automatically broadcasts one image to multiple prompts\n"
-                "  - Ideal for coarse-to-fine dense captioning scenarios\n"
+                "  - Unifies support for API and Local model deployment modes\n"
+                "  - Supports global batch processing for faster inference\n"
             )
 
     def run(self, storage: DataFlowStorage, input_prompts_key: str, input_image_key: str, output_key: str):
         self.logger.info(f"Running BatchVQAGenerator on {input_prompts_key}...")
-        df = storage.read("dataframe")
-        
-        all_answers_nested = []
+        df: pd.DataFrame = storage.read("dataframe")
         
+        use_api_mode = is_api_serving(self.serving)
+        if use_api_mode:
+            self.logger.info("Using API serving mode")
+        else:
+            self.logger.info("Using local serving mode")
+
+        # 1. 展平数据阶段 (Flatten Data)
+        # 将 [ [q1, q2], [q3] ] 展平为 [q1, q2, q3]，以便一次性送入大模型获得最高并发性能
+        flat_conversations = []
+        flat_images = []
+        row_question_counts = [] # 记录每一行有几个问题，用于后续重组答案
+
         for idx, row in df.iterrows():
             questions = row.get(input_prompts_key, [])
             image_path = row.get(input_image_key)
             
-            if not questions or not isinstance(questions, list) or not image_path:
-                all_answers_nested.append([])
-                continue
+            # 统一将图片路径处理为 List 格式
+            if isinstance(image_path, str):
+                image_path = [image_path]
+            elif not image_path:
+                image_path = []
+
+            if not isinstance(questions, list):
+                questions = []
+
+            row_question_counts.append(len(questions))
 
-            batch_prompts = []
-            batch_images = []
-            
             for q in questions:
-                raw = [
-                    {"role": "system", "content": self.system_prompt},
-                    {"role": "user", "content": [
-                        {"type": "image", "image": image_path},
-                        {"type": "text", "text": q}
-                    ]}
-                ]
-                image_inputs, _ = process_vision_info(raw)
-                final_p = self.serving.processor.apply_chat_template(raw, tokenize=False, add_generation_prompt=True)
-                
-                batch_prompts.append(final_p)
-                batch_images.append(image_inputs)
-            
-            if not batch_prompts:
-                all_answers_nested.append([])
-                continue
+                # 构造标准对话格式
+                if use_api_mode:
+                    # API 模式通常只需要标准文本，图片通过 image_list 单独传入
+                    conversation = [{"role": "user", "content": q}]
+                else:
+                    # Local 模式（如 vLLM）通常需要手动在文本前拼接 <image> 占位符
+                    img_tokens = "<image>" * len(image_path)
+                    conversation = [{"role": "user", "content": img_tokens + q}]
+
+                flat_conversations.append(conversation)
+                flat_images.append(image_path)
 
-            # 批量调用
-            row_answers = self.serving.generate_from_input(
+        # 2. 批量推理阶段 (Batch Inference)
+        if flat_conversations:
+            flat_outputs = self.serving.generate_from_input_messages(
+                conversations=flat_conversations,
+                image_list=flat_images,
                 system_prompt=self.system_prompt,
-                user_inputs=batch_prompts,
-                image_inputs=batch_images
             )
+        else:
+            flat_outputs = []
+
+        # 3. 重组数据阶段 (Unflatten Data)
+        # 将展平的输出 [a1, a2, a3] 根据 row_question_counts 重组回 [ [a1, a2], [a3] ]
+        all_answers_nested = []
+        current_idx = 0
+        for count in row_question_counts:
+            row_answers = flat_outputs[current_idx : current_idx + count]
             all_answers_nested.append(row_answers)
+            current_idx += count
             
         df[output_key] = all_answers_nested
-        storage.write(df)
-        return [output_key]
\ No newline at end of file
+        output_file = storage.write(df)
+        
+        self.logger.info("Results saved to %s", output_file)
+        return [output_key]
+
+
+# ==========================================
+# 测试用例 (Main Block)
+# ==========================================
+if __name__ == "__main__":
+    # 使用 API 模式测试
+    model = APIVLMServing_openai(
+        api_url="http://172.96.141.132:3001/v1",
+        key_name_of_api_key="DF_API_KEY",
+        model_name="gpt-5-nano-2025-08-07",
+        image_io=None,
+        send_request_stream=False,
+        max_workers=10,
+        timeout=1800
+    )
+
+    # 如果需要测试本地模型，可以解开注释：
+    # model = LocalModelVLMServing_vllm(
+    #     hf_model_name_or_path="Qwen/Qwen2.5-VL-3B-Instruct",
+    #     vllm_tensor_parallel_size=1,
+    #     ...
+    # )
+
+    generator = BatchVQAGenerator(
+        serving=model,
+        system_prompt="You are a helpful visual assistant."
+    )
+
+    storage = FileStorage(
+        first_entry_file_name="./dataflow/example/image_to_text_pipeline/sample_data.json",
+        cache_path="./cache_local",
+        file_name_prefix="batch_vqa",
+        cache_type="json",
+    )
+    
+    storage.step()
+
+    generator.run(
+        storage=storage,
+        input_prompts_key="questions",  # 假设输入列包含多个问题
+        input_image_key="image",
+        output_key="answers",           # 输出列表
+    )
\ No newline at end of file
diff --git a/dataflow/operators/core_vision/generate/fix_prompted_vqa_generator.py b/dataflow/operators/core_vision/generate/fix_prompted_vqa_generator.py
index 8a74ad1..df4a22a 100644
--- a/dataflow/operators/core_vision/generate/fix_prompted_vqa_generator.py
+++ b/dataflow/operators/core_vision/generate/fix_prompted_vqa_generator.py
@@ -1,18 +1,25 @@
 import pandas as pd
+from typing import List
+
 from dataflow.utils.registry import OPERATOR_REGISTRY
+from dataflow.utils.storage import FileStorage, DataFlowStorage
+from dataflow.core import OperatorABC, LLMServingABC
 from dataflow import get_logger
 
-from dataflow.utils.storage import FileStorage, DataFlowStorage
-from dataflow.core import OperatorABC
-from dataflow.core import LLMServingABC
 from dataflow.serving.local_model_vlm_serving import LocalModelVLMServing_vllm
-from qwen_vl_utils import process_vision_info
+from dataflow.serving.api_vlm_serving_openai import APIVLMServing_openai
+
+
+# 提取判断是否为 API Serving 的辅助函数
+def is_api_serving(serving):
+    return isinstance(serving, APIVLMServing_openai)
+
 
 @OPERATOR_REGISTRY.register()
 class FixPromptedVQAGenerator(OperatorABC):
-    '''
+    """
     FixPromptedVQAGenerator generate answers for questions based on provided context. The context can be image/video.
-    '''
+    """
     def __init__(self, 
                  serving: LLMServingABC, 
                  system_prompt: str = "You are a helpful assistant.",
@@ -24,38 +31,17 @@ def __init__(self,
             
     @staticmethod
     def get_desc(lang: str = "zh"):
-        return "基于给定的 system prompt 和 user prompt，并读取 image/video 生成答案" if lang == "zh" else "Generate answers for questions based on provided context. The context can be image/video."
-    
-    def _prepare_batch_inputs(self, input_media_paths, is_image: bool = True):
-        """
-        Construct batched prompts and multimodal inputs from media paths.
-        """
-        prompt_list = []
-        media_paths = []
-        type_media = "image" if is_image else "video"
-
-        for paths in input_media_paths:
-            raw_prompt = [
-                {"role": "system", "content": self.system_prompt},
-                {
-                    "role": "user",
-                    "content": [
-                    ],
-                },
-            ]
-            for path in paths:
-                raw_prompt[1]["content"].append({"type": type_media, type_media: path})
-            raw_prompt[1]["content"].append({"type": "text", "text": self.user_prompt})
-            # Get multimodal inputs
-            media_path, _ = process_vision_info(raw_prompt)
-
-            prompt = self.serving.processor.apply_chat_template(
-                raw_prompt, tokenize=False, add_generation_prompt=True
+        if lang == "zh":
+            return (
+                "固定提示词视觉问答生成算子 (FixPromptedVQAGenerator)。\n"
+                "基于给定的 system prompt 和 user prompt，读取 image/video 生成答案。\n\n"
+                "特点：\n"
+                "  - 支持图像或视频模态\n"
+                "  - 统一支持 API 和本地 Local 模型部署模式\n"
+                "  - 自动管理底层的 <image> 或 <video> 占位符\n"
             )
-            media_paths.append(media_path)
-            prompt_list.append(prompt)
-
-        return prompt_list, media_paths
+        else:
+            return "Generate answers for questions based on provided context. The context can be image/video."
 
     def run(self, 
             storage: DataFlowStorage,
@@ -63,79 +49,121 @@ def run(self,
             input_video_key: str = "video",
             output_answer_key: str = "answer",
             ):
-        if output_answer_key is None:
-            raise ValueError("At least one of output_answer_key must be provided.")
+        if not output_answer_key:
+            raise ValueError("'output_answer_key' must be provided.")
 
         self.logger.info("Running FixPromptedVQA...")
-        self.input_image_key = input_image_key
-        self.input_video_key = input_video_key
-        self.output_answer_key = output_answer_key
-
-        # Load the raw dataframe from the input file
-        dataframe = storage.read('dataframe')
-        self.logger.info(f"Loading, number of rows: {len(dataframe)}")
+        
+        # 加载 DataFrame
+        dataframe: pd.DataFrame = storage.read('dataframe')
+        self.logger.info(f"Loaded dataframe with {len(dataframe)} rows")
+
+        # 提取并清洗多模态列数据
+        image_column = dataframe.get(input_image_key, pd.Series([None] * len(dataframe))).tolist()
+        video_column = dataframe.get(input_video_key, pd.Series([None] * len(dataframe))).tolist()
+
+        # 统一转为 List 格式
+        image_column = [path if isinstance(path, list) else [path] if pd.notna(path) else [] for path in image_column]
+        video_column = [path if isinstance(path, list) else [path] if pd.notna(path) else [] for path in video_column]
+        
+        # 判断当前生效的模态
+        has_images = any(len(p) > 0 for p in image_column)
+        has_videos = any(len(p) > 0 for p in video_column)
+
+        if has_images and has_videos:
+            raise ValueError("Only one of input_image_key or input_video_key can be provided with valid data.")
+        if not has_images and not has_videos:
+            raise ValueError("At least one of input_image_key or input_video_key must contain valid media paths.")
+
+        use_api_mode = is_api_serving(self.serving)
+        if use_api_mode:
+            self.logger.info("Using API serving mode")
+        else:
+            self.logger.info("Using local serving mode")
 
-        image_column = dataframe.get(self.input_image_key, pd.Series([])).tolist()
-        video_column = dataframe.get(self.input_video_key, pd.Series([])).tolist()
+        # 构造对话与输入列表
+        conversations_list = []
+        image_inputs_list = None
+        video_inputs_list = None
 
-        image_column = [path if isinstance(path, list) else [path] for path in image_column]
-        video_column = [path if isinstance(path, list) else [path] for path in video_column]
+        if has_images:
+            image_inputs_list = image_column
+            for paths in image_column:
+                valid_media_count = len([p for p in paths if p])
                 
-        if len(image_column) == 0:
-            image_column = None
-        if len(video_column) == 0:
-            video_column = None
-        if image_column is None and video_column is None:
-            raise ValueError("At least one of input_image_key or input_video_key must be provided.")
-        if image_column is not None and video_column is not None:
-            raise ValueError("Only one of input_image_key or input_video_key must be provided.")
-
-        if image_column is not None:
-            prompt_list, image_inputs_list = self._prepare_batch_inputs(image_column)
-            video_inputs_list = None
-        elif video_column is not None:
-            prompt_list, video_inputs_list = self._prepare_batch_inputs(video_column, is_image=False)
-            image_inputs_list = None
-        else:
-            raise ValueError("At least one of input_image_key or input_video_key must be provided.")
-
-        outputs = self.serving.generate_from_input(
-            system_prompt=self.system_prompt,
-            user_inputs=prompt_list,
-            image_inputs=image_inputs_list,
-            video_inputs=video_inputs_list
+                if use_api_mode:
+                    content = self.user_prompt
+                else:
+                    content = ("<image>" * valid_media_count) + self.user_prompt
+                    
+                conversations_list.append([{"role": "user", "content": content}])
+                
+        elif has_videos:
+            video_inputs_list = video_column
+            for paths in video_column:
+                valid_media_count = len([p for p in paths if p])
+                
+                if use_api_mode:
+                    content = self.user_prompt
+                else:
+                    content = ("<video>" * valid_media_count) + self.user_prompt
+                    
+                conversations_list.append([{"role": "user", "content": content}])
+
+        # 统一调用基类的消息生成接口
+        outputs = self.serving.generate_from_input_messages(
+            conversations=conversations_list,
+            image_list=image_inputs_list,
+            video_list=video_inputs_list,
+            system_prompt=self.system_prompt
         )
 
-        dataframe[self.output_answer_key] = outputs
+        # 保存结果
+        dataframe[output_answer_key] = outputs
         output_file = storage.write(dataframe)
         self.logger.info(f"Results saved to {output_file}")
 
-        return output_answer_key
+        return [output_answer_key]
     
+
+# ==========================================
+# 测试用例 (Main Block)
+# ==========================================
 if __name__ == "__main__":
-    # Initialize model
-    model = LocalModelVLMServing_vllm(
-        hf_model_name_or_path="Qwen/Qwen2.5-VL-3B-Instruct",
-        vllm_tensor_parallel_size=1,
-        vllm_temperature=0.7,
-        vllm_top_p=0.9,
-        vllm_max_tokens=512,
+    # 使用 API 模式进行测试
+    model = APIVLMServing_openai(
+        api_url="http://172.96.141.132:3001/v1",
+        key_name_of_api_key="DF_API_KEY",
+        model_name="gpt-5-nano-2025-08-07",
+        image_io=None,
+        send_request_stream=False,
+        max_workers=10,
+        timeout=1800
     )
 
+    # 如需使用本地模式，解开下方注释：
+    # model = LocalModelVLMServing_vllm(
+    #     hf_model_name_or_path="Qwen/Qwen2.5-VL-3B-Instruct",
+    #     vllm_tensor_parallel_size=1,
+    #     vllm_temperature=0.7,
+    #     vllm_top_p=0.9,
+    #     vllm_max_tokens=512,
+    # )
+
     generator = FixPromptedVQAGenerator(
         serving=model,
         system_prompt="You are a helpful assistant.",
         user_prompt="Please caption the media in detail."
     )
 
-    # Prepare input
+    # 准备输入数据
     storage = FileStorage(
         first_entry_file_name="./dataflow/example/image_to_text_pipeline/fix_prompted_vqa.jsonl", 
         cache_path="./cache_prompted_vqa",
         file_name_prefix="fix_prompted_vqa",
         cache_type="jsonl",
     )
-    storage.step()  # Load the data
+    storage.step()  # 加载数据
 
     generator.run(
         storage=storage,
diff --git a/dataflow/operators/core_vision/generate/multirole_videoqa_generator.py b/dataflow/operators/core_vision/generate/multirole_videoqa_generator.py
index 389d802..5980d49 100644
--- a/dataflow/operators/core_vision/generate/multirole_videoqa_generator.py
+++ b/dataflow/operators/core_vision/generate/multirole_videoqa_generator.py
@@ -1,124 +1,107 @@
-from typing import List, Dict, Any, Union
 import os
 import json
 import pandas as pd
 import re
-from typing import List, Dict, Any
-from PIL import Image
+from typing import List, Dict, Any, Union
+
 from dataflow.core.Operator import OperatorABC
 from dataflow.utils.registry import OPERATOR_REGISTRY
 from dataflow import get_logger
-from dataflow.utils.storage import FileStorage, DataFlowStorage
+from dataflow.utils.storage import DataFlowStorage
 from dataflow.core import VLMServingABC
-from dataflow.serving.local_model_llm_serving import LocalModelLLMServing_vllm
-from qwen_vl_utils import process_vision_info
-from dataflow.prompts.video import (MultiroleQAInitialQAGenerationPrompt, 
-                                MultiroleQACallExpertAgentsPrompt, 
-                                MultiroleQAProfile4ExpertAgents, 
-                                MultiroleQAMasterAgentRevisionPrompt,
-                                MultiroleQADIYFinalQASynthesisPrompt, 
-                                MultiroleQAClassificationPrompt)
-
-# -----------------------------------------------------------------------------
-class Callvlm:
-    def __init__(self, vlm_serving):
-        self.llm_serving = vlm_serving
-    def call(self, prompt_text: str, image_paths: List[str], system_prompt: str) -> str:
-    
-        image_inputs_list = []
-
-        for path in image_paths:
-            for p in path:
-                raw_prompt = [
-                {"role": "system", "content": system_prompt}
-                ]
-                user_content = []
-
-                user_content.append({"type": "image", "image": p})
-                
-                user_content.append({"type": "text", "text": prompt_text})
-            
-                raw_prompt.append({
-                    "role": "user", 
-                    "content": user_content 
-                })         
 
-                image_inputs, _ = process_vision_info(raw_prompt)
+from dataflow.serving.api_vlm_serving_openai import APIVLMServing_openai
 
-                formatted_prompt = self.llm_serving.processor.apply_chat_template(
-                    raw_prompt, tokenize=False, add_generation_prompt=True
-                )
-                image_inputs_list.append(image_inputs)
+# 引入提示词模板
+from dataflow.prompts.video import (
+    MultiroleQAInitialQAGenerationPrompt, 
+    MultiroleQACallExpertAgentsPrompt, 
+    MultiroleQAProfile4ExpertAgents, 
+    MultiroleQAMasterAgentRevisionPrompt,
+    MultiroleQADIYFinalQASynthesisPrompt, 
+    MultiroleQAClassificationPrompt
+)
 
-        outputs = self.llm_serving.generate_from_input(
-            user_inputs=[formatted_prompt],
-            image_inputs=image_inputs_list
-        )
-        
-        if not outputs:
-            return "" 
-        
-        final_output = outputs[0]
-        
-        if isinstance(final_output, list) and final_output:
-            final_output = final_output[0]
-            
-        return str(final_output).strip()
 # -----------------------------------------------------------------------------
-@OPERATOR_REGISTRY.register()
-class MultiroleVideoQAInitialGenerator(OperatorABC):
+# 辅助函数与基类 (消除重复代码，统一调用规范)
+# -----------------------------------------------------------------------------
+
+def is_api_serving(serving):
+    return isinstance(serving, APIVLMServing_openai)
+
+class MultiroleVideoQABase(OperatorABC):
+    """
+    多智能体视频问答算子基类，提供统一的视频信息提取和模型调用接口。
+    """
     def __init__(self, llm_serving: VLMServingABC):
         self.logger = get_logger()
         self.llm_serving = llm_serving
-        self.initial_gen_prompt = MultiroleQAInitialQAGenerationPrompt()
 
-    def _serialize_v_input(self, v_input: Dict[str, Any]) -> Dict[str, Any]:
-        v_output = {
+    def _extract_video_info(self, v_input: Dict[str, Any]) -> tuple[Dict[str, Any], List[str]]:
+        """
+        提取视频 Meta 和 Clips 文本信息，并将所有有效图片路径展平为一个 List，
+        移除了极度消耗内存的 PIL.Image 预加载逻辑。
+        """
+        v_content = {
             "Meta": v_input.get("Meta", ""),
             "Clips": []
         }
-        raw_clips = v_input.get("Clips", [])
-        total_image_paths = []
+        flat_image_paths = []
 
-        for clip in raw_clips:
+        for clip in v_input.get("Clips", []):
             processed_clip = {
                 "Audio_Text": clip.get("Audio_Text", ""),
                 "Description": clip.get("Description", "")
             }
 
-            image_paths = clip.get("Frames_Images", [])
-            loaded_images = []
+            paths = clip.get("Frames_Images", [])
+            if isinstance(paths, str):
+                paths = [paths]
             
-            if isinstance(image_paths, list):
-                for path in image_paths:
-                    try:
-                        img = Image.open(path).convert("RGB")
-                        loaded_images.append(img)
-                    except Exception as e:
-                        if hasattr(self, 'logger'):
-                            self.logger.error(f"Failed to load image at {path}: {e}")
+            # 过滤并收集有效的图片路径
+            valid_paths = [p for p in paths if isinstance(p, str) and p.strip()]
+            flat_image_paths.extend(valid_paths)
             
-            processed_clip["Frames_Images"] = loaded_images
-            
-            v_output["Clips"].append(processed_clip)
-            total_image_paths.append(image_paths)
-
-        return v_output, total_image_paths
+            processed_clip["Frames_Images"] = valid_paths
+            v_content["Clips"].append(processed_clip)
 
-    def _process_single_video(self, v_input: Dict[str, Any]) -> Dict[str, Any]:
+        return v_content, flat_image_paths
 
-        v_content, all_image_paths = self._serialize_v_input(v_input)
-
-        self.logger.info("Executing Step 1: Initial QA Generation")
+    def _generate_answer(self, prompt_text: str, image_paths: List[str]) -> str:
+        """
+        统一的模型调用接口。自动处理 API/Local 模式和 <image> 占位符。
+        代替了原来臃肿且有逻辑缺陷的 Callvlm 类。
+        """
+        use_api_mode = is_api_serving(self.llm_serving)
+        
+        if use_api_mode:
+            content = prompt_text
+        else:
+            img_tokens = "<image>" * len(image_paths)
+            content = f"{img_tokens}\n{prompt_text}" if img_tokens else prompt_text
+
+        conversation = [{"role": "user", "content": content}]
+
+        outputs = self.llm_serving.generate_from_input_messages(
+            conversations=[conversation],
+            image_list=[image_paths] if image_paths else None,
+            system_prompt=""  # 保持与原逻辑一致，不使用系统提示词
+        )
+        
+        if outputs and len(outputs) > 0:
+            return str(outputs[0]).strip()
+        return ""
 
-        prompt_s1 = self.initial_gen_prompt.build_prompt(v_content)
 
-        call_vlm = Callvlm(self.llm_serving)
-        initial_qa_str = call_vlm.call(prompt_s1, all_image_paths, "")
+# -----------------------------------------------------------------------------
+# Operator 1: Initial QA Generator (阶段一：初始问答生成)
+# -----------------------------------------------------------------------------
 
-        v_output = v_input.copy()
-        v_output["QA"] = initial_qa_str
-        return v_output
+@OPERATOR_REGISTRY.register()
+class MultiroleVideoQAInitialGenerator(MultiroleVideoQABase):
+    def __init__(self, llm_serving: VLMServingABC):
+        super().__init__(llm_serving)
+        self.initial_gen_prompt = MultiroleQAInitialQAGenerationPrompt()
 
     def run(
         self,
@@ -127,376 +110,195 @@ def run(
         input_clips_key: str = "Clips", 
         output_key: str = "QA"
     ):
-        if output_key is None:
-            raise ValueError("output_key must be provided.")
-
-        data_list = storage.read(output_type="dict")
-        df = pd.DataFrame(data_list)
+        df: pd.DataFrame = storage.read("dataframe")
         
-        if not isinstance(df, pd.DataFrame):
-            raise ValueError("storage.read must return a pandas DataFrame")
-
         if input_meta_key not in df.columns or input_clips_key not in df.columns:
-             raise ValueError(f"Input columns {input_meta_key} or {input_clips_key} not found in DataFrame.")
+             raise ValueError(f"Columns '{input_meta_key}' or '{input_clips_key}' not found.")
 
         if output_key not in df.columns:
-            df[output_key] = [None for _ in range(len(df))]
+            df[output_key] = None
 
-        self.logger.info(f"Start processing {len(df)} videos...")
+        self.logger.info(f"[InitialGenerator] Start processing {len(df)} videos...")
 
         for idx, row in df.iterrows():
-            current_output = row[output_key]
-            if current_output is not None and isinstance(current_output, list) and len(current_output) > 0:
+            # 跳过已处理的数据
+            if row.get(output_key) and isinstance(row.get(output_key), list) and len(row[output_key]) > 0:
                 continue
 
-            meta_val = row.get(input_meta_key, "")
             clips_val = row.get(input_clips_key, [])
-
             if not isinstance(clips_val, list):
                 self.logger.warning(f"Row {idx}: 'Clips' is not a list. Skipping.")
                 df.at[idx, output_key] = [] 
                 continue
 
-            v_input = {
-                "Meta": meta_val, 
-                "Clips": clips_val 
-            }
+            v_input = {"Meta": row.get(input_meta_key, ""), "Clips": clips_val}
 
             try:
-                processed_output = self._process_single_video(v_input)
-                
-                qa_result = processed_output.get("QA", [])
-                
-                df.at[idx, output_key] = qa_result
+                v_content, all_image_paths = self._extract_video_info(v_input)
+                prompt_s1 = self.initial_gen_prompt.build_prompt(v_content)
                 
+                initial_qa_str = self._generate_answer(prompt_s1, all_image_paths)
+                df.at[idx, output_key] = initial_qa_str
+
             except Exception as e:
                 self.logger.error(f"Error processing row {idx}: {str(e)}")
                 df.at[idx, output_key] = [] 
 
-        return df
+        storage.write(df)
+        return [output_key]
+
 
 # -----------------------------------------------------------------------------
+# Operator 2: Multi Agent Generator (阶段二：多智能体专家迭代)
+# -----------------------------------------------------------------------------
+
 @OPERATOR_REGISTRY.register()
-class MultiroleVideoQAMultiAgentGenerator(OperatorABC):
-    def __init__(self, llm_serving: VLMServingABC, max_iterations: int):
-        self.logger = get_logger()
-        self.llm_serving = llm_serving
+class MultiroleVideoQAMultiAgentGenerator(MultiroleVideoQABase):
+    def __init__(self, llm_serving: VLMServingABC, max_iterations: int = 3):
+        super().__init__(llm_serving)
         self.max_iterations = max_iterations
         self.call_expert_prompt = MultiroleQACallExpertAgentsPrompt()
         self.expert_profile_prompt = MultiroleQAProfile4ExpertAgents()
         self.master_revision_prompt = MultiroleQAMasterAgentRevisionPrompt()
 
-
-    def _serialize_v_input(self, v_input: Dict[str, Any]) -> Dict[str, Any]:
-        v_output = {
-            "Meta": v_input.get("Meta", ""),
-            "Clips": []
-        }
-        raw_clips = v_input.get("Clips", [])
-        total_image_paths = []
-
-        for clip in raw_clips:
-            processed_clip = {
-                "Audio_Text": clip.get("Audio_Text", ""),
-                "Description": clip.get("Description", "")
-            }
-
-            image_paths = clip.get("Frames_Images", [])
-            loaded_images = []
-            
-            if isinstance(image_paths, list):
-                for path in image_paths:
-                    try:
-                        img = Image.open(path).convert("RGB")
-                        loaded_images.append(img)
-                    except Exception as e:
-                        if hasattr(self, 'logger'):
-                            self.logger.error(f"Failed to load image at {path}: {e}")
-            
-            processed_clip["Frames_Images"] = loaded_images
-            
-            v_output["Clips"].append(processed_clip)
-            total_image_paths.append(image_paths)
-
-        return v_output, total_image_paths
-
     def experts(self, call_for_experts_response: str) -> List[Dict[str, str]]:
-            """
-            """
-            experts_list: List[Dict[str, str]] = []
-
-            json_matches = re.findall(r'\{.*?\}', call_for_experts_response, re.DOTALL)
+        experts_list: List[Dict[str, str]] = []
+        json_matches = re.findall(r'\{.*?\}', call_for_experts_response, re.DOTALL)
 
-            for json_str in json_matches:
-                try:
-                    expert_data: Dict[str, Any] = json.loads(json_str.strip())
-
-                    role_raw = expert_data.get("Expert_Role", "")
-                    subtask_raw = expert_data.get("Subtask", "")
-
-                    role = role_raw.strip('<> ').strip()
-                    subtask = subtask_raw.strip('<> ').strip()
-
-                    if role and subtask:
-                        experts_list.append({
-                            "role": role,
-                            "subtask": subtask
-                        })
-
-                except json.JSONDecodeError:
-                    continue
-                except AttributeError:
-                    continue
-
-            return experts_list
-
-    def _process_single_video(self, v_input: Dict[str, Any], init_QA: str) -> Dict[str, Any]:
-
-        v_content, all_image_paths = self._serialize_v_input(v_input)
-
-        qa_history = [] 
-        qa_history.append(init_QA)
-        current_qa_pool_str = init_QA
-
-        # ---------------- Loop: Expert Iteration ----------------
-        iteration_count = 0
-        expert_history = []
-        while iteration_count < self.max_iterations:
-            self.logger.info(f"Iteration {iteration_count + 1}: Check for Experts")
-
-            prompt_s2 = self.call_expert_prompt.build_prompt(v_content, current_qa_pool_str, expert_history)
-
-            call_vlm = Callvlm(self.llm_serving)
-            call_for_experts_response = call_vlm.call(prompt_s2, all_image_paths, "")
-
-            if isinstance(call_for_experts_response, str):
-                if "NO_EXPERTS" in call_for_experts_response:
-                    self.logger.info("Master Agent decided to end iteration.")
-                    break
-
-            experts_list = self.experts(call_for_experts_response)
-            for exp in experts_list:
-                expert_history.append(exp)
-
-            for expert in experts_list:
-                expert_profile = expert["role"]
-                subtask = expert["subtask"]
-                
-                prompt_s3 = self.expert_profile_prompt.build_prompt(expert_profile, v_content, subtask)
-                expert_qa_str = call_vlm.call(prompt_s3, all_image_paths, "")
-
-                prompt_s4 = self.master_revision_prompt.build_prompt(v_content, expert_qa_str, current_qa_pool_str)
-                revised_qa_str = call_vlm.call(prompt_s4, all_image_paths, "")
-                
-                current_qa_pool_str += f"\n{revised_qa_str}"
-                qa_history.append(revised_qa_str)
+        for json_str in json_matches:
+            try:
+                expert_data = json.loads(json_str.strip())
+                role = expert_data.get("Expert_Role", "").strip('<> ').strip()
+                subtask = expert_data.get("Subtask", "").strip('<> ').strip()
 
-            iteration_count += 1
+                if role and subtask:
+                    experts_list.append({"role": role, "subtask": subtask})
+            except (json.JSONDecodeError, AttributeError):
+                continue
 
-        v_output = v_input.copy()
-        v_output["QA"] = qa_history
-        return v_output
+        return experts_list
 
     def run(
         self,
-        df: pd.DataFrame,
+        storage: DataFlowStorage,
         input_meta_key: str = "Meta", 
         input_clips_key: str = "Clips", 
         output_key: str = "QA"
     ):
-        if output_key is None:
-            raise ValueError("output_key must be provided.")
-        
-        if not isinstance(df, pd.DataFrame):
-            raise ValueError("df must be a pandas DataFrame")
-
-        if input_meta_key not in df.columns or input_clips_key not in df.columns:
-             raise ValueError(f"Input columns {input_meta_key} or {input_clips_key} not found in DataFrame.")
-
-        self.logger.info(f"Start processing {len(df)} videos...")
+        df: pd.DataFrame = storage.read("dataframe")
+        self.logger.info(f"[MultiAgentGenerator] Start processing {len(df)} videos...")
 
         for idx, row in df.iterrows():
-            current_output = row[output_key]
-
-            meta_val = row.get(input_meta_key, "")
             clips_val = row.get(input_clips_key, [])
-            init_QA = row.get(output_key, "")
+            init_qa = row.get(output_key, "")
 
             if not isinstance(clips_val, list):
-                self.logger.warning(f"Row {idx}: 'Clips' is not a list. Skipping.")
-                df.at[idx, output_key] = [] 
                 continue
 
-            v_input = {
-                "Meta": meta_val, 
-                "Clips": clips_val 
-            }
+            v_input = {"Meta": row.get(input_meta_key, ""), "Clips": clips_val}
 
             try:
-                processed_output = self._process_single_video(v_input, init_QA)
-                
-                qa_result = processed_output.get("QA", [])
-                
-                df.at[idx, output_key] = qa_result
-                
-            except Exception as e:
-                self.logger.error(f"Error processing row {idx}: {str(e)}")
-                df.at[idx, output_key] = [] 
-
-        return df
-
-# -----------------------------------------------------------------------------
-@OPERATOR_REGISTRY.register()
-class MultiroleVideoQAFinalGenerator(OperatorABC):
-    def __init__(self, llm_serving: VLMServingABC):
-        self.logger = get_logger()
-        self.llm_serving = llm_serving
-        self.final_synthesis_prompt = MultiroleQADIYFinalQASynthesisPrompt()
-        self.classification_prompt = MultiroleQAClassificationPrompt()
+                v_content, all_image_paths = self._extract_video_info(v_input)
 
+                qa_history = [init_qa]
+                current_qa_pool_str = str(init_qa)
+                expert_history = []
 
-    def _serialize_v_input(self, v_input: Dict[str, Any]) -> Dict[str, Any]:
-        v_output = {
-            "Meta": v_input.get("Meta", ""),
-            "Clips": []
-        }
-        raw_clips = v_input.get("Clips", [])
-        total_image_paths = []
+                for i in range(self.max_iterations):
+                    self.logger.info(f"Row {idx} - Iteration {i + 1}: Check for Experts")
+                    prompt_s2 = self.call_expert_prompt.build_prompt(v_content, current_qa_pool_str, expert_history)
+                    call_for_experts_response = self._generate_answer(prompt_s2, all_image_paths)
 
-        for clip in raw_clips:
-            processed_clip = {
-                "Audio_Text": clip.get("Audio_Text", ""),
-                "Description": clip.get("Description", "")
-            }
+                    if "NO_EXPERTS" in call_for_experts_response:
+                        self.logger.info("Master Agent decided to end iteration.")
+                        break
 
-            image_paths = clip.get("Frames_Images", [])
-            loaded_images = []
-            
-            if isinstance(image_paths, list):
-                for path in image_paths:
-                    try:
-                        img = Image.open(path).convert("RGB")
-                        loaded_images.append(img)
-                    except Exception as e:
-                        if hasattr(self, 'logger'):
-                            self.logger.error(f"Failed to load image at {path}: {e}")
-            
-            processed_clip["Frames_Images"] = loaded_images
-            
-            v_output["Clips"].append(processed_clip)
-            total_image_paths.append(image_paths)
+                    experts_list = self.experts(call_for_experts_response)
+                    expert_history.extend(experts_list)
 
-        return v_output, total_image_paths
+                    for expert in experts_list:
+                        prompt_s3 = self.expert_profile_prompt.build_prompt(expert["role"], v_content, expert["subtask"])
+                        expert_qa_str = self._generate_answer(prompt_s3, all_image_paths)
 
-    def extract(
-            self,
-            final_qa_json_str: str, 
-            logger: Any = None
-        ) -> Union[List[Dict[str, Any]], str]:
+                        prompt_s4 = self.master_revision_prompt.build_prompt(v_content, expert_qa_str, current_qa_pool_str)
+                        revised_qa_str = self._generate_answer(prompt_s4, all_image_paths)
+                        
+                        current_qa_pool_str += f"\n{revised_qa_str}"
+                        qa_history.append(revised_qa_str)
 
-            JSON_ARRAY_REGEX = re.compile(r"(\[.*\])", re.DOTALL)
+                df.at[idx, output_key] = qa_history
 
-            match = JSON_ARRAY_REGEX.search(final_qa_json_str)
-            
-            if not match:
-                if logger:
-                    logger.warning("Failed to find JSON array structure (missing [ or ]).")
-                return final_qa_json_str 
-
-            json_block = match.group(1)
-            
-            qa_list: Union[List[Dict[str, Any]], str]
-            try:
-                qa_list = json.loads(json_block)
-                
-                if not isinstance(qa_list, list):
-                    raise TypeError("Parsed result is not a list (e.g., VLM outputted a single object instead of an array).")
-
-            except json.JSONDecodeError as e:
-                if logger:
-                    logger.warning(f"Failed to parse extracted JSON block (Decode Error: {e}).")
-                qa_list = final_qa_json_str 
-                
             except Exception as e:
-                if logger:
-                    logger.warning(f"Failed to parse extracted JSON block (General Error: {e}).")
-                qa_list = final_qa_json_str 
-                
-            return qa_list
-
-    
-    def _process_single_video(self, v_input: Dict[str, Any], qa_history: List[str]) -> Dict[str, Any]:
-
-        v_content, all_image_paths = self._serialize_v_input(v_input)
-        
-        # ---------------- Step 5: Final QA Synthesis ----------------
-        self.logger.info("Executing Step 5: Final QA Synthesis")
+                self.logger.error(f"Error processing row {idx}: {str(e)}")
 
-        prompt_s5 = self.final_synthesis_prompt.build_prompt(qa_history)
+        storage.write(df)
+        return [output_key]
 
-        call_vlm = Callvlm(self.llm_serving)
-        synthesized_qa_str = call_vlm.call(prompt_s5, all_image_paths, "")
 
-        # ---------------- Step 6: Question Classification ----------------
-        self.logger.info("Executing Step 6: Question Classification")
-        prompt_s6 = self.classification_prompt.build_prompt(synthesized_qa_str)
+# -----------------------------------------------------------------------------
+# Operator 3: Final Generator (阶段三：最终合成与分类)
+# -----------------------------------------------------------------------------
 
-        final_qa_json_str = call_vlm.call(prompt_s6, all_image_paths, "")
+@OPERATOR_REGISTRY.register()
+class MultiroleVideoQAFinalGenerator(MultiroleVideoQABase):
+    def __init__(self, llm_serving: VLMServingABC):
+        super().__init__(llm_serving)
+        self.final_synthesis_prompt = MultiroleQADIYFinalQASynthesisPrompt()
+        self.classification_prompt = MultiroleQAClassificationPrompt()
 
-        # ---------------- Construct V_output ----------------
-        qa_list = self.extract(final_qa_json_str, self.logger)
+    def extract(self, final_qa_json_str: str) -> Union[List[Dict[str, Any]], str]:
+        JSON_ARRAY_REGEX = re.compile(r"(\[.*\])", re.DOTALL)
+        match = JSON_ARRAY_REGEX.search(final_qa_json_str)
         
-        if isinstance(qa_list, str):
-            self.logger.warning("Failed to parse Final QA JSON, returning raw string.")
-            qa_list = final_qa_json_str
-
-        v_output = v_input.copy()
-        v_output["QA"] = qa_list
-        return v_output
+        if not match:
+            self.logger.warning("Failed to find JSON array structure.")
+            return final_qa_json_str 
+
+        try:
+            qa_list = json.loads(match.group(1))
+            if not isinstance(qa_list, list):
+                raise TypeError("Parsed result is not a list.")
+            return qa_list
+        except Exception as e:
+            self.logger.warning(f"Failed to parse extracted JSON block: {e}")
+            return final_qa_json_str 
 
     def run(
         self,
         storage: DataFlowStorage,
-        df: pd.DataFrame,
         input_meta_key: str = "Meta", 
         input_clips_key: str = "Clips", 
         output_key: str = "QA"
     ):
-        if output_key is None:
-            raise ValueError("output_key must be provided.")
-        
-        if not isinstance(df, pd.DataFrame):
-            raise ValueError("df must be a pandas DataFrame")
-
-        if input_meta_key not in df.columns or input_clips_key not in df.columns:
-             raise ValueError(f"Input columns {input_meta_key} or {input_clips_key} not found in DataFrame.")
-
-        self.logger.info(f"Start processing {len(df)} videos...")
+        df: pd.DataFrame = storage.read("dataframe")
+        self.logger.info(f"[FinalGenerator] Start processing {len(df)} videos...")
 
         for idx, row in df.iterrows():
-            current_output = row[output_key]
-
-            meta_val = row.get(input_meta_key, "")
             clips_val = row.get(input_clips_key, [])
-            qa_history = row.get(output_key, "")
+            qa_history = row.get(output_key, [])
 
             if not isinstance(clips_val, list):
-                self.logger.warning(f"Row {idx}: 'Clips' is not a list. Skipping.")
-                df.at[idx, output_key] = [] 
                 continue
 
-            v_input = {
-                "Meta": meta_val, 
-                "Clips": clips_val 
-            }
+            v_input = {"Meta": row.get(input_meta_key, ""), "Clips": clips_val}
 
             try:
-                processed_output = self._process_single_video(v_input, qa_history)
-                
-                qa_result = processed_output.get("QA", [])
-                
-                df.at[idx, output_key] = qa_result
-                
+                v_content, all_image_paths = self._extract_video_info(v_input)
+
+                # Step 5: Final QA Synthesis
+                self.logger.info(f"Row {idx} - Step 5: Final QA Synthesis")
+                prompt_s5 = self.final_synthesis_prompt.build_prompt(qa_history)
+                synthesized_qa_str = self._generate_answer(prompt_s5, all_image_paths)
+
+                # Step 6: Question Classification
+                self.logger.info(f"Row {idx} - Step 6: Question Classification")
+                prompt_s6 = self.classification_prompt.build_prompt(synthesized_qa_str)
+                final_qa_json_str = self._generate_answer(prompt_s6, all_image_paths)
+
+                # Extract and Save
+                qa_list = self.extract(final_qa_json_str)
+                df.at[idx, output_key] = qa_list
+
             except Exception as e:
                 self.logger.error(f"Error processing row {idx}: {str(e)}")
                 df.at[idx, output_key] = [] 
@@ -505,8 +307,3 @@ def run(
         self.logger.info(f"All processing done. Results saved to {output_file}")
 
         return [output_key]
-
-
-    
-
-# -----------------------------------------------------------------------------
diff --git a/dataflow/operators/core_vision/generate/prompt_templated_vqa_generator.py b/dataflow/operators/core_vision/generate/prompt_templated_vqa_generator.py
index ad9a093..85d1d95 100644
--- a/dataflow/operators/core_vision/generate/prompt_templated_vqa_generator.py
+++ b/dataflow/operators/core_vision/generate/prompt_templated_vqa_generator.py
@@ -1,16 +1,21 @@
 import pandas as pd
+from typing import List
+
 from dataflow.utils.registry import OPERATOR_REGISTRY
 from dataflow import get_logger
-
 from dataflow.utils.storage import FileStorage, DataFlowStorage
-from dataflow.core import OperatorABC
-from dataflow.core import LLMServingABC
-from qwen_vl_utils import process_vision_info
 
+from dataflow.core import OperatorABC, LLMServingABC
 from dataflow.serving.local_model_vlm_serving import LocalModelVLMServing_vllm
+from dataflow.serving.api_vlm_serving_openai import APIVLMServing_openai
 from dataflow.prompts.prompt_template import NamedPlaceholderPromptTemplate
 
 
+# 提取判断是否为 API Serving 的辅助函数
+def is_api_serving(serving):
+    return isinstance(serving, APIVLMServing_openai)
+
+
 @OPERATOR_REGISTRY.register()
 class PromptTemplatedVQAGenerator(OperatorABC):
     """
@@ -18,9 +23,6 @@ class PromptTemplatedVQAGenerator(OperatorABC):
     1) 从 DataFrame 读取若干字段（由 input_keys 指定）
     2) 使用 prompt_template.build_prompt(...) 生成纯文本 prompt
     3) 将该 prompt 与 image/video 一起输入多模态模型，生成答案
-
-    其中 prompt_template 需要实现：
-        build_prompt(self, need_fields: set[str], **kwargs) -> str
     """
 
     def __init__(
@@ -43,10 +45,13 @@ def __init__(
     def get_desc(lang: str = "zh"):
         if lang == "zh":
             return (
-                "PromptTemplatedVQAGenerator：先用模板填充文本 prompt，再结合 image/video "
-                "进行多模态问答的算子。\n"
-                "JSONL/DataFrame 中包含若干字段（例如 descriptions、type 等），"
-                "通过 input_keys 将 DataFrame 列映射到模板字段，由 prompt_template 生成最终的文本 Prompt。"
+                "基于模板的动态多模态问答算子 (PromptTemplatedVQAGenerator)。\n"
+                "JSONL/DataFrame 中包含若干字段，通过 input_keys 将 DataFrame 列映射到模板字段，\n"
+                "由 prompt_template 生成最终的文本 Prompt，再结合 image/video 进行多模态问答。\n\n"
+                "特点：\n"
+                "  - 支持动态组装复杂的 Prompt\n"
+                "  - 统一支持 API 和本地 Local 模型部署模式\n"
+                "  - 自动管理底层的 <image> 或 <video> 占位符\n"
             )
         else:
             return (
@@ -55,35 +60,6 @@ def get_desc(lang: str = "zh"):
                 "performs VQA with image/video."
             )
 
-    def _prepare_batch_inputs(self, prompts, input_media_paths, is_image: bool = True):
-
-        prompt_list = []
-        media_paths = []
-        type_media = "image" if is_image else "video"
-
-        for paths, p in zip(input_media_paths, prompts):
-            raw_prompt = [
-                {"role": "system", "content": self.system_prompt},
-                {
-                    "role": "user",
-                    "content": [
-                    ],
-                },
-            ]
-            for path in paths:
-                raw_prompt[1]["content"].append({"type": type_media, type_media: path})
-            raw_prompt[1]["content"].append({"type": "text", "text": p})
-
-            media_path, _ = process_vision_info(raw_prompt)
-            prompt = self.serving.processor.apply_chat_template(
-                raw_prompt, tokenize=False, add_generation_prompt=True
-            )
-
-            media_paths.append(media_path)
-            prompt_list.append(prompt)
-
-        return prompt_list, media_paths
-
     def run(
         self,
         storage: DataFlowStorage,
@@ -98,16 +74,10 @@ def run(
         - input_image_key / input_video_key: 存放图片/视频路径的列名（只允许其一存在）
         - output_answer_key: 输出答案列名
         - **input_keys: 模板字段名 -> DataFrame 列名
-            例如：
-                descriptions="descriptions", type="type"
-
-        逻辑：
-        1. 从 DataFrame 每行抽取 input_keys 对应列，形成 key_dict
-        2. 用 prompt_template.build_prompt(need_fields, **key_dict) 得到文本 prompt
-        3. 再与 image/video 一起进入多模态模型
+            例如：descriptions="descriptions_col", type="type_col"
         """
-        if output_answer_key is None:
-            raise ValueError("output_answer_key must be provided.")
+        if not output_answer_key:
+            raise ValueError("'output_answer_key' must be provided.")
 
         if len(input_keys) == 0:
             raise ValueError(
@@ -116,13 +86,12 @@ def run(
             )
 
         self.logger.info("Running PromptTemplatedVQAGenerator...")
-        self.input_image_key = input_image_key
-        self.input_video_key = input_video_key
-        self.output_answer_key = output_answer_key
 
-        dataframe = storage.read("dataframe")
-        self.logger.info(f"Loading, number of rows: {len(dataframe)}")
+        # 1. 加载 DataFrame
+        dataframe: pd.DataFrame = storage.read("dataframe")
+        self.logger.info(f"Loaded dataframe with {len(dataframe)} rows")
 
+        # 2. 动态生成 Prompt 文本
         need_fields = set(input_keys.keys())
         prompt_column = []
 
@@ -130,70 +99,105 @@ def run(
             key_dict = {}
             for key in need_fields:
                 col_name = input_keys[key]  # 模板字段名 -> DataFrame 列名
-                key_dict[key] = row[col_name]
+                key_dict[key] = row.get(col_name, "")
             prompt_text = self.prompt_template.build_prompt(need_fields, **key_dict)
             prompt_column.append(prompt_text)
 
         self.logger.info(
-            f"Using prompt_template to build prompts with fields {need_fields}, "
-            f"prepared {len(prompt_column)} prompts."
+            f"Built {len(prompt_column)} prompts using fields: {need_fields}"
         )
 
-        # 3. 读取 image / video 列
-        image_column = dataframe.get(self.input_image_key, pd.Series([])).tolist()
-        video_column = dataframe.get(self.input_video_key, pd.Series([])).tolist()
+        # 3. 提取并清洗多模态列数据
+        image_column = dataframe.get(input_image_key, pd.Series([None] * len(dataframe))).tolist()
+        video_column = dataframe.get(input_video_key, pd.Series([None] * len(dataframe))).tolist()
 
-        image_column = [path if isinstance(path, list) else [path] for path in image_column]
-        video_column = [path if isinstance(path, list) else [path] for path in video_column]
+        image_column = [path if isinstance(path, list) else [path] if pd.notna(path) else [] for path in image_column]
+        video_column = [path if isinstance(path, list) else [path] if pd.notna(path) else [] for path in video_column]
 
-        if len(image_column) == 0:
-            image_column = None
-        if len(video_column) == 0:
-            video_column = None
+        has_images = any(len(p) > 0 for p in image_column)
+        has_videos = any(len(p) > 0 for p in video_column)
 
-        if image_column is None and video_column is None:
-            raise ValueError(
-                "At least one of input_image_key or input_video_key must be provided."
-            )
-        if image_column is not None and video_column is not None:
-            raise ValueError(
-                "Only one of input_image_key or input_video_key must be provided."
-            )
+        if has_images and has_videos:
+            raise ValueError("Only one of input_image_key or input_video_key can be provided with valid data.")
+        if not has_images and not has_videos:
+            raise ValueError("At least one of input_image_key or input_video_key must contain valid media paths.")
 
-        # 4. 组装多模态输入并推理
-        if image_column is not None:
-            prompt_list, image_inputs_list = self._prepare_batch_inputs(
-                prompt_column, image_column
-            )
-            video_inputs_list = None
+        use_api_mode = is_api_serving(self.serving)
+        if use_api_mode:
+            self.logger.info("Using API serving mode")
         else:
-            prompt_list, video_inputs_list = self._prepare_batch_inputs(
-                prompt_column, video_column, is_image=False
-            )
-            image_inputs_list = None
-
-        outputs = self.serving.generate_from_input(
+            self.logger.info("Using local serving mode")
+
+        # 4. 构造多模态对话结构
+        conversations_list = []
+        image_inputs_list = None
+        video_inputs_list = None
+
+        if has_images:
+            image_inputs_list = image_column
+            for prompt_text, paths in zip(prompt_column, image_column):
+                valid_media_count = len([p for p in paths if p])
+                
+                if use_api_mode:
+                    content = prompt_text
+                else:
+                    img_tokens = "<image>" * valid_media_count
+                    content = f"{img_tokens}\n{prompt_text}" if img_tokens else prompt_text
+                    
+                conversations_list.append([{"role": "user", "content": content}])
+                
+        elif has_videos:
+            video_inputs_list = video_column
+            for prompt_text, paths in zip(prompt_column, video_column):
+                valid_media_count = len([p for p in paths if p])
+                
+                if use_api_mode:
+                    content = prompt_text
+                else:
+                    vid_tokens = "<video>" * valid_media_count
+                    content = f"{vid_tokens}\n{prompt_text}" if vid_tokens else prompt_text
+                    
+                conversations_list.append([{"role": "user", "content": content}])
+
+        # 5. 统一调用基类接口
+        outputs = self.serving.generate_from_input_messages(
+            conversations=conversations_list,
+            image_list=image_inputs_list,
+            video_list=video_inputs_list,
             system_prompt=self.system_prompt,
-            user_inputs=prompt_list,
-            image_inputs=image_inputs_list,
-            video_inputs=video_inputs_list,
         )
 
-        dataframe[self.output_answer_key] = outputs
+        dataframe[output_answer_key] = outputs
         output_file = storage.write(dataframe)
         self.logger.info(f"Results saved to {output_file}")
 
-        return output_answer_key
+        return [output_answer_key]
 
 
+# ==========================================
+# 测试用例 (Main Block)
+# ==========================================
 if __name__ == "__main__":
-    model = LocalModelVLMServing_vllm(
-        hf_model_name_or_path="Qwen/Qwen2.5-VL-3B-Instruct",
-        vllm_tensor_parallel_size=1,
-        vllm_temperature=0.7,
-        vllm_top_p=0.9,
-        vllm_max_tokens=512,
+    
+    # 使用 API 模式测试
+    model = APIVLMServing_openai(
+        api_url="http://172.96.141.132:3001/v1",
+        key_name_of_api_key="DF_API_KEY",
+        model_name="gpt-5-nano-2025-08-07",
+        image_io=None,
+        send_request_stream=False,
+        max_workers=10,
+        timeout=1800
     )
+
+    # 如需测试 Local 模型，请解开注释
+    # model = LocalModelVLMServing_vllm(
+    #     hf_model_name_or_path="Qwen/Qwen2.5-VL-3B-Instruct",
+    #     vllm_tensor_parallel_size=1,
+    #     vllm_temperature=0.7,
+    #     vllm_top_p=0.9,
+    #     vllm_max_tokens=512,
+    # )
     
     TEMPLATE = (
         "Descriptions:\n"
@@ -211,20 +215,22 @@ def run(
         prompt_template=prompt_template,
     )
 
-    # Prepare input
+    # 准备数据流存储
     storage = FileStorage(
         first_entry_file_name="./dataflow/example/image_to_text_pipeline/prompttemplated_vqa.jsonl", 
         cache_path="./cache_prompted_vqa",
         file_name_prefix="prompttemplated_vqa",
         cache_type="jsonl",
     )
-    storage.step()  # Load the data
+    storage.step()  # 加载数据
 
     generator.run(
         storage=storage,
         input_image_key="image",
         input_video_key="video",
         output_answer_key="answer",
+        # 下方为 input_keys 参数，表示：模板中 {descriptions} 对应 DataFrame 中的 "descriptions" 列
         descriptions="descriptions",
         type="type",
     )
+    
\ No newline at end of file
diff --git a/dataflow/operators/core_vision/generate/vlm_bbox_generator.py b/dataflow/operators/core_vision/generate/vlm_bbox_generator.py
index 0779e27..d70e929 100644
--- a/dataflow/operators/core_vision/generate/vlm_bbox_generator.py
+++ b/dataflow/operators/core_vision/generate/vlm_bbox_generator.py
@@ -1,34 +1,47 @@
-import torch
-import gc
 import re
-from PIL import Image
+import pandas as pd
 from typing import List, Dict, Any
 
 from dataflow import get_logger
 from dataflow.utils.registry import OPERATOR_REGISTRY
-from dataflow.utils.storage import DataFlowStorage
+from dataflow.utils.storage import FileStorage, DataFlowStorage
 from dataflow.core import OperatorABC, LLMServingABC
-from qwen_vl_utils import process_vision_info
+from dataflow.serving.local_model_vlm_serving import LocalModelVLMServing_vllm
+from dataflow.serving.api_vlm_serving_openai import APIVLMServing_openai
+
+
+# 提取判断是否为 API Serving 的辅助函数
+def is_api_serving(serving):
+    return isinstance(serving, APIVLMServing_openai)
+
 
 def parse_bbox_logic(text: str) -> List[List[float]]:
-    """解析模型生成的 BBox 文本 (x1, y1, x2, y2)"""
-    if not text: return []
+    """解析模型生成的 BBox 文本 (x1, y1), (x2, y2)"""
+    if not text: 
+        return []
+    
     bboxes = []
     # 兼容 (0.1, 0.1), (0.2, 0.2) 格式
     pattern = r'\(\s*([0-9.]+)\s*,\s*([0-9.]+)\s*\)\s*,\s*\(\s*([0-9.]+)\s*,\s*([0-9.]+)\s*\)'
+    
     for match in re.finditer(pattern, text):
         try:
             coords = list(map(float, match.groups()))
             x1, y1, x2, y2 = coords
-            # 归一化处理 (适配 0-1000 输出)
+            
+            # 归一化处理 (适配 0-1000 输出，转换为 0-1 的相对坐标)
             if any(c > 1.05 for c in coords):
                 x1, y1, x2, y2 = x1/1000, y1/1000, x2/1000, y2/1000
+                
             x1, x2 = min(x1, x2), max(x1, x2)
             y1, y2 = min(y1, y2), max(y1, y2)
             bboxes.append([x1, y1, x2, y2])
-        except: continue
+        except Exception: 
+            continue
+            
     return bboxes
 
+
 @OPERATOR_REGISTRY.register()
 class VLMBBoxGenerator(OperatorABC):
     """
@@ -40,93 +53,158 @@ def __init__(self, serving: LLMServingABC, prompt_template: str = 'Detect "{keyw
         self.serving = serving
         self.prompt_tmpl = prompt_template
         self.logger = get_logger()
+        self.system_prompt = "You are a helpful assistant capable of visual grounding."
 
     @staticmethod
     def get_desc(lang: str = "zh"):
-        return "使用 VLM 模型检测关键词的 BBox (支持批量处理)。"
+        if lang == "zh":
+            return (
+                "视觉定位 BBox 生成算子 (VLMBBoxGenerator)。\n"
+                "输入图像和关键词列表，使用 VLM 模型检测并输出目标的边界框。\n\n"
+                "特点：\n"
+                "  - 自动过滤和去重关键词\n"
+                "  - 全局 Batch 展平处理，极大提升吞吐量\n"
+                "  - 统一支持 API 和本地 Local 模型部署模式，自动管理多模态占位符\n"
+                "  - 自动归一化坐标并提取前 3 个置信度最高的候选框\n"
+            )
+        else:
+            return "Uses a VLM to detect bounding boxes for a list of keywords (Supports batching)."
 
     def run(self, storage: DataFlowStorage, input_image_key: str, input_kws_key: str, output_key: str):
+        if not output_key:
+            raise ValueError("'output_key' must be provided.")
+
         self.logger.info("Running VLMBBoxGenerator...")
-        df = storage.read("dataframe")
-        bbox_maps = []
+        df: pd.DataFrame = storage.read("dataframe")
         
+        use_api_mode = is_api_serving(self.serving)
+        if use_api_mode:
+            self.logger.info("Using API serving mode")
+        else:
+            self.logger.info("Using local serving mode")
+
+        # ---------------------------------------------------------
+        # 1. 展平数据阶段 (Flatten Data)
+        # 将 N 张图片和 M 个关键词展平为 N*M 的一维请求列表
+        # ---------------------------------------------------------
+        flat_conversations = []
+        flat_images = []
+        row_mappings = []  # 记录这道 prompt 属于哪一行的哪个关键词：{"row_idx": int, "keyword": str}
+
         for idx, row in df.iterrows():
             img_path = row.get(input_image_key)
             keywords = row.get(input_kws_key, [])
-            row_map = {}
             
+            # 清洗图片路径
+            if isinstance(img_path, str):
+                img_path = [img_path]
+            elif not img_path:
+                img_path = []
+
             # 校验数据有效性
-            if not keywords or not isinstance(keywords, list) or not img_path:
-                bbox_maps.append({})
+            if not isinstance(keywords, list) or not img_path:
                 continue
             
             # 针对单张图片，去重关键词
-            unique_kws = list(set(keywords))
+            unique_kws = list(set([str(k) for k in keywords if k]))
             if not unique_kws:
-                bbox_maps.append({})
                 continue
-
-            # --- 构造 Batch Request (One Image vs N Keywords) ---
-            batch_prompts = []
-            batch_images = []
             
             for kw in unique_kws:
                 safe_kw = kw.replace('"', '\\"')
                 text_prompt = self.prompt_tmpl.format(keyword=safe_kw)
                 
-                # 构造符合 Serving 接口的 raw prompt
-                raw_prompt = [
-                    {"role": "system", "content": "You are a helpful assistant capable of visual grounding."},
-                    {"role": "user", "content": [
-                        {"type": "image", "image": img_path},
-                        {"type": "text", "text": text_prompt}
-                    ]}
-                ]
+                if use_api_mode:
+                    content = text_prompt
+                else:
+                    img_tokens = "<image>" * len(img_path)
+                    content = f"{img_tokens}\n{text_prompt}" if img_tokens else text_prompt
                 
-                # 处理 Vision Info
-                try:
-                    img_inp, _ = process_vision_info(raw_prompt)
-                    prompt_str = self.serving.processor.apply_chat_template(
-                        raw_prompt, tokenize=False, add_generation_prompt=True
-                    )
-                    
-                    # Qwen2.5-VL 防御性补丁 (防止 template 没加占位符)
-                    if "<|image_pad|>" not in prompt_str and "<image>" not in prompt_str:
-                        prompt_str = "<|vision_start|><|image_pad|><|vision_end|>" + prompt_str
-                        
-                    batch_prompts.append(prompt_str)
-                    batch_images.append(img_inp)
-                except Exception as e:
-                    self.logger.warning(f"Failed to prepare prompt for '{kw}': {e}")
+                flat_conversations.append([{"role": "user", "content": content}])
+                flat_images.append(img_path)
+                row_mappings.append({"row_idx": idx, "keyword": kw})
+
+        # ---------------------------------------------------------
+        # 2. 批量推理阶段 (Batch Inference)
+        # 一次性将所有组合送入大模型，最大化利用显存和并发
+        # ---------------------------------------------------------
+        if flat_conversations:
+            self.logger.info(f"Generating BBox for {len(flat_conversations)} image-keyword pairs...")
+            flat_outputs = self.serving.generate_from_input_messages(
+                conversations=flat_conversations,
+                image_list=flat_images,
+                system_prompt=self.system_prompt
+            )
+        else:
+            flat_outputs = []
+
+        # ---------------------------------------------------------
+        # 3. 重组解析阶段 (Unflatten & Parse Data)
+        # ---------------------------------------------------------
+        # 初始化一个与 df 等长的空字典列表
+        bbox_maps = [{} for _ in range(len(df))]
+        
+        for mapping, out_text in zip(row_mappings, flat_outputs):
+            idx = mapping["row_idx"]
+            kw = mapping["keyword"]
             
-            # --- 批量调用 Serving ---
-            if not batch_prompts:
-                bbox_maps.append({})
+            # 检查是否包含 "not found"
+            if not out_text or "not found" in str(out_text).lower():
                 continue
-
-            try:
-                outputs = self.serving.generate_from_input(
-                    user_inputs=batch_prompts,
-                    image_inputs=batch_images
-                )
-                
-                # --- 解析结果 ---
-                for kw, out_text in zip(unique_kws, outputs):
-                    # 检查是否包含 "not found"
-                    if "not found" in out_text.lower():
-                        continue
-                    
-                    boxes = parse_bbox_logic(out_text)
-                    if boxes:
-                        # 格式化为字符串列表
-                        box_strs = [f"[{b[0]:.3f}, {b[1]:.3f}, {b[2]:.3f}, {b[3]:.3f}]" for b in boxes]
-                        row_map[kw] = box_strs[:3] # 保留前3个
-                        
-            except Exception as e:
-                self.logger.error(f"Serving generation error at row {idx}: {e}")
             
-            bbox_maps.append(row_map)
-        
+            boxes = parse_bbox_logic(str(out_text))
+            if boxes:
+                # 格式化为字符串列表，仅保留前 3 个
+                box_strs = [f"[{b[0]:.3f}, {b[1]:.3f}, {b[2]:.3f}, {b[3]:.3f}]" for b in boxes]
+                bbox_maps[idx][kw] = box_strs[:3]
+
         df[output_key] = bbox_maps
-        storage.write(df)
-        return [output_key]
\ No newline at end of file
+        output_file = storage.write(df)
+        self.logger.info(f"Results saved to {output_file}")
+        
+        return [output_key]
+
+
+# ==========================================
+# 测试用例 (Main Block)
+# ==========================================
+if __name__ == "__main__":
+    # 使用 API 模式测试
+    model = APIVLMServing_openai(
+        api_url="http://172.96.141.132:3001/v1",
+        key_name_of_api_key="DF_API_KEY",
+        model_name="gpt-5-nano-2025-08-07",
+        image_io=None,
+        send_request_stream=False,
+        max_workers=10,
+        timeout=1800
+    )
+
+    # 如需测试 Local 模型，请解开注释
+    # model = LocalModelVLMServing_vllm(
+    #     hf_model_name_or_path="Qwen/Qwen2.5-VL-3B-Instruct",
+    #     vllm_tensor_parallel_size=1,
+    #     vllm_temperature=0.7,
+    #     vllm_top_p=0.9,
+    #     vllm_max_tokens=512,
+    # )
+
+    generator = VLMBBoxGenerator(
+        serving=model,
+        prompt_template='Detect "{keyword}". Please provide the bounding boxes in (x1, y1), (x2, y2) format.'
+    )
+
+    storage = FileStorage(
+        first_entry_file_name="./dataflow/example/image_to_text_pipeline/bbox_sample.jsonl", 
+        cache_path="./cache_bbox",
+        file_name_prefix="bbox_gen",
+        cache_type="jsonl",
+    )
+    storage.step()
+
+    generator.run(
+        storage=storage,
+        input_image_key="image",
+        input_kws_key="keywords",  # 假设这列的数据格式为: ["cat", "dog", "car"]
+        output_key="bbox_map",     # 输出将被存为: {"cat": ["[0.1, 0.2, 0.3, 0.4]"], "dog": [...]}
+    )
\ No newline at end of file