2021 年 10 月 29 日 — 作者:魏魏,开发者倡导者 游戏通常被用作各种 强化学习 (RL) 算法的测试场。虽然机器学习研究人员发明新的 RL 算法来掌握具有挑战性的游戏非常令人兴奋,但我们也很想知道游戏开发者如何在 TensorFlow 中使用 RL 来构建游戏机器人,以实现各种目的,例如质量测试、游戏平衡调整…
作者:魏魏,开发者倡导者
游戏通常被用作各种 强化学习 (RL) 算法的测试场。虽然机器学习研究人员发明新的 RL 算法来掌握具有挑战性的游戏非常令人兴奋,但我们也很想知道游戏开发者如何在 TensorFlow 中使用 RL 来构建游戏机器人,以实现各种目的,例如质量测试、游戏平衡调整 和游戏难度评估。
我们已经有一个详细的 教程,演示了如何使用 TensorFlow 为经典的 CartPole 健身房环境实现 actor-critic RL 方法。在这个端到端的教程中,我们将向您展示如何使用 TensorFlow 核心、TensorFlow Agents 和 TensorFlow Lite 来构建一个游戏代理,在小型棋盘游戏应用程序中与人类用户对抗。最终结果是一个 Android 参考应用程序,看起来如下所示,我们已在 tensorflow/examples 存储库中开源了所有代码,供您参考。
“Plane Strike” 演示游戏 |
这款游戏名为“Plane Strike”,是一款小型棋盘游戏,类似于棋盘游戏 “战舰”。规则非常简单
即使为这样的小型游戏创建手工规则可能是可能的,我们也转向强化学习来创建一个人类玩家无法轻易击败的智能代理。有关强化学习的一般介绍,请参考 DeepMind 和 UCL 的这门 RL 课程。
我们为这款游戏应用程序提供了 2 种训练和部署路径
在此路径中,要训练代理,我们首先创建一个自定义的 OpenAI 健身房 环境“PlaneStrike-v0”,它允许我们轻松地推出游戏玩法并收集游戏日志。然后,我们使用奖励到去 策略梯度 算法来训练代理。REINFORCE 是一种 RL 中的策略梯度算法。它的基本思想是根据游戏过程中收集的奖励信号调整策略网络参数,以便策略网络可以在未来的游戏中最大化收益。
在数学上,策略梯度定义为
其中
请参考 DeepMind 关于策略梯度的讲座,以详细讨论。为了用 TensorFlow 实现它,我们定义了一个简单的 3 层 MLP 作为我们的策略网络,它根据人类玩家的棋盘状态预测代理的下一个攻击位置。请注意,上述策略梯度在没有奖励部分的情况下对数表达等效于负交叉熵损失。在这种情况下,由于我们希望最大化奖励,我们可以简单地最小化分类交叉熵损失来实现这一点。
model.compile(loss='sparse_categorical_crossentropy', optimizer=sgd)
我们创建了一个 play_game() 函数来推出游戏并帮助我们收集游戏日志。在每集结束后,我们通过 Keras fit() 函数训练代理
model.fit(x=board_log, y=action_log, sample_weight=rewards)
请注意,我们将折扣奖励到去作为“sample_weight”传递到 Keras fit() 函数中作为快捷方式,以实现策略梯度算法,而无需编写自定义训练循环。直观地理解这一点,我们需要一个 (x, y, reward) 元组,而不是像监督学习那样只使用 (x, y)。奖励(可以为负值)帮助预测器根据 x 输出朝 y 移动或远离 y。这与监督学习不同(在这种情况下,您的“sample_weight”永远不能为负)。
由于我们所做的不是监督学习,我们不能真正使用训练损失来监控训练进度。相反,我们将使用一个代理指标“game_length”,它指示代理完成每集需要多少步。直观地理解,如果代理更聪明并做出更好的预测,游戏长度就会变短。
TensorBoard 中的训练进度 |
由于这是一款需要代理立即响应的游戏,我们希望在移动设备而不是服务器上部署模型。训练完模型后,我们使用 TFLite 转换器将 Keras 模型转换为 TFLite 模型,并将其集成到我们的 Android 应用程序中。
converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()
导出的模型非常快,在 Pixel 手机上执行时间不到 1 毫秒。在游戏过程中,代理在每一步都会查看用户的棋盘位置并预测它下一步的攻击位置,以便尽快获得 8 个红色方格。
convertBoardStateToByteBuffer(board);
tflite.run(boardData, outputProbArrays);
float[] probArray = outputProbArrays[0];
int agentStrikePosition = -1;
float maxProb = 0;
for (int i = 0; i < probArray.length; i++) {
int x = i / Constants.BOARD_SIZE;
int y = i % Constants.BOARD_SIZE;
if (board[x][y] == BoardCellStatus.UNTRIED && probArray[i] > maxProb) {
agentStrikePosition = i;
maxProb = probArray[i];
}
}
虽然使用 TensorFlow API 从头开始编写我们的代理是一个很好的练习,但最好利用现有的 RL 算法实现。 TensorFlow Agents 是一个用于 TensorFlow 中强化学习的库,它通过提供经过良好测试的模块化组件(可以修改和扩展)来简化新 RL 算法的设计、实现和测试。TF Agents 已实现了几种最先进的 RL 算法,包括 DQN、DDPG、REINFORCE、PPO、SAC 和 TD3。TF Agents 训练的策略可以直接 转换为 TFLite 并部署到移动应用程序中(请注意,此功能最近才启用,因此您将需要 TensorFlow 和 TensorFlow Agents 的夜间构建版本)。
我们使用 TF Agents REINFORCE 代理 来训练我们的代理。首先,我们需要像上一节使用健身房环境那样定义一个 TF Agents 训练环境。然后,我们可以定义一个 actor 网络作为我们的策略网络
actor_net = tfa.networks.Sequential([
tfa.keras_layers.InnerReshape([BOARD_SIZE, BOARD_SIZE], [BOARD_SIZE**2]),
tf.keras.layers.Dense(FC_LAYER_PARAMS, activation='relu'),
tf.keras.layers.Dense(BOARD_SIZE**2),
tf.keras.layers.Lambda(lambda t: tfp.distributions.Categorical(logits=t)),
], input_spec=train_py_env.observation_spec(
))
我们将使用 TF Agents 已经实现的内置 REINFORCE 代理。该代理构建在上面定义的“actor_net”之上
tf_agent = reinforce_agent.ReinforceAgent(
train_env.time_step_spec(),
train_env.action_spec(),
actor_network=actor_net,
optimizer=optimizer,
normalize_returns=True,
train_step_counter=train_step_counter)
为了训练代理,我们需要收集一些轨迹作为经验。我们使用 DeepMind Reverb 和 TF Agent PyDriver 为此定义了一个函数
def collect_episode(environment, policy, num_episodes, replay_buffer_observer):
"""Collect game episode trajectories."""
initial_time_step = environment.reset()
driver = py_driver.PyDriver(
environment,
py_tf_eager_policy.PyTFEagerPolicy(policy, use_tf_function=True),
[replay_buffer_observer],
max_episodes=num_episodes)
initial_time_step = environment.reset()
driver.run(initial_time_step)
现在我们已经准备好训练模型了
for i in range(iterations):
# Collect a few episodes using collect_policy and save to the replay buffer.
collect_episode(train_py_env, collect_policy,
COLLECT_EPISODES_PER_ITERATION, replay_buffer_observer)
# Use data from the buffer and update the agent's network.
iterator = iter(replay_buffer.as_dataset(sample_batch_size=1))
trajectories, _ = next(iterator)
tf_agent.train(experience=trajectories)
replay_buffer.clear()
您可以使用 TensorBoard 监控训练进度。在这种情况下,我们可视化了平均集长度和平均收益。
TensorBoard 中的 TF Agents 训练进度
一旦策略经过训练并作为 SavedModel 导出,您就可以将其转换为 TFLite 模型
converter = tf.lite.TFLiteConverter.from_saved_model(
policy_dir, signature_keys=['action'])
converter.target_spec.supported_ops = [
tf.lite.OpsSet.TFLITE_BUILTINS, # enable TensorFlow Lite ops.
tf.lite.OpsSet.SELECT_TF_OPS # enable TensorFlow ops.
]
tflite_policy = converter.convert()
with open(os.path.join(model_dir, 'planestrike_tf_agents.tflite'), 'wb') as f:
f.write(tflite_policy)
当前,转换期间需要一些 TensorFlow 操作。转换后的模型与我们直接使用 TensorFlow 训练的模型略有不同,因为它以 4 个张量作为输入。真正重要的是“observation”张量。我们的代理将查看此“observation”张量并预测下一步行动。其他 3 个可以在推理时安全地忽略。
使用 Netron 可视化从 TF Agents 转换的 TFLite 模型 |
此外,该模型直接输出攻击位置而不是概率分布,因此我们不再需要手动执行 argmax。
@Override
protected void runInference() {
Map output = new HashMap<>();
// TF Agent directly returns the predicted action
int[][] prediction = new int[1][1];
output.put(0, prediction);
tflite.runForMultipleInputsOutputs(inputs, output);
agentStrikePosition = prediction[0][0];
总之,在这篇文章中,我们向您展示了两种训练游戏代理、将训练后的模型转换为 TFLite 并将其部署到 Android 应用程序中的路径。希望这个端到端的教程能帮助您更好地理解如何利用 TensorFlow 生态系统来构建酷炫的游戏。
最后,如果这款小游戏对您来说很有趣,我们挑战您在手机上安装 这款应用程序,看看您是否能击败我们训练的代理 😃。