normalizingflow的概念

当我们讨论概率分布时，通常默认该分布能够被归一化。“正则分布”和“标准高斯分布”是两个经典的例子。但是对于很多概率分布，如“beta分布”，它们并不能被简单地归一化。此时，我们可以考虑采用normalizingflow。normalizingflow是一个可逆映射，它能够将一个简单的概率分布转换为我们感兴趣的概率分布，如高斯混合模型。这种方法不仅适用于生成模型，也适用于deepinference。

normalizingflow的基本原理

给定一个简单的概率分布z，普通的深度学习生成模型试图学习一个非线性函数f，把输入s转化为目标概率分布x。类似的，normalizingflow也定义了一个可逆函数f，将随机向量z转化为x，其中z是先验分布，x是要生成的分布。与其他深度生成模型类似，normalizingflow同样使用反向传播来学习f的参数。但与其他模型不同的是，我们需要设计一个可逆的函数f，这样我们就可以应对概率密度不能直接计算的情况。 一种简单的normalizingflow被称为可逆自回归流（invertible autoregressive flow，IAF），它针对神经网络进行了改进，使得变换是可逆的。在IAF中，f由多个单层的可逆自回归函数组合而成。在每一层中，我们定义一个可逆变换ϕ，将输入z映射为μ和σ。然后，我们将σ作为原始向量的元素来乘，然后加上μ。这种逐层自回归的结构可以应对概率密度不能直接计算的情况。

normalizingflow在生成模型中应用

normalizingflow作为一种流模型，可以被看作是一种更一般的生成模型。它非常适合生成从复杂分布中采样的高质量样本。通常，我们可以将生成模型的目标视为计算高维积分或求解高维函数的最大值。然而，这些问题在高维空间中是非常困难的。normalizingflow允许我们将这些问题转化为低维空间中的简单积分问题和函数优化问题，因此它是一个非常强大的生成模型类型。 常见的高维数据类型为图像，因此使用normalizingflow来生成图像是自然的选择。我们可以训练一个生成模型来生成高质量的图像，例如使用celebA、cifar、imagenet等数据集。具体来说，我们可以从一个简单的分布（如高斯分布）中采样，然后通过normalizingflow转换为目标分布（如人脸图像）。然后，我们可以使用生成模型从目标分布中进行采样。通过这种方式，我们可以生成高质量的人脸以及其他图像类型，从而为许多应用场景（如图像生成、数据增强和降噪）提供了新的解决方法。

normalizingflow在deepinference中应用

除了生成模型之外，normalizingflow还可用于深度推断。深度推断主要关注的是推理复杂概率分布的参数。这对于数值计算、分类、聚类和时间序列分析等应用非常重要。normalizingflow可以将通常难以处理的后验分布转换为其他分布。这种转换可以通过maximum likelihood training进行优化。 通常情况下，我们对变量的后验分布关注不高，而更关注分布中的某些重要值，如期望值、求和或积分。在这种情况下，我们可以计算每个变量的预期值，然后用推理结果来代替实际值。这样，我们就可以使用期望最大化（EM）、变分推理和其他推理方法来计算概率值。

normalizingflow的优缺点

normalizingflow作为一种生成模型，具有以下优点：

1. normalizingflow实现简单，没有复杂的训练过程，可以使用常规的反向传播算法进行优化；
2. normalizingflow可以对数据进行变换，使其更加高斯化，这可以提高一些流模型的训练效果；
3. normalizingflow可以在数据量较少的情况下生成高质量的训练样本。 当然，normalizingflow也存在一些缺点：
4. 由于其基本原理是嵌入低维空间，所以适用范围有限，很难在无监督学习或其他学习任务中实现更好的表现；
5. normalizingflow的计算复杂度很高，计算慢，因此需要大量的计算资源。

代码示例

import torch
import torch.nn.functional as F
from torch import nn
class NormalizingFlow(nn.Module):
    def __init__(self, dim, flow_length=16):
        super().__init__()
        self.transforms = nn.ModuleList()
        self.dims = []
        for i in range(flow_length):
            transform = NormalizingFlowTransform(dim)
            self.transforms.append(transform)
        self.dims = tuple(self.dims)
    def forward(self, x):
        log_det = torch.zeros(len(x))
        for transform in self.transforms:
            x, ld = transform(x)
            log_det += ld
        return x, log_det
class NormalizingFlowTransform(nn.Module):
    def __init__(self, dim):
        super().__init__()
        self.t1 = nn.Sequential(
            nn.Linear(dim, 2 * dim),
            nn.LeakyReLU(),
            nn.Linear(2 * dim, 2 * dim),
        )
        self.t2 = nn.Sequential(
            nn.Linear(2 * dim, 2 * dim),
            nn.LeakyReLU(),
            nn.Linear(2 * dim, dim),
        )
        self.t3 = nn.Sequential(
            nn.Linear(2 * dim, 2 * dim),
            nn.LeakyReLU(),
            nn.Linear(2 * dim, dim),
        )
        self.t4 = nn.Sequential(
            nn.Linear(2 * dim, 2 * dim),
            nn.LeakyReLU(),
            nn.Linear(2 * dim, dim),
        )
    def forward(self, x):
        px, log_det_x = self.t1(x).chunk(2, dim=-1)
        x = x + F.softplus(log_det_x) * px
        py, log_det_y = self.t2(x).chunk(2, dim=-1)
        y = x + F.softplus(log_det_y) * py
        pz, log_det_z = self.t3(y).chunk(2, dim=-1)
        z = y + F.softplus(log_det_z) * pz
        pw, log_det_w = self.t4(z).chunk(2, dim=-1)
        w = z + F.softplus(log_det_w) * pw
        log_det = log_det_x + log_det_y + log_det_z + log_det_w
        return w, log_det