feat(api): implement AsyncTrainingManager MVP with SQLite persistence

- Add api_server module with adapter pattern architecture
- Implement AsyncTrainingManager using asyncio.subprocess + SQLite
- Add TaskQueueAdapter abstract base class for future server mode
- Create domain models: Task, TaskStatus, ProgressInfo
- Add run_pipeline.py wrapper script for subprocess execution
- Create config module for centralized environment variables
- Add aiosqlite dependency to pyproject.toml
- Include test config files for pipeline validation

The AsyncTrainingManager provides:
- Async task queue with SQLite persistence
- Real-time progress tracking via stdout JSON parsing
- Task cancellation and status querying
- Progress subscription for SSE streaming
- Application restart recovery support

api_server/app/__init__.py

@@ -0,0 +1,5 @@
+"""
+GPT-SoVITS 训练 API Server
+"""
+
+__version__ = "0.1.0"

api_server/app/adapters/__init__.py

@@ -0,0 +1,9 @@
+"""
+适配器模块
+
+提供不同环境下的存储、任务队列等适配器实现
+"""
+
+from .base import TaskQueueAdapter
+
+__all__ = ["TaskQueueAdapter"]

api_server/app/adapters/base.py

@@ -0,0 +1,140 @@
+"""
+适配器抽象基类模块
+
+定义任务队列、存储、数据库等适配器的抽象接口
+"""
+
+from abc import ABC, abstractmethod
+from typing import Dict, Optional, AsyncGenerator
+
+
+class TaskQueueAdapter(ABC):
+    """
+    任务队列适配器抽象基类
+    
+    定义任务队列的通用接口，支持本地（asyncio.subprocess）和
+    服务器（Celery）两种实现方式。
+    
+    Example:
+        >>> adapter = AsyncTrainingManager(db_path="./data/tasks.db")
+        >>> job_id = await adapter.enqueue("task-123", {"exp_name": "test"})
+        >>> status = await adapter.get_status(job_id)
+        >>> async for progress in adapter.subscribe_progress("task-123"):
+        ...     print(progress)
+    """
+    
+    @abstractmethod
+    async def enqueue(self, task_id: str, config: Dict, priority: str = "normal") -> str:
+        """
+        将任务加入队列
+        
+        Args:
+            task_id: 任务唯一标识
+            config: 任务配置字典，包含训练所需的所有参数
+            priority: 任务优先级 ("low", "normal", "high")
+            
+        Returns:
+            job_id: 队列中的作业ID
+            
+        Raises:
+            ValueError: 配置无效时抛出
+        """
+        pass
+    
+    @abstractmethod
+    async def get_status(self, job_id: str) -> Dict:
+        """
+        获取任务状态
+        
+        Args:
+            job_id: 作业ID
+            
+        Returns:
+            状态字典，包含:
+            - status: 任务状态 (queued, running, completed, failed, cancelled)
+            - progress: 进度 (0.0-1.0)
+            - current_stage: 当前阶段名称
+            - message: 状态消息
+            - error_message: 错误信息（如果失败）
+        """
+        pass
+    
+    @abstractmethod
+    async def cancel(self, job_id: str) -> bool:
+        """
+        取消任务
+        
+        Args:
+            job_id: 作业ID
+            
+        Returns:
+            是否成功取消
+        """
+        pass
+    
+    @abstractmethod
+    async def subscribe_progress(self, task_id: str) -> AsyncGenerator[Dict, None]:
+        """
+        订阅任务进度（用于 SSE 流）
+        
+        Args:
+            task_id: 任务ID
+            
+        Yields:
+            进度信息字典，包含:
+            - type: 消息类型 ("progress", "log", "heartbeat")
+            - stage: 当前阶段
+            - progress: 进度值
+            - message: 进度消息
+            - status: 状态 (running, completed, failed, cancelled)
+            
+        Note:
+            当 status 为终态时，生成器会自动结束
+        """
+        pass
+
+
+class ProgressAdapter(ABC):
+    """
+    进度管理适配器抽象基类
+    
+    用于更新和订阅任务进度，支持本地（内存队列）和
+    服务器（Redis Pub/Sub）两种实现。
+    """
+    
+    @abstractmethod
+    async def update_progress(self, task_id: str, progress: Dict) -> None:
+        """
+        更新进度
+        
+        Args:
+            task_id: 任务ID
+            progress: 进度信息字典
+        """
+        pass
+    
+    @abstractmethod
+    async def get_progress(self, task_id: str) -> Optional[Dict]:
+        """
+        获取当前进度
+        
+        Args:
+            task_id: 任务ID
+            
+        Returns:
+            最新进度信息，不存在则返回 None
+        """
+        pass
+    
+    @abstractmethod
+    async def subscribe(self, task_id: str) -> AsyncGenerator[Dict, None]:
+        """
+        订阅进度更新
+        
+        Args:
+            task_id: 任务ID
+            
+        Yields:
+            进度信息字典
+        """
+        pass

api_server/app/adapters/local/__init__.py

@@ -0,0 +1,9 @@
+"""
+本地适配器模块
+
+提供基于 SQLite 和 asyncio.subprocess 的本地实现
+"""
+
+from .task_queue import AsyncTrainingManager
+
+__all__ = ["AsyncTrainingManager"]

api_server/app/adapters/local/task_queue.py

