From e32457b521d0ae67c026335c66c2de0b82846996 Mon Sep 17 00:00:00 2001 From: Dawid Laszuk Date: Sun, 11 Jul 2021 13:29:33 -0700 Subject: [PATCH] feat: support experiments & (breaking) interface change Current intended use is through namespaces rather than func names. Each entity has their own name space, e.g. agents or environments, which have limited scope functions. Experiments support means allowing agents interact with environments on the same data plane. A few more crazy minor changes and we're aiming for major release. --- agents_bar/__init__.py | 2 +- agents_bar/agents.py | 77 ++++++++++++++++++---------- agents_bar/environments.py | 69 ++++++++++++++++--------- agents_bar/experiments.py | 102 +++++++++++++++++++++++++++++++++++++ agents_bar/remote_agent.py | 87 ++++++++----------------------- agents_bar/types.py | 17 +++++-- agents_bar/utils.py | 14 +++++ setup.cfg | 2 +- 8 files changed, 247 insertions(+), 123 deletions(-) create mode 100644 agents_bar/experiments.py diff --git a/agents_bar/__init__.py b/agents_bar/__init__.py index 1565dd2..5d6d3c4 100644 --- a/agents_bar/__init__.py +++ b/agents_bar/__init__.py @@ -2,6 +2,6 @@ from agents_bar.remote_agent import * from agents_bar.utils import * -__version__ = '0.3.0' +__version__ = '0.4.0' __author__ = "Dawid Laszuk" diff --git a/agents_bar/agents.py b/agents_bar/agents.py index 2014dd5..bbb9cda 100644 --- a/agents_bar/agents.py +++ b/agents_bar/agents.py @@ -1,9 +1,12 @@ +from agents_bar.utils import response_raise_error_if_any from typing import Dict, List, Optional from agents_bar.client import Client +AGENTS_PREFIX = "/agents" -def get_agents(client: Client) -> List[Dict]: + +def get_many(client: Client) -> List[Dict]: """Gets agents belonging to authenticated user. Parameters: @@ -13,12 +16,11 @@ def get_agents(client: Client) -> List[Dict]: List of agents. """ - response = client.get('/agents/') - if not response.ok: - response.raise_for_status() + response = client.get(f'{AGENTS_PREFIX}/') + response_raise_error_if_any(response) return response.json() -def get_agent(client: Client, agent_name: str) -> Dict: +def get(client: Client, agent_name: str) -> Dict: """Get indepth information about a specific agent. Parameters: @@ -29,12 +31,11 @@ def get_agent(client: Client, agent_name: str) -> Dict: Details of an agent. """ - response = client.get(f'/agents/{agent_name}') - if not response.ok: - response.raise_for_status() + response = client.get(f'{AGENTS_PREFIX}/{agent_name}') + response_raise_error_if_any(response) return response.json() -def create_agent(client: Client, config: Dict) -> Dict: +def create(client: Client, config: Dict) -> Dict: """Creates an agent with specified configuration. Parameters: @@ -45,12 +46,11 @@ def create_agent(client: Client, config: Dict) -> Dict: Details of an agent. """ - response = client.post('/agents/', data=config) - if not response.ok: - response.raise_for_status() + response = client.post(f'{AGENTS_PREFIX}/', data=config) + response_raise_error_if_any(response) return response.json() -def delete_agent(client: Client, agent_name: str) -> bool: +def delete(client: Client, agent_name: str) -> bool: """Deletes specified agent. Parameters: @@ -61,13 +61,12 @@ def delete_agent(client: Client, agent_name: str) -> bool: Whether agent was delete. True if an agent was delete, False if there was no such agent. """ - response = client.delete('/agents/' + agent_name) - if not response.ok: - response.raise_for_status() + response = client.delete(f'{AGENTS_PREFIX}/' + agent_name) + response_raise_error_if_any(response) return response.status_code == 202 -def get_agent_loss(client: Client, agent_name: str) -> Dict: +def get_loss(client: Client, agent_name: str) -> Dict: """Recent loss metrics. Parameters: @@ -78,19 +77,41 @@ def get_agent_loss(client: Client, agent_name: str) -> Dict: Loss metrics in a dictionary. """ - response = client.get(f'/agents/{agent_name}/loss') - if not response.ok: - response.raise_for_status() + response = client.get(f'{AGENTS_PREFIX}/{agent_name}/loss') + response_raise_error_if_any(response) return response.json() -def agent_step(client, agent_name: str, step: Dict) -> None: - response = client.post(f"/agents/{agent_name}/step", step) - if not response.ok: - response.raise_for_status() +def step(client, agent_name: str, step: Dict) -> None: + """Steps forward in agents learning mechanism. + + This method is used to provide learning data, like recent observations and rewards, + and trigger learning mechanism. *Note* that majority of agents perform `step` after + every environment iteration (`step`). + + Parameters: + client (Client): Authenticated client. + agent_name (str): Name of agent. + step (Dict): Data required for the agent to use in learning. + Likely that's `observation`, `next_observation`, `reward`, `action` values and `done` flag. + + """ + response = client.post(f"{AGENTS_PREFIX}/{agent_name}/step", step) + response_raise_error_if_any(response) return -def agent_act(client, agent_name: str, obs: Dict, params: Optional[Dict] = None) -> Dict: - response = client.post(f"/agents/{agent_name}/act", obs, params=params) - if not response.ok: - response.raise_for_status() +def act(client, agent_name: str, obs: Dict, params: Optional[Dict] = None) -> Dict: + """Asks agent about its action on provided observation. + + Parameters: + client (Client): Authenticated client. + agent_name (str): Name of agent. + obs (Dict): Observation from current environment state. + params (Optional dict): Anything useful for the agent to learn, e.g. epsilon greedy value. + + Returns: + Dictionary container actions. + + """ + response = client.post(f"{AGENTS_PREFIX}/{agent_name}/act", obs, params=params) + response_raise_error_if_any(response) return response.json() diff --git a/agents_bar/environments.py b/agents_bar/environments.py index 7390272..9b2f77c 100644 --- a/agents_bar/environments.py +++ b/agents_bar/environments.py @@ -1,9 +1,12 @@ from typing import Any, Dict, List from agents_bar.client import Client +from agents_bar.utils import response_raise_error_if_any +ENV_PREFIX = "/environments" -def environment_get_all(client: Client) -> List[Dict]: + +def get_many(client: Client) -> List[Dict]: """Gets environments belonging to authenticated user. Parameters: @@ -13,12 +16,10 @@ def environment_get_all(client: Client) -> List[Dict]: List of environments. """ - response = client.get('/env/') - if not response.ok: - response.raise_for_status() + response = client.get(f"{ENV_PREFIX}/") return response.json() -def environment_get(client: Client, env_name: str) -> Dict: +def get(client: Client, env_name: str) -> Dict: """Get indepth information about a specific environment. Parameters: @@ -29,12 +30,11 @@ def environment_get(client: Client, env_name: str) -> Dict: Details of an environment. """ - response = client.get(f'/env/{env_name}') - if not response.ok: - response.raise_for_status() + response = client.get(f'{ENV_PREFIX}/{env_name}') + response_raise_error_if_any(response) return response.json() -def environment_create(client: Client, config: Dict) -> Dict: +def create(client: Client, config: Dict) -> Dict: """Creates an environment with specified configuration. Parameters: @@ -45,12 +45,11 @@ def environment_create(client: Client, config: Dict) -> Dict: Details of an environment. """ - response = client.post('/env/', data=config) - if not response.ok: - response.raise_for_status() + response = client.post(f'{ENV_PREFIX}/', data=config) + response_raise_error_if_any(response) return response.json() -def environment_delete(client: Client, env_name: str) -> bool: +def delete(client: Client, env_name: str) -> bool: """Deletes specified environment. Parameters: @@ -61,12 +60,11 @@ def environment_delete(client: Client, env_name: str) -> bool: Whether environment was delete. True if an environment was delete, False if there was no such environment. """ - response = client.delete('/env/' + env_name) - if not response.ok: - response.raise_for_status() + response = client.delete(f'{ENV_PREFIX}/{env_name}') + response_raise_error_if_any(response) return response.status_code == 202 -def environment_reset(client: Client, env_name: str) -> List[float]: +def reset(client: Client, env_name: str) -> List[float]: """Resets the environment to starting position. Parameters: @@ -77,12 +75,12 @@ def environment_reset(client: Client, env_name: str) -> List[float]: Environment state in the starting position. """ - response = client.post(f"/env/{env_name}/reset") - if not response.ok: - response.raise_for_status() + response = client.post(f"{ENV_PREFIX}/{env_name}/reset") + response_raise_error_if_any(response) return response.json() -def environment_step(client: Client, env_name: str, step) -> Dict[str, Any]: + +def step(client: Client, env_name: str, step) -> Dict[str, Any]: """Steps the environment based on provided data. Parameters: @@ -94,7 +92,30 @@ def environment_step(client: Client, env_name: str, step) -> Dict[str, Any]: Environment state after taking provided actions. Consists of "observation", "reward", "done" and "info". """ - response = client.post(f"/env/{env_name}/step", data=step) - if not response.ok: - response.raise_for_status() + response = client.post(f"{ENV_PREFIX}/{env_name}/step", data=step) + response_raise_error_if_any(response) + return response.json() + + +def commit(client: Client, env_name: str) -> Dict[str, Any]: + """Commits last provided data. Must be proceeded by environment `step`. + + Useful when environment requires many agents or when agent is allowed to make mistakes. + + Parameters: + client (Client): Authenticated client. + env_name (str): Name of the environment + + Returns: + Data about the state the environment has transtioned into. + This should be the same as when using `step` with `commit=True`. + + """ + response = client.post(f"{ENV_PREFIX}/{env_name}/commit") + response_raise_error_if_any(response) + return response.json() + +def info(client: Client, env_name: str) -> Dict[str, Any]: + response = client.get(f"{ENV_PREFIX}/{env_name}/info") + response_raise_error_if_any(response) return response.json() diff --git a/agents_bar/experiments.py b/agents_bar/experiments.py new file mode 100644 index 0000000..51d194f --- /dev/null +++ b/agents_bar/experiments.py @@ -0,0 +1,102 @@ +import dataclasses +from typing import Dict, List, Optional + +from agents_bar.client import Client +from agents_bar.types import ExperimentCreate +from agents_bar.utils import response_raise_error_if_any + +EXP_PREFIX = "/experiments" + + +def get_many(client: Client) -> List[Dict]: + """Gets experiments that belong to an authenticated user. + + Parameters: + client (Client): Authenticated client. + + Returns: + List of experiments. + + """ + response = client.get(f"{EXP_PREFIX}/") + return response.json() + +def get(client: Client, exp_name: str) -> Dict: + """Get indepth information about a specific experiment. + + Parameters: + client (Client): Authenticated client. + exp_name (str): Name of experiment. + + Returns: + Details of an experiment. + + """ + response = client.get(f'{EXP_PREFIX}/{exp_name}') + response_raise_error_if_any(response) + return response.json() + +def create(client: Client, experiment_create: ExperimentCreate) -> Dict: + """Creates an experiment with specified configuration. + + Parameters: + client (Client): Authenticated client. + config (dict): Configuration of an experiment. + + Returns: + Details of an experiment. + + """ + response = client.post(f'{EXP_PREFIX}/', data=dataclasses.asdict(experiment_create)) + response_raise_error_if_any(response) + return response.json() + +def delete(client: Client, exp_name: str) -> bool: + """Deletes specified experiment. + + Parameters: + client (Client): Authenticated client. + exp_name (str): Name of the experiment. + + Returns: + Whether experiment was delete. True if an experiment was delete, False if there was no such experiment. + + """ + response = client.delete(f'{EXP_PREFIX}/{exp_name}') + response_raise_error_if_any(response) + return response.status_code == 202 + +def reset(client: Client, exp_name: str) -> str: + """Resets the experiment to starting position. + + Doesn't affect Agent nor Environment. Only resets values related to the Experiment, + like keeping score of last N episodes or managing Epislon value. + + Parameters: + client (Client): Authenticated client. + exp_name (str): Name of the experiment. + + Returns: + Confirmation on reset experiment. + + """ + response = client.post(f"{EXP_PREFIX}/{exp_name}/reset") + response_raise_error_if_any(response) + return response.json() + + +def start(client: Client, exp_name: str, config: Optional[Dict] = None) -> str: + """Starts experiment, i.e. communication between selected Agent and Env entities. + + Parameters: + client (Client): Authenticated client. + exp_name (str): Name of the experiment. + + Returns: + Information about started experiment. + + """ + config = config or {} + response = client.post(f"{EXP_PREFIX}/{exp_name}/start", data=config) + response_raise_error_if_any(response) + return response.text diff --git a/agents_bar/remote_agent.py b/agents_bar/remote_agent.py index 6d266fd..c8c156f 100644 --- a/agents_bar/remote_agent.py +++ b/agents_bar/remote_agent.py @@ -1,15 +1,13 @@ import dataclasses -import json import logging -import os from typing import Any, Dict, Optional, Union -import requests from tenacity import after_log, retry, stop_after_attempt, wait_fixed -from .types import ActionType, EncodedAgentState, ObsType -from .utils import to_list +from agents_bar.client import Client +from .types import ActionType, EncodedAgentState, ObsType +from .utils import response_raise_error_if_any, to_list SUPPORTED_MODELS = ['dqn', 'ppo', 'ddpg', 'rainbow'] #: Supported models @@ -21,7 +19,7 @@ class RemoteAgent: default_url = "https://agents.bar" logger = logging.getLogger("RemoteAgent") - def __init__(self, agent_name: str, description: str = "", **kwargs): + def __init__(self, client: Client, agent_name: str, **kwargs): """ An instance of the agent in the Agents Bar. @@ -35,13 +33,7 @@ def __init__(self, agent_name: str, description: str = "", **kwargs): password (str): Default None. Overrides password from the env variables. """ - self.url = self.__parse_url(kwargs) - # Pop credentials so that they aren't in the app beyond this point - self.__access_token = self.get_access_token( - access_token=kwargs.pop("access_token", None), username=kwargs.pop("username", None), - password=kwargs.pop("password", None) - ) - self._headers = {"Authorization": f"Bearer {self.__access_token}", "accept": "application/json"} + self._client: Client = client self._config: Dict = {} self._config.update(**kwargs) @@ -53,7 +45,7 @@ def __init__(self, agent_name: str, description: str = "", **kwargs): self.loss: Dict[str, float] = {} self.agent_name = agent_name - self.description = description + self._description: Optional[str] = None @property def obs_size(self): @@ -73,40 +65,7 @@ def agent_model(self): self.sync() return self._agent_model - @staticmethod - def __parse_url(kwargs) -> str: - url = kwargs.pop("url", RemoteAgent.default_url) - if url[:4].lower() != "http": - url = "https://" + url - return url + "/api/v1" - - def get_access_token(self, username=None, password=None, access_token=None) -> str: - """Retrieves access token. - - """ - access_token = access_token if access_token is not None else os.environ.get('AGENTS_BAR_ACCESS_TOKEN') - if access_token is None: - access_token = self.__login(username=username, password=password) - return access_token - - def __login(self, username: Optional[str] = None, password: Optional[str] = None): - username = username if username is not None else os.environ.get('AGENTS_BAR_USER') - password = password if password is not None else os.environ.get('AGENTS_BAR_PASS') - if username is None or password is None: - raise ValueError("No credentials provided for logging in. Please pass either 'access_token' or " - "('username' and 'password'). These credentials should be related to your Agents Bar account.") - - data = dict(username=username, password=password) - response = requests.post(f"{self.url}/login/access-token", data=data) - if response.status_code >= 300: - self.logger.error(response.text) - raise ValueError( - f"Received an error while trying to authenticate as username='{username}'. " - f"Please double check your credentials. Error: {response.text}" - ) - return response.json()['access_token'] - - def create_agent(self, obs_size: int, action_size: int, agent_model: str, active: bool = True) -> Dict: + def create_agent(self, obs_size: int, action_size: int, agent_model: str, active: bool = True, description: Optional[str] = None) -> Dict: """Creates a new agent in the service. Uses provided information on RemoteAgent instantiation to create a new agent. @@ -135,15 +94,18 @@ def create_agent(self, obs_size: int, action_size: int, agent_model: str, active self._discrete = None self._config['obs_size'] = obs_size self._config['action_size'] = action_size + self._description = description + self.logger.debug("Creating an agent (name=%s, model=%s)", self.agent_name, self.agent_model) payload = dict( name=self.agent_name, model=self.agent_model, - description=self.description, + description=self._description, config=self._config, is_active=active, ) - response = requests.post(f"{self.url}/agents/", data=json.dumps(payload), headers=self._headers) + response = self._client.post('/agents', data=payload) + # response = requests.post(f"{self.url}/agents/", data=json.dumps(payload), headers=self._headers) if response.status_code >= 300: raise RuntimeError("Unable to create a new agent.\n%s" % response.json()) return response.json() @@ -169,7 +131,7 @@ def remove(self, *, agent_name: str, quite: bool = True) -> bool: raise ValueError("You wanted to delete an agent. Are you sure? If so, we need *again* its name.") self.logger.warning("Agent '%s' is being exterminated", agent_name) - response = requests.delete(f"{self.url}/agents/{agent_name}", headers=self._headers) + response = self._client.delete(f"/agents/{agent_name}") if response.status_code >= 300: raise RuntimeError(f"Error while deleting the agent '{agent_name}'. Message from server: {response.text}") return True @@ -177,12 +139,12 @@ def remove(self, *, agent_name: str, quite: bool = True) -> bool: @property def exists(self): """Whether the agent service exists and is accessible""" - response = requests.get(f"{self.url}/agents/{self.agent_name}", headers=self._headers) + response = self._client.get(f"/agents/{self.agent_name}") return response.ok @property def is_active(self): - response = requests.get(f"{self.url}/agents/{self.agent_name}", headers=self._headers) + response = self._client.get(f"/agents/{self.agent_name}") if not response.ok: response.raise_for_status() agent = response.json() @@ -215,7 +177,7 @@ def make_str_or_number(val): def info(self) -> Dict[str, Any]: """Gets agents meta-data from sever.""" - response = requests.get(f"{self.url}/agents/{self.agent_name}", headers=self._headers) + response = self._client.get(f"/agents/{self.agent_name}") info = response.json() self._config = info.get('config', self._config) return info @@ -242,7 +204,7 @@ def get_state(self) -> EncodedAgentState: Snapshot with config, buffer and network states being encoded. """ - response = requests.get(f"{self.url}/snapshots/{self.agent_name}", headers=self._headers) + response = self._client.get(f"/snapshots/{self.agent_name}") if not response.ok: response.raise_for_status() state = response.json() @@ -259,7 +221,7 @@ def upload_state(self, state: EncodedAgentState) -> bool: """ j_state = dataclasses.asdict(state) - response = requests.post(f"{self.url}/snapshots/{self.agent_name}", json=j_state, headers=self._headers) + response = self._client.post(f"/snapshots/{self.agent_name}", data=j_state) if not response.ok: response.raise_for_status() # Raises return False # Doesn't reach @@ -279,9 +241,7 @@ def act(self, obs, noise: float = 0) -> ActionType: a list of either floats or ints. """ - data = json.dumps(obs) - response = requests.post(f"{self.url}/agents/{self.agent_name}/act", - params={"noise": noise}, data=data, headers=self._headers) + response = self._client.post(f"/agents/{self.agent_name}/act", params={"noise": noise}, data=obs) if not response.ok: response.raise_for_status() # Raises http @@ -290,7 +250,7 @@ def act(self, obs, noise: float = 0) -> ActionType: return int(action[0]) return action - @retry(stop=stop_after_attempt(10), wait=wait_fixed(0.01), after=after_log(global_logger, logging.INFO)) + @retry(stop=stop_after_attempt(5), wait=wait_fixed(0.01), after=after_log(global_logger, logging.INFO), reraise=True) def step(self, obs: ObsType, action: ActionType, reward: float, next_obs: ObsType, done: bool) -> bool: """Providing information from taking a step in environment. @@ -311,9 +271,6 @@ def step(self, obs: ObsType, action: ActionType, reward: float, next_obs: ObsTyp } data = {"step_data": step_data} - response = requests.post(f"{self.url}/agents/{self.agent_name}/step", - data=json.dumps(data), headers=self._headers) - if not response.ok: - response.raise_for_status() # Raises http - return False + response = self._client.post(f"/agents/{self.agent_name}/step", data=data) + response_raise_error_if_any(response) return True diff --git a/agents_bar/types.py b/agents_bar/types.py index f93e433..7d39a1f 100644 --- a/agents_bar/types.py +++ b/agents_bar/types.py @@ -1,6 +1,8 @@ -from dataclasses import dataclass -from typing import List, Union +from dataclasses import Field, dataclass +from typing import Any, List, Dict, Optional, Union +ObsType = List[float] +ActionType = Union[int, List[Union[int, float]]] @dataclass class EncodedAgentState: @@ -11,5 +13,12 @@ class EncodedAgentState: encoded_network: str encoded_buffer: str -ObsType = List[float] -ActionType = Union[int, List[Union[int, float]]] + +@dataclass +class ExperimentCreate: + name: str + agent_name: str + environment_name: str + config: Dict[str, Any] + description: Optional[str] = None + is_active: Optional[bool] = True diff --git a/agents_bar/utils.py b/agents_bar/utils.py index 19ce9a8..8750c6f 100644 --- a/agents_bar/utils.py +++ b/agents_bar/utils.py @@ -1,6 +1,9 @@ import time from typing import List +import requests +from requests.models import HTTPError + def wait_until_agent_is_active(agent, max_seconds: int = 20, verbose: bool = True) -> bool: """ @@ -79,3 +82,14 @@ def to_list(x: object) -> List: return [x] # Just hoping... return list(x) + + +def response_raise_error_if_any(response: requests.Response) -> None: + """ + Checks if there is any error while make a request. + If status 400+ then raises HTTPError with provided reason. + """ + try: + response.raise_for_status() + except requests.exceptions.HTTPError as e: + raise HTTPError({"error": str(e), "reason": response.json()['detail']}) from None diff --git a/setup.cfg b/setup.cfg index 7964557..1170087 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,6 +1,6 @@ [metadata] name = agents-bar -version = 0.3.0 +version = 0.4.0 author = Dawid Laszuk author_email = agents-bar-client-python@dawid.lasz.uk description = A client to work with Agents Bar (https://agents.bar).