https://bentoml.com logo
Join Slack
Powered by
Hello team :wave: , I have question regarding the ...
# ask-for-help
t
Hello team 👋 , I have question regarding the Adaptive Batching feature, based on the official description, I understand it this way, for example, I have some requests coming with only 1 element array, it would help combine some arrays together to process, like make a 2 or 3 elements array and then pass to process, or it has some requests with say 5 elements array, and it may distribute with 2 and 3 elements array requests to process separately based on real-time traffic burden and also the settings (max batch size and max latency)? Is it how it dynamically works? Or there is some misunderstanding around the Adaptive batch?
Copy code
BentoML supports adaptive batching, a dynamic request dispatching mechanism that intelligently groups multiple requests for more efficient processing. It continuously adjusts batch size and window based on real-time traffic patterns.
The reason I ask is because I have set up some load test, and logged out the real batch size received from the inference function, but I don't see any dynamic size coming in. So I am wondering any setup was not correct or I misunderstand this feature. Can anyone help me with this? 🙏
j
Can you share how did you setup your adaptive batching code?
t
in the service.py
Copy code
@bentoml.service(workers=int(number_of_workers))
class RealModel:
    def __init__(self):
        self.custom_inference = load_custom_inference_module()
        self.model_dir = os.environ['ZD_INFERENCE_MODEL_DIR']
        self.loaded_model = self.custom_inference.model_fn(self.model_dir)

        if hasattr(self.custom_inference, 'warm_up'):
            self.custom_inference.warm_up(self.loaded_model)
        else:
            bentoml_logger.warning("The custom_inference module does not have a warm_up function.")
            
    @bentoml.api
    def model_loaded(self) -> bool:
        return self.loaded_model is not None

    @bentoml.api
    def input_array_fn(self, input_str: str, content_type: str) -> List[Any]:
        result = self.custom_inference.input_array_fn(input_str, content_type)
        return result
    
    @bentoml.api(batchable=True, max_batch_size=max_batch_size, max_latency_ms=max_latency_ms)
    def predict_array_fn(self, input_tensors: List[Any]) -> List[Any]:
        bentoml_logger.info(f"Batch size: {len(input_tensors)}")
        return self.custom_inference.predict_array_fn(input_tensors, self.loaded_model)
    
    @bentoml.api
    def output_array_fn(self, predictions: List[Any], content_type: str) -> Any:
        return self.custom_inference.output_array_fn(predictions, content_type)

    // other functions

@bentoml.service(
    workers=int(number_of_workers),
    traffic={"timeout": 60},
    logging={
        "access": {
            "enabled": True,
            "request_content_length": True,
            "request_content_type": True,
            "response_content_length": True,
            "response_content_type": True,
            "skip_paths": ["/metrics", "/healthz", "/livez", "/readyz"],
            "format": {
                "trace_id": "032x",
                "span_id": "016x"
            }
        }
    }
)
@bentoml.mount_asgi_app(app)
class ServeModel:
    name = "Zendesk Model Serving"
    model = bentoml.depends(RealModel)

    @bentoml.api(route='predictions/model')
    def invoke(
        self,
        inputs: Dict,
        ctx: bentoml.Context
    ) -> Dict:
        content_type = ctx.request.headers.get("content-type")
        input_str = json.dumps(inputs)
        if self.adaptive_batch_enabled:
            input_tensors = self.model.input_array_fn(input_str, content_type)
            predictions = self.model.predict_array_fn(input_tensors)
            predict_response = self.model.output_array_fn(predictions, 'application/json')
        else:
            input_tensors = self.model.input_fn(input_str, content_type)
            predictions = self.model.predict_fn(input_tensors)
            predict_response = self.model.output_fn(predictions, 'application/json')
        
        if isinstance(predict_response, str):
            dict_response = json.loads(predict_response)
        else:
            dict_response = predict_response

        if not isinstance(dict_response, Dict):
            raise ValueError("Output from custom inference is not a valid dictionary")



        return dict_response
and in our custom_inference file
Copy code
def predict_array_fn(
        input_data: Union[Dict, List[Dict]], model: EmbeddingModelE5MultilingualLargeOnnx
) -> List[List[float]]:
    """Performs embeddings on the input text, using the loaded model."""
    input_datas = [input_data] if not isinstance(input_data, list) else input_data
    print(f"Real Batch size: {len(input_datas)}")
    embeddings = model.get_embeddings(input_datas)
    return embeddings


