跳到内容

连续噪声变换

ContinuousExpNoiseTransform

基类：ABC

连续计划的基础类。

alpha = exp(- sigma)，其中 1 - alpha 控制掩码比例。

源代码在 bionemo/moco/schedules/noise/continuous_noise_transforms.py 中 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class ContinuousExpNoiseTransform(ABC):
    """A base class for continuous schedules.

    alpha = exp(- sigma) where 1 - alpha controls the masking fraction.
    """

    def __init__(self, direction: TimeDirection):
        """Initialize the DiscreteNoiseSchedule.

        Args:
            direction : TimeDirection, required this defines in which direction the scheduler was built
        """
        self.direction = string_to_enum(direction, TimeDirection)

    def calculate_sigma(
        self,
        t: Tensor,
        device: Union[str, torch.device] = "cpu",
        synchronize: Optional[TimeDirection] = None,
    ) -> Tensor:
        """Calculate the sigma for the given time steps.

        Args:
            t (Tensor): The input tensor representing the time steps, with values ranging from 0 to 1.
            device (Optional[str]): The device to place the schedule on. Defaults to "cpu".
            synchronize (optional[TimeDirection]): TimeDirection to synchronize the schedule with. If the schedule is defined with a different direction,
                this parameter allows to flip the direction to match the specified one. Defaults to None.

        Returns:
            Tensor: A tensor representing the sigma values for the given time steps.

        Raises:
            ValueError: If the input time steps exceed the maximum allowed value of 1.
        """
        if t.max() > 1:
            raise ValueError(f"Invalid value: max continuous time is 1, but got {t.max().item()}")

        if synchronize and self.direction != string_to_enum(synchronize, TimeDirection):
            t = 1 - t
        return self._calculate_sigma(t, device)

    @abstractmethod
    def _calculate_sigma(self, t: Tensor, device: Union[str, torch.device] = "cpu") -> Tensor:
        """Calculate the -log of the clean data value for the given time steps.

        Args:
            t (Tensor): The input tensor representing the time steps.
            device (Optional[str]): The device to place the schedule on. Defaults to "cpu".

        Returns:
            Tensor: A tensor representing the sigma values for the given time steps.
        """
        pass

    def sigma_to_alpha(self, sigma: Tensor) -> Tensor:
        """Converts sigma to alpha values by alpha = exp(- sigma).

        Args:
            sigma (Tensor): The input sigma tensor.

        Returns:
            Tensor: A tensor containing the alpha values.
        """
        return torch.exp(-1 * sigma)

__init__(direction)

初始化 DiscreteNoiseSchedule。

参数

名称 类型 描述 默认值
direction

TimeDirection，必需，这定义了构建调度器的方向

 必需
源代码在 bionemo/moco/schedules/noise/continuous_noise_transforms.py 中 
34
35
36
37
38
39
40
def __init__(self, direction: TimeDirection):
    """Initialize the DiscreteNoiseSchedule.

    Args:
        direction : TimeDirection, required this defines in which direction the scheduler was built
    """
    self.direction = string_to_enum(direction, TimeDirection)

calculate_sigma(t, device='cpu', synchronize=None)

计算给定时间步长的 sigma 值。

参数

名称 类型 描述 默认值
t Tensor

输入张量，表示时间步长，值范围为 0 到 1。

 必需
device Optional[str]

放置计划的设备。默认为 "cpu"。

 'cpu'
synchronize optional[TimeDirection]

用于同步计划的 TimeDirection。 如果计划使用不同的方向定义，则此参数允许翻转方向以匹配指定的方向。 默认为 None。

 None

返回值

名称 类型 描述
Tensor Tensor

表示给定时间步长的 sigma 值的张量。

Raises

类型 描述
ValueError

如果输入时间步长超过最大允许值 1。
源代码在 bionemo/moco/schedules/noise/continuous_noise_transforms.py 中 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def calculate_sigma(
    self,
    t: Tensor,
    device: Union[str, torch.device] = "cpu",
    synchronize: Optional[TimeDirection] = None,
) -> Tensor:
    """Calculate the sigma for the given time steps.

    Args:
        t (Tensor): The input tensor representing the time steps, with values ranging from 0 to 1.
        device (Optional[str]): The device to place the schedule on. Defaults to "cpu".
        synchronize (optional[TimeDirection]): TimeDirection to synchronize the schedule with. If the schedule is defined with a different direction,
            this parameter allows to flip the direction to match the specified one. Defaults to None.

    Returns:
        Tensor: A tensor representing the sigma values for the given time steps.

    Raises:
        ValueError: If the input time steps exceed the maximum allowed value of 1.
    """
    if t.max() > 1:
        raise ValueError(f"Invalid value: max continuous time is 1, but got {t.max().item()}")

    if synchronize and self.direction != string_to_enum(synchronize, TimeDirection):
        t = 1 - t
    return self._calculate_sigma(t, device)

