本地AI批量任务进度管理：从日志到状态接口的落地实践

“今天的进度”放在本地 AI 工具链里，通常不是一句闲聊，而是一组很具体的状态：任务队列跑到哪一条了、显存还剩多少、有没有任务失败、今天一共产出了多少结果。如果你同时跑过批量图片生成、批量文档解析、多个 TTS 合成任务，就知道“今天进度如何”这件事，本身比“今天做了什么”更难回答。这次我们从本地 AI 批量任务的进度管理角度，讲一套能落地的方法：任务目录怎么规划、日志怎么记录、接口怎么查询进度、批量任务怎么断点续跑。

这篇文章适合正在把本地 AI 工具接进日常生产的读者。无论你用的是图像生成、OCR 文档解析、语音合成还是视频抽帧类工具，只要涉及批量任务、多文件输入、WebUI 或 API 服务，都可以直接参考。不用把“进度”理解成一个复杂平台的功能，它完全可以靠一套合理的目录结构、日志规范和接口约定实现。

1. 核心能力速览

这套方案本质上不是一个单独的开源项目，而是一套围绕“本地 AI 服务 + 任务队列 + 进度查询”的通用做法。它的核心能力可以整理成下面这张表：

有一点需要明确：不同工具的任务提交方式和输出格式差异很大。下面给出的命令和接口示例是通用模板，具体路径、端口、参数需要替换成你实际使用的项目配置。

2. 适用场景与使用边界

2.1 适合谁

• 内容生产者：每天批量生成配图、缩略图，需要知道今天完成了多少张、失败了多少张。
• 文档处理用户：用 OCR 模型解析大量 PDF，需要记录每个文件的解析状态和输出 Markdown 路径。
• 语音合成用户：批量合成多段音频，需要按文本文件逐条执行，并记录每个片段的输出位置。
• 本地工具链维护者：把 WebUI 或 API 服务接到自己的脚本里，需要一套稳定的任务状态查询方式。

这类场景有一个共同特点：任务量大、单条任务时长短、失败需要重试。如果只是偶尔跑一张图、转一段文字，“进度”不是刚需；一旦进入批量阶段，进度管理就直接影响效率。

2.2 不适合什么场景

• 需要多人协作、实时看板、权限管理的团队任务调度，应该用更完整的任务队列系统。
• 任务之间有复杂依赖关系的流程，例如 A 任务完成后才能触发 B 任务，建议使用工作流引擎。
• 单次任务就要跑数小时的大模型训练，这类进度管理需要专门的训练日志和检查点机制。

2.3 使用边界与合规提醒

使用本地 AI 工具处理批量任务时，有几个底线必须守住：

• 处理人脸图片时，需要确认图片来源合法，并且获得了人脸信息的使用授权。
• 处理语音数据时，需要确认声音素材的授权范围，不能随意克隆或合成他人声线用于商用。
• 处理版权素材时，例如书籍扫描件、付费文档、影视截图，需要确认你是否有权进行解析和转写。
• 涉及用户隐私数据时，建议先在完全隔离的测试环境中验证，避免数据泄露。

本地部署的优势是数据不出本机，但这不等于可以随便处理他人数据。合规边界永远优先于技术效率。

3. 前置条件与目录规划

3.1 环境准备

在开始跑批量任务之前，先把基础环境核对一遍：

• 操作系统：建议使用 Linux 或 Windows，Mac 要看具体项目是否支持。
• Python 版本：多数本地 AI 工具依赖 Python 3.10 或 3.11，具体版本以项目文档为准。
• GPU 驱动与 CUDA：NVIDIA 显卡需要安装对应版本的驱动和 CUDA 工具包，AMD 或 Apple Silicon 需要看项目是否支持。
• 依赖管理：建议每个项目使用独立虚拟环境，避免依赖冲突。
• 磁盘空间：模型文件、输入素材、输出结果需要分开目录存放，建议预留至少模型体积 2 倍的剩余空间。

如果你不确定当前机器的显卡和显存情况，在系统里执行下方命令查看：

# Windows nvidia-smi # Linux nvidia-smi # macOS system_profiler SPDisplaysDataType

nvidia-smi可以看到显卡型号、驱动版本、显存总量和当前占用。这是后面观察显存占用最直接的命令。

3.2 目录结构设计

“今天的进度”能不能一眼看懂，很大程度上取决于目录设计。推荐按下面这种方式组织：

