Worker 编排器深度解析：AI 漫剧生成的状态机引擎

版本：v1.1（含 Seedance 2.0 接入）日期：2026-08-26

一、Worker 是什么

Worker 是 AI 漫剧项目的后台生成引擎。它是一个独立运行的 Python 进程，与 FastAPI Web 服务分离，专门负责耗时的 AI 生成任务。

┌──────────────┐         ┌─────────┐         ┌──────────────┐
│  FastAPI     │  LPUSH  │  Redis  │  BRPOP  │  Worker      │
│  (Web 服务)  │ ──────► │  队列   │ ──────► │  (生成引擎)   │
│  端口 8201   │         │ task:queue│        │  python -m app.worker
└──────────────┘         └─────────┘         └──────┬───────┘
                                                   │
                                    ┌──────────────┼──────────────┐
                                    ▼              ▼              ▼
                               豆包 LLM       Seedream 图像   Seedance 视频
                              (生成剧本)      (生成分镜图)    (生成动态视频)
                                    │              │              │
                                    ▼              ▼              ▼
                                 PostgreSQL     MinIO          FFmpeg 拼接
                                (状态持久化)    (资源存储)      (最终视频)

为什么不直接在 API 请求里生成？

一句话生成漫剧的完整链路需要 10-20 分钟（Seedance 2.0 视频生成是瓶颈）。如果阻塞 HTTP 请求，用户体验极差，且连接容易超时。Worker 模式让 API 立即返回 task_id，用户通过轮询查看进度。

二、启动流程

入口文件

文件：backend/app/worker.py

python

import logging
from app.init_db import init_db
from app.services.orchestrator import process_queue

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")

if __name__ == "__main__":
    init_db()
    process_queue()

启动命令：

bash

cd backend
.venv/bin/python -m app.worker

启动时做了两件事：

init_db() — 创建数据库表（Base.metadata.create_all），并确保存在默认审核员账号 reviewer / reviewer123
process_queue() — 进入阻塞式消费循环

启动日志：

INFO:root:创建数据库表...
INFO:root:数据库初始化完成
INFO:app.services.orchestrator:编排 Worker 启动，等待任务...

三、消费循环：`process_queue`

文件：backend/app/services/orchestrator.py 第 226-246 行

python

def process_queue() -> None:
    logger.info("编排 Worker 启动，等待任务...")
    while True:
        try:
            result = redis_client.brpop("task:queue", timeout=5)
            if result:
                _, task_id = result
                task_id = int(task_id)
                logger.info("取到任务 %s，开始生成", task_id)
                run_task(task_id)
        except KeyboardInterrupt:
            logger.info("Worker 收到退出信号，停止")
            break
        except Exception as e:
            logger.exception("Worker 处理异常: %s", e)
            time.sleep(1)

逐行解析

代码	作用
`while True`	无限循环，Worker 是常驻进程
`redis_client.brpop("task:queue", timeout=5)`	阻塞式从 Redis 列表右端弹出一个任务 ID，最多等 5 秒
`timeout=5`	5 秒超时设计：即使没有任务，每 5 秒也会回到循环顶部，便于检查 `KeyboardInterrupt` 信号实现优雅退出
`int(task_id)`	Redis 返回的是字符串，转为整数
`run_task(task_id)`	执行单个任务的完整生成链路
`except KeyboardInterrupt`	用户 `Ctrl+C` 时优雅退出，不会丢任务
`except Exception`	任何异常（包括 run_task 内部未捕获的）都不会让 Worker 崩溃，等 1 秒继续
`time.sleep(1)`	防止异常风暴（比如 Redis 连接断了会疯狂重试）

Redis 队列模型

API 写入端                      Worker 消费端

LPUSH task:queue 6     ──►    [6, 5, 4, 3]
                              BRPOP task:queue  ──►  取出 3（FIFO 先进先出）

LPUSH（左端推入）+ BRPOP（右端弹出）= FIFO 先进先出队列
BRPOP 是阻塞命令，没有任务时 Worker 会挂起等待，不消耗 CPU
decode_responses=True 配置让 Redis 返回字符串而非字节

