CrispASR Executor

LTAB 异步任务执行器 — 基于 CrispASR 的音频/视频语音转录服务。从 LTAB 拉取任务，下载音频，调用 CrispASR 进行 ASR 转写，生成 Markdown 文档，上传到 MinIO，回报结果。

架构位置

Caller ──→ LTAB ──→ [本项目] ──→ CrispASR 二进制 ──→ MinIO
                     Executor

本项目是 LTAB 的一个 Executor 实现，可作为模板参考，用于编写其他类型的 Executor。

核心流程

1. 心跳上报 (每30s)
   POST /executor/heartbeat
   Authorization: Bearer layla_xxxx
   Body: {ip, input_schema}

2. 拉取任务
   GET /executor/next?task_type=crispasr
   ← {task_id, task_type, source_name, source_url}

3. 解析来源
   source_url 可能是:
   - JSON 字符串: {"source":"bilibili","url":"...","language":"zh"}
   - JSON 字符串: {"source":"url","url":"https://direct-link/..."}
   - JSON 字符串: {"source":"s3","key":"submits/xxx.mp4"}

4. 下载音频
   - bilibili: yt-dlp (主) → Bilibili API (备)
   - url: 直接下载
   - s3: boto3 从 MinIO 下载

5. 音频转码
   ffmpeg -i input -vn -ar 16000 -ac 1 -c:a pcm_s16le output.wav

6. ASR 转写
   crispasr --backend qwen3 -m model.gguf -f audio.wav --output-txt -l zh

7. 生成 Markdown
   包含: 标题、任务ID、模型信息、时长、转录时间、转录内容

8. 上传结果
   boto3 → MinIO (transfer-station/results/)

9. 回报结果
   POST /executor/result/{task_id}?result_key=results/xxx&result_name=xxx.md

进度上报策略

原则

进度必须尽可能反映真实完成情况。上报规则：

1. 有真实里程碑时，上报真实进度：下载完成、转码完成、处理完成等
2. 处理黑盒工具时，使用时间估算：如果工具不输出中间进度，用渐近曲线估算
3. 估算进度不得超过下一个真实里程碑：比如转写阶段估算不能超过 79%（因为 80% 是转写完成）
4. 估算必须有客观依据：基于音频时长、模型速度等可计算因素，不能凭空填数字

CrispASR 进度分布

进度	类型	含义	依据
10%	✅ 真实	输入解析完成	source_url 解析成功
25%	✅ 真实	音频下载完成	文件落盘，字节数确认
50%	✅ 真实	ffmpeg 转码完成	WAV 文件生成
50-79%	⏳ 估算	CrispASR 转写中	渐近曲线：50 + 29 × (1 - e^(-3t/T))
80%	✅ 真实	转写完成	.txt 文件生成
90%	✅ 真实	Markdown 生成完成	.md 文件生成
95%	✅ 真实	MinIO 上传完成	S3 返回成功

为什么 50-79% 是估算

CrispASR CLI 是黑盒二进制，不输出中间进度：

• --print-progress 对分片音频无效
• CLI 模式不像 server 模式输出 slice N/M done
• stderr 只有启动日志和最终统计

因此 CrispASR 处理期间无法获取真实进度，只能用时间估算。

估算算法

使用渐近曲线（不是线性），特点：

• 前期快速推进（给用户信心）
• 后期逐渐减速（接近预期完成时间时变慢）
• 超过预期时间后继续缓慢推进（不会卡死在某个百分比）
• 永远不会达到 80%（那是真实完成的标志）

import math

audio_duration_sec = wav_path.stat().st_size / (16000 * 2)  # 16kHz 16-bit mono
estimated_sec = audio_duration_sec / 10.0  # GPU 模式约 10x 实时速度

elapsed = time.time() - start_time
pct = int(50 + 29 * (1 - math.exp(-3 * elapsed / estimated_sec)))
pct = min(pct, 79)

不同耗时下的进度表现（假设预估 128 秒）：