project/ ├── inputs/ # 输入素材，按日期或批次分目录 │ └── 20250115/ ├── outputs/ # 输出结果 │ └── 20250115/ ├── logs/ # 运行日志 │ └── 20250115.log ├── status/ # 任务状态文件 │ └── 20250115.json └── scripts/ # 启动脚本和批量任务脚本

按日期分目录的好处是：每天的任务输入、输出、日志、状态都可以对应起来。想查“昨天的进度”，直接看昨天的 logs 和 status 文件；想复现当天结果，直接找当天的 inputs 和 outputs。

如果你的任务量不大，也可以简化成 inputs、outputs、logs 三个目录，但状态文件建议保留。它是批量任务断点续跑的基础。

4. 任务启动与进度记录方式

4.1 手动运行单条任务

本地 AI 工具通常会有两种启动方式：WebUI 页面和命令行 API。先用命令行跑通一条单任务，确认模型能正常加载、输出目录能正常写入，再进入批量阶段。

以 API 服务为例，启动命令通常是这样的形式：

# 示例命令，具体参数需要按实际项目调整 python app.py --host 127.0.0.1 --port 7860

启动后，服务会监听本地的7860端口。浏览器打开http://127.0.0.1:7860，可以进入 WebUI 页面；命令行可以通过接口提交任务。

启动时需要注意几点：

• 如果端口被占用，更换端口号，或者先杀掉占用进程。
• 日志输出到控制台的同时，建议同时写入日志文件。
• 首次启动会加载模型文件，耗时可能较长，属于正常现象。

4.2 给日志加上时间戳

进度管理的第一个关键动作，是让日志带上时间戳。很多默认日志只输出任务名称和结果摘要，缺少时间信息，导致后来很难判断一个任务到底跑了多久。

Python 的logging模块配置时间戳并不复杂：

import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", handlers=[ logging.FileHandler("logs/20250115.log", encoding="utf-8"), logging.StreamHandler() ] )

之后在任务脚本里调用：

logging.info("开始处理任务：input_001.png") # 这里是调用模型的代码 logging.info("任务完成：input_001.png -> output_001.png")

这样日志文件里就会出现类似下面的内容：

2025-01-15 09:00:12 [INFO] 开始处理任务：input_001.png 2025-01-15 09:00:47 [INFO] 任务完成：input_001.png -> output_001.png 2025-01-15 09:00:47 [INFO] 开始处理任务：input_002.png

到这一步，你已经拥有了一个最基础但可用的进度记录系统。接下来可以通过状态文件实现批量任务的进度追踪。

5. 功能测试：从单任务到批量任务

5.1 功能验证维度

不管底层是什么模型，建议按以下维度做验证：

• 单任务能否成功执行，输出文件是否完整。
• 任务执行时间是否符合预期。
• 显存占用是否稳定，有没有溢出风险。
• 连续执行多个任务时，服务是否稳定。
• 失败任务能否定位原因并重试。
• 输出文件名是否与输入文件名有明确的对应关系。

5.2 批量任务脚本

批量任务的核心逻辑是：遍历输入目录，逐个提交任务，记录每个任务的状态。下面是一个通用模板，适用于大部分支持 API 调用的本地 AI 工具：

import json import logging import os import time from pathlib import Path import requests # 配置 INPUT_DIR = Path("inputs/20250115") OUTPUT_DIR = Path("outputs/20250115") STATUS_FILE = Path("status/20250115.json") API_URL = "http://127.0.0.1:7860/api/generate" OUTPUT_DIR.mkdir(parents=True, exist_ok=True) STATUS_FILE.parent.mkdir(parents=True, exist_ok=True) # 加载已有状态，用于断点续跑 status = {} if STATUS_FILE.exists(): with open(STATUS_FILE, "r", encoding="utf-8") as f: status = json.load(f) input_files = sorted(INPUT_DIR.glob("*")) for idx, file_path in enumerate(input_files, start=1): if file_path.name in status and status[file_path.name] == "done": logging.info("跳过已完成任务：%s", file_path.name) continue # 提交任务，这里需要按实际项目调整请求参数 payload = { "filename": file_path.name, "output_dir": str(OUTPUT_DIR), } try: logging.info("处理中（%d/%d）：%s", idx, len(input_files), file_path.name) response = requests.post(API_URL, json=payload, timeout=600) response.raise_for_status() # 标记成功 status[file_path.name] = "done" with open(STATUS_FILE, "w", encoding="utf-8") as f: json.dump(status, f, ensure_ascii=False, indent=2) except Exception as e: status[file_path.name] = f"error: {e}" with open(STATUS_FILE, "w", encoding="utf-8") as f: json.dump(status, f, ensure_ascii=False, indent=2) logging.error("任务失败：%s，错误：%s", file_path.name, e) # 每个任务之间稍作间隔，避免请求过密 time.sleep(1)

