We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
bug: unable to start BentoML container in RayServe
### Describe the bug

followed instructions at https://docs.bentoml.org/en/latest/integrations/ray.html

ray serve attempts to instantiate container, but trips up:

2023-10-30 23:45:22,446 INFO scripts.py:471 -- Running import path: 'bento_ray:classifier'.
Traceback (most recent call last):
  File "/home/thor/.local/bin/serve", line 8, in <module>
    sys.exit(cli())
  File "/home/thor/.local/lib/python3.8/site-packages/click/core.py", line 1157, in __call__
    return self.main(*args, **kwargs)
  File "/home/thor/.local/lib/python3.8/site-packages/click/core.py", line 1078, in main
    rv = self.invoke(ctx)
  File "/home/thor/.local/lib/python3.8/site-packages/click/core.py", line 1688, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/home/thor/.local/lib/python3.8/site-packages/click/core.py", line 1434, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/home/thor/.local/lib/python3.8/site-packages/click/core.py", line 783, in invoke
    return __callback(*args, **kwargs)
  File "/home/thor/.local/lib/python3.8/site-packages/ray/serve/scripts.py", line 473, in run
    import_attr(import_path), args_dict
  File "/home/thor/.local/lib/python3.8/site-packages/ray/_private/utils.py", line 1378, in import_attr
    module = importlib.import_module(module_name)
  File "/usr/lib/python3.8/importlib/__init__.py", line 127, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
  File "<frozen importlib._bootstrap>", line 991, in _find_and_load
  File "<frozen importlib._bootstrap>", line 975, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 671, in _load_unlocked
  File "<frozen importlib._bootstrap_external>", line 848, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/mnt/d/Data/AIRay/./bento_ray.py", line 5, in <module>
    classifier = bentoml.ray.deployment(
  File "/home/thor/.local/lib/python3.8/site-packages/bentoml/_internal/utils/lazy_loader.py", line 69, in __getattr__
    self._module = self._load()
  File "/home/thor/.local/lib/python3.8/site-packages/bentoml/_internal/utils/lazy_loader.py", line 53, in _load
    raise self._exc(f"{self._exc_msg} (reason: {err})") from None
bentoml.exceptions.MissingDependencyException: None (reason: No module named 'bentoml._internal.ray._internal')


### To reproduce

_No response_

### Expected behavior

Expected RayServe to spin up BentoML container within its cluster

### Environment

Ubuntu 20.04.6

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/bentoml/_internal/ray/__init__.py`
Content:
```
1 from __future__ import annotations
2 
3 import typing as t
4 from functools import partial
5 
6 import requests
7 
8 from ._internal.utils import pkg
9 
10 if pkg.pkg_version_info("ray")[:2] >= (2, 5):
11     from ray.serve._private.http_util import BufferedASGISender as ASGIHTTPSender
12 else:
13     from ray.serve._private.http_util import ASGIHTTPSender
14 
15 import bentoml
16 from bentoml import Tag
17 
18 from ...exceptions import MissingDependencyException
19 from ..runner.container import AutoContainer
20 from ..runner.utils import Params
21 from .runner_handle import RayRunnerHandle
22 
23 try:
24     from ray import serve
25     from ray.runtime_env import RuntimeEnv
26     from ray.serve.deployment import Deployment
27 except ImportError:  # pragma: no cover
28     raise MissingDependencyException(
29         """'ray[serve]' is required in order to use module 'bentoml.ray', install with 'pip install -U "ray[serve]"'. See https://docs.ray.io/ for more information.""",
30     )
31 
32 
33 def _get_runner_deployment(
34     svc: bentoml.Service,
35     runner_name: str,
36     runner_deployment_config: dict[str, t.Any],
37     enable_batching: bool,
38     batching_config: dict[str, dict[str, float | int]],
39 ) -> Deployment:
40     class RunnerDeployment:
41         def __init__(self):
42             _runner = next(
43                 (runner for runner in svc.runners if runner.name == runner_name),
44                 None,
45             )
46             _runner.init_local(quiet=True)
47             for method in _runner.runner_methods:
48                 if enable_batching and method.config.batchable:
49                     inp_batch_dim = method.config.batch_dim[0]
50                     out_batch_dim = method.config.batch_dim[1]
51                     ray_batch_args = (
52                         batching_config[method.name]
53                         if method.name in batching_config
54                         else {}
55                     )
56 
57                     @serve.batch(**ray_batch_args)
58                     async def _func(self, *args, **kwargs):
59                         params = Params(*args, **kwargs).map(
60                             partial(
61                                 AutoContainer.batches_to_batch, batch_dim=inp_batch_dim
62                             )
63                         )
64                         run_params = params.map(lambda arg: arg[0])
65                         indices = next(iter(params.items()))[1][1]
66                         results = await getattr(_runner, method.name).async_run(
67                             *run_params.args, **run_params.kwargs
68                         )
69                         return AutoContainer.batch_to_batches(
70                             results, indices=indices, batch_dim=out_batch_dim
71                         )
72 
73                     setattr(RunnerDeployment, method.name, _func)
74                 else:
75 
76                     async def _func(self, *args, **kwargs):
77                         return await getattr(_runner, method.name).async_run(
78                             *args, **kwargs
79                         )
80 
81                     setattr(RunnerDeployment, method.name, _func)
82 
83     return serve.deployment(
84         RunnerDeployment, name=f"bento-runner-{runner_name}", **runner_deployment_config
85     )
86 
87 
88 def _get_service_deployment(svc: bentoml.Service, **kwargs) -> Deployment:
89     @serve.deployment(name=f"bento-svc-{svc.name}", **kwargs)
90     class BentoDeployment:
91         def __init__(self, **runner_deployments: dict[str, Deployment]):
92             from ..server.http_app import HTTPAppFactory
93 
94             # Use Ray's metrics setup. Ray has already initialized a prometheus client
95             # for multi-process and this conflicts with BentoML's prometheus setup
96             self.app = HTTPAppFactory(svc, enable_metrics=False)()
97 
98             for runner in svc.runners:
99                 assert runner.name in runner_deployments
100                 runner._set_handle(RayRunnerHandle, runner_deployments[runner.name])
101 
102         async def __call__(self, request: requests.Request):
103             sender = ASGIHTTPSender()
104             await self.app(request.scope, receive=request.receive, send=sender)
105             return sender.build_asgi_response()
106 
107     return BentoDeployment
108 
109 
110 def get_bento_runtime_env(bento_tag: str | Tag) -> RuntimeEnv:
111     # TODO: create Ray RuntimeEnv from Bento env configs
112     return RuntimeEnv(
113         pip=[],
114         conda=[],
115         env_vars=[],
116     )
117 
118 
119 def _deploy_bento_runners(
120     svc: bentoml.Service,
121     runners_deployment_config_map: dict | None = None,
122     enable_batching: bool = False,
123     batching_config: dict | None = None,
124 ) -> dict[str, Deployment]:
125     # Deploy BentoML Runners as Ray serve deployments
126     runners_deployment_config_map = runners_deployment_config_map or {}
127     batching_config = batching_config or {}
128     runner_deployments = {}
129     for runner in svc.runners:
130         runner_deployment_config = (
131             runners_deployment_config_map.get(runner.name)
132             if runner.name in runners_deployment_config_map
133             else {}
134         )
135         batching_config = (
136             batching_config.get(runner.name) if runner.name in batching_config else {}
137         )
138         runner_deployment = _get_runner_deployment(
139             svc, runner.name, runner_deployment_config, enable_batching, batching_config
140         ).bind()
141         runner_deployments[runner.name] = runner_deployment
142 
143     return runner_deployments
144 
145 
146 def deployment(
147     target: str | Tag | bentoml.Bento | bentoml.Service,
148     service_deployment_config: dict[str, t.Any] | None = None,
149     runners_deployment_config_map: dict[str, dict[str, t.Any]] | None = None,
150     enable_batching: bool = False,
151     batching_config: dict[str, dict[str, dict[str, float | int]]] | None = None,
152 ) -> Deployment:
153     """
154     Deploy a Bento or bentoml.Service to Ray
155 
156     Args:
157         target: A bentoml.Service instance, Bento tag string, or Bento object
158         service_deployment_config: Ray deployment config for BentoML API server
159         runners_deployment_config_map: Ray deployment config map for all Runners
160         enable_batching: Enable Ray Serve batching for RunnerMethods with batchable=True
161         batching_config: Pass through Ray batching config by Runner name and method name
162 
163     Returns:
164         A bound ray.serve.Deployment instance
165 
166     Example:
167 
168     .. code-block:: python
169         :caption: `ray_demo.py`
170 
171         import bentoml
172 
173         classifier = bentoml.ray.deployment('iris_classifier:latest')
174 
175     .. code-block:: bash
176 
177         serve run ray_demo:classifier
178 
179     Configure BentoML-on-Ray deployment:
180 
181     .. code-block:: python
182 
183         import bentoml
184 
185         classifier = bentoml.ray.deployment(
186             'iris_classifier:latest',
187             {"route_prefix": "/hello", "num_replicas": 3, "ray_actor_options": {"num_cpus": 1}},
188             {"iris_clf": {"num_replicas": 1, "ray_actor_options": {"num_cpus": 5}}}
189         )
190 
191     Configure BentoML Runner's Batching behavior on Ray:
192 
193     .. code-block:: python
194 
195         import bentoml
196 
197         deploy = bentoml.ray.deployment(
198             'fraud_detection:latest',
199             {"num_replicas": 5, "ray_actor_options": {"num_cpus": 1}},
200             {"ieee-fraud-detection-sm": {"num_replicas": 1, "ray_actor_options": {"num_cpus": 5}}},
201             enable_batching=True,
202             batching_config={"ieee-fraud-detection-sm": {"predict_proba": {"max_batch_size": 5, "batch_wait_timeout_s": 0.2}}}
203         )
204 
205     """
206     # TODO: validate Ray deployment options
207     service_deployment_config = service_deployment_config or {}
208 
209     svc = target if isinstance(target, bentoml.Service) else bentoml.load(target)
210 
211     runner_deployments = _deploy_bento_runners(
212         svc, runners_deployment_config_map, enable_batching, batching_config
213     )
214 
215     return _get_service_deployment(svc, **service_deployment_config).bind(
216         **runner_deployments
217     )
218
```

Path: `src/bentoml/_internal/ray/runner_handle.py`
Content:
```
1 from __future__ import annotations
2 
3 import functools
4 import typing as t
5 
6 from ...exceptions import MissingDependencyException
7 from ..runner import Runner
8 from ..runner.runner import RunnerMethod
9 from ..runner.runner_handle import RunnerHandle
10 
11 try:
12     import ray
13     from ray.serve.deployment import Deployment as RayDeployment
14 except ImportError:  # pragma: no cover
15     raise MissingDependencyException(
16         "'ray' is required in order to use module 'bentoml.ray', install with 'pip install -U \"ray[serve]\"'. For more information, refer to https://docs.ray.io/",
17     )
18 
19 if t.TYPE_CHECKING:
20     P = t.ParamSpec("P")
21 else:
22     P = t.TypeVar("P")
23 
24 R = t.TypeVar("R")
25 
26 
27 class RayRunnerHandle(RunnerHandle):
28     def __init__(self, runner: Runner, ray_deployment: RayDeployment) -> None:
29         self._runner = runner
30         self._ray_deployment = ray_deployment
31 
32     async def is_ready(self, timeout: int) -> bool:
33         return True
34 
35     def run_method(
36         self,
37         __bentoml_method: RunnerMethod[t.Any, P, R],
38         *args: P.args,
39         **kwargs: P.kwargs,
40     ) -> R | tuple[R, ...]:
41         import anyio
42 
43         return t.cast(
44             "R | tuple[R, ...]",
45             anyio.from_thread.run(
46                 functools.partial(self.async_run_method, **kwargs),
47                 __bentoml_method,
48                 *args,
49             ),
50         )
51 
52     async def async_run_method(
53         self,
54         __bentoml_method: RunnerMethod[t.Any, P, R],
55         *args: P.args,
56         **kwargs: P.kwargs,
57     ) -> R | tuple[R, ...]:
58         return ray.get(
59             await getattr(self._ray_deployment, __bentoml_method.name).remote(
60                 *args, **kwargs
61             )
62         )
63
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/bentoml/_internal/ray/__init__.py b/src/bentoml/_internal/ray/__init__.py
--- a/src/bentoml/_internal/ray/__init__.py
+++ b/src/bentoml/_internal/ray/__init__.py
@@ -5,7 +5,7 @@
 
 import requests
 
-from ._internal.utils import pkg
+from ..._internal.utils import pkg
 
 if pkg.pkg_version_info("ray")[:2] >= (2, 5):
     from ray.serve._private.http_util import BufferedASGISender as ASGIHTTPSender
diff --git a/src/bentoml/_internal/ray/runner_handle.py b/src/bentoml/_internal/ray/runner_handle.py
--- a/src/bentoml/_internal/ray/runner_handle.py
+++ b/src/bentoml/_internal/ray/runner_handle.py
@@ -60,3 +60,11 @@
                 *args, **kwargs
             )
         )
