Arquitetura do Modelo e Design de Fusão Multimodal

O sistema Wan2.1 VACE (Video-Audio Creation and Editing) inrtoduz uma abordagem unificada para geração e manipulação de vídeo, combinando processamento visual e sonoro em uma arquitetura coesa. Seu núcleo é baseado em uma estrutura de transformador de difusão (DiT) com ramificações especializadas.

Estrutura Principal

O modelo emprega uma arquitetura de ramificação dupla: um tronco principal para processamento geral e um ramo VACE dedicado ao controle de edição. Isso permite a ingestão precisa de múltiplos inputs modais.

Mecanismo de Codificação de Recursos Multimodal

O sistema processa entradas diversas, como descrições textuais, quadros-fonte de vídeo, máscaras de edição e imagens de referência, através de codificadores especializados.

def codificar_entradas_vace(self, quadros, imagens_ref, mascaras=None, codificador_vae=None):
    if mascaras is None:
        recursos_latentes = codificador_vae.encode(quadros)
    else:
        # Isola regiões ativas e inativas com base na máscara
        mascaras_bin = [torch.where(m > 0.5, 1.0, 0.0) for m in mascaras]
        regioes_inativas = [q * (1 - m) for q, m in zip(quadros, mascaras_bin)]
        regioes_ativas = [q * m for q, m in zip(quadros, mascaras_bin)]
        recursos_inativos = codificador_vae.encode(regioes_inativas)
        recursos_ativos = codificador_vae.encode(regioes_ativas)
        recursos_latentes = [torch.cat((ri, ra), dim=0) for ri, ra in zip(recursos_inativos, recursos_ativos)]

    recursos_fusao = []
    for recurso_latente, refs in zip(recursos_latentes, imagens_ref):
        if refs is not None:
            recursos_ref = codificador_vae.encode(refs)
            recurso_latente = torch.cat([*recursos_ref, recurso_latente], dim=1)
        recursos_fusao.append(recurso_latente)
    return recursos_fusao

Estratégia de Fusão por Atenção Hierárquica

Os sinais de controle de edição são injetados em diferentes profundidades da rede através de um mecanismo de atenção em camadas.

Mecanismo de Garantia de Consistência Espaço-Temporal

Para assegurar coerência visual ao longo do tempo, o modelo implementa codificação 3D com sensibilidade tmeporal e um sistema de injeção de dicas em camadas.

self.incorporacao_recursos_vace = nn.Conv3d(
    self.dim_entrada_vace,
    self.dim_interna,
    kernel_size=self.tamanho_bloco,
    stride=self.tamanho_bloco)

def processar_recursos_vace(self, x, contexto_vace, comprimento_seq, kwargs):
    recursos_vace = [self.incorporacao_recursos_vace(u.unsqueeze(0)) for u in contexto_vace]
    recursos_vace = [u.flatten(2).transpose(1, 2) for u in recursos_vace]
    recursos_vace = torch.cat([torch.cat([u, u.new_zeros(1, comprimento_seq - u.size(1), u.size(2))],
                  dim=1) for u in recursos_vace])

    dicas_hierarquicas = []
    for bloco in self.blocos_vace:
        recursos_vace, dica_skip = bloco(recursos_vace, **kwargs)
        dicas_hierarquicas.append(dica_skip)
    return dicas_hierarquicas

Arquitetura de Suporte a Múltiplas Resoluções

O modelo suporta saída em várias resoluções configurando o stride do VAE e o tamanho dos blocos de processamento.

Mecanismo de Coordenação entre Edição de Vídeo e Geração de Áudio

O VACE integra a edição visual com a síntese de áudio através de uma arquitetura de fusão multimodal e um mecanismo de alinhamento temporal, garantindo sincronia precisa.

Mecanismo de Alinhamento Temporal

Um módulo dedicado alinha os recursos de vídeo e áudio no domínio do tempo utilizando atenção cruzada e incorporações temporais.

