fix: Enhance synthesis task management and export functionality (#465)

* feat: add export functionality for synthesis tasks with format selection

* fix: adjust chunk size and overlap parameters for synthesis tasks

* refactor: modularize generation service and optimize concurrency settings

* feat: enhance archive functionality with format selection for synthesis tasks

* feat: add synthesis data management functions including delete and update operations

* feat: add synthesis data management functions including delete and update operations

* feat: add synthesis data management functions including delete and update operations

* feat: implement task executor for background processing and enhance task management

* feat: implement task executor for background processing and enhance task management

* feat: implement task executor for background processing and enhance task management

Author: Dallas98

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: "",
   });
 

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,

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,18 +296,28 @@ 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);
       message.error(t('synthesisTask.home.archive.failed'));
     }
   };
 
+  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}

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>);
 }

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 头中解析文件名
    */
@@ -661,6 +683,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);
 

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 }) => {

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(

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")