We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
splitting aggregated batched requests parameters
The reverse of https://github.com/SeldonIO/MLServer/issues/915 and https://github.com/SeldonIO/MLServer/pull/916 should also be implemented since when a series of parameters for a batched request are sent in the model, the MLServer should be able to un-batch them correctly. E.g. following request:
```python
            ResponseOutput(
                name="foo",
                datatype="INT32",
                shape=[3, 3],
                data=[1, 2, 3, 4, 5, 6, 7, 8, 9],
                parameters=Parameters(
                    content_type="np",
                    foo=["foo_1", "foo_2"],
                    bar=["bar_1", "bar_2", "bar_3"],
                ),
            )
```
should be un-batched as follows:
```python
            [
                ResponseOutput(
                    name="foo",
                    datatype="INT32",
                    shape=[1, 3],
                    data=[1, 2, 3],
                    parameters=Parameters(
                        content_type="np", foo="['foo_1']", bar="['bar_1']"
                    ),
                ),
                ResponseOutput(
                    name="foo",
                    datatype="INT32",
                    shape=[1, 3],
                    data=[4, 5, 6],
                    parameters=Parameters(
                        content_type="np", foo="['foo_2']", bar="['bar_2']"
                    ),
                ),
                ResponseOutput(
                    name="foo",
                    datatype="INT32",
                    shape=[1, 3],
                    data=[7, 8, 9],
                    parameters=Parameters(content_type="np", bar="['bar_3']"),
                ),
            ],
```

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

