) -> Output:
        return self._call_with_config(self._invoke, input, config, **kwargs)
    async def _ainvoke(
        self,
        input: Input,
        run_manager: "AsyncCallbackManagerForChainRun",
        config: RunnableConfig,
    ) -> Output:
        async for attempt in self._async_retrying(reraise=True):
            with attempt:
                result = await super().ainvoke(
                    input,
                    self._patch_config(config, run_manager, attempt.retry_state),
                )
            if attempt.retry_state.outcome and not attempt.retry_state.outcome.failed:
                attempt.retry_state.set_result(result)
        return result
[docs]    async def ainvoke(
        self, input: Input, config: Optional[RunnableConfig] = None, **kwargs: Any
    ) -> Output:
        return await self._acall_with_config(self._ainvoke, input, config, **kwargs)
    def _batch(
        self,
        inputs: List[Input],
        run_manager: List["CallbackManagerForChainRun"],
        config: List[RunnableConfig],
    ) -> List[Union[Output, Exception]]:
        results_map: Dict[int, Output] = {}
        def pending(iterable: List[U]) -> List[U]:
            return [item for idx, item in enumerate(iterable) if idx not in results_map]
        try:
            for attempt in self._sync_retrying():
                with attempt:
                    # Get the results of the inputs that have not succeeded yet.
                    result = super().batch(
                        pending(inputs),
                        self._patch_config_list(
                            pending(config), pending(run_manager), attempt.retry_state
                        ),
                        return_exceptions=True,
                    )