From 8ffa131fad26ec39ff51faaf10a813d2155b240c Mon Sep 17 00:00:00 2001
From: Jerry Yan <792602257@qq.com>
Date: Mon, 9 Feb 2026 23:23:05 +0800
Subject: [PATCH] =?UTF-8?q?feat(annotation):=20=E8=87=AA=E5=8A=A8=E6=A0=87?=
 =?UTF-8?q?=E6=B3=A8=E4=BB=BB=E5=8A=A1=E6=94=AF=E6=8C=81=E9=9D=9E=E5=9B=BE?=
 =?UTF-8?q?=E5=83=8F=E7=B1=BB=E5=9E=8B=E6=95=B0=E6=8D=AE=E9=9B=86=EF=BC=88?=
 =?UTF-8?q?TEXT/AUDIO/VIDEO=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

移除自动标注任务创建流程中的 IMAGE-only 限制，使 TEXT、AUDIO、VIDEO
类型数据集均可用于自动标注任务。

- 新增数据库迁移：t_dm_auto_annotation_tasks 表添加 dataset_type 列
- 后端 schema/API/service 全链路传递 dataset_type
- Worker 动态构建 sample key（image/text/audio/video）和输出目录
- 前端移除数据集类型校验，下拉框显示数据集类型标识
- 输出数据集继承源数据集类型，不再硬编码为 IMAGE
- 保持向后兼容：默认值为 IMAGE，worker 有元数据回退和目录 fallback

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 .../OperatorCreate/CreateTask.tsx             |   19 +-
 .../app/db/models/annotation_management.py    |  174 +-
 .../app/module/annotation/interface/auto.py   |  121 +-
 .../app/module/annotation/schema/auto.py      |   92 +-
 .../app/module/annotation/service/auto.py     |  156 +-
 .../datamate/auto_annotation_worker.py        | 1656 +++++++++--------
 .../data-annotation-multitype-migration.sql   |   25 +
 7 files changed, 1161 insertions(+), 1082 deletions(-)
 create mode 100644 scripts/db/data-annotation-multitype-migration.sql

diff --git a/frontend/src/pages/DataAnnotation/OperatorCreate/CreateTask.tsx b/frontend/src/pages/DataAnnotation/OperatorCreate/CreateTask.tsx
index 29c8cc6..420cf8e 100644
--- a/frontend/src/pages/DataAnnotation/OperatorCreate/CreateTask.tsx
+++ b/frontend/src/pages/DataAnnotation/OperatorCreate/CreateTask.tsx
@@ -6,7 +6,7 @@ import { ArrowLeft } from "lucide-react";
 import { Link, useNavigate } from "react-router";
 import { queryDatasetsUsingGet } from "@/pages/DataManagement/dataset.api";
 import { mapDataset } from "@/pages/DataManagement/dataset.const";
-import { Dataset, DatasetType } from "@/pages/DataManagement/dataset.model";
+import { Dataset } from "@/pages/DataManagement/dataset.model";
 import { createAnnotationOperatorTaskUsingPost } from "../annotation.api";
 import { useCreateStepTwo } from "./hooks/useCreateStepTwo";
 import PipelinePreview from "./components/PipelinePreview";