谁在推入任务

两处 API 会推入任务队列：

用户提交创意：POST /api/creative/submit
python
```
redis_client.lpush("task:queue", task.id)
```
1
用户重试失败任务：POST /api/task/{task_id}/retry
python
```
redis_client.lpush("task:queue", task.id)
```
1

四、单任务执行：`run_task`

文件：backend/app/services/orchestrator.py 第 57-71 行

python

def run_task(task_id: int) -> None:
    db: Session = SessionLocal()
    try:
        task = db.get(Task, task_id)
        if not task:
            logger.error("任务 %s 不存在", task_id)
            return

        _run_pipeline(db, task)
    except Exception as e:
        logger.exception("任务 %s 生成失败: %s", task_id, e)
        _mark_failed(db, task_id, str(e))
    finally:
        db.close()

设计要点

要点	说明
独立 DB 会话	每个任务创建自己的 `SessionLocal()`，任务结束后 `db.close()` 释放连接
异常兜底	`_run_pipeline` 内任何异常（包括网络超时、API 401、JSON 解析失败）都被 catch
失败标记	异常时调 `_mark_failed()` 把任务标记为 `FAILED`，记录失败阶段
资源释放	`finally` 确保数据库连接一定被释放，即使 `mark_failed` 本身也抛异常

失败处理函数

python

def _mark_failed(db: Session, task_id: int, reason: str) -> None:
    task = db.get(Task, task_id)
    if task:
        task.status = TaskStatus.FAILED.value     # "FAILED"
        task.fail_stage = task.stage              # 记录在哪个阶段失败
        db.commit()
    _set_progress(task_id, "FAILED", 100, eta_seconds=0)
    logger.error("任务 %s 标记失败: %s", task_id, reason)

fail_stage 字段记录失败发生在哪个阶段（SCRIPT / IMAGE / TTS / VIDEO）
Redis 进度也更新为 FAILED，前端轮询能立即感知
用户可在"我的作品"页面点击"重新生成"重试 FAILED 任务

五、核心流水线：`_run_pipeline` 逐阶段剖析

这是 Worker 最核心的函数，约 130 行代码，驱动整个漫剧生成链路。

状态机总览

QUEUED ──► SCRIPT ──► IMAGE ──► TTS ──► VIDEO ──► PENDING_REVIEW ──► APPROVED
  │           │         │         │         │              │
  └───────────┴─────────┴─────────┴─────────┴──────────────┴──► FAILED (任意阶段)

每个阶段做三件事：

更新 Task.status 和 Task.stage 到数据库
写 Redis 进度缓存
调用对应 AI 适配器执行生成

阶段 1：剧本生成（SCRIPT）

python

_update_stage(db, task, TaskStatus.SCRIPT, "SCRIPT")
_set_progress(task_id, "SCRIPT", 10)

script = llm_adapter.gen_script(task.idea, task.style, task.duration_bucket)

调用链路：

task.idea = "一只小猫在月球上种花，遇到了会唱歌的星星"
task.style = "治愈系动漫"
task.duration_bucket = "short_30"
         │
         ▼
llm_adapter.gen_script(idea, style, duration_bucket)
         │
         ▼  OpenAI SDK → 方舟 API
POST https://ark.cn-beijing.volces.com/api/v3/chat/completions
model = doubao-seed-2-1-pro-260628
         │
         ▼  返回 JSON
{
  "title": "月球上的星光花",
  "frames": [ {...}, {...}, {...} ]
}

入库操作：

python

task.title = script.get("title", "未命名漫剧")
task.story_json = script                    # 完整 JSON 存入 JSONB 字段

# 逐镜写入 storyboards 表
for idx, frame in enumerate(frames_data):
    sb = Storyboard(
        task_id=task_id,
        seq=idx,                           # 分镜序号 0, 1, 2
        scene_desc=frame.get("scene_desc", ""),
        dialogue=frame.get("dialogue", ""),
        narration=frame.get("narration", ""),
        duration_ms=frame.get("duration_ms", frame_duration),  # 默认 10000ms
        image_prompt=frame.get("image_prompt", ...),
    )
    db.add(sb)
