Two runnable lanes with visible artifacts: a closed-loop simulator where the
environment and policy are both Flows, and a graph renderer that validates
the runtime IR and writes an inspectable HTML view. Both run without a real
simulator or GPU.
The Lift task is a feedback loop: the environment produces a LiftState, the
policy consumes it and returns a LiftAction, which feeds back into the
environment. LiftEnvFlow runs a deterministic mock so the smoke test needs no
robosuite install.
from retriever.flow import Flow, Latest, Pipeline, Rate, Trigger, io
@io
class LiftAction:
dz: float | None = None
grip: float | None = None
@io
class LiftState:
step: int | None = None
source: str | None = None
object_height: float | None = None
gripper_z: float | None = None
reward: float | None = None
done: bool | None = None
class LiftEnvFlow(Flow[LiftAction, LiftState]):
"""robosuite Lift, or a deterministic mock when robosuite is absent."""
def step(self, action: LiftAction | None) -> LiftState:
self.step_idx += 1
dz = 0.0 if action is None or action.dz is None else float(action.dz)
grip = 1.0 if action is None or action.grip is None else float(action.grip)
return self._step_mock(dz=dz, grip=grip) # or _step_robosuite in real mode
class HeuristicLiftPolicy(Flow[LiftState, LiftAction]):
"""Approach, close the gripper, then lift."""
def step(self, state: LiftState | None) -> LiftAction:
if state is None or state.object_height is None:
return LiftAction(dz=-0.4, grip=1.0)
if state.done or state.object_height >= self.target_height:
return LiftAction(dz=0.0, grip=-1.0)
if state.gripper_z > state.object_height + 0.08:
return LiftAction(dz=-0.5, grip=1.0)
return LiftAction(dz=0.6, grip=-1.0)
The feedback edge is explicit: env → policy and policy → env. The environment
ticks at env_hz, the policy at policy_hz, and the printer wakes on each new
step.
pipe = Pipeline("robosuite_lift_demo")
with pipe:
env = LiftEnvFlow(mode="mock", ...) @ Rate(hz=20)
policy = HeuristicLiftPolicy(target_height=1.05) @ Rate(hz=5)
printer = LiftPrinter(print_every=2) @ Trigger("step")
pipe.connect(env, policy, sync=Latest())
pipe.connect(policy, env, sync=Latest()) # feedback loop
pipe.connect(env, printer, sync=Latest())
pixi run demo-robosuite-mock
Real output — the mock is deterministic, so object_z rises once the gripper
closes near the block and reward tracks the lift:
[mock step=002] object_z=0.820 gripper_z=0.940 reward=0.000 done=False
[mock step=004] object_z=0.820 gripper_z=0.900 reward=0.000 done=False
[mock step=006] object_z=0.841 gripper_z=0.904 reward=0.021 done=False
[mock step=008] object_z=0.883 gripper_z=0.952 reward=0.063 done=False
[mock step=010] object_z=0.925 gripper_z=1.000 reward=0.105 done=False
[mock step=012] object_z=0.967 gripper_z=1.048 reward=0.147 done=False
Swap mode="robosuite" (after installing the optional dependency) and the same
graph drives the real simulator; LiftState.source reports which backend ran.
pipe.validate() returns the runtime IR; generate_ascii_graph and
save_interactive_html turn it into an inspectable graph — including feedback
edges, clocks, and sync policies.
from retriever.flow import Latest, Pipeline, Rate, Trigger, io
from retriever.ir.viz import generate_ascii_graph, save_interactive_html
pipe = Pipeline("visualization_demo")
env = DummyNode() @ Rate(hz=10.0)
perception = DummyNode() @ Trigger("obs")
planner = DummyNode() @ Trigger("state")
executor = DummyNode() @ Trigger("plan", "state")
pipe.connect(env, perception, map={"data": "data"}, sync=Latest())
pipe.connect(perception, planner, map={"data": "data"}, sync=Latest())
pipe.connect(planner, executor, map={"data": "data"}, sync=Latest())
pipe.connect(perception, executor, map={"data": "data"}, sync=Latest())
pipe.connect(executor, env, map={"data": "data"}, sync=Latest()) # closes the loop
ir = pipe.validate()
print(generate_ascii_graph(ir))
save_interactive_html(ir, "out/golden_retriever_closed_loop_viz.html")
pixi run demo-pipeline-html-viz
Real output — the ASCII graph shows each node’s clock and the detected cycle, and
the command writes a self-contained HTML artifact:
Pipeline: visualization_demo
============================
NOTE: Cycle detected in graph topology (expected for closed-loop systems).
[DummyNode] <Rate(10.0 Hz)>
--(data -> data: Latest)--> [DummyNode]
[DummyNode] <Trigger(on ['obs'])>
--(data -> data: Latest)--> [DummyNode]
--(data -> _fanin/DummyNode__2/data: Latest)--> [DummyNode]
[DummyNode] <Trigger(on ['state'])>
--(data -> _fanin/DummyNode__3/data: Latest)--> [DummyNode]
[DummyNode] <Trigger(on ['plan', 'state'])>
--(data -> data: Latest)--> [DummyNode] (feedback/cycle)
HTML visualization written to out/golden_retriever_closed_loop_viz.html
The generated graph, embedded from the artifact:
Run the visual lane from lightest to heaviest so each layer is verified before the next:
- HTML graph proof —
pixi run demo-pipeline-html-viz renders the graph structure with no simulator or model.
- Mock robosuite —
pixi run demo-robosuite-mock runs the env↔policy loop deterministically, without a robosuite install.
- Real robosuite / Rerun — swap
mode="robosuite" (optional dependency) only after the mock trace looks right.
Making the environment and policy ordinary Flows keeps timing and handoffs
visible in the graph instead of hidden in callbacks: the feedback edge, the two
clock rates, and the sync policies are all inspectable IR. That is why the same
graph renders to ASCII/HTML and runs against either a mock or a real backend.
Heavier visual lanes reuse the same runtime once their dependencies are present:
pixi run -e twist2 demo-twist2-rerun # MuJoCo + Rerun viewer (optional, heavy)
Source: examples/advanced/robosuite_lift/app.py and
examples/experimental/visualization/visualize_pipeline.py.
