剑桥大学等开源 RLgraph 框架，让强化学习测试可视化

from rlgraph.agents import agentsfrom rlgraph.spaces import *
# Describe inputs via powerful space objects which manage batch and time ranks.states = IntBox(  low=0,  high=1024,  add_time_rank=True)
# Nested container spaces are used throughout RLgraph# to describe state and perform shape inference.actions = Dict(  int_action=IntBox(low=0, high=5),  bool_action=BoolBox(),  float_action=FloatBox(shape=(3,))
# Create PPO agentagent = Agent.from_spec(    agent_config,    state_space=env.state_space,    action_space=env.action_space)
# Use agent locally, control behaviour with flags.actions = agent.get_actions(states, use_exploration=use_exploration, apply_preprocessing=True)
# Batch observe multi-environment settings.agent.observe(states, actions, rewards, terminals, batched=True , env_id="env_3")
# Updates by sampling from buffer after observing.loss = agent.update()
# Update from an external batch which may contain # arbitrary (non-terminal) sub-episode fragments from multiple environments,# identified via sequence indices:agent.update(batch=dict(states=states, actions=actions, rewards=rewards, terminals=terminals, sequence_indices=sequence_indices)
# Single-threaded, vectorized execution. env_spec = {"type": "openai", "env_id": "CartPole-v0"}worker = SingleThreadedWorker(env_spec=env_spec, agent=agent, num_envs=8)stats = worker.execute_timesteps(100000)
# Go large scale with Ray in 2 lines.ray_executor = SyncBatchExecutor(agent_config, env_spec)ray_stats = ray_executor.execute(steps=100000)
# Use agent trained on Ray just as before.agent = ray_executor.local_agent

record_space = Dict(    states=dict(state1=float, state2=float),    actions=dict(action1=float, action2=IntBox(10)),    # For scalar spaces, use Python primitives    reward=float,    terminals=BoolBox(),    next_states=dict(state1=float, state2=float),    add_batch_rank=True)
# Memory exposes insert, sample..methodsmemory = ReplayMemory(capacity=1000) 
# Input spaces contain spaces for all arguments.input_spaces = dict(records=record_space, num_records=int)
# Builds the memory with variables, placeholders for these spaces.graph = ComponentTest(component=memory, input_spaces=input_spaces)
# Generate a sample batch from the nested space.observation = record_space.sample(size=32)
# Calls session, fetches ops, prepares inputs, executes API method.graph.test(memory.insert_records, observation))
# Get some samples providing the "num_records" int-arg.samples = graph.test((memory.insert_records, 10))