这个脚本最关键的一点是：每完成一个任务，就把状态写入 JSON 文件。下次再运行脚本时，已完成的任务会被跳过，失败或者未执行的任务会继续处理，这就是断点续跑的基础。

5.3 状态文件示例

运行一段时间后，status/20250115.json的内容大概是这样的：

{ "input_001.png": "done", "input_002.png": "done", "input_003.png": "error: HTTP 500 Internal Server Error", "input_004.png": "done" }

查看这个文件，就能知道哪些任务成功了，哪些失败了，失败原因是什么。这比看一张进度条要可靠得多，因为批量任务往往不是顺序执行一次就完，而是需要反复排查和重试的。

5.4 判断成功的标准

一个任务是否真正成功，不能只看日志提示“完成”。建议做以下确认：

• 输出文件是否存在，且大小不为 0。
• 输出文件内容是否符合预期，例如图片能正常打开、音频文件可以播放、OCR 结果包含目标文本。
• API 是否返回了成功状态码。
• 日志中是否有隐性警告，例如显存接近上限导致的自动降级。

6. 接口 API 与页面化进度查询

6.1 为什么需要接口

日志文件能解决事后排查，但在任务运行过程中，你很可能想随时看一眼“现在跑到第几个了”。这时候需要 API 提供实时状态。

很多本地 AI 工具本身带有进度查询接口，例如 ComfyUI 的队列查询、TTS 服务的任务状态查询。如果你的工具没有现成接口，可以自己写一个简单的状态服务，读取 status JSON 文件并返回。

6.2 一个轻量级进度接口示例

下面这一段可以作为参考，用 Flask 或 FastAPI 写一个极简的状态查询接口：

from flask import Flask, jsonify import json from pathlib import Path app = Flask(__name__) STATUS_FILE = Path("status/20250115.json") @app.route("/api/progress", methods=["GET"]) def get_progress(): if not STATUS_FILE.exists(): return jsonify({"total": 0, "done": 0, "failed": 0, "tasks": {}}) with open(STATUS_FILE, "r", encoding="utf-8") as f: status = json.load(f) total = len(status) done = sum(1 for v in status.values() if v == "done") failed = sum(1 for v in status.values() if v.startswith("error")) return jsonify({ "total": total, "done": done, "failed": failed, "remaining": total - done - failed, "tasks": status }) if __name__ == "__main__": # 只监听本机地址，避免外部访问 app.run(host="127.0.0.1", port=8900)

启动这个服务后，浏览器打开http://127.0.0.1:8900/api/progress，看到的内容类似：

{ "total": 4, "done": 2, "failed": 1, "remaining": 1, "tasks": { "input_001.png": "done", "input_002.png": "done", "input_003.png": "error: HTTP 500 Internal Server Error", "input_004.png": "done" } }

这样可以通过 curl 直接查询：

curl http://127.0.0.1:8900/api/progress

接口能力可以让进度管理从“手动看日志”升级为“脚本自动查询”。如果你愿意，还可以写一个更简单的 Web 页面定时刷新，这样在 WebUI 旁边开一个小窗口就能看到整体进度。

6.3 批量任务队列设计建议

当任务量很大时，直接在一个循环里顺序跑会存在几个问题：

• 单个任务卡住会导致整个队列卡住。
• 中途中断后，虽然状态文件能帮助续跑，但已经输出的结果不会自动清理。
• 任务之间没有依赖关系时，其实可以考虑并发，但显存和内存需要评估。

建议采用简单的“单进程顺序执行 + 状态文件记录 + 失败重试”模式。顺序执行虽然慢，但稳定，显存占用可控。如果你的显卡显存足够大，可以尝试同时运行 2 到 4 个任务，但需要预先观察显存占用，避免 OOM。

7. 资源占用与性能观察

7.1 显存占用怎么看

观察显存占用最简单的方式是执行nvidia-smi，可以看到每个进程的显存使用量。

批量任务运行时，建议注意以下几点：

• 显存占用与模型大小、输入分辨率、批量大小直接相关，不同工具差异很大。
• 显存不足时，任务通常会报错，错误信息里可能包含out of memory、CUDA OOM等关键词。
• 如果显存比较紧张，可以选择降低分辨率、减少并发数、使用 CPU 推理（速度会慢很多）。

