Reproducing a Memory-Augmented CartPole Policy

No file to download here -- pip install gymnasium gives you a deterministic, freely available simulation (CartPole-v1) that produces a real state/action/outcome stream on your own machine. What's checkable: whether querying past episodes for similar states, and biasing action choice toward what succeeded before, beats a random policy -- and whether the SDK's retrieval agrees with an independent nearest-neighbor computed over the same logged data.

Requirements: a VAAS-X API key, pip install vaas-x gymnasium pandas.

Baseline: random policy

import gymnasium as gym

env = gym.make("CartPole-v1")

def random_episode():
    obs, _ = env.reset()
    steps = 0
    while True:
        action = env.action_space.sample()
        obs, _, terminated, truncated, _ = env.step(action)
        steps += 1
        if terminated or truncated:
            return steps

baseline = [random_episode() for _ in range(50)]
print("random policy, mean steps:", sum(baseline) / len(baseline))

Log and ingest a memory

Bucket each observation by sign so it collapses to a short, repeatable state description, and put the action taken into source_text so it comes back out of retrieval later.

from vaasx import Bootstrap

brain = Bootstrap(api_key="YOUR_API_KEY", device_id="cartpole")

def bucket(obs):
    labels = ["cart_left", "cart_right"], ["cart_still", "cart_moving_right"], \
              ["pole_left", "pole_right"], ["pole_falling_left", "pole_falling_right"]
    return " ".join(labels[i][int(v > 0)] for i, v in enumerate(obs))

log = []  # kept locally too, for the independent check later
for _ in range(200):
    obs, _ = env.reset()
    ep_items, ep_states, ep_actions = [], [], []
    while True:
        action = env.action_space.sample()
        state_text = bucket(obs)
        ep_items.append({
            "payload": {f"obs_{i}": float(v) for i, v in enumerate(obs)},
            "source_text": f"{state_text}; action_taken={action}",
        })
        ep_states.append(state_text)
        ep_actions.append(action)
        obs, _, terminated, truncated, _ = env.step(action)
        if terminated or truncated:
            break
    resp = brain.ingest(ep_items)
    for i, episode_id in enumerate(resp["episode_ids"]):
        if episode_id is None:
            continue
        # every step kept the pole up except the last one in the episode
        success = i < len(resp["episode_ids"]) - 1
        brain.outcome(episode_id, success=success)
        log.append({"state": ep_states[i], "action": ep_actions[i], "success": success})

Query it for an action

import re

def recall_action(obs, k=5):
    hits = brain.query(bucket(obs), k=k, prefer_success=True)
    votes = [int(m.group(1)) for h in hits if (m := re.search(r"action_taken=(\d)", h["text"]))]
    if len(votes) < 2:
        return env.action_space.sample()
    return max(set(votes), key=votes.count)

Evaluate

def memory_episode():
    obs, _ = env.reset()
    steps = 0
    while True:
        action = recall_action(obs)
        obs, _, terminated, truncated, _ = env.step(action)
        steps += 1
        if terminated or truncated:
            return steps

memory_run = [memory_episode() for _ in range(50)]
print("memory-augmented, mean steps:", sum(memory_run) / len(memory_run))

CartPole is noisy over 50 episodes -- run more if the gap over the random baseline isn't clear.

Verify independently

import pandas as pd

local = pd.DataFrame(log)
local_pref = local.groupby("state").apply(lambda g: g.loc[g["success"], "action"].mode())
print(local_pref)

For a handful of states, compare local_pref against what recall_action actually returns -- it's a majority vote over the same logged outcomes, so the two should agree.

More guides: Reproduction Guides. Full reference: Developer Guide.