@@ -0,0 +1,695 @@
+"""
+本地异步任务管理器
+
+基于 asyncio.subprocess + SQLite 的本地任务队列实现。
+适用于 macOS 本地训练和 Electron 集成场景。
+"""
+
+import asyncio
+import json
+import os
+import sqlite3
+import sys
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, Optional, AsyncGenerator, List
+
+import aiosqlite
+
+from ..base import TaskQueueAdapter
+from ...core.config import settings, PROJECT_ROOT, get_pythonpath
+
+# 进度消息标识符（与 run_pipeline.py 保持一致）
+PROGRESS_PREFIX = "##PROGRESS##"
+PROGRESS_SUFFIX = "##"
+
+
+class AsyncTrainingManager(TaskQueueAdapter):
+    """
+    基于 asyncio.subprocess 的异步任务管理器
+    
+    特点：
+    1. 使用 asyncio.create_subprocess_exec() 异步启动训练子进程
+    2. 完全非阻塞，与 FastAPI 异步模型完美契合
+    3. SQLite 持久化任务状态，支持应用重启后恢复
+    4. 实时解析子进程输出获取进度
+    
+    Example:
+        >>> manager = AsyncTrainingManager(db_path="./data/tasks.db")
+        >>> job_id = await manager.enqueue("task-123", {"exp_name": "test", ...})
+        >>> 
+        >>> # 订阅进度
+        >>> async for progress in manager.subscribe_progress("task-123"):
+        ...     print(f"{progress['stage']}: {progress['progress']*100:.1f}%")
+        >>> 
+        >>> # 取消任务
+        >>> await manager.cancel(job_id)
+    """
+
+    def __init__(self, db_path: str = None, max_concurrent: int = 1):
+        """
+        初始化任务管理器
+        
+        Args:
+            db_path: SQLite 数据库路径，默认使用 settings.SQLITE_PATH
+            max_concurrent: 最大并发任务数（本地通常为1）
+        """
+        self.db_path = db_path or str(settings.SQLITE_PATH)
+        self.max_concurrent = max_concurrent
+
+        # 运行时状态
+        self.running_processes: Dict[str, asyncio.subprocess.Process] = {}  # task_id -> Process
+        self.progress_channels: Dict[str, asyncio.Queue] = {}  # task_id -> Queue
+        self._running_count = 0
+        self._queue_lock = asyncio.Lock()
+
+        # 初始化数据库
+        self._init_db_sync()
+
+    def _init_db_sync(self) -> None:
+        """同步初始化数据库（启动时调用）"""
+        Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
+
+        with sqlite3.connect(self.db_path) as conn:
+            conn.execute('''
+                CREATE TABLE IF NOT EXISTS task_queue (
+                    job_id TEXT PRIMARY KEY,
+                    task_id TEXT NOT NULL UNIQUE,
+                    exp_name TEXT NOT NULL,
+                    config TEXT NOT NULL,
+                    status TEXT DEFAULT 'queued',
+                    current_stage TEXT,
+                    progress REAL DEFAULT 0,
+                    overall_progress REAL DEFAULT 0,
+                    message TEXT,
+                    error_message TEXT,
+                    created_at TEXT NOT NULL,
+                    started_at TEXT,
+                    completed_at TEXT
+                )
+            ''')
+            conn.execute('CREATE INDEX IF NOT EXISTS idx_task_queue_status ON task_queue(status)')
+            conn.execute('CREATE INDEX IF NOT EXISTS idx_task_queue_task_id ON task_queue(task_id)')
+            conn.execute('CREATE INDEX IF NOT EXISTS idx_task_queue_created ON task_queue(created_at)')
+            conn.commit()
+
+    async def enqueue(self, task_id: str, config: Dict, priority: str = "normal") -> str:
+        """
+        将任务加入队列并异步启动
+        
+        Args:
+            task_id: 任务唯一标识
+            config: 任务配置，需包含:
+                - exp_name: 实验名称
+                - version: 模型版本
+                - stages: 阶段配置列表
+            priority: 优先级（当前实现忽略此参数）
+            
+        Returns:
+            job_id: 作业ID
+        """
+        job_id = str(uuid.uuid4())
+        exp_name = config.get("exp_name", "unknown")
+
+        # 持久化到 SQLite
+        async with aiosqlite.connect(self.db_path) as db:
+            await db.execute(
+                '''INSERT INTO task_queue 
+                   (job_id, task_id, exp_name, config, status, created_at)
+                   VALUES (?, ?, ?, ?, 'queued', ?)''',
+                (job_id, task_id, exp_name, json.dumps(config, ensure_ascii=False),
+                 datetime.utcnow().isoformat())
+            )
+            await db.commit()
+
+        # 创建进度队列
+        self.progress_channels[task_id] = asyncio.Queue()
+
+        # 异步启动训练任务
+        asyncio.create_task(self._run_training_async(job_id, task_id, config))
+
+        return job_id
+
+    async def _run_training_async(self, job_id: str, task_id: str, config: Dict) -> None:
+        """
+        异步执行训练 Pipeline
+        
+        Args:
+            job_id: 作业ID
+            task_id: 任务ID
+            config: 任务配置
+        """
+        config_path = None
+
+        try:
+            # 更新状态为 running
+            await self._update_status(
+                job_id,
+                status='running',
+                started_at=datetime.utcnow().isoformat()
+            )
+            await self._send_progress(task_id, {
+                "type": "progress",
+                "status": "running",
+                "message": "训练任务启动中...",
+                "progress": 0.0,
+                "overall_progress": 0.0,
+            })
+
+            # 写入临时配置文件
+            config_path = await self._write_config_file(task_id, config)
+
+            # 获取 run_pipeline.py 脚本路径
+            script_path = self._get_pipeline_script_path()
+
+            # 构建环境变量
+            env = os.environ.copy()
+            env['PYTHONPATH'] = get_pythonpath()
+
+            # 创建子进程
+            process = await asyncio.create_subprocess_exec(
+                sys.executable, script_path,
+                '--config', config_path,
+                '--task-id', task_id,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                env=env,
+                cwd=str(PROJECT_ROOT),
+            )
+
+            self.running_processes[task_id] = process
+            self._running_count += 1
+
+            # 异步监控子进程输出
+            await self._monitor_process_output(task_id, job_id, process)
+
+            # 等待进程完成
+            returncode = await process.wait()
+
+            if returncode == 0:
+                await self._update_status(
+                    job_id,
+                    status='completed',
+                    progress=1.0,
+                    overall_progress=1.0,
+                    message='训练完成',
+                    completed_at=datetime.utcnow().isoformat()
+                )
+                await self._send_progress(task_id, {
+                    "type": "progress",
+                    "status": "completed",
+                    "message": "训练完成",
+                    "progress": 1.0,
+                    "overall_progress": 1.0,
+                })
+            else:
+                # 尝试读取剩余的 stderr
+                stderr_data = await process.stderr.read()
+                error_msg = stderr_data.decode() if stderr_data else f"进程退出码: {returncode}"
+
+                await self._update_status(
+                    job_id,
+                    status='failed',
+                    error_message=error_msg,
+                    completed_at=datetime.utcnow().isoformat()
+                )
+                await self._send_progress(task_id, {
+                    "type": "progress",
+                    "status": "failed",
+                    "message": f"训练失败: {error_msg[:200]}",
+                    "error": error_msg,
+                })
+
+        except asyncio.CancelledError:
+            await self._update_status(
+                job_id,
+                status='cancelled',
+                message='任务已取消',
+                completed_at=datetime.utcnow().isoformat()
+            )
+            await self._send_progress(task_id, {
+                "type": "progress",
+                "status": "cancelled",
+                "message": "任务已取消",
+            })
+
+        except Exception as e:
+            error_msg = str(e)
+            await self._update_status(
+                job_id,
+                status='failed',
+                error_message=error_msg,
+                completed_at=datetime.utcnow().isoformat()
+            )
+            await self._send_progress(task_id, {
+                "type": "progress",
+                "status": "failed",
+                "message": f"任务执行出错: {error_msg}",
+                "error": error_msg,
+            })
+
+        finally:
+            # 清理
+            self.running_processes.pop(task_id, None)
+            self._running_count = max(0, self._running_count - 1)
+
+            # 清理临时配置文件
+            if config_path:
+                await self._cleanup_config_file(config_path)
+
+    async def _monitor_process_output(
+        self,
+        task_id: str,
+        job_id: str,
+        process: asyncio.subprocess.Process
+    ) -> None:
+        """
+        异步监控子进程输出并解析进度
+        
+        Args:
+            task_id: 任务ID
+            job_id: 作业ID
+            process: 子进程对象
+        """
+        async def read_stdout():
+            """读取 stdout 并解析进度"""
+            while True:
+                line = await process.stdout.readline()
+                if not line:
+                    break
+
+                text = line.decode('utf-8', errors='replace').strip()
+                if not text:
+                    continue
+
+                # 检测进度标记
+                if text.startswith(PROGRESS_PREFIX) and text.endswith(PROGRESS_SUFFIX):
+                    json_str = text[len(PROGRESS_PREFIX):-len(PROGRESS_SUFFIX)]
+                    try:
+                        progress_info = json.loads(json_str)
+                        await self._handle_progress(task_id, job_id, progress_info)
+                    except json.JSONDecodeError as e:
+                        # 解析失败，作为普通日志处理
+                        await self._send_progress(task_id, {
+                            "type": "log",
+                            "level": "warning",
+                            "message": f"进度解析失败: {e}",
+                        })
+                else:
+                    # 普通输出，作为日志处理
+                    await self._send_progress(task_id, {
+                        "type": "log",
+                        "level": "info",
+                        "message": text,
+                    })
+
+        async def read_stderr():
+            """读取 stderr 作为错误日志"""
+            while True:
+                line = await process.stderr.readline()
+                if not line:
+                    break
+
+                text = line.decode('utf-8', errors='replace').strip()
+                if text:
+                    await self._send_progress(task_id, {
+                        "type": "log",
+                        "level": "error",
+                        "message": text,
+                    })
+
+        # 并发读取 stdout 和 stderr
+        await asyncio.gather(
+            read_stdout(),
+            read_stderr(),
+            return_exceptions=True
+        )
+
+    async def _handle_progress(
+        self,
+        task_id: str,
+        job_id: str,
+        progress_info: Dict
+    ) -> None:
+        """
+        处理进度信息
+        
+        Args:
+            task_id: 任务ID
+            job_id: 作业ID
+            progress_info: 进度信息字典
+        """
+        # 发送到订阅者
+        await self._send_progress(task_id, progress_info)
+
+        # 更新数据库中的进度
+        updates = {}
+
+        if 'stage' in progress_info:
+            updates['current_stage'] = progress_info['stage']
+        if 'progress' in progress_info:
+            updates['progress'] = progress_info['progress']
+        if 'overall_progress' in progress_info:
+            updates['overall_progress'] = progress_info['overall_progress']
+        if 'message' in progress_info:
+            updates['message'] = progress_info['message']
+        if 'status' in progress_info:
+            updates['status'] = progress_info['status']
+        if 'error' in progress_info:
+            updates['error_message'] = progress_info['error']
+
+        if updates:
+            await self._update_status(job_id, **updates)
+
+    async def _send_progress(self, task_id: str, progress_info: Dict) -> None:
+        """
+        发送进度到订阅队列
+        
+        Args:
+            task_id: 任务ID
+            progress_info: 进度信息
+        """
+        if task_id in self.progress_channels:
+            # 添加时间戳
+            if 'timestamp' not in progress_info:
+                progress_info['timestamp'] = datetime.utcnow().isoformat()
+
+            await self.progress_channels[task_id].put(progress_info)
+
+    async def _update_status(self, job_id: str, **kwargs) -> None:
+        """
+        更新任务状态
+        
+        Args:
+            job_id: 作业ID
+            **kwargs: 要更新的字段
+        """
+        if not kwargs:
+            return
+
+        async with aiosqlite.connect(self.db_path) as db:
+            updates = []
+            values = []
+
+            for key, value in kwargs.items():
+                updates.append(f"{key} = ?")
+                values.append(value)
+
+            values.append(job_id)
+
+            await db.execute(
+                f"UPDATE task_queue SET {', '.join(updates)} WHERE job_id = ?",
+                values
+            )
+            await db.commit()
+
+    async def get_status(self, job_id: str) -> Dict:
+        """
+        获取任务状态
+        
+        Args:
+            job_id: 作业ID
+            
+        Returns:
+            状态字典
+        """
+        async with aiosqlite.connect(self.db_path) as db:
+            db.row_factory = aiosqlite.Row
+            async with db.execute(
+                "SELECT * FROM task_queue WHERE job_id = ?", (job_id,)
+            ) as cursor:
+                row = await cursor.fetchone()
+                if row:
+                    return dict(row)
+
+        return {"status": "not_found", "message": "任务不存在"}
+
+    async def get_status_by_task_id(self, task_id: str) -> Dict:
+        """
+        通过 task_id 获取任务状态
+        
+        Args:
+            task_id: 任务ID
+            
+        Returns:
+            状态字典
+        """
+        async with aiosqlite.connect(self.db_path) as db:
+            db.row_factory = aiosqlite.Row
+            async with db.execute(
+                "SELECT * FROM task_queue WHERE task_id = ?", (task_id,)
+            ) as cursor:
+                row = await cursor.fetchone()
+                if row:
+                    return dict(row)
+
+        return {"status": "not_found", "message": "任务不存在"}
+
+    async def cancel(self, job_id: str) -> bool:
+        """
+        取消任务
+        
+        Args:
+            job_id: 作业ID
+            
+        Returns:
+            是否成功取消
+        """
+        # 查找 task_id
+        async with aiosqlite.connect(self.db_path) as db:
+            async with db.execute(
+                "SELECT task_id, status FROM task_queue WHERE job_id = ?", (job_id,)
+            ) as cursor:
+                row = await cursor.fetchone()
+                if not row:
+                    return False
+                task_id, status = row
+
+        # 如果任务已经完成，无法取消
+        if status in ('completed', 'failed', 'cancelled'):
+            return False
+
+        # 终止进程
+        if task_id in self.running_processes:
+            process = self.running_processes[task_id]
+
+            # 先尝试优雅终止
+            process.terminate()
+
+            try:
+                # 等待进程终止
+                await asyncio.wait_for(process.wait(), timeout=5.0)
+            except asyncio.TimeoutError:
+                # 超时则强制终止
+                process.kill()
+                await process.wait()
+
+            return True
+
+        # 如果进程不在运行（可能还在队列中），直接更新状态
+        await self._update_status(
+            job_id,
+            status='cancelled',
+            message='任务已取消',
+            completed_at=datetime.utcnow().isoformat()
+        )
+
+        # 通知订阅者
+        if task_id in self.progress_channels:
+            await self._send_progress(task_id, {
+                "type": "progress",
+                "status": "cancelled",
+                "message": "任务已取消",
+            })
+
+        return True
+
+    async def subscribe_progress(self, task_id: str) -> AsyncGenerator[Dict, None]:
+        """
+        订阅任务进度（用于 SSE 流）
+        
+        Args:
+            task_id: 任务ID
+            
+        Yields:
+            进度信息字典
+        """
+        # 确保队列存在
+        if task_id not in self.progress_channels:
+            self.progress_channels[task_id] = asyncio.Queue()
+
+        queue = self.progress_channels[task_id]
+
+        # 首先发送当前状态
+        status = await self.get_status_by_task_id(task_id)
+        if status.get("status") != "not_found":
+            yield {
+                "type": "progress",
+                "status": status.get("status"),
+                "stage": status.get("current_stage"),
+                "progress": status.get("progress", 0.0),
+                "overall_progress": status.get("overall_progress", 0.0),
+                "message": status.get("message"),
+                "timestamp": datetime.utcnow().isoformat(),
+            }
+
+        # 持续接收进度更新
+        while True:
+            try:
+                # 30秒超时，发送心跳
+                progress = await asyncio.wait_for(queue.get(), timeout=30.0)
+                yield progress
+
+                # 检查是否为终态
+                if progress.get('status') in ('completed', 'failed', 'cancelled'):
+                    break
+
+            except asyncio.TimeoutError:
+                # 发送心跳保持连接
+                yield {
+                    "type": "heartbeat",
+                    "timestamp": datetime.utcnow().isoformat(),
+                }
+
+    async def list_tasks(
+        self,
+        status: Optional[str] = None,
+        limit: int = 50,
+        offset: int = 0
+    ) -> List[Dict]:
+        """
+        列出任务
+        
+        Args:
+            status: 按状态筛选
+            limit: 返回数量限制
+            offset: 偏移量
+            
+        Returns:
+            任务列表
+        """
+        async with aiosqlite.connect(self.db_path) as db:
+            db.row_factory = aiosqlite.Row
+
+            if status:
+                query = """
+                    SELECT * FROM task_queue 
+                    WHERE status = ? 
+                    ORDER BY created_at DESC 
+                    LIMIT ? OFFSET ?
+                """
+                params = (status, limit, offset)
+            else:
+                query = """
+                    SELECT * FROM task_queue 
+                    ORDER BY created_at DESC 
+                    LIMIT ? OFFSET ?
+                """
+                params = (limit, offset)
+
+            async with db.execute(query, params) as cursor:
+                rows = await cursor.fetchall()
+                return [dict(row) for row in rows]
+
+    async def recover_pending_tasks(self) -> int:
+        """
+        应用重启后恢复未完成的任务
+        
+        将 running 状态的任务标记为 interrupted，
+        可选择重新启动 queued 状态的任务。
+        
+        Returns:
+            恢复的任务数量
+        """
+        async with aiosqlite.connect(self.db_path) as db:
+            # 将 running 状态的任务标记为 interrupted
+            await db.execute(
+                """UPDATE task_queue 
+                   SET status = 'interrupted', 
+                       message = '应用重启导致任务中断'
+                   WHERE status = 'running'"""
+            )
+            await db.commit()
+
+            # 获取 queued 状态的任务
+            db.row_factory = aiosqlite.Row
+            async with db.execute(
+                "SELECT * FROM task_queue WHERE status = 'queued' ORDER BY created_at"
+            ) as cursor:
+                queued_tasks = await cursor.fetchall()
+
+        # 重新启动 queued 状态的任务
+        recovered = 0
+        for task in queued_tasks:
+            task_id = task['task_id']
+            job_id = task['job_id']
+            config = json.loads(task['config'])
+
+            self.progress_channels[task_id] = asyncio.Queue()
+            asyncio.create_task(self._run_training_async(job_id, task_id, config))
+            recovered += 1
+
+        return recovered
+
+    async def cleanup_old_tasks(self, days: int = 7) -> int:
+        """
+        清理旧任务记录
+        
+        Args:
+            days: 保留天数
+            
+        Returns:
+            删除的任务数量
+        """
+        from datetime import timedelta
+
+        cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat()
+
+        async with aiosqlite.connect(self.db_path) as db:
+            cursor = await db.execute(
+                """DELETE FROM task_queue 
+                   WHERE status IN ('completed', 'failed', 'cancelled') 
+                   AND completed_at < ?""",
+                (cutoff,)
+            )
+            deleted = cursor.rowcount
+            await db.commit()
+
+        return deleted
+
+    def _get_pipeline_script_path(self) -> str:
+        """获取 run_pipeline.py 脚本路径"""
+        return str(settings.PIPELINE_SCRIPT_PATH)
+
+    async def _write_config_file(self, task_id: str, config: Dict) -> str:
+        """
+        写入临时配置文件
+        
+        Args:
+            task_id: 任务ID
+            config: 配置字典
+            
+        Returns:
+            配置文件路径
+        """
+        config_path = settings.CONFIGS_DIR / f"{task_id}.json"
+
+        with open(config_path, 'w', encoding='utf-8') as f:
+            json.dump(config, f, ensure_ascii=False, indent=2)
+
+        return str(config_path)
+
+    async def _cleanup_config_file(self, config_path: str) -> None:
+        """
+        清理临时配置文件
+        
+        Args:
+            config_path: 配置文件路径
+        """
+        try:
+            path = Path(config_path)
+            if path.exists():
+                path.unlink()
+        except Exception:
+            pass  # 忽略清理错误