+
+    def async_stream_method(
+        self,
+        __bentoml_method: RunnerMethod[t.Any, P, R],
+        *args: P.args,
+        **kwargs: P.kwargs,
+    ) -> t.AsyncGenerator[R, None]:
+        raise NotImplementedError

{"golden_diff": "diff --git a/src/bentoml/_internal/ray/__init__.py b/src/bentoml/_internal/ray/__init__.py\n--- a/src/bentoml/_internal/ray/__init__.py\n+++ b/src/bentoml/_internal/ray/__init__.py\n@@ -5,7 +5,7 @@\n \n import requests\n \n-from ._internal.utils import pkg\n+from ..._internal.utils import pkg\n \n if pkg.pkg_version_info(\"ray\")[:2] >= (2, 5):\n     from ray.serve._private.http_util import BufferedASGISender as ASGIHTTPSender\ndiff --git a/src/bentoml/_internal/ray/runner_handle.py b/src/bentoml/_internal/ray/runner_handle.py\n--- a/src/bentoml/_internal/ray/runner_handle.py\n+++ b/src/bentoml/_internal/ray/runner_handle.py\n@@ -60,3 +60,11 @@\n                 *args, **kwargs\n             )\n         )\n+\n+    def async_stream_method(\n+        self,\n+        __bentoml_method: RunnerMethod[t.Any, P, R],\n+        *args: P.args,\n+        **kwargs: P.kwargs,\n+    ) -> t.AsyncGenerator[R, None]:\n+        raise NotImplementedError\n", "issue": "bug: unable to start BentoML container in RayServe\n### Describe the bug\n\nfollowed instructions at https://docs.bentoml.org/en/latest/integrations/ray.html\r\n\r\nray serve attempts to instantiate container, but trips up:\r\n\r\n2023-10-30 23:45:22,446 INFO scripts.py:471 -- Running import path: 'bento_ray:classifier'.\r\nTraceback (most recent call last):\r\n  File \"/home/thor/.local/bin/serve\", line 8, in <module>\r\n    sys.exit(cli())\r\n  File \"/home/thor/.local/lib/python3.8/site-packages/click/core.py\", line 1157, in __call__\r\n    return self.main(*args, **kwargs)\r\n  File \"/home/thor/.local/lib/python3.8/site-packages/click/core.py\", line 1078, in main\r\n    rv = self.invoke(ctx)\r\n  File \"/home/thor/.local/lib/python3.8/site-packages/click/core.py\", line 1688, in invoke\r\n    return _process_result(sub_ctx.command.invoke(sub_ctx))\r\n  File \"/home/thor/.local/lib/python3.8/site-packages/click/core.py\", line 1434, in invoke\r\n    return ctx.invoke(self.callback, **ctx.params)\r\n  File \"/home/thor/.local/lib/python3.8/site-packages/click/core.py\", line 783, in invoke\r\n    return __callback(*args, **kwargs)\r\n  File \"/home/thor/.local/lib/python3.8/site-packages/ray/serve/scripts.py\", line 473, in run\r\n    import_attr(import_path), args_dict\r\n  File \"/home/thor/.local/lib/python3.8/site-packages/ray/_private/utils.py\", line 1378, in import_attr\r\n    module = importlib.import_module(module_name)\r\n  File \"/usr/lib/python3.8/importlib/__init__.py\", line 127, in import_module\r\n    return _bootstrap._gcd_import(name[level:], package, level)\r\n  File \"<frozen importlib._bootstrap>\", line 1014, in _gcd_import\r\n  File \"<frozen importlib._bootstrap>\", line 991, in _find_and_load\r\n  File \"<frozen importlib._bootstrap>\", line 975, in _find_and_load_unlocked\r\n  File \"<frozen importlib._bootstrap>\", line 671, in _load_unlocked\r\n  File \"<frozen importlib._bootstrap_external>\", line 848, in exec_module\r\n  File \"<frozen importlib._bootstrap>\", line 219, in _call_with_frames_removed\r\n  File \"/mnt/d/Data/AIRay/./bento_ray.py\", line 5, in <module>\r\n    classifier = bentoml.ray.deployment(\r\n  File \"/home/thor/.local/lib/python3.8/site-packages/bentoml/_internal/utils/lazy_loader.py\", line 69, in __getattr__\r\n    self._module = self._load()\r\n  File \"/home/thor/.local/lib/python3.8/site-packages/bentoml/_internal/utils/lazy_loader.py\", line 53, in _load\r\n    raise self._exc(f\"{self._exc_msg} (reason: {err})\") from None\r\nbentoml.exceptions.MissingDependencyException: None (reason: No module named 'bentoml._internal.ray._internal')\r\n\n\n### To reproduce\n\n_No response_\n\n### Expected behavior\n\nExpected RayServe to spin up BentoML container within its cluster\n\n### Environment\n\nUbuntu 20.04.6\n", "before_files": [{"content": "from __future__ import annotations\n\nimport typing as t\nfrom functools import partial\n\nimport requests\n\nfrom ._internal.utils import pkg\n\nif pkg.pkg_version_info(\"ray\")[:2] >= (2, 5):\n    from ray.serve._private.http_util import BufferedASGISender as ASGIHTTPSender\nelse:\n    from ray.serve._private.http_util import ASGIHTTPSender\n\nimport bentoml\nfrom bentoml import Tag\n\nfrom ...exceptions import MissingDependencyException\nfrom ..runner.container import AutoContainer\nfrom ..runner.utils import Params\nfrom .runner_handle import RayRunnerHandle\n\ntry:\n    from ray import serve\n    from ray.runtime_env import RuntimeEnv\n    from ray.serve.deployment import Deployment\nexcept ImportError:  # pragma: no cover\n    raise MissingDependencyException(\n        \"\"\"'ray[serve]' is required in order to use module 'bentoml.ray', install with 'pip install -U \"ray[serve]\"'. See https://docs.ray.io/ for more information.\"\"\",\n    )\n\n\ndef _get_runner_deployment(\n    svc: bentoml.Service,\n    runner_name: str,\n    runner_deployment_config: dict[str, t.Any],\n    enable_batching: bool,\n    batching_config: dict[str, dict[str, float | int]],\n) -> Deployment:\n    class RunnerDeployment:\n        def __init__(self):\n            _runner = next(\n                (runner for runner in svc.runners if runner.name == runner_name),\n                None,\n            )\n            _runner.init_local(quiet=True)\n            for method in _runner.runner_methods:\n                if enable_batching and method.config.batchable:\n                    inp_batch_dim = method.config.batch_dim[0]\n                    out_batch_dim = method.config.batch_dim[1]\n                    ray_batch_args = (\n                        batching_config[method.name]\n                        if method.name in batching_config\n                        else {}\n                    )\n\n                    @serve.batch(**ray_batch_args)\n                    async def _func(self, *args, **kwargs):\n                        params = Params(*args, **kwargs).map(\n                            partial(\n                                AutoContainer.batches_to_batch, batch_dim=inp_batch_dim\n                            )\n                        )\n                        run_params = params.map(lambda arg: arg[0])\n                        indices = next(iter(params.items()))[1][1]\n                        results = await getattr(_runner, method.name).async_run(\n                            *run_params.args, **run_params.kwargs\n                        )\n                        return AutoContainer.batch_to_batches(\n                            results, indices=indices, batch_dim=out_batch_dim\n                        )\n\n                    setattr(RunnerDeployment, method.name, _func)\n                else:\n\n                    async def _func(self, *args, **kwargs):\n                        return await getattr(_runner, method.name).async_run(\n                            *args, **kwargs\n                        )\n\n                    setattr(RunnerDeployment, method.name, _func)\n\n    return serve.deployment(\n        RunnerDeployment, name=f\"bento-runner-{runner_name}\", **runner_deployment_config\n    )\n\n\ndef _get_service_deployment(svc: bentoml.Service, **kwargs) -> Deployment:\n    @serve.deployment(name=f\"bento-svc-{svc.name}\", **kwargs)\n    class BentoDeployment:\n        def __init__(self, **runner_deployments: dict[str, Deployment]):\n            from ..server.http_app import HTTPAppFactory\n\n            # Use Ray's metrics setup. Ray has already initialized a prometheus client\n            # for multi-process and this conflicts with BentoML's prometheus setup\n            self.app = HTTPAppFactory(svc, enable_metrics=False)()\n\n            for runner in svc.runners:\n                assert runner.name in runner_deployments\n                runner._set_handle(RayRunnerHandle, runner_deployments[runner.name])\n\n        async def __call__(self, request: requests.Request):\n            sender = ASGIHTTPSender()\n            await self.app(request.scope, receive=request.receive, send=sender)\n            return sender.build_asgi_response()\n\n    return BentoDeployment\n\n\ndef get_bento_runtime_env(bento_tag: str | Tag) -> RuntimeEnv:\n    # TODO: create Ray RuntimeEnv from Bento env configs\n    return RuntimeEnv(\n        pip=[],\n        conda=[],\n        env_vars=[],\n    )\n\n\ndef _deploy_bento_runners(\n    svc: bentoml.Service,\n    runners_deployment_config_map: dict | None = None,\n    enable_batching: bool = False,\n    batching_config: dict | None = None,\n) -> dict[str, Deployment]:\n    # Deploy BentoML Runners as Ray serve deployments\n    runners_deployment_config_map = runners_deployment_config_map or {}\n    batching_config = batching_config or {}\n    runner_deployments = {}\n    for runner in svc.runners:\n        runner_deployment_config = (\n            runners_deployment_config_map.get(runner.name)\n            if runner.name in runners_deployment_config_map\n            else {}\n        )\n        batching_config = (\n            batching_config.get(runner.name) if runner.name in batching_config else {}\n        )\n        runner_deployment = _get_runner_deployment(\n            svc, runner.name, runner_deployment_config, enable_batching, batching_config\n        ).bind()\n        runner_deployments[runner.name] = runner_deployment\n\n    return runner_deployments\n\n\ndef deployment(\n    target: str | Tag | bentoml.Bento | bentoml.Service,\n    service_deployment_config: dict[str, t.Any] | None = None,\n    runners_deployment_config_map: dict[str, dict[str, t.Any]] | None = None,\n    enable_batching: bool = False,\n    batching_config: dict[str, dict[str, dict[str, float | int]]] | None = None,\n) -> Deployment:\n    \"\"\"\n    Deploy a Bento or bentoml.Service to Ray\n\n    Args:\n        target: A bentoml.Service instance, Bento tag string, or Bento object\n        service_deployment_config: Ray deployment config for BentoML API server\n        runners_deployment_config_map: Ray deployment config map for all Runners\n        enable_batching: Enable Ray Serve batching for RunnerMethods with batchable=True\n        batching_config: Pass through Ray batching config by Runner name and method name\n\n    Returns:\n        A bound ray.serve.Deployment instance\n\n    Example:\n\n    .. code-block:: python\n        :caption: `ray_demo.py`\n\n        import bentoml\n\n        classifier = bentoml.ray.deployment('iris_classifier:latest')\n\n    .. code-block:: bash\n\n        serve run ray_demo:classifier\n\n    Configure BentoML-on-Ray deployment:\n\n    .. code-block:: python\n\n        import bentoml\n\n        classifier = bentoml.ray.deployment(\n            'iris_classifier:latest',\n            {\"route_prefix\": \"/hello\", \"num_replicas\": 3, \"ray_actor_options\": {\"num_cpus\": 1}},\n            {\"iris_clf\": {\"num_replicas\": 1, \"ray_actor_options\": {\"num_cpus\": 5}}}\n        )\n\n    Configure BentoML Runner's Batching behavior on Ray:\n\n    .. code-block:: python\n\n        import bentoml\n\n        deploy = bentoml.ray.deployment(\n            'fraud_detection:latest',\n            {\"num_replicas\": 5, \"ray_actor_options\": {\"num_cpus\": 1}},\n            {\"ieee-fraud-detection-sm\": {\"num_replicas\": 1, \"ray_actor_options\": {\"num_cpus\": 5}}},\n            enable_batching=True,\n            batching_config={\"ieee-fraud-detection-sm\": {\"predict_proba\": {\"max_batch_size\": 5, \"batch_wait_timeout_s\": 0.2}}}\n        )\n\n    \"\"\"\n    # TODO: validate Ray deployment options\n    service_deployment_config = service_deployment_config or {}\n\n    svc = target if isinstance(target, bentoml.Service) else bentoml.load(target)\n\n    runner_deployments = _deploy_bento_runners(\n        svc, runners_deployment_config_map, enable_batching, batching_config\n    )\n\n    return _get_service_deployment(svc, **service_deployment_config).bind(\n        **runner_deployments\n    )\n", "path": "src/bentoml/_internal/ray/__init__.py"}, {"content": "from __future__ import annotations\n\nimport functools\nimport typing as t\n\nfrom ...exceptions import MissingDependencyException\nfrom ..runner import Runner\nfrom ..runner.runner import RunnerMethod\nfrom ..runner.runner_handle import RunnerHandle\n\ntry:\n    import ray\n    from ray.serve.deployment import Deployment as RayDeployment\nexcept ImportError:  # pragma: no cover\n    raise MissingDependencyException(\n        \"'ray' is required in order to use module 'bentoml.ray', install with 'pip install -U \\\"ray[serve]\\\"'. For more information, refer to https://docs.ray.io/\",\n    )\n\nif t.TYPE_CHECKING:\n    P = t.ParamSpec(\"P\")\nelse:\n    P = t.TypeVar(\"P\")\n\nR = t.TypeVar(\"R\")\n\n\nclass RayRunnerHandle(RunnerHandle):\n    def __init__(self, runner: Runner, ray_deployment: RayDeployment) -> None:\n        self._runner = runner\n        self._ray_deployment = ray_deployment\n\n    async def is_ready(self, timeout: int) -> bool:\n        return True\n\n    def run_method(\n        self,\n        __bentoml_method: RunnerMethod[t.Any, P, R],\n        *args: P.args,\n        **kwargs: P.kwargs,\n    ) -> R | tuple[R, ...]:\n        import anyio\n\n        return t.cast(\n            \"R | tuple[R, ...]\",\n            anyio.from_thread.run(\n                functools.partial(self.async_run_method, **kwargs),\n                __bentoml_method,\n                *args,\n            ),\n        )\n\n    async def async_run_method(\n        self,\n        __bentoml_method: RunnerMethod[t.Any, P, R],\n        *args: P.args,\n        **kwargs: P.kwargs,\n    ) -> R | tuple[R, ...]:\n        return ray.get(\n            await getattr(self._ray_deployment, __bentoml_method.name).remote(\n                *args, **kwargs\n            )\n        )\n", "path": "src/bentoml/_internal/ray/runner_handle.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport typing as t\nfrom functools import partial\n\nimport requests\n\nfrom ..._internal.utils import pkg\n\nif pkg.pkg_version_info(\"ray\")[:2] >= (2, 5):\n    from ray.serve._private.http_util import BufferedASGISender as ASGIHTTPSender\nelse:\n    from ray.serve._private.http_util import ASGIHTTPSender\n\nimport bentoml\nfrom bentoml import Tag\n\nfrom ...exceptions import MissingDependencyException\nfrom ..runner.container import AutoContainer\nfrom ..runner.utils import Params\nfrom .runner_handle import RayRunnerHandle\n\ntry:\n    from ray import serve\n    from ray.runtime_env import RuntimeEnv\n    from ray.serve.deployment import Deployment\nexcept ImportError:  # pragma: no cover\n    raise MissingDependencyException(\n        \"\"\"'ray[serve]' is required in order to use module 'bentoml.ray', install with 'pip install -U \"ray[serve]\"'. See https://docs.ray.io/ for more information.\"\"\",\n    )\n\n\ndef _get_runner_deployment(\n    svc: bentoml.Service,\n    runner_name: str,\n    runner_deployment_config: dict[str, t.Any],\n    enable_batching: bool,\n    batching_config: dict[str, dict[str, float | int]],\n) -> Deployment:\n    class RunnerDeployment:\n        def __init__(self):\n            _runner = next(\n                (runner for runner in svc.runners if runner.name == runner_name),\n                None,\n            )\n            _runner.init_local(quiet=True)\n            for method in _runner.runner_methods:\n                if enable_batching and method.config.batchable:\n                    inp_batch_dim = method.config.batch_dim[0]\n                    out_batch_dim = method.config.batch_dim[1]\n                    ray_batch_args = (\n                        batching_config[method.name]\n                        if method.name in batching_config\n                        else {}\n                    )\n\n                    @serve.batch(**ray_batch_args)\n                    async def _func(self, *args, **kwargs):\n                        params = Params(*args, **kwargs).map(\n                            partial(\n                                AutoContainer.batches_to_batch, batch_dim=inp_batch_dim\n                            )\n                        )\n                        run_params = params.map(lambda arg: arg[0])\n                        indices = next(iter(params.items()))[1][1]\n                        results = await getattr(_runner, method.name).async_run(\n                            *run_params.args, **run_params.kwargs\n                        )\n                        return AutoContainer.batch_to_batches(\n                            results, indices=indices, batch_dim=out_batch_dim\n                        )\n\n                    setattr(RunnerDeployment, method.name, _func)\n                else:\n\n                    async def _func(self, *args, **kwargs):\n                        return await getattr(_runner, method.name).async_run(\n                            *args, **kwargs\n                        )\n\n                    setattr(RunnerDeployment, method.name, _func)\n\n    return serve.deployment(\n        RunnerDeployment, name=f\"bento-runner-{runner_name}\", **runner_deployment_config\n    )\n\n\ndef _get_service_deployment(svc: bentoml.Service, **kwargs) -> Deployment:\n    @serve.deployment(name=f\"bento-svc-{svc.name}\", **kwargs)\n    class BentoDeployment:\n        def __init__(self, **runner_deployments: dict[str, Deployment]):\n            from ..server.http_app import HTTPAppFactory\n\n            # Use Ray's metrics setup. Ray has already initialized a prometheus client\n            # for multi-process and this conflicts with BentoML's prometheus setup\n            self.app = HTTPAppFactory(svc, enable_metrics=False)()\n\n            for runner in svc.runners:\n                assert runner.name in runner_deployments\n                runner._set_handle(RayRunnerHandle, runner_deployments[runner.name])\n\n        async def __call__(self, request: requests.Request):\n            sender = ASGIHTTPSender()\n            await self.app(request.scope, receive=request.receive, send=sender)\n            return sender.build_asgi_response()\n\n    return BentoDeployment\n\n\ndef get_bento_runtime_env(bento_tag: str | Tag) -> RuntimeEnv:\n    # TODO: create Ray RuntimeEnv from Bento env configs\n    return RuntimeEnv(\n        pip=[],\n        conda=[],\n        env_vars=[],\n    )\n\n\ndef _deploy_bento_runners(\n    svc: bentoml.Service,\n    runners_deployment_config_map: dict | None = None,\n    enable_batching: bool = False,\n    batching_config: dict | None = None,\n) -> dict[str, Deployment]:\n    # Deploy BentoML Runners as Ray serve deployments\n    runners_deployment_config_map = runners_deployment_config_map or {}\n    batching_config = batching_config or {}\n    runner_deployments = {}\n    for runner in svc.runners:\n        runner_deployment_config = (\n            runners_deployment_config_map.get(runner.name)\n            if runner.name in runners_deployment_config_map\n            else {}\n        )\n        batching_config = (\n            batching_config.get(runner.name) if runner.name in batching_config else {}\n        )\n        runner_deployment = _get_runner_deployment(\n            svc, runner.name, runner_deployment_config, enable_batching, batching_config\n        ).bind()\n        runner_deployments[runner.name] = runner_deployment\n\n    return runner_deployments\n\n\ndef deployment(\n    target: str | Tag | bentoml.Bento | bentoml.Service,\n    service_deployment_config: dict[str, t.Any] | None = None,\n    runners_deployment_config_map: dict[str, dict[str, t.Any]] | None = None,\n    enable_batching: bool = False,\n    batching_config: dict[str, dict[str, dict[str, float | int]]] | None = None,\n) -> Deployment:\n    \"\"\"\n    Deploy a Bento or bentoml.Service to Ray\n\n    Args:\n        target: A bentoml.Service instance, Bento tag string, or Bento object\n        service_deployment_config: Ray deployment config for BentoML API server\n        runners_deployment_config_map: Ray deployment config map for all Runners\n        enable_batching: Enable Ray Serve batching for RunnerMethods with batchable=True\n        batching_config: Pass through Ray batching config by Runner name and method name\n\n    Returns:\n        A bound ray.serve.Deployment instance\n\n    Example:\n\n    .. code-block:: python\n        :caption: `ray_demo.py`\n\n        import bentoml\n\n        classifier = bentoml.ray.deployment('iris_classifier:latest')\n\n    .. code-block:: bash\n\n        serve run ray_demo:classifier\n\n    Configure BentoML-on-Ray deployment:\n\n    .. code-block:: python\n\n        import bentoml\n\n        classifier = bentoml.ray.deployment(\n            'iris_classifier:latest',\n            {\"route_prefix\": \"/hello\", \"num_replicas\": 3, \"ray_actor_options\": {\"num_cpus\": 1}},\n            {\"iris_clf\": {\"num_replicas\": 1, \"ray_actor_options\": {\"num_cpus\": 5}}}\n        )\n\n    Configure BentoML Runner's Batching behavior on Ray:\n\n    .. code-block:: python\n\n        import bentoml\n\n        deploy = bentoml.ray.deployment(\n            'fraud_detection:latest',\n            {\"num_replicas\": 5, \"ray_actor_options\": {\"num_cpus\": 1}},\n            {\"ieee-fraud-detection-sm\": {\"num_replicas\": 1, \"ray_actor_options\": {\"num_cpus\": 5}}},\n            enable_batching=True,\n            batching_config={\"ieee-fraud-detection-sm\": {\"predict_proba\": {\"max_batch_size\": 5, \"batch_wait_timeout_s\": 0.2}}}\n        )\n\n    \"\"\"\n    # TODO: validate Ray deployment options\n    service_deployment_config = service_deployment_config or {}\n\n    svc = target if isinstance(target, bentoml.Service) else bentoml.load(target)\n\n    runner_deployments = _deploy_bento_runners(\n        svc, runners_deployment_config_map, enable_batching, batching_config\n    )\n\n    return _get_service_deployment(svc, **service_deployment_config).bind(\n        **runner_deployments\n    )\n", "path": "src/bentoml/_internal/ray/__init__.py"}, {"content": "from __future__ import annotations\n\nimport functools\nimport typing as t\n\nfrom ...exceptions import MissingDependencyException\nfrom ..runner import Runner\nfrom ..runner.runner import RunnerMethod\nfrom ..runner.runner_handle import RunnerHandle\n\ntry:\n    import ray\n    from ray.serve.deployment import Deployment as RayDeployment\nexcept ImportError:  # pragma: no cover\n    raise MissingDependencyException(\n        \"'ray' is required in order to use module 'bentoml.ray', install with 'pip install -U \\\"ray[serve]\\\"'. For more information, refer to https://docs.ray.io/\",\n    )\n\nif t.TYPE_CHECKING:\n    P = t.ParamSpec(\"P\")\nelse:\n    P = t.TypeVar(\"P\")\n\nR = t.TypeVar(\"R\")\n\n\nclass RayRunnerHandle(RunnerHandle):\n    def __init__(self, runner: Runner, ray_deployment: RayDeployment) -> None:\n        self._runner = runner\n        self._ray_deployment = ray_deployment\n\n    async def is_ready(self, timeout: int) -> bool:\n        return True\n\n    def run_method(\n        self,\n        __bentoml_method: RunnerMethod[t.Any, P, R],\n        *args: P.args,\n        **kwargs: P.kwargs,\n    ) -> R | tuple[R, ...]:\n        import anyio\n\n        return t.cast(\n            \"R | tuple[R, ...]\",\n            anyio.from_thread.run(\n                functools.partial(self.async_run_method, **kwargs),\n                __bentoml_method,\n                *args,\n            ),\n        )\n\n    async def async_run_method(\n        self,\n        __bentoml_method: RunnerMethod[t.Any, P, R],\n        *args: P.args,\n        **kwargs: P.kwargs,\n    ) -> R | tuple[R, ...]:\n        return ray.get(\n            await getattr(self._ray_deployment, __bentoml_method.name).remote(\n                *args, **kwargs\n            )\n        )\n\n    def async_stream_method(\n        self,\n        __bentoml_method: RunnerMethod[t.Any, P, R],\n        *args: P.args,\n        **kwargs: P.kwargs,\n    ) -> t.AsyncGenerator[R, None]:\n        raise NotImplementedError\n", "path": "src/bentoml/_internal/ray/runner_handle.py"}]}