def input_array_fn(request_body: str, request_content_type: str) -> List[Dict]:
    """Loads text content from request."""
    if request_content_type.startswith("application/json"):
        input_data = json.loads(request_body)
        if "prefix_type" in input_data:
            input_prefix_type = input_data["prefix_type"] if not isinstance(input_data["prefix_type"], list) \
                else input_data["prefix_type"][0]
            try:
                prefix_type = EmbeddingPrefixEnum[input_prefix_type]
            except KeyError:
                raise ValueError(f"'{input_prefix_type}' is not a valid name in {EmbeddingPrefixEnum.__name__}")
        else:
            prefix_type = EmbeddingPrefixEnum.passage

        return [{"text": text, "prefix_type": prefix_type.value} for text in input_data["texts"]]
    else:
        raise ValueError(
            f"request_content_type {request_content_type} not supported. Must be 'application/json'"
        )


def output_array_fn(predictions: List[List[float]], content_type: str) -> Dict[str, List[List]]:
    """Serializes the prediction result into a JSON formatted list of embeddings."""
    if content_type != "application/json":
        raise ValueError(
            f"content_type {content_type} not supported. Must be 'application/json'"
        )
    return {"embeddings": predictions}
thanks @Jian Shen Yap, I pasted the main things here, please let me know if you need more details 🙏
j
Hello,your understand of adaptive batching is correct. it will dynamically batch request together to optimize throughput while respective the
Copy code
max_batch_size=max_batch_size
max_latency_ms=max_latency_ms
parameter. can you share what is the value of these parameter, as well as the average latency for 1 prediction? also, are you firing alot of request at once and expect it to batch, but it doesn't?
another thing that you cna check, there is a 
bentoml_service_adaptive_batch_size_*
related metrics when you check at 
/metrics
endpoint, it wil give you the insights about the adaptive batching
t
Hey Jian, what I set is
Copy code
max_batch_size=5
max_latency_ms=200000
average latency for 1 prediction would be less than 100ms I am firing a lot of requests at once and keep adding more requests slowly and expect it to batch, and so far checking the batch size send to the group function is still same as what I have in the original requests. I haven't check the metrics so far, I will check it later. thanks!
Hey @Jian Shen Yap, I used metrics api, and get adaptive batching related metrics like below, just wondering how can I tell if they have dynamically combine some requests together to process?
Copy code
# HELP bentoml_service_adaptive_batch_size Service adaptive batch size
# TYPE bentoml_service_adaptive_batch_size histogram
bentoml_service_adaptive_batch_size_sum{method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="2"} 25015.0
bentoml_service_adaptive_batch_size_sum{method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="1"} 25506.0
bentoml_service_adaptive_batch_size_bucket{le="1.0",method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="2"} 25015.0
bentoml_service_adaptive_batch_size_bucket{le="2.0",method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="2"} 25015.0
bentoml_service_adaptive_batch_size_bucket{le="4.0",method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="2"} 25015.0
bentoml_service_adaptive_batch_size_bucket{le="5.0",method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="2"} 25015.0
bentoml_service_adaptive_batch_size_bucket{le="+Inf",method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="2"} 25015.0
bentoml_service_adaptive_batch_size_count{method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="2"} 25015.0
bentoml_service_adaptive_batch_size_bucket{le="1.0",method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="1"} 25506.0
bentoml_service_adaptive_batch_size_bucket{le="2.0",method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="1"} 25506.0
bentoml_service_adaptive_batch_size_bucket{le="4.0",method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="1"} 25506.0
bentoml_service_adaptive_batch_size_bucket{le="5.0",method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="1"} 25506.0
bentoml_service_adaptive_batch_size_bucket{le="+Inf",method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="1"} 25506.0
bentoml_service_adaptive_batch_size_count{method_name="predict_array_fn",runner_name="RealModel",service_name="RealModel",service_version="not available",worker_index="1"} 25506.0
j
the histogram shows that it is not batching, or rather, it is always a batch size of 1. you can look up how to read Prometheus histrogram metrics. can you share a payload of your requests? to see the batching mechanism in work, maybe you can try with a very simple endpoint and see
Open in Slack
PreviousNext
Powered by