Documentation: https://uncore-team.github.io/rl_spin_decoupler/
Example code: tutorial | complete executable example | RL skeleton | Agent skeleton
This is a simple Python module that allows to sync an RL algorithm (e.g., from Stable-Baselines3) to an agent, physical or simulated, that is able to get observations and execute actions.
Usually, this would not be necessary since in many cases you can implement in the same environment class all what is needed for both RL and the agent, but there are scenarios where the agent needs to execute some spin loop, in its own thread, while RL executes its own. Both loops may be quite difficult to put in sync. RL Spin Decoupler is intended for those cases.
Its use is pretty simple: you will have a Python program running RL and another one -a different process- running the agent. The former will use the RLSide class of this module, and the latter the AgentSide class in order to communicate to each other (communications are implemented with sockets). You can find further explanations in the code about when and how to call the methods of these classes in order to sync both processes.
In addition, two files called 'skeleton_...' contain incomplete implementations of a case of use of the decoupler, and examples/ contains a fully executable end-to-end demo.
Besides facilitating the link between different processes that carry out RL and agent simulation/control, this decoupler may be useful as well when different timings are involved and matter (for instance, if the agent must execute some action at a given time, while the RL algorithm has no notion of that).
Although RL Spin Decoupler has been implemented and tested with Stable Baselines3 in mind, it is quite general and could be used with other RL libraries: it assumes that there is a step() method at each RL step and a reset() method when an episode starts. The parameters and results of those methods are the same that Stable Baselines3 uses.
Why RL Spin Decoupler?
-
Non-invasive integration. The RL side exposes a Gymnasium-like interface (
reset()/step()), so you keep using your learning stack (policy networks, replay buffers, optimizers) unchanged. The library is validated with Stable-Baselines3 and applies to any RL code that follows the samestep/resetcontract. -
No shared clock, no blocking. The RL process and the agent's spin loop run as separate OS processes communicating over TCP sockets. The agent polls asynchronously (non-blocking) and keeps running at its own frequency instead of stalling while the policy computes.
-
Timing made observable. Every step reports the actual Last Action Time (LAT) — the wall-clock duration of the executed action — alongside the agent's own timestamp (ATO). This exposes latencies that are invisible in a monolithic setup and lets you build time-aware training loops on top.
-
Separation of concerns. Control logic and learning logic live in different processes and can be developed and run independently — even on different machines, so a GPU host can drive learning while the control loop stays on embedded hardware.
-
Learn from the examples. Three self-contained, runnable examples ship with the repo — a first-order plant controller, the Gymnasium LunarLander benchmark, and skeleton templates for both sides — covering the range from a minimal software-only demo to a standard RL benchmark.
The core library uses only Python standard library modules and does not require third-party packages.
External RL packages (for example Stable-Baselines3 or Gymnasium) are optional and only needed for user-side training scripts built on top of this library.
- Main examples index: examples/README.md
- First-order plant control example: examples/first_order_plant_control/
- LunarLander SB3 decoupled example: examples/lunar_lander/
The first-order plant example is the lightweight baseline demo.
The LunarLander example demonstrates the decoupled pattern where the agent process only transports observations/timing, while reward and episode termination logic are computed on the RL side.
- Python 3.8 or newer.
- No mandatory third-party dependencies for the core communication library.
Clone the repository and move into the project folder:
git clone https://github.com/uncore-team/rl_spin_decoupler.git
cd rl_spin_decouplerOptional: if you want to install optional RL packages listed in requirements.txt:
pip install -r requirements.txtEditable install for development:
pip install -e .Developer verification
Run the test suite with coverage:
pip install -e ".[dev]"
pytestBuild the documentation locally:
pip install -e ".[docs]"
python -m sphinx -b html docs docs/_build/htmlThe generated documentation home page is docs/_build/html/index.html.
Run lint and formatting checks locally:
python -m ruff check .
python -m ruff format --check .If you are new to GitHub, you can get this library in two simple ways:
- With
git(recommended)
git clone https://github.com/uncore-team/rl_spin_decoupler.git
cd rl_spin_decoupler- Without
git(download ZIP)
- Open the repository page in GitHub.
- Click the green
Codebutton. - Select
Download ZIP. - Extract the ZIP file to a local folder.
After downloading, open the project folder in your editor and start using spindecoupler.py and the skeleton files as a base.
- Implement your agent and RL wrappers from the provided skeleton files.
- Start the RL side process first.
- Start the agent side process second.
- Run learning and close communications with
stepExpFinished()when finished.
These templates illustrate a typical decoupled setup: the RL process drives training at a lower frequency, while the agent process runs a faster control loop and exchanges observations/actions through the communication wrappers.
This skeleton wraps the RL-facing communication API in a Gym-like environment, so you can plug your learning loop with minimal glue code.
""" RL side skeleton. Orchestrates the learning process. """
# imports
...
import time
from typing import Optional
from spindecoupler import RLSide # RL side comms wrapper
class RLEnv:
"""Minimal Gym-like wrapper around the decoupled communication API."""
def __init__(self, debug: bool = False):
self._debug = debug
self._commstoagent = RLSide(49054, verbose=debug) # blocks until agent connects
def resetGetObs(self):
"""Paper-level API name kept explicitly in the skeleton."""
obs0, t_agent = self._commstoagent.resetGetObs()
return obs0, t_agent
def stepSendActGetObs(self, action):
"""Paper-level tuple: (o_{t+1}, t_agent, t_wall, LAT)"""
lat, obs_next, _rew_from_agent, t_agent = self._commstoagent.stepSendActGetObs(action)
t_wall = time.time()
return obs_next, t_agent, t_wall, lat
def stepExpFinished(self):
"""Graceful end of experiment/socket lifecycle."""
self._commstoagent.stepExpFinished()
# -- Gym-like API --
def reset(self, seed: Optional[int] = None, options: Optional[dict] = None):
obs0, t_agent = self.resetGetObs()
info = {"t_agent": t_agent}
if self._debug:
print(f"reset -> t_agent={t_agent}")
return obs0, info
def step(self, action):
o_t1, t_agent, t_wall, lat = self.stepSendActGetObs(action)
reward = self._compute_reward(o_t1, action, lat)
terminated = self._is_terminal(o_t1)
truncated = False
info = {"t_agent": t_agent, "t_wall": t_wall, "lat": lat}
return o_t1, reward, terminated, truncated, info
# -- User hooks --
def _compute_reward(self, obs, action, lat: float) -> float:
return ...
def _is_terminal(self, obs) -> bool:
return ...
# -- Entry point --
if __name__ == "__main__":
print("Learning...")
env = RLEnv(debug=True)
model = ...
numstepsexp = 1_000
model.learn(total_timesteps=numstepsexp)
model.save(...)
env.stepExpFinished()This skeleton models a high-frequency control loop that polls RL commands asynchronously and reports observations at the configured RL timestep.
Use it as a starting point: keep the communication calls as shown, and customize the user hooks (_compute_reward, _is_terminal, _build_observation, _apply_action, _reset_workspace, _null_action) for your specific robot or simulator.
""" Agent side skeleton. Manages the high-frequency spin loop. """
# imports
...
import time
from enum import Enum
from typing import Any
from spindecoupler import AgentSide, BaseCommPoint # Agent side comms wrapper
class Agent:
"""Robot/simulator control loop that talks to the RL process."""
class StepState(Enum):
"""States of the agent during its step() execution."""
READYFORRLCOMMAND = 0 # Waiting for new RL command
EXECUTINGLASTACTION = 1 # Executing the last received action
AFTERRESET = 2 # Just reset; must send observation
def __init__(self, debug: bool = False) -> None:
self._debug = debug
self._rltimestep = 0.1 # seconds between RL actions (must be > control timestep)
self._control_timestep = 0.02 # seconds of local control loop
if self._rltimestep <= self._control_timestep:
raise ValueError("RL timestep must be > control timestep")
self._stepstate = Agent.StepState.READYFORRLCOMMAND
self._lastaction = self._null_action()
self._lastactiont0 = time.time()
self._commstoRL = AgentSide(BaseCommPoint.get_ip(), 49054, verbose=debug)
def readWhatToDo(self):
"""Paper-level API: poll for commands from RL side."""
return self._commstoRL.readWhatToDo()
def stepSendLastActDur(self, lat: float) -> None:
"""Paper-level API: report actual execution time of previous action."""
self._commstoRL.stepSendLastActDur(lat)
def step(self) -> Any:
"""Main agent tick: manages state transitions and communication."""
now_wall = time.time()
act = self._lastaction
if self._stepstate == Agent.StepState.EXECUTINGLASTACTION:
# Check if action duration threshold reached; send observation if so
if now_wall - self._lastactiont0 >= self._rltimestep:
observation = self._build_observation()
self._commstoRL.stepSendObs(observation, agenttime=now_wall)
self._stepstate = Agent.StepState.READYFORRLCOMMAND
elif self._stepstate == Agent.StepState.READYFORRLCOMMAND:
# Poll for new commands from RL
whattodo = self.readWhatToDo()
if whattodo is not None:
what, payload = whattodo
if what == AgentSide.WhatToDo.REC_ACTION_SEND_OBS:
# Receive action and report LAT of previous action
lat = now_wall - self._lastactiont0
self.stepSendLastActDur(lat)
self._lastactiont0 = now_wall
self._lastaction = payload
self._stepstate = Agent.StepState.EXECUTINGLASTACTION
elif what == AgentSide.WhatToDo.RESET_SEND_OBS:
# Reset episode and prepare to send observation
self._reset_workspace()
act = self._null_action()
self._lastactiont0 = now_wall
self._stepstate = Agent.StepState.AFTERRESET
elif what == AgentSide.WhatToDo.FINISH:
raise RuntimeError("Experiment finished")
elif self._stepstate == Agent.StepState.AFTERRESET:
# Send observation after reset
observation = self._build_observation()
self._commstoRL.resetSendObs(observation, agenttime=now_wall)
self._stepstate = Agent.StepState.READYFORRLCOMMAND
self._apply_action(self._lastaction)
return self._lastaction
def spinloop(self) -> None:
"""Typical control loop."""
while True:
self.step()
time.sleep(self._control_timestep)
# -- user hooks --
def _build_observation(self):
return ... # dict-like observation
def _apply_action(self, action: Any) -> None:
... # send action to low-level controller
def _reset_workspace(self) -> None:
... # reset robot/simulation state
def _null_action(self):
return ...
# -- Entry point --
if __name__ == "__main__":
agent = Agent(debug=True)
agent.spinloop()Contributions are welcome through pull requests. Please:
- Create a feature branch from
main. - Keep changes focused and documented.
- Open a pull request with a clear description of the motivation and changes.
This project is licensed under the GNU General Public License v3.0. See LICENSE.