Ray RLlib 外部环境集成指南RLlink 协议详解与 TCP 客户端接入实战【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray导读本文以 RLlib 环境包参考文档 为核心系统讲解 Ray RLlib 新 API 栈下连接外部环境External Envs的官方方案——RLlink 通信协议。当训练环境是一个拥有自身执行循环的复杂模拟器如游戏引擎、机器人仿真器时可以让模拟器作为 TCP 客户端自主推进仿真把收集到的经验批量回传 RLlib 服务端进行训练。读完本文你将掌握 RLlink 的协议帧格式、全部消息类型与收发 API理解EnvRunnerServerForExternalInference服务端的实现原理并能复现仓库中完整的 TCP 客户端接入实战示例。:align: left :width: 600 **客户端侧推理的外部应用接入架构**外部模拟器作为客户端连接以 TCP 方式运行的 RLlib 服务端自定义 EnvRunner周期性批量回传采样数据并接收权重更新动作推理在客户端本地完成以获得更好性能。为什么需要外部环境从 RLlib 步进环境到环境主动上报RLlib 常规的用法是由训练框架步进step一个 Gym 环境框架调用env.step(action)获得下一个观测与奖励。但有一类场景这样做并不合适——例如在游戏引擎或机器人仿真这类自带执行循环的复杂模拟器中仿真步进节奏完全由模拟器自身掌控无法也不应该被训练进程逐帧驱动。仓库在 doc/source/rllib/external-envs.md 中给出了这一问题的自然解法翻转控制关系——由模拟器中的智能体自行控制步进RLlib 以一个外部服务的形式存在负责回答单个动作查询或接收批量采样数据并训练策略但不限制模拟器每秒的步进频率。这正是 RLlib 新 API 栈Ray 2.40 起默认启用见 new_api_stack.rst中外部环境接入的定位。官方参考文档 external.rst 将这一能力收敛为对ray.rllib.env.external.rllink模块RLlink枚举类、get_rllink_message、send_rllink_message两个函数的公开 API 引用本文即围绕这三者展开。RLlink 协议概览简单、有状态的 RL 专用通信协议RLlinkray.rllib.env.external.rllink.RLlink是一个用于 RL 服务端如 RLlib与外部客户端环境模拟器之间通信的简单、有状态协议其当前协议版本为PROTOCOL_VERSION Version(0.0.1)。它在 rllib/env/external/rllink.py 中以枚举类实现承载交换 episode 数据、模型权重与算法配置等 RL 专属信息并原生支持 on-policy 训练工作流。协议的核心设计特征包括有状态设计协议在多次消息交换之间维持状态典型如GET_CONFIG→SET_CONFIG的请求-响应对客户端主动发起通信始终由客户端发起服务端从不主动发送未经请求的消息只有对PING、GET_CONFIG、GET_STATE、EPISODES_AND_GET_STATE这类期待响应的请求才应答而裸的EPISODES消息不会收到任何回复灵活的采样方式通过EPISODES_AND_GET_STATE支持 on-policy 数据收集通过EPISODES支持 off-policy 收集msgpack 编码消息体使用 msgpack 编码协议前几个版本未加密、不安全。消息帧结构8 字节长度头 msgpack 消息体一条 RLlink 消息由头部和消息体组成Header头部8 字节的长度字段以 ASCII 十进制数字表示消息体的字节数左补零。例如00000011表示消息体有 11 字节。该长度不含头部本身因此完整帧为 8 字节头部 消息体长度Body消息体msgpack 编码的字典其中type字段为必填用于标识消息类型。以官方文档 external-envs.md 中的PING消息为例其消息体是{type: PING}的 msgpack 编码共 11 字节因此完整帧为b00000011 b\x81\xa4type\xa4PING服务端应答的PONG帧结构相同b00000011 b\x81\xa4type\xa4PONG帧的读写实现源码级帧的组装与解析实现在 rllib/env/external/rllink.py 的两个函数中这也是参考文档 autosummary 中列出的两个公开函数send_rllink_message(sock_, message)先通过msgpack.packb(message, use_bin_typeTrue)序列化消息体再用str(len(body)).zfill(8).encode(utf-8)生成 8 位左补零的长度头最后sock_.sendall(header body)一次性发送get_rllink_message(sock_)先读取 8 字节长度头再按长度读取消息体msgpack.unpackb(body, rawFalse)反序列化后强制校验消息字典中必须包含type字段否则抛出ConnectionError(Protocol Error! ...)随后将type字段映射为RLlink枚举成员并返回(RLlink, message)元组。底层字节读取由私有辅助函数_get_num_bytes完成它循环recv直到收满指定字节数保证不会因 TCP 分包而读到不完整帧。需要注意的是get_rllink_message会从消息体中pop(type)因此调用方拿到的message字典中不再包含type键。全部消息类型详解RLlink枚举按方向划分为请求客户端 → 服务端与响应服务端 → 客户端两大类此外还保留了新 API 栈已弃用OldAPIStack即将移除的一批旧消息。以下内容综合 rllink.py 源码与 external-envs.md 文档整理。请求客户端 → 服务端消息类型用途消息体期望响应PING初始握手建立通信{type: PING}PONGGET_CONFIG请求算法配置客户端据此构建本地RLModule并确定收集多少步后再发送EPISODES_AND_GET_STATE{type: GET_CONFIG}SET_CONFIGGET_STATE请求当前状态如模型权重不附带任何 episode{type: GET_STATE}SET_STATEEPISODES批量发送收集到的 episode 供服务端 off-policy 训练服务端接收后不回包episodesSingleAgentEpisode.get_state()字典列表无EPISODES_AND_GET_STATE合并EPISODES与GET_STATE支持采集后立即同步更新的 on-policy 工作流episodes逐段 episode 的状态字典服务端用SingleAgentEpisode.from_state()重建timesteps本批环境步数SET_STATEEPISODES_AND_GET_STATE的完整发送示例来自 external-envs.mdsend_rllink_message( sock, { type: EPISODES_AND_GET_STATE, # 每个元素是一个 SingleAgentEpisode.get_state() 字典。 episodes: [episode.get_state() for episode in episodes], timesteps: 128, }, )响应服务端 → 客户端消息类型用途消息体PONG应答PING确认连通性{type: PONG}SET_CONFIG向客户端下发算法配置configpickle 序列化后的AlgorithmConfig客户端pickle.loads()后据此构建本地RLModule因载荷为 pickle务必只连接可信服务端SET_STATE向客户端下发当前状态模型权重state含rl_moduleRLModule.get_state()输出与weights_seq_no权重版本号两个键SET_STATE消息的典型形状{ type: SET_STATE, state: { rl_module: ..., # RLModule.get_state() 输出 weights_seq_no: 123, }, }其中weights_seq_no是模型权重的版本号跨消息对比它可以判断客户端采集的数据有多on-policy——即客户端是用最新权重还是旧权重采样的。旧 API 栈的遗留消息即将废弃源码中还保留了一组标注OldAPIStack (to be deprecated soon)的旧消息ACTION_SPACE、OBSERVATION_SPACE、GET_WORKER_ARGS、GET_WEIGHTS、REPORT_SAMPLES、START_EPISODE、GET_ACTION、LOG_ACTION、LOG_RETURNS、END_EPISODE。它们是旧 API 栈ExternalEnv/ExternalMultiAgentEnv逐帧动作问答模式的遗留协议新接入项目应使用上文的新消息类型。标准工作流从握手到 on-policy 训练external-envs.md 给出了四步标准流程与 dummy_external_client.py 中的实际客户端代码一一对应初始握手客户端发送PING服务端应答PONG配置请求客户端发送GET_CONFIG服务端应答SET_CONFIG客户端用配置构建本地RLModule并从中读取get_rollout_fragment_length()决定每批采集多少步初始权重请求客户端发送GET_STATE服务端应答SET_STATE客户端将权重加载进本地RLModuleOn-policy 训练循环客户端采集数据后发送EPISODES_AND_GET_STATE服务端接收 episode 后应答SET_STATE客户端阻塞等待响应收到新权重后再开始下一批采集从而保证同步的 on-policy 更新。服务端实现EnvRunnerServerForExternalInference 源码剖析参考文档引用的RLlink协议在服务端由自定义 EnvRunner 消费。仓库在 rllib/env/external/env_runner_server_for_external_inference.py 提供了参考实现EnvRunnerServerForExternalInference旧名TcpClientInferenceEnvRunner见 tcp_client_inference_env_runner.py 的兼容别名。该实现基于三个假设每个 EnvRunner 同一时刻只接受一个外部客户端连接外部客户端持有 connector 流水线与 RLModule推理在客户端本地完成样本以RLlib episode 列表的形式成批回传该 EnvRunner 上始终保留一份 RLModule 副本但只作为权重容器、不参与推理。其核心机制从源码看包括端口分配监听地址为localhost端口为env_config[port]默认 5555加上worker_index即每个 EnvRunner actor 监听不同端口多个客户端可并行连接后台监听线程构造函数启动一个 daemon 线程_client_message_listener先绑定 socket、listen(1)并accept()单个客户端随后进入消息循环按RLlink消息类型分发处理PING→ 回PONGEPISODES/EPISODES_AND_GET_STATE→ 调用_process_episodes_message用SingleAgentEpisode.from_state()重建并to_numpy()转成 numpy 后缓存GET_STATE→ 回SET_STATEGET_CONFIG→ 回SET_CONFIG配置以pickle.dumps(self.config)传输on-policy 阻塞收到EPISODES_AND_GET_STATE后置_blocked_on_state True暂停处理后续消息直到学习器调用set_state推送新权重此时_send_set_state_message()将状态发回客户端并解除阻塞采样接口sample()在_sample_lock保护下轮询等待客户端送来的 episode 块按len(eps)累计环境步数、区分已完成/进行中的 episode 并更新指标权重同步set_state通过weights_seq_no判断是否需要真正更新本地 RLModule 权重版本为 0 或落后于新版本才更新支持从ray.ObjectRef中ray.get拉取状态断线恢复任何ConnectionError都会触发_recycle_sockets(5.0)——关闭旧 socket、休眠 5 秒后重新监听、等待客户端重连。端到端实战用 TCP 客户端连接 RLlib 训练 CartPole仓库在 rllib/examples/envs/env_connecting_to_rllib_w_tcp_client.py 提供了完整可运行示例演示如何让外部模拟器通过 TCP 连接 RLlib 服务端进行训练。服务端配置要点服务端通过标准配置 API 指定自定义 EnvRunner 与环境空间关键点包括使用observation_space/action_space明确外部环境的观测与动作空间示例为 4 维连续观测 2 个离散动作的 CartPole在env_config{port: args.port}中指定监听端口通过.env_runners(env_runner_clsEnvRunnerServerForExternalInference)将默认 EnvRunner 替换为外部推理服务端base_config ( get_trainable_cls(args.algo) .get_default_config() .environment( observation_spacegym.spaces.Box(float(-inf), float(-inf), (4,), np.float32), action_spacegym.spaces.Discrete(2), # EnvRunners 监听 port worker_index 端口。 env_config{port: args.port}, ) .env_runners( # 指向自定义 EnvRunner。 env_runner_clsEnvRunnerServerForExternalInference, ) .training(num_epochs10, vf_loss_coeff0.01) .rl_module(model_configDefaultModelConfig(vf_share_layersTrue)) )运行方式python rllib/examples/envs/env_connecting_to_rllib_w_tcp_client.py --port 5555 --use-dummy-client--portRLlib EnvRunner 的监听端口默认 5555客户端侧需保持一致--use-dummy-client启动内置的哑客户端模拟器线程不带该参数时可自行从 C 应用等外部程序连接调试时可追加--no-tune --num-env-runners0便于在 RLlib 代码中打断点示例默认以 PPO 训练约 200 迭代、200 万步预期终端会输出类似episode_return_mean ≈ 458.68的训练结果结束时哑客户端会因服务端主动关闭 socket 而抛出ConnectionError属正常现象。哑客户端外部模拟器的实现模板内置哑客户端 _dummy_external_client.py 是外部应用接入 RLlink 协议的完整模板其流程为重试连接localhost:port→ 发送PING并断言收到PONG→ 发送GET_CONFIG用返回的配置构建本地RLModule→ 发送GET_STATE加载初始权重 → 进入环境循环用rl_module.forward_exploration本地推理出动作分布按 softmax 概率采样动作env.step推进仿真并用episode.add_env_step记录含ACTION_DIST_INPUTS与ACTION_LOGP等模型输出当累计步数达到config.get_rollout_fragment_length()时发送EPISODES_AND_GET_STATE并阻塞等待SET_STATE更新权重如此循环实现同步 on-policy 训练episode 结束后调用episode.cut()截断并开启新 episode。自定义 EnvRunner 与扩展方向RLlink 只是一个消息协议接入外部环境并不局限于 TCP。官方文档明确说明你可以自定义EnvRunner子类来改变底层通信机制例如用共享内存取代 TCP 实现更低延迟的通信层。参考文档 external.rst 指向的ray.rllib.env.external.rllink模块rllib/env/external/init.py 的公开导出正是这类自定义实现的协议基座旧模块ray.rllib.env.utils.external_env_protocol已发出弃用警告指向新位置见 external_env_protocol.py。从源码结构看外部环境接入仍处于新 API 栈的进行时状态RLlib 服务端暂不支持逐动作请求、服务端推理模式官方正为自定义 EnvRunner 与游戏引擎等仿真软件的非 Python 客户端适配器开发更多示例同时协议本身也被定位为初始草案未来预期引入安全层与压缩。安全提示与使用限制最后务必注意两点均出自官方文档与源码协议明文、不加密RLlink 前几个版本使用 msgpack 编码但无加密、不安全不应在不可信网络上传输敏感数据pickle 反序列化风险SET_CONFIG的config字段为 pickle 序列化载荷客户端执行pickle.loads()反序列化因此只能连接可信的服务端否则存在任意代码执行风险。综合来看RLlink 为外部模拟器 RLlib提供了一条轻量、简单、可自定义传输层的接入路径其协议基座、服务端参考实现与完整示例代码均可在当前仓库中直接查阅与复现。【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考