{"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@@ -1,5 +1,5 @@\n from collections import defaultdict, OrderedDict\n-from typing import Dict, List, Optional, Union\n+from typing import Dict, List, Optional, Union, Any, DefaultDict\n \n from ..types import (\n     InferenceRequest,\n@@ -16,6 +16,22 @@\n     return getattr(payload.data, \"__root__\", payload.data)\n \n \n+def _get_parameters(payload: ResponseOutput) -> DefaultDict[Any, Any]:\n+    parameters = defaultdict(list)\n+    if payload.parameters is not None:\n+        payload_parameters = payload.parameters.dict()\n+    for param_name, param_values in payload_parameters.items():\n+        if param_name in [\"content_type\", \"headers\"]:\n+            continue\n+        for param_value in param_values:\n+            parameters[param_name].append(param_value)\n+    if \"content_type\" in payload_parameters.keys():\n+        parameters[\"content_type\"] = payload_parameters[\"content_type\"]\n+    if \"headers\" in payload_parameters.keys():\n+        parameters[\"headers\"] = payload_parameters[\"headers\"]\n+    return parameters\n+\n+\n def _merge_parameters(\n     all_params: dict,\n     parametrised_obj: Union[\n@@ -206,6 +222,10 @@\n     ) -> Dict[str, ResponseOutput]:\n \n         all_data = self._split_data(response_output)\n+        if response_output.parameters is not None:\n+            all_parameters = self._split_parameters(response_output)\n+        else:\n+            all_parameters = None\n         response_outputs = {}\n         for internal_id, data in all_data.items():\n             shape = Shape(response_output.shape)\n@@ -215,12 +235,14 @@\n                 shape=shape.to_list(),\n                 data=data,\n                 datatype=response_output.datatype,\n-                parameters=response_output.parameters,\n+                parameters=all_parameters\n+                if all_parameters is None\n+                else all_parameters[internal_id],\n             )\n \n         return response_outputs\n \n-    def _split_data(self, response_output: ResponseOutput) -> Dict[str, ResponseOutput]:\n+    def _split_data(self, response_output: ResponseOutput) -> Dict[str, Any]:\n         merged_shape = Shape(response_output.shape)\n         element_size = merged_shape.elem_size\n         merged_data = _get_data(response_output)\n@@ -234,3 +256,32 @@\n             all_data[internal_id] = data\n \n         return all_data\n+\n+    def _split_parameters(\n+        self, response_output: ResponseOutput\n+    ) -> Dict[str, Parameters]:\n+        merged_parameters = _get_parameters(response_output)\n+        idx = 0\n+\n+        all_parameters = {}\n+        # TODO: Don't rely on array to have been flattened\n+        for internal_id, minibatch_size in self._minibatch_sizes.items():\n+            parameter_args = {}\n+            for parameter_name, parameter_values in merged_parameters.items():\n+                if parameter_name in [\"content_type\", \"headers\"]:\n+                    continue\n+                try:\n+                    parameter_value = parameter_values[idx]\n+                    if parameter_value != []:\n+                        parameter_args[parameter_name] = str(parameter_value)\n+                except IndexError:\n+                    pass\n+            if \"content_type\" in merged_parameters.keys():\n+                parameter_args[\"content_type\"] = merged_parameters[\"content_type\"]\n+            if \"headers\" in merged_parameters.keys():\n+                parameter_args[\"headers\"] = merged_parameters[\"headers\"]\n+            parameter_obj = Parameters(**parameter_args)\n+            all_parameters[internal_id] = parameter_obj\n+            idx += minibatch_size\n+\n+        return all_parameters\n", "issue": "splitting aggregated batched requests parameters\nThe reverse of https://github.com/SeldonIO/MLServer/issues/915 and https://github.com/SeldonIO/MLServer/pull/916 should also be implemented since when a series of parameters for a batched request are sent in the model, the MLServer should be able to un-batch them correctly. E.g. following request:\r\n```python\r\n            ResponseOutput(\r\n                name=\"foo\",\r\n                datatype=\"INT32\",\r\n                shape=[3, 3],\r\n                data=[1, 2, 3, 4, 5, 6, 7, 8, 9],\r\n                parameters=Parameters(\r\n                    content_type=\"np\",\r\n                    foo=[\"foo_1\", \"foo_2\"],\r\n                    bar=[\"bar_1\", \"bar_2\", \"bar_3\"],\r\n                ),\r\n            )\r\n```\r\nshould be un-batched as follows:\r\n```python\r\n            [\r\n                ResponseOutput(\r\n                    name=\"foo\",\r\n                    datatype=\"INT32\",\r\n                    shape=[1, 3],\r\n                    data=[1, 2, 3],\r\n                    parameters=Parameters(\r\n                        content_type=\"np\", foo=\"['foo_1']\", bar=\"['bar_1']\"\r\n                    ),\r\n                ),\r\n                ResponseOutput(\r\n                    name=\"foo\",\r\n                    datatype=\"INT32\",\r\n                    shape=[1, 3],\r\n                    data=[4, 5, 6],\r\n                    parameters=Parameters(\r\n                        content_type=\"np\", foo=\"['foo_2']\", bar=\"['bar_2']\"\r\n                    ),\r\n                ),\r\n                ResponseOutput(\r\n                    name=\"foo\",\r\n                    datatype=\"INT32\",\r\n                    shape=[1, 3],\r\n                    data=[7, 8, 9],\r\n                    parameters=Parameters(content_type=\"np\", bar=\"['bar_3']\"),\r\n                ),\r\n            ],\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_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"}], "after_files": [{"content": "from collections import defaultdict, OrderedDict\nfrom typing import Dict, List, Optional, Union, Any, DefaultDict\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 _get_parameters(payload: ResponseOutput) -> DefaultDict[Any, Any]:\n    parameters = defaultdict(list)\n    if payload.parameters is not None:\n        payload_parameters = payload.parameters.dict()\n    for param_name, param_values in payload_parameters.items():\n        if param_name in [\"content_type\", \"headers\"]:\n            continue\n        for param_value in param_values:\n            parameters[param_name].append(param_value)\n    if \"content_type\" in payload_parameters.keys():\n        parameters[\"content_type\"] = payload_parameters[\"content_type\"]\n    if \"headers\" in payload_parameters.keys():\n        parameters[\"headers\"] = payload_parameters[\"headers\"]\n    return parameters\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        if response_output.parameters is not None:\n            all_parameters = self._split_parameters(response_output)\n        else:\n            all_parameters = None\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=all_parameters\n                if all_parameters is None\n                else all_parameters[internal_id],\n            )\n\n        return response_outputs\n\n    def _split_data(self, response_output: ResponseOutput) -> Dict[str, Any]:\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\n    def _split_parameters(\n        self, response_output: ResponseOutput\n    ) -> Dict[str, Parameters]:\n        merged_parameters = _get_parameters(response_output)\n        idx = 0\n\n        all_parameters = {}\n        # TODO: Don't rely on array to have been flattened\n        for internal_id, minibatch_size in self._minibatch_sizes.items():\n            parameter_args = {}\n            for parameter_name, parameter_values in merged_parameters.items():\n                if parameter_name in [\"content_type\", \"headers\"]:\n                    continue\n                try:\n                    parameter_value = parameter_values[idx]\n                    if parameter_value != []:\n                        parameter_args[parameter_name] = str(parameter_value)\n                except IndexError:\n                    pass\n            if \"content_type\" in merged_parameters.keys():\n                parameter_args[\"content_type\"] = merged_parameters[\"content_type\"]\n            if \"headers\" in merged_parameters.keys():\n                parameter_args[\"headers\"] = merged_parameters[\"headers\"]\n            parameter_obj = Parameters(**parameter_args)\n            all_parameters[internal_id] = parameter_obj\n            idx += minibatch_size\n\n        return all_parameters\n", "path": "mlserver/batching/requests.py"}]}