PyEnvironment 的 time_step 与 time_step_spec 结构不匹配怎么办？
需要将 observation_spec() 定义为与实际 observation 完全对应的嵌套结构，并确保 _reset()、_step() 返回的每个叶子值都符合对应的 shape 和 dtype。当前异常显示，实际 observation 是嵌套字典，而规格声明的是一个 float32 标量。
报错发生在策略读取 time_step 时，尚未进入将该步数据写入 Reverb 的阶段。先修正环境与策略之间的观测约定，再检查网络和 replay buffer。
为什么现有规格不匹配？
当前声明：
self._observation_spec = array_spec.BoundedArraySpec(
shape=(),
dtype=np.float32,
minimum=0,
maximum=1000,
name='observation_spec',
)这表示 observation 是一个取值在 [0, 1000] 之间的 np.float32 标量，无法描述下面这样的数据：
{
'dev_strength': {'r': ..., 'best': ...},
'dev_100': {'start': ..., 'best': ..., 'shift': ..., 'k': ...},
'100_i': ...,
'inv_strength': ...,
't': ...,
}规格中的字典键和嵌套层级必须与实际数据对应，叶子位置放 ArraySpec 或 BoundedArraySpec。
原来注释掉的字典规格也不能直接启用，至少存在以下差异：
dev_{event}实际包含start、best、shift、k，注释中的定义却只有r、best。- 实际浮点字段为
float64，注释中的规格多为float32。 - 部分实际整数字段为
int64，注释中的规格使用了int32或float32。
_reset()、_step() 能返回结果，并不代表返回值符合环境规格。
按日志中的结构定义 observation_spec
下面的示例按照异常日志中的字段和 dtype 建立规格。现有材料无法确认每个字段的有效范围，因此先使用 ArraySpec。确定业务边界后，再将相应字段换成 BoundedArraySpec。
import numpy as np
from tf_agents.specs import array_spec
def scalar_spec(name, dtype):
return array_spec.ArraySpec(
shape=(),
dtype=dtype,
name=name,
)
def build_observation_spec(latent_abilities, events):
specs = {}
for ability in latent_abilities:
key = f'dev_{ability}'
specs[key] = {
'r': scalar_spec(f'{key}_r', np.float64),
'best': scalar_spec(f'{key}_best', np.int64),
}
for event in events:
key = f'dev_{event}_tech'
specs[key] = {
'r': scalar_spec(f'{key}_r', np.float64),
'best': scalar_spec(f'{key}_best', np.int64),
}
key = f'dev_{event}'
specs[key] = {
'start': scalar_spec(f'{key}_start', np.float64),
'best': scalar_spec(f'{key}_best', np.float64),
'shift': scalar_spec(f'{key}_shift', np.int64),
'k': scalar_spec(f'{key}_k', np.float64),
}
specs[f'{event}_i'] = scalar_spec(
f'{event}_i', np.float64
)
specs[f'{event}_o'] = scalar_spec(
f'{event}_o', np.float64
)
specs[f'inv_{event}_tech'] = scalar_spec(
f'inv_{event}_tech', np.int64
)
for ability in latent_abilities:
specs[f'inv_{ability}'] = scalar_spec(
f'inv_{ability}', np.int64
)
specs['t'] = scalar_spec('t', np.int64)
return specs在环境初始化时设置该规格，并通过 observation_spec() 返回：
self._observation_spec = build_observation_spec(
self.latent_abilities,
self.events,
)def observation_spec(self):
return self._observation_spec这里将能力发展和技术发展中的 best 定为 np.int64，是为了与所给日志一致。如果这些值在后续步骤中可能出现小数，应同时调整规格和实际输出类型，不能只凭初始值是整数就确定整个训练过程的 dtype。
让 reset 和 step 使用同一套观测转换
_reset() 和 _step() 都应返回 TimeStep，其中的 observation 由同一个转换函数生成。结构正确后，可以按规格显式转换每个叶子的 dtype：
import tensorflow as tf
import numpy as np
def convert_observation(raw_observation, observation_spec):
tf.nest.assert_same_structure(
observation_spec,
raw_observation,
)
def convert_leaf(spec, value):
value = np.asarray(value, dtype=spec.dtype)
if value.shape != spec.shape:
raise ValueError(
f'{spec.name}: expected shape {spec.shape}, '
f'got {value.shape}'
)
return value
return tf.nest.map_structure(
convert_leaf,
observation_spec,
raw_observation,
)将现有 observation_from_state(...) 的输出传给这个函数，再构造时间步：
from tf_agents.trajectories import time_step as ts
# 在 _reset() 完成状态初始化后：
observation = convert_observation(
raw_observation,
self.observation_spec(),
)
return ts.restart(observation)# 在 _step() 完成状态更新后：
observation = convert_observation(
raw_observation,
self.observation_spec(),
)
reward = np.asarray(reward, dtype=np.float32)
if episode_ended:
return ts.termination(observation, reward)
return ts.transition(
observation,
reward=reward,
discount=np.asarray(self.gamma_, dtype=np.float32),
)以上 raw_observation、reward、episode_ended 代表由现有环境逻辑计算出的值，需要接入实际实现。转换函数不会补齐缺失键，也不能修正错误的业务字段；它按已确定的规格转换数值类型并检查 shape。转成整数前，应确认该字段确实只允许整数，避免截断有效的小数。
日志中的 shape=(1,) 是否需要写进规格？
对于未批处理的 PyEnvironment，标量字段应声明为 shape=()，并返回相应的 NumPy 标量数组。
经过 TFPyEnvironment 或 PyTFEagerPolicy 的批处理转换后，单个环境的标量可能表现为 shape=(1,)。这个 1 是外层 batch 维度，不应因此将原始标量规格改成 shape=(1,)。
应先检查 env_train_py.reset() 中的原始 NumPy 值，不要仅凭策略内部 Tensor 的 shape 判断规格。本次异常首先报告的是字典与单个叶子之间的结构差异。
DQN 网络也需要接收嵌套观测
修正环境规格后，当前网络仍有输入问题：
input_spec=tf.TensorSpec(shape=(1,), dtype=tf.int32)它声明的是一个整数向量，无法对应整棵嵌套 observation。普通 Dense 层也不能直接将这个字典作为特征矩阵使用。
如果保留字典观测，可以使用 QNetwork 的预处理功能，将每个标量转成 float32，扩展为长度为 1 的特征，拼接后再送入全连接层。
import tensorflow as tf
from tf_agents.networks import q_network
observation_spec = train_env.observation_spec()
def make_preprocessing_layer(_):
return tf.keras.Sequential([
tf.keras.layers.Lambda(
lambda x: tf.cast(x, tf.float32)
),
tf.keras.layers.Reshape((1,)),
])
preprocessing_layers = tf.nest.map_structure(
make_preprocessing_layer,
observation_spec,
)
q_net = q_network.QNetwork(
input_tensor_spec=observation_spec,
action_spec=train_env.action_spec(),
preprocessing_layers=preprocessing_layers,
preprocessing_combiner=tf.keras.layers.Concatenate(axis=-1),
fc_layer_params=(100, 50),
)该示例假设所有观测叶子都是标量，与当前日志一致。如果后续增加向量或矩阵字段，应相应修改预处理层。不同量纲的成绩、计数和发展参数是否需要归一化，还需根据实际数值范围决定。
也可以在环境出口将观测转换为固定顺序的 float32 向量，再用一个向量规格描述它。但必须同步修改实际 observation、规格及网络输入，单独改变 shape 不会自动将字典转换为向量。
统一采集环境，并在接入 Reverb 前验证
原代码把 env 交给 driver，却用另一个实例 env_train_py 的时间步启动。即使两者规格相同，内部状态也可能不同。应使用同一个实例：
py_driver.PyDriver(
env_train_py,
py_tf_eager_policy.PyTFEagerPolicy(
policy_random,
use_tf_function=True,
),
[rb_observer],
max_steps=initial_collect_steps,
).run(env_train_py.reset())先在创建 TensorFlow 包装器之前验证 Python 环境：
from tf_agents.environments import utils
utils.validate_py_environment(
env_train_py,
episodes=5,
)这个检查有助于发现采样过程中输出与规格不一致的问题，但通过有限次数的运行，不代表所有动作路径和业务边界都已验证。
修改规格后，应重新创建训练和评估环境、TFPyEnvironment、网络、agent、policy，以及依赖 agent.collect_data_spec 的 Reverb table、buffer 和 observer。已有对象可能仍持有旧规格，也不能直接假定旧表中的数据与新结构兼容。
先确认 Python 环境验证和策略取动作都正常，再恢复 Reverb 数据采集。所给堆栈没有显示 Reverb 写入错误，目前没有依据认定 sequence_length=2 或 table signature 是这次异常的原因。
备注：内容仅供参考。