inverse¶

inverse ¶

Posterior sampling machinery for linear inverse problems.

v1 space decision. DeltaFlow v1 targets the pixel-space case by default: the velocity field v_theta(x, t) acts directly on images and the measurement operator A is defined on the same space. However, all components in deltaflow.inverse fully support the latent-space case: GaussianLikelihood accepts an optional decoder callable, and because the likelihood gradient is computed by autograd, gradients are pulled back through the decoder into the latent state automatically. See the docstring on GaussianLikelihood for the wiring.

Modules:

operators, measurement operators A: masks, blur, downsample.
tweedie, flow-matching Tweedie decomposition of (x_t, v_t, t) into a clean-signal estimate and a noise estimate (FlowDPS-style).
likelihood, -log p(y | x_clean_hat) objects for the PosteriorSolver.

GaussianLikelihood ¶

GaussianLikelihood(y, operator, sigma=1.0, decoder=None, reduction='sum')

Gaussian measurement likelihood for posterior sampling.

For a measurement \(y = A(x) + n\) with \(n \sim \mathcal{N}(0, \sigma^2 I)\), the negative log-likelihood of the clean-signal estimate is

\[ -\log p(y \mid \hat{x}_1) = \frac{1}{2\sigma^2}\,\bigl\| y - A\bigl(D(\hat{x}_1)\bigr) \bigr\|^2 \;+\; \text{const}, \]

where \(A\) is the measurement operator and \(D\) an optional decoder (\(D = \mathrm{id}\) when the field and operator share a space). The constant is dropped, and \(\sigma\) only scales the gradient magnitude, not its direction.

Parameters:

Name	Type	Description	Default
`y`	`Tensor`	measured tensor, shape matches \(A\)'s output.	required
`operator`	`Callable[[Tensor], Tensor]`	linear measurement operator \(A\) (any callable, e.g. an instance from `deltaflow.inverse.operators`).	required
`sigma`	`float`	measurement-noise standard deviation \(\sigma\). Only affects the scale of the likelihood gradient. The direction is independent of \(\sigma\).	`1.0`
`decoder`	`Optional[Callable[[Tensor], Tensor]]`	optional decoder \(D\) (latent to pixel), used when the velocity field is trained in a latent space but \(A\) is defined on pixels. Autograd flows through it automatically.	`None`
`reduction`	`str`	`"sum"` (default, matches the log-density scaling) or `"mean"`. The `PosteriorSolver` sums per-sample values, so `"sum"` is usually what you want.	`'sum'`

Source code in deltaflow/inverse/likelihood.py

def __init__(
    self,
    y: torch.Tensor,
    operator: Callable[[torch.Tensor], torch.Tensor],
    sigma: float = 1.0,
    decoder: Optional[Callable[[torch.Tensor], torch.Tensor]] = None,
    reduction: str = "sum",
):
    self.y = y
    self.operator = operator
    self.sigma = float(sigma)
    self.decoder = decoder
    if reduction not in ("sum", "mean"):
        raise ValueError(f"reduction must be 'sum' or 'mean', got {reduction!r}")
    self.reduction = reduction

neg_log_prob ¶

neg_log_prob(x_clean_hat)

Return -log p(y | x_clean_hat) as a scalar tensor.

The result stays in the autograd graph of x_clean_hat so the posterior solver can backprop through it.

Source code in deltaflow/inverse/likelihood.py

def neg_log_prob(self, x_clean_hat: torch.Tensor) -> torch.Tensor:
    """Return ``-log p(y | x_clean_hat)`` as a scalar tensor.

    The result stays in the autograd graph of ``x_clean_hat`` so the
    posterior solver can backprop through it.
    """
    pixel = self.decoder(x_clean_hat) if self.decoder is not None else x_clean_hat
    y_pred = self.operator(pixel)
    diff = self.y.to(dtype=y_pred.dtype, device=y_pred.device) - y_pred
    sq = diff.pow(2)
    if self.reduction == "mean":
        sq = sq.mean()
    else:
        sq = sq.sum()
    return 0.5 * sq / (self.sigma * self.sigma)

