input format for QueryPipeline._arun_multi?

[johnhu33]

what input and format is needed for QueryPipeline._arun_multi?

[dosubot[bot]]

The input needed for QueryPipeline._arun_multi is a dictionary where the keys are the root module keys and the values are the corresponding input data for those modules. The format is as follows:

module_input_dict: Dict[str, Any]

For example, if you have two components qc1_0 and qc1_1 that need inputs, the module_input_dict would look like this:

{
    "qc1_0": {"input1": 1, "input2": 2},
    "qc1_1": {"input1": 3, "input2": 4}
}

If you are running in batch mode, the inputs would be lists of values:

{
    "qc1_0": {"input1": [1, 5], "input2": [2, 1]},
    "qc1_1": {"input1": [3, 7], "input2": [4, 2]}
}

^[1][2][3]

To continue talking to Dosu, mention @dosu.

[johnhu33]

@dosu how to build a QueryPipeline that can retrieve multi inputs and return multi outputs in parallel?

[dosubot[bot]]

To build a QueryPipeline that can retrieve multiple inputs and return multiple outputs in parallel, you can use the run_multi method. This method is designed to handle multiple roots and execute them in parallel.

Here's an example of how you can use the run_multi method:

def run_multi(
        self,
        module_input_dict: Dict[str, Any],
        callback_manager: Optional[CallbackManager] = None,
        batch: bool = False,
    ) -> Dict[str, Any]:
    """Run the pipeline for multiple roots."""
    callback_manager = callback_manager or self.callback_manager
    self.set_callback_manager(callback_manager)
    with self.callback_manager.as_trace("query"):
        with self.callback_manager.event(
            CBEventType.QUERY,
            payload={EventPayload.QUERY_STR: json.dumps(module_input_dict)},
        ) as query_event:
            if batch:
                outputs: Dict[str, Any] = {}

                batch_lengths = {
                    len(values)
                    for subdict in module_input_dict.values()
                    for values in subdict.values()
                }

                if len(batch_lengths) != 1:
                    raise ValueError("Length of batch inputs must be the same.")

                batch_size = next(iter(batch_lengths))

                # List individual outputs from batch multi input.
                inputs = [
                    {
                        key: {
                            inner_key: inner_val[i]
                            for inner_key, inner_val in value.items()
                        }
                        for key, value in module_input_dict.items()
                    }
                    for i in range(batch_size)
                ]
                jobs = [self._arun_multi(input) for input in inputs]
                results = asyncio_run(run_jobs(jobs, workers=len(jobs)))

                for result in results:
                    outputs = self.merge_dicts(outputs, result[0])

                return outputs
            else:
                outputs, _ = self._run_multi(module_input_dict)
                return outputs

This method allows you to run multiple inputs in parallel and return their outputs. The batch parameter can be used to handle batch processing if needed.

Additionally, you can refer to the test_query_pipeline_multi function for an example of how to set up and use the run_multi method in a test scenario:

def test_query_pipeline_multi() -> None:
    """Test query pipeline."""
    # try run run_multi
    # link both qc1_0 and qc1_1 to qc2
    qc1_0 = QueryComponent1()
    qc1_1 = QueryComponent1()
    qc2 = QueryComponent2()
    p = QueryPipeline()
    p.add_modules({"qc1_0": qc1_0, "qc1_1": qc1_1, "qc2": qc2})
    p.add_link("qc1_0", "qc2", dest_key="input1")
    p.add_link("qc1_1", "qc2", dest_key="input2")
    output = p.run_multi(
        {"qc1_0": {"input1": 1, "input2": 2}, "qc1_1": {"input1": 3, "input2": 4}}
    )
    assert output == {"qc2": {"output": "3:7"}}

This test demonstrates how to configure a QueryPipeline with multiple components and links, and how to use the run_multi method to process multiple inputs in parallel.

For more advanced scenarios, such as handling batch processing or retrieving intermediate outputs, you can refer to the test_query_pipeline_multi_batch and test_query_pipeline_multi_intermediate_output functions ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[johnhu33]

@Duso that's not quite what i want. please write a code snipet that first build a query pipeline that can take multiple input and retrieve multiple output

[johnhu33]

@Duso that's not quite what i want. please write a code snipet that first build a query pipeline that can take multiple input and retrieve multiple output