diff --git a/frontend/src/pages/SynthesisTask/CreateTask.tsx b/frontend/src/pages/SynthesisTask/CreateTask.tsx
index 7b0c62e50..f224887e6 100644
--- a/frontend/src/pages/SynthesisTask/CreateTask.tsx
+++ b/frontend/src/pages/SynthesisTask/CreateTask.tsx
@@ -53,8 +53,8 @@ export default function SynthesisTaskCreate() {
       | "PARAGRAPH_CHUNK"
       | "FIXED_LENGTH_CHUNK"
       | "CUSTOM_SEPARATOR_CHUNK",
-    chunkSize: 3000,
-    overlapSize: 100,
+    chunkSize: 500,
+    overlapSize: 50,
     delimiter: "",
   });
 
diff --git a/frontend/src/pages/SynthesisTask/SynthFileTask.tsx b/frontend/src/pages/SynthesisTask/SynthFileTask.tsx
index d0a5bcf2d..660646c39 100644
--- a/frontend/src/pages/SynthesisTask/SynthFileTask.tsx
+++ b/frontend/src/pages/SynthesisTask/SynthFileTask.tsx
@@ -290,7 +290,7 @@ export default function SynthFileTask() {
       danger: true,
       confirm: {
         title: t('synthesisTask.fileTask.confirm.deleteTitle'),
-        description: t('synthesisTask.fileTask.confirm.deleteDescription', { name: synthesisTask?.name || '' }),
+        description: t('synthesisTask.fileTask.confirm.deleteDescription', { name: taskInfo?.name || '' }),
         okText: t('dataManagement.confirm.deleteConfirm'),
         cancelText: t('dataManagement.confirm.deleteCancel'),
         onConfirm: handleDelete,
diff --git a/frontend/src/pages/SynthesisTask/components/SynthesisTaskTab.tsx b/frontend/src/pages/SynthesisTask/components/SynthesisTaskTab.tsx
index 12716aae3..fe8562f5d 100644
--- a/frontend/src/pages/SynthesisTask/components/SynthesisTaskTab.tsx
+++ b/frontend/src/pages/SynthesisTask/components/SynthesisTaskTab.tsx
@@ -68,6 +68,9 @@ export default function SynthesisTaskTab() {
   const [evalLoading, setEvalLoading] = useState(false);
   const [models, setModels] = useState<ModelI[]>([]);
   const [modelLoading, setModelLoading] = useState(false);
+  const [archiveModalVisible, setArchiveModalVisible] = useState(false);
+  const [currentArchiveTask, setCurrentArchiveTask] = useState<SynthesisTask | null>(null);
+  const [archiveFormat, setArchiveFormat] = useState<string>("alpaca");
 
   const [evalForm] = Form.useForm();
 
@@ -210,7 +213,7 @@ export default function SynthesisTaskTab() {
       title: t('synthesisTask.home.columns.actions'),
       key: "actions",
       fixed: "right" as const,
-      width: 120,
+      width: 160,
       render: (_: unknown, task: SynthesisTask) => (
         <div className="flex items-center justify-start gap-1">
           <Tooltip title={t('synthesisTask.actions.viewDetail')}>
@@ -234,15 +237,7 @@ export default function SynthesisTaskTab() {
               type="text"
               className="hover:bg-green-50 p-1 h-7 w-7 flex items-center justify-center text-green-600"
               icon={<FolderOpenOutlined />}
-              onClick={() => {
-                Modal.confirm({
-                  title: t('synthesisTask.home.confirm.archiveTitle'),
-                  content: t('synthesisTask.home.confirm.archiveContent', { name: task.name }),
-                  okText: t('synthesisTask.actions.archive'),
-                  cancelText: t('synthesisTask.actions.cancel'),
-                  onOk: () => handleArchiveTask(task),
-                });
-              }}
+              onClick={() => openArchiveModal(task)}
             />
           </Tooltip>
           <Tooltip title={t('synthesisTask.actions.delete')}>
@@ -276,9 +271,8 @@ export default function SynthesisTaskTab() {
     },
   ];
 
-  const handleArchiveTask = async (task: SynthesisTask) => {
+  const handleArchiveTask = async (task: SynthesisTask, format: string = "alpaca") => {
     try {
-      // 1. 创建目标数据集（使用简单的默认命名 + 随机后缀，可后续扩展为弹窗自定义）
       const randomSuffix = Math.random().toString(36).slice(2, 8);
       const datasetReq: {
         name: string;
@@ -302,11 +296,9 @@ export default function SynthesisTaskTab() {
         return;
       }
 
-      // 2. 调用后端归档接口，将合成数据写入该数据集
-      await archiveSynthesisTaskToDatasetUsingPost(task.id, datasetId);
+      await archiveSynthesisTaskToDatasetUsingPost(task.id, datasetId, format);
 
       message.success(t('synthesisTask.home.archive.success'));
-      // 3. 可选：跳转到数据集详情页
       navigate(`/data/management/detail/${datasetId}`);
     } catch (e) {
       console.error(e);
@@ -314,6 +306,18 @@ export default function SynthesisTaskTab() {
     }
   };
 
+  const openArchiveModal = (task: SynthesisTask) => {
+    setCurrentArchiveTask(task);
+    setArchiveFormat("alpaca");
+    setArchiveModalVisible(true);
+  };
+
+  const handleArchiveConfirm = async () => {
+    if (!currentArchiveTask) return;
+    setArchiveModalVisible(false);
+    await handleArchiveTask(currentArchiveTask, archiveFormat);
+  };
+
   const openEvalModal = (task: SynthesisTask) => {
     setCurrentEvalTask(task);
     setEvalModalVisible(true);
@@ -464,6 +468,37 @@ export default function SynthesisTaskTab() {
         />
       </Card>
 
+      <Modal
+        title={t('synthesisTask.home.confirm.archiveTitle')}
+        open={archiveModalVisible}
+        onCancel={() => {
+          setArchiveModalVisible(false);
+          setCurrentArchiveTask(null);
+        }}
+        onOk={handleArchiveConfirm}
+        okText={t('synthesisTask.actions.archive')}
+        cancelText={t('synthesisTask.actions.cancel')}
+      >
+        <div>
+          <p>{t('synthesisTask.home.confirm.archiveContent', { name: currentArchiveTask?.name })}</p>
+          <div style={{ marginTop: 16 }}>
+            <span style={{ marginRight: 8 }}>导出格式：</span>
+            <Select
+              value={archiveFormat}
+              style={{ width: 120 }}
+              options={[
+                { label: "Alpaca", value: "alpaca" },
+                { label: "ShareGPT", value: "sharegpt" },
+                { label: "原始格式", value: "raw" },
+              ]}
+              onChange={(value) => {
+                setArchiveFormat(value);
+              }}
+            />
+          </div>
+        </div>
+      </Modal>
+
       <Modal
         title={t('synthesisTask.home.modal.evalTitle')}
         open={evalModalVisible}
diff --git a/frontend/src/pages/SynthesisTask/synthesis-api.ts b/frontend/src/pages/SynthesisTask/synthesis-api.ts
index 7be1bd690..15aae1071 100644
--- a/frontend/src/pages/SynthesisTask/synthesis-api.ts
+++ b/frontend/src/pages/SynthesisTask/synthesis-api.ts
@@ -1,4 +1,4 @@
-import { get, post, del } from "@/utils/request";
+import { get, post, del, patch } from "@/utils/request";
 
 // 创建数据合成任务
 export function createSynthesisTaskUsingPost(data: Record<string, unknown>) {
@@ -65,28 +65,25 @@ export function getPromptByTypeUsingGet(synthType: string) {
 }
 
 // 将合成任务数据归档到已存在的数据集中
-export function archiveSynthesisTaskToDatasetUsingPost(taskId: string, datasetId: string) {
-  return post(`/api/synthesis/gen/task/${taskId}/export-dataset/${datasetId}`);
+export function archiveSynthesisTaskToDatasetUsingPost(
+  taskId: string,
+  datasetId: string,
+  format: string = "alpaca"
+) {
+  return post(`/api/synthesis/gen/task/${taskId}/export-dataset/${datasetId}?format=${format}`);
 }
 
-// ---------------- 数据记录级别：chunk 与 synthesis data ----------------
-
-// 根据 chunkId 删除单个 chunk 及其下所有合成数据
+// 删除 chunk 及其关联的合成数据
 export function deleteChunkWithDataUsingDelete(chunkId: string) {
   return del(`/api/synthesis/gen/chunk/${chunkId}`);
 }
 
-// 删除某个 chunk 下的所有合成数据，返回删除条数
-export function deleteSynthesisDataByChunkUsingDelete(chunkId: string) {
-  return del(`/api/synthesis/gen/chunk/${chunkId}/data`);
-}
-
-// 批量删除合成数据记录
-export function batchDeleteSynthesisDataUsingDelete(body: { ids: string[] }) {
-  return del(`/api/synthesis/gen/data/batch`, null, { body: JSON.stringify(body) });
+// 批量删除合成数据
+export function batchDeleteSynthesisDataUsingDelete(data: { ids: string[] }) {
+  return del("/api/synthesis/gen/data/batch", data as unknown as Record<string, never>);
 }
 
-// 更新单条合成数据的完整 JSON 内容
-export function updateSynthesisDataUsingPatch(dataId: string, body: { data: Record<string, unknown> }) {
-  return post(`/api/synthesis/gen/data/${dataId}`, body, { method: "PATCH" });
+// 更新合成数据
+export function updateSynthesisDataUsingPatch(dataId: string, data: { data: Record<string, unknown> }) {
+  return patch(`/api/synthesis/gen/data/${dataId}`, data as unknown as Record<string, never>);
 }
diff --git a/frontend/src/utils/request.ts b/frontend/src/utils/request.ts
index f76c115e8..91aa77b55 100644
--- a/frontend/src/utils/request.ts
+++ b/frontend/src/utils/request.ts
@@ -405,6 +405,28 @@ class Request {
     return this.request(fullURL, config);
   }
 
+  /**
+   * PATCH请求
+   * @param {string} url - 请求URL
+   * @param {object} data - 请求体数据
+   * @param {object} options - 额外的fetch选项，包括showLoading, onUploadProgress, onDownloadProgress
+   */
+  async patch(url, data = null, options = {}) {
+    const config = {
+      method: "PATCH",
+      credentials: "include",
+      mode: "cors",
+      headers: {
+        ...this.defaultHeaders,
+        ...options.headers,
+      },
+      body: data ? JSON.stringify(data) : undefined,
+      ...options,
+    };
+
+    return this.request(this.baseURL + url, config);
+  }
+
   /**
    * 从 Content-Disposition 头中解析文件名
    */
@@ -662,6 +684,7 @@ export const get = request.get.bind(request);
 export const post = request.post.bind(request);
 export const put = request.put.bind(request);
 export const del = request.delete.bind(request);
+export const patch = request.patch.bind(request);
 export const download = request.download.bind(request);
 export const upload = request.upload.bind(request);
 
diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts
index e8a58c53e..46b5ea887 100644
--- a/frontend/vite.config.ts
+++ b/frontend/vite.config.ts
@@ -15,7 +15,7 @@ export default defineConfig({
     host: "0.0.0.0",
     proxy: (() => {
       const pythonProxyConfig = {
-        target: "http://localhost:32033",
+        target: "http://localhost:18000",
         changeOrigin: true,
         secure: false,
         configure: (proxy: { on: (event: string, handler: (arg: unknown) => void) => void }) => {
@@ -32,7 +32,7 @@ export default defineConfig({
       };
 
       const javaProxyConfig = {
-        target: "http://localhost:32033",
+        target: "http://localhost:8080",
         changeOrigin: true,
         secure: false,
         configure: (proxy: { on: (event: string, handler: (arg: unknown) => void) => void }) => {
diff --git a/runtime/datamate-python/app/core/exception/middleware.py b/runtime/datamate-python/app/core/exception/middleware.py
index 561d130df..82b03ca24 100644
--- a/runtime/datamate-python/app/core/exception/middleware.py
+++ b/runtime/datamate-python/app/core/exception/middleware.py
@@ -69,7 +69,7 @@ async def dispatch(self, request: Request, call_next):
         except Exception as exc:
             # 捕获所有未处理的异常
             logger.error(
-                f"Unhandled exception occurred at {request.method} {request.url.path}", exc,
+                f"Unhandled exception occurred at {request.method} {request.url.path}",
                 exc_info=True
             )
             return self._error_response(
diff --git a/runtime/datamate-python/app/main.py b/runtime/datamate-python/app/main.py
index a0a5499d0..1d837677c 100644
--- a/runtime/datamate-python/app/main.py
+++ b/runtime/datamate-python/app/main.py
@@ -25,6 +25,7 @@
     set_collection_scheduler,
 )
 from app.module.shared.schedule import Scheduler
+from app.module.generation.service.task_executor import init_executor, shutdown_executor
 
 setup_logging()
 logger = get_logger(__name__)
@@ -68,10 +69,15 @@ def mask_db_url(url: str) -> Literal[b""] | str:
     set_collection_scheduler(collection_scheduler)
     await load_scheduled_collection_tasks()
 
+    # Initialize generation task executor
+    init_executor(max_workers=10, max_concurrent_tasks=5)
+    logger.info("Generation task executor initialized")
+
     yield
 
     # @shutdown
     collection_scheduler.shutdown()
+    shutdown_executor()
     logger.info("DataMate Python Backend shutting down ...\n\n")
 
 
diff --git a/runtime/datamate-python/app/module/generation/interface/generation_api.py b/runtime/datamate-python/app/module/generation/interface/generation_api.py
index e186f53ad..246c300c6 100644
--- a/runtime/datamate-python/app/module/generation/interface/generation_api.py
+++ b/runtime/datamate-python/app/module/generation/interface/generation_api.py
@@ -1,37 +1,44 @@
-import uuid
-from typing import cast
+"""
+数据合成模块 API 接口层
+
+该模块提供数据合成任务的 RESTful API 接口。
+每个接口负责：参数解析 → 调用服务 → 返回响应
+"""
+from typing import List, Optional
 
 from fastapi import APIRouter, Depends, BackgroundTasks
-from sqlalchemy import select, func, delete
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.core.exception import ErrorCodes, BusinessError, SuccessResponse, transaction
+from app.core.exception import BusinessError, ErrorCodes, SuccessResponse
 from app.core.logging import get_logger
-from app.db.models.data_synthesis import (
-    save_synthesis_task,
-    DataSynthInstance,
-    DataSynthesisFileInstance,
-    DataSynthesisChunkInstance,
-    SynthesisData,
-)
-from app.db.models.dataset_management import DatasetFiles
 from app.db.session import get_db
 from app.module.generation.schema.generation import (
     CreateSynthesisTaskRequest,
     DataSynthesisTaskItem,
     PagedDataSynthesisTaskResponse,
     SynthesisType,
-    DataSynthesisFileTaskItem,
     PagedDataSynthesisFileTaskResponse,
-    DataSynthesisChunkItem,
     PagedDataSynthesisChunkResponse,
     SynthesisDataItem,
     SynthesisDataUpdateRequest,
     BatchDeleteSynthesisDataRequest,
+    BatchDeleteChunkInstancesRequest,
+    BatchDeleteChunkInstancesByFileRequest,
+    BatchDeleteChunkInstancesByTaskRequest,
+    SynthesisDataPatchItem,
+    ExportSynthesisDataRequest,
+    ExportSynthesisDataResponse,
 )
+from app.module.generation.service.chunk_query_service import ChunkQueryService
 from app.module.generation.service.export_service import SynthesisDatasetExporter, SynthesisExportError
+from app.module.generation.service.file_query_service import FileQueryService
 from app.module.generation.service.generation_service import GenerationService
 from app.module.generation.service.prompt import get_prompt
+from app.module.generation.service.synthesis_data_query_service import SynthesisDataQueryService
+from app.module.generation.service.task_create_service import TaskCreateService
+from app.module.generation.service.task_delete_service import TaskDeleteService
+from app.module.generation.service.task_query_service import TaskQueryService
+from app.module.generation.service.task_executor import execute_generation_task
 from app.module.shared.schema import StandardResponse
 
 router = APIRouter(
@@ -41,96 +48,72 @@
 
 logger = get_logger(__name__)
 
+
+# ==================== 任务接口 ====================
+
 @router.post("/task", response_model=StandardResponse[DataSynthesisTaskItem])
 async def create_synthesis_task(
     request: CreateSynthesisTaskRequest,
     background_tasks: BackgroundTasks,
     db: AsyncSession = Depends(get_db),
 ):
-    """创建数据合成任务"""
-    # 先根据 source_file_id 在 DatasetFiles 中查出已有文件信息
-    file_ids = request.source_file_id or []
-    dataset_files = []
-    if file_ids:
-        ds_result = await db.execute(
-            select(DatasetFiles).where(DatasetFiles.id.in_(file_ids))
-        )
-        dataset_files = ds_result.scalars().all()
-
-    # 保存任务到数据库（在事务中）
-    request.source_file_id = [str(f.id) for f in dataset_files]
-
-    async with transaction(db):
-        synthesis_task = await save_synthesis_task(db, request)
-
-        # 将已有的 DatasetFiles 记录保存到 t_data_synthesis_file_instances
-        synth_files = []
-        for f in dataset_files:
-            file_instance = DataSynthesisFileInstance(
-                id=str(uuid.uuid4()),  # 使用新的 UUID 作为文件任务记录的主键，避免与 DatasetFiles 主键冲突
-                synthesis_instance_id=synthesis_task.id,
-                file_name=f.file_name,
-                source_file_id=str(f.id),
-                status="pending",
-                total_chunks=0,
-                processed_chunks=0,
-                created_by="system",
-                updated_by="system",
-            )
-            synth_files.append(file_instance)
-
-        if dataset_files:
-            db.add_all(synth_files)
-
-    # 事务已提交，启动后台任务
-    generation_service = GenerationService(db)
-    # 异步处理任务：只传任务 ID，后台任务中使用新的 DB 会话重新加载任务对象
-    background_tasks.add_task(generation_service.process_task, synthesis_task.id)
-
-    # 将 ORM 对象包装成 DataSynthesisTaskItem，兼容新字段从 synth_config 还原
-
-    task_item = DataSynthesisTaskItem(
-        id=synthesis_task.id,
-        name=synthesis_task.name,
-        description=synthesis_task.description,
-        status=synthesis_task.status,
-        synthesis_type=synthesis_task.synth_type,
-        total_files=synthesis_task.total_files,
-        created_at=synthesis_task.created_at,
-        updated_at=synthesis_task.updated_at,
-        created_by=synthesis_task.created_by,
-        updated_by=synthesis_task.updated_by,
-    )
+    """
+    创建数据合成任务
+
+    使用改进的后台任务执行器，将任务提交到独立的线程池中执行，
+    避免阻塞主事件循环，确保前端接口的响应性。
+
+    Args:
+        request: 创建任务请求
+        background_tasks: 后台任务
+        db: 数据库会话
+
+    Returns:
+        创建的任务
+    """
+    create_service = TaskCreateService(db)
+    task = await create_service.create(request)
+
+    # 使用新的执行器提交任务到线程池
+    # 这样可以真正隔离后台任务，避免阻塞主事件循环
+    async def run_task_in_executor(task_id: str):
+        """在线程池中执行任务"""
+        # 定义一个包装函数来正确调用process_task
+        async def process_task_wrapper(tid: str):
+            # 创建新的GenerationService实例，使用独立的会话
+            from app.db.session import AsyncSessionLocal
+            async with AsyncSessionLocal() as session:
+                service = GenerationService(session)
+                await service.process_task(tid)
 
-    return SuccessResponse(data=task_item)
+        await execute_generation_task(task_id, process_task_wrapper)
+
+    background_tasks.add_task(run_task_in_executor, task.id)
+    logger.info(f"Submitted generation task {task.id} to background executor")
+
+    return SuccessResponse(data=DataSynthesisTaskItem.from_orm_with_config(task))
 
 
 @router.get("/task/{task_id}", response_model=StandardResponse[DataSynthesisTaskItem])
-async def get_synthesis_task(
-    task_id: str,
-    db: AsyncSession = Depends(get_db)
-):
-    """获取数据合成任务详情"""
-    synthesis_task = await db.get(DataSynthInstance, task_id)
-    if not synthesis_task:
-        raise BusinessError(ErrorCodes.GENERATION_TASK_NOT_FOUND, data={"task_id": task_id})
+async def get_synthesis_task(task_id: str, db: AsyncSession = Depends(get_db)):
+    """
+    获取数据合成任务详情
 
-    task_item = DataSynthesisTaskItem(
-        id=synthesis_task.id,
-        name=synthesis_task.name,
-        description=synthesis_task.description,
-        status=synthesis_task.status,
-        synthesis_type=synthesis_task.synth_type,
-        total_files=synthesis_task.total_files,
-        created_at=synthesis_task.created_at,
-        updated_at=synthesis_task.updated_at,
-        created_by=synthesis_task.created_by,
-        updated_by=synthesis_task.updated_by,
-    )
-    return SuccessResponse(data=task_item)
+    Args:
+        task_id: 任务 ID
+        db: 数据库会话
 
+    Returns:
+        任务详情
+    """
+    query_service = TaskQueryService(db)
+    task = await query_service.get_task(task_id)
+    if not task:
+        raise BusinessError(ErrorCodes.GENERATION_TASK_NOT_FOUND, data={"task_id": task_id})
+    return SuccessResponse(data=DataSynthesisTaskItem.from_orm_with_config(task))
 
-@router.get("/tasks", response_model=StandardResponse[PagedDataSynthesisTaskResponse], status_code=200)
+
+@router.get("/tasks", response_model=StandardResponse[PagedDataSynthesisTaskResponse])
 async def list_synthesis_tasks(
     page: int = 1,
     page_size: int = 10,
@@ -139,116 +122,42 @@ async def list_synthesis_tasks(
     name: str | None = None,
     db: AsyncSession = Depends(get_db)
 ):
-    """分页列出所有数据合成任务，默认按创建时间倒序"""
-    query = select(DataSynthInstance)
-    if synthesis_type:
-        query = query.filter(DataSynthInstance.synth_type == synthesis_type)
-    if status:
-        query = query.filter(DataSynthInstance.status == status)
-    if name:
-        query = query.filter(DataSynthInstance.name.like(f"%{name}%"))
-
-    # 默认按创建时间倒序排列
-    query = query.order_by(DataSynthInstance.created_at.desc())
-
-    count_q = select(func.count()).select_from(query.subquery())
-    total = (await db.execute(count_q)).scalar_one()
-
-    if page < 1:
-        page = 1
-    if page_size < 1:
-        page_size = 10
-
-    result = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
-    rows = result.scalars().all()
-
-    task_items: list[DataSynthesisTaskItem] = []
-    for row in rows:
-        synth_cfg = getattr(row, "synth_config", {}) or {}
-        text_split_cfg = synth_cfg.get("text_split_config") or {}
-        synthesis_cfg = synth_cfg.get("synthesis_config") or {}
-        source_file_ids = synth_cfg.get("source_file_id") or []
-        model_id = synth_cfg.get("model_id")
-        result_location = synth_cfg.get("result_data_location")
-
-        task_items.append(
-            DataSynthesisTaskItem(
-                id=str(row.id),
-                name=str(row.name),
-                description=cast(str | None, row.description),
-                status=cast(str | None, row.status),
-                synthesis_type=str(row.synth_type),
-                model_id=model_id or "",
-                progress=int(cast(int, row.progress)),
-                result_data_location=result_location,
-                text_split_config=text_split_cfg,
-                synthesis_config=synthesis_cfg,
-                source_file_id=list(source_file_ids),
-                total_files=int(cast(int, row.total_files)),
-                processed_files=int(cast(int, row.processed_files)),
-                total_chunks=int(cast(int, row.total_chunks)),
-                processed_chunks=int(cast(int, row.processed_chunks)),
-                total_synthesis_data=int(cast(int, row.total_synth_data)),
-                created_at=row.created_at,
-                updated_at=row.updated_at,
-                created_by=row.created_by,
-                updated_by=row.updated_by,
-            )
-        )
-
-    paged = PagedDataSynthesisTaskResponse(
-        content=task_items,
-        totalElements=total,
-        totalPages=(total + page_size - 1) // page_size,
+    """
+    分页列出所有数据合成任务
+
+    Args:
+        page: 页码
+        page_size: 每页数量
+        synthesis_type: 合成类型过滤
+        status: 状态过滤
+        name: 名称模糊搜索
+        db: 数据库会话
+
+    Returns:
+        分页任务列表
+    """
+    query_service = TaskQueryService(db)
+    result = await query_service.list_tasks(
         page=page,
-        size=page_size,
+        page_size=page_size,
+        synthesis_type=synthesis_type,
+        status=status,
+        name=name,
     )
-
-    return SuccessResponse(data=paged)
+    return SuccessResponse(data=result)
 
 
 @router.delete("/task/{task_id}", response_model=StandardResponse)
-async def delete_synthesis_task(
-    task_id: str,
-    db: AsyncSession = Depends(get_db)
-):
-    """删除数据合成任务"""
-    task = await db.get(DataSynthInstance, task_id)
-    if not task:
-        raise BusinessError(ErrorCodes.GENERATION_TASK_NOT_FOUND, data={"task_id": task_id})
-
-    # 1. 删除与该任务相关的 SynthesisData、Chunk、File 记录
-    # 先查出所有文件任务 ID
-    file_result = await db.execute(
-        select(DataSynthesisFileInstance.id).where(
-            DataSynthesisFileInstance.synthesis_instance_id == task_id
-        )
-    )
-    file_ids = [row[0] for row in file_result.all()]
-
-    if file_ids:
-        # 删除 SynthesisData（根据文件任务ID）
-        await db.execute(delete(SynthesisData).where(
-                SynthesisData.synthesis_file_instance_id.in_(file_ids)
-            )
-        )
-
-        # 删除 Chunk 记录
-        await db.execute(delete(DataSynthesisChunkInstance).where(
-                DataSynthesisChunkInstance.synthesis_file_instance_id.in_(file_ids)
-            )
-        )
-
-        # 删除文件任务记录
-        await db.execute(delete(DataSynthesisFileInstance).where(
-                DataSynthesisFileInstance.id.in_(file_ids)
-            )
-        )
-
-    # 2. 删除任务本身
-    await db.delete(task)
-    await db.commit()
+async def delete_synthesis_task(task_id: str, db: AsyncSession = Depends(get_db)):
+    """
+    删除数据合成任务
 
+    Args:
+        task_id: 任务 ID
+        db: 数据库会话
+    """
+    delete_service = TaskDeleteService(db)
+    await delete_service.delete_task(task_id)
     return SuccessResponse(data=None)
 
 
@@ -258,211 +167,168 @@ async def delete_synthesis_file_task(
     file_id: str,
     db: AsyncSession = Depends(get_db)
 ):
-    """删除数据合成任务中的文件任务，同时刷新任务表中的文件/切片数量"""
-    # 先获取任务和文件任务记录
-    task = await db.get(DataSynthInstance, task_id)
-    if not task:
-        raise BusinessError(ErrorCodes.GENERATION_TASK_NOT_FOUND, data={"task_id": task_id})
-
-    file_task = await db.get(DataSynthesisFileInstance, file_id)
-    if not file_task:
-        raise BusinessError(ErrorCodes.GENERATION_FILE_NOT_FOUND, data={"file_id": file_id})
-
-    # 删除 SynthesisData（根据文件任务ID）
-    await db.execute(
-        delete(SynthesisData).where(
-            SynthesisData.synthesis_file_instance_id == file_id
-        )
-    )
-
-    # 删除 Chunk 记录
-    await db.execute(delete(DataSynthesisChunkInstance).where(
-            DataSynthesisChunkInstance.synthesis_file_instance_id == file_id
-        )
-    )
+    """
+    删除数据合成任务中的文件任务
 
-    # 删除文件任务记录
-    await db.execute(
-        delete(DataSynthesisFileInstance).where(
-            DataSynthesisFileInstance.id == file_id
-        )
-    )
+    Args:
+        task_id: 任务 ID
+        file_id: 文件任务 ID
+        db: 数据库会话
+    """
+    delete_service = TaskDeleteService(db)
+    await delete_service.delete_file_task(task_id, file_id)
+    return SuccessResponse(data=None)
 
-    # 刷新任务级别统计字段：总文件数、总文本块数、已处理文本块数
-    if task.total_files and task.total_files > 0:
-        task.total_files -= 1
-        if task.total_files < 0:
-            task.total_files = 0
 
-    await db.commit()
-    await db.refresh(task)
+# ==================== 提示词接口 ====================
 
-    return SuccessResponse(data=None)
+@router.get("/prompt", response_model=StandardResponse[str])
+async def get_prompt_by_type(synth_type: SynthesisType):
+    """
+    获取提示词
 
+    Args:
+        synth_type: 合成类型
 
-@router.get("/prompt", response_model=StandardResponse[str])
-async def get_prompt_by_type(
-    synth_type: SynthesisType,
-):
+    Returns:
+        提示词内容
+    """
     prompt = get_prompt(synth_type)
     return SuccessResponse(data=prompt)
 
 
-@router.get("/task/{task_id}/files", response_model=StandardResponse[PagedDataSynthesisFileTaskResponse])
+# ==================== 文件任务接口 ====================
+
+@router.get(
+    "/task/{task_id}/files",
+    response_model=StandardResponse[PagedDataSynthesisFileTaskResponse]
+)
 async def list_synthesis_file_tasks(
     task_id: str,
     page: int = 1,
     page_size: int = 10,
     db: AsyncSession = Depends(get_db),
 ):
-    """分页获取某个数据合成任务下的文件任务列表"""
-    # 先校验任务是否存在
-    task = await db.get(DataSynthInstance, task_id)
-    if not task:
-        raise BusinessError(ErrorCodes.GENERATION_TASK_NOT_FOUND, data={"task_id": task_id})
-
-    base_query = select(DataSynthesisFileInstance).where(
-        DataSynthesisFileInstance.synthesis_instance_id == task_id
-    )
+    """
+    分页获取任务下的文件任务列表
 
-    count_q = select(func.count()).select_from(base_query.subquery())
-    total = (await db.execute(count_q)).scalar_one()
+    Args:
+        task_id: 任务 ID
+        page: 页码
+        page_size: 每页数量
+        db: 数据库会话
 
-    if page < 1:
-        page = 1
-    if page_size < 1:
-        page_size = 10
+    Returns:
+        分页文件任务列表
+    """
+    query_service = TaskQueryService(db)
+    task = await query_service.get_task(task_id)
+    if not task:
+        raise BusinessError(ErrorCodes.GENERATION_TASK_NOT_FOUND, data={"task_id": task_id})
 
-    result = await db.execute(
-        base_query.offset((page - 1) * page_size).limit(page_size)
-    )
-    rows = result.scalars().all()
-
-    file_items = [
-        DataSynthesisFileTaskItem(
-            id=row.id,
-            synthesis_instance_id=row.synthesis_instance_id,
-            file_name=row.file_name,
-            source_file_id=row.source_file_id,
-            status=row.status,
-            total_chunks=row.total_chunks,
-            processed_chunks=row.processed_chunks,
-            created_at=row.created_at,
-            updated_at=row.updated_at,
-            created_by=row.created_by,
-            updated_by=row.updated_by,
-        )
-        for row in rows
-    ]
+    file_query = FileQueryService(db)
+    items, total = await file_query.list_by_task(task_id, page, page_size)
 
-    paged = PagedDataSynthesisFileTaskResponse(
-        content=file_items,
+    return SuccessResponse(data=PagedDataSynthesisFileTaskResponse(
+        content=items,
         totalElements=total,
         totalPages=(total + page_size - 1) // page_size,
         page=page,
         size=page_size,
-    )
+    ))
 
-    return SuccessResponse(data=paged)
 
+# ==================== 切片接口 ====================
 
-@router.get("/file/{file_id}/chunks", response_model=StandardResponse[PagedDataSynthesisChunkResponse])
+@router.get(
+    "/file/{file_id}/chunks",
+    response_model=StandardResponse[PagedDataSynthesisChunkResponse]
+)
 async def list_chunks_by_file(
     file_id: str,
     page: int = 1,
     page_size: int = 10,
     db: AsyncSession = Depends(get_db),
 ):
-    """根据文件任务 ID 分页查询 chunk 记录"""
-    # 校验文件任务是否存在
-    file_task = await db.get(DataSynthesisFileInstance, file_id)
-    if not file_task:
-        raise BusinessError(ErrorCodes.GENERATION_FILE_NOT_FOUND, data={"file_id": file_id})
-
-    base_query = select(DataSynthesisChunkInstance).where(
-        DataSynthesisChunkInstance.synthesis_file_instance_id == file_id
-    )
+    """
+    根据文件任务 ID 分页查询切片列表
 
-    count_q = select(func.count()).select_from(base_query.subquery())
-    total = (await db.execute(count_q)).scalar_one()
+    Args:
+        file_id: 文件任务 ID
+        page: 页码
+        page_size: 每页数量
+        db: 数据库会话
 
-    if page < 1:
-        page = 1
-    if page_size < 1:
-        page_size = 10
+    Returns:
+        分页切片列表
+    """
+    file_query = FileQueryService(db)
+    file_task = await file_query.get_by_id(file_id)
+    if not file_task:
+        raise BusinessError(ErrorCodes.GENERATION_FILE_NOT_FOUND, data={"file_id": file_id})
 
-    result = await db.execute(
-        base_query.order_by(DataSynthesisChunkInstance.chunk_index.asc())
-        .offset((page - 1) * page_size)
-        .limit(page_size)
-    )
-    rows = result.scalars().all()
-
-    chunk_items = [
-        DataSynthesisChunkItem(
-            id=row.id,
-            synthesis_file_instance_id=row.synthesis_file_instance_id,
-            chunk_index=row.chunk_index,
-            chunk_content=row.chunk_content,
-            chunk_metadata=getattr(row, "chunk_metadata", None),
-        )
-        for row in rows
-    ]
+    chunk_query = ChunkQueryService(db)
+    items, total = await chunk_query.list_by_file(file_id, page, page_size)
 
-    paged = PagedDataSynthesisChunkResponse(
-        content=chunk_items,
+    return SuccessResponse(data=PagedDataSynthesisChunkResponse(
+        content=items,
         totalElements=total,
         totalPages=(total + page_size - 1) // page_size,
         page=page,
         size=page_size,
-    )
+    ))
 
-    return SuccessResponse(data=paged)
 
+@router.get(
+    "/chunk/{chunk_id}/data",
+    response_model=StandardResponse[List[SynthesisDataItem]]
+)
+async def list_synthesis_data_by_chunk(chunk_id: str, db: AsyncSession = Depends(get_db)):
+    """
+    根据切片 ID 查询所有合成结果数据
 
-@router.get("/chunk/{chunk_id}/data", response_model=StandardResponse[list[SynthesisDataItem]])
-async def list_synthesis_data_by_chunk(
-    chunk_id: str,
-    db: AsyncSession = Depends(get_db),
-):
-    """根据 chunk ID 查询所有合成结果数据"""
-    # 可选：校验 chunk 是否存在
-    chunk = await db.get(DataSynthesisChunkInstance, chunk_id)
+    Args:
+        chunk_id: 切片 ID
+        db: 数据库会话
+
+    Returns:
+        合成数据列表
+    """
+    chunk_query = ChunkQueryService(db)
+    chunk = await chunk_query.get_by_id(chunk_id)
     if not chunk:
         raise BusinessError(ErrorCodes.GENERATION_CHUNK_NOT_FOUND, data={"chunk_id": chunk_id})
 
-    result = await db.execute(
-        select(SynthesisData).where(SynthesisData.chunk_instance_id == chunk_id)
-    )
-    rows = result.scalars().all()
-
-    items = [
-        SynthesisDataItem(
-            id=row.id,
-            data=row.data,
-            synthesis_file_instance_id=row.synthesis_file_instance_id,
-            chunk_instance_id=row.chunk_instance_id,
-        )
-        for row in rows
-    ]
-
+    data_query = SynthesisDataQueryService(db)
+    items = await data_query.list_by_chunk(chunk_id)
     return SuccessResponse(data=items)
 
 
-@router.post("/task/{task_id}/export-dataset/{dataset_id}", response_model=StandardResponse[str])
+# ==================== 导出接口 ====================
+
+@router.post(
+    "/task/{task_id}/export-dataset/{dataset_id}",
+    response_model=StandardResponse[str]
+)
 async def export_synthesis_task_to_dataset(
     task_id: str,
     dataset_id: str,
+    format: str = "alpaca",
     db: AsyncSession = Depends(get_db),
 ):
-    """将指定合成任务的全部合成数据归档到已有数据集中。
+    """
+    将合成任务的数据归档到已有数据集
+
+    Args:
+        task_id: 任务 ID
+        dataset_id: 目标数据集 ID
+        format: 导出格式
+        db: 数据库会话
 
-    规则：
-    - 以原始文件为维度，每个原始文件生成一个 JSONL 文件；
-    - JSONL 文件名称与原始文件名称完全一致；
-    - 仅写入文件，不再创建数据集。
+    Returns:
+        数据集 ID
     """
-    exporter = SynthesisDatasetExporter(db)
+    exporter = SynthesisDatasetExporter(db, format=format)
     generation = GenerationService(db)
     try:
         dataset = await exporter.export_task_to_dataset(task_id, dataset_id)
@@ -470,59 +336,73 @@ async def export_synthesis_task_to_dataset(
     except SynthesisExportError as e:
         logger.error(
             "Failed to export synthesis task %s to dataset %s: %s",
-            task_id,
-            dataset_id,
-            e,
+            task_id, dataset_id, e
         )
         raise BusinessError(ErrorCodes.OPERATION_FAILED, data={"error": str(e)})
 
     return SuccessResponse(data=dataset.id)
 
 
-@router.delete("/chunk/{chunk_id}", response_model=StandardResponse)
-async def delete_chunk_with_data(
-    chunk_id: str,
+@router.post(
+    "/data/export",
+    response_model=StandardResponse[ExportSynthesisDataResponse],
+)
+async def export_synthesis_data(
+    request: ExportSynthesisDataRequest,
     db: AsyncSession = Depends(get_db),
 ):
-    """删除单条 t_data_synthesis_chunk_instances 记录及其关联的所有 t_data_synthesis_data"""
-    chunk = await db.get(DataSynthesisChunkInstance, chunk_id)
-    if not chunk:
-        raise BusinessError(ErrorCodes.GENERATION_CHUNK_NOT_FOUND, data={"chunk_id": chunk_id})
+    """
+    将合成数据导出为指定格式的 JSONL 文件
 
-    # 先删除与该 chunk 关联的合成数据
-    await db.execute(
-        delete(SynthesisData).where(SynthesisData.chunk_instance_id == chunk_id)
-    )
+    Args:
+        request: 导出请求
+        db: 数据库会话
 
-    # 再删除 chunk 本身
-    await db.execute(
-        delete(DataSynthesisChunkInstance).where(
-            DataSynthesisChunkInstance.id == chunk_id
-        )
+    Returns:
+        导出结果
+    """
+    exporter = SynthesisDatasetExporter(
+        db,
+        format=request.format.value,
+        output_path=request.output_path,
     )
+    result = await exporter.export_data(
+        task_id=request.task_id,
+        file_instance_ids=request.file_instance_ids,
+    )
+    return SuccessResponse(data=result)
+
+
+# ==================== 删除接口 ====================
 
-    await db.commit()
+@router.delete("/chunk/{chunk_id}", response_model=StandardResponse)
+async def delete_chunk_with_data(chunk_id: str, db: AsyncSession = Depends(get_db)):
+    """
+    删除切片及其关联的合成数据
 
+    Args:
+        chunk_id: 切片 ID
+        db: 数据库会话
+    """
+    delete_service = TaskDeleteService(db)
+    await delete_service.delete_chunk(chunk_id)
     return SuccessResponse(data=None)
 
 
 @router.delete("/chunk/{chunk_id}/data", response_model=StandardResponse)
-async def delete_synthesis_data_by_chunk(
-    chunk_id: str,
-    db: AsyncSession = Depends(get_db),
-):
-    """仅删除指定 chunk 下的全部 t_data_synthesis_data 记录，返回删除条数"""
-    chunk = await db.get(DataSynthesisChunkInstance, chunk_id)
-    if not chunk:
-        raise BusinessError(ErrorCodes.GENERATION_CHUNK_NOT_FOUND, data={"chunk_id": chunk_id})
-
-    result = await db.execute(
-        delete(SynthesisData).where(SynthesisData.chunk_instance_id == chunk_id)
-    )
-    deleted = int(getattr(result, "rowcount", 0) or 0)
+async def delete_synthesis_data_by_chunk(chunk_id: str, db: AsyncSession = Depends(get_db)):
+    """
+    删除切片下的所有合成数据
 
-    await db.commit()
+    Args:
+        chunk_id: 切片 ID
+        db: 数据库会话
 
+    Returns:
+        删除的记录数
+    """
+    delete_service = TaskDeleteService(db)
+    deleted = await delete_service.delete_synthesis_data_by_chunk(chunk_id)
     return SuccessResponse(data=deleted)
 
 
@@ -531,16 +411,18 @@ async def batch_delete_synthesis_data(
     request: BatchDeleteSynthesisDataRequest,
     db: AsyncSession = Depends(get_db),
 ):
-    """批量删除 t_data_synthesis_data 记录"""
-    if not request.ids:
-        return SuccessResponse(data=0)
+    """
+    批量删除合成数据
 
-    result = await db.execute(
-        delete(SynthesisData).where(SynthesisData.id.in_(request.ids))
-    )
-    deleted = int(getattr(result, "rowcount", 0) or 0)
-    await db.commit()
+    Args:
+        request: 批量删除请求
+        db: 数据库会话
 
+    Returns:
+        删除的记录数
+    """
+    delete_service = TaskDeleteService(db)
+    deleted = await delete_service.batch_delete_synthesis_data(request.ids)
     return SuccessResponse(data=deleted)
 
 
@@ -550,27 +432,138 @@ async def update_synthesis_data_field(
     body: SynthesisDataUpdateRequest,
     db: AsyncSession = Depends(get_db),
 ):
-    """修改单条 t_data_synthesis_data.data 的完整 JSON
+    """
+    修改合成数据
+
+    Args:
+        data_id: 数据 ID
+        body: 更新请求
+        db: 数据库会话
 
-    前端传入完整 JSON，后端直接覆盖原有 data 字段，不做局部 merge。
+    Returns:
+        更新后的数据
     """
-    record = await db.get(SynthesisData, data_id)
+    data_query = SynthesisDataQueryService(db)
+    record = await data_query.update(data_id, body.data)
     if not record:
         raise BusinessError(ErrorCodes.GENERATION_DATA_NOT_FOUND, data={"data_id": data_id})
+    return SuccessResponse(data=record)
 
-    # 直接整体覆盖 data 字段
-    record.data = body.data
-
-    await db.commit()
-    await db.refresh(record)
-
-    return StandardResponse(
-        code="0",
-        message="success",
-        data=SynthesisDataItem(
-            id=record.id,
-            data=record.data,
-            synthesis_file_instance_id=record.synthesis_file_instance_id,
-            chunk_instance_id=record.chunk_instance_id,
-        ),
-    )
+
+@router.post(
+    "/batch-delete/chunks",
+    response_model=StandardResponse,
+)
+async def batch_delete_chunk_instances(
+    request: BatchDeleteChunkInstancesRequest,
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    批量删除切片
+
+    Args:
+        request: 批量删除请求
+        db: 数据库会话
+    """
+    delete_service = TaskDeleteService(db)
+    await delete_service.batch_delete_chunks(request.chunk_ids)
+    return SuccessResponse(data=None)
+
+
+@router.post(
+    "/batch-delete/chunks-by-file",
+    response_model=StandardResponse,
+)
+async def batch_delete_chunk_instances_by_file(
+    request: BatchDeleteChunkInstancesByFileRequest,
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    按文件任务维度删除切片
+
+    Args:
+        request: 删除请求
+        db: 数据库会话
+    """
+    file_query = FileQueryService(db)
+    file_task = await file_query.get_by_id(request.file_id)
+    if not file_task:
+        raise BusinessError(ErrorCodes.GENERATION_FILE_NOT_FOUND)
+
+    delete_service = TaskDeleteService(db)
+    await delete_service.delete_chunks_by_file(request.file_id)
+    return SuccessResponse(data=None)
+
+
+@router.post(
+    "/batch-delete/chunks-by-task",
+    response_model=StandardResponse,
+)
+async def batch_delete_chunk_instances_by_task(
+    request: BatchDeleteChunkInstancesByTaskRequest,
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    按任务维度删除切片
+
+    Args:
+        request: 删除请求
+        db: 数据库会话
+    """
+    query_service = TaskQueryService(db)
+    task = await query_service.get_task(request.task_id)
+    if not task:
+        raise BusinessError(ErrorCodes.GENERATION_TASK_NOT_FOUND)
+
+    delete_service = TaskDeleteService(db)
+    await delete_service.delete_chunks_by_task(request.task_id)
+    return SuccessResponse(data=None)
+
+
+# ==================== 批量查询接口 ====================
+
+@router.post(
+    "/data/patch",
+    response_model=StandardResponse[List[SynthesisDataPatchItem]],
+)
+async def get_synthesis_data_patch(
+    task_id: Optional[str] = None,
+    file_instance_id: Optional[str] = None,
+    chunk_instance_id: Optional[str] = None,
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    批量查询合成数据
+
+    Args:
+        task_id: 任务 ID
+        file_instance_id: 文件任务 ID
+        chunk_instance_id: 切片 ID
+        db: 数据库会话
+
+    Returns:
+        合成数据列表
+    """
+    data_query = SynthesisDataQueryService(db)
+
+    if chunk_instance_id:
+        items = await data_query.list_by_chunk(chunk_instance_id)
+        return SuccessResponse(data=[SynthesisDataPatchItem.model_validate(item) for item in items])
+
+    if file_instance_id:
+        file_query = FileQueryService(db)
+        file_task = await file_query.get_by_id(file_instance_id)
+        if not file_task:
+            raise BusinessError(ErrorCodes.GENERATION_FILE_NOT_FOUND)
+        items = await data_query.list_by_file(file_instance_id)
+        return SuccessResponse(data=items)
+
+    if task_id:
+        query_service = TaskQueryService(db)
+        task = await query_service.get_task(task_id)
+        if not task:
+            raise BusinessError(ErrorCodes.GENERATION_TASK_NOT_FOUND)
+        items = await data_query.list_by_task(task_id)
+        return SuccessResponse(data=items)
+
+    return SuccessResponse(data=[])
diff --git a/runtime/datamate-python/app/module/generation/schema/generation.py b/runtime/datamate-python/app/module/generation/schema/generation.py
index 0c0b35262..ff5332c4e 100644
--- a/runtime/datamate-python/app/module/generation/schema/generation.py
+++ b/runtime/datamate-python/app/module/generation/schema/generation.py
@@ -1,6 +1,6 @@
 from datetime import datetime
 from enum import Enum
-from typing import List, Optional, Dict, Any
+from typing import Any, List, Optional, Dict
 
 from pydantic import BaseModel, Field, field_validator, ConfigDict
 
@@ -68,8 +68,53 @@ class DataSynthesisTaskItem(BaseModel):
     created_by: Optional[str] = None
     updated_by: Optional[str] = None
 
+    # 扩展字段（带默认值，用于列表查询等复杂场景）
+    model_id: str = ""
+    progress: int = 0
+    result_data_location: Optional[str] = None
+    text_split_config: Optional[Dict[str, Any]] = None
+    synthesis_config: Optional[Dict[str, Any]] = None
+    source_file_id: List[str] = []
+    processed_files: int = 0
+    total_chunks: int = 0
+    processed_chunks: int = 0
+    total_synthesis_data: int = 0
+
     model_config = ConfigDict(from_attributes=True)
 
+    @classmethod
+    def from_orm_with_config(cls, task: Any) -> "DataSynthesisTaskItem":
+        """从 ORM 对象转换，包含完整配置（用于列表查询）"""
+        synth_cfg = getattr(task, "synth_config", {}) or {}
+        text_split_cfg = synth_cfg.get("text_split_config") or {}
+        synthesis_cfg = synth_cfg.get("synthesis_config") or {}
+        source_file_ids = synth_cfg.get("source_file_id") or []
+        model_id = synth_cfg.get("model_id", "")
+        result_location = synth_cfg.get("result_data_location")
+
+        return cls(
+            id=str(task.id),
+            name=str(task.name),
+            description=task.description,
+            status=task.status,
+            synthesis_type=str(task.synth_type),
+            model_id=model_id or "",
+            progress=int(task.progress or 0),
+            result_data_location=result_location,
+            text_split_config=text_split_cfg,
+            synthesis_config=synthesis_cfg,
+            source_file_id=list(source_file_ids),
+            total_files=int(task.total_files or 0),
+            processed_files=int(task.processed_files or 0),
+            total_chunks=int(task.total_chunks or 0),
+            processed_chunks=int(task.processed_chunks or 0),
+            total_synthesis_data=int(task.total_synth_data or 0),
+            created_at=task.created_at,
+            updated_at=task.updated_at,
+            created_by=task.created_by,
+            updated_by=task.updated_by,
+        )
+
 
 class PagedDataSynthesisTaskResponse(BaseModel):
     """分页数据合成任务响应"""
@@ -175,3 +220,25 @@ class SynthesisDataPatchItem(BaseModel):
     synthesis_file_instance_id: str
 
     model_config = ConfigDict(from_attributes=True)
+
+
+class ExportFormat(str, Enum):
+    """导出格式枚举"""
+    ALPACA = "alpaca"
+    SHAREGPT = "sharegpt"
+    RAW = "raw"
+
+
+class ExportSynthesisDataRequest(BaseModel):
+    """导出合成数据请求"""
+    task_id: str = Field(..., description="合成任务ID")
+    file_instance_ids: Optional[List[str]] = Field(None, description="文件实例ID列表，为空则导出全部")
+    format: ExportFormat = Field(ExportFormat.ALPACA, description="导出格式")
+    output_path: Optional[str] = Field(None, description="输出路径，为空则使用默认路径")
+
+
+class ExportSynthesisDataResponse(BaseModel):
+    """导出合成数据响应"""
+    file_paths: List[str] = Field(..., description="导出文件路径列表")
+    total_records: int = Field(..., description="总记录数")
+    format: str = Field(..., description="导出格式")
diff --git a/runtime/datamate-python/app/module/generation/service/chunk_processor.py b/runtime/datamate-python/app/module/generation/service/chunk_processor.py
new file mode 100644
index 000000000..0fcebf707
--- /dev/null
+++ b/runtime/datamate-python/app/module/generation/service/chunk_processor.py
@@ -0,0 +1,109 @@
+"""切片处理器模块 - 负责文本切片和持久化操作"""
+import uuid
+from typing import Any
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models.data_synthesis import (
+    DataSynthInstance,
+    DataSynthesisFileInstance,
+    DataSynthesisChunkInstance,
+)
+from app.db.session import logger
+
+
+class ChunkProcessor:
+    """切片处理器"""
+
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def persist_chunks(
+        self,
+        synthesis_task: DataSynthInstance,
+        file_task: DataSynthesisFileInstance,
+        file_id: str,
+        chunks: list[Any],
+    ) -> int:
+        """将切片结果保存到数据库。
+
+        Args:
+            synthesis_task: 合成任务实例
+            file_task: 文件任务实例
+            file_id: 原始文件ID
+            chunks: 切片后的文档列表
+
+        Returns:
+            保存的切片数量
+        """
+        for idx, doc in enumerate(chunks, start=1):
+            base_metadata = dict(getattr(doc, "metadata", {}) or {})
+            base_metadata.update({
+                "task_id": str(synthesis_task.id),
+                "file_id": file_id
+            })
+
+            chunk_record = DataSynthesisChunkInstance(
+                id=str(uuid.uuid4()),
+                synthesis_file_instance_id=file_task.id,
+                chunk_index=idx,
+                chunk_content=doc.page_content,
+                chunk_metadata=base_metadata,
+            )
+            self.db.add(chunk_record)
+
+        file_task.total_chunks = len(chunks)
+        file_task.status = "processing"
+
+        await self.db.commit()
+        await self.db.refresh(file_task)
+
+        logger.info(f"Persisted {len(chunks)} chunks for file_task={file_task.id}")
+        return len(chunks)
+
+    async def load_chunk_batch(
+        self,
+        file_task_id: str,
+        start_index: int,
+        end_index: int,
+    ) -> list[DataSynthesisChunkInstance]:
+        """按索引范围加载指定文件任务下的一批切片记录。
+
+        Args:
+            file_task_id: 文件任务ID
+            start_index: 起始索引（包含）
+            end_index: 结束索引（包含）
+
+        Returns:
+            切片实例列表
+        """
+        from sqlalchemy import select
+
+        result = await self.db.execute(
+            select(DataSynthesisChunkInstance)
+            .where(
+                DataSynthesisChunkInstance.synthesis_file_instance_id == file_task_id,
+                DataSynthesisChunkInstance.chunk_index >= start_index,
+                DataSynthesisChunkInstance.chunk_index <= end_index,
+            )
+            .order_by(DataSynthesisChunkInstance.chunk_index.asc())
+        )
+        return list(result.scalars().all())
+
+    async def count_chunks_for_file(self, synth_file_instance_id: str) -> int:
+        """统计指定任务与文件下的切片总数。
+
+        Args:
+            synth_file_instance_id: 文件实例ID
+
+        Returns:
+            切片总数
+        """
+        from sqlalchemy import func, select
+
+        result = await self.db.execute(
+            select(func.count(DataSynthesisChunkInstance.id)).where(
+                DataSynthesisChunkInstance.synthesis_file_instance_id == synth_file_instance_id
+            )
+        )
+        return int(result.scalar() or 0)
diff --git a/runtime/datamate-python/app/module/generation/service/chunk_query_service.py b/runtime/datamate-python/app/module/generation/service/chunk_query_service.py
new file mode 100644
index 000000000..7868f1021
--- /dev/null
+++ b/runtime/datamate-python/app/module/generation/service/chunk_query_service.py
@@ -0,0 +1,67 @@
+"""
+切片查询服务 - 负责切片相关的查询操作
+"""
+from typing import List, Tuple
+
+from sqlalchemy import select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models.data_synthesis import DataSynthesisChunkInstance
+from app.module.generation.schema.generation import (
+    DataSynthesisChunkItem,
+    PagedDataSynthesisChunkResponse,
+)
+
+
+class ChunkQueryService:
+    """切片查询服务 - 负责切片相关的查询操作"""
+
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def get_by_id(self, chunk_id: str) -> DataSynthesisChunkInstance | None:
+        """
+        根据 ID 获取切片
+
+        Args:
+            chunk_id: 切片 ID
+
+        Returns:
+            切片实例，不存在则返回 None
+        """
+        return await self.db.get(DataSynthesisChunkInstance, chunk_id)
+
+    async def list_by_file(
+        self, file_id: str, page: int, page_size: int
+    ) -> Tuple[List[DataSynthesisChunkItem], int]:
+        """
+        分页查询文件任务下的切片列表
+
+        Args:
+            file_id: 文件任务 ID
+            page: 页码
+            page_size: 每页数量
+
+        Returns:
+            (切片列表, 总数)
+        """
+        base_query = select(DataSynthesisChunkInstance).where(
+            DataSynthesisChunkInstance.synthesis_file_instance_id == file_id
+        )
+
+        count_result = await self.db.execute(
+            select(func.count()).select_from(base_query.subquery())
+        )
+        total = count_result.scalar_one() or 0
+
+        offset = max(page, 1) - 1
+        limit = max(page_size, 1)
+        result = await self.db.execute(
+            base_query.order_by(DataSynthesisChunkInstance.chunk_index.asc())
+            .offset(offset * limit)
+            .limit(limit)
+        )
+        rows = result.scalars().all()
+
+        items = [DataSynthesisChunkItem.model_validate(row) for row in rows]
+        return items, total
diff --git a/runtime/datamate-python/app/module/generation/service/export_service.py b/runtime/datamate-python/app/module/generation/service/export_service.py
index ee3eb24e1..4bba6c2d4 100644
--- a/runtime/datamate-python/app/module/generation/service/export_service.py
+++ b/runtime/datamate-python/app/module/generation/service/export_service.py
@@ -1,8 +1,10 @@
+"""
+数据导出服务 - 负责将合成数据导出为各种格式
+"""
 import datetime
 import json
 import os
-import time
-from typing import Iterable, List, Sequence, cast
+from typing import Iterable, List, Optional, Sequence
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -14,34 +16,50 @@
     SynthesisData,
 )
 from app.db.models.dataset_management import Dataset, DatasetFiles
+from app.module.generation.schema.generation import (
+    ExportSynthesisDataResponse,
+)
 
 logger = get_logger(__name__)
 
 
 class SynthesisExportError(Exception):
-    """Raised when exporting synthesis data to dataset fails."""
+    """导出失败时抛出的异常"""
 
 
 class SynthesisDatasetExporter:
-    """Export synthesis data of a task into an existing dataset.
+    """
+    数据导出器 - 将合成数据导出为各种格式
 
-    Export rules:
-    - Dimension: original file (DatasetFiles)
-    - One JSONL file per original file
-    - JSONL file name is exactly the same as the original file name
+    导出规则：
+    - 维度：原始文件 (DatasetFiles)
+    - 每个原始文件生成一个 JSONL 文件
+    - JSONL 文件名称与原始文件名称一致
+    - 支持格式转换：alpaca、sharegpt、raw
     """
 
-    def __init__(self, db: AsyncSession):
+    SUPPORTED_FORMATS = ["alpaca", "sharegpt", "raw"]
+    DEFAULT_FORMAT = "alpaca"
+
+    def __init__(self, db: AsyncSession, format: str = "alpaca", output_path: Optional[str] = None):
         self._db = db
+        self._format = format if format in self.SUPPORTED_FORMATS else self.DEFAULT_FORMAT
+        self._output_path = output_path
 
     async def export_task_to_dataset(
         self,
         task_id: str,
         dataset_id: str,
     ) -> Dataset:
-        """Export the full synthesis data of the given task into an existing dataset.
+        """
+        将任务的全部合成数据导出到已有数据集
+
+        Args:
+            task_id: 任务 ID
+            dataset_id: 目标数据集 ID
 
-        Optimized to process one file at a time to reduce memory usage.
+        Returns:
+            更新后的数据集
         """
         task = await self._db.get(DataSynthInstance, task_id)
         if not task:
@@ -60,22 +78,19 @@ async def export_task_to_dataset(
         created_files: list[DatasetFiles] = []
         total_size = 0
 
-        # 一个文件一个文件处理，避免一次性加载所有合成数据
         for file_instance in file_instances:
             records = await self._load_synthesis_data_for_file(file_instance.id)
             if not records:
                 continue
 
-            # 归档文件名称：原始文件名称.xxx -> 原始文件名称.jsonl
             original_name = file_instance.file_name or "unknown"
             base_name, _ = os.path.splitext(original_name)
             archived_file_name = f"{base_name}.jsonl"
 
             file_path = os.path.join(base_path, archived_file_name)
             os.makedirs(os.path.dirname(file_path), exist_ok=True)
-            self._write_jsonl(file_path, records)
+            self._write_jsonl(file_path, records, self._format)
 
-            # 计算文件大小
             try:
                 file_size = os.path.getsize(file_path)
             except OSError:
@@ -93,7 +108,6 @@ async def export_task_to_dataset(
             created_files.append(df)
             total_size += file_size
 
-        # 更新数据集的文件数、总大小和状态
         if created_files:
             dataset.file_count = (dataset.file_count or 0) + len(created_files)
             dataset.size_bytes = (dataset.size_bytes or 0) + total_size
@@ -111,7 +125,58 @@ async def export_task_to_dataset(
 
         return dataset
 
+    async def export_data(
+        self,
+        task_id: str,
+        file_instance_ids: Optional[List[str]] = None,
+    ) -> ExportSynthesisDataResponse:
+        """
+        将合成数据导出为指定格式的 JSONL 文件
+
+        Args:
+            task_id: 任务 ID
+            file_instance_ids: 文件实例 ID 列表，为空则导出全部
+
+        Returns:
+            导出结果
+        """
+        if file_instance_ids:
+            file_instances = await self._load_specific_file_instances(file_instance_ids)
+        else:
+            file_instances = await self._load_file_instances(task_id)
+
+        if not file_instances:
+            raise SynthesisExportError("No file instances to export")
+
+        output_dir = self._output_path or self._get_default_output_dir()
+        os.makedirs(output_dir, exist_ok=True)
+
+        file_paths: List[str] = []
+        total_records = 0
+
+        for file_instance in file_instances:
+            records = await self._load_synthesis_data_for_file(file_instance.id)
+            if not records:
+                continue
+
+            original_name = file_instance.file_name or "unknown"
+            base_name, _ = os.path.splitext(original_name)
+            output_file_name = f"{base_name}.jsonl"
+
+            file_path = os.path.join(output_dir, output_file_name)
+            self._write_jsonl(file_path, records, self._format)
+
+            file_paths.append(file_path)
+            total_records += len(records)
+
+        return ExportSynthesisDataResponse(
+            file_paths=file_paths,
+            total_records=total_records,
+            format=self._format,
+        )
+
     async def _load_file_instances(self, task_id: str) -> Sequence[DataSynthesisFileInstance]:
+        """加载任务下的所有文件实例"""
         result = await self._db.execute(
             select(DataSynthesisFileInstance).where(
                 DataSynthesisFileInstance.synthesis_instance_id == task_id
@@ -119,13 +184,21 @@ async def _load_file_instances(self, task_id: str) -> Sequence[DataSynthesisFile
         )
         return result.scalars().all()
 
+    async def _load_specific_file_instances(
+        self, file_ids: List[str]
+    ) -> Sequence[DataSynthesisFileInstance]:
+        """加载指定的文件实例"""
+        result = await self._db.execute(
+            select(DataSynthesisFileInstance).where(
+                DataSynthesisFileInstance.id.in_(file_ids)
+            )
+        )
+        return result.scalars().all()
+
     async def _load_synthesis_data_for_file(
         self, file_instance_id: str
     ) -> List[dict]:
-        """Load all synthesis data records for a single file instance.
-
-        Each returned item is a plain JSON-serialisable dict based on SynthesisData.data.
-        """
+        """加载单个文件实例的所有合成数据"""
         result = await self._db.execute(
             select(SynthesisData).where(
                 SynthesisData.synthesis_file_instance_id == file_instance_id
@@ -139,21 +212,62 @@ async def _load_synthesis_data_for_file(
             records.append(payload)
         return records
 
-    @staticmethod
-    def _write_jsonl(path: str, records: Iterable[dict]) -> None:
+    def _write_jsonl(self, path: str, records: Iterable[dict], format: Optional[str] = None) -> None:
+        """写入 JSONL 文件"""
+        fmt = format or self._format
+        os.makedirs(os.path.dirname(path), exist_ok=True)
         with open(path, "w", encoding="utf-8") as f:
             for record in records:
-                f.write(json.dumps(record, ensure_ascii=False))
+                formatted = self._format_record(record, fmt)
+                f.write(json.dumps(formatted, ensure_ascii=False))
                 f.write("\n")
 
+    def _format_record(self, record: dict, format: str) -> dict:
+        """根据格式转换记录"""
+        if format == "alpaca":
+            return self._format_as_alpaca(record)
+        elif format == "sharegpt":
+            return self._format_as_sharegpt(record)
+        else:
+            return self._format_as_raw(record)
+
+    def _format_as_alpaca(self, record: dict) -> dict:
+        """转换为 Alpaca 格式"""
+        return {
+            "instruction": record.get("instruction", ""),
+            "output": record.get("output", ""),
+        }
+
+    def _format_as_sharegpt(self, record: dict) -> dict:
+        """转换为 ShareGPT 格式"""
+        instruction = record.get("instruction", "")
+        output = record.get("output", "")
+        input_text = record.get("input", "")
+
+        conversations = [
+            {"from": "human", "value": instruction},
+            {"from": "gpt", "value": output},
+        ]
+
+        result = {"conversations": conversations}
+
+        if input_text:
+            result["context"] = input_text
+
+        return result
+
+    def _format_as_raw(self, record: dict) -> dict:
+        """原始格式"""
+        return record
+
+    def _get_default_output_dir(self) -> str:
+        """获取默认输出目录"""
+        import tempfile
+        return tempfile.gettempdir()
+
     @staticmethod
     def _ensure_dataset_path(dataset: Dataset) -> str:
-        """Ensure dataset.path is available and the directory exists.
-
-        The actual value of dataset.path should come from Dataset's default
-        path generation logic or external configuration, not from the
-        synthesis task's result_data_location.
-        """
+        """确保数据集路径存在"""
         if not dataset.path:
             raise SynthesisExportError("Dataset path is empty")
         os.makedirs(dataset.path, exist_ok=True)
diff --git a/runtime/datamate-python/app/module/generation/service/file_query_service.py b/runtime/datamate-python/app/module/generation/service/file_query_service.py
new file mode 100644
index 000000000..78c26520e
--- /dev/null
+++ b/runtime/datamate-python/app/module/generation/service/file_query_service.py
@@ -0,0 +1,65 @@
+"""
+文件查询服务 - 负责文件任务相关的查询操作
+"""
+from typing import List, Tuple
+
+from sqlalchemy import select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models.data_synthesis import DataSynthesisFileInstance
+from app.module.generation.schema.generation import (
+    DataSynthesisFileTaskItem,
+    PagedDataSynthesisFileTaskResponse,
+)
+
+
+class FileQueryService:
+    """文件查询服务 - 负责文件任务相关的查询操作"""
+
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def get_by_id(self, file_id: str) -> DataSynthesisFileInstance | None:
+        """
+        根据 ID 获取文件任务
+
+        Args:
+            file_id: 文件任务 ID
+
+        Returns:
+            文件任务实例，不存在则返回 None
+        """
+        return await self.db.get(DataSynthesisFileInstance, file_id)
+
+    async def list_by_task(
+        self, task_id: str, page: int, page_size: int
+    ) -> Tuple[List[DataSynthesisFileTaskItem], int]:
+        """
+        分页查询任务下的文件任务列表
+
+        Args:
+            task_id: 任务 ID
+            page: 页码
+            page_size: 每页数量
+
+        Returns:
+            (文件任务列表, 总数)
+        """
+        base_query = select(DataSynthesisFileInstance).where(
+            DataSynthesisFileInstance.synthesis_instance_id == task_id
+        )
+
+        count_result = await self.db.execute(
+            select(func.count()).select_from(base_query.subquery())
+        )
+        total = count_result.scalar_one() or 0
+
+        offset = max(page, 1) - 1
+        limit = max(page_size, 1)
+        result = await self.db.execute(
+            base_query.offset(offset * limit).limit(limit)
+        )
+        rows = result.scalars().all()
+
+        items = [DataSynthesisFileTaskItem.model_validate(row) for row in rows]
+        return items, total
diff --git a/runtime/datamate-python/app/module/generation/service/generation_service.py b/runtime/datamate-python/app/module/generation/service/generation_service.py
index 3dec30df6..13342d230 100644
--- a/runtime/datamate-python/app/module/generation/service/generation_service.py
+++ b/runtime/datamate-python/app/module/generation/service/generation_service.py
@@ -1,12 +1,9 @@
+"""数据合成服务 - 核心业务逻辑层"""
 import asyncio
-import json
-import random
-import re
-import uuid
+from contextlib import asynccontextmanager
 
-from langchain_core.language_models import BaseChatModel
 from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from app.db.models.base_entity import LineageNode, LineageEdge
 from app.db.models.data_synthesis import (
@@ -16,133 +13,219 @@
     SynthesisData,
 )
 from app.db.models.dataset_management import DatasetFiles, Dataset
-from app.db.session import logger
+from app.db.session import logger, AsyncSessionLocal
 from app.module.generation.schema.generation import Config, SyntheConfig
-from app.module.generation.service.prompt import (
-    QUESTION_GENERATOR_PROMPT,
-    ANSWER_GENERATOR_PROMPT,
-)
+from app.module.generation.service.chunk_processor import ChunkProcessor
+from app.module.generation.service.qa_generator import QAGenerator
+from app.module.generation.service.qa_generator import _filter_docs_by_size
+from app.module.generation.service.task_executor import run_in_thread
 from app.module.shared.common.document_loaders import load_documents
 from app.module.shared.common.text_split import DocumentSplitter
-from app.module.shared.util.model_chat import extract_json_substring
 from app.module.shared.llm import LLMFactory
 from app.module.system.service.common_service import get_model_by_id
 from app.module.shared.common.lineage import LineageService
 from app.module.shared.schema import NodeType, EdgeType
 
 
-def _filter_docs(split_docs, chunk_size):
-    """
-    过滤文档，移除长度小于 chunk_size 的文档
-    """
-    filtered_docs = []
-    for doc in split_docs:
-        if len(doc.page_content) >= chunk_size * 0.7:
-            filtered_docs.append(doc)
-    return filtered_docs
-
-
-def extract_img_urls(doc):
-    """提取文档中的图片地址"""
-    pattern = r"!\[\]\((.*?)\)"
-    # 查找所有匹配的地址
-    img_urls = re.findall(pattern, doc)
-    return img_urls
-
 class GenerationService:
+    """数据合成服务 - 使用模块化架构"""
+
     def __init__(self, db: AsyncSession):
         self.db = db
-        # 全局并发信号量：保证任意时刻最多 10 次模型调用
-        self.question_semaphore = asyncio.Semaphore(20)
-        self.answer_semaphore = asyncio.Semaphore(100)
+        self.chunk_processor = ChunkProcessor(db)
+        self.qa_generator = QAGenerator(db)
+
+    @asynccontextmanager
+    async def _session_scope(self):
+        """
+        创建独立的数据库会话上下文管理器
+        用于长时间运行的任务，避免长时间占用连接
+        """
+        session = AsyncSessionLocal()
+        try:
+            yield session
+            await session.commit()
+        except Exception:
+            await session.rollback()
+            raise
+        finally:
+            await session.close()
+
+    def _load_and_split_sync(
+        self, file_path: str, chunk_size: int, chunk_overlap: int
+    ):
+        """
+        同步方法：加载并切分文件
+        在线程池中执行，不阻塞事件循环
+        """
+        try:
+            docs = load_documents(file_path)
+            split_docs = DocumentSplitter.auto_split(docs, chunk_size, chunk_overlap)
+            return _filter_docs_by_size(split_docs, chunk_size)
+        except Exception as e:
+            logger.error(f"Error loading or splitting file {file_path}: {e}")
+            raise
 
     async def process_task(self, task_id: str):
-        """处理数据合成任务入口：根据任务ID加载任务并逐个处理源文件。"""
-        synth_task: DataSynthInstance | None = await self.db.get(DataSynthInstance, task_id)
-        if not synth_task:
-            logger.error(f"Synthesis task {task_id} not found, abort processing")
-            return
+        """
+        处理数据合成任务入口 - 使用独立会话和线程池
+
+        Args:
+            task_id: 合成任务ID
+        """
+        # 使用独立的数据库会话，不与API请求共享
+        async with self._session_scope() as session:
+            synth_task: DataSynthInstance | None = await session.get(
+                DataSynthInstance, task_id
+            )
+            if not synth_task:
+                logger.error(f"Synthesis task {task_id} not found, abort processing")
+                return
+
+            logger.info(f"Start processing synthesis task {task_id}")
+
+            # 更新任务状态为处理中
+            synth_task.status = "processing"
+            await session.commit()
+
+            # 从 synth_config 中读取 max_qa_pairs
+            max_qa_pairs = self._parse_max_qa_pairs(synth_task)
+
+            # 获取任务关联的文件ID列表
+            file_ids = await self._get_file_ids_for_task(session, task_id)
+            if not file_ids:
+                logger.warning(
+                    f"No files associated with task {task_id}, abort processing"
+                )
+                await self._mark_task_failed(session, task_id, "no_files_associated")
+                return
+
+            # 逐个文件处理
+            processed_count = 0
+            total_task_chunks = 0
+            processed_task_chunks = 0
+            for file_id in file_ids:
+                try:
+                    file_task = await self._get_or_create_file_instance(
+                        session, str(synth_task.id), file_id
+                    )
+                    success = await self._process_single_file(
+                        session, synth_task, file_id, max_qa_pairs=max_qa_pairs
+                    )
+                    # 累加任务级别的切片统计
+                    if file_task:
+                        total_task_chunks += file_task.total_chunks or 0
+                        processed_task_chunks += file_task.processed_chunks or 0
+                except Exception as e:
+                    logger.exception(
+                        f"Unexpected error when processing file {file_id} for task {task_id}: {e}"
+                    )
+                    await self._mark_file_failed(
+                        session, str(synth_task.id), file_id, str(e)
+                    )
+                    success = False
+
+                if success:
+                    processed_count += 1
+                    synth_task.processed_files = processed_count
+                    # 更新任务级别的切片统计
+                    synth_task.total_chunks = total_task_chunks
+                    synth_task.processed_chunks = processed_task_chunks
+                    await session.commit()
+
+            # 更新最终任务状态
+            if processed_count == len(file_ids):
+                synth_task.status = "completed"
+            else:
+                synth_task.status = "partially_completed"
+            # 确保任务完成时切片统计准确
+            if synth_task.status in ("completed", "partially_completed"):
+                synth_task.total_chunks = total_task_chunks
+                synth_task.processed_chunks = processed_task_chunks
+            await session.commit()
 
-        logger.info(f"Start processing synthe task {task_id}")
+            logger.info(f"Finished processing synthesis task {synth_task.id}")
 
-        # 从 synth_config 中读取 max_qa_pairs，全局控制 QA 总量上限；<=0 或异常则视为不限制
+    def _parse_max_qa_pairs(self, synth_task: DataSynthInstance) -> int | None:
+        """解析最大QA对数量配置"""
         try:
             cfg = Config(**(synth_task.synth_config or {}))
-            max_qa_pairs = cfg.max_qa_pairs if (cfg and cfg.max_qa_pairs and cfg.max_qa_pairs > 0) else None
+            max_qa_pairs = (
+                cfg.max_qa_pairs
+                if (cfg and cfg.max_qa_pairs and cfg.max_qa_pairs > 0)
+                else None
+            )
         except Exception:
             max_qa_pairs = None
+        return max_qa_pairs
 
-        # 获取任务关联的文件原始ID列表
-        file_ids = await self._get_file_ids_for_task(task_id)
-        if not file_ids:
-            logger.warning(f"No files associated with task {task_id}, abort processing")
-            return
-
-        # 逐个文件处理
-        for file_id in file_ids:
-            try:
-                success = await self._process_single_file(synth_task, file_id, max_qa_pairs=max_qa_pairs)
-            except Exception as e:
-                logger.exception(f"Unexpected error when processing file {file_id} for task {task_id}: {e}")
-                # 确保对应文件任务状态标记为失败
-                await self._mark_file_failed(str(synth_task.id), file_id, str(e))
-                success = False
-
-            if success:
-                # 每处理完一个文件，简单增加 processed_files 计数
-                synth_task.processed_files = (synth_task.processed_files or 0) + 1
-                await self.db.commit()
-                await self.db.refresh(synth_task)
-
-        logger.info(f"Finished processing synthesis task {synth_task.id}")
-
-    # ==================== 高层文件处理流程 ====================
     async def _process_single_file(
         self,
+        session: AsyncSession,
         synth_task: DataSynthInstance,
         file_id: str,
         max_qa_pairs: int | None = None,
     ) -> bool:
-        """按 chunk 批量流式处理单个源文件。
-
-        流程：
-        1. 切片并将所有 chunk 持久化到 DB 后释放内存；
-        2. 从 DB 按 chunk_index 升序批量读取 chunk；
-        3. 对批次中的每个 chunk：先生成指定数量的问题，再基于这些问题生成答案；
-        4. 每成功处理完一个 chunk（即该 chunk 至少生成一条 QA）就更新一次 processed_chunks；
-        5. 全部完成后将文件实例标记为 completed。
         """
-        # 解析文件路径与配置
-        file_path = await self._resolve_file_path(file_id)
+        处理单个源文件 - 使用线程池执行阻塞操作
+
+        Args:
+            session: 数据库会话
+            synth_task: 合成任务实例
+            file_id: 源文件ID
+            max_qa_pairs: 最大QA对数量限制
+
+        Returns:
+            处理是否成功
+        """
+        # 解析文件路径
+        file_path = await self._resolve_file_path(session, file_id)
         if not file_path:
             logger.warning(f"File path not found for file_id={file_id}, skip")
-            await self._mark_file_failed(str(synth_task.id), file_id, "file_path_not_found")
+            await self._mark_file_failed(
+                session, str(synth_task.id), file_id, "file_path_not_found"
+            )
             return False
 
         logger.info(f"Processing file_id={file_id}, path={file_path}")
 
+        # 解析配置
         try:
             config = Config(**(synth_task.synth_config or {}))
         except Exception as e:
             logger.error(f"Invalid synth_config for task={synth_task.id}: {e}")
-            await self._mark_file_failed(str(synth_task.id), file_id, "invalid_synth_config")
+            await self._mark_file_failed(
+                session, str(synth_task.id), file_id, "invalid_synth_config"
+            )
+            return False
+
+        # 1. 加载并切片 - 在线程池中执行阻塞操作
+        try:
+            chunks = await run_in_thread(
+                self._load_and_split_sync,
+                file_path,
+                config.text_split_config.chunk_size,
+                config.text_split_config.chunk_overlap,
+            )
+        except Exception as e:
+            logger.error(f"Failed to load and split file {file_id}: {e}")
+            await self._mark_file_failed(
+                session, str(synth_task.id), file_id, f"load_split_failed: {e}"
+            )
             return False
 
-        # 1. 加载并切片（仅在此处占用内存）
-        chunks = self._load_and_split(
-            file_path,
-            config.text_split_config.chunk_size,
-            config.text_split_config.chunk_overlap,
-        )
         if not chunks:
             logger.warning(f"No chunks generated for file_id={file_id}")
-            await self._mark_file_failed(str(synth_task.id), file_id, "no_chunks_generated")
+            await self._mark_file_failed(
+                session, str(synth_task.id), file_id, "no_chunks_generated"
+            )
             return False
 
-        logger.info(f"File {file_id} split into {len(chunks)} chunks by LangChain")
+        logger.info(f"File {file_id} split into {len(chunks)} chunks")
 
-        # 2. 获取文件实例并持久化 chunk 记录
+        # 2. 获取文件实例并持久化切片
         file_task = await self._get_or_create_file_instance(
+            session,
             synthesis_task_id=str(synth_task.id),
             source_file_id=file_id,
         )
@@ -150,31 +233,36 @@ async def _process_single_file(
             logger.error(
                 f"DataSynthesisFileInstance not found for task={synth_task.id}, file_id={file_id}"
             )
-            await self._mark_file_failed(str(synth_task.id), file_id, "file_instance_not_found")
+            await self._mark_file_failed(
+                session, str(synth_task.id), file_id, "file_instance_not_found"
+            )
             return False
 
-        await self._persist_chunks(synth_task, file_task, file_id, chunks)
+        # 使用 ChunkProcessor 持久化切片
+        chunk_processor = ChunkProcessor(session)
+        await chunk_processor.persist_chunks(synth_task, file_task, file_id, chunks)
         total_chunks = len(chunks)
-        # 释放内存中的切片
-        del chunks
+        del chunks  # 释放内存
 
-        # 3. 读取问答配置
+        # 3. 验证问答配置
         question_cfg: SyntheConfig | None = config.question_synth_config
         answer_cfg: SyntheConfig | None = config.answer_synth_config
         if not question_cfg or not answer_cfg:
             logger.error(
                 f"Question/Answer synth config missing for task={synth_task.id}, file={file_id}"
             )
-            await self._mark_file_failed(str(synth_task.id), file_id, "qa_config_missing")
+            await self._mark_file_failed(
+                session, str(synth_task.id), file_id, "qa_config_missing"
+            )
             return False
 
         logger.info(
             f"Start QA generation for task={synth_task.id}, file={file_id}, total_chunks={total_chunks}"
         )
 
-        # 为本文件构建模型 client
-        question_model = await get_model_by_id(self.db, question_cfg.model_id)
-        answer_model = await get_model_by_id(self.db, answer_cfg.model_id)
+        # 4. 创建模型客户端
+        question_model = await get_model_by_id(session, question_cfg.model_id)
+        answer_model = await get_model_by_id(session, answer_cfg.model_id)
         question_chat = LLMFactory.create_chat(
             question_model.model_name, question_model.base_url, question_model.api_key
         )
@@ -182,312 +270,171 @@ async def _process_single_file(
             answer_model.model_name, answer_model.base_url, answer_model.api_key
         )
 
-        # 分批次从 DB 读取并处理 chunk
-        batch_size = 100
+        # 5. 分批次处理切片
+        await self._process_chunks_in_batches(
+            session=session,
+            file_task=file_task,
+            total_chunks=total_chunks,
+            question_cfg=question_cfg,
+            answer_cfg=answer_cfg,
+            question_chat=question_chat,
+            answer_chat=answer_chat,
+            synth_task_id=str(synth_task.id),
+            max_qa_pairs=max_qa_pairs,
+        )
+
+        # 6. 标记文件任务完成
+        file_task.status = "completed"
+        await session.commit()
+        await session.refresh(file_task)
+        return True
+
+    async def _process_chunks_in_batches(
+        self,
+        session: AsyncSession,
+        file_task: DataSynthesisFileInstance,
+        total_chunks: int,
+        question_cfg: SyntheConfig,
+        answer_cfg: SyntheConfig,
+        question_chat,
+        answer_chat,
+        synth_task_id: str,
+        max_qa_pairs: int | None,
+    ) -> None:
+        """分批次处理切片
+
+        Args:
+            session: 数据库会话
+            file_task: 文件任务实例
+            total_chunks: 总切片数
+            question_cfg: 问题生成配置
+            answer_cfg: 答案生成配置
+            question_chat: 问题生成模型
+            answer_chat: 答案生成模型
+            synth_task_id: 合成任务ID
+            max_qa_pairs: 最大QA对数量限制
+        """
+        batch_size = 50  # 减小批次大小，降低内存占用
         current_index = 1
 
+        chunk_processor = ChunkProcessor(session)
+        qa_generator = QAGenerator(session)
+
         while current_index <= total_chunks:
             end_index = min(current_index + batch_size - 1, total_chunks)
-            chunk_batch = await self._load_chunk_batch(
+
+            # 使用 ChunkProcessor 加载切片批次
+            chunk_batch = await chunk_processor.load_chunk_batch(
                 file_task_id=file_task.id,
                 start_index=current_index,
                 end_index=end_index,
             )
+
             if not chunk_batch:
                 logger.warning(
-                    f"Empty chunk batch loaded for file={file_id}, range=[{current_index}, {end_index}]"
+                    f"Empty chunk batch loaded for file={file_task.id}, range=[{current_index}, {end_index}]"
                 )
                 current_index = end_index + 1
                 continue
 
-            # 对本批中的每个 chunk 并发处理（内部受 semaphore 限流）
-            async def process_one(chunk: DataSynthesisChunkInstance) -> bool:
-                return await self._process_single_chunk_qa(
-                    file_task=file_task,
-                    chunk=chunk,
-                    question_cfg=question_cfg,
-                    answer_cfg=answer_cfg,
-                    question_chat=question_chat,
-                    answer_chat=answer_chat,
-                    synth_task_id=str(synth_task.id),
-                    max_qa_pairs=max_qa_pairs,
-                )
-
-            tasks = [process_one(chunk) for chunk in chunk_batch]
-            await asyncio.gather(*tasks, return_exceptions=True)
+            # 顺序处理每个切片，避免并发过高
+            for chunk in chunk_batch:
+                try:
+                    success = await self._process_single_chunk(
+                        session=session,
+                        file_task=file_task,
+                        chunk=chunk,
+                        question_cfg=question_cfg,
+                        answer_cfg=answer_cfg,
+                        question_chat=question_chat,
+                        answer_chat=answer_chat,
+                        synth_task_id=synth_task_id,
+                        max_qa_pairs=max_qa_pairs,
+                        qa_generator=qa_generator,
+                    )
+                    if success:
+                        # 更新已处理的切片计数
+                        file_task.processed_chunks = (file_task.processed_chunks or 0) + 1
+                        # 实时刷新进度到数据库
+                        await session.commit()
+                        await session.refresh(file_task)
+                        logger.info(f"File {file_task.id} progress: {file_task.processed_chunks}/{file_task.total_chunks} chunks processed")
+                except Exception as e:
+                    logger.error(f"Error processing chunk {chunk.id}: {e}")
+                    # 即使处理失败，也要更新进度
+                    file_task.processed_chunks = (file_task.processed_chunks or 0) + 1
+                    await session.commit()
+                    await session.refresh(file_task)
+
+                # 短暂释放控制权，让出事件循环，处理其他请求
+                await asyncio.sleep(0)
 
             current_index = end_index + 1
 
-        # 全部完成
-        file_task.status = "completed"
-        await self.db.commit()
-        await self.db.refresh(file_task)
-        return True
-
-    async def _process_single_chunk_qa(
+    async def _process_single_chunk(
         self,
+        session: AsyncSession,
         file_task: DataSynthesisFileInstance,
         chunk: DataSynthesisChunkInstance,
         question_cfg: SyntheConfig,
         answer_cfg: SyntheConfig,
-        question_chat: BaseChatModel,
-        answer_chat: BaseChatModel,
+        question_chat,
+        answer_chat,
         synth_task_id: str,
-        max_qa_pairs: int | None = None,
+        max_qa_pairs: int | None,
+        qa_generator: QAGenerator,
     ) -> bool:
-        """处理单个 chunk：生成问题列表，然后为每个问题生成答案并落库。
-
-        为了全局控制 QA 总量：在本方法开始处，根据 synth_task_id 查询当前已落盘的
-        SynthesisData 条数，如果 >= max_qa_pairs，则不再对当前 chunk 做任何 QA 生成，
-        并将当前文件任务标记为 completed，processed_chunks = total_chunks。
+        """处理单个切片
 
-        已经进入后续流程的任务（例如其它协程正在生成答案）允许自然执行完。
+        Args:
+            session: 数据库会话
+            file_task: 文件任务实例
+            chunk: 切片实例
+            question_cfg: 问题生成配置
+            answer_cfg: 答案生成配置
+            question_chat: 问题生成模型
+            answer_chat: 答案生成模型
+            synth_task_id: 合成任务ID
+            max_qa_pairs: 最大QA对数量限制
+            qa_generator: QA生成器
+
+        Returns:
+            处理是否成功
         """
-        # 随机决定是否对当前 chunk 进行 QA 生成
-        if random.random() > question_cfg.temperature:
-            logger.info(
-                f"Skip QA generation for chunk_index={chunk.chunk_index} in file_task={file_task.id} due to random decision."
-            )
-            # 更新文件任务的 processed_chunks 计数
-            await self._increment_processed_chunks(file_task.id, 1)
-            return False
-
-        # 如果没有全局上限配置，维持原有行为
-        if max_qa_pairs is not None and max_qa_pairs > 0:
-            from sqlalchemy import func
-
-            # 统计当前整个任务下已生成的 QA 总数
-            result = await self.db.execute(
-                select(func.count(SynthesisData.id)).where(
-                    SynthesisData.synthesis_file_instance_id.in_(
-                        select(DataSynthesisFileInstance.id).where(
-                            DataSynthesisFileInstance.synthesis_instance_id == synth_task_id
-                        )
-                    )
-                )
-            )
-            current_qa_count = int(result.scalar() or 0)
-
-            if current_qa_count >= max_qa_pairs:
-                logger.info(
-                    "max_qa_pairs reached: current=%s, max=%s, task_id=%s, file_task_id=%s, skip new QA generation for this chunk.",
-                    current_qa_count,
-                    max_qa_pairs,
-                    synth_task_id,
-                    file_task.id,
-                )
-                # 将文件任务标记为已完成，并认为所有 chunk 均已处理
+        # 检查QA对数量上限
+        if max_qa_pairs and max_qa_pairs > 0:
+            if await qa_generator.check_qa_limit(synth_task_id, max_qa_pairs):
+                logger.info(f"max_qa_pairs reached, skipping chunk {chunk.chunk_index}")
                 file_task.status = "completed"
                 if file_task.total_chunks is not None:
                     file_task.processed_chunks = file_task.total_chunks
-                await self.db.commit()
-                await self.db.refresh(file_task)
+                await session.commit()
+                await session.refresh(file_task)
                 return False
 
-        # ---- 下面保持原有逻辑不变 ----
-        chunk_index = chunk.chunk_index
-        chunk_text = chunk.chunk_content or ""
-        if not chunk_text.strip():
-            logger.warning(
-                f"Empty chunk text for file_task={file_task.id}, chunk_index={chunk_index}"
-            )
-            # 无论成功或失败，均视为该 chunk 已处理完成
-            try:
-                await self._increment_processed_chunks(file_task.id, 1)
-            except Exception as e:
-                logger.exception(
-                    f"Failed to increment processed_chunks for file_task={file_task.id}, chunk_index={chunk_index}: {e}"
-                )
-            return False
-
-        success_any = False
-
-        # 1. 生成问题
-        try:
-            questions = await self._generate_questions_for_one_chunk(
-                chunk_text=chunk_text,
-                question_cfg=question_cfg,
-                question_chat=question_chat,
-            )
-        except Exception as e:
-            logger.error(
-                f"Generate questions failed for file_task={file_task.id}, chunk_index={chunk_index}: {e}"
-            )
-            questions = []
-
-        if not questions:
-            logger.info(
-                f"No questions generated for file_task={file_task.id}, chunk_index={chunk_index}"
-            )
-        else:
-            # 2. 针对每个问题生成答案并入库
-            qa_success = await self._generate_answers_for_one_chunk(
-                file_task=file_task,
-                chunk=chunk,
-                questions=questions,
-                answer_cfg=answer_cfg,
-                answer_chat=answer_chat,
-            )
-            success_any = bool(qa_success)
-
-        # 无论本 chunk 处理是否成功，都增加 processed_chunks 计数，避免任务长时间卡住
-        try:
-            await self._increment_processed_chunks(file_task.id, 1)
-        except Exception as e:
-            logger.exception(
-                f"Failed to increment processed_chunks for file_task={file_task.id}, chunk_index={chunk_index}: {e}"
-            )
-
-        return success_any
-
-    async def _generate_questions_for_one_chunk(
-        self,
-        chunk_text: str,
-        question_cfg: SyntheConfig,
-        question_chat: BaseChatModel,
-    ) -> list[str]:
-        """针对单个 chunk 文本，调用 question_chat 生成问题列表。"""
-        number = question_cfg.number or 5
-        number = number if number is not None else 5
-        number = max(int(len(chunk_text) / 1000 * number), 1)
-        template = getattr(question_cfg, "prompt_template", QUESTION_GENERATOR_PROMPT)
-        template = template if (template is not None and template.strip() != "") else QUESTION_GENERATOR_PROMPT
-
-        prompt = (
-            template
-            .replace("{text}", chunk_text)
-            .replace("{number}", str(number))
-            .replace("{textLength}", str(len(chunk_text)))
+        # 使用 QAGenerator 处理问答生成
+        success_count = await qa_generator.process_chunk_qa(
+            file_task=file_task,
+            chunk=chunk,
+            question_cfg=question_cfg,
+            answer_cfg=answer_cfg,
+            question_chat=question_chat,
+            answer_chat=answer_chat,
         )
 
-        async with self.question_semaphore:
-            loop = asyncio.get_running_loop()
-            raw_answer = await loop.run_in_executor(
-                None,
-                LLMFactory.invoke_sync,
-                question_chat,
-                prompt,
-            )
+        return success_count > 0
 
-        # 解析为问题列表
-        questions = self._parse_questions_from_answer(
-            raw_answer,
-        )
-        return questions
+    async def _resolve_file_path(self, session: AsyncSession, file_id: str) -> str | None:
+        """根据文件ID获取文件路径
 
-    async def _generate_answers_for_one_chunk(
-        self,
-        file_task: DataSynthesisFileInstance,
-        chunk: DataSynthesisChunkInstance,
-        questions: list[str],
-        answer_cfg: SyntheConfig,
-        answer_chat: BaseChatModel,
-    ) -> bool:
-        """为一个 chunk 的所有问题生成答案并写入 SynthesisData。
+        Args:
+            session: 数据库会话
+            file_id: 文件ID
 
-        返回：是否至少成功写入一条 QA。
+        Returns:
+            文件路径
         """
-        if not questions:
-            return False
-
-        chunk_text = chunk.chunk_content or ""
-        template = getattr(answer_cfg, "prompt_template", ANSWER_GENERATOR_PROMPT)
-        template = template if (template is not None and template.strip() != "") else ANSWER_GENERATOR_PROMPT
-        extra_vars = getattr(answer_cfg, "extra_prompt_vars", {}) or {}
-
-        success_flags: list[bool] = []
-
-        async def process_single_question(question: str):
-            prompt = template.replace("{text}", chunk_text).replace("{question}", question)
-            for k, v in extra_vars.items():
-                prompt.replace(f"{{{{{k}}}}}", str(v))
-            else:
-                prompt_local = prompt
-
-            async with self.answer_semaphore:
-                loop = asyncio.get_running_loop()
-                answer = await loop.run_in_executor(
-                    None,
-                    LLMFactory.invoke_sync,
-                    answer_chat,
-                    prompt_local,
-                )
-
-            # 默认结构：与 ANSWER_GENERATOR_PROMPT 一致，并补充 instruction 字段
-            base_obj: dict[str, object] = {
-                "input": chunk_text,
-                "output": answer,
-            }
-
-            # 如果模型已经按照 ANSWER_GENERATOR_PROMPT 返回了 JSON，则尝试解析并在其上增加 instruction
-            parsed_obj: dict[str, object] | None = None
-            if isinstance(answer, str):
-                cleaned = extract_json_substring(answer)
-                try:
-                  parsed = json.loads(cleaned)
-                  if isinstance(parsed, dict):
-                      parsed_obj = parsed
-                except Exception:
-                  parsed_obj = None
-
-            if parsed_obj is not None:
-                parsed_obj["instruction"] = question
-                data_obj = parsed_obj
-            else:
-                base_obj["instruction"] = question
-                data_obj = base_obj
-
-            # 提取图片URL
-            img_urls = extract_img_urls(chunk_text)
-            if img_urls:
-                data_obj["img_urls"] = img_urls
-
-            record = SynthesisData(
-                id=str(uuid.uuid4()),
-                data=data_obj,
-                synthesis_file_instance_id=file_task.id,
-                chunk_instance_id=chunk.id,
-            )
-            self.db.add(record)
-            success_flags.append(True)
-
-        tasks = [process_single_question(q) for q in questions]
-        await asyncio.gather(*tasks, return_exceptions=True)
-
-        if success_flags:
-            await self.db.commit()
-            return True
-        return False
-
-    @staticmethod
-    def _parse_questions_from_answer(
-        raw_answer: str,
-    ) -> list[str]:
-        """从大模型返回中解析问题数组。"""
-        if not raw_answer:
-            return []
-
-        cleaned = extract_json_substring(raw_answer)
-        try:
-            data = json.loads(cleaned)
-        except Exception as e:
-            logger.error(
-                f"Failed to parse question list JSON for task: {e}. "
-            )
-            return []
-
-        if isinstance(data, list):
-            return [str(q) for q in data if isinstance(q, str) and q.strip()]
-        # 容错：如果是单个字符串
-        if isinstance(data, str) and data.strip():
-            return [data.strip()]
-        return []
-
-
-    # ==================== 原有辅助方法（文件路径/切片/持久化等） ====================
-    async def _resolve_file_path(self, file_id: str) -> str | None:
-        """根据文件ID查询 t_dm_dataset_files 并返回 file_path（仅 ACTIVE 文件）。"""
-        result = await self.db.execute(
+        result = await session.execute(
             select(DatasetFiles).where(DatasetFiles.id == file_id)
         )
         file_obj = result.scalar_one_or_none()
@@ -495,86 +442,47 @@ async def _resolve_file_path(self, file_id: str) -> str | None:
             return None
         return file_obj.file_path
 
-    @staticmethod
-    def _load_and_split(file_path: str, chunk_size: int, chunk_overlap: int):
-        """使用 LangChain 加载文本并进行切片，直接返回 Document 列表。
-        Args:
-            file_path: 待切片的文件路径
-            chunk_size: 切片大小
-            chunk_overlap: 切片重叠大小
-        """
-        try:
-            docs = load_documents(file_path)
-            split_docs = DocumentSplitter.auto_split(docs, chunk_size, chunk_overlap)
-            return _filter_docs(split_docs, chunk_size)
-        except Exception as e:
-            logger.error(f"Error loading or splitting file {file_path}: {e}")
-            raise
-
-    async def _persist_chunks(
-        self,
-        synthesis_task: DataSynthInstance,
-        file_task: DataSynthesisFileInstance,
-        file_id: str,
-        chunks,
-    ) -> None:
-        """将切片结果保存到 t_data_synthesis_chunk_instances，并更新文件级分块计数。"""
-        for idx, doc in enumerate(chunks, start=1):
-            # 先复制原始 Document.metadata，再在其上追加任务相关字段，避免覆盖原有元数据
-            base_metadata = dict(getattr(doc, "metadata", {}) or {})
-            base_metadata.update(
-                {
-                    "task_id": str(synthesis_task.id),
-                    "file_id": file_id
-                }
-            )
-
-            chunk_record = DataSynthesisChunkInstance(
-                id=str(uuid.uuid4()),
-                synthesis_file_instance_id=file_task.id,
-                chunk_index=idx,
-                chunk_content=doc.page_content,
-                chunk_metadata=base_metadata,
-            )
-            self.db.add(chunk_record)
-
-        # 更新文件任务的分块数量
-        file_task.total_chunks = len(chunks)
-        file_task.status = "processing"
-
-        await self.db.commit()
-        await self.db.refresh(file_task)
-
     async def _get_or_create_file_instance(
         self,
+        session: AsyncSession,
         synthesis_task_id: str,
         source_file_id: str,
-    ) -> DataSynthesisFileInstance:
-        """根据任务ID和原始文件ID，查找或创建对应的 DataSynthesisFileInstance 记录。
+    ) -> DataSynthesisFileInstance | None:
+        """获取或创建文件任务实例
 
-        - 如果已存在（同一任务 + 同一 source_file_id），直接返回；
-        - 如果不存在，则创建一条新的文件任务记录，file_name 来自文件路径，
-          target_file_location 先复用任务的 result_data_location。
+        Args:
+            session: 数据库会话
+            synthesis_task_id: 合成任务ID
+            source_file_id: 源文件ID
+
+        Returns:
+            文件任务实例
         """
-        # 尝试查询已有文件任务记录
-        result = await self.db.execute(
+        result = await session.execute(
             select(DataSynthesisFileInstance).where(
                 DataSynthesisFileInstance.synthesis_instance_id == synthesis_task_id,
                 DataSynthesisFileInstance.source_file_id == source_file_id,
             )
         )
-        file_task = result.scalar_one_or_none()
-        return file_task
+        return result.scalar_one_or_none()
 
-    async def _mark_file_failed(self, synth_task_id: str, file_id: str, reason: str | None = None) -> None:
-        """将指定任务下的单个文件任务标记为失败状态，兜底错误处理。
+    async def _mark_file_failed(
+        self,
+        session: AsyncSession,
+        synth_task_id: str,
+        file_id: str,
+        reason: str | None = None,
+    ) -> None:
+        """标记文件任务失败
 
-        - 如果找到对应的 DataSynthesisFileInstance，则更新其 status="failed"。
-        - 如果未找到，则静默返回，仅记录日志。
-        - reason 参数仅用于日志记录，方便排查。
+        Args:
+            session: 数据库会话
+            synth_task_id: 合成任务ID
+            file_id: 文件ID
+            reason: 失败原因
         """
         try:
-            result = await self.db.execute(
+            result = await session.execute(
                 select(DataSynthesisFileInstance).where(
                     DataSynthesisFileInstance.synthesis_instance_id == synth_task_id,
                     DataSynthesisFileInstance.source_file_id == file_id,
@@ -588,93 +496,83 @@ async def _mark_file_failed(self, synth_task_id: str, file_id: str, reason: str
                 return
 
             file_task.status = "failed"
-            await self.db.commit()
-            await self.db.refresh(file_task)
+            await session.commit()
+            await session.refresh(file_task)
             logger.info(
                 f"Marked file task as failed for task={synth_task_id}, file_id={file_id}, reason={reason}"
             )
         except Exception as e:
-            # 兜底日志，避免异常向外传播影响其它文件处理
             logger.exception(
                 f"Unexpected error when marking file failed for task={synth_task_id}, file_id={file_id}, original_reason={reason}, error={e}"
             )
 
-    async def _get_file_ids_for_task(self, synth_task_id: str):
-        """根据任务ID查询关联的文件原始ID列表"""
-        result = await self.db.execute(
-            select(DataSynthesisFileInstance.source_file_id)
-            .where(DataSynthesisFileInstance.synthesis_instance_id == synth_task_id)
-        )
-        file_ids = result.scalars().all()
-        return file_ids
+    async def _get_file_ids_for_task(
+        self, session: AsyncSession, synth_task_id: str
+    ):
+        """获取任务关联的文件ID列表
 
-    # ========== 新增：chunk 计数与批量加载、processed_chunks 安全更新辅助方法 ==========
-    async def _count_chunks_for_file(self, synth_file_instance_id: str) -> int:
-        """统计指定任务与文件下的 chunk 总数。"""
-        from sqlalchemy import func
+        Args:
+            session: 数据库会话
+            synth_task_id: 合成任务ID
 
-        result = await self.db.execute(
-            select(func.count(DataSynthesisChunkInstance.id)).where(
-                DataSynthesisChunkInstance.synthesis_file_instance_id == synth_file_instance_id
+        Returns:
+            文件ID列表
+        """
+        result = await session.execute(
+            select(DataSynthesisFileInstance.source_file_id).where(
+                DataSynthesisFileInstance.synthesis_instance_id == synth_task_id
             )
         )
-        return int(result.scalar() or 0)
+        return result.scalars().all()
 
-    async def _load_chunk_batch(
+    async def _mark_task_failed(
         self,
-        file_task_id: str,
-        start_index: int,
-        end_index: int,
-    ) -> list[DataSynthesisChunkInstance]:
-        """按索引范围加载指定文件任务下的一批 chunk 记录（含边界）。"""
-        result = await self.db.execute(
-            select(DataSynthesisChunkInstance)
-            .where(
-                DataSynthesisChunkInstance.synthesis_file_instance_id == file_task_id,
-                DataSynthesisChunkInstance.chunk_index >= start_index,
-                DataSynthesisChunkInstance.chunk_index <= end_index,
-            )
-            .order_by(DataSynthesisChunkInstance.chunk_index.asc())
-        )
-        return list(result.scalars().all())
-
-    async def _increment_processed_chunks(self, file_task_id: str, delta: int) -> None:
-        result = await self.db.execute(
-            select(DataSynthesisFileInstance).where(
-                DataSynthesisFileInstance.id == file_task_id,
-            )
-        )
-        file_task = result.scalar_one_or_none()
-        if not file_task:
-            logger.error(f"Failed to increment processed_chunks: file_task {file_task_id} not found")
-            return
-
-        # 原始自增
-        new_value = (file_task.processed_chunks or 0) + int(delta)
+        session: AsyncSession,
+        task_id: str,
+        reason: str | None = None,
+    ) -> None:
+        """标记任务失败
 
-        # 如果存在 total_chunks，上限为 total_chunks，避免超过
-        total = file_task.total_chunks
-        if isinstance(total, int) and total >= 0:
-            new_value = min(new_value, total)
+        Args:
+            session: 数据库会话
+            task_id: 任务ID
+            reason: 失败原因
+        """
+        try:
+            task = await session.get(DataSynthInstance, task_id)
+            if task:
+                task.status = "failed"
+                await session.commit()
+                logger.info(
+                    f"Marked task {task_id} as failed, reason={reason}"
+                )
+        except Exception as e:
+            logger.exception(f"Error marking task {task_id} as failed: {e}")
 
-        file_task.processed_chunks = new_value
-        await self.db.commit()
-        await self.db.refresh(file_task)
+    async def add_synthesis_to_graph(
+        self, db: AsyncSession, task_id: str, dest_dataset_id: str
+    ) -> None:
+        """记录数据合成血缘关系
 
-    async def add_synthesis_to_graph(self, db: AsyncSession, task_id: str, dest_dataset_id: str) -> None:
-        """记录数据合成血缘关系：源数据集 -> 合成数据集 via DATA_SYNTHESIS"""
+        Args:
+            db: 数据库会话
+            task_id: 任务ID
+            dest_dataset_id: 目标数据集ID
+        """
         try:
-            # 获取任务和目标数据集信息
-            task = await self.db.get(DataSynthInstance, task_id)
+            task = await db.get(DataSynthInstance, task_id)
             src_dataset_result = await db.execute(
                 select(DatasetFiles.dataset_id)
-                .join(DataSynthesisFileInstance, DatasetFiles.id == DataSynthesisFileInstance.source_file_id)
+                .join(
+                    DataSynthesisFileInstance,
+                    DatasetFiles.id == DataSynthesisFileInstance.source_file_id,
+                )
                 .where(DataSynthesisFileInstance.synthesis_instance_id == task_id)
                 .limit(1)
             )
             src_dataset_id = src_dataset_result.scalar_one_or_none()
-            src_dataset = await self.db.get(Dataset, src_dataset_id)
-            dst_dataset = await self.db.get(Dataset, dest_dataset_id)
+            src_dataset = await db.get(Dataset, src_dataset_id)
+            dst_dataset = await db.get(Dataset, dest_dataset_id)
 
             if not task or not dst_dataset:
                 logger.warning("Missing task or destination dataset for lineage graph")
@@ -684,13 +582,13 @@ async def add_synthesis_to_graph(self, db: AsyncSession, task_id: str, dest_data
                 id=src_dataset.id,
                 node_type=NodeType.DATASET.value,
                 name=src_dataset.name,
-                description=src_dataset.description
+                description=src_dataset.description,
             )
             dest_node = LineageNode(
                 id=dst_dataset.id,
                 node_type=NodeType.DATASET.value,
                 name=dst_dataset.name,
-                description=dst_dataset.description
+                description=dst_dataset.description,
             )
             synthesis_edge = LineageEdge(
                 process_id=task_id,
@@ -698,13 +596,12 @@ async def add_synthesis_to_graph(self, db: AsyncSession, task_id: str, dest_data
                 edge_type=EdgeType.DATA_SYNTHESIS.value,
                 description=task.description,
                 from_node_id=src_node.id,
-                to_node_id=dst_dataset.id
+                to_node_id=dst_dataset.id,
             )
 
-            # 生成血缘图
             lineage_service = LineageService(db=db)
             await lineage_service.generate_graph(src_node, synthesis_edge, dest_node)
-            await self.db.commit()
+            await db.commit()
 
             logger.info(f"Added synthesis lineage: {src_node.name} -> {dest_dataset.name}")
         except Exception as exc:
diff --git a/runtime/datamate-python/app/module/generation/service/qa_generator.py b/runtime/datamate-python/app/module/generation/service/qa_generator.py
new file mode 100644
index 000000000..d7c52fd47
--- /dev/null
+++ b/runtime/datamate-python/app/module/generation/service/qa_generator.py
@@ -0,0 +1,349 @@
+"""问答生成器模块 - 负责问题生成和答案生成"""
+import asyncio
+import json
+import re
+import uuid
+from typing import Any
+
+from langchain_core.language_models import BaseChatModel
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models.data_synthesis import (
+    DataSynthesisChunkInstance,
+    DataSynthesisFileInstance,
+    SynthesisData,
+)
+from app.db.session import logger
+from app.module.generation.schema.generation import SyntheConfig
+from app.module.generation.service.prompt import (
+    ANSWER_GENERATOR_PROMPT,
+    QUESTION_GENERATOR_PROMPT,
+)
+from app.module.shared.util.model_chat import extract_json_substring
+from app.module.shared.llm import LLMFactory
+
+
+# 图片URL提取正则
+IMG_URL_PATTERN = re.compile(r"!\[\]\((.*?)\)")
+
+
+def extract_img_urls(text: str) -> list[str]:
+    """提取文本中的图片URL"""
+    return IMG_URL_PATTERN.findall(text)
+
+
+def _filter_docs_by_size(docs: list[Any], chunk_size: int, ratio: float = 0.7) -> list[Any]:
+    """过滤文档，保留长度大于 chunk_size * ratio 的文档
+
+    Args:
+        docs: 文档列表
+        chunk_size: 切片大小
+        ratio: 长度比例阈值
+
+    Returns:
+        过滤后的文档列表
+    """
+    min_length = chunk_size * ratio
+    return [doc for doc in docs if len(doc.page_content) >= min_length]
+
+
+class QuestionGenerator:
+    """问题生成器"""
+
+    def __init__(self, db: AsyncSession, semaphore: asyncio.Semaphore | None = None):
+        self.db = db
+        self.semaphore = semaphore or asyncio.Semaphore(10)
+
+    async def generate_questions(
+        self,
+        chunk_text: str,
+        question_cfg: SyntheConfig,
+        question_chat: BaseChatModel,
+    ) -> list[str]:
+        """针对单个切片文本，调用模型生成问题列表
+
+        Args:
+            chunk_text: 切片文本内容
+            question_cfg: 问题生成配置
+            question_chat: 问题生成模型
+
+        Returns:
+            问题列表
+        """
+        # 计算问题数量：每千tokens生成指定数量的问题
+        base_number = question_cfg.number or 5
+        calculated_number = max(int(len(chunk_text) / 1000 * base_number), 1)
+
+        # 获取提示模板
+        template = getattr(question_cfg, "prompt_template", None)
+        template = template if template and template.strip() else QUESTION_GENERATOR_PROMPT
+
+        # 构建提示
+        prompt = (
+            template
+            .replace("{text}", chunk_text)
+            .replace("{number}", str(calculated_number))
+            .replace("{textLength}", str(len(chunk_text)))
+        )
+
+        # 调用模型
+        async with self.semaphore:
+            loop = asyncio.get_running_loop()
+            raw_answer = await loop.run_in_executor(
+                None,
+                LLMFactory.invoke_sync,
+                question_chat,
+                prompt,
+            )
+
+        # 解析问题列表
+        return self._parse_questions(raw_answer)
+
+    def _parse_questions(self, raw_answer: str) -> list[str]:
+        """从模型返回中解析问题列表
+
+        Args:
+            raw_answer: 模型原始返回
+
+        Returns:
+            问题列表
+        """
+        if not raw_answer:
+            return []
+
+        cleaned = extract_json_substring(raw_answer)
+        try:
+            data = json.loads(cleaned)
+        except Exception as e:
+            logger.error(f"Failed to parse question list JSON: {e}")
+            return []
+
+        if isinstance(data, list):
+            return [str(q) for q in data if isinstance(q, str) and q.strip()]
+
+        # 容错：如果是单个字符串
+        if isinstance(data, str) and data.strip():
+            return [data.strip()]
+
+        return []
+
+
+class AnswerGenerator:
+    """答案生成器"""
+
+    def __init__(self, db: AsyncSession, semaphore: asyncio.Semaphore | None = None):
+        self.db = db
+        self.semaphore = semaphore or asyncio.Semaphore(10)
+
+    async def generate_answers_for_chunk(
+        self,
+        file_task: DataSynthesisFileInstance,
+        chunk: DataSynthesisChunkInstance,
+        questions: list[str],
+        answer_cfg: SyntheConfig,
+        answer_chat: BaseChatModel,
+    ) -> int:
+        """为一个切片的所有问题生成答案并写入数据库
+
+        Args:
+            file_task: 文件任务实例
+            chunk: 切片实例
+            questions: 问题列表
+            answer_cfg: 答案生成配置
+            answer_chat: 答案生成模型
+
+        Returns:
+            成功生成的QA对数量
+        """
+        if not questions:
+            return 0
+
+        chunk_text = chunk.chunk_content or ""
+        template = getattr(answer_cfg, "prompt_template", None)
+        template = template if template and template.strip() else ANSWER_GENERATOR_PROMPT
+
+        success_count = 0
+
+        async def process_question(question: str) -> bool:
+            nonlocal success_count
+            try:
+                prompt = template.replace("{text}", chunk_text).replace("{question}", question)
+
+                async with self.semaphore:
+                    loop = asyncio.get_running_loop()
+                    answer = await loop.run_in_executor(
+                        None,
+                        LLMFactory.invoke_sync,
+                        answer_chat,
+                        prompt,
+                    )
+
+                # 构建数据对象
+                data_obj = self._build_synthesis_data(chunk_text, question, answer)
+
+                # 保存记录
+                record = SynthesisData(
+                    id=str(uuid.uuid4()),
+                    data=data_obj,
+                    synthesis_file_instance_id=file_task.id,
+                    chunk_instance_id=chunk.id,
+                )
+                self.db.add(record)
+                success_count += 1
+                return True
+
+            except Exception as e:
+                logger.error(f"Failed to generate answer for question '{question[:50]}...': {e}")
+                return False
+
+        # 并行处理所有问题
+        tasks = [process_question(q) for q in questions]
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+        if success_count > 0:
+            await self.db.commit()
+
+        return success_count
+
+    def _build_synthesis_data(
+        self,
+        chunk_text: str,
+        question: str,
+        answer: str,
+    ) -> dict[str, Any]:
+        """构建合成数据结构
+
+        Args:
+            chunk_text: 切片原文
+            question: 问题
+            answer: 答案
+
+        Returns:
+            结构化数据字典
+        """
+        # 基础结构
+        base_obj: dict[str, Any] = {
+            "input": chunk_text,
+            "output": answer,
+        }
+
+        # 尝试解析模型返回的JSON
+        parsed_obj: dict[str, Any] | None = None
+        if isinstance(answer, str):
+            cleaned = extract_json_substring(answer)
+            try:
+                parsed = json.loads(cleaned)
+                if isinstance(parsed, dict):
+                    parsed_obj = parsed
+            except Exception:
+                parsed_obj = None
+
+        # 合并解析结果
+        if parsed_obj is not None:
+            parsed_obj["instruction"] = question
+            data_obj = parsed_obj
+        else:
+            base_obj["instruction"] = question
+            data_obj = base_obj
+
+        # 提取图片URL
+        img_urls = extract_img_urls(chunk_text)
+        if img_urls:
+            data_obj["img_urls"] = img_urls
+
+        return data_obj
+
+
+class QAGenerator:
+    """问答生成器 - 整合问题和答案生成"""
+
+    def __init__(self, db: AsyncSession):
+        self.db = db
+        self.question_semaphore = asyncio.Semaphore(10)
+        self.answer_semaphore = asyncio.Semaphore(10)
+        self.question_generator = QuestionGenerator(db, self.question_semaphore)
+        self.answer_generator = AnswerGenerator(db, self.answer_semaphore)
+
+    async def process_chunk_qa(
+        self,
+        file_task: DataSynthesisFileInstance,
+        chunk: DataSynthesisChunkInstance,
+        question_cfg: SyntheConfig,
+        answer_cfg: SyntheConfig,
+        question_chat: Any,
+        answer_chat: Any,
+    ) -> int:
+        """处理单个切片的QA生成
+
+        Args:
+            file_task: 文件任务实例
+            chunk: 切片实例
+            question_cfg: 问题生成配置
+            answer_cfg: 答案生成配置
+            question_chat: 问题生成模型
+            answer_chat: 答案生成模型
+
+        Returns:
+            成功生成的QA对数量
+        """
+        chunk_text = chunk.chunk_content or ""
+        if not chunk_text.strip():
+            logger.warning(f"Empty chunk text for chunk_index={chunk.chunk_index}")
+            return 0
+
+        # 1. 生成问题
+        try:
+            questions = await self.question_generator.generate_questions(
+                chunk_text=chunk_text,
+                question_cfg=question_cfg,
+                question_chat=question_chat,
+            )
+        except Exception as e:
+            logger.error(f"Generate questions failed: {e}")
+            questions = []
+
+        if not questions:
+            logger.info(f"No questions generated for chunk_index={chunk.chunk_index}")
+            return 0
+
+        # 2. 生成答案
+        success_count = await self.answer_generator.generate_answers_for_chunk(
+            file_task=file_task,
+            chunk=chunk,
+            questions=questions,
+            answer_cfg=answer_cfg,
+            answer_chat=answer_chat,
+        )
+
+        return success_count
+
+    async def check_qa_limit(
+        self,
+        synth_task_id: str,
+        max_qa_pairs: int | None,
+    ) -> bool:
+        """检查是否已达到QA对数量上限
+
+        Args:
+            synth_task_id: 合成任务ID
+            max_qa_pairs: 最大QA对数量限制
+
+        Returns:
+            是否已达到上限
+        """
+        if max_qa_pairs is None or max_qa_pairs <= 0:
+            return False
+
+        result = await self.db.execute(
+            select(func.count(SynthesisData.id)).where(
+                SynthesisData.synthesis_file_instance_id.in_(
+                    select(DataSynthesisFileInstance.id).where(
+                        DataSynthesisFileInstance.synthesis_instance_id == synth_task_id
+                    )
+                )
+            )
+        )
+        current_count = int(result.scalar() or 0)
+
+        return current_count >= max_qa_pairs
diff --git a/runtime/datamate-python/app/module/generation/service/synthesis_data_query_service.py b/runtime/datamate-python/app/module/generation/service/synthesis_data_query_service.py
new file mode 100644
index 000000000..6d05a8f24
--- /dev/null
+++ b/runtime/datamate-python/app/module/generation/service/synthesis_data_query_service.py
@@ -0,0 +1,98 @@
+"""
+合成数据查询服务 - 负责合成数据相关的查询操作
+"""
+from typing import List
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models.data_synthesis import (
+    DataSynthesisFileInstance,
+    SynthesisData,
+)
+from app.module.generation.schema.generation import (
+    SynthesisDataItem,
+    SynthesisDataPatchItem,
+)
+
+
+class SynthesisDataQueryService:
+    """合成数据查询服务 - 负责合成数据相关的查询操作"""
+
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def list_by_chunk(self, chunk_id: str) -> List[SynthesisDataItem]:
+        """
+        根据切片 ID 查询所有合成数据
+
+        Args:
+            chunk_id: 切片 ID
+
+        Returns:
+            合成数据列表
+        """
+        result = await self.db.execute(
+            select(SynthesisData).where(SynthesisData.chunk_instance_id == chunk_id)
+        )
+        rows = result.scalars().all()
+        return [SynthesisDataItem.model_validate(row) for row in rows]
+
+    async def list_by_file(self, file_id: str) -> List[SynthesisDataPatchItem]:
+        """
+        根据文件任务 ID 查询所有合成数据
+
+        Args:
+            file_id: 文件任务 ID
+
+        Returns:
+            合成数据列表
+        """
+        result = await self.db.execute(
+            select(SynthesisData).where(
+                SynthesisData.synthesis_file_instance_id == file_id
+            )
+        )
+        rows = result.scalars().all()
+        return [SynthesisDataPatchItem.model_validate(row) for row in rows]
+
+    async def list_by_task(self, task_id: str) -> List[SynthesisDataPatchItem]:
+        """
+        根据任务 ID 查询所有合成数据
+
+        Args:
+            task_id: 任务 ID
+
+        Returns:
+            合成数据列表
+        """
+        result = await self.db.execute(
+            select(SynthesisData).where(
+                SynthesisData.synthesis_file_instance_id.in_(
+                    select(DataSynthesisFileInstance.id).where(
+                        DataSynthesisFileInstance.synthesis_instance_id == task_id
+                    )
+                )
+            )
+        )
+        rows = result.scalars().all()
+        return [SynthesisDataPatchItem.model_validate(row) for row in rows]
+
+    async def update(self, data_id: str, data: dict) -> SynthesisDataItem | None:
+        """
+        更新合成数据
+
+        Args:
+            data_id: 数据 ID
+            data: 新的数据内容
+
+        Returns:
+            更新后的数据，不存在则返回 None
+        """
+        record = await self.db.get(SynthesisData, data_id)
+        if not record:
+            return None
+        record.data = data
+        await self.db.commit()
+        await self.db.refresh(record)
+        return SynthesisDataItem.model_validate(record)
diff --git a/runtime/datamate-python/app/module/generation/service/task_create_service.py b/runtime/datamate-python/app/module/generation/service/task_create_service.py
new file mode 100644
index 000000000..a684ec524
--- /dev/null
+++ b/runtime/datamate-python/app/module/generation/service/task_create_service.py
@@ -0,0 +1,89 @@
+"""
+任务命令服务 - 负责任务的创建操作
+"""
+import uuid
+from typing import List
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.exception import transaction
+from app.db.models.data_synthesis import (
+    DataSynthInstance,
+    DataSynthesisFileInstance,
+    save_synthesis_task,
+)
+from app.db.models.dataset_management import DatasetFiles
+from app.module.generation.schema.generation import CreateSynthesisTaskRequest
+
+
+class TaskCreateService:
+    """任务创建服务 - 负责任务的创建操作"""
+
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def create(self, request: CreateSynthesisTaskRequest) -> DataSynthInstance:
+        """
+        创建数据合成任务
+
+        Args:
+            request: 创建任务请求
+
+        Returns:
+            创建的任务实例
+        """
+        dataset_files = await self._query_source_files(request.source_file_id)
+        request.source_file_id = [str(f.id) for f in dataset_files]
+
+        async with transaction(self.db):
+            task = await save_synthesis_task(self.db, request)
+            await self._create_file_instances(task.id, dataset_files)
+
+        return task
+
+    async def _query_source_files(self, file_ids: List[str]) -> List[DatasetFiles]:
+        """
+        查询源文件列表
+
+        Args:
+            file_ids: 文件 ID 列表
+
+        Returns:
+            源文件列表
+        """
+        if not file_ids:
+            return []
+        result = await self.db.execute(
+            select(DatasetFiles).where(DatasetFiles.id.in_(file_ids))
+        )
+        return list(result.scalars().all())
+
+    async def _create_file_instances(
+        self, task_id: str, dataset_files: List[DatasetFiles]
+    ) -> None:
+        """
+        创建文件任务实例
+
+        Args:
+            task_id: 任务 ID
+            dataset_files: 源文件列表
+        """
+        if not dataset_files:
+            return
+
+        file_instances = [
+            DataSynthesisFileInstance(
+                id=str(uuid.uuid4()),
+                synthesis_instance_id=task_id,
+                file_name=f.file_name,
+                source_file_id=str(f.id),
+                status="pending",
+                total_chunks=0,
+                processed_chunks=0,
+                created_by="system",
+                updated_by="system",
+            )
+            for f in dataset_files
+        ]
+        self.db.add_all(file_instances)
diff --git a/runtime/datamate-python/app/module/generation/service/task_delete_service.py b/runtime/datamate-python/app/module/generation/service/task_delete_service.py
new file mode 100644
index 000000000..221b73277
--- /dev/null
+++ b/runtime/datamate-python/app/module/generation/service/task_delete_service.py
@@ -0,0 +1,219 @@
+"""
+任务删除服务 - 负责任务的删除操作
+"""
+from typing import List
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.exception import BusinessError, ErrorCodes
+from app.db.models.data_synthesis import (
+    DataSynthInstance,
+    DataSynthesisFileInstance,
+    DataSynthesisChunkInstance,
+    SynthesisData,
+)
+
+
+class TaskDeleteService:
+    """任务删除服务 - 负责任务的删除操作"""
+
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def delete_task(self, task_id: str) -> None:
+        """
+        删除任务及其所有关联数据
+
+        Args:
+            task_id: 任务 ID
+        """
+        task = await self._get_task(task_id)
+        if not task:
+            raise BusinessError(ErrorCodes.GENERATION_TASK_NOT_FOUND)
+
+        file_ids = await self._get_file_instance_ids(task_id)
+
+        if file_ids:
+            await self._delete_synthesis_data_by_files(file_ids)
+            await self._delete_chunks_by_files(file_ids)
+            await self._delete_file_instances(file_ids)
+
+        await self.db.delete(task)
+        await self.db.commit()
+
+    async def delete_file_task(self, task_id: str, file_id: str) -> None:
+        """
+        删除文件任务并更新任务统计
+
+        Args:
+            task_id: 任务 ID
+            file_id: 文件任务 ID
+        """
+        task = await self._get_task(task_id)
+        if not task:
+            raise BusinessError(ErrorCodes.GENERATION_TASK_NOT_FOUND)
+
+        file_task = await self._get_file_task(file_id)
+        if not file_task:
+            raise BusinessError(ErrorCodes.GENERATION_FILE_NOT_FOUND)
+
+        await self._delete_synthesis_data_by_files([file_id])
+        await self._delete_chunks_by_files([file_id])
+        await self.db.delete(file_task)
+
+        self._decrement_file_count(task)
+        await self.db.commit()
+
+    async def delete_chunk(self, chunk_id: str) -> None:
+        """
+        删除切片及其关联的合成数据
+
+        Args:
+            chunk_id: 切片 ID
+        """
+        chunk = await self._get_chunk(chunk_id)
+        if not chunk:
+            raise BusinessError(ErrorCodes.GENERATION_CHUNK_NOT_FOUND)
+
+        await self._delete_synthesis_data_by_chunk(chunk_id)
+        await self._delete_chunk(chunk_id)
+        await self.db.commit()
+
+    async def delete_synthesis_data_by_chunk(self, chunk_id: str) -> int:
+        """
+        删除切片下的所有合成数据
+
+        Args:
+            chunk_id: 切片 ID
+
+        Returns:
+            删除的记录数
+        """
+        result = await self._delete_synthesis_data_by_chunk(chunk_id)
+        deleted = int(getattr(result, "rowcount", 0) or 0)
+        await self.db.commit()
+        return deleted
+
+    async def batch_delete_synthesis_data(self, ids: List[str]) -> int:
+        """
+        批量删除合成数据
+
+        Args:
+            ids: 合成数据 ID 列表
+
+        Returns:
+            删除的记录数
+        """
+        if not ids:
+            return 0
+        result = await self.db.execute(
+            delete(SynthesisData).where(SynthesisData.id.in_(ids))
+        )
+        deleted = int(getattr(result, "rowcount", 0) or 0)
+        await self.db.commit()
+        return deleted
+
+    async def delete_chunks_batch(self, chunk_ids: List[str]) -> None:
+        """
+        批量删除切片
+
+        Args:
+            chunk_ids: 切片 ID 列表
+        """
+        for chunk_id in chunk_ids:
+            chunk = await self._get_chunk(chunk_id)
+            if chunk:
+                await self._delete_synthesis_data_by_chunk(chunk_id)
+                await self._delete_chunk(chunk_id)
+        await self.db.commit()
+
+    async def delete_chunks_by_file(self, file_id: str) -> None:
+        """
+        按文件任务维度删除切片
+
+        Args:
+            file_id: 文件任务 ID
+        """
+        await self._delete_synthesis_data_by_files([file_id])
+        await self._delete_chunks_by_files([file_id])
+        await self.db.commit()
+
+    async def delete_chunks_by_task(self, task_id: str) -> None:
+        """
+        按任务维度删除切片
+
+        Args:
+            task_id: 任务 ID
+        """
+        file_ids = await self._get_file_instance_ids(task_id)
+        if file_ids:
+            await self._delete_synthesis_data_by_files(file_ids)
+            await self._delete_chunks_by_files(file_ids)
+        await self.db.commit()
+
+    async def _get_task(self, task_id: str) -> DataSynthInstance | None:
+        """获取任务"""
+        return await self.db.get(DataSynthInstance, task_id)
+
+    async def _get_file_task(self, file_id: str) -> DataSynthesisFileInstance | None:
+        """获取文件任务"""
+        return await self.db.get(DataSynthesisFileInstance, file_id)
+
+    async def _get_chunk(self, chunk_id: str) -> DataSynthesisChunkInstance | None:
+        """获取切片"""
+        return await self.db.get(DataSynthesisChunkInstance, chunk_id)
+
+    async def _get_file_instance_ids(self, task_id: str) -> List[str]:
+        """获取任务下的所有文件任务 ID"""
+        result = await self.db.execute(
+            select(DataSynthesisFileInstance.id).where(
+                DataSynthesisFileInstance.synthesis_instance_id == task_id
+            )
+        )
+        return [row[0] for row in result.all()]
+
+    def _decrement_file_count(self, task: DataSynthInstance) -> None:
+        """递减任务的文件数量"""
+        if task.total_files and task.total_files > 0:
+            task.total_files -= 1
+            if task.total_files < 0:
+                task.total_files = 0
+
+    async def _delete_synthesis_data_by_files(self, file_ids: List[str]) -> None:
+        """批量删除文件任务下的合成数据"""
+        await self.db.execute(
+            delete(SynthesisData).where(
+                SynthesisData.synthesis_file_instance_id.in_(file_ids)
+            )
+        )
+
+    async def _delete_synthesis_data_by_chunk(self, chunk_id: str) -> None:
+        """删除切片下的合成数据"""
+        await self.db.execute(
+            delete(SynthesisData).where(SynthesisData.chunk_instance_id == chunk_id)
+        )
+
+    async def _delete_chunks_by_files(self, file_ids: List[str]) -> None:
+        """批量删除文件任务下的切片"""
+        await self.db.execute(
+            delete(DataSynthesisChunkInstance).where(
+                DataSynthesisChunkInstance.synthesis_file_instance_id.in_(file_ids)
+            )
+        )
+
+    async def _delete_chunk(self, chunk_id: str) -> None:
+        """删除切片"""
+        await self.db.execute(
+            delete(DataSynthesisChunkInstance).where(
+                DataSynthesisChunkInstance.id == chunk_id
+            )
+        )
+
+    async def _delete_file_instances(self, file_ids: List[str]) -> None:
+        """批量删除文件任务"""
+        await self.db.execute(
+            delete(DataSynthesisFileInstance).where(
+                DataSynthesisFileInstance.id.in_(file_ids)
+            )
+        )
diff --git a/runtime/datamate-python/app/module/generation/service/task_executor.py b/runtime/datamate-python/app/module/generation/service/task_executor.py
new file mode 100644
index 000000000..9daa44d34
--- /dev/null
+++ b/runtime/datamate-python/app/module/generation/service/task_executor.py
@@ -0,0 +1,137 @@
+"""
+独立任务执行器 - 真正隔离后台任务与主事件循环
+
+使用线程池执行CPU密集型/阻塞操作，避免阻塞主事件循环
+"""
+import asyncio
+import concurrent.futures
+from typing import Any, Callable
+from functools import partial
+
+from app.core.logging import get_logger
+
+logger = get_logger(__name__)
+
+# 全局线程池 - 用于执行阻塞操作
+# 使用 ProcessPoolExecutor 处理CPU密集型任务，ThreadPoolExecutor 处理I/O密集型任务
+_thread_pool: concurrent.futures.ThreadPoolExecutor | None = None
+_process_pool: concurrent.futures.ProcessPoolExecutor | None = None
+
+# 任务队列 - 限制并发任务数量
+_task_semaphore: asyncio.Semaphore | None = None
+
+
+def init_executor(
+    max_workers: int = 10,
+    max_concurrent_tasks: int = 5,
+):
+    """
+    初始化执行器
+
+    Args:
+        max_workers: 线程池最大工作线程数
+        max_concurrent_tasks: 最大并发任务数（同时执行的数据合成任务数）
+    """
+    global _thread_pool, _process_pool, _task_semaphore
+
+    if _thread_pool is None:
+        _thread_pool = concurrent.futures.ThreadPoolExecutor(
+            max_workers=max_workers,
+            thread_name_prefix="generation_task_"
+        )
+        logger.info(f"Initialized thread pool with {max_workers} workers")
+
+    if _task_semaphore is None:
+        _task_semaphore = asyncio.Semaphore(max_concurrent_tasks)
+        logger.info(f"Initialized task semaphore with limit {max_concurrent_tasks}")
+
+
+def get_thread_pool() -> concurrent.futures.ThreadPoolExecutor:
+    """获取线程池，如果未初始化则自动初始化"""
+    global _thread_pool
+    if _thread_pool is None:
+        init_executor()
+    return _thread_pool
+
+
+def get_task_semaphore() -> asyncio.Semaphore:
+    """获取任务信号量，如果未初始化则自动初始化"""
+    global _task_semaphore
+    if _task_semaphore is None:
+        init_executor()
+    return _task_semaphore
+
+
+async def run_in_thread[
+    T
+](func: Callable[..., T], *args, **kwargs) -> T:
+    """
+    在线程池中执行阻塞函数
+
+    Args:
+        func: 要执行的阻塞函数
+        *args: 位置参数
+        **kwargs: 关键字参数
+
+    Returns:
+        函数执行结果
+    """
+    pool = get_thread_pool()
+    loop = asyncio.get_running_loop()
+
+    # 包装函数，添加超时和异常处理
+    def wrapper():
+        try:
+            return func(*args, **kwargs)
+        except Exception as e:
+            logger.error(f"Error in thread execution: {e}")
+            raise
+
+    try:
+        # 使用 asyncio.wait_for 添加超时控制
+        return await asyncio.wait_for(
+            loop.run_in_executor(pool, wrapper),
+            timeout=300  # 5分钟超时
+        )
+    except asyncio.TimeoutError:
+        logger.error(f"Thread execution timeout after 300s")
+        raise
+
+
+async def execute_generation_task(task_id: str, process_task_func: Callable[..., Any]):
+    """
+    执行数据合成任务，带并发控制和资源隔离
+
+    Args:
+        task_id: 任务ID
+        process_task_func: 异步任务处理函数，接收task_id作为参数
+    """
+    semaphore = get_task_semaphore()
+
+    async with semaphore:
+        logger.info(f"Starting generation task {task_id} (semaphore acquired)")
+        try:
+            # 直接调用异步函数，而不是通过run_in_thread
+            # 因为process_task_func已经是异步函数
+            await process_task_func(task_id)
+            logger.info(f"Generation task {task_id} completed successfully")
+        except Exception as e:
+            logger.exception(f"Generation task {task_id} failed: {e}")
+            raise
+
+
+def shutdown_executor():
+    """关闭执行器，释放资源"""
+    global _thread_pool, _process_pool, _task_semaphore
+
+    if _thread_pool:
+        _thread_pool.shutdown(wait=True)
+        _thread_pool = None
+        logger.info("Thread pool shutdown")
+
+    if _process_pool:
+        _process_pool.shutdown(wait=True)
+        _process_pool = None
+        logger.info("Process pool shutdown")
+
+    _task_semaphore = None
diff --git a/runtime/datamate-python/app/module/generation/service/task_query_service.py b/runtime/datamate-python/app/module/generation/service/task_query_service.py
new file mode 100644
index 000000000..928de0514
--- /dev/null
+++ b/runtime/datamate-python/app/module/generation/service/task_query_service.py
@@ -0,0 +1,59 @@
+from sqlalchemy import select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models.data_synthesis import DataSynthInstance
+from app.module.generation.schema.generation import (
+    DataSynthesisTaskItem,
+    PagedDataSynthesisTaskResponse,
+)
+
+
+class TaskQueryService:
+    """任务查询服务 - 专注于任务列表和详情查询"""
+
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def get_task(self, task_id: str) -> DataSynthInstance | None:
+        """获取单个任务"""
+        return await self.db.get(DataSynthInstance, task_id)
+
+    async def list_tasks(
+        self,
+        page: int,
+        page_size: int,
+        synthesis_type: str | None = None,
+        status: str | None = None,
+        name: str | None = None,
+    ) -> PagedDataSynthesisTaskResponse:
+        """分页查询任务列表"""
+        query = select(DataSynthInstance)
+
+        if synthesis_type:
+            query = query.where(DataSynthInstance.synth_type == synthesis_type)
+        if status:
+            query = query.where(DataSynthInstance.status == status)
+        if name:
+            query = query.where(DataSynthInstance.name.like(f"%{name}%"))
+
+        query = query.order_by(DataSynthInstance.created_at.desc())
+
+        count_result = await self.db.execute(
+            select(func.count()).select_from(query.subquery())
+        )
+        total = count_result.scalar_one() or 0
+
+        offset = max(page, 1) - 1
+        limit = max(page_size, 1)
+        result = await self.db.execute(query.offset(offset * limit).limit(limit))
+        rows = result.scalars().all()
+
+        items = [DataSynthesisTaskItem.from_orm_with_config(row) for row in rows]
+
+        return PagedDataSynthesisTaskResponse(
+            content=items,
+            totalElements=total,
+            totalPages=(total + page_size - 1) // page_size,
+            page=page,
+            size=page_size,
+        )