Args: x1: shape (bs, *dim), represents the target minibatch (data) x0: shape (bs, *dim), represents the source minibatch, if None we sample x0 from a standard normal distribution. cond_kwargs: additional arguments for the conditional flow
(self, x1: Tensor, x0: Tensor = None, **cond_kwargs)
| 190 | |
| 191 | |
| 192 | def training_losses(self, x1: Tensor, x0: Tensor = None, **cond_kwargs): |
| 193 | """ |
| 194 | Args: |
| 195 | x1: shape (bs, *dim), represents the target minibatch (data) |
| 196 | x0: shape (bs, *dim), represents the source minibatch, if None |
| 197 | we sample x0 from a standard normal distribution. |
| 198 | cond_kwargs: additional arguments for the conditional flow |
| 199 | network (e.g. conditioning information) |
| 200 | Returns: |
| 201 | loss: scalar, the training loss for the flow model |
| 202 | """ |
| 203 | if x0 is None: |
| 204 | x0 = torch.randn_like(x1) |
| 205 | |
| 206 | bs, dev, dtype = x1.shape[0], x1.device, x1.dtype |
| 207 | |
| 208 | # Sample time t from uniform distribution U(0, 1) |
| 209 | t = torch.rand(bs, device=dev, dtype=dtype) |
| 210 | |
| 211 | # sample xt and ut |
| 212 | xt = self.compute_xt(x0=x0, x1=x1, t=t) |
| 213 | ut = self.compute_ut(x0=x0, x1=x1, t=t) |
| 214 | vt = self.sample_vt(fm_x=xt, fm_t=t, **cond_kwargs) |
| 215 | |
| 216 | return (vt - ut).square() |
nothing calls this directly
no test coverage detected