db.commit()

进度更新：progress = 20%，total_frames = 分镜数

阶段 2：图片生成（IMAGE）

python

_update_stage(db, task, TaskStatus.IMAGE, "IMAGE")
_set_progress(task_id, "IMAGE", 30, 0, len(storyboards))

for i, sb in enumerate(storyboards):
    image_url = None
    fallback = False
    for attempt in range(2):
        try:
            raw_url = image_adapter.gen_frame(sb.image_prompt)
            image_bytes = video_composer._download_image(raw_url)
            obj_name = storage.upload_bytes(image_bytes, "png", "image/png")
            image_url = storage.get_object_url(obj_name)
            break
        except Exception as e:
            logger.warning("分镜 %s 第 %s 次失败: %s", sb.seq, attempt+1, e)
    if not image_url:
        # 兜底占位图
        fb_bytes = _gen_fallback_image(task.style)
        ...

逐镜串行调用，每镜的完整流程：

Storyboard.image_prompt (英文)
    │
    ▼  image_adapter.gen_frame()
POST https://ark.cn-beijing.volces.com/api/v3/images/generations
model = doubao-seedream-5-0-260128
size = 1920x1920
    │
    ▼  返回方舟 TOS 临时 URL（有效期约 24 小时）
https://ark-acg-cn-beijing.tos-cn-beijing.volces.com/doubao-seedream-5-0/xxx.jpeg
    │
    ▼  httpx GET 下载图片字节
image_bytes
    │
    ▼  storage.upload_bytes(bytes, "png", "image/png")
MinIO: ai-comic bucket → png/{uuid}.png
    │
    ▼  storage.get_object_url(obj_name)
http://localhost:9000/ai-comic/png/{uuid}.png
    │
    ▼  写入 storyboard.image_url

重试策略：每镜最多 2 次尝试。如果 Seedream 两次都失败，用 1x1 透明 PNG 兜底，标记 fallback = true，不阻塞后续阶段。

进度更新（假设 3 镜）：

完成镜数	progress	frames_ready
0	30%	0
1	40%	1
2	50%	2
3	60%	3

公式：progress = 30 + (i+1) / total * 30

阶段 3：配音合成（TTS）

python

_update_stage(db, task, TaskStatus.TTS, "TTS")
_set_progress(task_id, "TTS", 60, total, total)

for i, sb in enumerate(storyboards):
    text = sb.dialogue.strip() if sb.dialogue else sb.narration
    if text:
        try:
            audio_bytes = tts_adapter.synth(text)
            obj_name = storage.upload_bytes(audio_bytes, "mp3", "audio/mpeg")
            sb.audio_url = storage.get_object_url(obj_name)
        except Exception as e:
            sb.audio_url = None
    db.commit()

文本优先级：有台词用台词，无台词用旁白，两者都空则跳过。

TTS 适配器内部：

text = "小糖：「今天一定要把这种子种下去！」"
    │
    ▼  tts_adapter.synth(text)
POST https://openspeech.bytedance.com/api/v1/tts
鉴权：Bearer;{ACCESS_TOKEN}  ──失败──►  X-Api-Key HMAC-SHA256 签名
    │
    ▼  返回 base64 MP3
    │
    ▼  base64 解码 → bytes
    │
    ▼  MinIO: mp3/{uuid}.mp3

异常兜底：TTS 失败时返回 b""（空字节），storyboard.audio_url = None，不阻塞。后续视频阶段用静音兜底。

当前账号 TTS 未开通（403/401），此阶段实际全部跳过。

进度更新：progress = 60% → 75%

阶段 4：视频生成（VIDEO）— Seedance 2.0

这是最复杂、最耗时的阶段。接入 Seedance 2.0 后，每个分镜都会调用一次视频大模型生成动态视频，然后 FFmpeg 拼接为完整 MP4。

