code: def _get_runner_deployment(
      svc: bentoml.Service,
      runner_name: str,
      runner_deployment_config: dict[str, t.Any],
      enable_batching: bool,
      batching_config: dict[str, dict[str, float | int]],
  ) -> Deployment:
      class RunnerDeployment:
          def __init__(self):
              _runner = next(
                  (
                      runner
                      for runner in svc.runners
                      if runner.name == runner_name
                  ),
                  None,
              )
              _runner.init_local(quiet=True)
              for method in _runner.runner_methods:
                  if enable_batching and method.config.batchable:
                      inp_batch_dim = method.config.batch_dim[0]
                      out_batch_dim = method.config.batch_dim[1]
                      ray_batch_args = (
                          batching_config[method.name]
                          if method.name in batching_config
                          else {}
                      )
  
                      @serve.batch(**ray_batch_args)
                      async def _func(self, *args, **kwargs):
                          params = Params(*args, **kwargs).map(
                              partial(
                                  AutoContainer.batches_to_batch,
                                  batch_dim=inp_batch_dim,
                              )
                          )
                          run_params = params.map(lambda arg: arg[0])
                          indices = next(iter(params.items()))[1][1]
                          results = await getattr(
                              _runner, method.name
                          ).async_run(
                              *run_params.args, **run_params.kwargs
                          )
                          return AutoContainer.batch_to_batches(
                              results,
                              indices=indices,
                              batch_dim=out_batch_dim,
                          )
  
                      setattr(RunnerDeployment, method.name, _func)
                  else:
  
                      async def _func(self, *args, **kwargs):
                          return await getattr(
                              _runner, method.name
                          ).async_run(*args, **kwargs)
  
                      setattr(RunnerDeployment, method.name, _func)
  
      return serve.deployment(
          RunnerDeployment,
          name=f"bento-runner-{runner_name}",
          **runner_deployment_config,
      )
  
file_path: BentoML/grpc-client/bentoml/_internal/ray/__init__.py
imports: from __future__ import annotations
  import typing
  from functools import partial
  import requests
  from ..._internal.utils import pkg
  import bentoml
  from bentoml import Tag
  from ...exceptions import MissingDependencyException
  from ..runner.container import AutoContainer
  from ..runner.utils import Params
  from .runner_handle import RayRunnerHandle
label_desc: Use image node_id 1 for calling a global function with example usage: _get_runner_deployment(svc, runner_name, runner_deployment_config, enable_batching, batching_config) and returns: serve, AutoContainer, unknown