Invoke the chain on a batch of inputs in async mode :param inputs: A batch of inputs :return: A list of dicts with the defined json schema
(self, inputs: list[dict])
| 107 | return results |
| 108 | |
| 109 | async def async_batch_invoke(self, inputs: list[dict]) -> list[dict]: |
| 110 | """ |
| 111 | Invoke the chain on a batch of inputs in async mode |
| 112 | :param inputs: A batch of inputs |
| 113 | :return: A list of dicts with the defined json schema |
| 114 | """ |
| 115 | with self.callback() as cb: |
| 116 | tasks = [self.chain.ainvoke(chain_input) for chain_input in inputs] |
| 117 | all_res = await self.retry_operation(tasks) |
| 118 | self.accumulate_usage += cb.total_cost |
| 119 | if self.parser_func is not None: |
| 120 | return [self.parser_func(t.result()) for t in list(all_res)] |
| 121 | return [t.result() for t in list(all_res)] |
| 122 | |
| 123 | def batch_invoke(self, inputs: list[dict], num_workers: int): |
| 124 | """ |
no test coverage detected