api_server/app/core/__init__.py

@@ -0,0 +1,9 @@
+"""
+核心模块
+
+包含配置、枚举等核心组件
+"""
+
+from .config import settings, PROJECT_ROOT, API_SERVER_ROOT
+
+__all__ = ["settings", "PROJECT_ROOT", "API_SERVER_ROOT"]

api_server/app/core/config.py

@@ -0,0 +1,142 @@
+"""
+环境变量和配置模块
+
+统一管理项目路径、环境配置等
+"""
+
+import os
+from pathlib import Path
+from typing import Literal
+
+# ============================================================
+# 路径常量
+# ============================================================
+
+USER_HOME_ROOT = Path.home()
+
+# api_server/app/core/config.py -> api_server/app/core -> api_server/app -> api_server -> 项目根目录
+API_SERVER_ROOT = Path(__file__).parent.parent.parent.resolve()
+PROJECT_ROOT = API_SERVER_ROOT.parent.resolve()
+
+# GPT_SoVITS 模块路径
+GPT_SOVITS_ROOT = PROJECT_ROOT / "GPT_SoVITS"
+
+# 默认数据目录
+DEFAULT_DATA_DIR = USER_HOME_ROOT / '.moyoyo-tts' / "data"
+
+# 预训练模型目录
+PRETRAINED_MODELS_DIR = GPT_SOVITS_ROOT / "pretrained_models"
+
+# 日志目录
+LOGS_DIR = PROJECT_ROOT / "logs"
+
+
+# ============================================================
+# 配置类
+# ============================================================
+
+class Settings:
+    """
+    API Server 配置
+    
+    支持从环境变量读取配置，提供合理的默认值
+    
+    Example:
+        >>> from api_server.app.core.config import settings
+        >>> print(settings.PROJECT_ROOT)
+        >>> print(settings.DEPLOYMENT_MODE)
+    """
+
+    # 部署模式
+    DEPLOYMENT_MODE: Literal["local", "server"] = os.getenv("DEPLOYMENT_MODE", "local")
+
+    # API 配置
+    API_V1_PREFIX: str = os.getenv("API_V1_PREFIX", "/api/v1")
+    API_HOST: str = os.getenv("API_HOST", "0.0.0.0")
+    API_PORT: int = int(os.getenv("API_PORT", "8000"))
+
+    # 路径配置（可通过环境变量覆盖）
+    PROJECT_ROOT: Path = Path(os.getenv("PROJECT_ROOT", str(PROJECT_ROOT)))
+    API_SERVER_ROOT: Path = Path(os.getenv("API_SERVER_ROOT", str(API_SERVER_ROOT)))
+    DATA_DIR: Path = Path(os.getenv("DATA_DIR", str(DEFAULT_DATA_DIR)))
+
+    # SQLite 数据库路径
+    SQLITE_PATH: Path = Path(os.getenv("SQLITE_PATH", str(DEFAULT_DATA_DIR / "tasks.db")))
+
+    # 任务配置
+    LOCAL_MAX_WORKERS: int = int(os.getenv("LOCAL_MAX_WORKERS", "1"))
+
+    # 预训练模型路径
+    BERT_PRETRAINED_DIR: str = os.getenv(
+        "BERT_PRETRAINED_DIR",
+        str(PRETRAINED_MODELS_DIR / "chinese-roberta-wwm-ext-large")
+    )
+    SSL_PRETRAINED_DIR: str = os.getenv(
+        "SSL_PRETRAINED_DIR",
+        str(PRETRAINED_MODELS_DIR / "chinese-hubert-base")
+    )
+    PRETRAINED_S2G: str = os.getenv(
+        "PRETRAINED_S2G",
+        str(PRETRAINED_MODELS_DIR / "gsv-v2final-pretrained" / "s2G2333k.pth")
+    )
+    PRETRAINED_S2D: str = os.getenv(
+        "PRETRAINED_S2D",
+        str(PRETRAINED_MODELS_DIR / "gsv-v2final-pretrained" / "s2D2333k.pth")
+    )
+    PRETRAINED_S1: str = os.getenv(
+        "PRETRAINED_S1",
+        str(PRETRAINED_MODELS_DIR / "gsv-v2final-pretrained" / "s1bert25hz-5kh-longer-epoch=12-step=369668.ckpt")
+    )
+
+    # Pipeline 脚本路径
+    @property
+    def PIPELINE_SCRIPT_PATH(self) -> Path:
+        """Pipeline 执行脚本路径"""
+        return self.API_SERVER_ROOT / "app" / "scripts" / "run_pipeline.py"
+
+    # 临时配置文件目录
+    @property
+    def CONFIGS_DIR(self) -> Path:
+        """临时配置文件目录"""
+        path = self.DATA_DIR / "configs"
+        path.mkdir(parents=True, exist_ok=True)
+        return path
+
+    def __repr__(self) -> str:
+        return (
+            f"Settings(\n"
+            f"  DEPLOYMENT_MODE={self.DEPLOYMENT_MODE!r},\n"
+            f"  PROJECT_ROOT={self.PROJECT_ROOT},\n"
+            f"  API_SERVER_ROOT={self.API_SERVER_ROOT},\n"
+            f"  DATA_DIR={self.DATA_DIR},\n"
+            f"  SQLITE_PATH={self.SQLITE_PATH},\n"
+            f")"
+        )
+
+
+# 全局配置实例
+settings = Settings()
+
+
+def get_pythonpath() -> str:
+    """
+    获取 PYTHONPATH 环境变量值
+    
+    用于子进程启动时设置正确的模块搜索路径
+    
+    Returns:
+        PYTHONPATH 字符串
+    """
+    paths = [
+        str(PROJECT_ROOT),
+        str(GPT_SOVITS_ROOT),
+    ]
+    return os.pathsep.join(paths)
+
+
+def ensure_data_dirs() -> None:
+    """
+    确保必要的数据目录存在
+    """
+    settings.DATA_DIR.mkdir(parents=True, exist_ok=True)
+    settings.CONFIGS_DIR.mkdir(parents=True, exist_ok=True)

