在 Modal（modal.com）上为指导蒸馏的联合视频音频扩散 Transformer MiniMaxAI/Mini...

Prompt

在 Modal（modal.com）上为指导蒸馏的联合视频音频扩散 Transformer MiniMaxAI/MiniMaxH3（约 330 亿参数）构建一个完整的、单 GPU 的生产级服务系统。完整输出全部 18 个文件的代码，要求为纯可执行 Python，仅包含有效的函数式语句、import 声明、类与函数主体及配置 2. 配置 — config.py 定义一个冻结的 dataclass Settings，通过 load_settings() 实例化，并使用下列默认值从环境变量填充所有字段： APP_NAME: "videoaudiodiffusionserver" GPU_TYPE: "B200"（支持 "B300" 与 "H200"） MODEL_REPO_ID: "MiniMaxAI/MiniMaxH3" MODEL_BLOCKS_MODULE: "diffusers.modular_pipelines.minimax_h3" FL_SUBFOLDER: "FL2VA" REF_SUBFOLDER: "Ref2VA" TURBO_TRANSFORMER_REPO: "" TURBO_TRANSFORMER_SUBFOLDER: "transformer" TURBO_STEPS: 8 QUALITY_STEPS: 20 FAST_STEPS: 20 FAST_CACHE_THRESHOLD: 0.12 CACHE_WARMUP_STEPS: 3 CACHE_TAIL_STEPS: 2 QUALITY_REPAIR_TAU: 0.5 FAST_REPAIR_TAU: 1.0 TURBO_REPAIR_TAU: 1.0 DEFAULT_PROFILE: "fast" WEIGHT_PRECISION: "fp8" FP8_EXCLUDE_PATTERNS: "embed,norm,adaln,modulation,time,proj_out,final,head" FP8_KEEP_EDGE_BLOCKS: 2 ATTENTION_BACKEND: "fp8" ATTN_FP8_MIN_SEQ: 4096 COMPILE_MODE: "maxautotunenocudagraphs" VAE_TILING: 0 RESIDENCY_RESERVE_GIB: 40 LORA_GPU_CACHE_SIZE: 8 FUSED_LORAS: "" WARMUP_BUCKETS: "768p:16:9:124,480p:16:9:124" WARMUP_STEPS: 2 MIN_CONTAINERS: 1 SCALEDOWN_WINDOW: 600 GPU_MAX_INPUTS: 4 GPU_TARGET_INPUTS: 1 HOST_MEMORY_MIB: 262144 CPU_CORES: 16 FUNCTION_TIMEOUT: 3600 MODELS_VOLUME: "vamodels" OUTPUTS_VOLUME: "vaoutputs" JOBS_DICT: "vajobs" MODELS_DIR: "/models" OUTPUTS_DIR: "/outputs" FPS: 24 X264_PRESET: "veryfast" X264_CRF: 17 AUDIO_BITRATE: "192k" 导出 ENV_KEYS（包含全部 46 个配置键的列表）以及 settings_env()（返回当前所有有效配置的字符串字典）。 3. 分辨率与桶信息 — shapes.py FPS = 24 RESOLUTIONS: "480p": {"16:9": (832, 480), "9:16": (480, 832), "1:1": (640, 640), "4:3": (736, 544), "3:4": (544, 736)} "768p": {"16:9": (1344, 768), "9:16": (768, 1344), "1:1": (1024, 1024), "4:3": (1152, 864), "3:4": (864, 1152)} 帧网格：合法的片段帧数满足 17n + 5。单段原生范围为 n ∈ [7, 20]（124–345 帧）。 closest_aspect(w, h): 返回使 abs(math.log(w / h) math.log(rw / rh)) 最小的宽高比键。 frames_for_duration(seconds): 返回 max(1, round(seconds FPS))。 validate_num_frames(n): 校验 (n 5) % 17 == 0 且 (n 5) // 17 >= 1。 plan_segments(F): 若 F <= 345：返回 ([最小的 17n + 5 >= max(F, 124)], F)。 若 F > 345：计算 k = math.ceil((F 1) / 344)；返回 ([最小的 L = 17n + 5 (n <= 20) 且 1 + k (L 1) >= F] k, F)。 4. 数据模式 — schemas.py 定义 Pydantic v2 模型： MediaInput: source: str ReferenceInput: type: Literal["image", "video", "audio"], source: str, strength: float | None = None, start_seconds: float | None = None, end_seconds: float | None = None LoraSpec: source: str, weight_name: str | None = None, scale: float = 1.0 GenerationRequest: 字段：prompt: str, mode: Literal["t2v", "i2v", "fl2v", "ref2v"] | None = None, resolution: Literal["480p", "768p"] = "768p", aspect_ratio: Literal["16:9", "9:16", "1:1", "4:3", "3:4", "auto"] = "16:9", duration_seconds: float | None = None, num_frames: int | None = None, profile: Literal["quality", "fast", "turbo"] | None = None, num_inference_steps: int | None = None, seed: int | None = None, first_frame: MediaInput | None = None, last_frame: MediaInput | None = None, references: list[ReferenceInput] = [], loras: list[LoraSpec] = [], include_audio: bool = True 模型验证器： 当 duration_seconds 与 num_frames 均未提供时，默认 duration_seconds = 5.0 同时提供 duration_seconds 与 num_frames 时抛出 ValueError 使用 validate_num_frames 校验 num_frames 未指定 mode 时自动推断：有 references > "ref2v"；同时有 first 与 last > "fl2v"；仅有 first > "i2v"；否则 > "t2v" 强制依赖："i2v" 需 first_frame；"fl2v" 需 first_frame 与 last_frame；"ref2v" 至少 1 个 reference 且 last_frame 为空；"t2v" 不得提供 first_frame、last_frame、references aspect_ratio="auto" 需 first_frame 或 图像 reference ResolvedParams: width: int, height: int, num_frames: int, fps: int, duration_seconds: float, seed: int, profile: str, steps: int, mode: str, segments: list[int] JobStatus: id: str, status: Literal["queued", "running", "succeeded", "failed"], created_at: str, updated_at: str, result: ResolvedParams | None = None, video_url: str | None = None, error: str | None = None ErrorBody: error: dict[str, str]，包含 code 与 message 5. 媒体处理 — media_io.py 使用 FFmpeg 子进程、Pillow 与 NumPy 实现： fetch_bytes(source): 解析 base64 data: URI；对 HTTP(S) URL 采用 httpx.AsyncClient(follow_redirects=True, timeout=120) 抓取 load_image(data, width, height): 执行 EXIF 转置，转为 RGB；使用 Lanczos 插值缩放并居中裁剪至 (width, height) image_size(data): 返回执行 EXIF 转置后的 (w, h) load_video_frames(data, width, height, start, end): 向 FFmpeg 输入数据，按 24 fps 提取原始 RGB24 帧并重采样 load_audio(data, sample_rate, start, end): 从 FFmpeg 提取 float32le 立体声音频并重采样 to_uint8_frames(video): 接受 4D/5D torch 张量、NumPy 数组或 PIL 图像列表。若输入为 CUDA 张量，则先在 GPU 上映射动态范围到 [0, 255] 并转 torch.uint8，再转出到 CPU 固定内存 to_stereo(audio): 将音频张量/数组整理为 float32 NumPy [T, 2]；单声道复制，两声道以上取前两声道 stitch_segments(video_list, audio_list, sample_rate, fps): 拼接视频段（从索引 1 开始的段丢弃首帧），音频边界使用等功率交叉淡化，重叠样本数为 sample_rate // fps trim(frames, audio, F, sample_rate, fps): 视频截至 F 帧，音频截至 int(F sample_rate / fps) 样本，不足则零填充 encode_mp4(frames, audio_or_None, sample_rate, fps, out_path, preset, crf, audio_bitrate): 先写临时 float32 音频文件，再通过 stdin 流式向 FFmpeg 发送原始视频帧，使用 libx264（ pix_fmt yuv420p, movflags +faststart）及 aac 编码，并写入 out_path 6. FP8 指数转换注意力 — kernels/fp8_attention.py 以 Triton 实现 is_supported(q, k, v) > bool 及 fp8_attention(q, k, v, scale=None, key_mask=None, key_bias=None, repair_tau=1.0) > Tensor： 输入：q, k, v 形状 [B, H, S, D]（D ∈ {64, 128}，dtype bf16/fp16） is_supported：当张量在 CUDA 上、D 合法、dtype 合法、且 GPU 计算能力 ≥ 8.9 时返回 True 量化准备： 常数 c = (scale or (1.0 / math.sqrt(D))) 1.4426950408889634 Q Kernel：乘以 c，对每 (b, h, 128token block) 求绝对最大值 amax；scale_q = max(amax, 1e12) / 448.0；结果截断到 [448.0, 448.0]，转 torch.float8_e4m3fn，并保存 scale_q float32 [B, H, ceil(Sq / 128)] K 平滑 Kernel：k_mean = k.float().mean(dim=2, keepdim=True)；减去均值后按 block 量化生成 k8 与 scale_k V Kernel：按 (b, h, channel) 求绝对最大值，得到 scale_v [B, H, D]，并生成转置打包的 v8t [B, H, D, Sk] 主注意力 Triton Kernel： 网格：(cdiv(Sq, BLOCK_M), B H)，BLOCK_M = 128 自动调优：BLOCK_N ∈ [64, 128], num_stages ∈ [2, 3, 4], num_warps ∈ [4, 8]，以 (D, Sq // 512, Sk // 512, key_mask 是否存在, key_bias 是否存在, repair_tau ≥ 0.0) 为键 计算能力 ≥ 10.0 时，关键循环使用 tl.range(..., warp_specialize=True) 关键路径：nq = cdiv(Sq, BLOCK_M), nk = cdiv(Sk, BLOCK_N)，起始块 j0 = (pid_m nk) // nq，序列块索引 (j0 + i) % nk 分数：s = tl.dot(q8, tl.trans(k8)) scale_q[:, None] scale_k[j]；若存在 key_bias 则加 1.4426950408889634；被 mask 的位置设为 1e9 重新缩放：计算 rowmax；若 rowmax > m + 8.0 或 m == 1e9，则设 m_new = rowmax；否则保持 m_new = m；alpha = tl.math.exp2(m m_new)；更新运行项 ExpCast：cst = 56.0 8.0 m；bits_f = s 8.0 + cst[:, None]；小于 8.0 置 0，四舍五入到最近整数，截断到 [0, 126]，并按位转换为 tl.float8e4nv 得 p8 累积：l += tl.sum(p8.to(tl.float32), axis=1)；acc += tl.dot(p8, v8_tile) 残差修复：当 repair_tau ≥ 0.0 时启用；计算 p_exact, p_apx, delta；并累加到 acc_r 与 l 归一化：(acc + acc_r) scale_v[None, :] / tl.maximum(l[:, None], 1e12)；最终转回输入 dtype 7. FP8 线性层 — kernels/fp8_linear.py Triton Kernel quantize_rows(x2d)：按行量化为 x8 (torch.float8_e4m3fn) 与 x_scale float32 [M, 1]，x_scale = max(rowamax, 1e12) / 448.0 类 FP8Linear(nn.Module)： from_linear(linear): 保存 w8, w_scale, 可选 BF16 bias，并将原权重置为 None forward(x): 展平为 2D，量化输入，执行 torch._scaled_mm，恢复形状，并加上激活的 LoRA 分支 暴露 in_features, out_features convert_transformer_to_fp8(transformer, exclude_patterns, keep_edge_blocks) > int： 找到主要 block 列表（最大的 nn.ModuleList） 替换满足条件的 nn.Linear：in/out_features % 16 == 0，索引在 [keep_edge_blocks, total keep_edge_blocks)，且模块路径不匹配 exclude_patterns 调用 torch.cuda.empty_cache()，返回转换数量 8. 注意力路由器 — kernels/attention_router.py 模块状态：enabled: bool, repair_tau: float, min_seq: int, 线程局部的 denoiser 深度计数 install(): 猴子补丁 torch.nn.functional.scaled_dot_product_attention 及 torch._C._nn 内部引用 routed_sdpa(...): 当以下条件全部满足时路由到 fp8_attention： denoiser 深度 > 0 enabled 为 True dropout_p == 0.0 非因果 (is_causal 为 False) enable_gqa 为 False q, k, v 头维一致 is_supported(q, k, v) 为 True Sq >= min_seq mask 为 None 或 perkey（倒数第二维为 1） 其他情况回退到原生 SDPA。布尔 mask 映射为 key_mask [B, Sk]，浮点 mask 映射为 key_bias [B, Sk] 提供 denoiser_scope(transformer) 上下文管理器跟踪深度；configure(enabled, repair_tau, min_seq) 9. 步残差缓存 — runtime/step_cache.py 类 StepResidualCache： attach(transformer): 给 transformer.forward 计步，并包装各 block 的 forward begin(num_steps, threshold, warmup, tail): 重置状态；若 threshold <= 0.0 或 num_steps <= 8 则禁用缓存 Block 0：计算相对残差 rel 并累加到 acc_rel 跳过决策：在缓存启用、warmup <= step < num_steps tail、有缓存、且 acc_rel < threshold 时跳过 1 至 N2 号 block，直接返回输入；否则正常执行并重置 acc_rel 最后一个 block：跳过时返回 input + residual；执行时记录 residual end(): 释放缓存张量 10. LoRA 运行时 — runtime/lora.py resolve_lora(source, weight_name) > Path: 从 HTTP(S) 或 Hugging Face 下载 safetensors 到 MODELS_DIR/lora_cache/<sha256>.safetensors parse_lora(path, module_names) > dict: 解析权重张量，去前缀，处理 lora_unet_ 名称，将 A/B 或 down/up 矩阵配对，应用 alpha / rank（缺省 1.0）。若有未匹配键，抛出 ValueError（列举最多 10 条） 线性模块扩展：给 nn.Linear 和 FP8Linear 附加 adapters，在 forward 时计算并加和 LoraManager: 维护 GPU 侧 LRU 缓存，容量 LORA_GPU_CACHE_SIZE。提供 activate(transformer, specs) 与 deactivate(transformer) fuse_into_weights(transformer, specs): 将 LoRA 融合到 BF16 基线权重，然后再进行 FP8 转换 11. 模型权重管理 — runtime/weights.py ensure_weights(settings) > dict[str, Path]： 通过 snapshot_download 下载 MODEL_REPO_ID 到 MODELS_DIR/base（带 .complete 标记） 若 TURBO_TRANSFORMER_REPO 非空则下载到 MODELS_DIR/turbo 下载并解析 FUSED_LORAS 中指定的 LoRA 返回本地目录路径字典 commit_if_changed(volume): 提交 Modal 存储卷变更 12. 模块化 Pipeline 适配器 — runtime/pipeline_adapter.py 类 PipelineAdapter(pipe, blocks_module)： 通过 pipe.blocks.input_names（或子模块输入名）映射标准参数： prompt > "prompt" height > "height" width > "width" num_frames > "num_frames" steps > "num_inference_steps" generator > "generator" fps > "frame_rate" 或 "fps" references > "references" 或 "conditions" output_type > "output_type" build_references(first_frame, last_frame, references, num_frames): 利用 inspect.signature 生成 blocks_module 的 reference 实例；将 first_frame 锚定索引 0，last_frame 锚定 num_frames 1 run(...): 调用 pipeline，指定输出 ["videos", "audio"]，返回 (video, audio) 张量 sample_rate(): 从 pipeline 组件 (audio_vae 或 vocoder) 提取采样率 13. 模型生命周期与驻留 — runtime/model_manager.py 类 ModelManager(settings)： load()： 设置 TORCHINDUCTOR_CACHE_DIR, TRITON_CACHE_DIR 于 MODELS_DIR torch.backends.cuda.matmul.allow_tf32 = True 加载 FLpipeline（子文件夹 FL_SUBFOLDER，BF16），放入 GPU 加载 Refpipeline（子文件夹 REF_SUBFOLDER），与 FL 共享子模块 若有 Turbo transformer 则加载并共享 将 FUSED_LORAS 融合进所有 transformer 若 WEIGHT_PRECISION == "fp8" 则转换指定层为 FP8Linear 驻留管理：计算显存占用。当可用显存低于 RESIDENCY_RESERVE_GIB 时，将多余 transformer 参数迁移至 CPU 固定内存；提供 ensure_resident(key) 做 LRU 交换 配置注意力路由器，使用 denoiser_scope，后端设 "native" 为所有 transformer 附加 StepResidualCache 与 LoRA 注册表 当启用编译时，对 transformer block 进行 torch.compile（mode=COMPILE_MODE, dynamic=False, fullgraph=False），并设 torch._dynamo.config.cache_size_limit = 64 若 VAE_TILING 启用则开启 按 WARMUP_BUCKETS 撞库，执行 WARMUP_STEPS pipeline_for(profile, mode): 若 profile 为 turbo 返回 "turbo"；mode=="ref2v" 返回 "ref"；否则返回 "fl" 14. 请求执行引擎 — runtime/generator.py 类 Generator(manager, settings)： 持有单线程 GPU 执行器、2 线程编码执行器、以及 asyncio.Lock resolve(req) > ResolvedParams: 校验 profile 兼容性，决定步数，解析宽高比，初始化随机种子，并通过 plan_segments 计算片段计划 prepare_media(req, params): 通过 asyncio.gather 并发抓取解码所有输入，转换成目标分辨率与采样率 run(req) > (frames, audio, sample_rate, params)： GPU 锁外：并发媒体解码与 LoRA 解析 GPU 锁内：激活对应 transformer，配置路由器，激活 LoRA，遍历片段；后一片段用前一片段最后一帧作条件；将输出转为 uint8 视频与立体声音频；finally 中确保 LoRA 停用与缓存清理 GPU 锁外：拼接片段、截断到最终长度，并格式化音频 render_mp4(req, out_path): 调用 run，并把 encode_mp4 提交到编码执行器以并行化 15. 任务存储 — runtime/jobs.py 类 JobStore(settings)： 封装 modal.Dict.from_name(JOBS_DICT, create_if_missing=True) 异步方法：create(job_id, request_json), set_call(job_id, call_id), mark_running(job_id), mark_succeeded(job_id, params, video_path), mark_failed(job_id, message), get(job_id), delete(job_id) 所有时间戳均保存为 UTC ISO8601 16. REST API — api.py create_api(generator_cls, job_store, settings, outputs_volume) > FastAPI： GET /health：返回 {"status": "ok"} GET /v1/models：返回可用模式、支持的 profile、分辨率映射、帧率、帧网格规则、原生片段范围及默认 profile POST /v1/generations：接受 GenerationRequest，生成 UUID4，注册任务，触发 generator_cls().generate.spawn，记录 call ID，返回 202 与 JobStatus GET /v1/generations/{id}：返回 JobStatus；若有 call ID 则检查运行状态；成功后填充 video_url = f"/v1/generations/{id}/video" GET /v1/generations/{id}/video：重新加载 outputs_volume，返回 MP4 文件；若不存在返回 404；未完成返回 409 DELETE /v1/generations/{id}：删除视频、清除任务、提交卷，返回 204 POST /v1/generations/sync：调用 generator_cls().generate_bytes.remote 同步生成，返回 MP4 二进制并附加头 XGenerationSeed, XWidth, XHeight, XNumFrames, XSteps, XProfile 错误处理：Pydantic 校验错误与 ValueError > 400；ID 不存在 > 404；生成失败 > 500 17. Modal 应用 — app.py 实例化 app = modal.App(settings.APP_NAME) 配置 modal.Volume.from_name 用于 MODELS_VOLUME 与 OUTPUTS_VOLUME 定义 gpu_image： modal.Image.from_registry("nvidia/cuda:13.0.1develubuntu22.04", add_python="3.11") .apt_install("ffmpeg", "git", "libgl1", "libglib2.00") .pip_install("torch==2.9.", "torchaudio==2.9.", index_url="") .pip_install("triton>=3.5", "git+", "transformers", "accelerate", "safetensors", "huggingface_hub[hf_transfer]", "sentencepiece", "protobuf", "einops", "ftfy", "pillow", "numpy", "httpx", "fastapi[standard]", "pydantic>=2") .env({settings_env(), "HF_HOME": f"{MODELS_DIR}/hf", "HF_HUB_ENABLE_HF_TRANSFER": "1"}) .add_local_python_source("config", "shapes", "schemas", "media_io", "api", "kernels", "runtime") 定义 api_image: modal.Image.debian_slim(python_version="3.11")，安装 fastapi[standard], pydantic>=2, httpx，并挂载同样环境及源码 download_weights：运行于 gpu_image，挂载模型卷，执行 ensure_weights，并提交卷 类 VideoAudioGenerator： 装饰 @modal.cls(gpu=settings.GPU_TYPE, image=gpu_image, volumes={MODELS_DIR: models_volume, OUTPUTS_DIR: outputs_volume}, timeout=FUNCTION_TIMEOUT, scaledown_window=SCALEDOWN_WINDOW, min_containers=MIN_CONTAINERS, memory=HOST_MEMORY_MIB, cpu=CPU_CORES) 应用 @modal.concurrent(max_inputs=GPU_MAX_INPUTS, target_inputs=GPU_TARGET_INPUTS) @modal.enter() start(): 初始化 ModelManager 并加载；提交 models_volume；构建 Generator 与 JobStore @modal.method() generate(request_json, job_id): 将任务设为 running，调用 render_mp4 输出到 OUTPUTS_DIR/{job_id}.mp4，提交卷，标记成功或失败 @modal.method() generate_bytes(request_json): 渲染到临时文件并返回 (mp4_bytes, params_dict) web 函数：@app.function(image=api_image, volumes={OUTPUTS_DIR: outputs_volume})、@modal.concurrent(max_inputs=200)、@modal.asgi_app()，返回 create_api(VideoAudioGenerator, JobStore(settings), settings, outputs_volume) 本地入口 main(...)：CLI 接收生成参数，将本地图片编码为 data URI，调用 VideoAudioGenerator().generate_bytes.remote，保存 MP4 并打印参数 18. 输出交付格式 按文件名整洁地给出全部 18 个文件的完整代码，保持原样且不裁剪，保证立即可执行。