4.1 逐镜调用 Seedance

python

for i, sb in enumerate(storyboards):
    # 构造提示词
    prompt = (
        f"{task.style}，漫剧第 {sb.seq + 1} 个分镜。"
        f"场景：{sb.scene_desc}。"
        f"画面要求：{sb.image_prompt}。"
        f"生成连续稳定的电影级动态镜头，保持角色、服装和场景一致。"
    )

    # 调用 Seedance（传入 base64 图片字节）
    video_source_url = video_adapter.generate_video(
        prompt=prompt,
        image_bytes=video_composer._download_image(sb.image_url),
        duration_seconds=max(2, min(12, round(sb.duration_ms / 1000))),
    )

    # 下载 Seedance 返回的临时视频，上传到 MinIO
    video_bytes = video_composer._download_image(video_source_url)
    video_object_name = storage.upload_bytes(video_bytes, "mp4", "video/mp4")

Seedance 适配器内部流程（video_adapter.py）：

Step 1: 把分镜图片转 base64 Data URI
   image_bytes → "data:image/png;base64,iVBORw0KGgo..."

   为什么用 base64 而不直接传 URL？
   → MinIO 运行在 localhost:9000，方舟服务无法访问本地地址
   → base64 内嵌到请求体，避免外部下载依赖

Step 2: 创建异步视频任务
   POST https://ark.cn-beijing.volces.com/api/v3/contents/generations/tasks
   {
     "model": "doubao-seedance-2-0-260128",
     "content": [
       {"type": "image_url", "image_url": {"url": "data:image/png;base64,..."}},
       {"type": "text", "text": "治愈系动漫，漫剧第1个分镜..."}
     ],
     "ratio": "16:9",
     "resolution": "1080p",
     "duration": 10,
     "generate_audio": false,
     "watermark": false
   }
   → 返回 {"id": "cgt-20260825174317-fx5ng", "status": "pending"}

Step 3: 轮询任务状态
   while time.monotonic() < deadline:  # deadline = 当前时间 + 900秒
       time.sleep(10)  # 每 10 秒查一次
       GET .../tasks/cgt-20260825174317-fx5ng
       → {"status": "running"}  → 继续等
       → {"status": "running"}  → 继续等
       ...（通常 3-5 分钟）
       → {"status": "succeeded", "content": {"video_url": "https://...mp4"}}
       → 返回 video_url

   超时 900 秒未完成 → 抛 VideoGenerationTimeout
   状态为 failed → 抛 VideoGenerationError

Step 4: 返回方舟 TOS 临时视频 URL

时间参数计算：

python

duration_seconds = max(2, min(12, round(sb.duration_ms / 1000)))

sb.duration_ms 通常为 10000（10 秒）
round(10000 / 1000) = 10
max(2, min(12, 10)) = 10（夹在 [2, 12] 之间）

4.2 FFmpeg 拼接

python

video_url, cover_url = video_composer.concat_videos(
    [frame.image_url for frame in frame_assets],
    cover_image_url=storyboards[0].image_url if storyboards else None,
)

拼接器内部：

3 个 MinIO 视频片段 URL
    │
    ▼  下载到临时目录
/tmp/tmpxxxxx/
├── seedance_0.mp4
├── seedance_1.mp4
├── seedance_2.mp4
├── videos.concat.txt       ← "file 'seedance_0.mp4'\nfile 'seedance_1.mp4'..."
└── output.mp4               ← FFmpeg 输出
    │
    ▼  FFmpeg concat + 转码
ffmpeg -y -f concat -safe 0 -i videos.concat.txt \
  -c:v libx264 -preset fast -pix_fmt yuv420p \
  -an output.mp4
    │
    ▼  读取 output.mp4 → upload_bytes → MinIO
http://localhost:9000/ai-comic/mp4/{uuid}.mp4
    │
    ▼  封面图 = 第一镜分镜图 → MinIO
