diff --git a/python/ray/actor.py b/python/ray/actor.py
index 7085cb9d6..e380dcc9f 100644
--- a/python/ray/actor.py
+++ b/python/ray/actor.py
@@ -95,7 +95,7 @@ class ActorMethod:
             self._actor_hard_ref = None
 
     def __call__(self, *args, **kwargs):
-        raise Exception("Actor methods cannot be called directly. Instead "
+        raise TypeError("Actor methods cannot be called directly. Instead "
                         "of running 'object.{}()', try "
                         "'object.{}.remote()'.".format(self._method_name,
                                                        self._method_name))
@@ -158,7 +158,7 @@ class ActorClassMethodMetadata(object):
 
     def __init__(self):
         class_name = type(self).__name__
-        raise Exception("{} can not be constructed directly, "
+        raise TypeError("{} can not be constructed directly, "
                         "instead of running '{}()', try '{}.create()'".format(
                             class_name, class_name, class_name))
 
@@ -310,7 +310,7 @@ class ActorClass:
         Raises:
             Exception: Always.
         """
-        raise Exception("Actors cannot be instantiated directly. "
+        raise TypeError("Actors cannot be instantiated directly. "
                         "Instead of '{}()', use '{}.remote()'.".format(
                             self.__ray_metadata__.class_name,
                             self.__ray_metadata__.class_name))
@@ -465,13 +465,13 @@ class ActorClass:
 
         worker = ray.worker.get_global_worker()
         if worker.mode is None:
-            raise Exception("Actors cannot be created before ray.init() "
-                            "has been called.")
+            raise RuntimeError("Actors cannot be created before ray.init() "
+                               "has been called.")
 
         if detached and name is None:
-            raise Exception("Detached actors must be named. "
-                            "Please use Actor._remote(name='some_name') "
-                            "to associate the name.")
+            raise ValueError("Detached actors must be named. "
+                             "Please use Actor._remote(name='some_name') "
+                             "to associate the name.")
 
         # Check whether the name is already taken.
         if name is not None:
@@ -730,7 +730,7 @@ class ActorHandle:
 
             class FakeActorMethod(object):
                 def __call__(self, *args, **kwargs):
-                    raise Exception(
+                    raise TypeError(
                         "Actor methods cannot be called directly. Instead "
                         "of running 'object.{}()', try 'object.{}.remote()'.".
                         format(item, item))
@@ -872,7 +872,7 @@ def modify_class(cls):
             """
             worker = ray.worker.global_worker
             if not isinstance(self, ray.actor.Checkpointable):
-                raise Exception(
+                raise TypeError(
                     "__ray_checkpoint__.remote() may only be called on actors "
                     "that implement ray.actor.Checkpointable")
             return worker._save_actor_checkpoint()
@@ -902,9 +902,9 @@ def make_actor(cls, num_cpus, num_gpus, memory, object_store_memory, resources,
 
     if not (ray_constants.NO_RECONSTRUCTION <= max_reconstructions <=
             ray_constants.INFINITE_RECONSTRUCTION):
-        raise Exception("max_reconstructions must be in range [%d, %d]." %
-                        (ray_constants.NO_RECONSTRUCTION,
-                         ray_constants.INFINITE_RECONSTRUCTION))
+        raise ValueError("max_reconstructions must be in range [%d, %d]." %
+                         (ray_constants.NO_RECONSTRUCTION,
+                          ray_constants.INFINITE_RECONSTRUCTION))
 
     return ActorClass._ray_from_modified_class(
         Class, ActorClassID.from_random(), max_reconstructions, num_cpus,
@@ -934,7 +934,7 @@ def exit_actor():
         raise exit
         assert False, "This process should have terminated."
     else:
-        raise Exception("exit_actor called on a non-actor worker.")
+        raise TypeError("exit_actor called on a non-actor worker.")
 
 
 ray.worker.global_worker.make_actor = make_actor
diff --git a/python/ray/cluster_utils.py b/python/ray/cluster_utils.py
index 0fa997c1f..f5a4aba41 100644
--- a/python/ray/cluster_utils.py
+++ b/python/ray/cluster_utils.py
@@ -141,8 +141,8 @@ class Cluster:
                 exception.
 
         Raises:
-            Exception: An exception is raised if the timeout expires before the
-                node appears in the client table.
+            TimeoutError: An exception is raised if the timeout expires before
+                the node appears in the client table.
         """
         ip_address, port = self.redis_address.split(":")
         redis_client = redis.StrictRedis(
@@ -158,7 +158,7 @@ class Cluster:
                 return
             else:
                 time.sleep(0.1)
-        raise Exception("Timed out while waiting for nodes to join.")
+        raise TimeoutError("Timed out while waiting for nodes to join.")
 
     def wait_for_nodes(self, timeout=30):
         """Waits for correct number of nodes to be registered.
@@ -174,8 +174,8 @@ class Cluster:
                 before failing.
 
         Raises:
-            Exception: An exception is raised if we time out while waiting for
-                nodes to join.
+            TimeoutError: An exception is raised if we time out while waiting
+                for nodes to join.
         """
         ip_address, port = self.address.split(":")
         redis_client = redis.StrictRedis(
@@ -195,7 +195,7 @@ class Cluster:
                     "{} nodes are currently registered, but we are expecting "
                     "{}".format(len(live_clients), expected))
                 time.sleep(0.1)
-        raise Exception("Timed out while waiting for nodes to join.")
+        raise TimeoutError("Timed out while waiting for nodes to join.")
 
     def list_all_nodes(self):
         """Lists all nodes.
diff --git a/python/ray/cross_language.py b/python/ray/cross_language.py
index 90cf3a239..d07e75c4d 100644
--- a/python/ray/cross_language.py
+++ b/python/ray/cross_language.py
@@ -23,10 +23,10 @@ def format_args(worker, args, kwargs):
         List of args and kwargs (if supported).
     """
     if not worker.load_code_from_local:
-        raise Exception("Cross language feature needs "
-                        "--load-code-from-local to be set.")
+        raise ValueError("Cross language feature needs "
+                         "--load-code-from-local to be set.")
     if kwargs:
-        raise Exception("Cross language remote functions "
+        raise TypeError("Cross language remote functions "
                         "does not support kwargs.")
     return args
 
diff --git a/python/ray/function_manager.py b/python/ray/function_manager.py
index 570ce6db8..e3b212312 100644
--- a/python/ray/function_manager.py
+++ b/python/ray/function_manager.py
@@ -172,7 +172,7 @@ class FunctionActorManager:
         # This is a placeholder in case the function can't be unpickled. This
         # will be overwritten if the function is successfully registered.
         def f(*args, **kwargs):
-            raise Exception("This function was not imported properly.")
+            raise RuntimeError("This function was not imported properly.")
 
         # This function is called by ImportThread. This operation needs to be
         # atomic. Otherwise, there is race condition. Another thread may use
@@ -275,7 +275,7 @@ class FunctionActorManager:
             self._num_task_executions[job_id][function_id] = 0
         except Exception:
             logger.exception("Failed to load function %s.", function_name)
-            raise Exception(
+            raise RuntimeError(
                 "Function {} failed to be loaded from local code.".format(
                     function_descriptor))
 
@@ -447,7 +447,7 @@ class FunctionActorManager:
                 return actor_class
         except Exception:
             logger.exception("Failed to load actor_class %s.", class_name)
-            raise Exception(
+            raise RuntimeError(
                 "Actor {} failed to be imported from local code.".format(
                     class_name))
 
@@ -456,7 +456,7 @@ class FunctionActorManager:
             pass
 
         def temporary_actor_method(*args, **kwargs):
-            raise Exception(
+            raise RuntimeError(
                 "The actor with name {} failed to be imported, "
                 "and so cannot execute this method.".format(actor_class_name))
 
diff --git a/python/ray/import_thread.py b/python/ray/import_thread.py
index 47b766b8d..f1da987eb 100644
--- a/python/ray/import_thread.py
+++ b/python/ray/import_thread.py
@@ -150,7 +150,7 @@ class ImportThread:
         # TODO(rkn): We may need to bring back the case of
         # fetching actor classes here.
         else:
-            raise Exception("This code should be unreachable.")
+            assert False, "This code should be unreachable."
 
     def fetch_and_execute_function_to_run(self, key):
         """Run on arbitrary function on the worker."""
diff --git a/python/ray/monitor.py b/python/ray/monitor.py
index b325205b7..7c6675847 100644
--- a/python/ray/monitor.py
+++ b/python/ray/monitor.py
@@ -234,7 +234,7 @@ class Monitor:
                       ray.ray_constants.AUTOSCALER_RESOURCE_REQUEST_CHANNEL):
                     message_handler = self.autoscaler_resource_request_handler
                 else:
-                    raise Exception("This code should be unreachable.")
+                    assert False, "This code should be unreachable."
 
                 # Call the handler.
                 message_handler(channel, data)
diff --git a/python/ray/node.py b/python/ray/node.py
index 28030d8fa..6673e095b 100644
--- a/python/ray/node.py
+++ b/python/ray/node.py
@@ -397,7 +397,8 @@ class Node:
         """
         if socket_path is not None:
             if os.path.exists(socket_path):
-                raise Exception("Socket file {} exists!".format(socket_path))
+                raise RuntimeError(
+                    "Socket file {} exists!".format(socket_path))
             socket_dir = os.path.dirname(socket_path)
             try_to_create_directory(socket_dir)
             return socket_path
@@ -677,9 +678,10 @@ class Node:
             # Handle the case where the process has already exited.
             if process.poll() is not None:
                 if check_alive:
-                    raise Exception("Attempting to kill a process of type "
-                                    "'{}', but this process is already dead."
-                                    .format(process_type))
+                    raise RuntimeError(
+                        "Attempting to kill a process of type "
+                        "'{}', but this process is already dead."
+                        .format(process_type))
                 else:
                     continue
 
@@ -696,7 +698,7 @@ class Node:
                     if process_info.stderr_file is not None:
                         with open(process_info.stderr_file, "r") as f:
                             message += "\nPROCESS STDERR:\n" + f.read()
-                    raise Exception(message)
+                    raise RuntimeError(message)
                 continue
 
             if process_info.use_valgrind_profiler:
diff --git a/python/ray/remote_function.py b/python/ray/remote_function.py
index 02f3807f9..948fa305b 100644
--- a/python/ray/remote_function.py
+++ b/python/ray/remote_function.py
@@ -97,7 +97,7 @@ class RemoteFunction:
         self.remote = _remote_proxy
 
     def __call__(self, *args, **kwargs):
-        raise Exception("Remote functions cannot be called directly. Instead "
+        raise TypeError("Remote functions cannot be called directly. Instead "
                         "of running '{}()', try '{}.remote()'.".format(
                             self._function_name, self._function_name))
 
diff --git a/python/ray/resource_spec.py b/python/ray/resource_spec.py
index 082b6fc37..142e65239 100644
--- a/python/ray/resource_spec.py
+++ b/python/ray/resource_spec.py
@@ -144,9 +144,9 @@ class ResourceSpec(
         # excede the amount allowed by CUDA_VISIBLE_DEVICES.
         if (num_gpus is not None and gpu_ids is not None
                 and num_gpus > len(gpu_ids)):
-            raise Exception("Attempting to start raylet with {} GPUs, "
-                            "but CUDA_VISIBLE_DEVICES contains {}.".format(
-                                num_gpus, gpu_ids))
+            raise ValueError("Attempting to start raylet with {} GPUs, "
+                             "but CUDA_VISIBLE_DEVICES contains {}.".format(
+                                 num_gpus, gpu_ids))
         if num_gpus is None:
             # Try to automatically detect the number of GPUs.
             num_gpus = _autodetect_num_gpus()
diff --git a/python/ray/services.py b/python/ray/services.py
index ec8df1cda..93039528e 100644
--- a/python/ray/services.py
+++ b/python/ray/services.py
@@ -143,7 +143,7 @@ def get_address_info_from_redis_helper(redis_address,
 
     client_table = ray.state._parse_client_table(redis_client)
     if len(client_table) == 0:
-        raise Exception(
+        raise RuntimeError(
             "Redis has started but no raylets have registered yet.")
 
     relevant_client = None
@@ -155,7 +155,7 @@ def get_address_info_from_redis_helper(redis_address,
             relevant_client = client_info
             break
     if relevant_client is None:
-        raise Exception(
+        raise RuntimeError(
             "Redis has started but no raylets have registered yet.")
 
     return {
@@ -207,8 +207,8 @@ def remaining_processes_alive():
             ray.init().
     """
     if ray.worker._global_node is None:
-        raise Exception("This process is not in a position to determine "
-                        "whether all processes are alive or not.")
+        raise RuntimeError("This process is not in a position to determine "
+                           "whether all processes are alive or not.")
     return ray.worker._global_node.remaining_processes_alive()
 
 
@@ -526,9 +526,9 @@ def wait_for_redis_to_start(redis_ip_address,
         else:
             break
     if counter == num_retries:
-        raise Exception("Unable to connect to Redis. If the Redis instance is "
-                        "on a different machine, check that your firewall is "
-                        "configured properly.")
+        raise RuntimeError("Unable to connect to Redis. If the Redis instance "
+                           "is on a different machine, check that your "
+                           "firewall is configured properly.")
 
 
 def _compute_version_info():
@@ -586,7 +586,7 @@ def check_version_info(redis_client):
                          "    Ray: " + version_info[0] + "\n"
                          "    Python: " + version_info[1] + "\n")
         if version_info[:2] != true_version_info[:2]:
-            raise Exception(error_message)
+            raise RuntimeError(error_message)
         else:
             logger.warning(error_message)
 
@@ -682,8 +682,8 @@ def start_redis(node_ip_address,
     if redis_shard_ports is None:
         redis_shard_ports = num_redis_shards * [None]
     elif len(redis_shard_ports) != num_redis_shards:
-        raise Exception("The number of Redis shard ports does not match the "
-                        "number of Redis shards.")
+        raise RuntimeError("The number of Redis shard ports does not match "
+                           "the number of Redis shards.")
 
     processes = []
 
@@ -693,10 +693,10 @@ def start_redis(node_ip_address,
         if password is not None:
             # TODO(pschafhalter) remove this once credis supports
             # authenticating Redis ports
-            raise Exception("Setting the `redis_password` argument is not "
-                            "supported in credis. To run Ray with "
-                            "password-protected Redis ports, ensure that "
-                            "the environment variable `RAY_USE_NEW_GCS=off`.")
+            raise ValueError("Setting the `redis_password` argument is not "
+                             "supported in credis. To run Ray with "
+                             "password-protected Redis ports, ensure that "
+                             "the environment variable `RAY_USE_NEW_GCS=off`.")
         assert num_redis_shards == 1, (
             "For now, RAY_USE_NEW_GCS supports 1 shard, and credis "
             "supports 1-node chain for that shard only.")
@@ -909,8 +909,9 @@ def _start_redis_instance(executable,
         port = new_port()
         counter += 1
     if counter == num_retries:
-        raise Exception("Couldn't start Redis. Check log files: {} {}".format(
-            stdout_file.name, stderr_file.name))
+        raise RuntimeError("Couldn't start Redis. "
+                           "Check log files: {} {}".format(
+                               stdout_file.name, stderr_file.name))
 
     # Create a Redis client just for configuring Redis.
     redis_client = redis.StrictRedis(
@@ -1227,7 +1228,7 @@ def start_raylet(redis_address,
     config_str = ",".join(["{},{}".format(*kv) for kv in config.items()])
 
     if use_valgrind and use_profiler:
-        raise Exception("Cannot use valgrind and profiler at the same time.")
+        raise ValueError("Cannot use valgrind and profiler at the same time.")
 
     assert resource_spec.resolved()
     num_initial_workers = resource_spec.num_cpus
@@ -1321,9 +1322,9 @@ def get_ray_jars_dir():
     current_dir = os.path.abspath(os.path.dirname(__file__))
     jars_dir = os.path.abspath(os.path.join(current_dir, "jars"))
     if not os.path.exists(jars_dir):
-        raise Exception("Ray jars is not packaged into ray. "
-                        "Please build ray with java enabled "
-                        "(set env var RAY_INSTALL_JAVA=1)")
+        raise RuntimeError("Ray jars is not packaged into ray. "
+                           "Please build ray with java enabled "
+                           "(set env var RAY_INSTALL_JAVA=1)")
     return os.path.abspath(os.path.join(current_dir, "jars"))
 
 
@@ -1446,7 +1447,7 @@ def determine_plasma_store_config(object_store_memory,
 
         # Do some sanity checks.
         if object_store_memory > system_memory:
-            raise Exception(
+            raise ValueError(
                 "The requested object store memory size is greater "
                 "than the total available memory.")
     else:
@@ -1455,7 +1456,7 @@ def determine_plasma_store_config(object_store_memory,
                        "plasma_directory is set.")
 
     if not os.path.isdir(plasma_directory):
-        raise Exception(
+        raise ValueError(
             "The file {} does not exist or is not a directory.".format(
                 plasma_directory))
 
@@ -1496,16 +1497,16 @@ def _start_plasma_store(plasma_store_memory,
             plasma store process.
     """
     if use_valgrind and use_profiler:
-        raise Exception("Cannot use valgrind and profiler at the same time.")
+        raise ValueError("Cannot use valgrind and profiler at the same time.")
 
     if huge_pages and not (sys.platform == "linux"
                            or sys.platform == "linux2"):
-        raise Exception("The huge_pages argument is only supported on "
-                        "Linux.")
+        raise ValueError("The huge_pages argument is only supported on "
+                         "Linux.")
 
     if huge_pages and plasma_directory is None:
-        raise Exception("If huge_pages is True, then the "
-                        "plasma_directory argument must be provided.")
+        raise ValueError("If huge_pages is True, then the "
+                         "plasma_directory argument must be provided.")
 
     if not isinstance(plasma_store_memory, int):
         plasma_store_memory = int(plasma_store_memory)
diff --git a/python/ray/signature.py b/python/ray/signature.py
index eaa2beb9e..20e49d93b 100644
--- a/python/ray/signature.py
+++ b/python/ray/signature.py
@@ -96,9 +96,9 @@ def extract_signature(func, ignore_first=False):
 
     if ignore_first:
         if len(signature_parameters) == 0:
-            raise Exception("Methods must take a 'self' argument, but the "
-                            "method '{}' does not have one.".format(
-                                func.__name__))
+            raise ValueError("Methods must take a 'self' argument, but the "
+                             "method '{}' does not have one.".format(
+                                 func.__name__))
         signature_parameters = signature_parameters[1:]
 
     return _scrub_parameters(signature_parameters)
diff --git a/python/ray/state.py b/python/ray/state.py
index 31d172861..00622f3b2 100644
--- a/python/ray/state.py
+++ b/python/ray/state.py
@@ -105,8 +105,8 @@ def _parse_resource_table(redis_client, client_id):
     gcs_entry = gcs_utils.GcsEntry.FromString(message)
     entries_len = len(gcs_entry.entries)
     if entries_len % 2 != 0:
-        raise Exception("Invalid entry size for resource lookup: " +
-                        str(entries_len))
+        raise ValueError("Invalid entry size for resource lookup: " +
+                         str(entries_len))
 
     for i in range(0, entries_len, 2):
         resource_table_data = gcs_utils.ResourceTableData.FromString(
@@ -139,16 +139,16 @@ class GlobalState:
         """Check that the object has been initialized before it is used.
 
         Raises:
-            Exception: An exception is raised if ray.init() has not been called
-                yet.
+            RuntimeError: An exception is raised if ray.init() has not been
+                called yet.
         """
         if self.redis_client is None:
-            raise Exception("The ray global state API cannot be used before "
-                            "ray.init has been called.")
+            raise RuntimeError("The ray global state API cannot be used "
+                               "before ray.init has been called.")
 
         if self.redis_clients is None:
-            raise Exception("The ray global state API cannot be used before "
-                            "ray.init has been called.")
+            raise RuntimeError("The ray global state API cannot be used "
+                               "before ray.init has been called.")
 
     def disconnect(self):
         """Disconnect global state from GCS."""
@@ -184,9 +184,9 @@ class GlobalState:
                 time.sleep(1)
                 continue
             num_redis_shards = int(num_redis_shards)
-            if num_redis_shards < 1:
-                raise Exception("Expected at least one Redis shard, found "
-                                "{}.".format(num_redis_shards))
+            assert num_redis_shards >= 1, (
+                "Expected at least one Redis "
+                "shard, found {}.".format(num_redis_shards))
 
             # Attempt to get all of the Redis shards.
             redis_shard_addresses = self.redis_client.lrange(
@@ -201,10 +201,10 @@ class GlobalState:
 
         # Check to see if we timed out.
         if time.time() - start_time >= timeout:
-            raise Exception("Timed out while attempting to initialize the "
-                            "global state. num_redis_shards = {}, "
-                            "redis_shard_addresses = {}".format(
-                                num_redis_shards, redis_shard_addresses))
+            raise TimeoutError("Timed out while attempting to initialize the "
+                               "global state. num_redis_shards = {}, "
+                               "redis_shard_addresses = {}".format(
+                                   num_redis_shards, redis_shard_addresses))
 
         # Get the rest of the information.
         self.redis_clients = []
diff --git a/python/ray/worker.py b/python/ray/worker.py
index cc345c037..79f092574 100644
--- a/python/ray/worker.py
+++ b/python/ray/worker.py
@@ -439,8 +439,8 @@ def get_gpu_ids():
         A list of GPU IDs.
     """
     if _mode() == LOCAL_MODE:
-        raise Exception("ray.get_gpu_ids() currently does not work in LOCAL "
-                        "MODE.")
+        raise RuntimeError("ray.get_gpu_ids() currently does not work in "
+                           "local_mode.")
 
     all_resource_ids = global_worker.core_worker.resource_ids()
     assigned_ids = [
@@ -466,9 +466,8 @@ def get_resource_ids():
         resource reserved for this worker.
     """
     if _mode() == LOCAL_MODE:
-        raise Exception(
-            "ray.get_resource_ids() currently does not work in LOCAL "
-            "MODE.")
+        raise RuntimeError("ray.get_resource_ids() currently does not work in "
+                           "local_mode.")
 
     return global_worker.core_worker.resource_ids()
 
@@ -482,7 +481,7 @@ def get_webui_url():
         The URL of the web UI as a string.
     """
     if _global_node is None:
-        raise Exception("Ray has not been initialized/connected.")
+        raise RuntimeError("Ray has not been initialized/connected.")
     return _global_node.webui_url
 
 
@@ -687,10 +686,10 @@ def init(address=None,
                          "called.")
             return
         else:
-            raise Exception("Perhaps you called ray.init twice by accident? "
-                            "This error can be suppressed by passing in "
-                            "'ignore_reinit_error=True' or by calling "
-                            "'ray.shutdown()' prior to 'ray.init()'.")
+            raise RuntimeError("Maybe you called ray.init twice by accident? "
+                               "This error can be suppressed by passing in "
+                               "'ignore_reinit_error=True' or by calling "
+                               "'ray.shutdown()' prior to 'ray.init()'.")
 
     # Convert hostnames to numerical IP address.
     if node_ip_address is not None:
@@ -743,44 +742,45 @@ def init(address=None,
     else:
         # In this case, we are connecting to an existing cluster.
         if num_cpus is not None or num_gpus is not None:
-            raise Exception("When connecting to an existing cluster, num_cpus "
-                            "and num_gpus must not be provided.")
+            raise ValueError(
+                "When connecting to an existing cluster, num_cpus "
+                "and num_gpus must not be provided.")
         if resources is not None:
-            raise Exception("When connecting to an existing cluster, "
-                            "resources must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "resources must not be provided.")
         if num_redis_shards is not None:
-            raise Exception("When connecting to an existing cluster, "
-                            "num_redis_shards must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "num_redis_shards must not be provided.")
         if redis_max_clients is not None:
-            raise Exception("When connecting to an existing cluster, "
-                            "redis_max_clients must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "redis_max_clients must not be provided.")
         if memory is not None:
-            raise Exception("When connecting to an existing cluster, "
-                            "memory must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "memory must not be provided.")
         if object_store_memory is not None:
-            raise Exception("When connecting to an existing cluster, "
-                            "object_store_memory must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "object_store_memory must not be provided.")
         if redis_max_memory is not None:
-            raise Exception("When connecting to an existing cluster, "
-                            "redis_max_memory must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "redis_max_memory must not be provided.")
         if plasma_directory is not None:
-            raise Exception("When connecting to an existing cluster, "
-                            "plasma_directory must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "plasma_directory must not be provided.")
         if huge_pages:
-            raise Exception("When connecting to an existing cluster, "
-                            "huge_pages must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "huge_pages must not be provided.")
         if temp_dir is not None:
-            raise Exception("When connecting to an existing cluster, "
-                            "temp_dir must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "temp_dir must not be provided.")
         if plasma_store_socket_name is not None:
-            raise Exception("When connecting to an existing cluster, "
-                            "plasma_store_socket_name must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "plasma_store_socket_name must not be provided.")
         if raylet_socket_name is not None:
-            raise Exception("When connecting to an existing cluster, "
-                            "raylet_socket_name must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "raylet_socket_name must not be provided.")
         if _internal_config is not None:
-            raise Exception("When connecting to an existing cluster, "
-                            "_internal_config must not be provided.")
+            raise ValueError("When connecting to an existing cluster, "
+                             "_internal_config must not be provided.")
 
         # In this case, we only need to connect the node.
         ray_params = ray.parameter.RayParams(
@@ -1604,13 +1604,13 @@ def wait(object_ids, num_returns=1, timeout=None):
             return [], []
 
         if len(object_ids) != len(set(object_ids)):
-            raise Exception("Wait requires a list of unique object IDs.")
+            raise ValueError("Wait requires a list of unique object IDs.")
         if num_returns <= 0:
-            raise Exception(
+            raise ValueError(
                 "Invalid number of objects to return %d." % num_returns)
         if num_returns > len(object_ids):
-            raise Exception("num_returns cannot be greater than the number "
-                            "of objects provided to ray.wait.")
+            raise ValueError("num_returns cannot be greater than the number "
+                             "of objects provided to ray.wait.")
 
         timeout = timeout if timeout is not None else 10**6
         timeout_milliseconds = int(timeout * 1000)
@@ -1676,8 +1676,8 @@ def make_decorator(num_return_vals=None,
                 or is_cython(function_or_class)):
             # Set the remote function default resources.
             if max_reconstructions is not None:
-                raise Exception("The keyword 'max_reconstructions' is not "
-                                "allowed for remote functions.")
+                raise ValueError("The keyword 'max_reconstructions' is not "
+                                 "allowed for remote functions.")
 
             return ray.remote_function.RemoteFunction(
                 Language.PYTHON, function_or_class, None, num_cpus, num_gpus,
@@ -1686,17 +1686,17 @@ def make_decorator(num_return_vals=None,
 
         if inspect.isclass(function_or_class):
             if num_return_vals is not None:
-                raise Exception("The keyword 'num_return_vals' is not allowed "
-                                "for actors.")
+                raise TypeError("The keyword 'num_return_vals' is not "
+                                "allowed for actors.")
             if max_calls is not None:
-                raise Exception("The keyword 'max_calls' is not allowed for "
-                                "actors.")
+                raise TypeError("The keyword 'max_calls' is not "
+                                "allowed for actors.")
 
             return worker.make_actor(function_or_class, num_cpus, num_gpus,
                                      memory, object_store_memory, resources,
                                      max_reconstructions)
 
-        raise Exception("The @ray.remote decorator must be applied to "
+        raise TypeError("The @ray.remote decorator must be applied to "
                         "either a function or to a class.")
 
     return decorator
@@ -1815,7 +1815,7 @@ def remote(*args, **kwargs):
     num_gpus = kwargs["num_gpus"] if "num_gpus" in kwargs else None
     resources = kwargs.get("resources")
     if not isinstance(resources, dict) and resources is not None:
-        raise Exception("The 'resources' keyword argument must be a "
+        raise TypeError("The 'resources' keyword argument must be a "
                         "dictionary, but received type {}.".format(
                             type(resources)))
     if resources is not None: