企业官网建设流程全解析

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 专用通信协议

RLlink（ray.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 字节，因此完整帧为：

b"00000011" + b"\x81\xa4type\xa4PING"

服务端应答的PONG帧结构相同：

b"00000011" + b"\x81\xa4type\xa4PONG"

帧的读写实现（源码级）

帧的组装与解析实现在 rllib/env/external/rllink.py 的两个函数中，这也是参考文档 autosummary 中列出的两个公开函数：

• send_rllink_message(sock_, message)：先通过msgpack.packb(message, use_bin_type=True)序列化消息体，再用str(len(body)).zfill(8).encode("utf-8")生成 8 位左补零的长度头，最后sock_.sendall(header + body)一次性发送；
• get_rllink_message(sock_)：先读取 8 字节长度头，再按长度读取消息体，msgpack.unpackb(body, raw=False)反序列化后，强制校验消息字典中必须包含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 文档整理。

请求：客户端 → 服务端

EPISODES_AND_GET_STATE的完整发送示例（来自 external-envs.md）：

send_rllink_message( sock, { "type": "EPISODES_AND_GET_STATE", # 每个元素是一个 `SingleAgentEpisode.get_state()` 字典。 "episodes": [episode.get_state() for episode in episodes], "timesteps": 128, }, )

响应：服务端 → 客户端

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 中的实际客户端代码一一对应：

1. 初始握手：客户端发送PING，服务端应答PONG；
2. 配置请求：客户端发送GET_CONFIG，服务端应答SET_CONFIG，客户端用配置构建本地RLModule，并从中读取get_rollout_fragment_length()决定每批采集多少步；
3. 初始权重请求：客户端发送GET_STATE，服务端应答SET_STATE，客户端将权重加载进本地RLModule；
4. On-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→ 回PONG；EPISODES/EPISODES_AND_GET_STATE→ 调用_process_episodes_message用SingleAgentEpisode.from_state()重建并to_numpy()转成 numpy 后缓存；GET_STATE→ 回SET_STATE；GET_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_cls=EnvRunnerServerForExternalInference)将默认 EnvRunner 替换为外部推理服务端：

base_config = ( get_trainable_cls(args.algo) .get_default_config() .environment( observation_space=gym.spaces.Box(float("-inf"), float("-inf"), (4,), np.float32), action_space=gym.spaces.Discrete(2), # EnvRunners 监听 `port` + worker_index 端口。 env_config={"port": args.port}, ) .env_runners( # 指向自定义 EnvRunner。 env_runner_cls=EnvRunnerServerForExternalInference, ) .training(num_epochs=10, vf_loss_coeff=0.01) .rl_module(model_config=DefaultModelConfig(vf_share_layers=True)) )

运行方式

python rllib/examples/envs/env_connecting_to_rllib_w_tcp_client.py --port 5555 --use-dummy-client

• --port：RLlib EnvRunner 的监听端口（默认 5555），客户端侧需保持一致；
• --use-dummy-client：启动内置的哑客户端（模拟器）线程；不带该参数时，可自行从 C++ 应用等外部程序连接；
• 调试时可追加--no-tune --num-env-runners=0，便于在 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 客户端适配器开发更多示例；同时协议本身也被定位为"初始草案"，未来预期引入安全层与压缩。

安全提示与使用限制

最后务必注意两点（均出自官方文档与源码）：

1. 协议明文、不加密：RLlink 前几个版本使用 msgpack 编码但无加密、不安全，不应在不可信网络上传输敏感数据；
2. 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