class ModuloAlinhamentoTemporal(nn.Module):
    def __init__(self, dim=2048, num_cabecas=16):
        super().__init__()
        self.incorp_temp_video = nn.Linear(1, dim)
        self.incorp_temp_audio = nn.Linear(1, dim)
        self.atencao_cruzada = nn.MultiheadAttention(dim, num_cabecas)

    def forward(self, recursos_video, recursos_audio, carimbos_tempo):
        inc_video = self.incorp_temp_video(carimbos_tempo.unsqueeze(-1))
        inc_audio = self.incorp_temp_audio(carimbos_tempo.unsqueeze(-1))

        video_apr = recursos_video + inc_video
        audio_apr = recursos_audio + inc_audio

        video_alinhado, _ = self.atencao_cruzada(video_apr, audio_apr, audio_apr)
        audio_alinhado, _ = self.atencao_cruzada(audio_apr, video_apr, video_apr)

        return video_alinhado, audio_alinhado

Controle de Sincronização em Tempo Real

Um controlador gerencia buffers de vídeo e áudio, emparelhando quadros e amostras com base em seus carimbos de tempo para produzir saída sincronizada.

class ControladorSincronizacao:
    def __init__(self, taxa_quadros=30, taxa_amostragem=44100):
        self.taxa_quadros = taxa_quadros
        self.taxa_amostragem = taxa_amostragem
        self.buffer_video = []
        self.buffer_audio = []

    def sincronizar_buffers(self):
        dados_sincronizados = []
        idx_v, idx_a = A', 0

        while idx_v < len(self.buffer_video) and idx_a < len(self.buffer_audio):
            ts_v = self.buffer_video[idx_v][1]
            ts_a = self.buffer_audio[idx_a][1]

            if abs(ts_v - ts_a) < 0.033:
                dados_sincronizados.append((self.buffer_video[idx_v][0], self.buffer_audio[idx_a][0]))
                idx_v += 1
                idx_a += 1
            elif ts_v < ts_a:
                idx_v += 1
            else:
                idx_a += 1
        return dados_sincronizados

Técnicas de Processamento de Máscaras e Integração de Imagens de Referência

Para controle preciso da edição, o sistema VACE emprega processamento avançado de máscaras e integração de imagens-guia no espaço latente.

Processamento de Máscaras no Espaço Latente

Máscaras de segmentação são aplicadas para isolar regiões de interesse antes da codificação, permitindo edições localizadas.

def codificar_com_mascara(self, quadros, imagens_ref, mascaras, codificador_vae):
    mascaras_bin = [torch.where(m > 0.5, 1.0, 0.0) for m in mascaras]
    # Separa e codifica regiões inativas e ativas
    quadros_inativos = [q * (1 - m) for q, m in zip(quadros, mascaras_bin)]
    quadros_ativos = [q * m for q, m in zip(quadros, mascaras_bin)]
    latentes_inativos = codificador_vae.encode(quadros_inativos)
    latentes_ativos = codificador_vae.encode(quadros_ativos)
    # Concatena recursos das duas regiões
    latentes_combinados = [torch.cat((li, la), dim=0) for li, la in zip(latentes_inativos, latentes_ativos)]

    resultado_final = []
    for latente, refs in zip(latentes_combinados, imagens_ref):
        if refs is not None:
            latentes_ref = codificador_vae.encode(refs)
            latente = torch.cat([*latentes_ref, latente], dim=1)
        resultado_final.append(latente)
    return resultado_final

Redimensionamento e Alinhamento Espacial de Máscaras

As máscaras são redimensionadas para corresponder à resolução do espaço latente do VAE, utilizando interpolação, para garantir alinhamento correto com os recursos visuais codificados.

def ajustar_mascara_latente(self, mascaras, stride_vae):
    resultado = []
    for mascara in mascaras:
        c, prof, alt, larg = mascara.shape
        nova_profundidade = int((prof + 3) // stride_vae[0])
        nova_altura = 2 * (alt // (stride_vae[1] * 2))
        nova_largura = 2 * (larg // (stride_vae[2] * 2))

        mascara = mascara[0, :, :, :]
        mascara = mascara.view(prof, alt, stride_vae[1], larg, stride_vae[2])
        mascara = mascara.permute(2, 4, 0, 1, 3)
        mascara = mascara.reshape(stride_vae[1] * stride_vae[2], prof, alt, larg)

        mascara_redim = F.interpolate(
            mascara.unsqueeze(0),
            size=(nova_profundidade, nova_altura, nova_largura),
            mode='nearest-exact').squeeze(0)
        resultado.append(mascara_redim)
    return resultado