阿里云将在 Qwen Conference 泰国站演示 Qwen-Image 3.0、Wan 3.0 与 WonderClip 的视觉 AI 流水线。这个议程真正值得开发者关注的不是某一张图或某一段视频比上一个模型好看多少，而是“视觉 AI 流水线”这几个字背后的一系列工程问题：图像模型生成的是静止帧，视频模型要基于静止帧继续生成动态内容，剪辑工具又要拿一堆视频片段组织成带字幕、带节奏、带包装的成片。它们各自的输出格式、字段、参数、失败方式都不一样，如何把它们串成一条可追踪、可重跑、可干预的流水线，才是落到项目里真正花时间的部分。
这篇文章不讨论会议新闻本身，而是以这个技术组合为参照，落地一条最小可运行的视觉 AI 流水线。你会看到为什么要按阶段拆分模型调用，中间产物应该长什么样，配置如何与代码解耦，以及本地验证通过后，距离生产环境还差哪些东西。
1. 先搞清楚 Qwen-Image、Wan 和 WonderClip 在流水线里分别承担什么
1.1 单模型演示和流水线演示的差别
单模型演示通常是这样：用户输入一句提示词，模型返回一张图片或一段视频。模型内部处理过程可以看作一个黑盒，使用方只能看到输入和输出。
流水线演示则不一样。它把“一句话创意”拆成多个阶段：
- 根据创意规划分镜或镜头脚本。
- 根据分镜描述生成静态画面。
- 让静态画面运动起来，生成视频片段。
- 把多个视频片段按顺序、字幕、转场和背景声组织成成片。
- 人工审核成片，确认后才进入发布流程。
从标题给出的组合看，Qwen-Image、Wan 和 WonderClip 分别对应图像生成、视频生成和剪辑包装这一类能力。具体每个产品的官方接口和限制要以公开文档为准，但它们在流水线中的位置是可以确定的：图像输出是视频输入的素材之一，视频输出又是剪辑工具的输入素材。
这也是视觉 AI 流水线最核心的规律：前一个阶段的产物，必须能被后一个阶段消费。如果只是把三个模型分别调用三次，然后再手动把文件拖进剪辑软件，那还不能叫流水线，只能叫“三个独立工具”。
1.2 一条典型视觉 AI 流水线的五个阶段
为了后续代码不至于混乱，先用一张表把阶段、输入、输出和可能承担能力的模块列清楚。
| 阶段 | 输入 | 输出 | 常见能力模块 |
|---|---|---|---|
| 分镜规划 | 自然语言创意、文案 | 分镜列表、镜头描述 | 语言模型、规则脚本 |
| 图像生成或编辑 | 镜头描述、参考图 | 静态图片 | Qwen-Image 3.0 这类图像能力 |
| 图像生成视频 | 图片、运动描述 | 视频片段 | Wan 3.0 这类视频能力 |
| 剪辑包装 | 多个视频片段、字幕、风格配置 | 成片元数据或最终视频 | WonderClip 这类剪辑能力 |
| 人工审稿 | 成片、字幕、画面 | 审核结果或修改建议 | 业务系统、审核后台 |
需要注意，分镜规划阶段并不一定要用大模型。实际项目里可以用一个简单的语言模型调用，也可以用预先写好的分镜模板。重点是它对后续阶段提供结构化输入，而不是让用户一次性把所有要求塞进一句话。
1.3 数据依赖决定串行还是并行
视觉 AI 流水线不一定是“一条直线跑到底”。是否串行，取决于后一个阶段是否需要前一个阶段的产物。
如果只生成一个镜头，顺序就是：
镜头描述 -> 图像 -> 视频 -> 剪辑片段但如果要生成多个镜头，图像生成阶段可以按镜头数量并行调用。视频生成阶段也只需要等待对应图像完成。剪辑阶段则必须等所有视频片段都回来，因为剪辑要对时间线做整体排序。
因此在架构上不要把流水线写成一个阻塞的大函数。比较稳妥的做法是：
- 每个阶段有明确输入和输出。
- 每份中间产物有唯一标识。
- 状态可以写入数据库或日志，方便随时查看哪个镜头卡住了。
- 只有下游阶段需要上游多个结果时，才做汇合和等待。
广告片、电商短视频、产品宣传片这类场景最典型。它们不只有单镜头，还包含多个分镜：产品特写、场景环境、文字包装。每个分镜可以独立生成，但最终成片必须等所有分镜就绪。这也是后面代码示例选择“图像、视频、剪辑”三段式的原因。
2. 环境与项目结构：先把编排层和模型层分开
2.1 学习环境与生产环境的要求不一样
搭建示例前，先确定运行边界。能跑通和能上线是两个不同目标。
| 维度 | 学习环境 | 生产环境 |
|---|---|---|
| Python | 3.10 及以上 | 使用固定的运行时版本和镜像 |
| 依赖 | pyyaml、pillow | 增加日志库、消息队列 SDK、对象存储 SDK |
| 模型 | Mock Provider，不真实调用 | 通过服务地址调用模型，或部署自有模型服务 |
| 存储 | 本地 workdir 目录 | 对象存储，如阿里云 OSS |
| 密钥 | 不需要 | API Key、RAM 权限、密钥管理 |
| 运行方式 | 命令行同步执行 | 异步任务、队列、重试和监控 |
这里不建议一上来就追求“真实调用模型”。原因是图像生成、视频生成接口的请求体、图片回传方式、异步任务查询方式并不统一。先用 Mock Provider 把编排层验证清楚，再替换成真实模型，错误定位会容易很多。
2.2 项目目录规划
示例项目结构可以这样设计：
visual-ai-pipeline/ ├── configs/ │ └── demo.yaml ├── pipeline_demo/ │ ├── __init__.py │ ├── core.py │ ├── config.py │ └── main.py ├── workdir/ ├── requirements.txt └── README.md目录拆分的目的是让“配置”和“执行逻辑”分离。configs 只放链路定义和参数。workdir 放中间产物，学习环境里可以手动删除，生产环境应指向对象存储而不是本地磁盘。
2.3 配置驱动链路，而不是把链路写死在代码里
如果链路顺序写在代码里，每调整一次模型参数都要改代码。视觉 AI 项目通常还在快速变化阶段，推荐用 YAML 描述流水线。
# configs/demo.yaml storage: base_dir: ./workdir pipeline: name: visual-ai-demo steps: - type: image model_ref: mock-qwen-image params: width: 1024 height: 1024 - type: video model_ref: mock-wan params: duration: 5 frame_count: 25 - type: clip model_ref: mock-wonderclip params: style: tech subtitle: true上面 model_ref 在 Mock 环境中只是一个标签，并不代表已经真实接通了对应模型。将来接入真实服务时，可以把这个字段作为选择具体 Adapter 的依据。
对应配置加载代码并不复杂：
# pipeline_demo/config.py from pathlib import Path from typing import Any import yaml def load_config(config_path: str | Path) -> dict[str, Any]: path = Path(config_path) with path.open("r", encoding="utf-8") as f: return yaml.safe_load(f)这段代码只负责把 YAML 读成 Python 字典。链路是什么、每步传什么参数、产物放哪里，都由配置决定。业务代码不需要关心模型参数的每次微调。
3. 用最小代码实现一条可运行的视觉 AI 流水线
3.1 用统一数据结构承载中间产物
视觉 AI 流水线里，各阶段输入输出并不一样，但都需要几个通用字段：任务 ID、当前阶段、状态、提示词、产物路径、错误信息、扩展元数据。没有统一数据结构，阶段之间就只能依赖字典满天飞，既不好调试，也容易在字段命名上出错。
下面这个数据类可以作为基础：
# pipeline_demo/core.py import json import uuid from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Dict from PIL import Image @dataclass class Asset: asset_id: str prompt: str current_stage: str = "" status: str = "pending" file_path: Path | None = None error: str = "" meta: Dict[str, Any] = field(default_factory=dict)字段含义很明确：
- asset_id：整条流水线任务的唯一 ID。
- current_stage：当前处理到哪一步。
- status：pending、success、failed。
- file_path：当前步骤产物的本地路径。
- error：失败原因。
- meta：保存请求参数、模型名、产物类型等上下文。
音频、图片、视频、JSON 都先包成同一个 Asset，后续步骤只对字段做约定。例如视频阶段要求 asset.file_path 存在，且读取 asset.prompt 作为运动描述。
3.2 Provider 接口只约定“输入 Asset，输出 Asset”
真实模型的调用差异很大，但流水线只关心能否把中间产物推进到下一步。因此每个阶段的 Provider 只需要实现一个方法。
class Provider: def __init__(self, params: Dict[str, Any]): self.params = params def process(self, asset: Asset, workdir: Path) -> Asset: raise NotImplementedError初始化时接收该阶段的参数，例如图片宽高、视频时长、字幕风格。真正的业务逻辑写在 process 里。这样做的好处是流水线执行器不需要关心具体模型，它只负责按顺序调用 Provider，并保存产物。
3.3 三个 Mock Provider 模拟图像生成、视频生成和剪辑包装
Mock 阶段不真实调用 Qwen-Image 3.0、Wan 3.0 和 WonderClip，而是用最少代码模拟阶段之间的数据流转。
图像生成 Provider 使用 Pillow 生成一张纯色图片：
class ImageGenProvider(Provider): def process(self, asset: Asset, workdir: Path) -> Asset: width = self.params.get("width", 1024) height = self.params.get("height", 1024) out_path = workdir / f"{asset.asset_id}_image.png" Image.new("RGB", (width, height), (48, 54, 82)).save(out_path) asset.file_path = out_path asset.current_stage = "image" asset.meta["output_type"] = "image" return asset视频生成 Provider 检查上游图片是否存在，然后把视频请求参数写入 JSON。这一步对应真实项目里“调用视频生成 API”之前构造请求体的过程。
class VideoGenProvider(Provider): def process(self, asset: Asset, workdir: Path) -> Asset: if not asset.file_path or not asset.file_path.exists(): raise ValueError("video stage requires existing image artifact") payload = { "input_image": str(asset.file_path), "prompt": asset.prompt, "duration": self.params.get("duration", 5), "frame_count": self.params.get("frame_count", 25), } out_path = workdir / f"{asset.asset_id}_video_request.json" out_path.write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", ) asset.file_path = out_path asset.current_stage = "video" asset.meta["output_type"] = "video" asset.meta["request_payload"] = payload return asset剪辑包装 Provider 读取视频请求上下文，输出一个剪辑清单。
class ClipProvider(Provider): def process(self, asset: Asset, workdir: Path) -> Asset: if "request_payload" not in asset.meta: raise ValueError("clip stage expects video request metadata") video_request = asset.meta["request_payload"] manifest = { "asset_id": asset.asset_id, "source_video_request": str(asset.file_path), "source_image": video_request.get("input_image"), "prompt": asset.prompt, "style": self.params.get("style", "tech"), "subtitle": self.params.get("subtitle", True), } out_path = workdir / f"{asset.asset_id}_clip_manifest.json" out_path.write_text( json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8", ) asset.file_path = out_path asset.current_stage = "clip" asset.meta["output_type"] = "clip" return asset真实项目中，VideoGenProvider 的 payload 可能需要通过 HTTPS 提交给模型服务，然后等待异步任务回调。ClripProvider 也要上传多个视频片段到剪辑服务。这里把“发起请求、等待回调、保存产物”之间的差异收敛到 Provider 内部，流水线主逻辑不会被单个 API 的细节打断。
3.4 注册中心 + 执行器
Provider 注册中心本质是一个字典。新增一个云厂商或一种模型，只需要在字典里多注册一个实现。
PROVIDER_REGISTRY = { "image": ImageGenProvider, "video": VideoGenProvider, "clip": ClipProvider, }执行器读取配置，按顺序执行每个 step。
def run_pipeline(prompt: str, config: dict) -> Asset: base_dir = Path(config["storage"]["base_dir"]) run_dir = base_dir / datetime.now().strftime("%Y%m%d_%H%M%S") run_dir.mkdir(parents=True, exist_ok=True) asset = Asset( asset_id=uuid.uuid4().hex[:12], prompt=prompt, ) asset.meta["step_models"] = {} for idx, step_cfg in enumerate(config["pipeline"]["steps"], start=1): step_type = step_cfg["type"] if step_type not in PROVIDER_REGISTRY: asset.status = "failed" asset.error = f"unknown step type: {step_type}" return asset asset.meta["step_models"][step_type] = step_cfg.get("model_ref", "mock") provider = PROVIDER_REGISTRY[step_type](
