We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Issue in aggregating batched requests parameters
When sending batched requests with custom parameters (using extra field in [here](https://github.com/SeldonIO/MLServer/blob/749cd02bc5f3ee0b915151eedeff50075ac946ad/mlserver/types/dataplane.py#L27)) like the following:

```python
# request 1
payload = types.InferenceRequest(
    inputs=[
        types.RequestInput(
            name="parameters-np",
            shape=[1],
            datatype="BYTES",
            data=[],
            parameters=types.Parameters(
                custom-param='value-1,
        )
    ]
)
# request 2
payload = types.InferenceRequest(
    inputs=[
        types.RequestInput(
            name="parameters-np",
            shape=[1],
            datatype="BYTES",
            data=[],
            parameters=types.Parameters(
                custom-param='value-2',
        )
    ]
)
```
the results param field for the batched inference request will only contain a single `custom-param` field with the last request value for that, this is because a [dictionary is used](https://github.com/SeldonIO/MLServer/blob/749cd02bc5f3ee0b915151eedeff50075ac946ad/mlserver/batching/requests.py#L112) to aggregate parameters and for the case of having a same parameter with different values in the same batch only one value will be stored for them. E.g. in the case above we will only have `custom-param` equals to `value-2`. A better solution would be able to store all the values of common request parameter separately in a list e.g. `{custom-param=['value-1', 'value-2']`


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mlserver/batching/requests.py`
Content:
```
1 from collections import defaultdict, OrderedDict
2 from typing import Dict, List, Optional, Union
3 
4 from ..types import (
5     InferenceRequest,
6     InferenceResponse,
7     Parameters,
8     RequestInput,
9     RequestOutput,
10     ResponseOutput,
11 )
12 from .shape import Shape
13 
14 
15 def _get_data(payload: Union[RequestInput, ResponseOutput]):
16     return getattr(payload.data, "__root__", payload.data)
17 
18 
19 def _merge_parameters(
20     all_params: dict,
21     parametrised_obj: Union[
22         InferenceRequest, InferenceResponse, RequestInput, RequestOutput
23     ],
24 ) -> dict:
25     if not parametrised_obj.parameters:
26         return all_params
27 
28     obj_params = parametrised_obj.parameters.dict()
29     return {**all_params, **obj_params}
30 
31 
32 def _merge_data(
33     all_data: Union[list, List[str], List[bytes]]
34 ) -> Union[list, str, bytes]:
35     sampled_datum = all_data[0]
36 
37     if isinstance(sampled_datum, str):
38         return "".join(all_data)  # type: ignore
39 
40     if isinstance(sampled_datum, bytes):
41         return b"".join(all_data)  # type: ignore
42 
43     if isinstance(sampled_datum, list):
44         return sum(all_data, [])
45 
46     # TODO: Should we raise an error if we couldn't merge the data?
47     return all_data
48 
49 
50 class BatchedRequests:
51     def __init__(self, inference_requests: Dict[str, InferenceRequest] = {}):
52         self.inference_requests = inference_requests
53 
54         # External IDs represent the incoming prediction IDs that need to match
55         # 1:1 between request and response.
56         # Since we can't ensure the uniqueness (or even presence) of the
57         # external IDs, we'll also maintain our own list of internal IDs.
58         self._ids_mapping: Dict[str, Optional[str]] = OrderedDict()
59 
60         # Minibatch here refers to the individual batch size of the input head
61         # of each input request (i.e. the number of datapoints on each input
62         # request)
63         self._minibatch_sizes: Dict[str, int] = OrderedDict()
64 
65         self.merged_request = self._merge_requests()
66 
67     def _merge_requests(self) -> InferenceRequest:
68         inputs_index: Dict[str, Dict[str, RequestInput]] = defaultdict(OrderedDict)
69         outputs_index: Dict[str, Dict[str, RequestOutput]] = defaultdict(OrderedDict)
70         all_params: dict = {}
71         has_outputs = False  # if no outputs are defined, then outputs=None
72 
73         for internal_id, inference_request in self.inference_requests.items():
74             self._ids_mapping[internal_id] = inference_request.id
75             all_params = _merge_parameters(all_params, inference_request)
76             for request_input in inference_request.inputs:
77                 inputs_index[request_input.name][internal_id] = request_input
78 
79             if inference_request.outputs is not None:
80                 has_outputs = True
81                 for request_output in inference_request.outputs:
82                     outputs_index[request_output.name][internal_id] = request_output
83 
84         inputs = [
85             self._merge_request_inputs(request_inputs)
86             for request_inputs in inputs_index.values()
87         ]
88 
89         outputs = (
90             [
91                 self._merge_request_outputs(request_outputs)
92                 for request_outputs in outputs_index.values()
93             ]
94             if has_outputs
95             else None
96         )
97 
98         # TODO: Should we add a 'fake' request ID?
99         params = Parameters(**all_params) if all_params else None
100         return InferenceRequest(inputs=inputs, outputs=outputs, parameters=params)
101 
102     def _merge_request_inputs(
103         self, request_inputs: Dict[str, RequestInput]
104     ) -> RequestInput:
105         # Note that minibatch sizes could be different on each input head,
106         # however, to simplify the implementation, here we assume that it will
107         # be the same across all of them
108         batch_size = 0
109         all_data = []
110         all_params: dict = {}
111         for internal_id, request_input in request_inputs.items():
112             all_params = _merge_parameters(all_params, request_input)
113             all_data.append(_get_data(request_input))
114             minibatch_shape = Shape(request_input.shape)
115             self._minibatch_sizes[internal_id] = minibatch_shape.batch_size
116             batch_size += minibatch_shape.batch_size
117 
118         data = _merge_data(all_data)
119         parameters = Parameters(**all_params) if all_params else None
120 
121         # TODO: What should we do if list is empty?
122         sampled = next(iter(request_inputs.values()))
123         shape = Shape(sampled.shape)
124         shape.batch_size = batch_size
125 
126         return RequestInput(
127             name=sampled.name,
128             datatype=sampled.datatype,
129             shape=shape.to_list(),
130             data=data,
131             parameters=parameters,
132         )
133 
134     def _merge_request_outputs(
135         self, request_outputs: Dict[str, RequestOutput]
136     ) -> RequestOutput:
137         all_params: dict = {}
138         for internal_id, request_output in request_outputs.items():
139             all_params = _merge_parameters(all_params, request_output)
140 
141         parameters = Parameters(**all_params) if all_params else None
142 
143         # TODO: What should we do if list is empty?
144         sampled = next(iter(request_outputs.values()))
145 
146         return RequestOutput(name=sampled.name, parameters=parameters)
147 
148     def split_response(
149         self, batched_response: InferenceResponse
150     ) -> Dict[str, InferenceResponse]:
151         responses: Dict[str, InferenceResponse] = {}
152 
153         for response_output in batched_response.outputs:
154             response_outputs = self._split_response_output(response_output)
155 
156             for internal_id, response_output in response_outputs.items():
157                 if internal_id not in responses:
158                     responses[internal_id] = InferenceResponse(
159                         id=self._ids_mapping[internal_id],
160                         model_name=batched_response.model_name,
161                         model_version=batched_response.model_version,
162                         outputs=[],
163                         parameters=batched_response.parameters,
164                     )
165 
166                 responses[internal_id].outputs.append(response_output)
167 
168         return responses
169 
170     def _split_response_output(
171         self, response_output: ResponseOutput
172     ) -> Dict[str, ResponseOutput]:
173 
174         all_data = self._split_data(response_output)
175         response_outputs = {}
176         for internal_id, data in all_data.items():
177             shape = Shape(response_output.shape)
178             shape.batch_size = self._minibatch_sizes[internal_id]
179             response_outputs[internal_id] = ResponseOutput(
180                 name=response_output.name,
181                 shape=shape.to_list(),
182                 data=data,
183                 datatype=response_output.datatype,
184                 parameters=response_output.parameters,
185             )
186 
187         return response_outputs
188 
189     def _split_data(self, response_output: ResponseOutput) -> Dict[str, ResponseOutput]:
190         merged_shape = Shape(response_output.shape)
191         element_size = merged_shape.elem_size
192         merged_data = _get_data(response_output)
193         idx = 0
194 
195         all_data = {}
196         # TODO: Don't rely on array to have been flattened
197         for internal_id, minibatch_size in self._minibatch_sizes.items():
198             data = merged_data[idx : idx + minibatch_size * element_size]
199             idx += minibatch_size * element_size
200             all_data[internal_id] = data
201 
202         return all_data
203
```
--- 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/mlserver/batching/requests.py b/mlserver/batching/requests.py
--- a/mlserver/batching/requests.py
+++ b/mlserver/batching/requests.py
@@ -29,6 +29,40 @@
     return {**all_params, **obj_params}
 
 
+def _merge_input_parameters(
+    all_params: dict,
+    parametrised_obj: Union[
+        InferenceRequest, InferenceResponse, RequestInput, RequestOutput
+    ],
+) -> dict:
+    if not parametrised_obj.parameters:
+        return all_params
+    obj_params = parametrised_obj.parameters.dict()
+    if all_params == {}:
+        return obj_params
+    else:
+        common_keys = set(all_params).intersection(set(obj_params)) - {
+            "content_type",
+            "headers",
+        }
+        uncommon_keys = set(all_params).union(set(obj_params)) - common_keys
+        new_all_params = {}
+        for key in common_keys:
+            if type(all_params[key]) == list:
+                new_value = all_params[key] + [obj_params[key]]
+                new_all_params[key] = new_value
+            else:
+                new_value = [all_params[key]]
+                new_value.append(obj_params[key])
+                new_all_params[key] = new_value
+        for key in uncommon_keys:
+            if key in all_params.keys():
+                new_all_params[key] = all_params[key]
+            if key in obj_params.keys():
+                new_all_params[key] = obj_params[key]
+    return new_all_params
+
+
 def _merge_data(
     all_data: Union[list, List[str], List[bytes]]
 ) -> Union[list, str, bytes]:
@@ -109,7 +143,7 @@
         all_data = []
         all_params: dict = {}
         for internal_id, request_input in request_inputs.items():
-            all_params = _merge_parameters(all_params, request_input)
+            all_params = _merge_input_parameters(all_params, request_input)
             all_data.append(_get_data(request_input))
             minibatch_shape = Shape(request_input.shape)
             self._minibatch_sizes[internal_id] = minibatch_shape.batch_size

{"golden_diff": "diff --git a/mlserver/batching/requests.py b/mlserver/batching/requests.py\n--- a/mlserver/batching/requests.py\n+++ b/mlserver/batching/requests.py\n@@ -29,6 +29,40 @@\n     return {**all_params, **obj_params}\n \n \n+def _merge_input_parameters(\n+    all_params: dict,\n+    parametrised_obj: Union[\n+        InferenceRequest, InferenceResponse, RequestInput, RequestOutput\n+    ],\n+) -> dict:\n+    if not parametrised_obj.parameters:\n+        return all_params\n+    obj_params = parametrised_obj.parameters.dict()\n+    if all_params == {}:\n+        return obj_params\n+    else:\n+        common_keys = set(all_params).intersection(set(obj_params)) - {\n+            \"content_type\",\n+            \"headers\",\n+        }\n+        uncommon_keys = set(all_params).union(set(obj_params)) - common_keys\n+        new_all_params = {}\n+        for key in common_keys:\n+            if type(all_params[key]) == list:\n+                new_value = all_params[key] + [obj_params[key]]\n+                new_all_params[key] = new_value\n+            else:\n+                new_value = [all_params[key]]\n+                new_value.append(obj_params[key])\n+                new_all_params[key] = new_value\n+        for key in uncommon_keys:\n+            if key in all_params.keys():\n+                new_all_params[key] = all_params[key]\n+            if key in obj_params.keys():\n+                new_all_params[key] = obj_params[key]\n+    return new_all_params\n+\n+\n def _merge_data(\n     all_data: Union[list, List[str], List[bytes]]\n ) -> Union[list, str, bytes]:\n@@ -109,7 +143,7 @@\n         all_data = []\n         all_params: dict = {}\n         for internal_id, request_input in request_inputs.items():\n-            all_params = _merge_parameters(all_params, request_input)\n+            all_params = _merge_input_parameters(all_params, request_input)\n             all_data.append(_get_data(request_input))\n             minibatch_shape = Shape(request_input.shape)\n             self._minibatch_sizes[internal_id] = minibatch_shape.batch_size\n", "issue": "Issue in aggregating batched requests parameters\nWhen sending batched requests with custom parameters (using extra field in [here](https://github.com/SeldonIO/MLServer/blob/749cd02bc5f3ee0b915151eedeff50075ac946ad/mlserver/types/dataplane.py#L27)) like the following:\r\n\r\n```python\r\n# request 1\r\npayload = types.InferenceRequest(\r\n    inputs=[\r\n        types.RequestInput(\r\n            name=\"parameters-np\",\r\n            shape=[1],\r\n            datatype=\"BYTES\",\r\n            data=[],\r\n            parameters=types.Parameters(\r\n                custom-param='value-1,\r\n        )\r\n    ]\r\n)\r\n# request 2\r\npayload = types.InferenceRequest(\r\n    inputs=[\r\n        types.RequestInput(\r\n            name=\"parameters-np\",\r\n            shape=[1],\r\n            datatype=\"BYTES\",\r\n            data=[],\r\n            parameters=types.Parameters(\r\n                custom-param='value-2',\r\n        )\r\n    ]\r\n)\r\n```\r\nthe results param field for the batched inference request will only contain a single `custom-param` field with the last request value for that, this is because a [dictionary is used](https://github.com/SeldonIO/MLServer/blob/749cd02bc5f3ee0b915151eedeff50075ac946ad/mlserver/batching/requests.py#L112) to aggregate parameters and for the case of having a same parameter with different values in the same batch only one value will be stored for them. E.g. in the case above we will only have `custom-param` equals to `value-2`. A better solution would be able to store all the values of common request parameter separately in a list e.g. `{custom-param=['value-1', 'value-2']`\r\n\n", "before_files": [{"content": "from collections import defaultdict, OrderedDict\nfrom typing import Dict, List, Optional, Union\n\nfrom ..types import (\n    InferenceRequest,\n    InferenceResponse,\n    Parameters,\n    RequestInput,\n    RequestOutput,\n    ResponseOutput,\n)\nfrom .shape import Shape\n\n\ndef _get_data(payload: Union[RequestInput, ResponseOutput]):\n    return getattr(payload.data, \"__root__\", payload.data)\n\n\ndef _merge_parameters(\n    all_params: dict,\n    parametrised_obj: Union[\n        InferenceRequest, InferenceResponse, RequestInput, RequestOutput\n    ],\n) -> dict:\n    if not parametrised_obj.parameters:\n        return all_params\n\n    obj_params = parametrised_obj.parameters.dict()\n    return {**all_params, **obj_params}\n\n\ndef _merge_data(\n    all_data: Union[list, List[str], List[bytes]]\n) -> Union[list, str, bytes]:\n    sampled_datum = all_data[0]\n\n    if isinstance(sampled_datum, str):\n        return \"\".join(all_data)  # type: ignore\n\n    if isinstance(sampled_datum, bytes):\n        return b\"\".join(all_data)  # type: ignore\n\n    if isinstance(sampled_datum, list):\n        return sum(all_data, [])\n\n    # TODO: Should we raise an error if we couldn't merge the data?\n    return all_data\n\n\nclass BatchedRequests:\n    def __init__(self, inference_requests: Dict[str, InferenceRequest] = {}):\n        self.inference_requests = inference_requests\n\n        # External IDs represent the incoming prediction IDs that need to match\n        # 1:1 between request and response.\n        # Since we can't ensure the uniqueness (or even presence) of the\n        # external IDs, we'll also maintain our own list of internal IDs.\n        self._ids_mapping: Dict[str, Optional[str]] = OrderedDict()\n\n        # Minibatch here refers to the individual batch size of the input head\n        # of each input request (i.e. the number of datapoints on each input\n        # request)\n        self._minibatch_sizes: Dict[str, int] = OrderedDict()\n\n        self.merged_request = self._merge_requests()\n\n    def _merge_requests(self) -> InferenceRequest:\n        inputs_index: Dict[str, Dict[str, RequestInput]] = defaultdict(OrderedDict)\n        outputs_index: Dict[str, Dict[str, RequestOutput]] = defaultdict(OrderedDict)\n        all_params: dict = {}\n        has_outputs = False  # if no outputs are defined, then outputs=None\n\n        for internal_id, inference_request in self.inference_requests.items():\n            self._ids_mapping[internal_id] = inference_request.id\n            all_params = _merge_parameters(all_params, inference_request)\n            for request_input in inference_request.inputs:\n                inputs_index[request_input.name][internal_id] = request_input\n\n            if inference_request.outputs is not None:\n                has_outputs = True\n                for request_output in inference_request.outputs:\n                    outputs_index[request_output.name][internal_id] = request_output\n\n        inputs = [\n            self._merge_request_inputs(request_inputs)\n            for request_inputs in inputs_index.values()\n        ]\n\n        outputs = (\n            [\n                self._merge_request_outputs(request_outputs)\n                for request_outputs in outputs_index.values()\n            ]\n            if has_outputs\n            else None\n        )\n\n        # TODO: Should we add a 'fake' request ID?\n        params = Parameters(**all_params) if all_params else None\n        return InferenceRequest(inputs=inputs, outputs=outputs, parameters=params)\n\n    def _merge_request_inputs(\n        self, request_inputs: Dict[str, RequestInput]\n    ) -> RequestInput:\n        # Note that minibatch sizes could be different on each input head,\n        # however, to simplify the implementation, here we assume that it will\n        # be the same across all of them\n        batch_size = 0\n        all_data = []\n        all_params: dict = {}\n        for internal_id, request_input in request_inputs.items():\n            all_params = _merge_parameters(all_params, request_input)\n            all_data.append(_get_data(request_input))\n            minibatch_shape = Shape(request_input.shape)\n            self._minibatch_sizes[internal_id] = minibatch_shape.batch_size\n            batch_size += minibatch_shape.batch_size\n\n        data = _merge_data(all_data)\n        parameters = Parameters(**all_params) if all_params else None\n\n        # TODO: What should we do if list is empty?\n        sampled = next(iter(request_inputs.values()))\n        shape = Shape(sampled.shape)\n        shape.batch_size = batch_size\n\n        return RequestInput(\n            name=sampled.name,\n            datatype=sampled.datatype,\n            shape=shape.to_list(),\n            data=data,\n            parameters=parameters,\n        )\n\n    def _merge_request_outputs(\n        self, request_outputs: Dict[str, RequestOutput]\n    ) -> RequestOutput:\n        all_params: dict = {}\n        for internal_id, request_output in request_outputs.items():\n            all_params = _merge_parameters(all_params, request_output)\n\n        parameters = Parameters(**all_params) if all_params else None\n\n        # TODO: What should we do if list is empty?\n        sampled = next(iter(request_outputs.values()))\n\n        return RequestOutput(name=sampled.name, parameters=parameters)\n\n    def split_response(\n        self, batched_response: InferenceResponse\n    ) -> Dict[str, InferenceResponse]:\n        responses: Dict[str, InferenceResponse] = {}\n\n        for response_output in batched_response.outputs:\n            response_outputs = self._split_response_output(response_output)\n\n            for internal_id, response_output in response_outputs.items():\n                if internal_id not in responses:\n                    responses[internal_id] = InferenceResponse(\n                        id=self._ids_mapping[internal_id],\n                        model_name=batched_response.model_name,\n                        model_version=batched_response.model_version,\n                        outputs=[],\n                        parameters=batched_response.parameters,\n                    )\n\n                responses[internal_id].outputs.append(response_output)\n\n        return responses\n\n    def _split_response_output(\n        self, response_output: ResponseOutput\n    ) -> Dict[str, ResponseOutput]:\n\n        all_data = self._split_data(response_output)\n        response_outputs = {}\n        for internal_id, data in all_data.items():\n            shape = Shape(response_output.shape)\n            shape.batch_size = self._minibatch_sizes[internal_id]\n            response_outputs[internal_id] = ResponseOutput(\n                name=response_output.name,\n                shape=shape.to_list(),\n                data=data,\n                datatype=response_output.datatype,\n                parameters=response_output.parameters,\n            )\n\n        return response_outputs\n\n    def _split_data(self, response_output: ResponseOutput) -> Dict[str, ResponseOutput]:\n        merged_shape = Shape(response_output.shape)\n        element_size = merged_shape.elem_size\n        merged_data = _get_data(response_output)\n        idx = 0\n\n        all_data = {}\n        # TODO: Don't rely on array to have been flattened\n        for internal_id, minibatch_size in self._minibatch_sizes.items():\n            data = merged_data[idx : idx + minibatch_size * element_size]\n            idx += minibatch_size * element_size\n            all_data[internal_id] = data\n\n        return all_data\n", "path": "mlserver/batching/requests.py"}], "after_files": [{"content": "from collections import defaultdict, OrderedDict\nfrom typing import Dict, List, Optional, Union\n\nfrom ..types import (\n    InferenceRequest,\n    InferenceResponse,\n    Parameters,\n    RequestInput,\n    RequestOutput,\n    ResponseOutput,\n)\nfrom .shape import Shape\n\n\ndef _get_data(payload: Union[RequestInput, ResponseOutput]):\n    return getattr(payload.data, \"__root__\", payload.data)\n\n\ndef _merge_parameters(\n    all_params: dict,\n    parametrised_obj: Union[\n        InferenceRequest, InferenceResponse, RequestInput, RequestOutput\n    ],\n) -> dict:\n    if not parametrised_obj.parameters:\n        return all_params\n\n    obj_params = parametrised_obj.parameters.dict()\n    return {**all_params, **obj_params}\n\n\ndef _merge_input_parameters(\n    all_params: dict,\n    parametrised_obj: Union[\n        InferenceRequest, InferenceResponse, RequestInput, RequestOutput\n    ],\n) -> dict:\n    if not parametrised_obj.parameters:\n        return all_params\n    obj_params = parametrised_obj.parameters.dict()\n    if all_params == {}:\n        return obj_params\n    else:\n        common_keys = set(all_params).intersection(set(obj_params)) - {\n            \"content_type\",\n            \"headers\",\n        }\n        uncommon_keys = set(all_params).union(set(obj_params)) - common_keys\n        new_all_params = {}\n        for key in common_keys:\n            if type(all_params[key]) == list:\n                new_value = all_params[key] + [obj_params[key]]\n                new_all_params[key] = new_value\n            else:\n                new_value = [all_params[key]]\n                new_value.append(obj_params[key])\n                new_all_params[key] = new_value\n        for key in uncommon_keys:\n            if key in all_params.keys():\n                new_all_params[key] = all_params[key]\n            if key in obj_params.keys():\n                new_all_params[key] = obj_params[key]\n    return new_all_params\n\n\ndef _merge_data(\n    all_data: Union[list, List[str], List[bytes]]\n) -> Union[list, str, bytes]:\n    sampled_datum = all_data[0]\n\n    if isinstance(sampled_datum, str):\n        return \"\".join(all_data)  # type: ignore\n\n    if isinstance(sampled_datum, bytes):\n        return b\"\".join(all_data)  # type: ignore\n\n    if isinstance(sampled_datum, list):\n        return sum(all_data, [])\n\n    # TODO: Should we raise an error if we couldn't merge the data?\n    return all_data\n\n\nclass BatchedRequests:\n    def __init__(self, inference_requests: Dict[str, InferenceRequest] = {}):\n        self.inference_requests = inference_requests\n\n        # External IDs represent the incoming prediction IDs that need to match\n        # 1:1 between request and response.\n        # Since we can't ensure the uniqueness (or even presence) of the\n        # external IDs, we'll also maintain our own list of internal IDs.\n        self._ids_mapping: Dict[str, Optional[str]] = OrderedDict()\n\n        # Minibatch here refers to the individual batch size of the input head\n        # of each input request (i.e. the number of datapoints on each input\n        # request)\n        self._minibatch_sizes: Dict[str, int] = OrderedDict()\n\n        self.merged_request = self._merge_requests()\n\n    def _merge_requests(self) -> InferenceRequest:\n        inputs_index: Dict[str, Dict[str, RequestInput]] = defaultdict(OrderedDict)\n        outputs_index: Dict[str, Dict[str, RequestOutput]] = defaultdict(OrderedDict)\n        all_params: dict = {}\n        has_outputs = False  # if no outputs are defined, then outputs=None\n\n        for internal_id, inference_request in self.inference_requests.items():\n            self._ids_mapping[internal_id] = inference_request.id\n            all_params = _merge_parameters(all_params, inference_request)\n            for request_input in inference_request.inputs:\n                inputs_index[request_input.name][internal_id] = request_input\n\n            if inference_request.outputs is not None:\n                has_outputs = True\n                for request_output in inference_request.outputs:\n                    outputs_index[request_output.name][internal_id] = request_output\n\n        inputs = [\n            self._merge_request_inputs(request_inputs)\n            for request_inputs in inputs_index.values()\n        ]\n\n        outputs = (\n            [\n                self._merge_request_outputs(request_outputs)\n                for request_outputs in outputs_index.values()\n            ]\n            if has_outputs\n            else None\n        )\n\n        # TODO: Should we add a 'fake' request ID?\n        params = Parameters(**all_params) if all_params else None\n        return InferenceRequest(inputs=inputs, outputs=outputs, parameters=params)\n\n    def _merge_request_inputs(\n        self, request_inputs: Dict[str, RequestInput]\n    ) -> RequestInput:\n        # Note that minibatch sizes could be different on each input head,\n        # however, to simplify the implementation, here we assume that it will\n        # be the same across all of them\n        batch_size = 0\n        all_data = []\n        all_params: dict = {}\n        for internal_id, request_input in request_inputs.items():\n            all_params = _merge_input_parameters(all_params, request_input)\n            all_data.append(_get_data(request_input))\n            minibatch_shape = Shape(request_input.shape)\n            self._minibatch_sizes[internal_id] = minibatch_shape.batch_size\n            batch_size += minibatch_shape.batch_size\n\n        data = _merge_data(all_data)\n        parameters = Parameters(**all_params) if all_params else None\n\n        # TODO: What should we do if list is empty?\n        sampled = next(iter(request_inputs.values()))\n        shape = Shape(sampled.shape)\n        shape.batch_size = batch_size\n\n        return RequestInput(\n            name=sampled.name,\n            datatype=sampled.datatype,\n            shape=shape.to_list(),\n            data=data,\n            parameters=parameters,\n        )\n\n    def _merge_request_outputs(\n        self, request_outputs: Dict[str, RequestOutput]\n    ) -> RequestOutput:\n        all_params: dict = {}\n        for internal_id, request_output in request_outputs.items():\n            all_params = _merge_parameters(all_params, request_output)\n\n        parameters = Parameters(**all_params) if all_params else None\n\n        # TODO: What should we do if list is empty?\n        sampled = next(iter(request_outputs.values()))\n\n        return RequestOutput(name=sampled.name, parameters=parameters)\n\n    def split_response(\n        self, batched_response: InferenceResponse\n    ) -> Dict[str, InferenceResponse]:\n        responses: Dict[str, InferenceResponse] = {}\n\n        for response_output in batched_response.outputs:\n            response_outputs = self._split_response_output(response_output)\n\n            for internal_id, response_output in response_outputs.items():\n                if internal_id not in responses:\n                    responses[internal_id] = InferenceResponse(\n                        id=self._ids_mapping[internal_id],\n                        model_name=batched_response.model_name,\n                        model_version=batched_response.model_version,\n                        outputs=[],\n                        parameters=batched_response.parameters,\n                    )\n\n                responses[internal_id].outputs.append(response_output)\n\n        return responses\n\n    def _split_response_output(\n        self, response_output: ResponseOutput\n    ) -> Dict[str, ResponseOutput]:\n\n        all_data = self._split_data(response_output)\n        response_outputs = {}\n        for internal_id, data in all_data.items():\n            shape = Shape(response_output.shape)\n            shape.batch_size = self._minibatch_sizes[internal_id]\n            response_outputs[internal_id] = ResponseOutput(\n                name=response_output.name,\n                shape=shape.to_list(),\n                data=data,\n                datatype=response_output.datatype,\n                parameters=response_output.parameters,\n            )\n\n        return response_outputs\n\n    def _split_data(self, response_output: ResponseOutput) -> Dict[str, ResponseOutput]:\n        merged_shape = Shape(response_output.shape)\n        element_size = merged_shape.elem_size\n        merged_data = _get_data(response_output)\n        idx = 0\n\n        all_data = {}\n        # TODO: Don't rely on array to have been flattened\n        for internal_id, minibatch_size in self._minibatch_sizes.items():\n            data = merged_data[idx : idx + minibatch_size * element_size]\n            idx += minibatch_size * element_size\n            all_data[internal_id] = data\n\n        return all_data\n", "path": "mlserver/batching/requests.py"}]}