7.2 性能影响要素

影响批量任务整体耗时的因素主要包括：

• 模型大小：模型参数越多，单次推理越慢。
• 输入大小：图片分辨率、音频时长、文本长度都会影响推理时间。
• 批量大小：适当地提高批量大小可以提升 GPU 利用率，但会提高显存占用。
• 并发任务数：并发任务越多，资源竞争越明显，单个任务速度反而可能下降。

建议第一次跑批量任务时，先用 5 到 10 个小样本测试，记录单任务耗时和显存占用，再放大到完整数据集。这样可以避免大规模任务跑到一半发现显存不够或者耗时长到无法接受。

7.3 降低显存占用的常用做法

如果你的显卡显存有限，常规思路包括：

• 降低输入分辨率，比如图片从 1024 降到 768。
• 减少批量大小，从 4 降到 2 或 1。
• 使用模型 fp16 或 int8 量化版本，前提是工具支持且质量可接受。
• 清理显存缓存，检查是否有残留进程占住显存。
• 尽量避免多个 WebUI 服务同时运行，互相挤占显存。

8. 常见问题与排查方法

9. 最佳实践与使用建议

9.1 第一次先小参数测试

批量任务最容易踩的坑，是一上来就用大批量、高分辨率测试。建议先跑 5 到 10 个样本，确认单任务耗时、显存占用、输出质量符合预期，再放大任务量。如果小样本测试就出现显存不足或者服务崩溃，放大任务量只会更严重。

9.2 保留一套最小可运行配置

每次调试完，把能跑通的最小配置单独保存下来。包括启动命令、Python 依赖清单、模型文件路径、输入素材样例。这样后续环境变化时可以快速恢复到可用状态。

9.3 目录与状态文件规范化

输入素材、输出结果、运行日志、状态文件分目录管理，并且按日期命名。看似多花了几秒钟，但后续排查问题、重跑任务、汇报进度都会非常省事。输出文件命名建议包含输入文件名和任务标识，例如input_001__done.png。

9.4 批量任务加入日志与失败重试

不要让失败任务静默跳过。每个失败任务至少记录日志和状态文件。重试时不要无脑重跑全部任务，而是通过状态文件找出失败或未完成的任务单独处理。

9.5 接口服务限制访问范围

如果启动 API 服务，建议默认绑定127.0.0.1，不要绑定0.0.0.0，避免局域网其他设备直接访问。如果确实需要远程使用，要加上简单的访问控制或至少使用防火墙限制端口。本地 AI 服务通常没有内置权限认证，暴露在公网有风险。

9.6 涉及人脸、声音、版权素材必须确认授权

无论是图像生成、声音克隆还是文档解析，只要素材涉及他人肖像、声音、版权内容，都要先确认授权范围。本地部署不能作为免责理由，合规要求仍然存在。发布或商用前需要做效果复核。

9.7 发布或商用前复核输出

AI 工具的批量输出可以节省大量时间，但质量并不总是稳定。建议在正式发布或商用前，抽查输出结果，确认没有明显瑕疵或侵权内容。尤其是人脸、文字、标识等敏感信息，需要人工复核。

9.8 定期清理旧日志和旧输出

日志文件和输出文件会快速累积。建议每个批次结束后归档，每月清理一次早期日志，避免磁盘空间被无限占用。

10. 总结与下一步

“今天的进度”在本地 AI 批量任务中并不难实现。核心是三个动作：按日期规划输入输出目录，用状态文件记录每个任务的成功与失败，用日志时间戳记录执行耗时。这三个动作做完，你已经可以从容回答任何一天的任务进度。如果想要更进一步，可以再加一个轻量级进度查询接口，把状态文件变成可以随时访问的 API。

这套方法最适合的场景，是每天固定跑一批 AI 任务的个人或小团队。它不依赖复杂平台，不需要额外数据库，只要会写简单的 Python 脚本就可以维护。需要最先验证的是：你的服务能不能稳定处理批量任务，状态文件在断点续跑时是否准确，显存占用在连续运行后是否保持不变。最容易踩的坑，则是没有给任务设置合理的超时时间，导致一个失败任务卡住整个队列。

建议先把你当前最常用的本地 AI 工具接进来，跑一个 10 条任务的样本，把状态文件生成出来。确认这套流程能流畅运转后，再逐步扩展到日常生产。等积累一段时间，这些日志和状态文件会成为你复盘效率、优化参数最有价值的数据。