http://localhost:9000/ai-comic/png/{uuid}.png

-an 参数说明：Seedance 当前 generate_audio=false，视频片段没有音频轨。如果不加 -an，FFmpeg 会尝试编码不存在的音频流导致失败（exit code 254）。

进度更新：progress = 80% → 95%

阶段 5：收尾入库

python

work = Work(
    task_id=task_id,
    video_url=video_url,        # MinIO 最终视频 URL
    cover_url=cover_url,        # MinIO 封面图 URL
    status=TaskStatus.PENDING_REVIEW,
)
db.add(work)

task.status = TaskStatus.PENDING_REVIEW
task.stage = "REVIEW"
db.commit()
_set_progress(task_id, "REVIEW", 100, total, total)

redis_client.lpush("review:queue", task_id)
logger.info("任务 %s 生成完成，进入审核队列", task_id)

做了什么：

在 works 表创建作品记录，关联最终视频和封面
tasks 表状态改为 PENDING_REVIEW，阶段改为 REVIEW
Redis 进度设为 100%
task_id 推入 review:queue，管理员审核面板会看到

六、进度缓存机制

Worker 通过 Redis Hash 写进度，API 通过 HGETALL 读进度。

写入端（Worker）

python

def _set_progress(task_id, stage, progress, frames_ready=0, total_frames=0, eta_seconds=None):
    redis_client.hset(f"task:{task_id}:progress", mapping={
        "stage": stage,
        "progress": progress,
        "frames_ready": frames_ready,
        "total_frames": total_frames,
        "eta_seconds": eta_seconds or 0,
    })

读取端（API）

python

# creative.py get_task_status()
prog = redis_client.hgetall(f"task:{task_id}:progress")
frames_ready = int(prog.get("frames_ready", 0))
total_frames = int(prog.get("total_frames", 0))

Redis 存储结构

Key:  task:6:progress
Type: Hash
Fields:
  stage         = "IMAGE"
  progress      = "45"
  frames_ready  = "1"
  total_frames  = "3"
  eta_seconds   = "0"

进度百分比分布

 0%──5%──10%──15%──20%────30%──────60%──75%──80%────95%──100%
 │   │    │    │    │       │       │    │    │       │     │
QUEUED  SCRIPT完成  IMAGE开始  IMAGE完成 TTS完成 VIDEO开始  VIDEO完成 REVIEW

为什么用 Redis 而不是直接查数据库

数据库存的是阶段级状态（SCRIPT / IMAGE / VIDEO），粒度粗
Redis 存的是镜级进度（第几张图已完成、总共几张），前端可展示"生成第 2/3 张分镜图"
Redis 读写性能远高于 PostgreSQL，适合高频轮询（前端每 3-5 秒查一次）

七、异常处理全景

分层异常策略

process_queue()
  ├── KeyboardInterrupt → 优雅退出
  ├── Exception → log + sleep(1) + 继续循环（Worker 不崩）
  │
  └── run_task(task_id)
        ├── Exception → _mark_failed() → log
        │
        └── _run_pipeline(db, task)
              ├── 阶段1 SCRIPT 失败 → 冒泡到 run_task → FAILED
              ├── 阶段2 IMAGE 单镜失败 → 重试 2 次 → 兜底图 → 继续
              ├── 阶段3 TTS 失败 → 返回空字节 → 继续
              ├── 阶段4 VIDEO Seedance 失败 → 冒泡 → FAILED
              └── 阶段4 FFmpeg 失败 → 冒泡 → FAILED

各阶段异常影响矩阵

阶段	异常类型	是否阻塞流程	处理方式
SCRIPT	LLM 401/500	是	冒泡 → FAILED
SCRIPT	JSON 解析失败	是	冒泡 → FAILED
IMAGE	Seedream 单次失败	否	重试 1 次
IMAGE	Seedream 2 次失败	否	1x1 透明 PNG 兜底
TTS	401/403	否	返回 `b""`，静音兜底
VIDEO	Seedance 创建 400	是	冒泡 → FAILED
VIDEO	Seedance 轮询超时	是	冒泡 → FAILED
VIDEO	Seedance 状态 failed	是	冒泡 → FAILED
VIDEO	FFmpeg 拼接失败	是	冒泡 → FAILED

