From 311f89c20f5cc363b3ca38340680282afd2d70e8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Alexandre=20C=C3=B4t=C3=A9?= Date: Mon, 24 Oct 2022 16:31:02 -0400 Subject: [PATCH 1/2] Use ScienceWorld v1.1 --- README.md | 22 ++++---- drrn/closeLeftoverEnvs.py | 18 ------- drrn/train-scienceworld.py | 56 ++++++++++---------- drrn/vec_env.py | 101 +++++++++++++++++-------------------- 4 files changed, 83 insertions(+), 114 deletions(-) delete mode 100644 drrn/closeLeftoverEnvs.py diff --git a/README.md b/README.md index 30d6ce5..bf5336c 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # DRRN Agent (Modified for ScienceWorld) -This repository contains a reference implementation DRRN as mentioned in [Interactive Fiction Games: A Colossal Adventure](https://arxiv.org/abs/1909.05398), that has been modified for use with the [ScienceWorld](https://www.github.com/allenai/ScienceWorld) environment. +This repository contains a reference implementation DRRN as mentioned in [Interactive Fiction Games: A Colossal Adventure](https://arxiv.org/abs/1909.05398), that has been modified for use with the [ScienceWorld](https://www.github.com/allenai/ScienceWorld) environment. # Quickstart @@ -12,20 +12,20 @@ git clone https://github.com/cognitiveailab/drrn-scienceworld.git cd drrn-scienceworld # Create conda environment -conda create --name drrn1 python=3.8 -conda activate drrn1 +conda create --name drrn-scienceworld python=3.8 +conda activate drrn-scienceworld pip install -r requirements.txt ``` -An example of training the DRRN model (using 8 threads, for 10k training steps, evaluating on dev every 1k steps): +An example of training the DRRN model (using 8 parallel envs, for 10k training steps, evaluating on dev every 1k steps): ```bash cd drrn -python3 train-scienceworld.py --num_envs=8 --max_steps=10000 --task_idx=13 --simplification_str=easy --priority_fraction=0.50 --memory_size=100000 --env_step_limit=100 --eval_freq=1000 --eval_set=dev --historySavePrefix=drrn-task13-results-seed0-dev +python train-scienceworld.py --num_envs=8 --max_steps=10000 --task_idx=13 --simplification_str=easy --priority_fraction=0.50 --memory_size=100000 --env_step_limit=100 --eval_freq=1000 --eval_set=dev --historySavePrefix=drrn-task13-results-seed0-dev ``` Here: -- **max_steps:** Maximum number of steps to train for (per environment thread) -- **num_envs:** The number of environment threads to simultaneously use during training (8 is a common number) +- **max_steps:** Maximum number of steps to train for (per environment) +- **num_envs:** The number of environments to simultaneously use during training (8 is a common number) - **task_idx:** The ScienceWorld task index (0-29). *See **task list** below* - **env_step_limit:** the maximum number of steps to run an environment for, before it times out and resets (100 typical) - **eval_freq:** the number of steps between evaluations @@ -37,7 +37,7 @@ This configuration generally takes about 1-2 hours to run (to 10k steps). ## ScienceWorld Task List ``` -TASK LIST: +TASK LIST: 0: task-1-boil (30 variations) 1: task-1-change-the-state-of-matter-of (30 variations) 2: task-1-freeze (30 variations) @@ -71,15 +71,13 @@ TASK LIST: ``` # Hardware requirements -This code generally runs best with at least num_threads+1 CPU cores (e.g. about 10 cores for an 8-thread environment). +This code generally runs best with at least num_envs+1 CPU cores. -The GPU memory requirements are variable, but generally stay below 8gb. +The GPU memory requirements are variable, but generally stay below 8gb. # Known issues -- *Many threads*: If you are attempting to use a large number of threads (e.g. 20+), you may need to add an additional several-second delay after the threads spawn before the rest of the program runs. (The ScienceWorld API already adds a 5 second delay, which handles small numbers of threads well.) - - *Model saving with manys steps*: Very occassionally, on very long runs (generally 1M+ steps), the periodic pickling the model when saving checkpoints runs into issues and freezes. The cause is unknown, but as a workaround the save has been wrapped in a timeout, so that if it takes longer than 2 minutes to save the model, the checkpoint is not saved and training continues. Subsequent checkpoints usually save without issue. diff --git a/drrn/closeLeftoverEnvs.py b/drrn/closeLeftoverEnvs.py deleted file mode 100644 index e671bc9..0000000 --- a/drrn/closeLeftoverEnvs.py +++ /dev/null @@ -1,18 +0,0 @@ -# -# If ScienceWorld environments have been left open (from a crash, etc), -# this will send close signals to any that exist across a large port range. -# - -from scienceworld import ScienceWorldEnv - -MAX_THREADS = 200 - -for threadNum in range(0, MAX_THREADS): - print("Thread num: " + str(threadNum) + " / " + str(MAX_THREADS)) - try: - env = ScienceWorldEnv("", None, 10, threadNum, launchServer=False) - env.shutdown() - except: - print("\tNo Server Found") - - diff --git a/drrn/train-scienceworld.py b/drrn/train-scienceworld.py index c706bbb..416a180 100644 --- a/drrn/train-scienceworld.py +++ b/drrn/train-scienceworld.py @@ -29,17 +29,17 @@ def clean(strIn): return strIn.strip() -def evaluate(agent, args, env_step_limit, bufferedHistorySaverEval, extraSaveInfo, nb_episodes=10): - # Initialize a ScienceWorld thread for serial evaluation - env = initializeEnv(threadNum = args.num_envs+10, args=args) # A threadNum (and therefore port) that shouldn't be used by any of the regular training workers +def evaluate(agent, args, env_step_limit, bufferedHistorySaverEval, extraSaveInfo, nb_episodes=10): + # Initialize a ScienceWorld server for serial evaluation + env = initializeEnv(args=args) scoresOut = [] with torch.no_grad(): - + for ep in range(nb_episodes): total_score = 0 - log("Starting evaluation episode {}".format(ep)) - print("Starting evaluation episode " + str(ep) + " / " + str(nb_episodes)) + log("Starting evaluation episode {}".format(ep)) + print("Starting evaluation episode " + str(ep) + " / " + str(nb_episodes)) extraSaveInfo['evalIdx'] = ep score = evaluate_episode(agent, env, env_step_limit, args.simplification_str, bufferedHistorySaverEval, extraSaveInfo, args.eval_set) log("Evaluation episode {} ended with score {}\n\n".format(ep, score)) @@ -48,12 +48,10 @@ def evaluate(agent, args, env_step_limit, bufferedHistorySaverEval, extraSaveInf print("") avg_score = total_score / nb_episodes - - env.shutdown() - + return scoresOut, avg_score - + def evaluate_episode(agent, env, env_step_limit, simplificationStr, bufferedHistorySaverEval, extraSaveInfo, evalSet): @@ -74,22 +72,20 @@ def evaluate_episode(agent, env, env_step_limit, simplificationStr, bufferedHist else: print("evaluate_episode: unknown evaluation set (expected 'dev' or 'test', found: " + str(evalSet) + ")") - env.shutdown() - exit(1) state = agent.build_state([ob], [info])[0] - log('Obs{}: {} Inv: {} Desc: {}'.format(step, clean(ob), clean(info['inv']), clean(info['look']))) + log('Obs{}: {} Inv: {} Desc: {}'.format(step, clean(ob), clean(info['inv']), clean(info['look']))) while not done: - #print("numSteps: " + str(numSteps)) + #print("numSteps: " + str(numSteps)) valid_acts = info['valid'] - valid_ids = agent.encode(valid_acts) - _, action_idx, action_values = agent.act([state], [valid_ids], sample=False) + valid_ids = agent.encode(valid_acts) + _, action_idx, action_values = agent.act([state], [valid_ids], sample=False) action_idx = action_idx[0] action_values = action_values[0] action_str = valid_acts[action_idx] - log('Action{}: {}, Q-Value {:.2f}'.format(step, action_str, action_values[action_idx].item())) + log('Action{}: {}, Q-Value {:.2f}'.format(step, action_str, action_values[action_idx].item())) s = '' maxToDisplay = 10 # Max Q values to display, to limit the log size @@ -105,16 +101,16 @@ def evaluate_episode(agent, env, env_step_limit, simplificationStr, bufferedHist info = sanitizeInfo(info) ob = sanitizeObservation(ob, info) - - log("Reward{}: {}, Score {}, Done {}".format(step, rew, info['score'], done)) + + log("Reward{}: {}, Score {}, Done {}".format(step, rew, info['score'], done)) step += 1 log('Obs{}: {} Inv: {} Desc: {}'.format(step, clean(ob), clean(info['inv']), clean(info['look']))) - state = agent.build_state([ob], [info])[0] + state = agent.build_state([ob], [info])[0] - numSteps +=1 + numSteps +=1 if (numSteps > env_step_limit): print("Maximum number of evaluation steps reached (" + str(env_step_limit) + ").") - break + break print("Completed one evaluation episode") # Save @@ -158,7 +154,7 @@ def train(agent, envs, max_steps, update_freq, eval_freq, checkpoint_freq, log_f # Choose action(s) action_ids, action_idxs, _ = agent.act(states, valid_ids) - action_strs = [info['valid'][idx] for info, idx in zip(infos, action_idxs)] + action_strs = [info['valid'][idx] for info, idx in zip(infos, action_idxs)] # Perform the action(s) in the environment obs, rewards, dones, infos = envs.step(action_strs) @@ -179,14 +175,14 @@ def train(agent, envs, max_steps, update_freq, eval_freq, checkpoint_freq, log_f numEpisodes += 1 next_states = agent.build_state(obs, infos) - next_valids = [agent.encode(info['valid']) for info in infos] + next_valids = [agent.encode(info['valid']) for info in infos] for state, act, rew, next_state, valids, done in \ zip(states, action_ids, rewards, next_states, next_valids, dones): agent.observe(state, act, rew, next_state, valids, done) states = next_states valid_ids = next_valids - if step % log_freq == 0: + if step % log_freq == 0: tb.logkv('Step', step) tb.logkv('StepsFunctional', step*envs.num_envs) tb.logkv("FPS", int((step*envs.num_envs)/(time.time()-startTime))) @@ -203,8 +199,8 @@ def train(agent, envs, max_steps, update_freq, eval_freq, checkpoint_freq, log_f print("GPU_mem: " + str(agent.getMemoryUsage())) print("*************************") - if step % update_freq == 0: - loss = agent.update() + if step % update_freq == 0: + loss = agent.update() if loss is not None: tb.logkv_mean('Loss', loss) @@ -220,7 +216,7 @@ def train(agent, envs, max_steps, update_freq, eval_freq, checkpoint_freq, log_f # Do the evaluation procedure extraSaveInfo = {'numEpisodes':numEpisodes, 'numSteps':step, 'stepsFunctional:':stepsFunctional, 'maxHistoriesPerFile':args.maxHistoriesPerFile} eval_scores, avg_eval_score = evaluate(agent, args, args.env_step_limit, bufferedHistorySaverEval, extraSaveInfo) - + tb.logkv('EvalScore', avg_eval_score) tb.logkv('numEpisodes', numEpisodes) tb.dumpkvs() @@ -265,7 +261,7 @@ def parse_args(): parser.add_argument('--embedding_dim', default=128, type=int) parser.add_argument('--hidden_dim', default=128, type=int) - parser.add_argument('--task_idx', default=0, type=int) + parser.add_argument('--task_idx', default=0, type=int) parser.add_argument('--maxHistoriesPerFile', default=1000, type=int) parser.add_argument('--historySavePrefix', default='saveout', type=str) @@ -281,7 +277,7 @@ def main(): ## assert jericho.__version__ == '2.1.0', "This code is designed to be run with Jericho version 2.1.0." args = parse_args() print(args) - configure_logger(args.output_dir) + configure_logger(args.output_dir) agent = DRRN_Agent(args) # Initialize a threaded wrapper for the ScienceWorld environment diff --git a/drrn/vec_env.py b/drrn/vec_env.py index 1ff0ccd..6c355a5 100644 --- a/drrn/vec_env.py +++ b/drrn/vec_env.py @@ -3,7 +3,6 @@ import numpy as np import random -import time import sys # @@ -21,7 +20,7 @@ def sanitizeInfo(infoIn): 'reward': infoIn['reward'], 'score': infoIn['score'], 'look': infoIn['look'], - 'inv': infoIn['inv'], + 'inv': infoIn['inv'], 'valid': recastList, 'taskDesc': infoIn['taskDesc'] } @@ -36,43 +35,43 @@ def sanitizeObservation(obsIn, infoIn): # # Reset the environment (with a new randomly selected variation) # -def resetWithVariation(env, variationMin, variationMax, simplificationStr): - variationIdx = random.randrange(variationMin, variationMax) # train on range 0-20 - env.reset() +def resetWithVariation(env, variationMin, variationMax, simplificationStr): + variationIdx = random.randrange(variationMin, variationMax) # train on range 0-20 + env.reset() initialObs, initialDict = env.resetWithVariation(variationIdx, simplificationStr) print("Simplifications: " + env.getSimplificationsUsed() ) - + return initialObs, initialDict def resetWithVariationTrain(env, simplificationStr): variationIdx = env.getRandomVariationTrain() ## Random variation on train - env.reset() + env.reset() initialObs, initialDict = env.resetWithVariation(variationIdx, simplificationStr) print("Simplifications: " + env.getSimplificationsUsed() ) - return initialObs, initialDict + return initialObs, initialDict def resetWithVariationDev(env, simplificationStr): variationIdx = env.getRandomVariationDev() ## Random variation on dev - env.reset() + env.reset() initialObs, initialDict = env.resetWithVariation(variationIdx, simplificationStr) print("Simplifications: " + env.getSimplificationsUsed() ) - - return initialObs, initialDict -def resetWithVariationTest(env, simplificationStr): + return initialObs, initialDict + +def resetWithVariationTest(env, simplificationStr): variationIdx = env.getRandomVariationTest() ## Random variation on test - env.reset() + env.reset() initialObs, initialDict = env.resetWithVariation(variationIdx, simplificationStr) print("Simplifications: " + env.getSimplificationsUsed() ) - - return initialObs, initialDict + + return initialObs, initialDict # Initialize a ScienceWorld environment directly from the API -def initializeEnv(threadNum, args): - env = ScienceWorldEnv("", None, args.env_step_limit, threadNum) +def initializeEnv(args): + env = ScienceWorldEnv("", None, args.env_step_limit) - taskNames = env.getTaskNames() + taskNames = env.getTaskNames() taskName = taskNames[args.task_idx] # Just reset to variation 0, as another call (e.g. resetWithVariation...) will setup an appropriate variation (train/dev/test) @@ -87,68 +86,67 @@ def initializeEnv(threadNum, args): # # Worker # -def worker(remote, parent_remote, threadNum, args): +def worker(remote, parent_remote, args): parent_remote.close() - print ("------------------------------------ NEW (Thread " + str(threadNum) + ")") - # Create unique thread - # Note, it doesn't matter what the variation is initially initialized to -- we reset it to a proper variation # (train/dev/test) before use. - env = initializeEnv(threadNum = 100+threadNum, args=args) + # Create unique server instance + # Note, it doesn't matter what the variation is initially initialized to -- we reset it to a proper variation # (train/dev/test) before use. + env = initializeEnv(args=args) + port = env._gateway.gateway_parameters.port + print ("------------------------------------ NEW (Port: " + str(port) + ")") try: done = False while True: cmd, data = remote.recv() - if cmd == 'step': - if done: - # If the thread is done, reset it - print("Thread " + str(threadNum) + " is DONE -- resetting with new variation") + if cmd == 'step': + if done: + # If the server instance is done, reset it + print("Port: " + str(port) + " is DONE -- resetting with new variation") ob, info = resetWithVariationTrain(env, args.simplification_str) - print("Thread " + str(threadNum) + " reset complete") + print("Port: " + str(port) + " reset complete") reward = 0 done = False - else: + else: # Otherwise, complete one step ob, reward, done, info = env.step(data) - + # Sanitize the 'observation' and 'info' receieved from the API info = sanitizeInfo(info) ob = sanitizeObservation(ob, info) - # Make sure any stdout from this thread is printed to the console in a timely fashion + # Make sure any stdout from this server instance is printed to the console in a timely fashion sys.stdout.flush() if (done == True): - print("DONE -- SCORE " + str(info['score']) + " (Thread " + str(threadNum) + ")") + print("DONE -- SCORE " + str(info['score']) + " (Port " + str(port) + ")") # If we're done, store history in 'info' info['runHistory'] = env.getRunHistory() - + remote.send((ob, reward, done, info)) - + elif cmd == 'reset': - print ("------------------------------------ RESET (Thread " + str(threadNum) + ")") - ob, info = resetWithVariationTrain(env, args.simplification_str) + print ("------------------------------------ RESET (Port " + str(port) + ")") + ob, info = resetWithVariationTrain(env, args.simplification_str) info = sanitizeInfo(info) ob = sanitizeObservation(ob, info) - + remote.send((ob, info)) elif cmd == 'get_state': - print ("------------------------------------ GETSTATE (Thread " + str(threadNum) + ")") + print ("------------------------------------ GETSTATE (Port " + str(port) + ")") remote.send((env.env.get_state(), done)) - + elif cmd == 'set_state': - print ("------------------------------------ SETSTATE (Thread " + str(threadNum) + ")") + print ("------------------------------------ SETSTATE (Port " + str(port) + ")") done = data[1] - env.env.set_state(data[0]) + env.env.set_state(data[0]) remote.send(True) - + elif cmd == 'close': - print ("------------------------------------ CLOSE (Thread " + str(threadNum) + ")") - env.shutdown() # Shut down ScienceWorld server for this thread - time.sleep(2) + print ("------------------------------------ CLOSE (Port " + str(port) + ")") break else: @@ -157,9 +155,7 @@ def worker(remote, parent_remote, threadNum, args): except KeyboardInterrupt: print('SubprocVecEnv worker: got KeyboardInterrupt') finally: - print ("------------------------------------ SHUTDOWN (Thread " + str(threadNum) + ")") - env.shutdown() # Shut down ScienceWorld server for this thread - time.sleep(2) + print ("------------------------------------ SHUTDOWN (Port " + str(port) + ")") # @@ -167,13 +163,12 @@ def worker(remote, parent_remote, threadNum, args): # class VecEnv: def __init__(self, num_envs, programArgs): - self.closed = False + self.closed = False self.num_envs = num_envs - self.workerThreadNums = [x for x in range(num_envs)] # A different thread number (0-numEnvs) for each worker thread, so the ScienceWorld servers spawn on different ports self.remotes, self.work_remotes = zip(*[Pipe() for _ in range(num_envs)]) - self.ps = [Process(target=worker, args=(work_remote, remote, threadNum, programArgs)) - for (work_remote, remote, threadNum) in zip(self.work_remotes, self.remotes, self.workerThreadNums)] + self.ps = [Process(target=worker, args=(work_remote, remote, programArgs)) + for (work_remote, remote) in zip(self.work_remotes, self.remotes)] for p in self.ps: p.daemon = True # if the main process crashes, we should not cause things to hang p.start() @@ -219,8 +214,6 @@ def close_extras(self): for remote in self.remotes: remote.send(('close', None)) - time.sleep(5) - for p in self.ps: p.join() From 547b1b71cf3834d0572f7cd17319353104d248a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Alexandre=20C=C3=B4t=C3=A9?= Date: Tue, 8 Nov 2022 10:22:18 -0500 Subject: [PATCH 2/2] WIP --- Dockerfile | 30 ++-- beaker/batchSubmission.py | 18 ++- drrn/drrn.py | 21 +-- drrn/train-scienceworld.py | 312 +++++++++++++++++++------------------ drrn/vec_env.py | 4 +- requirements.txt | 2 +- 6 files changed, 198 insertions(+), 189 deletions(-) diff --git a/Dockerfile b/Dockerfile index c4191d4..d35635f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -26,29 +26,29 @@ ENV LD_LIBRARY_PATH=/usr/local/nvidia/lib:/usr/local/nvidia/lib64 ENV NVIDIA_VISIBLE_DEVICES=all ENV NVIDIA_DRIVER_CAPABILITIES=compute,utility -EXPOSE 5001 8883 8888 9000 -EXPOSE 25300-25600 +#EXPOSE 5001 8883 8888 9000 +#EXPOSE 25300-25600 USER root:root WORKDIR /opt -RUN wget http://nlp.stanford.edu/software/stanford-corenlp-full-2018-10-05.zip +#RUN wget http://nlp.stanford.edu/software/stanford-corenlp-full-2018-10-05.zip RUN apt-get update \ && apt-get install -y --no-install-recommends default-jre -RUN apt-get install -y --no-install-recommends unzip -RUN unzip stanford-corenlp-full-2018-10-05.zip \ - && mv $(ls -d stanford-corenlp-full-*/) corenlp \ - && rm *.zip -EXPOSE 5002-5100 -EXPOSE 5002 5003 5004 5005 5006 5007 5008 5009 5010 5011 5012 5013 5014 5015 5016 5017 5018 -EXPOSE 50022 50023 50024 50025 50026 50027 50028 50029 50030 50031 50032 50034 50035 50036 50037 50038 50039 -EXPOSE 50022 50032 50042 50052 50062 50072 50082 50092 50102 50112 50122 50132 50142 50152 50162 50172 50182 -COPY . /tdqn-scienceworld -RUN pip install -r /tdqn-scienceworld/requirements.txt +#RUN apt-get install -y --no-install-recommends unzip +#RUN unzip stanford-corenlp-full-2018-10-05.zip \ +# && mv $(ls -d stanford-corenlp-full-*/) corenlp \ +# && rm *.zip +#EXPOSE 5002-5100 +#EXPOSE 5002 5003 5004 5005 5006 5007 5008 5009 5010 5011 5012 5013 5014 5015 5016 5017 5018 +#EXPOSE 50022 50023 50024 50025 50026 50027 50028 50029 50030 50031 50032 50034 50035 50036 50037 50038 50039 +#EXPOSE 50022 50032 50042 50052 50062 50072 50082 50092 50102 50112 50122 50132 50142 50152 50162 50172 50182 +COPY . /drrn-scienceworld +RUN pip install -r /drrn-scienceworld/requirements.txt RUN pip3 install torch==1.10.1+cu113 -f https://download.pytorch.org/whl/cu113/torch_stable.html WORKDIR / -ENV PYTHONPATH=/tdqn-scienceworld/drrn +ENV PYTHONPATH=/drrn-scienceworld/drrn ENV HOME="" -WORKDIR /tdqn-scienceworld/drrn +WORKDIR /drrn-scienceworld/drrn diff --git a/beaker/batchSubmission.py b/beaker/batchSubmission.py index a933ae1..1863972 100644 --- a/beaker/batchSubmission.py +++ b/beaker/batchSubmission.py @@ -11,7 +11,7 @@ - name: sciworld-may31-drrn-8x100k-taskTASKID-seedSEEDNUM image: beaker: peterj/sciworld-drrn2c - arguments: [python3, train-scienceworld.py, --num_envs=8, --max_steps=100000, --task_idx=TASKID, --simplification_str=easy, --priority_fraction=0.50, --memory_size=100000, --env_step_limit=100, --log_freq=100, --checkpoint_freq=100000, --eval_freq=2000, --seed=SEEDNUM, --maxHistoriesPerFile=1000, --historySavePrefix=/results1/drrn1/results-seedSEEDNUM] + arguments: [python3, train-scienceworld.py, --num_envs=8, --max_steps=100000, --task_idx=TASKID, --simplification_str=easy, --priority_fraction=0.50, --memory_size=100000, --env_step_limit=100, --log_freq=100, --checkpoint_freq=5000, --eval_freq=1000, --seed=SEEDNUM] result: path: /results1/drrn1/ resources: @@ -20,10 +20,11 @@ cluster: ai2/raja_p100 priority: normal """ +template_command = "python train-scienceworld.py --num_envs=8 --max_steps=100000 --task_idx=TASKID --simplification_str=easy --priority_fraction=0.50 --memory_size=100000 --env_step_limit=100 --log_freq=100 --checkpoint_freq=5000 --eval_freq=1000 --seed=SEEDNUM --output_dir logs/drrn-8x100k-taskTASKID-seedSEEDNUM" def populateTemplate(taskId, seedNum): - outStr = templateStr + outStr = template_command outStr = outStr.replace("SEEDNUM", str(seedNum)) outStr = outStr.replace("TASKID", str(taskId)) @@ -47,17 +48,18 @@ def submitJob(filenameToRun): numJobs = 0 for seed in range(0, 1): - for taskIdx in range(0, 30): + for taskIdx in range(0, 30): tempFilename = "submit.yml" - print("Creating job (" + str(numJobs) + "): Task: " + str(taskIdx) + " seed: " + str(seed)) + #print("Creating job (" + str(numJobs) + "): Task: " + str(taskIdx) + " seed: " + str(seed)) scriptStr = populateTemplate(taskIdx, seed) - writeTemplate(tempFilename, scriptStr) - submitJob(tempFilename) + #writeTemplate(tempFilename, scriptStr) + print(scriptStr) + #submitJob(tempFilename) - time.sleep(1) + #time.sleep(1) numJobs += 1 - print("") + #print("") #print(populateTemplate(10, 2)) print("Submitted " + str(numJobs) + " jobs.") diff --git a/drrn/drrn.py b/drrn/drrn.py index f5eef08..ab61696 100644 --- a/drrn/drrn.py +++ b/drrn/drrn.py @@ -73,14 +73,17 @@ def build_state(self, obs, infos): """ Returns a state representation built from various info sources. """ obs_ids = [self.sp.EncodeAsIds(o) for o in obs] # TextWorld - look_ids = [self.sp.EncodeAsIds(info['look']) for info in infos] - inv_ids = [self.sp.EncodeAsIds(info['inv']) for info in infos] + #look_ids = [self.sp.EncodeAsIds(info['look']) for info in infos] + #inv_ids = [self.sp.EncodeAsIds(info['inv']) for info in infos] + look_ids = [self.sp.EncodeAsIds(look) for look in infos['look']] + inv_ids = [self.sp.EncodeAsIds(inv) for inv in infos['inv']] + # ScienceWorld #print("obs:") #print(obs) #print("infos:") - #print(infos) + #print(infos) #look_ids = [self.sp.EncodeAsIds(info['look']) for info in infos] #inv_ids = [self.sp.EncodeAsIds(info['inv']) for info in infos] @@ -146,11 +149,11 @@ def save(self, suffixStr=""): print("Saving agent to path: " + str(self.save_path)) print("Started saving at: " + str(startTime)) sys.stdout.flush() - + # First, remove any old backups print("Removing old backups") sys.stdout.flush() - try: + try: files = os.listdir(self.save_path + "/bak") for filename in files: if (filename.startswith("memory")) or (filename.startswith("model") or (filename.startswith("progress") or (filename.startswith("log")))): @@ -167,9 +170,9 @@ def save(self, suffixStr=""): os.makedirs(self.save_path + "/bak", exist_ok=True) files = os.listdir(self.save_path) for filename in files: - if filename.startswith("memory") or filename.startswith("model"): + if filename.startswith("memory") or filename.startswith("model"): shutil.move(self.save_path + "/" + filename, self.save_path + "/bak/" + filename) - if filename.startswith("progress") or filename.startswith("log"): + if filename.startswith("progress") or filename.startswith("log"): shutil.copy(self.save_path + "/" + filename, self.save_path + "/bak/" + filename) @@ -181,7 +184,7 @@ def save(self, suffixStr=""): self.lastSaveSuccessful = False with timeout(120): - print("Pickle") + print("Pickle") print("Length: " + str(len(self.memory)) ) sys.stdout.flush() pickle.dump(self.memory, open(pjoin(self.save_path, "memory" + str(suffixStr) + ".pkl"), 'wb')) @@ -195,7 +198,7 @@ def save(self, suffixStr=""): if (self.lastSaveSuccessful == False): print("* Model failed to save (timeout).") self.numSaveErrors += 1 - + print("Total number of save timeouts since running: " + str(self.numSaveErrors)) sys.stdout.flush() diff --git a/drrn/train-scienceworld.py b/drrn/train-scienceworld.py index 416a180..b028d08 100644 --- a/drrn/train-scienceworld.py +++ b/drrn/train-scienceworld.py @@ -1,13 +1,14 @@ -import subprocess import time -import math import timeit import torch import logger import argparse from drrn import DRRN_Agent -from vec_env import VecEnv -import random +import numpy as np + +import gymnasium as gym +from gymnasium.wrappers import TimeLimit, AutoResetWrapper +from functools import partial from scienceworld import ScienceWorldEnv, BufferedHistorySaver from vec_env import resetWithVariation, resetWithVariationDev, resetWithVariationTest, initializeEnv, sanitizeInfo, sanitizeObservation @@ -16,112 +17,97 @@ def configure_logger(log_dir): logger.configure(log_dir, format_strs=['log']) global tb - tb = logger.Logger(log_dir, [logger.make_output_format('tensorboard', log_dir), + tb = logger.Logger(log_dir, [#logger.make_output_format('tensorboard', log_dir), logger.make_output_format('csv', log_dir), - logger.make_output_format('stdout', log_dir)]) + #logger.make_output_format('stdout', log_dir) + ]) global log log = logger.log -def clean(strIn): - charsToFilter = ['\t', '\n', '*', '-'] - for c in charsToFilter: - strIn = strIn.replace(c, ' ') - return strIn.strip() +# def clean(strIn): +# charsToFilter = ['\t', '\n', '*', '-'] +# for c in charsToFilter: +# strIn = strIn.replace(c, ' ') +# return strIn.strip() -def evaluate(agent, args, env_step_limit, bufferedHistorySaverEval, extraSaveInfo, nb_episodes=10): - # Initialize a ScienceWorld server for serial evaluation - env = initializeEnv(args=args) +def evaluate(agent, envs_eval, options_eval, args, bufferedHistorySaverEval, extraSaveInfo): - scoresOut = [] + rng = np.random.default_rng(args.seed) + seeds = list(map(int, rng.integers(2**32, size=envs_eval.num_envs))) with torch.no_grad(): + obs, infos = envs_eval.reset(seed=seeds, options=options_eval) + dones = np.array([False] * envs_eval.num_envs) + + #log("Starting evaluation") + while not np.all(dones): + # Encode state and valid actions. + states = agent.build_state(obs, infos) + valid_ids = [agent.encode(valid) for valid in infos['valid']] + + # Choose actions + action_ids, action_idxs, _ = agent.act(states, valid_ids, sample=False) + action_strs = [valid[idx] for valid, idx in zip(infos['valid'], action_idxs)] + + # Perform the actions in the environments + obs, rewards, terminateds, truncateds, infos = envs_eval.step(action_strs) + dones = np.logical_or(terminateds, truncateds) + + print("Completed evaluation") + for runHistory, evalIdx in zip(infos["runHistory"], infos["variationIdx"]): + episodeIdx = str(extraSaveInfo["stepsFunctional"]) + "-" + str(evalIdx) + bufferedHistorySaverEval.storeRunHistory(runHistory, episodeIdx, notes=dict(extraSaveInfo)) + bufferedHistorySaverEval.saveRunHistoriesBufferIfFull(maxPerFile=extraSaveInfo['maxHistoriesPerFile']) + + avg_score = np.mean(infos['score']) + return infos['score'], avg_score + + +class SkipDone(gym.Wrapper): + + def reset(self, **kwargs): + observation, info = self.env.reset(**kwargs) + self._last_state = None + self._is_done = False + return observation, info + + def step(self, action): + if not self._is_done: + observation, reward, terminated, truncated, info = self.env.step(action) + self._is_done = terminated or truncated + info["_terminated"] = terminated + info["_truncated"] = truncated + + terminated = truncated = False # To avoid being autoreset. + self._last_state = (observation, reward, terminated, truncated, info) + + return self._last_state + +class PostProcessSkipDone(gym.vector.VectorWrapper): - for ep in range(nb_episodes): - total_score = 0 - log("Starting evaluation episode {}".format(ep)) - print("Starting evaluation episode " + str(ep) + " / " + str(nb_episodes)) - extraSaveInfo['evalIdx'] = ep - score = evaluate_episode(agent, env, env_step_limit, args.simplification_str, bufferedHistorySaverEval, extraSaveInfo, args.eval_set) - log("Evaluation episode {} ended with score {}\n\n".format(ep, score)) - total_score += score - scoresOut.append(total_score) - print("") - - avg_score = total_score / nb_episodes - - return scoresOut, avg_score - - - - -def evaluate_episode(agent, env, env_step_limit, simplificationStr, bufferedHistorySaverEval, extraSaveInfo, evalSet): - step = 0 - done = False - numSteps = 0 - ob = "" - info = {} - if (evalSet == "dev"): - ob, info = resetWithVariationDev(env, simplificationStr) - info = sanitizeInfo(info) - ob = sanitizeObservation(ob, info) - - elif (evalSet == "test"): - ob, info = resetWithVariationTest(env, simplificationStr) - info = sanitizeInfo(info) - ob = sanitizeObservation(ob, info) - - else: - print("evaluate_episode: unknown evaluation set (expected 'dev' or 'test', found: " + str(evalSet) + ")") - exit(1) - - - state = agent.build_state([ob], [info])[0] - log('Obs{}: {} Inv: {} Desc: {}'.format(step, clean(ob), clean(info['inv']), clean(info['look']))) - while not done: - #print("numSteps: " + str(numSteps)) - valid_acts = info['valid'] - valid_ids = agent.encode(valid_acts) - _, action_idx, action_values = agent.act([state], [valid_ids], sample=False) - action_idx = action_idx[0] - action_values = action_values[0] - action_str = valid_acts[action_idx] - log('Action{}: {}, Q-Value {:.2f}'.format(step, action_str, action_values[action_idx].item())) - s = '' - - maxToDisplay = 10 # Max Q values to display, to limit the log size - numDisplayed = 0 - for idx, (act, val) in enumerate(sorted(zip(valid_acts, action_values), key=lambda x: x[1], reverse=True), 1): - s += "{}){:.2f} {} ".format(idx, val.item(), act) - numDisplayed += 1 - if (numDisplayed > maxToDisplay): - break - - log('Q-Values: {}'.format(s)) - ob, rew, done, info = env.step(action_str) - info = sanitizeInfo(info) - ob = sanitizeObservation(ob, info) - - - log("Reward{}: {}, Score {}, Done {}".format(step, rew, info['score'], done)) - step += 1 - log('Obs{}: {} Inv: {} Desc: {}'.format(step, clean(ob), clean(info['inv']), clean(info['look']))) - state = agent.build_state([ob], [info])[0] - - numSteps +=1 - if (numSteps > env_step_limit): - print("Maximum number of evaluation steps reached (" + str(env_step_limit) + ").") - break - - print("Completed one evaluation episode") - # Save - runHistory = env.getRunHistory() - episodeIdx = str(extraSaveInfo['numEpisodes']) + "-" + str(extraSaveInfo['evalIdx']) - bufferedHistorySaverEval.storeRunHistory(runHistory, episodeIdx, notes=extraSaveInfo) - bufferedHistorySaverEval.saveRunHistoriesBufferIfFull(maxPerFile=extraSaveInfo['maxHistoriesPerFile']) - print("Completed saving") - - - return info['score'] + def step(self, actions): + observations, rewards, terminateds, truncateds, infos = self.env.step(actions) + terminateds = infos["_terminated"] + truncateds = infos["_truncated"] + return observations, rewards, terminateds, truncateds, infos + + +class SanitizeScienceWorld(gym.Wrapper): + + def reset(self, **kwargs): + obs, info = self.env.reset(**kwargs) + obs = info['taskDesc'] + " OBSERVATION " + obs + info["runHistory"] = "" + return obs, info + + def step(self, action): + obs, reward, terminated, truncated, info = self.env.step(action) + obs = info['taskDesc'] + " OBSERVATION " + obs + info["runHistory"] = "" + if terminated or truncated: + info["runHistory"] = self.env.unwrapped.env.getRunHistory() + + return obs, reward, terminated, truncated, info def train(agent, envs, max_steps, update_freq, eval_freq, checkpoint_freq, log_freq, args, bufferedHistorySaverTrain, bufferedHistorySaverEval): @@ -133,49 +119,77 @@ def train(agent, envs, max_steps, update_freq, eval_freq, checkpoint_freq, log_f stepsFunctional = 0 start1 = timeit.default_timer() + # Initialize a threaded wrapper for the ScienceWorld environment + TimeLimitWrapper = partial(TimeLimit, max_episode_steps=args.env_step_limit) + + envs_train = gym.make_vec( + "ScienceWorld-v0", + num_envs=args.num_envs, + vectorization_mode="async", + vector_kwargs={"shared_memory": False}, + wrappers=[TimeLimitWrapper, SanitizeScienceWorld] + ) + envs_eval = gym.make_vec( + "ScienceWorld-v0", + num_envs=10, + vectorization_mode="async", + vector_kwargs={"shared_memory": False}, + wrappers=[TimeLimitWrapper, SanitizeScienceWorld, SkipDone] + ) + envs_eval = PostProcessSkipDone(envs_eval) + + options_train = { + "task": args.task_idx, + "variation": "train", + "simplification": args.simplification_str, + } + options_eval = { + "task": args.task_idx, + "variation": args.eval_set, + "simplification": args.simplification_str, + } # Reinit environments - obs, infos = envs.reset() + rng = np.random.default_rng(args.seed) + seeds = list(map(int, rng.integers(2**32, size=envs_train.num_envs))) + obs, infos = envs_train.reset(seed=seeds, options=options_train) states = agent.build_state(obs, infos) - valid_ids = [agent.encode(info['valid']) for info in infos] + valid_ids = [agent.encode(valid) for valid in infos['valid']] + loss = np.inf for step in range(1, max_steps+1): - stepsFunctional = step * envs.num_envs + stepsFunctional = step * envs_train.num_envs # Summary statistics - print("-------------------") - print("Step " + str(step)) - print("") + #print("-------------------") end = timeit.default_timer() deltaTime = end - start1 deltaTimeMins = deltaTime / 60 - print("Started at runtime: " + str(deltaTime) + " seconds (" + str(deltaTimeMins) + " minutes)") - print("") + print(f"Step {step}. Loss: {loss:.4f} ({deltaTimeMins:.2f} minutes)") # Choose action(s) action_ids, action_idxs, _ = agent.act(states, valid_ids) - action_strs = [info['valid'][idx] for info, idx in zip(infos, action_idxs)] + action_strs = [valid[idx] for valid, idx in zip(infos['valid'], action_idxs)] # Perform the action(s) in the environment - obs, rewards, dones, infos = envs.step(action_strs) + obs, rewards, terminateds, truncateds, infos = envs_train.step(action_strs) + dones = np.logical_or(terminateds, truncateds) # Check for any completed episodes - for done, info in zip(dones, infos): + for i, (done, score) in enumerate(zip(dones, infos['score'])): if done: # An episode has completed - tb.logkv('EpisodeScore', info['score']) - print("EPISODE SCORE: " + str(info['score'])) - print("EPISODE SCORE: " + str(info['score']) + " STEPS: " + str(step) + " STEPS (functional): " + str(stepsFunctional) + " EPISODES: " + str(numEpisodes)) + tb.logkv('EpisodeScore', score) + print("EPISODE SCORE: " + str(score) + " STEPS: " + str(step) + " STEPS (functional): " + str(stepsFunctional) + " EPISODES: " + str(numEpisodes)) # Save the environment's history in the history logs - runHistory = info['runHistory'] - bufferedHistorySaverTrain.storeRunHistory(runHistory, numEpisodes, notes={'step':step}) + bufferedHistorySaverTrain.storeRunHistory(infos["final_info"][i]["runHistory"], numEpisodes, notes={'step':step}) bufferedHistorySaverTrain.saveRunHistoriesBufferIfFull(maxPerFile=args.maxHistoriesPerFile) numEpisodes += 1 next_states = agent.build_state(obs, infos) - next_valids = [agent.encode(info['valid']) for info in infos] + next_valids = [agent.encode(valid) for valid in infos['valid']] for state, act, rew, next_state, valids, done in \ zip(states, action_ids, rewards, next_states, next_valids, dones): agent.observe(state, act, rew, next_state, valids, done) @@ -184,16 +198,16 @@ def train(agent, envs, max_steps, update_freq, eval_freq, checkpoint_freq, log_f if step % log_freq == 0: tb.logkv('Step', step) - tb.logkv('StepsFunctional', step*envs.num_envs) - tb.logkv("FPS", int((step*envs.num_envs)/(time.time()-startTime))) + tb.logkv('StepsFunctional', stepsFunctional) + tb.logkv("FPS", int((stepsFunctional)/(time.time()-startTime))) tb.logkv('numEpisodes', numEpisodes) tb.logkv('taskIdx', args.task_idx) tb.logkv('GPU_mem', agent.getMemoryUsage()) print("*************************") print("Step: " + str(step)) - print("StepsFunctional: " + str(step*envs.num_envs)) - print("FPS: " + str( (step*envs.num_envs)/(time.time()-startTime)) ) + print("StepsFunctional: " + str(stepsFunctional)) + print("FPS: " + str(stepsFunctional/(time.time()-startTime)) ) print("numEpisodes: " + str(numEpisodes)) print("taskIdx: " + str(args.task_idx)) print("GPU_mem: " + str(agent.getMemoryUsage())) @@ -203,6 +217,8 @@ def train(agent, envs, max_steps, update_freq, eval_freq, checkpoint_freq, log_f loss = agent.update() if loss is not None: tb.logkv_mean('Loss', loss) + else: + loss = np.inf if step % checkpoint_freq == 0: # Save model checkpoints @@ -214,18 +230,17 @@ def train(agent, envs, max_steps, update_freq, eval_freq, checkpoint_freq, log_f if step % eval_freq == 0: # Do the evaluation procedure - extraSaveInfo = {'numEpisodes':numEpisodes, 'numSteps':step, 'stepsFunctional:':stepsFunctional, 'maxHistoriesPerFile':args.maxHistoriesPerFile} - eval_scores, avg_eval_score = evaluate(agent, args, args.env_step_limit, bufferedHistorySaverEval, extraSaveInfo) + extraSaveInfo = {'numEpisodes':numEpisodes, 'numSteps':step, 'stepsFunctional':stepsFunctional, 'maxHistoriesPerFile':args.maxHistoriesPerFile} + eval_scores, avg_eval_score = evaluate(agent, envs_eval, options_eval, args, bufferedHistorySaverEval, extraSaveInfo) tb.logkv('EvalScore', avg_eval_score) tb.logkv('numEpisodes', numEpisodes) tb.dumpkvs() for eval_score in eval_scores: - print("EVAL EPISODE SCORE: " + str(eval_score)) - print("EVAL EPISODE SCORE: " + str(eval_score) + " STEPS: " + str(step) + " STEPS: " + str(stepsFunctional) + " EPISODES: " + str(numEpisodes)) + print("EVAL EPISODE SCORE: " + str(eval_score) + " STEPS: " + str(step) + " STEPS (functional): " + str(stepsFunctional) + " EPISODES: " + str(numEpisodes)) - envs.reset() + envs_train.reset() # Save anything left in history buffers @@ -236,24 +251,25 @@ def train(agent, envs, max_steps, update_freq, eval_freq, checkpoint_freq, log_f print("Training complete.") # Final save agent.save("-steps" + str(stepsFunctional) + "-eps" + str(numEpisodes)) + # Close environments - envs.close_extras() + #envs.close_extras() + def parse_args(): parser = argparse.ArgumentParser() parser.add_argument('--output_dir', default='logs') parser.add_argument('--spm_path', default='../spm_models/unigram_8k.model') - parser.add_argument('--rom_path', default='zork1.z5') parser.add_argument('--env_step_limit', default=100, type=int) - parser.add_argument('--seed', default=0, type=int) - parser.add_argument('--num_envs', default=16, type=int) + parser.add_argument('--seed', default=20221108, type=int) + parser.add_argument('--num_envs', default=8, type=int) parser.add_argument('--max_steps', default=100000, type=int) parser.add_argument('--update_freq', default=1, type=int) - parser.add_argument('--checkpoint_freq', default=500, type=int) - parser.add_argument('--eval_freq', default=500, type=int) + parser.add_argument('--checkpoint_freq', default=5000, type=int) + parser.add_argument('--eval_freq', default=1000, type=int) parser.add_argument('--log_freq', default=100, type=int) - parser.add_argument('--memory_size', default=5000000, type=int) - parser.add_argument('--priority_fraction', default=0.0, type=float) + parser.add_argument('--memory_size', default=100000, type=int) + parser.add_argument('--priority_fraction', default=0.5, type=float) parser.add_argument('--batch_size', default=64, type=int) parser.add_argument('--gamma', default=.9, type=float) parser.add_argument('--learning_rate', default=0.0001, type=float) @@ -263,35 +279,32 @@ def parse_args(): parser.add_argument('--task_idx', default=0, type=int) parser.add_argument('--maxHistoriesPerFile', default=1000, type=int) - parser.add_argument('--historySavePrefix', default='saveout', type=str) + parser.add_argument('--historySavePrefix') - parser.add_argument('--eval_set', default='dev', type=str) # 'dev' or 'test' + parser.add_argument('--eval_set', default='test', type=str) # 'dev' or 'test' - parser.add_argument('--simplification_str', default='', type=str) + parser.add_argument('--simplification_str', default='easy', type=str) return parser.parse_args() def main(): - ## assert jericho.__version__ == '2.1.0', "This code is designed to be run with Jericho version 2.1.0." args = parse_args() print(args) configure_logger(args.output_dir) agent = DRRN_Agent(args) - # Initialize a threaded wrapper for the ScienceWorld environment - envs = VecEnv(args.num_envs, args) - # Initialize the save buffers taskIdx = args.task_idx - bufferedHistorySaverTrain = BufferedHistorySaver(filenameOutPrefix = args.historySavePrefix + "-task" + str(taskIdx) + "-train") - bufferedHistorySaverEval = BufferedHistorySaver(filenameOutPrefix = args.historySavePrefix + "-task" + str(taskIdx) + "-eval") + history_save_prefix = args.historySavePrefix or args.output_dir + bufferedHistorySaverTrain = BufferedHistorySaver(filenameOutPrefix=f"{history_save_prefix}/history-seed{args.seed}-task{taskIdx}-train") + bufferedHistorySaverEval = BufferedHistorySaver(filenameOutPrefix=f"{history_save_prefix}/history-seed{args.seed}-task{taskIdx}-{args.eval_set}") # Start training start = timeit.default_timer() - train(agent, envs, args.max_steps, args.update_freq, args.eval_freq, + train(agent, None, args.max_steps, args.update_freq, args.eval_freq, args.checkpoint_freq, args.log_freq, args, bufferedHistorySaverTrain, bufferedHistorySaverEval) end = timeit.default_timer() @@ -303,15 +316,6 @@ def main(): print("SimplificationStr: " + str(args.simplification_str)) -def interactive_run(env): - ob, info = env.reset() - while True: - print(clean(ob), 'Reward', reward, 'Done', done, 'Valid', info) - ob, reward, done, info = env.step(input()) - info = sanitizeInfo(info) - ob = sanitizeObservation(ob, info) - - if __name__ == "__main__": main() diff --git a/drrn/vec_env.py b/drrn/vec_env.py index 6c355a5..08563f6 100644 --- a/drrn/vec_env.py +++ b/drrn/vec_env.py @@ -162,12 +162,12 @@ def worker(remote, parent_remote, args): # VecEnv: Handles spawning up 'num_envs' workers. # class VecEnv: - def __init__(self, num_envs, programArgs): + def __init__(self, num_envs, programArgs, is_eval=False): self.closed = False self.num_envs = num_envs self.remotes, self.work_remotes = zip(*[Pipe() for _ in range(num_envs)]) - self.ps = [Process(target=worker, args=(work_remote, remote, programArgs)) + self.ps = [Process(target=worker, args=(work_remote, remote, programArgs, is_eval)) for (work_remote, remote) in zip(self.work_remotes, self.remotes)] for p in self.ps: p.daemon = True # if the main process crashes, we should not cause things to hang diff --git a/requirements.txt b/requirements.txt index 233597d..568ec9d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,4 +7,4 @@ torch==1.8.1 py4j pyyaml protobuf==3.19.4 -scienceworld>=1.0.0 +scienceworld>=1.0.3rc2