sigma_to_alpha(sigma)

通过 alpha = exp(- sigma) 将 sigma 值转换为 alpha 值。

参数

名称 类型 描述 默认值
sigma Tensor

输入 sigma 张量。

 必需

返回值

名称 类型 描述
Tensor Tensor

包含 alpha 值的张量。
源代码在 bionemo/moco/schedules/noise/continuous_noise_transforms.py 中 
82
83
84
85
86
87
88
89
90
91
def sigma_to_alpha(self, sigma: Tensor) -> Tensor:
    """Converts sigma to alpha values by alpha = exp(- sigma).

    Args:
        sigma (Tensor): The input sigma tensor.

    Returns:
        Tensor: A tensor containing the alpha values.
    """
    return torch.exp(-1 * sigma)

CosineExpNoiseTransform

基类：ContinuousExpNoiseTransform

余弦指数噪声计划。

源代码在 bionemo/moco/schedules/noise/continuous_noise_transforms.py 中 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class CosineExpNoiseTransform(ContinuousExpNoiseTransform):
    """A cosine Exponential noise schedule."""

    def __init__(self, eps: Float = 1.0e-3):
        """Initialize the CosineNoiseSchedule.

        Args:
            eps (Float): small number to prevent numerical issues.
        """
        self.direction = TimeDirection.DIFFUSION
        self.eps = eps

    def _calculate_sigma(self, t: Tensor, device: Union[str, torch.device] = "cpu") -> Tensor:
        """Calculate negative log of data interpolant fraction.

        Args:
            t (Tensor): The input tensor representing the time steps.
            device (Optional[str]): The device to place the schedule on. Defaults to "cpu".

        Returns:
            Tensor: A tensor representing the sigma values for the given time steps.
        """
        cos = torch.cos(t * torch.pi / 2).to(device)
        return -torch.log(self.eps + (1 - self.eps) * cos)

    def d_dt_sigma(self, t: Tensor, device: Union[str, torch.device] = "cpu") -> Tensor:
        """Compute the derivative of sigma with respect to time.

        Args:
            t (Tensor): The input tensor representing the time steps.
            device (Optional[str]): The device to place the schedule on. Defaults to "cpu".

        Returns:
            Tensor: A tensor representing the derivative of sigma with respect to time.

        Notes:
            The derivative of sigma as a function of time is given by:

            d/dt sigma(t) = d/dt (-log(cos(t * pi / 2) + eps))

            Using the chain rule, we get:

            d/dt sigma(t) = (-1 / (cos(t * pi / 2) + eps)) * (-sin(t * pi / 2) * pi / 2)

            This is the derivative that is computed and returned by this method.
        """
        cos = (1 - self.eps) * torch.cos(t * torch.pi / 2)
        sin = (1 - self.eps) * torch.sin(t * torch.pi / 2)
        scale = torch.pi / 2
        derivative = scale * sin / (cos + self.eps)
        return derivative.to(device)

__init__(eps=0.001)

初始化 CosineNoiseSchedule。

参数

名称 类型 描述 默认值
eps Float

防止数值问题的小数字。

 0.001
源代码在 bionemo/moco/schedules/noise/continuous_noise_transforms.py 中 
 97
 98
 99
100
101
102
103
104
def __init__(self, eps: Float = 1.0e-3):
    """Initialize the CosineNoiseSchedule.

    Args:
        eps (Float): small number to prevent numerical issues.
    """
    self.direction = TimeDirection.DIFFUSION
    self.eps = eps

d_dt_sigma(t, device='cpu')

计算 sigma 对时间的导数。

参数

名称 类型 描述 默认值
t Tensor

输入张量，表示时间步长。

 必需
device Optional[str]

放置计划的设备。默认为 "cpu"。

 'cpu'

返回值

名称 类型 描述
Tensor Tensor

表示 sigma 对时间的导数的张量。
注意

sigma 作为时间函数的导数由下式给出

d/dt sigma(t) = d/dt (-log(cos(t * pi / 2) + eps))

