MultiTaskFlow 是一个轻量级的多任务流管理工具，用于按顺序执行和监控一系列任务。它可以帮助您管理数据处理、模型训练、评估等一系列需要顺序执行的任务，并提供实时状态更新和执行结果跟踪。
- 现代化界面：全新的 React + TypeScript Web 界面
- 多队列管理：同时管理多个任务队列
- 实时日志：WebSocket 推送实时日志
- 任务操作：创建、编辑、删除、排序任务
- 消息推送：任务完成/失败时推送微信通知
- 认证保护：密码保护，会话持久化
- 基于YAML配置文件定义任务流
- 支持Python脚本和Shell命令的执行
- 提供任务状态实时监控
- 自动执行失败任务的重试逻辑
- 完整的日志记录和任务执行历史
- 进程PID跟踪与管理
- 优雅的信号处理和任务终止
- 支持静默模式，可跳过消息通知
- 智能环境变量加载：优先从配置文件同目录查找
.env文件 - 环境变量配置检查：启动时自动显示环境变量配置状态
- 彩色输出支持：可选安装 colorama 获得更友好的输出体验
- Python 3.7+
- PyYAML
- 其他依赖库（如有）
MultiTaskFlow 采用智能的 .env 文件加载策略，按以下优先级查找：
-
配置文件同目录（推荐）：与
tasks.yaml放在同一目录/your/project/ ├── tasks.yaml └── .env # 优先加载这个 -
当前工作目录：运行命令时的工作目录
/current/directory/ └── .env # 次优先 -
向上递归查找：从当前目录向上查找直到找到
.env文件
优势：
- 不同项目可以有独立的环境配置
- 配置文件和环境变量放在一起，便于管理
- 启动时会显示加载的
.env文件位置，方便确认
启动时，MultiTaskFlow 会自动显示环境变量配置状态：
============================================================
[环境变量配置检查]
============================================================
.env 文件: /path/to/your/.env
MSG_PUSH_TOKEN: AT_abc123...xyz789 (已配置)
MTF_SILENT_MODE: false (消息推送已启用)
============================================================
- Token 会脱敏显示（仅显示前后各6位）
- 如果未找到
.env文件，会提示推荐的创建位置 - 运行过程中修改
.env文件，下一个任务会自动检测并显示变化
在使用消息推送功能前，需要配置 MSG_PUSH_TOKEN 环境变量。以下是配置方法：
在您的任务配置文件（如 tasks.yaml）同目录下创建 .env 文件：
# .env 文件内容
MSG_PUSH_TOKEN=your_pushplus_token_here
MTF_SILENT_MODE=false优势：配置随项目走，不影响其他项目。
在 ~/.bashrc 或 ~/.zshrc 文件中添加：
# MultiTaskFlow 消息推送配置
export MSG_PUSH_TOKEN="your_pushplus_token_here"然后重新加载配置：
source ~/.bashrc # 或 source ~/.zshrc在运行命令前设置：
MSG_PUSH_TOKEN=your_token python your_script.py如果您不希望收到消息通知，可以启用静默模式。
# .env 文件内容
MTF_SILENT_MODE=true在 ~/.bashrc 或 ~/.zshrc 文件中添加：
# MultiTaskFlow 静默模式配置
export MTF_SILENT_MODE=true然后重新加载配置：
source ~/.bashrc # 或 source ~/.zshrc在运行命令前设置：
MTF_SILENT_MODE=true taskflow your_tasks.yaml支持的值：
- 启用静默模式：
true,1,yes,on - 禁用静默模式：
false,0,no,off或不设置
- 访问 PushPlus 官网
- 注册并登录
- 在个人中心获取您的 token
# 基础安装（仅 CLI 命令行工具）
pip install multitaskflow
# 完整安装（CLI + Web UI）⭐ 推荐
pip install multitaskflow[web]
# 安装并启用彩色输出
pip install multitaskflow[color]
# 全部功能
pip install multitaskflow[web,color]Web UI 功能：
- 安装
web扩展后，可使用taskflow web命令启动可视化管理界面 - 包含 FastAPI + React 前端，支持多队列管理、实时日志、消息推送等功能
彩色输出功能：
- 安装
colorama后，环境变量配置检查会以彩色显示 - 绿色表示已配置，红色表示未设置，黄色表示警告
- 如果不安装，会自动回退到普通文本输出，完全不影响功能
# 克隆仓库
git clone https://github.com/Polaris-F/MultiTaskFlow.git
cd MultiTaskFlow
# 方法1: 使用pip直接安装
pip install .
# 方法2: 开发模式安装
pip install -e .如果您想构建wheel包或源码分发包，可以使用以下命令：
# 安装构建工具
pip install build
# 构建分发包
python -m build
# 构建的包会在dist/目录下生成如果需要使用 消息接收功能，请访问 https://www.pushplus.plus/ 获取您的token
创建一个YAML格式的任务配置文件，定义您要执行的任务序列：
# tasks.yaml 示例
- name: "任务1-数据准备"
command: "python scripts/prepare_data.py --input data/raw --output data/processed"
status: "pending"
- name: "任务2-模型训练"
command: "python scripts/train_model.py --data data/processed --epochs 10"
status: "pending"
- name: "任务3-结果评估"
command: "python scripts/evaluate.py --model-path models/latest.pt"
status: "pending"在您的Python代码中使用MultiTaskFlow：
from multitaskflow import TaskFlow
# 创建任务流管理器
task_manager = TaskFlow("path/to/your/tasks.yaml")
# 启动任务执行
task_manager.run()
# 您也可以动态添加任务
task_manager.add_task_by_config(
name="额外任务",
command="echo '这是一个动态添加的任务'"
)安装后，您可以直接使用taskflow命令行工具：
# 使用配置文件运行任务流
taskflow path/to/your/tasks.yaml
# 使用默认配置
# 如果不提供配置文件路径，将在examples/tasks.yaml创建示例配置
taskflow
# 查看帮助
taskflow --help启动 Web UI 可视化管理界面：
# 使用当前目录作为工作空间
taskflow web
# 加载指定 YAML 配置文件
taskflow web tasks.yaml
# 指定端口
taskflow web --port 9000
# 指定工作空间目录
taskflow web -w /path/to/workspace
# 查看帮助
taskflow web --helpWeb UI 功能：
- 📋 多队列管理：同时管理多个任务队列
- ✏️ 任务操作：创建、编辑、删除、排序任务
▶️ 执行控制：单任务执行、队列自动执行- 📊 实时日志：WebSocket 推送实时日志
- 🔔 消息推送：任务完成/失败时推送微信通知
- 🔐 认证保护：首次使用需设置密码
首先taskflowPro.sh修改脚本中 TASK_CONFIG为任务流yaml路径
chmod +x taskflowPro.sh
./taskflowPro.sh start # 开始运行
./taskflowPro.sh stop # 结束运行您可以运行我们提供的演示脚本，查看任务管理和消息接收的实际效果。演示脚本模拟了一个完整的深度学习工作流，包括数据预处理、模型训练、模型评估和数据归档等步骤。
# 安装完成后，直接运行示例脚本
python -m multitaskflow.examples.demo
# 或使用命令行工具
taskflow examples/tasks.yaml演示脚本将依次执行以下任务：
- 数据预处理 - 模拟数据集加载、清洗和处理过程
- 模型训练-阶段1 - 模拟第一阶段模型训练过程
- 模型评估-阶段1 - 模拟对第一阶段训练模型的评估
- 模型训练-阶段2 - 模拟基于第一阶段模型继续训练
- 模型评估-阶段2 - 模拟对第二阶段训练模型的评估
- 数据归档 - 模拟模型和结果数据的归档过程
每个任务都会显示详细的执行进度和模拟输出，让您直观了解MultiTaskFlow的任务管理能力。所有演示任务都是模拟执行，不会创建实际文件或占用大量资源。
运行示例后，您将看到：
- 任务管理器启动和初始化过程
- 任务状态的实时更新（等待中→执行中→完成/失败）
- 每个任务的详细输出和进度信息
- 任务完成后的状态汇总
通过观察演示效果，您可以了解MultiTaskFlow如何帮助管理复杂的多步骤工作流程，以及它如何提供清晰的任务执行状态和结果反馈。
实际运行时在控制台中会看到详细的输出，显示任务状态和进度信息
任务配置文件支持以下选项：
- name: "示例任务"
command: "python script.py"
status: "pending" # pending, running, completed, failed, skipped
retry: 3 # 失败后重试次数 (TODO)
timeout: 3600 # 任务超时时间（秒）(TODO)
depends_on: ["前置任务名称"] # 依赖的任务 (TODO)status 字段说明：
pending: 待执行（默认值）running: 执行中（系统自动设置）completed: 已完成（系统自动设置）failed: 执行失败（系统自动设置）skipped: 跳过执行（手动设置，加载时会被过滤）
使用 skipped 状态：
如果您不想执行某个任务，可以在配置文件中将其 status 设置为 skipped，该任务将不会被加载到执行队列中。
- name: "临时禁用的任务"
command: "python old_script.py"
status: "skipped" # 这个任务不会执行MultiTaskFlow 支持静默模式，在此模式下不会发送任何消息通知。这对于以下场景非常有用：
- 生产环境部署：在生产环境中运行时，可能不需要消息通知
- 调试阶段：开发和调试过程中避免频繁接收通知
- 批量任务：执行大量批处理任务时，只关注最终结果而非每个任务
- CI/CD 流程：在自动化构建流水线中使用，避免触发过多通知
静默模式通过环境变量 MTF_SILENT_MODE 控制：
# 启用静默模式
export MTF_SILENT_MODE=true
# 临时启用
MTF_SILENT_MODE=true taskflow tasks.yaml支持的值：
- 设为
true,1,yes,on表示启用静默模式 - 不设置或设为其他值表示禁用静默模式
当启用静默模式时：
- 所有任务执行完成后不会发送消息通知
- 任务管理器完成时不会发送总结报告
- 所有操作和结果仍会记录在日志文件中
- 控制台输出不受影响，仍然会显示所有信息
注意，静默模式只影响消息通知行为，不会改变任务的实际执行过程。
您可以配置系统在任务状态变更时发送通知：
from multitaskflow import TaskFlow, Msg_push
# 创建消息推送实例
notifier = Msg_push(
webhook_url="your_webhook_url",
channel="your_channel"
)
# 创建带通知功能的任务流管理器
task_manager = TaskFlow(
"tasks.yaml",
msg_push=notifier
)MultiTaskFlow设计为可扩展的，您可以：
- 自定义任务状态处理逻辑
- 添加新的任务类型
- 扩展监控和报告功能
from multitaskflow import TaskFlow
class CustomTaskFlow(TaskFlow):
def process_task_output(self, task, output):
# 自定义输出处理逻辑
print(f"处理任务 {task.name} 的输出: {output}")
# 继续处理...
super().process_task_output(task, output)功能描述：支持运行时监控配置文件变化，实现任务的动态增删改
实现方案：
- 使用
watchdog库监控 YAML 配置文件变化 - 基于任务 ID 的增量更新机制
- 支持以下动态操作：
- ✅ 新增任务：在配置中添加新任务（需有唯一 ID）会自动加入执行队列
- ✅ 标记跳过：将任务 status 改为
skipped可从队列中移除 - ✅ 删除任务：从配置中删除的任务会从队列中移除
- ✅ 更新参数：修改现有任务参数（保持 ID 不变）会更新队列中的任务
配置要求：
# 每个任务必须有唯一的 id 字段
tasks:
- id: task_a # 必须唯一
name: "数据预处理"
command: "python preprocess.py"
status: pending # pending | running | completed | failed | skipped
- id: task_b
name: "模型训练"
command: "python train.py"
status: pending
- id: task_c
name: "旧任务"
status: skipped # 标记为跳过，不会执行使用场景：
- 长时间运行的任务流中，临时禁用某些任务
- 根据前面任务的结果动态添加新任务
- 修正配置错误而不需要重启整个流程
技术实现：
- 依赖：
watchdog>=2.1.0 - 核心逻辑：文件变化监听 + ID 去重 + 队列增量更新
- 日志输出：实时显示任务的添加、移除、更新操作
当前状态：设计阶段，欢迎反馈和建议
Q: XXXX？
欢迎贡献代码、报告问题或提出新功能建议！
- Fork 这个仓库
- 创建您的特性分支 (
git checkout -b feature/amazing-feature) - 提交您的更改 (
git commit -m 'Add some amazing feature') - 推送到分支 (
git push origin feature/amazing-feature) - 打开一个 Pull Request
-
1.0.0 - 2026-01-18 🎉 正式发布
- 🖥️ Web UI 可视化管理界面：全新的 React + TypeScript 前端
- 📟 CLI 集成：
taskflow web命令一键启动 Web UI - 📋 多队列管理：支持同时管理多个任务队列
- 🔔 消息推送：集成 PushPlus，任务完成/失败时推送微信通知
- 🔐 认证保护：密码保护 + 会话持久化
- 📊 实时日志：WebSocket 推送实时日志
-
0.1.5 - 2025-11-10
- 任务跳过功能（status: skipped）
- 智能环境变量加载
-
0.1.4 - 2025-11-10
- 环境变量配置检查
- 彩色输出支持
本项目采用MIT许可证 - 详情请查看 LICENSE 文件
- 主要开发者: Polaris
- 感谢所有贡献者和使用者的宝贵反馈