在 TensorFlow Probability 中实现“观察”行为
Achieving `observe` behaviour in TensorFlow Probability
考虑概率编程中 observe
语句的定义,如 [1] 中所定义:
The observe statement blocks runs which do not satisfy the boolean expression E and does not permit those executions to happen.
现在,考虑以下理论程序:
def f():
x ~ Normal(0, 1)
observe(x > 0) # only allow samples x > 0
return x
截断 Normal(0, 1)
分布中的 return 值。
因此,我的问题是:如何在 TensorFlow Probability 中实现 observe
,或者它的等价物是什么?请注意,observe
的参数应该是任何(符号)布尔表达式 E
:(例如 lambda x: x > 0
)。
注意:当然,对于上面的程序,可以使用 HalfNormal
分布,但我将其用于 observe
.
的实际示例
[1] 戈登、安德鲁 D. 等人。 “概率编程。” 关于软件工程未来的论文集。 2014. 167-181.
通常实现此目的的唯一方法是使用昂贵的拒绝采样器。然后你没有一个易于处理的密度。一般来说,TFP 要求我们所有的分布都具有易于处理的密度(即 dist.prob(x)
)。我们确实有一个友好的 autodiff TruncatedNormal
,或者如您所说 HalfNormal
。
如果你想实现其他东西,它可以像这样简单:
class Rejection(tfd.Distribution):
def __init__(self, underlying, condition, name=None):
self._u = underlying
self._c = condition
super().__init__(dtype=underlying.dtype,
name=name or f'rejection_{underlying}',
reparameterization_type=tfd.NOT_REPARAMETERIZED,
validate_args=underlying.validate_args,
allow_nan_stats=underlying.allow_nan_stats)
def _batch_shape(self):
return self._u.batch_shape
def _batch_shape_tensor(self):
return self._u.batch_shape_tensor()
def _event_shape(self):
return self._u.event_shape
def _event_shape_tensor(self):
return self._u.event_shape_tensor()
def _sample_n(self, n, seed=None):
return tf.while_loop(
lambda samples: not tf.reduce_all(self._c(samples)),
lambda samples: (tf.where(self._c(samples), samples, self._u.sample(n, seed=seed)),),
(self._u.sample(n, seed=seed),))[0]
d = Rejection(tfd.Normal(0,1), lambda x: x > -.3)
s = d.sample(100).numpy()
print(s.min())
考虑概率编程中 observe
语句的定义,如 [1] 中所定义:
The observe statement blocks runs which do not satisfy the boolean expression E and does not permit those executions to happen.
现在,考虑以下理论程序:
def f():
x ~ Normal(0, 1)
observe(x > 0) # only allow samples x > 0
return x
截断 Normal(0, 1)
分布中的 return 值。
因此,我的问题是:如何在 TensorFlow Probability 中实现 observe
,或者它的等价物是什么?请注意,observe
的参数应该是任何(符号)布尔表达式 E
:(例如 lambda x: x > 0
)。
注意:当然,对于上面的程序,可以使用 HalfNormal
分布,但我将其用于 observe
.
[1] 戈登、安德鲁 D. 等人。 “概率编程。” 关于软件工程未来的论文集。 2014. 167-181.
通常实现此目的的唯一方法是使用昂贵的拒绝采样器。然后你没有一个易于处理的密度。一般来说,TFP 要求我们所有的分布都具有易于处理的密度(即 dist.prob(x)
)。我们确实有一个友好的 autodiff TruncatedNormal
,或者如您所说 HalfNormal
。
如果你想实现其他东西,它可以像这样简单:
class Rejection(tfd.Distribution):
def __init__(self, underlying, condition, name=None):
self._u = underlying
self._c = condition
super().__init__(dtype=underlying.dtype,
name=name or f'rejection_{underlying}',
reparameterization_type=tfd.NOT_REPARAMETERIZED,
validate_args=underlying.validate_args,
allow_nan_stats=underlying.allow_nan_stats)
def _batch_shape(self):
return self._u.batch_shape
def _batch_shape_tensor(self):
return self._u.batch_shape_tensor()
def _event_shape(self):
return self._u.event_shape
def _event_shape_tensor(self):
return self._u.event_shape_tensor()
def _sample_n(self, n, seed=None):
return tf.while_loop(
lambda samples: not tf.reduce_all(self._c(samples)),
lambda samples: (tf.where(self._c(samples), samples, self._u.sample(n, seed=seed)),),
(self._u.sample(n, seed=seed),))[0]
d = Rejection(tfd.Normal(0,1), lambda x: x > -.3)
s = d.sample(100).numpy()
print(s.min())