Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
mathy.start()
while True:
print(f"Iteration: {counter}")
print(f"Difficulty: {difficulty}")
counter = counter + 1
eval_run = (
bool(counter % eval_interval == 0)
and experience.count >= min_train_experience
)
num_solved = 0
num_failed = 0
if eval_run:
print("\n\n=== Evaluating model with exploitation strategy ===")
mathy.stop()
mathy_eval = MathModel(
action_size,
model_dir,
init_model_dir=os.path.abspath(mathy.model_dir),
# We want to initialize from the training model for each evaluation. (?)
init_model_overwrite=True,
is_eval_model=True,
learning_rate=learning_rate,
epochs=training_epochs,
long_term_size=long_term_size,
)
mathy_eval.start()
model = mathy_eval if eval_run else mathy
# we fill this with episode rewards and when it's a fixed size we
# dump the average value to tensorboard
ep_reward_buffer: List[float] = []
# Fill up a certain amount of experience per problem type
def mathy_load_model(gym_env: MathyGymEnv):
global __model
if __model is None:
import os
os.environ["TF_CPP_MIN_LOG_LEVEL"] = "5"
tf.compat.v1.logging.set_verbosity("CRITICAL")
__model = MathModel(gym_env.mathy.action_size, "agents/ablated")
__model.start()
train_all = True
train_number = 2048 if not train_all else 1e6
controller = MathyEnv(verbose=True)
input_examples = Path(examples_file)
model_dir = Path(model_dir)
if not model_dir.is_dir():
print("Making model_dir: {}".format(model_dir))
model_dir.mkdir(parents=True, exist_ok=True)
if input_examples.is_file():
print("Copying examples into place: {}".format(model_dir))
train_dir = model_dir / "train"
if not train_dir.is_dir():
train_dir.mkdir(parents=True, exist_ok=True)
copyfile(str(input_examples), model_dir / "train" / INPUT_EXAMPLES_FILE_NAME)
mathy = MathModel(
controller.action_size,
model_dir,
init_model_dir=transfer_from,
init_model_overwrite=True,
learning_rate=learning_rate,
dropout=dropout,
long_term_size=train_number,
)
experience = MathExperience(mathy.model_dir)
mathy.start()
mathy.epochs = epochs
train_examples = mathy.train(
experience.short_term,
experience.long_term,
train_all=train_all,
sampling_fn=balanced_reward_experience_samples,
from typing import Optional
import gym
import numpy as np
import plac
import tensorflow as tf
from gym.envs.registration import register
from mathy.agent.controller import MathModel
from mathy.agent.training.mcts import MCTS
from mathy.gym import MathyGymEnv
from mathy.a3c import A3CAgent, A3CArgs
__mcts: Optional[MCTS] = None
__model: Optional[MathModel] = None
__agent: Optional[A3CAgent] = None
def mathy_load_model(gym_env: MathyGymEnv):
global __model
if __model is None:
import os
os.environ["TF_CPP_MIN_LOG_LEVEL"] = "5"
tf.compat.v1.logging.set_verbosity("CRITICAL")
__model = MathModel(gym_env.mathy.action_size, "agents/ablated")
__model.start()
def mathy_free_model():
global __model
def main(
model_dir, lesson_id=None, mcts_sims=500, num_exploration_moves=0, epsilon=0.0
):
controller = MathyEnv(verbose=True)
mathy = MathModel(controller.action_size, model_dir)
short_term_size = 128
experience = MathExperience(mathy.model_dir, short_term_size)
mathy.start()
if lesson_id is None:
lesson_id = list(lessons)[0]
if lesson_id not in lessons:
raise ValueError(
f"[exam] ERROR: '{lesson_id}' not found in ids. Valid lessons are: {', '.join(lessons)} "
)
plan = lessons[lesson_id]
plan_lessons = plan.lessons
num_solved = 0
num_failed = 0
model = mathy
print("[exam] using {} MCTS rollouts".format(mcts_sims))
print("[exam] lesson order: {}".format([l.name for l in plan_lessons]))
MathyComplexTermSimplificationEnv,
MathyBinomialDistributionEnv,
],
}
if env not in environments:
raise EnvironmentError(f"Invalid env, must be one of: {environments.keys()}")
# How many observations to gather between training sessions.
iter_episodes = 10
min_train_experience = 128
eval_interval = 2
short_term_size = 2048
long_term_size = 8192 * 3
counter = 0
training_epochs = 10
action_size = len(mathy_core_rules())
mathy = MathModel(
action_size,
model_dir,
init_model_dir=transfer_from,
learning_rate=learning_rate,
long_term_size=long_term_size,
epochs=training_epochs,
)
experience = MathExperience(mathy.model_dir, short_term_size)
mathy.start()
while True:
print(f"Iteration: {counter}")
print(f"Difficulty: {difficulty}")
counter = counter + 1
eval_run = (
bool(counter % eval_interval == 0)
and experience.count >= min_train_experience
def get_predictor(self, game, all_memory=False):
return MathModel(game, model_dir, all_memory)
def get_predictor(self, game, all_memory=False):
return MathModel(game, model_dir, all_memory)