将张量流概率分布作为双射器参数传递

Passing tensorflow-probability distributions as bijector parameters

我想创建一个 TransformedDistribution,它的变换双射器(双射器链)的一些组件本身被参数化为分布,objective 每次我通过双射器变换一些张量时都会有不同的结果(因为bijector的参数每次也会被采样)

让我们举一个非常简单的例子来说明我所说的(没有链):

import tensorflow as tf
import tensorflow_probability as tfp
tfd = tfp.distributions
tfb = tfp.bijectors

base_distribution = tfd.Normal(loc=0, scale=1.5)
x = base_distribution.sample(10) # Sample 10 times, to get fixed values
# Define the distribution for the parameter of the bijector
scale_distribution = tfd.Uniform(low=2/3-0.5, high=2/3+0.5)
# This is how I wish it was, but fails
bijector = tfb.Scale(scale=scale_distribution)
# ValueError: TypeError: object of type 'Uniform' has no len()

进一步详细说明示例:

transformed_dist = tfd.TransformedDistribution(
    base_distribution,
    bijector)

transformed_dist.sample() # get one sample of the original distribution, 
                          # scaled with some scale factor drawn from the
                          # uniform distribution.

我知道有一种方法可以使用 JointDistribution 构建分层分布模型,它让我可以使用一个(或多个)分布作为另一个分布的参数,然后从联合分布中采样。

但我还没有找到随机参数化双射器的等效方法。一种“有效”但有点麻烦的方法是:

# Works, but the sample is generated and becomes fixed when defining the bijector
bijector = tfb.Scale(scale=tf.random.uniform(minval=2/3-0.5, maxval=2/3+0.5,))

transformed_dist = tfd.TransformedDistribution(
    base_distribution,
    bijector)

transformed_dist.sample() # Now the distribution will always sample from
                          # a Normal what is Scaled by a deterministic 
                          # parameter, that was randomly generated at 
                          # definition time.

正如我在代码中解释的那样,每次我希望参数不同时都需要重新运行整个代码块。

我想这样做的原因是我想以某种旋转随机化的方式自动生成样本,即每次采样时获得不同的分布。

注意:我使用的是 tensorflow >2.1 和 eager execution。

我仍然认为 JointDistribution 是解决问题的方法,可以做类似

的事情
joint = tfd.JointDistributionSequential([
   tfd.Uniform(...),
   lambda unif: tfb.MyBijector(param=unif)(tfd.Normal(0., 1.))
])

输入双射器参数。

不过,您也可以用 tfp.util.DeferredTensor 拼凑一些东西。这是一个可以在张量所在的任何地方使用的对象,但其值取自给定函数的执行。例如

rand_dt = tfp.util.DeferredTensor(
    pretransformed_input=1.,  # value doesn't matter here
    transform_fn=lambda _: tf.random.uniform(minval=1., maxval=2.)
)

td = tfb.Scale(scale=rand_dt)(tfd.Normal(0., 1.))

for i in range(5):
  print(td.sample())
  # will print a sample with a different random scale on each call

这是一个 colab,对上述示例进行了稍微修改,以说明正在发生的事情:https://colab.research.google.com/drive/1akjX6a1W-RJoUjsy0hVOrrRAQiBIYjY-

HTH!

更新:

经过进一步思考,我实际上应该对第二种模式给予一些非常强烈的警告。不幸的是,很难保证 DeferredTensor 的单个 Tensor 化值将被一致使用——即使在例如单个 Distribution 方法调用的上下文中——所以做任何有副作用的事情都是危险的。例如:

rand_dt = tfp.util.DeferredTensor(
    pretransformed_input=1.,  # value doesn't matter here
    transform_fn=lambda _: tf.random.uniform(minval=1., maxval=2.)
)

td = tfb.Scale(scale=rand_dt)(tfd.Normal(0., 1.))

sample = td.sample()  # sample from TD with some random scale
print(td.log_prob(sample))  # new scale; wrong log_prob!

一种解决方法是使用无状态 TF 采样 API,直接使用或通过将列表值或张量值种子传递给 TFP(请参阅有关无状态采样语义的更多信息 here),并明确管理种子。不过,这可能会使上面的示例更加丑陋且难以使用(您可能需要有一个浮动变量作为种子输入并在每次需要新样本时得到 assigned)

您最好的选择可能是使用 JointDistribution 方法!