{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#default_exp agents.cartpole"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# CartPole Agent\n",
"> An agent for solving the CartPole-v1 environment"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#export\n",
"\n",
"# OpenCog\n",
"from opencog.pln import *\n",
"from opencog.type_constructors import *\n",
"from opencog.utilities import set_default_atomspace\n",
"\n",
"# ROCCA\n",
"from rocca.envs.wrappers import CartPoleWrapper\n",
"from rocca.agents import OpencogAgent\n",
"from rocca.agents.utils import *\n",
"\n",
"from rocca.utils import *"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "eaf7640f",
"metadata": {},
"outputs": [],
"source": [
"import gym\n",
"import time\n",
"import logging\n",
"\n",
"from opencog.logger import log\n",
"from rocca.agents.core import logger as ac_logger\n",
"from tensorboardX import SummaryWriter"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Agent Definition"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Fixed Rule Agent"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#export\n",
"\n",
"class FixedCartPoleAgent(OpencogAgent):\n",
" def __init__(self, env: CartPoleWrapper, atomspace: AtomSpace):\n",
" set_default_atomspace(atomspace)\n",
" \n",
" # Create Action Space. The set of allowed actions an agent can take.\n",
" # TODO take care of action parameters.\n",
" action_space = {ExecutionLink(SchemaNode(a)) for a in env.action_names}\n",
"\n",
" # Create Goal\n",
" pgoal = EvaluationLink(PredicateNode(\"Reward\"), NumberNode(\"1\"))\n",
" ngoal = EvaluationLink(PredicateNode(\"Reward\"), NumberNode(\"0\"))\n",
"\n",
" # Call super ctor\n",
" super().__init__(env, atomspace, action_space, pgoal, ngoal)\n",
"\n",
" def plan(self, goal, expiry) -> list:\n",
" \"\"\"Plan the next actions given a goal and its expiry time offset\n",
"\n",
" Return a python list of cognitive schematics. Whole cognitive\n",
" schematics are output (instead of action plans) in order to\n",
" make a decision based on their truth values. Alternatively it\n",
" could return a pair (action plan, tv), where tv has been\n",
" evaluated to take into account the truth value of the context\n",
" as well (which would differ from the truth value of rule in\n",
" case the context is uncertain).\n",
"\n",
" The format for a cognitive schematic is as follows\n",
"\n",
" BackPredictiveImplicationScope <tv>\n",
" <vardecl>\n",
" <expiry>\n",
" And (or SimultaneousAnd?)\n",
" <context>\n",
" Execution\n",
" <action>\n",
" <input> [optional]\n",
" <output> [optional]\n",
" <goal>\n",
"\n",
" \"\"\"\n",
"\n",
" # For now we provide 2 hardwired rules\n",
" #\n",
" # 1. Push cart to the left (0) if angle is negative\n",
" # 2. Push cart to the right (1) if angle is positive\n",
" #\n",
" # with some arbitrary truth value (stv 0.9, 0.1)\n",
" angle = VariableNode(\"$angle\")\n",
" numt = TypeNode(\"NumberNode\")\n",
" time_offset = to_nat(1)\n",
" pole_angle = PredicateNode(\"Pole Angle\")\n",
" go_right = SchemaNode(\"Go Right\")\n",
" go_left = SchemaNode(\"Go Left\")\n",
" reward = PredicateNode(\"Reward\")\n",
" epsilon = NumberNode(\"0.01\")\n",
" mepsilon = NumberNode(\"-0.01\")\n",
" unit = NumberNode(\"1\")\n",
" hTV = TruthValue(0.9, 0.1) # High TV\n",
" lTV = TruthValue(0.1, 0.1) # Low TV\n",
"\n",
" # BackPredictiveImplicationScope <high TV>\n",
" # TypedVariable\n",
" # Variable \"$angle\"\n",
" # Type \"NumberNode\"\n",
" # Time \"1\"\n",
" # And\n",
" # Evaluation\n",
" # Predicate \"Pole Angle\"\n",
" # Variable \"$angle\"\n",
" # GreaterThan\n",
" # Variable \"$angle\"\n",
" # 0\n",
" # Execution\n",
" # Schema \"Go Right\"\n",
" # Evaluation\n",
" # Predicate \"Reward\"\n",
" # Number \"1\"\n",
" cs_rr = BackPredictiveImplicationScopeLink(\n",
" TypedVariableLink(angle, numt),\n",
" time_offset,\n",
" AndLink(\n",
" # Context\n",
" EvaluationLink(pole_angle, angle),\n",
" GreaterThanLink(angle, epsilon),\n",
" # Action\n",
" ExecutionLink(go_right),\n",
" ),\n",
" # Goal\n",
" EvaluationLink(reward, unit),\n",
" # TV\n",
" tv=hTV,\n",
" )\n",
"\n",
" # BackPredictiveImplicationScope <high TV>\n",
" # TypedVariable\n",
" # Variable \"$angle\"\n",
" # Type \"NumberNode\"\n",
" # Time \"1\"\n",
" # And\n",
" # Evaluation\n",
" # Predicate \"Pole Angle\"\n",
" # Variable \"$angle\"\n",
" # GreaterThan\n",
" # 0\n",
" # Variable \"$angle\"\n",
" # Execution\n",
" # Schema \"Go Left\"\n",
" # Evaluation\n",
" # Predicate \"Reward\"\n",
" # Number \"1\"\n",
" cs_ll = BackPredictiveImplicationScopeLink(\n",
" TypedVariableLink(angle, numt),\n",
" time_offset,\n",
" AndLink(\n",
" # Context\n",
" EvaluationLink(pole_angle, angle),\n",
" GreaterThanLink(mepsilon, angle),\n",
" # Action\n",
" ExecutionLink(go_left),\n",
" ),\n",
" # Goal\n",
" EvaluationLink(reward, unit),\n",
" # TV\n",
" tv=hTV,\n",
" )\n",
"\n",
" # To cover all possibilities we shouldn't forget the complementary\n",
" # actions, i.e. going left when the pole is falling to the right\n",
" # and such, which should make the situation worse.\n",
"\n",
" # BackPredictiveImplicationScope <low TV>\n",
" # TypedVariable\n",
" # Variable \"$angle\"\n",
" # Type \"NumberNode\"\n",
" # Time \"1\"\n",
" # And (or SimultaneousAnd?)\n",
" # Evaluation\n",
" # Predicate \"Pole Angle\"\n",
" # Variable \"$angle\"\n",
" # GreaterThan\n",
" # Variable \"$angle\"\n",
" # 0\n",
" # Execution\n",
" # Schema \"Go Left\"\n",
" # Evaluation\n",
" # Predicate \"Reward\"\n",
" # Number \"1\"\n",
" cs_rl = BackPredictiveImplicationScopeLink(\n",
" TypedVariableLink(angle, numt),\n",
" time_offset,\n",
" AndLink(\n",
" # Context\n",
" EvaluationLink(pole_angle, angle),\n",
" GreaterThanLink(angle, epsilon),\n",
" # Action\n",
" ExecutionLink(go_left),\n",
" ),\n",
" # Goal\n",
" EvaluationLink(reward, unit),\n",
" # TV\n",
" tv=lTV,\n",
" )\n",
"\n",
" # BackPredictiveImplicationScope <low TV>\n",
" # TypedVariable\n",
" # Variable \"$angle\"\n",
" # Type \"NumberNode\"\n",
" # Time \"1\"\n",
" # And (or SimultaneousAnd?)\n",
" # Evaluation\n",
" # Predicate \"Pole Angle\"\n",
" # Variable \"$angle\"\n",
" # GreaterThan\n",
" # 0\n",
" # Variable \"$angle\"\n",
" # Execution\n",
" # Schema \"Go Right\"\n",
" # Evaluation\n",
" # Predicate \"Reward\"\n",
" # Number \"1\"\n",
" cs_lr = BackPredictiveImplicationScopeLink(\n",
" TypedVariableLink(angle, numt),\n",
" time_offset,\n",
" AndLink(\n",
" # Context\n",
" EvaluationLink(pole_angle, angle),\n",
" GreaterThanLink(mepsilon, angle),\n",
" # Action\n",
" ExecutionLink(go_right),\n",
" ),\n",
" # Goal\n",
" EvaluationLink(reward, unit),\n",
" # TV\n",
" tv=lTV,\n",
" )\n",
"\n",
" # Ideally we want to return only relevant cognitive schematics\n",
" # (i.e. with contexts probabilistically currently true) for\n",
" # now however we return everything and let to the deduction\n",
" # process deal with it, as it should be able to.\n",
" return [cs_ll, cs_rr, cs_rl, cs_lr]"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"def seed_with(agent, knowledge): # TODO: figure out how to pass atoms to be inserted.\n",
" set_default_atomspace(agent.atomspace)\n",
"\n",
" angle = VariableNode(\"$angle\")\n",
" numt = TypeNode(\"NumberNode\")\n",
" time_offset = to_nat(1)\n",
" pole_angle = PredicateNode(\"Pole Angle\")\n",
" go_right = SchemaNode(\"Go Right\")\n",
" go_left = SchemaNode(\"Go Left\")\n",
" reward = PredicateNode(\"Reward\")\n",
" epsilon = NumberNode(\"0.01\")\n",
" mepsilon = NumberNode(\"-0.01\")\n",
" unit = NumberNode(\"1\")\n",
" hTV = TruthValue(0.9, 0.1) # High TV\n",
" lTV = TruthValue(0.1, 0.1) # Low TV\n",
"\n",
" cs_rr = BackPredictiveImplicationScopeLink(\n",
" TypedVariableLink(angle, numt),\n",
" time_offset,\n",
" AndLink(\n",
" # Context\n",
" EvaluationLink(pole_angle, angle),\n",
" GreaterThanLink(angle, epsilon),\n",
" # Action\n",
" ExecutionLink(go_right),\n",
" ),\n",
" # Goal\n",
" EvaluationLink(reward, unit),\n",
" # TV\n",
" tv=hTV,\n",
" )\n",
"\n",
" cs_ll = BackPredictiveImplicationScopeLink(\n",
" TypedVariableLink(angle, numt),\n",
" time_offset,\n",
" AndLink(\n",
" # Context\n",
" EvaluationLink(pole_angle, angle),\n",
" GreaterThanLink(mepsilon, angle),\n",
" # Action\n",
" ExecutionLink(go_left),\n",
" ),\n",
" # Goal\n",
" EvaluationLink(reward, unit),\n",
" # TV\n",
" tv=hTV,\n",
" )\n",
"\n",
" cs_rl = BackPredictiveImplicationScopeLink(\n",
" TypedVariableLink(angle, numt),\n",
" time_offset,\n",
" AndLink(\n",
" # Context\n",
" EvaluationLink(pole_angle, angle),\n",
" GreaterThanLink(angle, epsilon),\n",
" # Action\n",
" ExecutionLink(go_left),\n",
" ),\n",
" # Goal\n",
" EvaluationLink(reward, unit),\n",
" # TV\n",
" tv=lTV,\n",
" )\n",
" \n",
" cs_lr = BackPredictiveImplicationScopeLink(\n",
" TypedVariableLink(angle, numt),\n",
" time_offset,\n",
" AndLink(\n",
" # Context\n",
" EvaluationLink(pole_angle, angle),\n",
" GreaterThanLink(mepsilon, angle),\n",
" # Action\n",
" ExecutionLink(go_right),\n",
" ),\n",
" # Goal\n",
" EvaluationLink(reward, unit),\n",
" # TV\n",
" tv=lTV,\n",
" )\n",
"\n",
" agent.cognitive_schematics.update(set([cs_ll, cs_rr, cs_rl, cs_lr])) # TODO: the code should update Python-side automatically."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Learning Agent"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#export\n",
"\n",
"class LearningCartPoleAgent(OpencogAgent):\n",
" def __init__(self, env: CartPoleWrapper, atomspace: AtomSpace, log_level=\"debug\"):\n",
" set_default_atomspace(atomspace)\n",
" \n",
" # Create Action Space. The set of allowed actions an agent can take.\n",
" # TODO take care of action parameters.\n",
" action_space = {ExecutionLink(SchemaNode(a)) for a in env.action_names}\n",
"\n",
" # Create Goal\n",
" pgoal = EvaluationLink(PredicateNode(\"Reward\"), NumberNode(\"1\"))\n",
" ngoal = EvaluationLink(PredicateNode(\"Reward\"), NumberNode(\"0\"))\n",
"\n",
" # Call super ctor\n",
" super().__init__(env, atomspace, action_space, pgoal, ngoal, log_level=log_level)\n",
"\n",
" # Overwrite some OpencogAgent parameters\n",
" self.monoaction_general_succeedent_mining = False\n",
" self.polyaction_mining = False\n",
" self.temporal_deduction = False"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Experiment"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"env = gym.make(\"CartPole-v1\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"env.action_space"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"env.observation_space"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Fixed Agent"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"atomspace = AtomSpace()\n",
"set_default_atomspace(atomspace)\n",
"wrapped_env = CartPoleWrapper(env, atomspace)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# tb_writer = SummaryWriter(comment=\"-cartpole-fixed\")\n",
"\n",
"cpa = FixedCartPoleAgent(wrapped_env, atomspace)\n",
"cpa.delta = 1.0e-16\n",
"\n",
"# Run control loop\n",
"while not cpa.control_cycle():\n",
" time.sleep(0.1)\n",
" log.debug(\"cycle_count = {}\".format(cpa.cycle_count))\n",
" # tb_writer.add_scalar(\n",
" # \"accumulated_reward\", cpa.accumulated_reward, cpa.cycle_count\n",
" # )\n",
"\n",
"log_msg(agent_log, f\"The final reward is {cpa.accumulated_reward}.\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Learning Agent"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"atomspace = AtomSpace()\n",
"set_default_atomspace(atomspace)\n",
"wrapped_env = CartPoleWrapper(env, atomspace)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"agent = LearningCartPoleAgent(wrapped_env, atomspace, log_level=\"fine\")\n",
"agent.delta = 1.0e-16\n",
"\n",
"# seed_with(agent, [\"fixme\"])"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"tb_writer = SummaryWriter(comment=\"-cartpole-learning-seeded\")\n",
"ac_logger.setLevel(logging.DEBUG) # The agents.core logger\n",
"\n",
"epochs = 1 # Number of epochs (learning / interacting episodes)\n",
"epoch_len = 200\n",
"\n",
"for i in range(epochs):\n",
" wrapped_env.restart()\n",
" agent.reset_action_counter()\n",
" accreward = agent.accumulated_reward # Keep track of the reward before\n",
"\n",
" # Learning phase: discover patterns to make more informed decisions\n",
" log_msg(agent_log, f\"Learning phase started. ({i + 1}/{epochs})\")\n",
" agent.learn()\n",
" \n",
" # Run agent to accumulate percepta\n",
" log_msg(agent_log, f\"Interaction phase started. ({i + 1}/{epochs})\")\n",
" for j in range(epoch_len):\n",
" done = agent.control_cycle()\n",
" # wrapped_env.render() uncomment to see the rendered env\n",
" time.sleep(0.01)\n",
" log.debug(\"cycle_count = {}\".format(agent.cycle_count))\n",
" if done:\n",
" break\n",
" \n",
" new_reward = agent.accumulated_reward - accreward\n",
" tb_writer.add_scalar(\"train/accumulated_reward\", new_reward, agent.cycle_count)\n",
" log_msg(agent_log, \"Accumulated reward during {}th epoch = {}\".format(i + 1, new_reward))\n",
" log_msg(agent_log, \"Action counter during {}th epoch:\\n{}\".format(i + 1, agent.action_counter)) # TODO: make the action counter look good\n",
"\n",
"log_msg(agent_log, f\"The average total reward over {epochs} trials (training): {agent.accumulated_reward / epochs}.\")\n",
"# TODO: add a separate testing loop and measure average total reward."
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3.10.0 64-bit",
"language": "python",
"name": "python3"
}
},
"nbformat": 4,
"nbformat_minor": 5
}