实际耗时	进度	说明
30s	~57%	快速推进
64s	~66%	中速
128s (预估)	~73%	接近预估时间
256s (2x)	~77%	超时但仍在动
512s (4x)	~79%	极慢但不卡死

编写新 Executor 的进度规范

# ✅ 正确：基于实际步骤
update_progress(task_id, 10)   # 解析完成
update_progress(task_id, 25)   # 下载完成
update_progress(task_id, 50)   # 预处理完成
# ... 黑盒处理中 ...
update_progress(task_id, 80)   # 处理完成

# ❌ 错误：凭空填数字
update_progress(task_id, 55)   # 没有任何依据
update_progress(task_id, 73)   # 纯粹按时间线性填

# ✅ 正确：黑盒处理期间用有依据的估算
if tool_provides_progress:
    # 解析工具输出的真实进度
    pct = parse_tool_progress(tool_stderr)
else:
    # 用时间渐近曲线估算
    pct = int(base + range * (1 - math.exp(-3 * elapsed / estimated)))

取消传播与优雅停机

取消传播（v4.1）

当 Caller 或管理面板取消一个正在运行的任务时：

Caller POST /task/cancel/{id}
  → LTAB 标记任务为 canceled
  → Executor 下一次调 update_progress 时收到 HTTP 400:
    {"code": 423102, "msg": "Task not running: canceled"}
  → update_progress 抛出 TaskCanceled 异常
  → process_task 捕获，kill CrispASR 子进程，清理临时文件
  → 不调用 report_fail（LTAB 已标记终态）

实现要点：

• TaskCanceled 异常类：标识任务被取消
• update_progress()：检测 423102 响应，抛出 TaskCanceled
• CrispASR 子进程：使用 Popen + cancel_event，支持中途 kill
• 进度估算线程：检测到取消后通过 cancel_event 通知主线程
• 取消延迟：最多 5 秒（progress 更新间隔）

编写新 Executor 时必须实现取消传播：

class TaskCanceled(Exception):
    pass

def update_progress(task_id, progress):
    r = requests.post(f"{LTAB}/executor/progress/{task_id}", ...)
    if r.status_code == 400 and r.json().get("code") == 423102:
        raise TaskCanceled(r.json().get("msg", ""))

def process_task(task_data):
    try:
        # ... 处理逻辑 ...
    except TaskCanceled:
        logger.info("Task canceled by LTAB, aborting")
        # 不要 report_fail
    except Exception as e:
        report_fail(task_id, str(e))

优雅停机（Drain，v4.1）

LTAB POST /executor/drain {"drain": true}
  → Executor 进入 drain 模式
  → /executor/next 不再返回新任务
  → 当前任务继续完成
  → 完成后可安全停机

典型流程（通过 Manager）：

1. Manager 发 POST /api/manager/action {"action":"stop","target":"crispasr-executor.service"}
2. 先调 POST /executor/drain {"drain": true}（让 Executor 不再接新任务）
3. 等当前任务完成
4. 再 systemctl stop crispasr-executor

支持的输入源

source 类型	说明	示例
bilibili	B站视频链接	{"source":"bilibili","url":"https://www.bilibili.com/video/BVxxx"}
url	直链音频/视频	{"source":"url","url":"https://example.com/audio.mp3"}
s3	MinIO 对象 key	{"source":"s3","key":"submits/abc123.mp4"}

依赖

系统依赖

• Python 3.10+
• ffmpeg / ffprobe
• yt-dlp（B站下载）
• CrispASR 二进制（编译时需含 CUDA 支持）

Python 依赖

boto3
requests

模型

• Qwen3 ASR 1.7B Q4_K: /root/.cache/crispasr/qwen3-asr-1.7b-q4_k.gguf (1.3GB)

部署

1. 安装 CrispASR

cd /opt/CrispASR
cmake -B build -DGGML_CUDA=ON
cmake --build build -j$(nproc)
# 二进制: /opt/CrispASR/build/bin/crispasr

2. 下载模型

