阿里万相Wan3.0全面开放：多模态输入一键生成30秒视频的技术深度解析

输入模态          编码器             融合层           解码器          输出
文本     ---->  Text Encoder  -->                    Video     ----> 视频
图像     ---->  Image Encoder -->  Cross Attention   Decoder
视频     ---->  Video Encoder -->  Fusion
音频     ---->  Multi modal   -->
网页/PDF ---->  Encoder       -->

# Wan3.0 时间一致性机制的概念实现（简化版）
import torch
import torch.nn as nn

class SpatioTemporalAttention(nn.Module):
    '''
    时空注意力机制
    同时处理空间维度的视觉特征和时间维度的运动特征
    '''
    def __init__(self, dim, heads=8, dim_head=64):
        super().__init__()
        self.heads = heads
        self.scale = dim_head ** -0.5

        # 空间注意力：关注同一帧内的不同区域
        self.spatial_attn = nn.MultiheadAttention(dim, heads, batch_first=True)

        # 时间注意力：关注不同帧之间的对应区域
        self.temporal_attn = nn.MultiheadAttention(dim, heads, batch_first=True)

        # 门控融合：动态决定空间和时间注意力的权重
        self.gate = nn.Sequential(
            nn.Linear(dim * 2, dim),
            nn.GELU(),
            nn.Linear(dim, 2),
            nn.Softmax(dim=-1)
        )

        self.norm = nn.LayerNorm(dim)

    def forward(self, x, frames, height, width):
        '''
        x: (B, C, T, H, W) - 批次、通道、时间、高度、宽度
        '''
        B, C, T, H, W = x.shape

        # 重排为 (B*T, H*W, C) 进行空间注意力
        spatial_x = x.permute(0, 2, 3, 4, 1).reshape(B * T, H * W, C)
        spatial_out, _ = self.spatial_attn(spatial_x, spatial_x, spatial_x)

        # 重排为 (B*H*W, T, C) 进行时间注意力
        temporal_x = x.permute(0, 3, 4, 2, 1).reshape(B * H * W, T, C)
        temporal_out, _ = self.temporal_attn(temporal_x, temporal_x, temporal_x)

        # 重排回统一形状
        spatial_out = spatial_out.reshape(B, T, H, W, C).permute(0, 4, 1, 2, 3)
        temporal_out = temporal_out.reshape(B, H, W, T, C).permute(0, 4, 3, 1, 2)

        # 门控融合
        gate_input = torch.cat([
            spatial_out.mean(dim=2),
            temporal_out.mean(dim=2)
        ], dim=1)
        gate_weights = self.gate(gate_input)

        w_spatial = gate_weights[:, :, 0:1].unsqueeze(-1).unsqueeze(-1)
        w_temporal = gate_weights[:, :, 1:2].unsqueeze(-1).unsqueeze(-1)

        output = self.norm(
            w_spatial * spatial_out + w_temporal * temporal_out + x
        )

        return output

# 模型初始化示例
model = SpatioTemporalAttention(dim=512, heads=8)
print(f"模型参数量: {sum(p.numel() for p in model.parameters()):,}")

输入提示词：
"一只金色的猫咪坐在窗台上，窗外是霓虹闪烁的东京夜景，
猫咪的绿色眼睛反射着城市的灯光，微风吹动窗帘，
镜头缓慢推近，呈现电影级景深效果"

输出：30秒高质量视频，包含连贯的动作和逼真的光影效果

# 使用Wan3.0 API进行图像到视频生成的示例代码
import requests
import base64
import time

class WanVideoGenerator:
    '''万相Wan3.0 视频生成API封装'''

    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://dashscope.aliyuncs.com/api/v1"

    def image_to_video(self, image_path, prompt, duration=30, resolution="1080p"):
        '''
        图像转视频
        :param image_path: 输入图像路径
        :param prompt: 动作描述提示词
        :param duration: 视频时长（秒）
        :param resolution: 分辨率
        '''
        # 读取并编码图像
        with open(image_path, "rb") as f:
            image_base64 = base64.b64encode(f.read()).decode()

        payload = {
            "model": "wan3.0",
            "input": {
                "image": image_base64,
                "prompt": prompt
            },
            "parameters": {
                "duration": duration,
                "resolution": resolution,
                "fps": 30,
                "motion_strength": 0.7,
                "style_transfer": False
            }
        }

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        # 提交生成任务
        resp = requests.post(
            f"{self.base_url}/services/aigc/video-generation/generation",
            json=payload,
            headers=headers
        )

        if resp.status_code == 200:
            task_id = resp.json().get("output", {}).get("task_id")
            return self._poll_task(task_id, headers)
        else:
            raise Exception(f"API请求失败: {resp.status_code}")

    def _poll_task(self, task_id, headers, max_wait=300):
        '''轮询任务状态'''
        start_time = time.time()
        while time.time() - start_time < max_wait:
            resp = requests.get(
                f"{self.base_url}/tasks/{task_id}",
                headers=headers
            )
            data = resp.json()
            status = data.get("output", {}).get("task_status")

            if status == "SUCCEEDED":
                video_url = data["output"]["video_url"]
                print(f"视频生成成功！下载地址: {video_url}")
                return video_url
            elif status == "FAILED":
                raise Exception(f"视频生成失败: {data}")
            else:
                print(f"生成中... 状态: {status}")
                time.sleep(5)

        raise TimeoutError("视频生成超时")

    def text_to_video(self, prompt, duration=30, resolution="1080p"):
        '''文本转视频'''
        payload = {
            "model": "wan3.0",
            "input": {"prompt": prompt},
            "parameters": {
                "duration": duration,
                "resolution": resolution,
                "fps": 30
            }
        }
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        resp = requests.post(
            f"{self.base_url}/services/aigc/video-generation/generation",
            json=payload, headers=headers
        )
        task_id = resp.json().get("output", {}).get("task_id")
        return self._poll_task(task_id, headers)


# 使用示例
# generator = WanVideoGenerator("your-api-key")
# video_url = generator.text_to_video(
#     prompt="未来城市中无人机穿梭于摩天大楼之间",
#     duration=30
# )

# 电商视频自动化生成流程
class EcommerceVideoPipeline:
    '''电商视频自动化生成流水线'''

    def __init__(self, wan_api):
        self.wan = wan_api

    def generate_product_video(self, product_info):
        '''
        根据产品信息自动生成营销视频
        '''
        # Step 1: 生成不同角度的产品展示
        prompts = [
            f"产品特写镜头，{product_info['name']}在柔和灯光下展示细节",
            f"产品使用场景，{product_info['name']}在{product_info['scene']}中使用",
            f"产品全景展示，{product_info['name']}360度旋转展示"
        ]

        videos = []
        for i, prompt in enumerate(prompts):
            print(f"正在生成第{i+1}个场景: {prompt[:30]}...")
            video_url = self.wan.text_to_video(
                prompt=prompt,
                duration=10,
                resolution="1080p"
            )
            videos.append(video_url)

        # Step 2: 添加产品文字信息
        # Step 3: 添加背景音乐
        # Step 4: 拼接视频片段

        return {
            "status": "success",
            "videos": videos,
            "total_duration": len(videos) * 10,
            "resolution": "1080p"
        }

# 使用示例
# pipeline = EcommerceVideoPipeline(generator)
# result = pipeline.generate_product_video({
#     "name": "智能手表",
#     "scene": "办公室",
#     "features": ["心率监测", "NFC支付", "GPS导航"]
# })