八、数据库会话管理

连接池配置

python

# database.py
engine = create_engine(settings.database_url, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

pool_pre_ping=True：每次借出连接前先发 ping，避免使用已断开的连接
autoflush=False：不在每次查询前自动 flush，减少不必要的 SQL

Worker 中的会话生命周期

python

def run_task(task_id: int) -> None:
    db: Session = SessionLocal()       # ← 创建
    try:
        task = db.get(Task, task_id)
        _run_pipeline(db, task)         # ← 整个流水线共用这一个会话
    except Exception as e:
        _mark_failed(db, task_id, str(e))
    finally:
        db.close()                     # ← 释放

关键点：

一个任务 = 一个 DB 会话，贯穿所有阶段
每个阶段的 db.commit() 确保数据持久化到 PostgreSQL
finally 确保连接一定被释放回连接池
如果 Worker 进程崩溃（如 OOM），未 commit 的数据丢失，但已 commit 的数据安全

九、任务重试机制

API 端

python

# creative.py retry_task()
if task.status != TaskStatus.FAILED.value:
    raise HTTPException(409, "仅失败任务可重试")

task.status = TaskStatus.QUEUED
task.stage = "SCRIPT"
task.fail_stage = None
task.retry_count += 1
db.commit()

redis_client.lpush("task:queue", task.id)   # 重新入队

重试不扣配额（quota 不变），但 retry_count 递增。

Worker 端处理

Worker 不关心任务是首次还是重试，统一走 _run_pipeline。重试时 LLM 会重新生成剧本，图片和视频都会重新生成。

已知问题

重试时旧的 Storyboard 行不会被物理删除，导致重复分镜：

首次运行: storyboards = [{seq:0}, {seq:1}, {seq:2}]
重试运行: storyboards = [{seq:0}, {seq:1}, {seq:2}, {seq:0}, {seq:1}, {seq:2}]
                                              ↑ 新插入的行

Worker 遍历时会处理 6 镜而非 3 镜，导致 Seedance 调用翻倍。

建议修复：重试时先 DELETE FROM storyboards WHERE task_id = ?，或在表上加 UNIQUE(task_id, seq) 约束。

十、Redis 的三个用途

Worker 和 API 共享 Redis，用于三个独立功能：

1. 任务队列

Key:  task:queue
Type: List
方向: LPUSH（API 写） → BRPOP（Worker 读）
内容: task_id 字符串
生命周期: 消费即删除

2. 进度缓存

Key:  task:{task_id}:progress
Type: Hash
方向: HSET（Worker 写） → HGETALL（API 读）
内容: stage / progress / frames_ready / total_frames / eta_seconds
生命周期: 任务完成后不主动清除，下次任务覆盖更新

3. 审核队列

Key:  review:queue
Type: List
方向: LPUSH（Worker 写） → LPOP（管理员 API 读）
内容: task_id 字符串
生命周期: 审核完成后从队列移除

4. 每日配额

Key:  quota:{user_id}:{date}
Type: String (integer)
方向: INCR（API 写） → GET（API 读）
TTL:  86400 秒（自动过期）

十一、典型运行日志时间线

以下是一个 3 镜 short_30 任务的完整日志：

[00:00] INFO: 取到任务 6，开始生成
[00:00] INFO: POST chat/completions → 200 OK (剧本生成)
[00:03] INFO: SCRIPT 完成，3 个分镜入库
[00:03] INFO: POST images/generations → 200 OK (第1镜)
[00:05] INFO: GET TOS 图片下载 → 200 OK
[00:05] INFO: MinIO 上传 png/{uuid1}.png
[00:08] INFO: POST images/generations → 200 OK (第2镜)
[00:10] INFO: GET TOS 图片下载 → 200 OK
[00:10] INFO: MinIO 上传 png/{uuid2}.png
[00:13] INFO: POST images/generations → 200 OK (第3镜)
[00:15] INFO: GET TOS 图片下载 → 200 OK
[00:15] INFO: MinIO 上传 png/{uuid3}.png
[00:15] INFO: TTS 合成 → 403 → 401（失败，静音兜底）× 3
[00:16] INFO: GET localhost:9000/png/{uuid1}.png (下载首帧图)
[00:16] INFO: POST contents/generations/tasks → 200 OK (Seedance 创建第1镜)
[00:16] INFO: GET .../tasks/cgt-xxx → 200 OK (轮询: pending)
[00:26] INFO: GET .../tasks/cgt-xxx → 200 OK (轮询: pending)
[00:36] INFO: GET .../tasks/cgt-xxx → 200 OK (轮询: pending)
[04:30] INFO: GET .../tasks/cgt-xxx → 200 OK (轮询: succeeded!)
[04:30] INFO: GET TOS 视频下载 → 200 OK (下载第1镜视频)
[04:30] INFO: MinIO 上传 mp4/{uuid1}.mp4
[04:31] INFO: POST contents/generations/tasks → 200 OK (Seedance 创建第2镜)
[08:30] INFO: 第2镜成功，下载上传
[08:31] INFO: POST contents/generations/tasks → 200 OK (Seedance 创建第3镜)
[12:30] INFO: 第3镜成功，下载上传
[12:31] INFO: FFmpeg concat → output.mp4 (拼接完成)
[12:31] INFO: MinIO 上传 mp4/{final}.mp4 + png/{cover}.png
[12:31] INFO: LPUSH review:queue 6
[12:31] INFO: 任务 6 生成完成，进入审核队列

总耗时约 12 分钟，其中 Seedance 占 ~12 分钟（3 镜 × ~4 分钟/镜），其余阶段总计约 1 分钟。

十二、当前架构的局限与优化方向

局限 1：串行 Seedance 调用

现状：3 镜串行调用，每镜等待 4 分钟，总计 12 分钟。

优化：并行创建 3 个 Seedance 任务，集中轮询：

python

# 伪代码
task_ids = [create_seedance_task(sb) for sb in storyboards]  # 并行创建
results = poll_all(task_ids)                                  # 集中轮询
video_urls = [extract_video_url(r) for r in results]

预期效果：总时长从 N × 4分钟 降到 max(4分钟, N × 10秒创建间隔)，即约 5 分钟。

局限 2：单 Worker 进程

现状：一个 Worker 进程串行消费队列，同时只能处理一个任务。

优化：启动多个 Worker 进程（Docker 多副本 / Supervisor 多进程），Redis 队列天然支持多消费者。

局限 3：重试幂等问题

现状：重试时旧 Storyboard 不删除，导致重复分镜。

优化：

python

# 方案 A：重试时物理删除旧分镜
db.execute(delete(Storyboard).where(Storyboard.task_id == task_id))

# 方案 B：数据库约束
UniqueConstraint("task_id", "seq", name="uq_task_seq")

局限 4：无任务超时全局保护

现状：如果 Seedance 轮询卡住（比如方舟服务故障），Worker 会阻塞 900 秒。

优化：加全局任务超时，超过 30 分钟直接标记 FAILED。

局限 5：进程崩溃后任务丢失

现状：Worker 从 BRPOP 取出任务后如果进程崩溃，任务已从队列消失但未执行。

优化：使用 Redis RPOPLPUSH 或 Stream 消费组，实现"处理中"队列，崩溃后可恢复。

十三、核心文件速查

功能	文件	关键函数
Worker 入口	`app/worker.py`	`main` → `init_db()` + `process_queue()`
消费循环	`app/services/orchestrator.py`	`process_queue()`
单任务执行	`app/services/orchestrator.py`	`run_task(task_id)`
核心流水线	`app/services/orchestrator.py`	`_run_pipeline(db, task)`
进度写入	`app/services/orchestrator.py`	`_set_progress(task_id, ...)`
失败标记	`app/services/orchestrator.py`	`_mark_failed(db, task_id, reason)`
状态更新	`app/services/orchestrator.py`	`_update_stage(db, task, status, stage)`
Seedance 适配器	`app/adapters/video_adapter.py`	`generate_video(prompt, image_bytes, ...)`
FFmpeg 拼接器	`app/services/video_composer.py`	`concat_videos(video_urls, cover_url)`
Redis 客户端	`app/redis_client.py`	`redis_client`
DB 会话工厂	`app/database.py`	`SessionLocal`
数据模型	`app/models.py`	`Task` / `Storyboard` / `Work` / `TaskStatus`
配置	`app/config.py`	`Settings`

Worker 编排器深度解析：AI 漫剧生成的状态机引擎 ​

一、Worker 是什么 ​

二、启动流程 ​

入口文件 ​

三、消费循环：process_queue ​

逐行解析 ​

Redis 队列模型 ​

谁在推入任务 ​

四、单任务执行：run_task ​

设计要点 ​

失败处理函数 ​

五、核心流水线：_run_pipeline 逐阶段剖析 ​

状态机总览 ​

阶段 1：剧本生成（SCRIPT） ​

阶段 2：图片生成（IMAGE） ​

阶段 3：配音合成（TTS） ​

阶段 4：视频生成（VIDEO）— Seedance 2.0 ​

4.1 逐镜调用 Seedance ​

4.2 FFmpeg 拼接 ​

阶段 5：收尾入库 ​

六、进度缓存机制 ​

写入端（Worker） ​

读取端（API） ​

Redis 存储结构 ​

进度百分比分布 ​

为什么用 Redis 而不是直接查数据库 ​

七、异常处理全景 ​

分层异常策略 ​

各阶段异常影响矩阵 ​

八、数据库会话管理 ​

连接池配置 ​

Worker 中的会话生命周期 ​

九、任务重试机制 ​

API 端 ​

Worker 端处理 ​

已知问题 ​

十、Redis 的三个用途 ​

1. 任务队列 ​

2. 进度缓存 ​

3. 审核队列 ​

4. 每日配额 ​

十一、典型运行日志时间线 ​

十二、当前架构的局限与优化方向 ​

局限 1：串行 Seedance 调用 ​

局限 2：单 Worker 进程 ​

局限 3：重试幂等问题 ​

局限 4：无任务超时全局保护 ​

局限 5：进程崩溃后任务丢失 ​

十三、核心文件速查 ​

Worker 编排器深度解析：AI 漫剧生成的状态机引擎

一、Worker 是什么

二、启动流程

入口文件

三、消费循环：`process_queue`

逐行解析

Redis 队列模型

谁在推入任务

四、单任务执行：`run_task`

设计要点

失败处理函数

五、核心流水线：`_run_pipeline` 逐阶段剖析

状态机总览

阶段 1：剧本生成（SCRIPT）

阶段 2：图片生成（IMAGE）

阶段 3：配音合成（TTS）

阶段 4：视频生成（VIDEO）— Seedance 2.0

4.1 逐镜调用 Seedance

4.2 FFmpeg 拼接

阶段 5：收尾入库

六、进度缓存机制

写入端（Worker）

读取端（API）

Redis 存储结构

进度百分比分布

为什么用 Redis 而不是直接查数据库

七、异常处理全景

分层异常策略

各阶段异常影响矩阵

八、数据库会话管理

连接池配置

Worker 中的会话生命周期

九、任务重试机制

API 端

Worker 端处理

已知问题

十、Redis 的三个用途

1. 任务队列

2. 进度缓存

3. 审核队列

4. 每日配额

十一、典型运行日志时间线

十二、当前架构的局限与优化方向

局限 1：串行 Seedance 调用

局限 2：单 Worker 进程

局限 3：重试幂等问题

局限 4：无任务超时全局保护

局限 5：进程崩溃后任务丢失

十三、核心文件速查