@@ -85,11 +85,6 @@ export default function AnnotationOperatorTaskCreate() {
     try {
       if (currentStep === 1) {
         await form.validateFields();
-
-        if (selectedDataset?.datasetType !== DatasetType.IMAGE) {
-          message.error("自动标注算子编排当前仅支持图片数据集");
-          return;
-        }
       }
       setCurrentStep((prev) => Math.min(prev + 1, 2));
     } catch {
@@ -109,11 +104,6 @@ export default function AnnotationOperatorTaskCreate() {
         return;
       }
 
-      if (selectedDataset?.datasetType !== DatasetType.IMAGE) {
-        message.error("自动标注算子编排当前仅支持图片数据集");
-        return;
-      }
-
       const outputDatasetName = values.outputDatasetName?.trim();
       const pipeline = selectedOperators.map((operator, index) => {
         const overrides = {
@@ -200,10 +190,10 @@ export default function AnnotationOperatorTaskCreate() {
                 label="选择数据集"
                 name="datasetId"
                 rules={[{ required: true, message: "请选择数据集" }]}
-                extra="自动标注算子编排当前仅支持图片数据集"
+                extra="请选择用于自动标注的数据集"
               >
                 <Select
-                  placeholder="请选择图片数据集"
+                  placeholder="请选择数据集"
                   optionFilterProp="label"
                   options={datasets.map((dataset) => ({
                     label: (
@@ -215,12 +205,11 @@ export default function AnnotationOperatorTaskCreate() {
                           {dataset.name}
                         </div>
                         <div className="text-xs text-gray-500">
-                          {dataset?.fileCount} 文件 • {dataset.size}
+                          {dataset.datasetType} &bull; {dataset?.fileCount} 文件 &bull; {dataset.size}
                         </div>
                       </div>
                     ),
                     value: dataset.id,
-                    disabled: dataset.datasetType !== DatasetType.IMAGE,
                   }))}
                 />
               </Form.Item>
diff --git a/runtime/datamate-python/app/db/models/annotation_management.py b/runtime/datamate-python/app/db/models/annotation_management.py
index 79ff1e5..96f2bef 100644
--- a/runtime/datamate-python/app/db/models/annotation_management.py
+++ b/runtime/datamate-python/app/db/models/annotation_management.py
@@ -197,7 +197,7 @@ class AnnotationResult(Base):
         return f"<AnnotationResult(id={self.id}, project_id={self.project_id}, file_id={self.file_id})>"
 
 
-class AutoAnnotationTask(Base):
+class AutoAnnotationTask(Base):
     """自动标注任务模型，对应表 t_dm_auto_annotation_tasks"""
 
     __tablename__ = "t_dm_auto_annotation_tasks"
@@ -206,94 +206,98 @@ class AutoAnnotationTask(Base):
         String(36), primary_key=True, default=lambda: str(uuid.uuid4()), comment="UUID"
     )
     name = Column(String(255), nullable=False, comment="任务名称")
-    dataset_id = Column(String(36), nullable=False, comment="数据集ID")
-    dataset_name = Column(
-        String(255), nullable=True, comment="数据集名称（冗余字段，方便查询）"
-    )
-    created_by = Column(String(255), nullable=True, comment="任务创建人")
-    config = Column(JSON, nullable=False, comment="任务配置（模型规模、置信度等）")
-    file_ids = Column(
-        JSON, nullable=True, comment="要处理的文件ID列表，为空则处理数据集所有图像"
-    )
-    status = Column(
-        String(50),
-        nullable=False,
-        default="pending",
-        comment="任务状态: pending/running/completed/failed/stopped",
-    )
-    task_mode = Column(
-        String(32),
-        nullable=False,
-        default="legacy_yolo",
-        comment="任务模式: legacy_yolo/pipeline",
-    )
-    executor_type = Column(
-        String(32),
-        nullable=False,
-        default="annotation_local",
-        comment="执行器类型",
-    )
-    pipeline = Column(JSON, nullable=True, comment="算子编排定义")
-    progress = Column(Integer, default=0, comment="任务进度 0-100")
-    stop_requested = Column(Boolean, default=False, comment="是否请求停止")
-    total_images = Column(Integer, default=0, comment="总图片数")
-    processed_images = Column(Integer, default=0, comment="已处理图片数")
-    detected_objects = Column(Integer, default=0, comment="检测到的对象总数")
-    output_path = Column(String(500), nullable=True, comment="输出路径")
-    output_dataset_id = Column(String(36), nullable=True, comment="输出数据集ID")
-    error_message = Column(Text, nullable=True, comment="错误信息")
-    created_at = Column(
-        TIMESTAMP, server_default=func.current_timestamp(), comment="创建时间"
-    )
+    dataset_id = Column(String(36), nullable=False, comment="数据集ID")
+    dataset_name = Column(
+        String(255), nullable=True, comment="数据集名称（冗余字段，方便查询）"
+    )
+    dataset_type = Column(
+        String(50), nullable=False, default="IMAGE",
+        comment="数据集类型: IMAGE/TEXT/AUDIO/VIDEO",
+    )
+    created_by = Column(String(255), nullable=True, comment="任务创建人")
+    config = Column(JSON, nullable=False, comment="任务配置（模型规模、置信度等）")
+    file_ids = Column(
+        JSON, nullable=True, comment="要处理的文件ID列表，为空则处理数据集所有图像"
+    )
+    status = Column(
+        String(50),
+        nullable=False,
+        default="pending",
+        comment="任务状态: pending/running/completed/failed/stopped",
+    )
+    task_mode = Column(
+        String(32),
+        nullable=False,
+        default="legacy_yolo",
+        comment="任务模式: legacy_yolo/pipeline",
+    )
+    executor_type = Column(
+        String(32),
+        nullable=False,
+        default="annotation_local",
+        comment="执行器类型",
+    )
+    pipeline = Column(JSON, nullable=True, comment="算子编排定义")
+    progress = Column(Integer, default=0, comment="任务进度 0-100")
+    stop_requested = Column(Boolean, default=False, comment="是否请求停止")
+    total_images = Column(Integer, default=0, comment="总图片数")
+    processed_images = Column(Integer, default=0, comment="已处理图片数")
+    detected_objects = Column(Integer, default=0, comment="检测到的对象总数")
+    output_path = Column(String(500), nullable=True, comment="输出路径")
+    output_dataset_id = Column(String(36), nullable=True, comment="输出数据集ID")
+    error_message = Column(Text, nullable=True, comment="错误信息")
+    created_at = Column(
+        TIMESTAMP, server_default=func.current_timestamp(), comment="创建时间"
+    )
     updated_at = Column(
         TIMESTAMP,
-        server_default=func.current_timestamp(),
-        onupdate=func.current_timestamp(),
-        comment="更新时间",
-    )
-    started_at = Column(TIMESTAMP, nullable=True, comment="任务启动时间")
-    heartbeat_at = Column(TIMESTAMP, nullable=True, comment="worker心跳时间")
-    run_token = Column(String(64), nullable=True, comment="运行令牌")
-    completed_at = Column(TIMESTAMP, nullable=True, comment="完成时间")
-    deleted_at = Column(TIMESTAMP, nullable=True, comment="删除时间（软删除）")
+        server_default=func.current_timestamp(),
+        onupdate=func.current_timestamp(),
+        comment="更新时间",
+    )
+    started_at = Column(TIMESTAMP, nullable=True, comment="任务启动时间")
+    heartbeat_at = Column(TIMESTAMP, nullable=True, comment="worker心跳时间")
+    run_token = Column(String(64), nullable=True, comment="运行令牌")
+    completed_at = Column(TIMESTAMP, nullable=True, comment="完成时间")
+    deleted_at = Column(TIMESTAMP, nullable=True, comment="删除时间（软删除）")
 
     def __repr__(self) -> str:  # pragma: no cover - repr 简单返回
         return f"<AutoAnnotationTask(id={self.id}, name={self.name}, status={self.status})>"
 
     @property
-    def is_deleted(self) -> bool:
-        """检查是否已被软删除"""
-        return self.deleted_at is not None
-
-
-class AnnotationTaskOperatorInstance(Base):
-    """自动标注任务内算子实例模型，对应表 t_dm_annotation_task_operator_instance"""
-
-    __tablename__ = "t_dm_annotation_task_operator_instance"
-
-    id = Column(BigInteger, primary_key=True, autoincrement=True, comment="自增主键")
-    task_id = Column(String(36), nullable=False, comment="自动标注任务ID")
-    op_index = Column(Integer, nullable=False, comment="算子顺序（从1开始）")
-    operator_id = Column(String(64), nullable=False, comment="算子ID（raw_id）")
-    settings_override = Column(JSON, nullable=True, comment="任务级算子参数覆盖")
-    inputs = Column(String(64), nullable=True, comment="输入模态")
-    outputs = Column(String(64), nullable=True, comment="输出模态")
-    created_at = Column(
-        TIMESTAMP,
-        server_default=func.current_timestamp(),
-        nullable=False,
-        comment="创建时间",
-    )
-    updated_at = Column(
-        TIMESTAMP,
-        server_default=func.current_timestamp(),
-        onupdate=func.current_timestamp(),
-        nullable=False,
-        comment="更新时间",
-    )
-
-    __table_args__ = (
-        UniqueConstraint("task_id", "op_index", name="uk_task_op_index"),
-        Index("idx_task_id", "task_id"),
-        Index("idx_operator_id", "operator_id"),
-    )
+    def is_deleted(self) -> bool:
+        """检查是否已被软删除"""
+        return self.deleted_at is not None
+
+
+class AnnotationTaskOperatorInstance(Base):
+    """自动标注任务内算子实例模型，对应表 t_dm_annotation_task_operator_instance"""
+
+    __tablename__ = "t_dm_annotation_task_operator_instance"
+
+    id = Column(BigInteger, primary_key=True, autoincrement=True, comment="自增主键")
+    task_id = Column(String(36), nullable=False, comment="自动标注任务ID")
+    op_index = Column(Integer, nullable=False, comment="算子顺序（从1开始）")
+    operator_id = Column(String(64), nullable=False, comment="算子ID（raw_id）")
+    settings_override = Column(JSON, nullable=True, comment="任务级算子参数覆盖")
+    inputs = Column(String(64), nullable=True, comment="输入模态")
+    outputs = Column(String(64), nullable=True, comment="输出模态")
+    created_at = Column(
+        TIMESTAMP,
+        server_default=func.current_timestamp(),
+        nullable=False,
+        comment="创建时间",
+    )
+    updated_at = Column(
+        TIMESTAMP,
+        server_default=func.current_timestamp(),
+        onupdate=func.current_timestamp(),
+        nullable=False,
+        comment="更新时间",
+    )
+
+    __table_args__ = (
+        UniqueConstraint("task_id", "op_index", name="uk_task_op_index"),
+        Index("idx_task_id", "task_id"),
+        Index("idx_operator_id", "operator_id"),
+    )
diff --git a/runtime/datamate-python/app/module/annotation/interface/auto.py b/runtime/datamate-python/app/module/annotation/interface/auto.py
index 60f3b43..e2b78e5 100644
--- a/runtime/datamate-python/app/module/annotation/interface/auto.py
+++ b/runtime/datamate-python/app/module/annotation/interface/auto.py
@@ -15,9 +15,9 @@ from typing import List, Literal
 from fastapi import APIRouter, Depends, HTTPException, Path
 from fastapi.responses import StreamingResponse
 from sqlalchemy.ext.asyncio import AsyncSession
-
-from app.db.session import get_db
-from app.module.shared.schema import StandardResponse
+
+from app.db.session import get_db
+from app.module.shared.schema import StandardResponse
 from app.module.dataset import DatasetManagementService
 from app.core.logging import get_logger
 
@@ -29,15 +29,15 @@ from ..security import (
 from ..schema.auto import (
     CreateAutoAnnotationTaskRequest,
     AutoAnnotationTaskResponse,
-)
+)
 from ..service.auto import AutoAnnotationTaskService
 
 
 router = APIRouter(
     tags=["annotation/auto"],
 )
-
-logger = get_logger(__name__)
+
+logger = get_logger(__name__)
 service = AutoAnnotationTaskService()
 
 
@@ -85,23 +85,28 @@ async def _create_task_internal(
     await assert_dataset_access(db, normalized_request.dataset_id, user_context)
     # 尝试获取数据集名称和总量用于冗余字段
     dataset_name = None
+    dataset_type = "IMAGE"
     total_images = len(normalized_request.file_ids) if normalized_request.file_ids else 0
     try:
         dm_client = DatasetManagementService(db)
         dataset = await dm_client.get_dataset(normalized_request.dataset_id)
         if dataset is not None:
             dataset_name = dataset.name
+            dataset_type = getattr(dataset, "datasetType", None) or "IMAGE"
             if not normalized_request.file_ids:
                 total_images = getattr(dataset, "fileCount", 0) or 0
     except Exception as e:  # pragma: no cover - 容错
         logger.warning("Failed to fetch dataset summary for annotation task: %s", e)
 
+    resolved_dataset_type = normalized_request.dataset_type or dataset_type
+
     return await service.create_task(
         db,
         normalized_request,
         user_context=user_context,
         dataset_name=dataset_name,
         total_images=total_images,
+        dataset_type=resolved_dataset_type,
     )
 
 
@@ -177,10 +182,10 @@ async def get_auto_annotation_task_status(
     task = await service.get_task(db, task_id, user_context)
     if not task:
         raise HTTPException(status_code=404, detail="Task not found")
-
-    return StandardResponse(
-        code=200,
-        message="success",
+
+    return StandardResponse(
+        code=200,
+        message="success",
         data=task,
     )
 
@@ -192,12 +197,12 @@ async def delete_auto_annotation_task(
     db: AsyncSession = Depends(get_db),
     user_context: RequestUserContext = Depends(get_request_user_context),
 ):
-    """删除（软删除）自动标注任务，仅标记 deleted_at。"""
-
+    """删除（软删除）自动标注任务，仅标记 deleted_at。"""
+
     ok = await service.soft_delete_task(db, task_id, user_context)
-    if not ok:
-        raise HTTPException(status_code=404, detail="Task not found")
-
+    if not ok:
+        raise HTTPException(status_code=404, detail="Task not found")
+
     return StandardResponse(
         code=200,
         message="success",
@@ -232,50 +237,50 @@ async def download_auto_annotation_result(
     db: AsyncSession = Depends(get_db),
     user_context: RequestUserContext = Depends(get_request_user_context),
 ):
-    """下载指定自动标注任务的结果 ZIP。"""
-
+    """下载指定自动标注任务的结果 ZIP。"""
+
     import os
     import zipfile
     import tempfile
-
-    # 复用服务层获取任务信息
+
+    # 复用服务层获取任务信息
     task = await service.get_task(db, task_id, user_context)
-    if not task:
-        raise HTTPException(status_code=404, detail="Task not found")
-
-    if not task.output_path:
-        raise HTTPException(status_code=400, detail="Task has no output path")
-
-    output_dir = task.output_path
-    if not os.path.isdir(output_dir):
-        raise HTTPException(status_code=404, detail="Output directory not found")
-
-    tmp_fd, tmp_path = tempfile.mkstemp(suffix=".zip")
-    os.close(tmp_fd)
-
-    with zipfile.ZipFile(tmp_path, "w", zipfile.ZIP_DEFLATED) as zf:
-        for root, _, files in os.walk(output_dir):
-            for filename in files:
-                file_path = os.path.join(root, filename)
-                arcname = os.path.relpath(file_path, output_dir)
-                zf.write(file_path, arcname)
-
-    file_size = os.path.getsize(tmp_path)
-    if file_size == 0:
-        raise HTTPException(status_code=500, detail="Generated ZIP is empty")
-
-    def iterfile():
-        with open(tmp_path, "rb") as f:
-            while True:
-                chunk = f.read(8192)
-                if not chunk:
-                    break
-                yield chunk
-
-    filename = f"{task.name}_annotations.zip"
-    headers = {
-        "Content-Disposition": f'attachment; filename="{filename}"',
-        "Content-Length": str(file_size),
-    }
-
-    return StreamingResponse(iterfile(), media_type="application/zip", headers=headers)
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found")
+
+    if not task.output_path:
+        raise HTTPException(status_code=400, detail="Task has no output path")
+
+    output_dir = task.output_path
+    if not os.path.isdir(output_dir):
+        raise HTTPException(status_code=404, detail="Output directory not found")
+
+    tmp_fd, tmp_path = tempfile.mkstemp(suffix=".zip")
+    os.close(tmp_fd)
+
+    with zipfile.ZipFile(tmp_path, "w", zipfile.ZIP_DEFLATED) as zf:
+        for root, _, files in os.walk(output_dir):
+            for filename in files:
+                file_path = os.path.join(root, filename)
+                arcname = os.path.relpath(file_path, output_dir)
+                zf.write(file_path, arcname)
+
+    file_size = os.path.getsize(tmp_path)
+    if file_size == 0:
+        raise HTTPException(status_code=500, detail="Generated ZIP is empty")
+
+    def iterfile():
+        with open(tmp_path, "rb") as f:
+            while True:
+                chunk = f.read(8192)
+                if not chunk:
+                    break
+                yield chunk
+
+    filename = f"{task.name}_annotations.zip"
+    headers = {
+        "Content-Disposition": f'attachment; filename="{filename}"',
+        "Content-Length": str(file_size),
+    }
+
+    return StreamingResponse(iterfile(), media_type="application/zip", headers=headers)
diff --git a/runtime/datamate-python/app/module/annotation/schema/auto.py b/runtime/datamate-python/app/module/annotation/schema/auto.py
index b403fdc..a8117d9 100644
--- a/runtime/datamate-python/app/module/annotation/schema/auto.py
+++ b/runtime/datamate-python/app/module/annotation/schema/auto.py
@@ -1,4 +1,4 @@
-"""Schemas for Auto Annotation tasks"""
+"""Schemas for Auto Annotation tasks"""
 from __future__ import annotations
 
 import json
@@ -7,24 +7,24 @@ from typing import List, Optional, Dict, Any
 from datetime import datetime
 
 from pydantic import BaseModel, Field, ConfigDict, model_validator
-
-
+
+
 class AutoAnnotationConfig(BaseModel):
-    """自动标注任务配置（与前端 payload 对齐）"""
-
-    model_size: str = Field(alias="modelSize", description="模型规模: n/s/m/l/x")
-    conf_threshold: float = Field(alias="confThreshold", description="置信度阈值 0-1")
-    target_classes: List[int] = Field(
-        default_factory=list,
-        alias="targetClasses",
-        description="目标类别ID列表，空表示全部类别",
-    )
-    output_dataset_name: Optional[str] = Field(
-        default=None,
-        alias="outputDatasetName",
-        description="自动标注结果要写入的新数据集名称（可选）",
-    )
-
+    """自动标注任务配置（与前端 payload 对齐）"""
+
+    model_size: str = Field(alias="modelSize", description="模型规模: n/s/m/l/x")
+    conf_threshold: float = Field(alias="confThreshold", description="置信度阈值 0-1")
+    target_classes: List[int] = Field(
+        default_factory=list,
+        alias="targetClasses",
+        description="目标类别ID列表，空表示全部类别",
+    )
+    output_dataset_name: Optional[str] = Field(
+        default=None,
+        alias="outputDatasetName",
+        description="自动标注结果要写入的新数据集名称（可选）",
+    )
+
     model_config = ConfigDict(populate_by_name=True)
 
 
@@ -68,13 +68,18 @@ class OperatorPipelineStep(BaseModel):
         return normalized
 
     model_config = ConfigDict(populate_by_name=True)
-
-
+
+
 class CreateAutoAnnotationTaskRequest(BaseModel):
     """创建自动标注任务的请求体，对齐前端 CreateAutoAnnotationDialog 发送的结构"""
 
     name: str = Field(..., min_length=1, max_length=255, description="任务名称")
     dataset_id: str = Field(..., alias="datasetId", description="数据集ID")
+    dataset_type: Optional[str] = Field(
+        default=None,
+        alias="datasetType",
+        description="数据集类型: IMAGE/TEXT/AUDIO/VIDEO（不传时由后端自动获取）",
+    )
     config: Optional[AutoAnnotationConfig] = Field(
         default=None,
         description="兼容旧版 YOLO 任务配置",
@@ -111,15 +116,16 @@ class CreateAutoAnnotationTaskRequest(BaseModel):
         return self
 
     model_config = ConfigDict(populate_by_name=True)
-
-
+
+
 class AutoAnnotationTaskResponse(BaseModel):
-    """自动标注任务响应模型（列表/详情均可复用）"""
-
-    id: str = Field(..., description="任务ID")
-    name: str = Field(..., description="任务名称")
-    dataset_id: str = Field(..., alias="datasetId", description="数据集ID")
+    """自动标注任务响应模型（列表/详情均可复用）"""
+
+    id: str = Field(..., description="任务ID")
+    name: str = Field(..., description="任务名称")
+    dataset_id: str = Field(..., alias="datasetId", description="数据集ID")
     dataset_name: Optional[str] = Field(None, alias="datasetName", description="数据集名称")
+    dataset_type: Optional[str] = Field(None, alias="datasetType", description="数据集类型: IMAGE/TEXT/AUDIO/VIDEO")
     task_mode: Optional[str] = Field(None, alias="taskMode", description="任务模式")
     executor_type: Optional[str] = Field(None, alias="executorType", description="执行器类型")
     pipeline: Optional[List[Dict[str, Any]]] = Field(None, description="算子编排定义")
@@ -128,11 +134,11 @@ class AutoAnnotationTaskResponse(BaseModel):
         alias="sourceDatasets",
         description="本任务实际处理涉及到的所有数据集名称列表",
     )
-    config: Dict[str, Any] = Field(..., description="任务配置")
-    status: str = Field(..., description="任务状态")
-    progress: int = Field(..., description="任务进度 0-100")
-    total_images: int = Field(..., alias="totalImages", description="总图片数")
-    processed_images: int = Field(..., alias="processedImages", description="已处理图片数")
+    config: Dict[str, Any] = Field(..., description="任务配置")
+    status: str = Field(..., description="任务状态")
+    progress: int = Field(..., description="任务进度 0-100")
+    total_images: int = Field(..., alias="totalImages", description="总图片数")
+    processed_images: int = Field(..., alias="processedImages", description="已处理图片数")
     detected_objects: int = Field(..., alias="detectedObjects", description="检测到的对象总数")
     output_path: Optional[str] = Field(None, alias="outputPath", description="输出路径")
     output_dataset_id: Optional[str] = Field(
@@ -152,14 +158,14 @@ class AutoAnnotationTaskResponse(BaseModel):
     created_at: datetime = Field(..., alias="createdAt", description="创建时间")
     updated_at: Optional[datetime] = Field(None, alias="updatedAt", description="更新时间")
     completed_at: Optional[datetime] = Field(None, alias="completedAt", description="完成时间")
-
-    model_config = ConfigDict(populate_by_name=True, from_attributes=True)
-
-
-class AutoAnnotationTaskListResponse(BaseModel):
-    """自动标注任务列表响应，目前前端直接使用数组，这里预留分页结构"""
-
-    content: List[AutoAnnotationTaskResponse] = Field(..., description="任务列表")
-    total: int = Field(..., description="总数")
-
-    model_config = ConfigDict(populate_by_name=True)
+
+    model_config = ConfigDict(populate_by_name=True, from_attributes=True)
+
+
+class AutoAnnotationTaskListResponse(BaseModel):
+    """自动标注任务列表响应，目前前端直接使用数组，这里预留分页结构"""
+
+    content: List[AutoAnnotationTaskResponse] = Field(..., description="任务列表")
+    total: int = Field(..., description="总数")
+
+    model_config = ConfigDict(populate_by_name=True)
diff --git a/runtime/datamate-python/app/module/annotation/service/auto.py b/runtime/datamate-python/app/module/annotation/service/auto.py
index 970321c..2220908 100644
--- a/runtime/datamate-python/app/module/annotation/service/auto.py
+++ b/runtime/datamate-python/app/module/annotation/service/auto.py
@@ -15,13 +15,13 @@ from app.db.models.annotation_management import (
 )
 from app.db.models.dataset_management import Dataset, DatasetFiles
 from app.module.annotation.security import RequestUserContext
-
-from ..schema.auto import (
-    CreateAutoAnnotationTaskRequest,
-    AutoAnnotationTaskResponse,
-)
-
-
+
+from ..schema.auto import (
+    CreateAutoAnnotationTaskRequest,
+    AutoAnnotationTaskResponse,
+)
+
+
 class AutoAnnotationTaskService:
     """自动标注任务服务（仅管理任务元数据，真正执行由 runtime 负责）"""
 
@@ -141,11 +141,12 @@ class AutoAnnotationTaskService:
         user_context: RequestUserContext,
         dataset_name: Optional[str] = None,
         total_images: int = 0,
+        dataset_type: str = "IMAGE",
     ) -> AutoAnnotationTaskResponse:
-        """创建自动标注任务，初始状态为 pending。
-
-        这里仅插入任务记录，不负责真正执行 YOLO 推理，
-        后续可以由调度器/worker 读取该表并更新进度。
+        """创建自动标注任务，初始状态为 pending。
+
+        这里仅插入任务记录，不负责真正执行 YOLO 推理，
+        后续可以由调度器/worker 读取该表并更新进度。
         """
 
         now = datetime.now()
@@ -170,6 +171,7 @@ class AutoAnnotationTaskService:
             name=request.name,
             dataset_id=request.dataset_id,
             dataset_name=dataset_name,
+            dataset_type=dataset_type,
             created_by=user_context.user_id,
             config=normalized_config,
             task_mode=request.task_mode,
@@ -192,15 +194,15 @@ class AutoAnnotationTaskService:
             db.add_all(operator_instances)
         await db.commit()
         await db.refresh(task)
-
-        # 创建后附带 sourceDatasets 信息（通常只有一个原始数据集）
-        resp = AutoAnnotationTaskResponse.model_validate(task)
-        try:
-            resp.source_datasets = await self._compute_source_datasets(db, task)
-        except Exception:
-            resp.source_datasets = [dataset_name] if dataset_name else [request.dataset_id]
-        return resp
-
+
+        # 创建后附带 sourceDatasets 信息（通常只有一个原始数据集）
+        resp = AutoAnnotationTaskResponse.model_validate(task)
+        try:
+            resp.source_datasets = await self._compute_source_datasets(db, task)
+        except Exception:
+            resp.source_datasets = [dataset_name] if dataset_name else [request.dataset_id]
+        return resp
+
     def _apply_dataset_scope(self, query, user_context: RequestUserContext):
         if user_context.is_admin:
             return query
@@ -222,21 +224,21 @@ class AutoAnnotationTaskService:
             query.order_by(AutoAnnotationTask.created_at.desc())
         )
         tasks: List[AutoAnnotationTask] = list(result.scalars().all())
-
-        responses: List[AutoAnnotationTaskResponse] = []
-        for task in tasks:
-            resp = AutoAnnotationTaskResponse.model_validate(task)
-            try:
-                resp.source_datasets = await self._compute_source_datasets(db, task)
-            except Exception:
-                # 出错时降级为单个 datasetName/datasetId
-                fallback_name = getattr(task, "dataset_name", None)
-                fallback_id = getattr(task, "dataset_id", "")
-                resp.source_datasets = [fallback_name] if fallback_name else [fallback_id]
-            responses.append(resp)
-
-        return responses
-
+
+        responses: List[AutoAnnotationTaskResponse] = []
+        for task in tasks:
+            resp = AutoAnnotationTaskResponse.model_validate(task)
+            try:
+                resp.source_datasets = await self._compute_source_datasets(db, task)
+            except Exception:
+                # 出错时降级为单个 datasetName/datasetId
+                fallback_name = getattr(task, "dataset_name", None)
+                fallback_id = getattr(task, "dataset_id", "")
+                resp.source_datasets = [fallback_name] if fallback_name else [fallback_id]
+            responses.append(resp)
+
+        return responses
+
     async def get_task(
         self,
         db: AsyncSession,
@@ -252,43 +254,43 @@ class AutoAnnotationTaskService:
         task = result.scalar_one_or_none()
         if not task:
             return None
-
-        resp = AutoAnnotationTaskResponse.model_validate(task)
-        try:
-            resp.source_datasets = await self._compute_source_datasets(db, task)
-        except Exception:
-            fallback_name = getattr(task, "dataset_name", None)
-            fallback_id = getattr(task, "dataset_id", "")
-            resp.source_datasets = [fallback_name] if fallback_name else [fallback_id]
-        return resp
-
-    async def _compute_source_datasets(
-        self,
-        db: AsyncSession,
-        task: AutoAnnotationTask,
-    ) -> List[str]:
-        """根据任务的 file_ids 推断实际涉及到的所有数据集名称。
-
-        - 如果存在 file_ids，则通过 t_dm_dataset_files 反查 dataset_id，再关联 t_dm_datasets 获取名称；
-        - 如果没有 file_ids，则退回到任务上冗余的 dataset_name/dataset_id。
-        """
-
-        file_ids = task.file_ids or []
-        if file_ids:
-            stmt = (
-                select(Dataset.name)
-                .join(DatasetFiles, Dataset.id == DatasetFiles.dataset_id)
-                .where(DatasetFiles.id.in_(file_ids))
-                .distinct()
-            )
-            result = await db.execute(stmt)
-            names = [row[0] for row in result.fetchall() if row[0]]
-            if names:
-                return names
-
-        # 回退：只显示一个数据集
-        if task.dataset_name:
-            return [task.dataset_name]
+
+        resp = AutoAnnotationTaskResponse.model_validate(task)
+        try:
+            resp.source_datasets = await self._compute_source_datasets(db, task)
+        except Exception:
+            fallback_name = getattr(task, "dataset_name", None)
+            fallback_id = getattr(task, "dataset_id", "")
+            resp.source_datasets = [fallback_name] if fallback_name else [fallback_id]
+        return resp
+
+    async def _compute_source_datasets(
+        self,
+        db: AsyncSession,
+        task: AutoAnnotationTask,
+    ) -> List[str]:
+        """根据任务的 file_ids 推断实际涉及到的所有数据集名称。
+
+        - 如果存在 file_ids，则通过 t_dm_dataset_files 反查 dataset_id，再关联 t_dm_datasets 获取名称；
+        - 如果没有 file_ids，则退回到任务上冗余的 dataset_name/dataset_id。
+        """
+
+        file_ids = task.file_ids or []
+        if file_ids:
+            stmt = (
+                select(Dataset.name)
+                .join(DatasetFiles, Dataset.id == DatasetFiles.dataset_id)
+                .where(DatasetFiles.id.in_(file_ids))
+                .distinct()
+            )
+            result = await db.execute(stmt)
+            names = [row[0] for row in result.fetchall() if row[0]]
+            if names:
+                return names
+
+        # 回退：只显示一个数据集
+        if task.dataset_name:
+            return [task.dataset_name]
         if task.dataset_id:
             return [task.dataset_id]
         return []
@@ -331,7 +333,7 @@ class AutoAnnotationTaskService:
             fallback_id = getattr(task, "dataset_id", "")
             resp.source_datasets = [fallback_name] if fallback_name else [fallback_id]
         return resp
-
+
     async def soft_delete_task(
         self,
         db: AsyncSession,
@@ -347,7 +349,7 @@ class AutoAnnotationTaskService:
         task = result.scalar_one_or_none()
         if not task:
             return False
-
-        task.deleted_at = datetime.now()
-        await db.commit()
-        return True
+
+        task.deleted_at = datetime.now()
+        await db.commit()
+        return True
diff --git a/runtime/python-executor/datamate/auto_annotation_worker.py b/runtime/python-executor/datamate/auto_annotation_worker.py
index 59ab7a1..fd4a613 100644
--- a/runtime/python-executor/datamate/auto_annotation_worker.py
+++ b/runtime/python-executor/datamate/auto_annotation_worker.py
@@ -17,41 +17,41 @@ frontend can display real-time status.
 - 为了保持简单，目前不处理 "running" 状态的恢复逻辑；容器重启时，
   已处于 running 的任务不会被重新拉起，需要后续扩展。
 """
-from __future__ import annotations
-
-import importlib
-import json
-import os
-import sys
-import threading
-import time
+from __future__ import annotations
+
+import importlib
+import json
+import os
+import sys
+import threading
+import time
 import uuid
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
-from loguru import logger
-from sqlalchemy import text
-
-try:
-    import datamate.ops  # noqa: F401
-except Exception as import_ops_err:  # pragma: no cover - 兜底日志
-    logger.warning("Failed to import datamate.ops package for operator registry: {}", import_ops_err)
-
-try:
-    import ops.annotation  # type: ignore # noqa: F401
-except Exception as import_annotation_ops_err:  # pragma: no cover - 兜底日志
-    logger.warning(
-        "Failed to import ops.annotation package for operator registry: {}",
-        import_annotation_ops_err,
-    )
-
-try:
-    from datamate.core.base_op import OPERATORS
-except Exception:  # pragma: no cover - 兜底
-    OPERATORS = None  # type: ignore
-
-from datamate.sql_manager.sql_manager import SQLManager
+from loguru import logger
+from sqlalchemy import text
+
+try:
+    import datamate.ops  # noqa: F401
+except Exception as import_ops_err:  # pragma: no cover - 兜底日志
+    logger.warning("Failed to import datamate.ops package for operator registry: {}", import_ops_err)
+
+try:
+    import ops.annotation  # type: ignore # noqa: F401
+except Exception as import_annotation_ops_err:  # pragma: no cover - 兜底日志
+    logger.warning(
+        "Failed to import ops.annotation package for operator registry: {}",
+        import_annotation_ops_err,
+    )
+
+try:
+    from datamate.core.base_op import OPERATORS
+except Exception:  # pragma: no cover - 兜底
+    OPERATORS = None  # type: ignore
+
+from datamate.sql_manager.sql_manager import SQLManager
 
 # 尝试多种导入路径，适配不同的打包/安装方式
 ImageObjectDetectionBoundingBox = None  # type: ignore
@@ -116,114 +116,114 @@ if ImageObjectDetectionBoundingBox is None:
 
 POLL_INTERVAL_SECONDS = float(os.getenv("AUTO_ANNOTATION_POLL_INTERVAL", "5"))
 
-DEFAULT_OUTPUT_ROOT = os.getenv(
-    "AUTO_ANNOTATION_OUTPUT_ROOT", "/dataset"
-)
-
-DEFAULT_OPERATOR_WHITELIST = os.getenv(
-    "AUTO_ANNOTATION_OPERATOR_WHITELIST",
-    "ImageObjectDetectionBoundingBox,test_annotation_marker",
-)
+DEFAULT_OUTPUT_ROOT = os.getenv(
+    "AUTO_ANNOTATION_OUTPUT_ROOT", "/dataset"
+)
+
+DEFAULT_OPERATOR_WHITELIST = os.getenv(
+    "AUTO_ANNOTATION_OPERATOR_WHITELIST",
+    "ImageObjectDetectionBoundingBox,test_annotation_marker",
+)
 
 
-def _fetch_pending_task() -> Optional[Dict[str, Any]]:
-    """原子 claim 一个 pending 任务并返回任务详情。"""
-
-    def _parse_json_field(value: Any, default: Any) -> Any:
-        if value is None:
-            return default
-        if isinstance(value, (dict, list)):
-            return value
-        if isinstance(value, str):
-            text_value = value.strip()
-            if not text_value:
-                return default
-            try:
-                return json.loads(text_value)
-            except Exception:
-                return default
-        return default
-
-    run_token = str(uuid.uuid4())
-    now = datetime.now()
-
-    claim_sql = text(
-        """
-        UPDATE t_dm_auto_annotation_tasks
-        SET status = 'running',
-            run_token = :run_token,
-            started_at = COALESCE(started_at, :now),
-            heartbeat_at = :now,
-            updated_at = :now,
-            error_message = NULL
-        WHERE id = (
-            SELECT id FROM (
-                SELECT id
-                FROM t_dm_auto_annotation_tasks
-                WHERE status = 'pending'
-                  AND deleted_at IS NULL
-                  AND COALESCE(stop_requested, 0) = 0
-                ORDER BY created_at ASC
-                LIMIT 1
-            ) AS pending_task
-        )
-          AND status = 'pending'
-          AND deleted_at IS NULL
-          AND COALESCE(stop_requested, 0) = 0
-        """
-    )
-
-    query_sql = text(
-        """
-        SELECT id, name, dataset_id, dataset_name, created_by,
-               config, file_ids, pipeline,
-               task_mode, executor_type,
-               status, stop_requested, run_token,
-               total_images, processed_images, detected_objects,
-               output_path, output_dataset_id
-        FROM t_dm_auto_annotation_tasks
-        WHERE run_token = :run_token
-        LIMIT 1
-        """
-    )
-
-    with SQLManager.create_connect() as conn:
-        claim_result = conn.execute(claim_sql, {"run_token": run_token, "now": now})
-        if not claim_result or int(getattr(claim_result, "rowcount", 0) or 0) <= 0:
-            return None
-
-        result = conn.execute(query_sql, {"run_token": run_token}).fetchone()
-        if not result:
-            return None
-
-        row = dict(result._mapping)  # type: ignore[attr-defined]
-
-    row["config"] = _parse_json_field(row.get("config"), {})
-
-    parsed_file_ids = _parse_json_field(row.get("file_ids"), None)
-    row["file_ids"] = parsed_file_ids if parsed_file_ids else None
-
-    parsed_pipeline = _parse_json_field(row.get("pipeline"), None)
-    row["pipeline"] = parsed_pipeline if parsed_pipeline else None
-    return row
+def _fetch_pending_task() -> Optional[Dict[str, Any]]:
+    """原子 claim 一个 pending 任务并返回任务详情。"""
+
+    def _parse_json_field(value: Any, default: Any) -> Any:
+        if value is None:
+            return default
+        if isinstance(value, (dict, list)):
+            return value
+        if isinstance(value, str):
+            text_value = value.strip()
+            if not text_value:
+                return default
+            try:
+                return json.loads(text_value)
+            except Exception:
+                return default
+        return default
+
+    run_token = str(uuid.uuid4())
+    now = datetime.now()
+
+    claim_sql = text(
+        """
+        UPDATE t_dm_auto_annotation_tasks
+        SET status = 'running',
+            run_token = :run_token,
+            started_at = COALESCE(started_at, :now),
+            heartbeat_at = :now,
+            updated_at = :now,
+            error_message = NULL
+        WHERE id = (
+            SELECT id FROM (
+                SELECT id
+                FROM t_dm_auto_annotation_tasks
+                WHERE status = 'pending'
+                  AND deleted_at IS NULL
+                  AND COALESCE(stop_requested, 0) = 0
+                ORDER BY created_at ASC
+                LIMIT 1
+            ) AS pending_task
+        )
+          AND status = 'pending'
+          AND deleted_at IS NULL
+          AND COALESCE(stop_requested, 0) = 0
+        """
+    )
+
+    query_sql = text(
+        """
+        SELECT id, name, dataset_id, dataset_name, dataset_type, created_by,
+               config, file_ids, pipeline,
+               task_mode, executor_type,
+               status, stop_requested, run_token,
+               total_images, processed_images, detected_objects,
+               output_path, output_dataset_id
+        FROM t_dm_auto_annotation_tasks
+        WHERE run_token = :run_token
+        LIMIT 1
+        """
+    )
+
+    with SQLManager.create_connect() as conn:
+        claim_result = conn.execute(claim_sql, {"run_token": run_token, "now": now})
+        if not claim_result or int(getattr(claim_result, "rowcount", 0) or 0) <= 0:
+            return None
+
+        result = conn.execute(query_sql, {"run_token": run_token}).fetchone()
+        if not result:
+            return None
+
+        row = dict(result._mapping)  # type: ignore[attr-defined]
+
+    row["config"] = _parse_json_field(row.get("config"), {})
+
+    parsed_file_ids = _parse_json_field(row.get("file_ids"), None)
+    row["file_ids"] = parsed_file_ids if parsed_file_ids else None
+
+    parsed_pipeline = _parse_json_field(row.get("pipeline"), None)
+    row["pipeline"] = parsed_pipeline if parsed_pipeline else None
+    return row
 
 
-def _update_task_status(
-    task_id: str,
-    *,
-    status: str,
-    run_token: Optional[str] = None,
-    progress: Optional[int] = None,
-    processed_images: Optional[int] = None,
-    detected_objects: Optional[int] = None,
-    total_images: Optional[int] = None,
-    output_path: Optional[str] = None,
-    output_dataset_id: Optional[str] = None,
-    error_message: Optional[str] = None,
-    completed: bool = False,
-    clear_run_token: bool = False,
-) -> None:
-    """更新任务的状态和统计字段。"""
+def _update_task_status(
+    task_id: str,
+    *,
+    status: str,
+    run_token: Optional[str] = None,
+    progress: Optional[int] = None,
+    processed_images: Optional[int] = None,
+    detected_objects: Optional[int] = None,
+    total_images: Optional[int] = None,
+    output_path: Optional[str] = None,
+    output_dataset_id: Optional[str] = None,
+    error_message: Optional[str] = None,
+    completed: bool = False,
+    clear_run_token: bool = False,
+) -> None:
+    """更新任务的状态和统计字段。"""
 
     fields: List[str] = ["status = :status", "updated_at = :updated_at"]
     params: Dict[str, Any] = {
@@ -244,325 +244,354 @@ def _update_task_status(
     if total_images is not None:
         fields.append("total_images = :total_images")
         params["total_images"] = int(total_images)
-    if output_path is not None:
-        fields.append("output_path = :output_path")
-        params["output_path"] = output_path
-    if output_dataset_id is not None:
-        fields.append("output_dataset_id = :output_dataset_id")
-        params["output_dataset_id"] = output_dataset_id
-    if error_message is not None:
-        fields.append("error_message = :error_message")
-        params["error_message"] = error_message[:2000]
-    if status == "running":
-        fields.append("heartbeat_at = :heartbeat_at")
-        params["heartbeat_at"] = datetime.now()
-    if completed:
-        fields.append("completed_at = :completed_at")
-        params["completed_at"] = datetime.now()
-    if clear_run_token:
-        fields.append("run_token = NULL")
-
-    where_clause = "id = :task_id"
-    if run_token:
-        where_clause += " AND run_token = :run_token"
-        params["run_token"] = run_token
-
-    sql = text(
-        f"""
-        UPDATE t_dm_auto_annotation_tasks
-        SET {', '.join(fields)}
-        WHERE {where_clause}
-        """
-    )
-
-    with SQLManager.create_connect() as conn:
-        result = conn.execute(sql, params)
-        if int(getattr(result, "rowcount", 0) or 0) <= 0:
-            logger.warning(
-                "No rows updated for task status change: task_id={}, status={}, run_token={}",
-                task_id,
-                status,
-                run_token,
-            )
-
-
-def _is_stop_requested(task_id: str, run_token: Optional[str] = None) -> bool:
-    """检查任务是否请求停止。"""
-
-    where_clause = "id = :task_id"
-    params: Dict[str, Any] = {"task_id": task_id}
-    if run_token:
-        where_clause += " AND run_token = :run_token"
-        params["run_token"] = run_token
-
-    sql = text(
-        f"""
-        SELECT COALESCE(stop_requested, 0)
-        FROM t_dm_auto_annotation_tasks
-        WHERE {where_clause}
-        LIMIT 1
-        """
-    )
-
-    with SQLManager.create_connect() as conn:
-        row = conn.execute(sql, params).fetchone()
-        if not row:
-            # 找不到任务（或 run_token 已失效）时保守停止
-            return True
-        return bool(row[0])
-
-
-def _extract_step_overrides(step: Dict[str, Any]) -> Dict[str, Any]:
-    """合并 pipeline 节点中的参数覆盖。"""
-
-    overrides: Dict[str, Any] = {}
-    for key in ("overrides", "settingsOverride", "settings_override"):
-        value = step.get(key)
-        if value is None:
-            continue
-        if isinstance(value, str):
-            try:
-                value = json.loads(value)
-            except Exception:
-                continue
-        if isinstance(value, dict):
-            overrides.update(value)
-    return overrides
-
-
-def _build_legacy_pipeline(config: Dict[str, Any]) -> List[Dict[str, Any]]:
-    """将 legacy_yolo 配置映射为单步 pipeline。"""
-
-    return [
-        {
-            "operatorId": "ImageObjectDetectionBoundingBox",
-            "overrides": {
-                "modelSize": config.get("modelSize", "l"),
-                "confThreshold": float(config.get("confThreshold", 0.7)),
-                "targetClasses": config.get("targetClasses", []) or [],
-            },
-        }
-    ]
-
-
-def _get_output_dataset_name(
-    task_id: str,
-    dataset_id: str,
-    source_dataset_name: str,
-    task_name: str,
-    config: Dict[str, Any],
-    pipeline_raw: Optional[List[Any]],
-) -> str:
-    """确定输出数据集名称。"""
-
-    output_name = config.get("outputDatasetName")
-    if output_name:
-        return str(output_name)
-
-    if pipeline_raw:
-        for step in pipeline_raw:
-            if not isinstance(step, dict):
-                continue
-            overrides = _extract_step_overrides(step)
-            output_name = overrides.get("outputDatasetName") or overrides.get("output_dataset_name")
-            if output_name:
-                return str(output_name)
-
-    base_name = source_dataset_name or task_name or f"dataset-{dataset_id[:8]}"
-    return f"{base_name}_auto_{task_id[:8]}"
-
-
-def _normalize_pipeline(
-    task_mode: str,
-    config: Dict[str, Any],
-    pipeline_raw: Optional[List[Any]],
-    output_dir: str,
-) -> List[Dict[str, Any]]:
-    """标准化 pipeline 结构并注入 outputDir。"""
-
-    source_pipeline = pipeline_raw
-    if task_mode == "legacy_yolo" or not source_pipeline:
-        source_pipeline = _build_legacy_pipeline(config)
-
-    normalized: List[Dict[str, Any]] = []
-    for step in source_pipeline:
-        if not isinstance(step, dict):
-            continue
-
-        operator_id: Optional[str] = None
-        overrides: Dict[str, Any] = {}
-
-        # 兼容 [{"OpName": {...}}] 风格
-        if (
-            "operatorId" not in step
-            and "operator_id" not in step
-            and "id" not in step
-            and len(step) == 1
-        ):
-            first_key = next(iter(step.keys()))
-            first_value = step.get(first_key)
-            if isinstance(first_key, str):
-                operator_id = first_key
-            if isinstance(first_value, dict):
-                overrides.update(first_value)
-
-        operator_id = operator_id or step.get("operatorId") or step.get("operator_id") or step.get("id")
-        if not operator_id:
-            continue
-
-        overrides.update(_extract_step_overrides(step))
-        overrides.setdefault("outputDir", output_dir)
-
-        normalized.append(
-            {
-                "operator_id": str(operator_id),
-                "overrides": overrides,
-            }
-        )
-
-    return normalized
-
-
-def _resolve_operator_class(operator_id: str):
-    """根据 operator_id 解析算子类。"""
-
-    if operator_id == "ImageObjectDetectionBoundingBox":
-        if ImageObjectDetectionBoundingBox is None:
-            raise ImportError("ImageObjectDetectionBoundingBox is not available")
-        return ImageObjectDetectionBoundingBox
-
-    registry_item = OPERATORS.get(operator_id) if OPERATORS is not None else None
-    if registry_item is None:
-        try:
-            from core.base_op import OPERATORS as relative_operators  # type: ignore
-            registry_item = relative_operators.get(operator_id)
-        except Exception:
-            registry_item = None
-
-    if registry_item is None:
-        raise ImportError(f"Operator not found in registry: {operator_id}")
-
-    if isinstance(registry_item, str):
-        submodule = importlib.import_module(registry_item)
-        operator_cls = getattr(submodule, operator_id, None)
-        if operator_cls is None:
-            raise ImportError(
-                f"Operator class {operator_id} not found in module {registry_item}"
-            )
-        return operator_cls
-
-    return registry_item
-
-
-def _build_operator_chain(pipeline: List[Dict[str, Any]]) -> List[Tuple[str, Any]]:
-    """初始化算子链。"""
-
-    chain: List[Tuple[str, Any]] = []
-    for step in pipeline:
-        operator_id = step.get("operator_id")
-        overrides = dict(step.get("overrides") or {})
-        if not operator_id:
-            continue
-
-        operator_cls = _resolve_operator_class(str(operator_id))
-        operator = operator_cls(**overrides)
-        chain.append((str(operator_id), operator))
-    return chain
-
-
-def _run_pipeline_sample(sample: Dict[str, Any], chain: List[Tuple[str, Any]]) -> Dict[str, Any]:
-    """在单个样本上执行 pipeline。"""
-
-    current_sample: Dict[str, Any] = dict(sample)
-    for operator_id, operator in chain:
-        if hasattr(operator, "execute") and callable(getattr(operator, "execute")):
-            result = operator.execute(current_sample)
-        elif callable(operator):
-            result = operator(current_sample)
-        else:
-            raise RuntimeError(f"Operator {operator_id} is not executable")
-
-        if result is None:
-            continue
-
-        if isinstance(result, dict):
-            current_sample.update(result)
-            continue
-
-        if isinstance(result, list):
-            # 仅取第一个 dict 结果，兼容部分返回 list 的算子
-            if result and isinstance(result[0], dict):
-                current_sample.update(result[0])
-            continue
-
-        logger.debug(
-            "Operator {} returned unsupported result type: {}",
-            operator_id,
-            type(result).__name__,
-        )
-    return current_sample
-
-
-def _count_detections(sample: Dict[str, Any]) -> int:
-    """从样本中提取检测数量。"""
-
-    annotations = sample.get("annotations")
-    if isinstance(annotations, dict):
-        detections = annotations.get("detections")
-        if isinstance(detections, list):
-            return len(detections)
-
-    detection_count = sample.get("detection_count")
-    if detection_count is None:
-        return 0
-    try:
-        return max(int(detection_count), 0)
-    except Exception:
-        return 0
-
-
-def _get_operator_whitelist() -> Optional[set[str]]:
-    """获取灰度白名单；返回 None 表示放开全部。"""
-
-    raw = str(DEFAULT_OPERATOR_WHITELIST or "").strip()
-    if not raw:
-        return None
-
-    normalized = raw.lower()
-    if normalized in {"*", "all", "any"}:
-        return None
-
-    allow_set = {
-        item.strip()
-        for item in raw.split(",")
-        if item and item.strip()
-    }
-    return allow_set or None
-
-
-def _validate_pipeline_whitelist(pipeline: List[Dict[str, Any]]) -> None:
-    """校验 pipeline 是否命中灰度白名单。"""
-
-    allow_set = _get_operator_whitelist()
-    if allow_set is None:
-        return
-
-    blocked = []
-    for step in pipeline:
-        operator_id = str(step.get("operator_id") or "")
-        if not operator_id:
-            continue
-        if operator_id not in allow_set:
-            blocked.append(operator_id)
-
-    if blocked:
-        raise RuntimeError(
-            "Operator not in whitelist: " + ", ".join(sorted(set(blocked)))
-        )
+    if output_path is not None:
+        fields.append("output_path = :output_path")
+        params["output_path"] = output_path
+    if output_dataset_id is not None:
+        fields.append("output_dataset_id = :output_dataset_id")
+        params["output_dataset_id"] = output_dataset_id
+    if error_message is not None:
+        fields.append("error_message = :error_message")
+        params["error_message"] = error_message[:2000]
+    if status == "running":
+        fields.append("heartbeat_at = :heartbeat_at")
+        params["heartbeat_at"] = datetime.now()
+    if completed:
+        fields.append("completed_at = :completed_at")
+        params["completed_at"] = datetime.now()
+    if clear_run_token:
+        fields.append("run_token = NULL")
+
+    where_clause = "id = :task_id"
+    if run_token:
+        where_clause += " AND run_token = :run_token"
+        params["run_token"] = run_token
+
+    sql = text(
+        f"""
+        UPDATE t_dm_auto_annotation_tasks
+        SET {', '.join(fields)}
+        WHERE {where_clause}
+        """
+    )
+
+    with SQLManager.create_connect() as conn:
+        result = conn.execute(sql, params)
+        if int(getattr(result, "rowcount", 0) or 0) <= 0:
+            logger.warning(
+                "No rows updated for task status change: task_id={}, status={}, run_token={}",
+                task_id,
+                status,
+                run_token,
+            )
 
 
-def _load_dataset_files(dataset_id: str) -> List[Tuple[str, str, str]]:
-    """加载指定数据集下的所有已完成文件。"""
+def _is_stop_requested(task_id: str, run_token: Optional[str] = None) -> bool:
+    """检查任务是否请求停止。"""
+
+    where_clause = "id = :task_id"
+    params: Dict[str, Any] = {"task_id": task_id}
+    if run_token:
+        where_clause += " AND run_token = :run_token"
+        params["run_token"] = run_token
+
+    sql = text(
+        f"""
+        SELECT COALESCE(stop_requested, 0)
+        FROM t_dm_auto_annotation_tasks
+        WHERE {where_clause}
+        LIMIT 1
+        """
+    )
+
+    with SQLManager.create_connect() as conn:
+        row = conn.execute(sql, params).fetchone()
+        if not row:
+            # 找不到任务（或 run_token 已失效）时保守停止
+            return True
+        return bool(row[0])
+
+
+def _extract_step_overrides(step: Dict[str, Any]) -> Dict[str, Any]:
+    """合并 pipeline 节点中的参数覆盖。"""
+
+    overrides: Dict[str, Any] = {}
+    for key in ("overrides", "settingsOverride", "settings_override"):
+        value = step.get(key)
+        if value is None:
+            continue
+        if isinstance(value, str):
+            try:
+                value = json.loads(value)
+            except Exception:
+                continue
+        if isinstance(value, dict):
+            overrides.update(value)
+    return overrides
+
+
+def _build_legacy_pipeline(config: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """将 legacy_yolo 配置映射为单步 pipeline。"""
+
+    return [
+        {
+            "operatorId": "ImageObjectDetectionBoundingBox",
+            "overrides": {
+                "modelSize": config.get("modelSize", "l"),
+                "confThreshold": float(config.get("confThreshold", 0.7)),
+                "targetClasses": config.get("targetClasses", []) or [],
+            },
+        }
+    ]
+
+
+def _get_output_dataset_name(
+    task_id: str,
+    dataset_id: str,
+    source_dataset_name: str,
+    task_name: str,
+    config: Dict[str, Any],
+    pipeline_raw: Optional[List[Any]],
+) -> str:
+    """确定输出数据集名称。"""
+
+    output_name = config.get("outputDatasetName")
+    if output_name:
+        return str(output_name)
+
+    if pipeline_raw:
+        for step in pipeline_raw:
+            if not isinstance(step, dict):
+                continue
+            overrides = _extract_step_overrides(step)
+            output_name = overrides.get("outputDatasetName") or overrides.get("output_dataset_name")
+            if output_name:
+                return str(output_name)
+
+    base_name = source_dataset_name or task_name or f"dataset-{dataset_id[:8]}"
+    return f"{base_name}_auto_{task_id[:8]}"
+
+
+def _normalize_pipeline(
+    task_mode: str,
+    config: Dict[str, Any],
+    pipeline_raw: Optional[List[Any]],
+    output_dir: str,
+) -> List[Dict[str, Any]]:
+    """标准化 pipeline 结构并注入 outputDir。"""
+
+    source_pipeline = pipeline_raw
+    if task_mode == "legacy_yolo" or not source_pipeline:
+        source_pipeline = _build_legacy_pipeline(config)
+
+    normalized: List[Dict[str, Any]] = []
+    for step in source_pipeline:
+        if not isinstance(step, dict):
+            continue
+
+        operator_id: Optional[str] = None
+        overrides: Dict[str, Any] = {}
+
+        # 兼容 [{"OpName": {...}}] 风格
+        if (
+            "operatorId" not in step
+            and "operator_id" not in step
+            and "id" not in step
+            and len(step) == 1
+        ):
+            first_key = next(iter(step.keys()))
+            first_value = step.get(first_key)
+            if isinstance(first_key, str):
+                operator_id = first_key
+            if isinstance(first_value, dict):
+                overrides.update(first_value)
+
+        operator_id = operator_id or step.get("operatorId") or step.get("operator_id") or step.get("id")
+        if not operator_id:
+            continue
+
+        overrides.update(_extract_step_overrides(step))
+        overrides.setdefault("outputDir", output_dir)
+
+        normalized.append(
+            {
+                "operator_id": str(operator_id),
+                "overrides": overrides,
+            }
+        )
+
+    return normalized
+
+
+def _resolve_operator_class(operator_id: str):
+    """根据 operator_id 解析算子类。"""
+
+    if operator_id == "ImageObjectDetectionBoundingBox":
+        if ImageObjectDetectionBoundingBox is None:
+            raise ImportError("ImageObjectDetectionBoundingBox is not available")
+        return ImageObjectDetectionBoundingBox
+
+    registry_item = OPERATORS.get(operator_id) if OPERATORS is not None else None
+    if registry_item is None:
+        try:
+            from core.base_op import OPERATORS as relative_operators  # type: ignore
+            registry_item = relative_operators.get(operator_id)
+        except Exception:
+            registry_item = None
+
+    if registry_item is None:
+        raise ImportError(f"Operator not found in registry: {operator_id}")
+
+    if isinstance(registry_item, str):
+        submodule = importlib.import_module(registry_item)
+        operator_cls = getattr(submodule, operator_id, None)
+        if operator_cls is None:
+            raise ImportError(
+                f"Operator class {operator_id} not found in module {registry_item}"
+            )
+        return operator_cls
+
+    return registry_item
+
+
+def _build_operator_chain(pipeline: List[Dict[str, Any]]) -> List[Tuple[str, Any]]:
+    """初始化算子链。"""
+
+    chain: List[Tuple[str, Any]] = []
+    for step in pipeline:
+        operator_id = step.get("operator_id")
+        overrides = dict(step.get("overrides") or {})
+        if not operator_id:
+            continue
+
+        operator_cls = _resolve_operator_class(str(operator_id))
+        operator = operator_cls(**overrides)
+        chain.append((str(operator_id), operator))
+    return chain
+
+
+def _run_pipeline_sample(sample: Dict[str, Any], chain: List[Tuple[str, Any]]) -> Dict[str, Any]:
+    """在单个样本上执行 pipeline。"""
+
+    current_sample: Dict[str, Any] = dict(sample)
+    for operator_id, operator in chain:
+        if hasattr(operator, "execute") and callable(getattr(operator, "execute")):
+            result = operator.execute(current_sample)
+        elif callable(operator):
+            result = operator(current_sample)
+        else:
+            raise RuntimeError(f"Operator {operator_id} is not executable")
+
+        if result is None:
+            continue
+
+        if isinstance(result, dict):
+            current_sample.update(result)
+            continue
+
+        if isinstance(result, list):
+            # 仅取第一个 dict 结果，兼容部分返回 list 的算子
+            if result and isinstance(result[0], dict):
+                current_sample.update(result[0])
+            continue
+
+        logger.debug(
+            "Operator {} returned unsupported result type: {}",
+            operator_id,
+            type(result).__name__,
+        )
+    return current_sample
+
+
+def _count_detections(sample: Dict[str, Any]) -> int:
+    """从样本中提取检测数量。"""
+
+    annotations = sample.get("annotations")
+    if isinstance(annotations, dict):
+        detections = annotations.get("detections")
+        if isinstance(detections, list):
+            return len(detections)
+
+    detection_count = sample.get("detection_count")
+    if detection_count is None:
+        return 0
+    try:
+        return max(int(detection_count), 0)
+    except Exception:
+        return 0
+
+
+# ---------------------------------------------------------------------------
+# 数据集类型 → sample key / 输出子目录 映射
+# ---------------------------------------------------------------------------
+
+DATASET_TYPE_SAMPLE_KEY: Dict[str, str] = {
+    "IMAGE": "image",
+    "TEXT": "text",
+    "AUDIO": "audio",
+    "VIDEO": "video",
+}
+
+DATASET_TYPE_DATA_DIR: Dict[str, str] = {
+    "IMAGE": "images",
+    "TEXT": "data",
+    "AUDIO": "data",
+    "VIDEO": "data",
+}
+
+
+def _get_sample_key(dataset_type: str) -> str:
+    """根据数据集类型返回 sample dict 中主数据对应的 key。"""
+    return DATASET_TYPE_SAMPLE_KEY.get(dataset_type.upper(), "image")
+
+
+def _get_data_dir_name(dataset_type: str) -> str:
+    """根据数据集类型返回输出子目录名。"""
+    return DATASET_TYPE_DATA_DIR.get(dataset_type.upper(), "images")
+
+
+def _get_operator_whitelist() -> Optional[set[str]]:
+    """获取灰度白名单；返回 None 表示放开全部。"""
+
+    raw = str(DEFAULT_OPERATOR_WHITELIST or "").strip()
+    if not raw:
+        return None
+
+    normalized = raw.lower()
+    if normalized in {"*", "all", "any"}:
+        return None
+
+    allow_set = {
+        item.strip()
+        for item in raw.split(",")
+        if item and item.strip()
+    }
+    return allow_set or None
+
+
+def _validate_pipeline_whitelist(pipeline: List[Dict[str, Any]]) -> None:
+    """校验 pipeline 是否命中灰度白名单。"""
+
+    allow_set = _get_operator_whitelist()
+    if allow_set is None:
+        return
+
+    blocked = []
+    for step in pipeline:
+        operator_id = str(step.get("operator_id") or "")
+        if not operator_id:
+            continue
+        if operator_id not in allow_set:
+            blocked.append(operator_id)
+
+    if blocked:
+        raise RuntimeError(
+            "Operator not in whitelist: " + ", ".join(sorted(set(blocked)))
+        )
+
+
+def _load_dataset_files(dataset_id: str) -> List[Tuple[str, str, str]]:
+    """加载指定数据集下的所有已完成文件。"""
 
     sql = text(
         """
@@ -574,45 +603,45 @@ def _load_dataset_files(dataset_id: str) -> List[Tuple[str, str, str]]:
         """
     )
 
-    with SQLManager.create_connect() as conn:
-        rows = conn.execute(sql, {"dataset_id": dataset_id}).fetchall()
-        return [(str(r[0]), str(r[1]), str(r[2])) for r in rows]
-
-
-def _load_dataset_meta(dataset_id: str) -> Optional[Dict[str, Any]]:
-    """加载数据集基础信息（含父ID与路径）。"""
-
-    sql = text(
-        """
-        SELECT id, name, parent_dataset_id, path
-        FROM t_dm_datasets
-        WHERE id = :dataset_id
-        """
-    )
-    with SQLManager.create_connect() as conn:
-        row = conn.execute(sql, {"dataset_id": dataset_id}).fetchone()
-        if not row:
-            return None
-        return dict(row._mapping)  # type: ignore[attr-defined]
-
-
-def _resolve_output_parent(source_dataset_id: str) -> Tuple[Optional[str], str]:
-    """根据源数据集确定产出数据集的父级与基路径（产出挂在父级下）。"""
-
-    base_path = DEFAULT_OUTPUT_ROOT.rstrip("/") or "/dataset"
-    source_meta = _load_dataset_meta(source_dataset_id)
-    if not source_meta:
-        return None, base_path
-
-    parent_dataset_id = source_meta.get("parent_dataset_id")
-    if not parent_dataset_id:
-        return None, base_path
-
-    parent_meta = _load_dataset_meta(str(parent_dataset_id))
-    parent_path = parent_meta.get("path") if parent_meta else None
-    if not parent_path:
-        return None, base_path
-    return str(parent_dataset_id), str(parent_path)
+    with SQLManager.create_connect() as conn:
+        rows = conn.execute(sql, {"dataset_id": dataset_id}).fetchall()
+        return [(str(r[0]), str(r[1]), str(r[2])) for r in rows]
+
+
+def _load_dataset_meta(dataset_id: str) -> Optional[Dict[str, Any]]:
+    """加载数据集基础信息（含父ID与路径）。"""
+
+    sql = text(
+        """
+        SELECT id, name, parent_dataset_id, path, dataset_type
+        FROM t_dm_datasets
+        WHERE id = :dataset_id
+        """
+    )
+    with SQLManager.create_connect() as conn:
+        row = conn.execute(sql, {"dataset_id": dataset_id}).fetchone()
+        if not row:
+            return None
+        return dict(row._mapping)  # type: ignore[attr-defined]
+
+
+def _resolve_output_parent(source_dataset_id: str) -> Tuple[Optional[str], str]:
+    """根据源数据集确定产出数据集的父级与基路径（产出挂在父级下）。"""
+
+    base_path = DEFAULT_OUTPUT_ROOT.rstrip("/") or "/dataset"
+    source_meta = _load_dataset_meta(source_dataset_id)
+    if not source_meta:
+        return None, base_path
+
+    parent_dataset_id = source_meta.get("parent_dataset_id")
+    if not parent_dataset_id:
+        return None, base_path
+
+    parent_meta = _load_dataset_meta(str(parent_dataset_id))
+    parent_path = parent_meta.get("path") if parent_meta else None
+    if not parent_path:
+        return None, base_path
+    return str(parent_dataset_id), str(parent_path)
 
 
 def _load_files_by_ids(file_ids: List[str]) -> List[Tuple[str, str, str]]:
@@ -638,44 +667,46 @@ def _load_files_by_ids(file_ids: List[str]) -> List[Tuple[str, str, str]]:
         return [(str(r[0]), str(r[1]), str(r[2])) for r in rows]
 
 
-def _ensure_output_dir(output_dir: str) -> str:
-    """确保输出目录及其 images/、annotations/ 子目录存在。"""
+def _ensure_output_dir(output_dir: str, dataset_type: str = "IMAGE") -> str:
+    """确保输出目录及其数据/annotations 子目录存在。"""
 
     os.makedirs(output_dir, exist_ok=True)
-    os.makedirs(os.path.join(output_dir, "images"), exist_ok=True)
+    data_dir_name = _get_data_dir_name(dataset_type)
+    os.makedirs(os.path.join(output_dir, data_dir_name), exist_ok=True)
     os.makedirs(os.path.join(output_dir, "annotations"), exist_ok=True)
     return output_dir
 
 
-def _create_output_dataset(
-    source_dataset_id: str,
-    source_dataset_name: str,
-    output_dataset_name: str,
-) -> Tuple[str, str]:
-    """为自动标注结果创建一个新的数据集并返回 (dataset_id, path)。"""
-
-    new_dataset_id = str(uuid.uuid4())
-    parent_dataset_id, dataset_base_path = _resolve_output_parent(source_dataset_id)
-    output_dir = os.path.join(dataset_base_path, new_dataset_id)
+def _create_output_dataset(
+    source_dataset_id: str,
+    source_dataset_name: str,
+    output_dataset_name: str,
+    dataset_type: str = "IMAGE",
+) -> Tuple[str, str]:
+    """为自动标注结果创建一个新的数据集并返回 (dataset_id, path)。"""
+
+    new_dataset_id = str(uuid.uuid4())
+    parent_dataset_id, dataset_base_path = _resolve_output_parent(source_dataset_id)
+    output_dir = os.path.join(dataset_base_path, new_dataset_id)
 
     description = (
         f"Auto annotations for dataset {source_dataset_name or source_dataset_id}"[:255]
     )
 
-    sql = text(
-        """
-        INSERT INTO t_dm_datasets (id, parent_dataset_id, name, description, dataset_type, path, status)
-        VALUES (:id, :parent_dataset_id, :name, :description, :dataset_type, :path, :status)
-        """
-    )
-    params = {
-        "id": new_dataset_id,
-        "parent_dataset_id": parent_dataset_id,
-        "name": output_dataset_name,
-        "description": description,
-        "dataset_type": "IMAGE",
-        "path": output_dir,
-        "status": "ACTIVE",
+    sql = text(
+        """
+        INSERT INTO t_dm_datasets (id, parent_dataset_id, name, description, dataset_type, path, status)
+        VALUES (:id, :parent_dataset_id, :name, :description, :dataset_type, :path, :status)
+        """
+    )
+    params = {
+        "id": new_dataset_id,
+        "parent_dataset_id": parent_dataset_id,
+        "name": output_dataset_name,
+        "description": description,
+        "dataset_type": dataset_type,
+        "path": output_dir,
+        "status": "ACTIVE",
     }
 
     with SQLManager.create_connect() as conn:
@@ -690,31 +721,38 @@ def _register_output_dataset(
     output_dir: str,
     output_dataset_name: str,
     total_images: int,
+    dataset_type: str = "IMAGE",
 ) -> None:
     """将自动标注结果注册到新建的数据集。"""
 
-    images_dir = os.path.join(output_dir, "images")
-    if not os.path.isdir(images_dir):
-        logger.warning(
-            "Auto-annotation images directory not found for task {}: {}",
-            task_id,
-            images_dir,
-        )
-        return
+    data_dir_name = _get_data_dir_name(dataset_type)
+    data_dir = os.path.join(output_dir, data_dir_name)
+    # 兼容旧任务和 IMAGE 算子（它们写入 images/ 目录）
+    if not os.path.isdir(data_dir):
+        fallback_dir = os.path.join(output_dir, "images")
+        if os.path.isdir(fallback_dir):
+            data_dir = fallback_dir
+        else:
+            logger.warning(
+                "Auto-annotation data directory not found for task {}: {}",
+                task_id,
+                data_dir,
+            )
+            return
 
-    image_files: List[Tuple[str, str, int]] = []
+    data_files: List[Tuple[str, str, int]] = []
     annotation_files: List[Tuple[str, str, int]] = []
     total_size = 0
 
-    for file_name in sorted(os.listdir(images_dir)):
-        file_path = os.path.join(images_dir, file_name)
+    for file_name in sorted(os.listdir(data_dir)):
+        file_path = os.path.join(data_dir, file_name)
         if not os.path.isfile(file_path):
             continue
         try:
             file_size = os.path.getsize(file_path)
         except OSError:
             file_size = 0
-        image_files.append((file_name, file_path, int(file_size)))
+        data_files.append((file_name, file_path, int(file_size)))
         total_size += int(file_size)
 
     annotations_dir = os.path.join(output_dir, "annotations")
@@ -730,23 +768,23 @@ def _register_output_dataset(
             annotation_files.append((file_name, file_path, int(file_size)))
             total_size += int(file_size)
 
-    if not image_files:
+    if not data_files:
         logger.warning(
-            "No image files found in auto-annotation output for task {}: {}",
+            "No data files found in auto-annotation output for task {}: {}",
             task_id,
-            images_dir,
+            data_dir,
         )
         return
 
-    insert_file_sql = text(
-        """
-        INSERT INTO t_dm_dataset_files (
-            id, dataset_id, file_name, file_path, logical_path, version, file_type, file_size, status
-        ) VALUES (
-            :id, :dataset_id, :file_name, :file_path, :logical_path, :version, :file_type, :file_size, :status
-        )
-        """
-    )
+    insert_file_sql = text(
+        """
+        INSERT INTO t_dm_dataset_files (
+            id, dataset_id, file_name, file_path, logical_path, version, file_type, file_size, status
+        ) VALUES (
+            :id, :dataset_id, :file_name, :file_path, :logical_path, :version, :file_type, :file_size, :status
+        )
+        """
+    )
     update_dataset_stat_sql = text(
         """
         UPDATE t_dm_datasets
@@ -759,43 +797,43 @@ def _register_output_dataset(
     with SQLManager.create_connect() as conn:
         added_count = 0
 
-        for file_name, file_path, file_size in image_files:
-            ext = os.path.splitext(file_name)[1].lstrip(".").upper() or None
-            logical_path = os.path.relpath(file_path, output_dir).replace("\\", "/")
-            conn.execute(
-                insert_file_sql,
-                {
-                    "id": str(uuid.uuid4()),
-                    "dataset_id": output_dataset_id,
-                    "file_name": file_name,
-                    "file_path": file_path,
-                    "logical_path": logical_path,
-                    "version": 1,
-                    "file_type": ext,
-                    "file_size": int(file_size),
-                    "status": "ACTIVE",
-                },
-            )
-            added_count += 1
-
-        for file_name, file_path, file_size in annotation_files:
-            ext = os.path.splitext(file_name)[1].lstrip(".").upper() or None
-            logical_path = os.path.relpath(file_path, output_dir).replace("\\", "/")
-            conn.execute(
-                insert_file_sql,
-                {
-                    "id": str(uuid.uuid4()),
-                    "dataset_id": output_dataset_id,
-                    "file_name": file_name,
-                    "file_path": file_path,
-                    "logical_path": logical_path,
-                    "version": 1,
-                    "file_type": ext,
-                    "file_size": int(file_size),
-                    "status": "ACTIVE",
-                },
-            )
-            added_count += 1
+        for file_name, file_path, file_size in data_files:
+            ext = os.path.splitext(file_name)[1].lstrip(".").upper() or None
+            logical_path = os.path.relpath(file_path, output_dir).replace("\\", "/")
+            conn.execute(
+                insert_file_sql,
+                {
+                    "id": str(uuid.uuid4()),
+                    "dataset_id": output_dataset_id,
+                    "file_name": file_name,
+                    "file_path": file_path,
+                    "logical_path": logical_path,
+                    "version": 1,
+                    "file_type": ext,
+                    "file_size": int(file_size),
+                    "status": "ACTIVE",
+                },
+            )
+            added_count += 1
+
+        for file_name, file_path, file_size in annotation_files:
+            ext = os.path.splitext(file_name)[1].lstrip(".").upper() or None
+            logical_path = os.path.relpath(file_path, output_dir).replace("\\", "/")
+            conn.execute(
+                insert_file_sql,
+                {
+                    "id": str(uuid.uuid4()),
+                    "dataset_id": output_dataset_id,
+                    "file_name": file_name,
+                    "file_path": file_path,
+                    "logical_path": logical_path,
+                    "version": 1,
+                    "file_type": ext,
+                    "file_size": int(file_size),
+                    "status": "ACTIVE",
+                },
+            )
+            added_count += 1
 
         if added_count > 0:
             conn.execute(
@@ -811,238 +849,248 @@ def _register_output_dataset(
         "Registered auto-annotation output into dataset: dataset_id={}, name={}, added_files={}, added_size_bytes={}, task_id={}, output_dir={}",
         output_dataset_id,
         output_dataset_name,
-        len(image_files) + len(annotation_files),
+        len(data_files) + len(annotation_files),
         total_size,
         task_id,
         output_dir,
     )
 
 
-def _process_single_task(task: Dict[str, Any]) -> None:
-    """执行单个自动标注任务。"""
-
-    task_id = str(task["id"])
-    dataset_id = str(task["dataset_id"])
-    task_name = str(task.get("name") or "")
-    source_dataset_name = str(task.get("dataset_name") or "")
-    run_token = str(task.get("run_token") or "")
-    task_mode = str(task.get("task_mode") or "legacy_yolo")
-    executor_type = str(task.get("executor_type") or "annotation_local")
-    cfg: Dict[str, Any] = task.get("config") or {}
-    pipeline_raw = task.get("pipeline")
-    selected_file_ids: Optional[List[str]] = task.get("file_ids") or None
-
-    output_dataset_name = _get_output_dataset_name(
-        task_id=task_id,
-        dataset_id=dataset_id,
-        source_dataset_name=source_dataset_name,
-        task_name=task_name,
-        config=cfg,
-        pipeline_raw=pipeline_raw if isinstance(pipeline_raw, list) else None,
-    )
-
-    logger.info(
-        "Start processing auto-annotation task: id={}, dataset_id={}, task_mode={}, executor_type={}, output_dataset_name={}",
-        task_id,
-        dataset_id,
-        task_mode,
-        executor_type,
-        output_dataset_name,
-    )
-
-    if _is_stop_requested(task_id, run_token):
-        logger.info("Task stop requested before processing started: {}", task_id)
-        _update_task_status(
-            task_id,
-            run_token=run_token,
-            status="stopped",
-            completed=True,
-            clear_run_token=True,
-            error_message="Task stopped before start",
-        )
-        return
-
-    _update_task_status(task_id, run_token=run_token, status="running", progress=0)
-
-    if selected_file_ids:
-        all_files = _load_files_by_ids(selected_file_ids)
-    else:
-        all_files = _load_dataset_files(dataset_id)
-
-    files = [(path, name) for _, path, name in all_files]
-
-    total_images = len(files)
-    if total_images == 0:
-        logger.warning("No files found for dataset {} when running auto-annotation task {}", dataset_id, task_id)
-        _update_task_status(
-            task_id,
-            run_token=run_token,
-            status="completed",
-            progress=100,
-            total_images=0,
-            processed_images=0,
-            detected_objects=0,
-            completed=True,
-            output_path=None,
-            clear_run_token=True,
-        )
-        return
-
-    output_dataset_id, output_dir = _create_output_dataset(
-        source_dataset_id=dataset_id,
-        source_dataset_name=source_dataset_name,
-        output_dataset_name=output_dataset_name,
-    )
-    output_dir = _ensure_output_dir(output_dir)
-
-    _update_task_status(
-        task_id,
-        run_token=run_token,
-        status="running",
-        total_images=total_images,
-        output_path=output_dir,
-        output_dataset_id=output_dataset_id,
-    )
-
-    try:
-        normalized_pipeline = _normalize_pipeline(
-            task_mode=task_mode,
-            config=cfg,
-            pipeline_raw=pipeline_raw if isinstance(pipeline_raw, list) else None,
-            output_dir=output_dir,
-        )
-
-        if not normalized_pipeline:
-            raise RuntimeError("Pipeline is empty after normalization")
-
-        _validate_pipeline_whitelist(normalized_pipeline)
-
-        chain = _build_operator_chain(normalized_pipeline)
-        if not chain:
-            raise RuntimeError("No valid operator instances initialized")
-    except Exception as e:
-        logger.error("Failed to init operator pipeline for task {}: {}", task_id, e)
-        _update_task_status(
-            task_id,
-            run_token=run_token,
-            status="failed",
-            total_images=total_images,
-            processed_images=0,
-            detected_objects=0,
-            error_message=f"Init pipeline failed: {e}",
-            clear_run_token=True,
-        )
-        return
-
-    processed = 0
-    detected_total = 0
-
-    try:
-
-        for file_path, file_name in files:
-            if _is_stop_requested(task_id, run_token):
-                logger.info("Task stop requested during processing: {}", task_id)
-                _update_task_status(
-                    task_id,
-                    run_token=run_token,
-                    status="stopped",
-                    progress=int(processed * 100 / total_images) if total_images > 0 else 0,
-                    processed_images=processed,
-                    detected_objects=detected_total,
-                    total_images=total_images,
-                    output_path=output_dir,
-                    output_dataset_id=output_dataset_id,
-                    completed=True,
-                    clear_run_token=True,
-                    error_message="Task stopped by request",
-                )
-                return
-
-            try:
-                sample = {
-                    "image": file_path,
-                    "filename": file_name,
-                }
-
-                result = _run_pipeline_sample(sample, chain)
-                detected_total += _count_detections(result)
-                processed += 1
-
-                progress = int(processed * 100 / total_images) if total_images > 0 else 100
-
-                _update_task_status(
-                    task_id,
-                    run_token=run_token,
-                    status="running",
-                    progress=progress,
-                    processed_images=processed,
-                    detected_objects=detected_total,
-                    total_images=total_images,
-                    output_path=output_dir,
-                    output_dataset_id=output_dataset_id,
-                )
-            except Exception as e:
-                logger.error(
-                    "Failed to process image for task {}: file_path={}, error={}",
-                    task_id,
-                    file_path,
-                    e,
-                )
-                continue
-
-        _update_task_status(
-            task_id,
-            run_token=run_token,
-            status="completed",
-            progress=100,
-            processed_images=processed,
-            detected_objects=detected_total,
-            total_images=total_images,
-            output_path=output_dir,
-            output_dataset_id=output_dataset_id,
-            completed=True,
-            clear_run_token=True,
-        )
-
-        logger.info(
-            "Completed auto-annotation task: id={}, total_images={}, processed={}, detected_objects={}, output_path={}",
-            task_id,
-            total_images,
-            processed,
-            detected_total,
-            output_dir,
-        )
-
-        if output_dataset_name and output_dataset_id:
-            try:
-                _register_output_dataset(
-                    task_id=task_id,
-                    output_dataset_id=output_dataset_id,
-                    output_dir=output_dir,
-                    output_dataset_name=output_dataset_name,
-                    total_images=total_images,
-                )
-            except Exception as e:  # pragma: no cover - 防御性日志
-                logger.error(
-                    "Failed to register auto-annotation output as dataset for task {}: {}",
-                    task_id,
-                    e,
-                )
-    except Exception as e:
-        logger.error("Task execution failed for task {}: {}", task_id, e)
-        _update_task_status(
-            task_id,
-            run_token=run_token,
-            status="failed",
-            progress=int(processed * 100 / total_images) if total_images > 0 else 0,
-            processed_images=processed,
-            detected_objects=detected_total,
-            total_images=total_images,
-            output_path=output_dir,
-            output_dataset_id=output_dataset_id,
-            error_message=f"Execute pipeline failed: {e}",
-            clear_run_token=True,
-        )
+def _process_single_task(task: Dict[str, Any]) -> None:
+    """执行单个自动标注任务。"""
+
+    task_id = str(task["id"])
+    dataset_id = str(task["dataset_id"])
+    task_name = str(task.get("name") or "")
+    source_dataset_name = str(task.get("dataset_name") or "")
+    run_token = str(task.get("run_token") or "")
+    task_mode = str(task.get("task_mode") or "legacy_yolo")
+    executor_type = str(task.get("executor_type") or "annotation_local")
+    cfg: Dict[str, Any] = task.get("config") or {}
+    pipeline_raw = task.get("pipeline")
+    selected_file_ids: Optional[List[str]] = task.get("file_ids") or None
+
+    # 解析数据集类型，兜底从数据集元数据获取
+    dataset_type = str(task.get("dataset_type") or "").upper() or "IMAGE"
+    if dataset_type == "IMAGE" and not task.get("dataset_type"):
+        source_meta = _load_dataset_meta(dataset_id)
+        if source_meta and source_meta.get("dataset_type"):
+            dataset_type = str(source_meta["dataset_type"]).upper()
+
+    output_dataset_name = _get_output_dataset_name(
+        task_id=task_id,
+        dataset_id=dataset_id,
+        source_dataset_name=source_dataset_name,
+        task_name=task_name,
+        config=cfg,
+        pipeline_raw=pipeline_raw if isinstance(pipeline_raw, list) else None,
+    )
+
+    logger.info(
+        "Start processing auto-annotation task: id={}, dataset_id={}, task_mode={}, executor_type={}, output_dataset_name={}",
+        task_id,
+        dataset_id,
+        task_mode,
+        executor_type,
+        output_dataset_name,
+    )
+
+    if _is_stop_requested(task_id, run_token):
+        logger.info("Task stop requested before processing started: {}", task_id)
+        _update_task_status(
+            task_id,
+            run_token=run_token,
+            status="stopped",
+            completed=True,
+            clear_run_token=True,
+            error_message="Task stopped before start",
+        )
+        return
+
+    _update_task_status(task_id, run_token=run_token, status="running", progress=0)
+
+    if selected_file_ids:
+        all_files = _load_files_by_ids(selected_file_ids)
+    else:
+        all_files = _load_dataset_files(dataset_id)
+
+    files = [(path, name) for _, path, name in all_files]
+
+    total_images = len(files)
+    if total_images == 0:
+        logger.warning("No files found for dataset {} when running auto-annotation task {}", dataset_id, task_id)
+        _update_task_status(
+            task_id,
+            run_token=run_token,
+            status="completed",
+            progress=100,
+            total_images=0,
+            processed_images=0,
+            detected_objects=0,
+            completed=True,
+            output_path=None,
+            clear_run_token=True,
+        )
+        return
+
+    output_dataset_id, output_dir = _create_output_dataset(
+        source_dataset_id=dataset_id,
+        source_dataset_name=source_dataset_name,
+        output_dataset_name=output_dataset_name,
+        dataset_type=dataset_type,
+    )
+    output_dir = _ensure_output_dir(output_dir, dataset_type=dataset_type)
+
+    _update_task_status(
+        task_id,
+        run_token=run_token,
+        status="running",
+        total_images=total_images,
+        output_path=output_dir,
+        output_dataset_id=output_dataset_id,
+    )
+
+    try:
+        normalized_pipeline = _normalize_pipeline(
+            task_mode=task_mode,
+            config=cfg,
+            pipeline_raw=pipeline_raw if isinstance(pipeline_raw, list) else None,
+            output_dir=output_dir,
+        )
+
+        if not normalized_pipeline:
+            raise RuntimeError("Pipeline is empty after normalization")
+
+        _validate_pipeline_whitelist(normalized_pipeline)
+
+        chain = _build_operator_chain(normalized_pipeline)
+        if not chain:
+            raise RuntimeError("No valid operator instances initialized")
+    except Exception as e:
+        logger.error("Failed to init operator pipeline for task {}: {}", task_id, e)
+        _update_task_status(
+            task_id,
+            run_token=run_token,
+            status="failed",
+            total_images=total_images,
+            processed_images=0,
+            detected_objects=0,
+            error_message=f"Init pipeline failed: {e}",
+            clear_run_token=True,
+        )
+        return
+
+    processed = 0
+    detected_total = 0
+
+    try:
+
+        for file_path, file_name in files:
+            if _is_stop_requested(task_id, run_token):
+                logger.info("Task stop requested during processing: {}", task_id)
+                _update_task_status(
+                    task_id,
+                    run_token=run_token,
+                    status="stopped",
+                    progress=int(processed * 100 / total_images) if total_images > 0 else 0,
+                    processed_images=processed,
+                    detected_objects=detected_total,
+                    total_images=total_images,
+                    output_path=output_dir,
+                    output_dataset_id=output_dataset_id,
+                    completed=True,
+                    clear_run_token=True,
+                    error_message="Task stopped by request",
+                )
+                return
+
+            try:
+                sample_key = _get_sample_key(dataset_type)
+                sample = {
+                    sample_key: file_path,
+                    "filename": file_name,
+                }
+
+                result = _run_pipeline_sample(sample, chain)
+                detected_total += _count_detections(result)
+                processed += 1
+
+                progress = int(processed * 100 / total_images) if total_images > 0 else 100
+
+                _update_task_status(
+                    task_id,
+                    run_token=run_token,
+                    status="running",
+                    progress=progress,
+                    processed_images=processed,
+                    detected_objects=detected_total,
+                    total_images=total_images,
+                    output_path=output_dir,
+                    output_dataset_id=output_dataset_id,
+                )
+            except Exception as e:
+                logger.error(
+                    "Failed to process file for task {}: file_path={}, error={}",
+                    task_id,
+                    file_path,
+                    e,
+                )
+                continue
+
+        _update_task_status(
+            task_id,
+            run_token=run_token,
+            status="completed",
+            progress=100,
+            processed_images=processed,
+            detected_objects=detected_total,
+            total_images=total_images,
+            output_path=output_dir,
+            output_dataset_id=output_dataset_id,
+            completed=True,
+            clear_run_token=True,
+        )
+
+        logger.info(
+            "Completed auto-annotation task: id={}, total_images={}, processed={}, detected_objects={}, output_path={}",
+            task_id,
+            total_images,
+            processed,
+            detected_total,
+            output_dir,
+        )
+
+        if output_dataset_name and output_dataset_id:
+            try:
+                _register_output_dataset(
+                    task_id=task_id,
+                    output_dataset_id=output_dataset_id,
+                    output_dir=output_dir,
+                    output_dataset_name=output_dataset_name,
+                    total_images=total_images,
+                    dataset_type=dataset_type,
+                )
+            except Exception as e:  # pragma: no cover - 防御性日志
+                logger.error(
+                    "Failed to register auto-annotation output as dataset for task {}: {}",
+                    task_id,
+                    e,
+                )
+    except Exception as e:
+        logger.error("Task execution failed for task {}: {}", task_id, e)
+        _update_task_status(
+            task_id,
+            run_token=run_token,
+            status="failed",
+            progress=int(processed * 100 / total_images) if total_images > 0 else 0,
+            processed_images=processed,
+            detected_objects=detected_total,
+            total_images=total_images,
+            output_path=output_dir,
+            output_dataset_id=output_dataset_id,
+            error_message=f"Execute pipeline failed: {e}",
+            clear_run_token=True,
+        )
 
 
 def _worker_loop() -> None:
diff --git a/scripts/db/data-annotation-multitype-migration.sql b/scripts/db/data-annotation-multitype-migration.sql
new file mode 100644
index 0000000..518b7ab
--- /dev/null
+++ b/scripts/db/data-annotation-multitype-migration.sql
@@ -0,0 +1,25 @@
+-- =============================================
+-- 自动标注任务支持多数据集类型迁移
+-- 为 t_dm_auto_annotation_tasks 表添加 dataset_type 列
+-- =============================================
+
+USE datamate;
+SET @db_name = DATABASE();
+
+-- 添加 dataset_type 列（IMAGE/TEXT/AUDIO/VIDEO），已有记录默认为 IMAGE
+SET @ddl = (
+    SELECT IF(
+        EXISTS(
+            SELECT 1
+            FROM information_schema.COLUMNS
+            WHERE TABLE_SCHEMA = @db_name
+              AND TABLE_NAME = 't_dm_auto_annotation_tasks'
+              AND COLUMN_NAME = 'dataset_type'
+        ),
+        'SELECT ''skip: column dataset_type already exists''',
+        'ALTER TABLE t_dm_auto_annotation_tasks ADD COLUMN dataset_type VARCHAR(50) NOT NULL DEFAULT ''IMAGE'' COMMENT ''数据集类型: IMAGE/TEXT/AUDIO/VIDEO'' AFTER dataset_name'
+    )
+);
+PREPARE stmt FROM @ddl;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;