Likelihood ¶

Bases: Protocol

Structural type for measurement-likelihood objects.

Any object exposing a differentiable neg_log_prob(x_clean_hat) method (returning a scalar or per-sample tensor that stays in x_clean_hat's autograd graph) satisfies this protocol and can be passed to PosteriorSolver.

BlurOperator ¶

BlurOperator(kernel_size=5, sigma=1.0, channels=1)

Bases: Module

Gaussian blur with a fixed kernel.

Parameters:

Name	Type	Description	Default
`kernel_size`	`int`	odd integer, side length of the Gaussian kernel.	`5`
`sigma`	`float`	standard deviation of the Gaussian.	`1.0`
`channels`	`int`	number of channels the operator will see. Required so the depthwise convolution weight is registered up front.	`1`

Source code in deltaflow/inverse/operators.py

def __init__(self, kernel_size: int = 5, sigma: float = 1.0, channels: int = 1):
    super().__init__()
    if kernel_size % 2 == 0:
        raise ValueError("kernel_size must be odd")
    self.kernel_size = kernel_size
    self.sigma = sigma
    self.channels = channels
    self.register_buffer("kernel", self._make_kernel(kernel_size, sigma, channels))

DownsampleOperator ¶

DownsampleOperator(factor=2)

Bases: Module

Average-pool downsampling by an integer factor.

Source code in deltaflow/inverse/operators.py

def __init__(self, factor: int = 2):
    super().__init__()
    self.factor = int(factor)

IdentityOperator ¶

Bases: Module

A(x) = x. Useful as a no-op default and for denoising tasks.

MaskOperator ¶

MaskOperator(mask=None)

Bases: Module

Elementwise masking (inpainting).

Parameters:

Name	Type	Description	Default
`mask`	`Optional[Tensor]`	broadcastable to `x`. Values of 1 keep, values of 0 drop. Passed either as a `torch.Tensor` or, at call time, via the `mask` keyword argument to `forward`.	`None`

Source code in deltaflow/inverse/operators.py

def __init__(self, mask: Optional[torch.Tensor] = None):
    super().__init__()
    if mask is not None:
        self.register_buffer("mask", mask.to(torch.float32))
    else:
        self.mask = None

BaseTweedie ¶

Bases: ABC

Base class for path-specific (x_t, v_t, t) -> (x_clean, x_noise).

LinearTweedie ¶

Bases: BaseTweedie

Tweedie decomposition for the linear (rectified-flow) path.

On the linear path \(x_t = (1-t)x_0 + t x_1\) with velocity \(v_t = x_1 - x_0\), the two endpoints are recovered in closed form by solving the \(2\times 2\) linear system:

\[ \hat{x}_1 = x_t + (1 - t)\,v_t, \qquad \hat{x}_0 = x_t - t\,v_t, \]

where \(\hat{x}_1\) is the clean-data estimate and \(\hat{x}_0\) the noise estimate.

VPTweedie ¶

Bases: BaseTweedie

Tweedie decomposition for the trigonometric variance-preserving path.

With \(\alpha_t = \sin(\tfrac{\pi}{2}t)\), \(\sigma_t = \cos(\tfrac{\pi}{2}t)\), the path is \(x_t = \alpha_t x_1 + \sigma_t x_0\) and its velocity \(v_t = \tfrac{\pi}{2}(\sigma_t x_1 - \alpha_t x_0)\). Inverting this \(2\times 2\) system (using \(\alpha_t^2 + \sigma_t^2 = 1\)) gives

\[ \hat{x}_1 = \alpha_t\,x_t + \frac{2}{\pi}\,\sigma_t\,v_t, \qquad \hat{x}_0 = \sigma_t\,x_t - \frac{2}{\pi}\,\alpha_t\,v_t. \]