摘要
在大规模 AI 漫剧本地生产场景中，串行调用 ComfyUI API 存在明显缺陷：任务阻塞、无法实时获取节点执行状态、任务崩溃丢失上下文、批量渲染时无法动态管控 GPU 资源。本文设计一套基于aiohttp+Redis 的异步调度框架，通过 ComfyUI WebSocket 长连接实时监听节点执行事件，实现任务优先级排队、失败重试、任务状态持久化；引入光流法做帧间运动一致性检测，在渲染阶段提前识别画面抖动与角色漂移。整套系统完全本地部署，不依赖第三方云端 API，适配 8G/12G 消费级显卡，适合长序列 AI 漫剧分镜自动化生产。附带异步客户端、任务队列、光流质量评估、异常自动重试完整代码。
资源，模型可以看这一篇文章：https://developer.aliyun.com/article/1756481?spm=a2c6h.12873639.article-detail.5.571a435eUakA13
1 系统架构概述
整体四层架构：
- 任务接入层：接收漫剧分镜任务，支持配置优先级、最大重试次数、分辨率、种子、参考图路径；
- 任务队列层：Redis 有序集合实现优先级队列，持久化任务状态（pending/running/success/failed）；
- ComfyUI 通信层：WebSocket 长连接监听工作流节点执行事件，HTTP API 提交 prompt；
- 质量校验层：光流计算帧间运动幅值，自动判定画面闪烁，不合格任务自动触发重渲染。
传统同步 API 方式只能轮询查询任务结果，本方案通过 WebSocket 推送事件，任务状态实时回调，可精准捕获节点 OOM、采样失败、VAE 解码异常等细粒度错误。
2 前置环境准备指令
# 安装异步网络、redis、图像处理、光流依赖 pip install aiohttp redis opencv-python numpy torch pynvml # Windows启动redis服务（本地缓存任务） redis-server.exe redis.windows.conf
3 ComfyUI WebSocket 异步客户端完整代码
import aiohttp import asyncio import json import uuid class ComfyAsyncClient: def __init__(self, host="127.0.0.1", port=8188): self.host = host self.port = port self.ws_url = f"ws://{host}:{port}/ws" self.http_url = f"http://{host}:{port}/prompt" self.client_id = str(uuid.uuid4()) self.ws = None async def connect(self): session = aiohttp.ClientSession() self.ws = await session.ws_connect(f"{self.ws_url}?clientId={self.client_id}") print("WebSocket长连接已建立，监听ComfyUI事件") return session async def submit_workflow(self, workflow_json): payload = { "prompt": workflow_json, "client_id": self.client_id } async with aiohttp.ClientSession() as session: async with session.post(self.http_url, json=payload) as resp: res = await resp.json() prompt_id = res["prompt_id"] return prompt_id async def listen_event(self, callback): """持续监听节点执行事件，执行完成触发回调函数""" async for msg in self.ws: if msg.type == aiohttp.WSMsgType.TEXT: data = json.loads(msg.data) msg_type = data.get("type") if msg_type == "executing": node_info = data["data"] await callback(node_info) # 全部节点执行完毕 if msg_type == "executing" and data["data"]["node"] is None: await callback({"status": "finished"}) break elif msg.type == aiohttp.WSMsgType.ERROR: print("WebSocket连接异常") break async def event_callback(info): print(f"节点事件回调: {info}") async def main(): comfy_client = ComfyAsyncClient() session = await comfy_client.connect() # 加载本地工作流JSON with open("comic_workflow.json", "r", encoding="utf-8") as f: wf = json.load(f) pid = await comfy_client.submit_workflow(wf) print(f"任务已提交，prompt_id={pid}") await comfy_client.listen_event(event_callback) await session.close() if __name__ == "__main__": asyncio.run(main())
4 Redis 优先级任务队列（漫剧批量任务管理）
import redis import time class ComicTaskQueue: def __init__(self): self.r = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True) self.queue_key = "comic_drama_task_queue" self.task_info_key = "comic_task_detail" def add_task(self, task_data, priority=5): """priority值越小优先级越高，紧急分镜优先渲染""" task_id = str(uuid.uuid4()) self.r.hset(self.task_info_key, task_id, json.dumps({ "data": task_data, "status": "pending", "retry":0, "max_retry":2, "create_time":time.time() })) # zadd 有序集合，score=优先级 self.r.zadd(self.queue_key, {task_id: priority}) print(f"任务 {task_id} 入队成功，优先级：{priority}") return task_id def fetch_task(self): # 取出分数最小的一条任务 res = self.r.zpopmin(self.queue_key, count=1) if not res: return None task_id, priority = res[0] raw = self.r.hget(self.task_info_key, task_id) task = json.loads(raw) task["status"] = "running" self.r.hset(self.task_info_key, task_id, json.dumps(task)) return task_id, task def mark_status(self, task_id, status): task_raw = self.r.hget(self.task_info_key, task_id) task = json.loads(task_raw) task["status"] = status self.r.hset(self.task_info_key, task_id, json.dumps(task)) # 入队示例 if __name__ == "__main__": q = ComicTaskQueue() scene_task = { "prompt":"漫剧分镜，少年人物，室内对话，柔和漫风", "neg_prompt":"扭曲，变形，闪烁，多手指", "seed":987654, "ref_img":"./ref_char.png" } q.add_task(scene_task, priority=3)
5 光流法帧间时序一致性检测代码（质量自动校验）
漫剧最核心指标：相邻帧光流幅值，如果幅值突变，代表人物发生漂移、画面闪烁，自动判定渲染失败，触发重试。
import cv2 import numpy as np def calc_optical_flow_score(prev_img_path, curr_img_path): """ 返回运动幅值均值，数值越大代表帧间变化剧烈 漫剧推荐阈值：大于12判定为异常闪烁 """ prev = cv2.imread(prev_img_path) curr = cv2.imread(curr_img_path) gray_prev = cv2.cvtColor(prev, cv2.COLOR_BGR2GRAY) gray_curr = cv2.cvtColor(curr, cv2.COLOR_BGR2GRAY) flow = cv2.calcOpticalFlowFarneback( gray_prev, gray_curr, None, pyr_scale=0.5, levels=3, winsize=15, iterations=3, poly_n=5, poly_sigma=1.2, flags=0 ) mag, ang = cv2.cartToPolar(flow[...,0], flow[...,1]) mean_magnitude = np.mean(mag) return mean_magnitude def check_scene_frames(frame_list, threshold=12): bad_frames = [] for i in range(1, len(frame_list)): score = calc_optical_flow_score(frame_list[i-1], frame_list[i]) print(f"帧{i-1}→帧{i} 光流分数：{score:.2f}") if score > threshold: bad_frames.append(i) return bad_frames if __name__ == "__main__": frames = ["./out/frame_001.png","./out/frame_002.png","./out/frame_003.png"] bad = check_scene_frames(frames) if len(bad) > 0: print(f"检测到闪烁异常帧索引：{bad}，任务标记为重试") else: print("帧间时序校验通过")
6 GPU 资源保护 + 任务失败自动重试调度主逻辑
import pynvml pynvml.nvmlInit() gpu_handle = pynvml.nvmlDeviceGetHandleByIndex(0) def get_gpu_free_vram_gb(): mem_info = pynvml.nvmlDeviceGetMemoryInfo(gpu_handle) free_gb = mem_info.free / (1024**3) return free_gb async def task_worker(): queue = ComicTaskQueue() client = ComfyAsyncClient() session = await client.connect() while True: task_item = queue.fetch_task() if task_item is None: await asyncio.sleep(1) continue tid, task = task_item free_vram = get_gpu_free_vram_gb() if free_vram < 2.0: print("显存不足，任务重新放回队列，等待空闲") queue.r.zadd(queue.queue_key, {tid:5}) await asyncio.sleep(3) continue try: # 提交工作流，监听WebSocket事件 wf_data = task["data"] workflow = build_workflow_from_task(wf_data) prompt_id = await client.submit_workflow(workflow) await client.listen_event(event_callback) # 渲染完成后执行光流校验 frame_paths = get_output_frames(prompt_id) bad_list = check_scene_frames(frame_paths) if len(bad_list) > 0: raise Exception("帧时序校验失败，画面闪烁") queue.mark_status(tid, "success") print(f"任务{tid}渲染+校验全部通过") except Exception as e: print(f"任务异常 {e}") task["retry"] += 1 if task["retry"] < task["max_retry"]: # 重试，优先级降低 queue.r.hset(queue.task_info_key, tid, json.dumps(task)) queue.r.zadd(queue.queue_key, {tid: priority+1}) else: queue.mark_status(tid, "failed") await session.close() def build_workflow_from_task(task_data): with open("comic_workflow.json","r",encoding="utf-8") as f: wf = json.load(f) wf["3"]["inputs"]["seed"] = task_data["seed"] wf["6"]["inputs"]["text"] = task_data["prompt"] wf["7"]["inputs"]["text"] = task_data["neg_prompt"] wf["10"]["inputs"]["image"] = task_data["ref_img"] return wf def get_output_frames(prompt_id): # 读取ComfyUI输出目录图片路径 import os out_dir = "./comic_out" files = sorted([os.path.join(out_dir,f) for f in os.listdir(out_dir) if f.endswith(".png")]) return files
7 视频合成 + 批量音频对齐 FFmpeg 指令
# 图片序列生成视频，加入光流插值平滑 ffmpeg -framerate 20 -pattern_type glob -i './comic_out/*.png' \ -filter:v "minterpolate=fps=24:mi_mode=2" \ -c:v libx264 -crf 19 comic_scene_smooth.mp4 # 批量合并TTS配音音频 ffmpeg -i comic_scene_smooth.mp4 -i dub_audio.wav -c:v copy -c:a aac comic_final_scene.mp4 # 批量压缩漫剧成片，降低码率便于预览 ffmpeg -i comic_final_scene.mp4 -b:v 1200k comic_preview.mp4
8 工程优化要点与坑点解析
- WebSocket 断线重连：生产环境需要增加断线检测，断开时自动重建连接，避免长时间任务监听中断；
- 显存水位动态调度：设置显存阈值，当剩余显存低于阈值，停止消费型任务入队，防止 OOM；
- 光流参数调优：漫剧人物动作幅度小，阈值不能设置过高；动态镜头需要动态调整阈值，防止误判；
- 任务隔离：不同角色场景分开调度，避免 IPAdapter 特征缓存互相污染；
- 日志持久化：每个任务单独输出日志，记录每帧光流分数、显存曲线，方便回溯画面闪烁问题。
9 总结
本文构建的本地部署 AI 漫剧异步调度系统，解决传统同步 API 批量渲染中状态不可感知、任务丢失、画质无法自动化质检的痛点。基于 WebSocket 实时监听 ComfyUI 节点事件，Redis 实现任务排队与持久化，光流算法自动检测帧间闪烁，整套系统实现漫剧渲染无人值守批量生产。该方案可以直接部署在消费级显卡，适用于长篇 AI 漫剧分镜流水线，将人工校验画面闪烁的大量工作交给程序自动完成，显著提升本地 AI 漫剧生产效率。
