code: def build_app(
      intermediate_handles: bool,
      num_replicas: int,
      max_batch_size: int,
      max_concurrent_queries: int,
  ):
      @serve.deployment(max_concurrent_queries=1000)
      class Upstream:
          def __init__(self, handle: RayServeHandle):
              self._handle = handle
  
              # Turn off access log.
              logging.getLogger("ray.serve").setLevel(logging.WARNING)
  
          async def __call__(self, req: Request):
              return await self._handle.remote(await req.body())
  
      @serve.deployment(
          num_replicas=num_replicas,
          max_concurrent_queries=max_concurrent_queries,
      )
      class Downstream:
          def __init__(self):
              # Turn off access log.
              logging.getLogger("ray.serve").setLevel(logging.WARNING)
  
          @serve.batch(max_batch_size=max_batch_size)
          async def batch(self, reqs):
              return [b"ok"] * len(reqs)
  
          async def __call__(self, req: Union[bytes, Request]):
              if max_batch_size > 1:
                  return await self.batch(req)
              else:
                  return b"ok"
  
      if intermediate_handles:
          return Upstream.bind(Downstream.bind())
      else:
          return Downstream.bind()
  
file_path: ray/python/ray/serve/_private/benchmarks/microbenchmark.py
imports: import asyncio
  import logging
  from pprint import pprint
  from typing import Dict, Union
  import aiohttp
  from starlette.requests import Request
  import ray
  from ray import serve
  from ray.serve._private.benchmarks.common import run_throughput_benchmark
  from ray.serve.handle import RayServeHandle
label_desc: Use image node_id 1 for calling a global function with example usage: build_app(intermediate_handles, num_replicas, max_batch_size, max_concurrent_queries) and returns: Upstream, Downstream, unknown, unknown, unknown, bytes