api_server/app/models/__init__.py

@@ -0,0 +1,9 @@
+"""
+模型模块
+
+包含领域模型和 Pydantic Schema
+"""
+
+from .domain import Task, TaskStatus, ProgressInfo
+
+__all__ = ["Task", "TaskStatus", "ProgressInfo"]

api_server/app/models/domain.py

@@ -0,0 +1,172 @@
+"""
+领域模型模块
+
+定义训练任务相关的核心数据结构
+"""
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Enum
+from typing import Dict, Optional, Any
+
+
+class TaskStatus(Enum):
+    """任务状态枚举"""
+    QUEUED = "queued"           # 已入队，等待执行
+    RUNNING = "running"         # 执行中
+    COMPLETED = "completed"     # 已完成
+    FAILED = "failed"           # 失败
+    CANCELLED = "cancelled"     # 已取消
+    INTERRUPTED = "interrupted" # 被中断（应用重启时运行中的任务）
+
+
+@dataclass
+class Task:
+    """
+    训练任务领域模型
+    
+    Attributes:
+        id: 任务唯一标识
+        job_id: 队列作业ID（由任务队列生成）
+        exp_name: 实验名称
+        status: 任务状态
+        config: 任务配置（包含所有训练参数）
+        current_stage: 当前执行阶段
+        progress: 总体进度 (0.0-1.0)
+        stage_progress: 当前阶段进度 (0.0-1.0)
+        message: 最新状态消息
+        error_message: 错误信息（失败时）
+        created_at: 创建时间
+        started_at: 开始执行时间
+        completed_at: 完成时间
+        
+    Example:
+        >>> task = Task(
+        ...     id="task-123",
+        ...     exp_name="my_voice",
+        ...     config={"version": "v2", "batch_size": 4}
+        ... )
+        >>> task.status
+        <TaskStatus.QUEUED: 'queued'>
+    """
+    id: str
+    exp_name: str
+    config: Dict[str, Any]
+    job_id: Optional[str] = None
+    status: TaskStatus = TaskStatus.QUEUED
+    current_stage: Optional[str] = None
+    progress: float = 0.0
+    stage_progress: float = 0.0
+    message: Optional[str] = None
+    error_message: Optional[str] = None
+    created_at: datetime = field(default_factory=datetime.utcnow)
+    started_at: Optional[datetime] = None
+    completed_at: Optional[datetime] = None
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典"""
+        return {
+            "id": self.id,
+            "job_id": self.job_id,
+            "exp_name": self.exp_name,
+            "status": self.status.value,
+            "config": self.config,
+            "current_stage": self.current_stage,
+            "progress": self.progress,
+            "stage_progress": self.stage_progress,
+            "message": self.message,
+            "error_message": self.error_message,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "started_at": self.started_at.isoformat() if self.started_at else None,
+            "completed_at": self.completed_at.isoformat() if self.completed_at else None,
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "Task":
+        """从字典创建实例"""
+        # 处理状态枚举
+        status = data.get("status", "queued")
+        if isinstance(status, str):
+            status = TaskStatus(status)
+        
+        # 处理日期时间
+        def parse_datetime(value):
+            if value is None:
+                return None
+            if isinstance(value, datetime):
+                return value
+            return datetime.fromisoformat(value)
+        
+        return cls(
+            id=data["id"],
+            job_id=data.get("job_id"),
+            exp_name=data["exp_name"],
+            status=status,
+            config=data.get("config", {}),
+            current_stage=data.get("current_stage"),
+            progress=data.get("progress", 0.0),
+            stage_progress=data.get("stage_progress", 0.0),
+            message=data.get("message"),
+            error_message=data.get("error_message"),
+            created_at=parse_datetime(data.get("created_at")),
+            started_at=parse_datetime(data.get("started_at")),
+            completed_at=parse_datetime(data.get("completed_at")),
+        )
+
+
+@dataclass
+class ProgressInfo:
+    """
+    进度信息数据结构
+    
+    用于在子进程和主进程之间传递进度更新
+    
+    Attributes:
+        type: 消息类型 ("progress", "log", "error", "heartbeat")
+        stage: 当前阶段名称
+        stage_index: 当前阶段索引
+        total_stages: 总阶段数
+        progress: 阶段内进度 (0.0-1.0)
+        overall_progress: 总体进度 (0.0-1.0)
+        message: 进度消息
+        status: 状态
+        data: 附加数据
+    """
+    type: str = "progress"
+    stage: Optional[str] = None
+    stage_index: Optional[int] = None
+    total_stages: Optional[int] = None
+    progress: float = 0.0
+    overall_progress: float = 0.0
+    message: Optional[str] = None
+    status: Optional[str] = None
+    data: Dict[str, Any] = field(default_factory=dict)
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典"""
+        return {
+            "type": self.type,
+            "stage": self.stage,
+            "stage_index": self.stage_index,
+            "total_stages": self.total_stages,
+            "progress": self.progress,
+            "overall_progress": self.overall_progress,
+            "message": self.message,
+            "status": self.status,
+            "data": self.data,
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "ProgressInfo":
+        """从字典创建实例"""
+        return cls(
+            type=data.get("type", "progress"),
+            stage=data.get("stage"),
+            stage_index=data.get("stage_index"),
+            total_stages=data.get("total_stages"),
+            progress=data.get("progress", 0.0),
+            overall_progress=data.get("overall_progress", 0.0),
+            message=data.get("message"),
+            status=data.get("status"),
+            data=data.get("data", {}),
+        )

api_server/app/scripts/__init__.py

@@ -0,0 +1,5 @@
+"""
+脚本模块
+
+包含用于子进程执行的独立脚本
+"""

api_server/app/scripts/run_pipeline.py

@@ -0,0 +1,368 @@
+#!/usr/bin/env python3
+"""
+Pipeline 包装脚本
+
+此脚本作为独立子进程运行，执行 TrainingPipeline 并将进度以 JSON 格式输出到 stdout。
+主进程（AsyncTrainingManager）通过解析 stdout 来获取实时进度。
+
+进度消息格式:
+    ##PROGRESS##{"type": "progress", "stage": "...", ...}##
+
+Usage:
+    python run_pipeline.py --config /path/to/config.json --task-id task-123
+"""
+
+import argparse
+import json
+import sys
+import os
+import traceback
+from datetime import datetime
+from typing import Dict, Any
+
+# 确保可以导入项目模块（在导入其他模块之前）
+from pathlib import Path
+_SCRIPT_DIR = Path(__file__).parent.resolve()
+_API_SERVER_ROOT = _SCRIPT_DIR.parent.parent
+_PROJECT_ROOT = _API_SERVER_ROOT.parent
+sys.path.insert(0, str(_PROJECT_ROOT))
+
+# 导入配置模块
+from api_server.app.core.config import settings, PROJECT_ROOT, get_pythonpath
+
+
+# 进度消息前缀和后缀，用于主进程解析
+PROGRESS_PREFIX = "##PROGRESS##"
+PROGRESS_SUFFIX = "##"
+
+
+def emit_progress(progress_info: Dict[str, Any]) -> None:
+    """
+    输出进度消息到 stdout
+    
+    Args:
+        progress_info: 进度信息字典
+    """
+    # 确保有时间戳
+    if "timestamp" not in progress_info:
+        progress_info["timestamp"] = datetime.utcnow().isoformat()
+    
+    json_str = json.dumps(progress_info, ensure_ascii=False)
+    print(f"{PROGRESS_PREFIX}{json_str}{PROGRESS_SUFFIX}", flush=True)
+
+
+def emit_log(level: str, message: str, **extra) -> None:
+    """
+    输出日志消息
+    
+    Args:
+        level: 日志级别 (info, warning, error)
+        message: 日志消息
+        **extra: 额外数据
+    """
+    emit_progress({
+        "type": "log",
+        "level": level,
+        "message": message,
+        **extra
+    })
+
+
+def load_config(config_path: str) -> Dict[str, Any]:
+    """
+    加载配置文件
+    
+    Args:
+        config_path: 配置文件路径
+        
+    Returns:
+        配置字典
+    """
+    with open(config_path, 'r', encoding='utf-8') as f:
+        return json.load(f)
+
+
+def build_pipeline(config: Dict[str, Any]):
+    """
+    根据配置构建 TrainingPipeline
+    
+    Args:
+        config: 配置字典，包含:
+            - exp_name: 实验名称
+            - version: 模型版本
+            - stages: 要执行的阶段列表
+            - 各阶段的具体配置
+            
+    Returns:
+        TrainingPipeline 实例
+    """
+    from training_pipeline import (
+        TrainingPipeline,
+        ModelVersion,
+        # 配置类
+        AudioSliceConfig,
+        ASRConfig,
+        DenoiseConfig,
+        FeatureExtractionConfig,
+        SoVITSTrainConfig,
+        GPTTrainConfig,
+        InferenceConfig,
+        # 阶段类
+        AudioSliceStage,
+        ASRStage,
+        DenoiseStage,
+        TextFeatureStage,
+        HuBERTFeatureStage,
+        SemanticTokenStage,
+        SoVITSTrainStage,
+        GPTTrainStage,
+        InferenceStage,
+    )
+    
+    pipeline = TrainingPipeline()
+    
+    exp_name = config["exp_name"]
+    version_str = config.get("version", "v2")
+    version = ModelVersion(version_str) if isinstance(version_str, str) else version_str
+    
+    # 通用配置参数
+    base_params = {
+        "exp_name": exp_name,
+        "exp_root": config.get("exp_root", "logs"),
+        "gpu_numbers": config.get("gpu_numbers", "0"),
+        "is_half": config.get("is_half", True),
+    }
+    
+    # 阶段配置映射
+    stage_builders = {
+        "audio_slice": lambda cfg: AudioSliceStage(AudioSliceConfig(
+            **base_params,
+            input_path=cfg.get("input_path", ""),
+            threshold=cfg.get("threshold", -34),
+            min_length=cfg.get("min_length", 4000),
+            min_interval=cfg.get("min_interval", 300),
+            hop_size=cfg.get("hop_size", 10),
+            max_sil_kept=cfg.get("max_sil_kept", 500),
+            max_amp=cfg.get("max_amp", 0.9),
+            alpha=cfg.get("alpha", 0.25),
+            n_parts=cfg.get("n_parts", 4),
+        )),
+        
+        "asr": lambda cfg: ASRStage(ASRConfig(
+            **base_params,
+            model=cfg.get("model", "达摩 ASR (中文)"),
+            model_size=cfg.get("model_size", "large"),
+            language=cfg.get("language", "zh"),
+            precision=cfg.get("precision", "float32"),
+        )),
+        
+        "denoise": lambda cfg: DenoiseStage(DenoiseConfig(
+            **base_params,
+            input_dir=cfg.get("input_dir", ""),
+            output_dir=cfg.get("output_dir", "output/denoise_opt"),
+        )),
+        
+        "text_feature": lambda cfg: TextFeatureStage(FeatureExtractionConfig(
+            **base_params,
+            version=version,
+            bert_pretrained_dir=cfg.get("bert_pretrained_dir", 
+                "GPT_SoVITS/pretrained_models/chinese-roberta-wwm-ext-large"),
+            ssl_pretrained_dir=cfg.get("ssl_pretrained_dir",
+                "GPT_SoVITS/pretrained_models/chinese-hubert-base"),
+            pretrained_s2G=cfg.get("pretrained_s2G",
+                "GPT_SoVITS/pretrained_models/gsv-v2final-pretrained/s2G2333k.pth"),
+        )),
+        
+        "hubert_feature": lambda cfg: HuBERTFeatureStage(FeatureExtractionConfig(
+            **base_params,
+            version=version,
+            bert_pretrained_dir=cfg.get("bert_pretrained_dir",
+                "GPT_SoVITS/pretrained_models/chinese-roberta-wwm-ext-large"),
+            ssl_pretrained_dir=cfg.get("ssl_pretrained_dir",
+                "GPT_SoVITS/pretrained_models/chinese-hubert-base"),
+            pretrained_s2G=cfg.get("pretrained_s2G",
+                "GPT_SoVITS/pretrained_models/gsv-v2final-pretrained/s2G2333k.pth"),
+        )),
+        
+        "semantic_token": lambda cfg: SemanticTokenStage(FeatureExtractionConfig(
+            **base_params,
+            version=version,
+            bert_pretrained_dir=cfg.get("bert_pretrained_dir",
+                "GPT_SoVITS/pretrained_models/chinese-roberta-wwm-ext-large"),
+            ssl_pretrained_dir=cfg.get("ssl_pretrained_dir",
+                "GPT_SoVITS/pretrained_models/chinese-hubert-base"),
+            pretrained_s2G=cfg.get("pretrained_s2G",
+                "GPT_SoVITS/pretrained_models/gsv-v2final-pretrained/s2G2333k.pth"),
+        )),
+        
+        "sovits_train": lambda cfg: SoVITSTrainStage(SoVITSTrainConfig(
+            **base_params,
+            version=version,
+            batch_size=cfg.get("batch_size", 4),
+            total_epoch=cfg.get("total_epoch", 8),
+            text_low_lr_rate=cfg.get("text_low_lr_rate", 0.4),
+            save_every_epoch=cfg.get("save_every_epoch", 4),
+            if_save_latest=cfg.get("if_save_latest", True),
+            if_save_every_weights=cfg.get("if_save_every_weights", True),
+            pretrained_s2G=cfg.get("pretrained_s2G",
+                "GPT_SoVITS/pretrained_models/gsv-v2final-pretrained/s2G2333k.pth"),
+            pretrained_s2D=cfg.get("pretrained_s2D",
+                "GPT_SoVITS/pretrained_models/gsv-v2final-pretrained/s2D2333k.pth"),
+            if_grad_ckpt=cfg.get("if_grad_ckpt", False),
+            lora_rank=cfg.get("lora_rank", 32),
+        )),
+        
+        "gpt_train": lambda cfg: GPTTrainStage(GPTTrainConfig(
+            **base_params,
+            version=version,
+            batch_size=cfg.get("batch_size", 4),
+            total_epoch=cfg.get("total_epoch", 15),
+            save_every_epoch=cfg.get("save_every_epoch", 5),
+            if_save_latest=cfg.get("if_save_latest", True),
+            if_save_every_weights=cfg.get("if_save_every_weights", True),
+            if_dpo=cfg.get("if_dpo", False),
+            pretrained_s1=cfg.get("pretrained_s1",
+                "GPT_SoVITS/pretrained_models/gsv-v2final-pretrained/s1bert25hz-5kh-longer-epoch=12-step=369668.ckpt"),
+        )),
+        
+        "inference": lambda cfg: InferenceStage(InferenceConfig(
+            **base_params,
+            version=version,
+            gpt_path=cfg.get("gpt_path", ""),
+            sovits_path=cfg.get("sovits_path", ""),
+            ref_text=cfg.get("ref_text", ""),
+            ref_audio_path=cfg.get("ref_audio_path", ""),
+            target_text=cfg.get("target_text", ""),
+            text_split_method=cfg.get("text_split_method", "cut1"),
+        )),
+    }
+    
+    # 按顺序添加阶段
+    stages = config.get("stages", [])
+    for stage_config in stages:
+        stage_type = stage_config.get("type")
+        if stage_type in stage_builders:
+            stage = stage_builders[stage_type](stage_config)
+            pipeline.add_stage(stage)
+            emit_log("info", f"已添加阶段: {stage.name}")
+        else:
+            emit_log("warning", f"未知阶段类型: {stage_type}")
+    
+    return pipeline
+
+
+def run_pipeline(config: Dict[str, Any], task_id: str) -> bool:
+    """
+    执行 Pipeline
+    
+    Args:
+        config: 配置字典
+        task_id: 任务ID
+        
+    Returns:
+        是否成功完成
+    """
+    emit_progress({
+        "type": "progress",
+        "status": "running",
+        "message": "正在初始化训练流水线...",
+        "task_id": task_id,
+        "progress": 0.0,
+        "overall_progress": 0.0,
+    })
+    
+    try:
+        pipeline = build_pipeline(config)
+        
+        stages = pipeline.get_stages()
+        if not stages:
+            emit_progress({
+                "type": "progress",
+                "status": "failed",
+                "message": "没有配置任何训练阶段",
+                "task_id": task_id,
+            })
+            return False
+        
+        emit_log("info", f"训练流水线已初始化，共 {len(stages)} 个阶段")
+        
+        # 执行 Pipeline
+        for progress in pipeline.run():
+            # 转换进度格式
+            emit_progress({
+                "type": "progress",
+                "status": "running",
+                "stage": progress.get("stage"),
+                "stage_index": progress.get("stage_index"),
+                "total_stages": progress.get("total_stages"),
+                "progress": progress.get("progress", 0.0),
+                "overall_progress": progress.get("overall_progress", 0.0),
+                "message": progress.get("message"),
+                "task_id": task_id,
+                "data": progress.get("data", {}),
+            })
+            
+            # 检查是否失败
+            if progress.get("status") == "failed":
+                emit_progress({
+                    "type": "progress",
+                    "status": "failed",
+                    "stage": progress.get("stage"),
+                    "message": progress.get("message", "阶段执行失败"),
+                    "task_id": task_id,
+                })
+                return False
+        
+        # 完成
+        emit_progress({
+            "type": "progress",
+            "status": "completed",
+            "message": "训练流水线执行完成",
+            "task_id": task_id,
+            "progress": 1.0,
+            "overall_progress": 1.0,
+        })
+        return True
+        
+    except Exception as e:
+        error_msg = str(e)
+        error_trace = traceback.format_exc()
+        emit_progress({
+            "type": "progress",
+            "status": "failed",
+            "message": f"执行出错: {error_msg}",
+            "error": error_msg,
+            "traceback": error_trace,
+            "task_id": task_id,
+        })
+        return False
+
+
+def main():
+    """主函数"""
+    parser = argparse.ArgumentParser(description="执行 GPT-SoVITS 训练流水线")
+    parser.add_argument("--config", required=True, help="配置文件路径 (JSON)")
+    parser.add_argument("--task-id", required=True, help="任务ID")
+    
+    args = parser.parse_args()
+    
+    emit_log("info", f"启动训练任务: {args.task_id}")
+    emit_log("info", f"配置文件: {args.config}")
+    
+    try:
+        config = load_config(args.config)
+    except Exception as e:
+        emit_progress({
+            "type": "progress",
+            "status": "failed",
+            "message": f"加载配置文件失败: {e}",
+            "task_id": args.task_id,
+        })
+        sys.exit(1)
+    
+    success = run_pipeline(config, args.task_id)
+    sys.exit(0 if success else 1)
+
+
+if __name__ == "__main__":
+    main()