# 模型会在首次运行时自动下载，或手动:
# /opt/CrispASR/build/bin/crispasr --backend qwen3 -m auto -f test.wav

3. 安装 Python 依赖

python3 -m venv /opt/executor-crispasr/venv
/opt/executor-crispasr/venv/bin/pip install boto3 requests

4. 配置 config.py

LTAB_BASE_URL = "http://ltab.cloverdo.cn:1111"
LTAB_API_KEY = "layla_xxxx"  # 从 LTAB 管理面板获取
TASK_TYPE = "crispasr"
CRISPASR_BIN = "/opt/CrispASR/build/bin/crispasr"
CRISPASR_MODEL = "/root/.cache/crispasr/qwen3-asr-1.7b-q4_k.gguf"
S3_ENDPOINT = "http://10.10.10.100:9000"
S3_ACCESS_KEY = "your_key"
S3_SECRET_KEY = "your_secret"
S3_BUCKET = "transfer-station"

5. 注册到 LTAB

在 LTAB 管理面板的「接入管理」页面新增接入：

• 名称: CrispASR
• 任务类型: crispasr
• 有效期: 按需设置

6. systemd 服务

# /etc/systemd/system/crispasr-executor.service
[Unit]
Description=CrispASR Executor for LTAB
After=network.target

[Service]
Type=simple
ExecStart=/opt/executor-crispasr/venv/bin/python3 /opt/executor-crispasr/executor.py
WorkingDirectory=/opt/executor-crispasr
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

systemctl enable --now crispasr-executor

编写新 Executor 的指南

如果你要基于此模板编写新的 Executor，需要修改以下部分：

必须修改

1. config.py — 修改 TASK_TYPE、二进制路径、模型路径
2. executor.py 的 process_task() — 实现你的处理逻辑
3. executor.py 的 INPUT_SCHEMA — 定义你的输入格式
4. config.py 的 S3 配置 — MinIO 连接信息

处理逻辑模板

def process_task(task_data: dict):
    task_id = task_data["task_id"]
    source_url = task_data["source_url"]  # LTAB 传来的输入
    source_name = task_data.get("source_name", "")

    # 1. 解析输入
    input_data = json.loads(source_url)

    # 2. 下载/获取源文件
    update_progress(task_id, 25)
    source_file = download(input_data)

    # 3. 处理
    update_progress(task_id, 50)
    result = your_process_function(source_file)

    # 4. 上传结果
    update_progress(task_id, 90)
    result_key = upload_result(result, task_id)

    # 5. 回报成功
    report_result(task_id, result_key, "output.md")

LTAB 通信函数（直接复用）

def heartbeat_loop():
    """每30秒心跳"""
    requests.post(f"{LTAB_BASE}/executor/heartbeat", headers=HEADERS, json={"ip": get_local_ip()})

def update_progress(task_id, progress):
    requests.post(f"{LTAB_BASE}/executor/progress/{task_id}", params={"progress": progress}, headers=HEADERS)

def report_result(task_id, result_key, result_name):
    requests.post(f"{LTAB_BASE}/executor/result/{task_id}", params={"result_key": result_key, "result_name": result_name}, headers=HEADERS)

def report_fail(task_id, error_msg):
    requests.post(f"{LTAB_BASE}/executor/fail/{task_id}", params={"error_msg": error_msg}, headers=HEADERS)

性能参考

音频时长	GPU (4070 Ti Super)	CPU (8线程)
5 分钟	~30 秒	~3 分钟
21 分钟	~3 分钟	~15 分钟

项目结构

crispasr-executor/
├── executor.py    # 主程序（心跳、拉取、处理、回报）
├── config.py      # 配置文件（LTAB 地址、密钥、路径）
└── README.md      # 本文档

当前部署信息

项目	地址
本服务	10.10.10.50 (WSL2, RTX 4070 Ti Super)
LTAB	http://ltab.cloverdo.cn:1111
MinIO	http://10.10.10.100:9000
Forgejo	http://10.10.10.100:3333/Cloverdo/crispasr-executor