使用链式法则，我们得到

d/dt sigma(t) = (-1 / (cos(t * pi / 2) + eps)) * (-sin(t * pi / 2) * pi / 2)

这是此方法计算并返回的导数。

源代码在 bionemo/moco/schedules/noise/continuous_noise_transforms.py 中 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def d_dt_sigma(self, t: Tensor, device: Union[str, torch.device] = "cpu") -> Tensor:
    """Compute the derivative of sigma with respect to time.

    Args:
        t (Tensor): The input tensor representing the time steps.
        device (Optional[str]): The device to place the schedule on. Defaults to "cpu".

    Returns:
        Tensor: A tensor representing the derivative of sigma with respect to time.

    Notes:
        The derivative of sigma as a function of time is given by:

        d/dt sigma(t) = d/dt (-log(cos(t * pi / 2) + eps))

        Using the chain rule, we get:

        d/dt sigma(t) = (-1 / (cos(t * pi / 2) + eps)) * (-sin(t * pi / 2) * pi / 2)

        This is the derivative that is computed and returned by this method.
    """
    cos = (1 - self.eps) * torch.cos(t * torch.pi / 2)
    sin = (1 - self.eps) * torch.sin(t * torch.pi / 2)
    scale = torch.pi / 2
    derivative = scale * sin / (cos + self.eps)
    return derivative.to(device)

LogLinearExpNoiseTransform

基类：ContinuousExpNoiseTransform

对数线性指数计划。

源代码在 bionemo/moco/schedules/noise/continuous_noise_transforms.py 中 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class LogLinearExpNoiseTransform(ContinuousExpNoiseTransform):
    """A log linear exponential schedule."""

    def __init__(self, eps: Float = 1.0e-3):
        """Initialize the CosineNoiseSchedule.

        Args:
            eps (Float): small value to prevent numerical issues.
        """
        self.direction = TimeDirection.DIFFUSION
        self.eps = eps

    def _calculate_sigma(self, t: Tensor, device: Union[str, torch.device] = "cpu") -> Tensor:
        """Calculate negative log of data interpolant fraction.

        Args:
            t (Tensor): The input tensor representing the time steps.
            device (Optional[str]): The device to place the schedule on. Defaults to "cpu".

        Returns:
            Tensor: A tensor representing the sigma values for the given time steps.
        """
        return -torch.log1p(-(1 - self.eps) * t).to(device)

    def d_dt_sigma(self, t: Tensor, device: Union[str, torch.device] = "cpu") -> Tensor:
        """Compute the derivative of sigma with respect to time.

        Args:
            t (Tensor): The input tensor representing the time steps.
            device (Optional[str]): The device to place the schedule on. Defaults to "cpu".

        Returns:
            Tensor: A tensor representing the derivative of sigma with respect to time.
        """
        derivative = (1 - self.eps) / (1 - (1 - self.eps) * t)
        return derivative.to(device)

__init__(eps=0.001)

初始化 CosineNoiseSchedule。

参数

名称 类型 描述 默认值
eps Float

防止数值问题的小值。

 0.001
源代码在 bionemo/moco/schedules/noise/continuous_noise_transforms.py 中 
150
151
152
153
154
155
156
157
def __init__(self, eps: Float = 1.0e-3):
    """Initialize the CosineNoiseSchedule.

    Args:
        eps (Float): small value to prevent numerical issues.
    """
    self.direction = TimeDirection.DIFFUSION
    self.eps = eps

d_dt_sigma(t, device='cpu')

计算 sigma 对时间的导数。

参数

名称 类型 描述 默认值
t Tensor

输入张量，表示时间步长。

 必需
device Optional[str]

放置计划的设备。默认为 "cpu"。

 'cpu'

返回值

名称 类型 描述
Tensor Tensor

表示 sigma 对时间的导数的张量。
源代码在 bionemo/moco/schedules/noise/continuous_noise_transforms.py 中 
171
172
173
174
175
176
177
178
179
180
181
182
def d_dt_sigma(self, t: Tensor, device: Union[str, torch.device] = "cpu") -> Tensor:
    """Compute the derivative of sigma with respect to time.

    Args:
        t (Tensor): The input tensor representing the time steps.
        device (Optional[str]): The device to place the schedule on. Defaults to "cpu".

    Returns:
        Tensor: A tensor representing the derivative of sigma with respect to time.
    """
    derivative = (1 - self.eps) / (1 - (1 - self.eps) * t)
    return derivative.to(device)