config: RunnableConfig,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        assert isinstance(
            input, dict
        ), "The input to RunnablePassthrough.assign() must be a dict."
        return {
            **input,
            **self.mapper.invoke(
                input,
                patch_config(config, callbacks=run_manager.get_child()),
                **kwargs,
            ),
        }
[docs]    def invoke(
        self,
        input: Dict[str, Any],
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        return self._call_with_config(self._invoke, input, config, **kwargs)
    async def _ainvoke(
        self,
        input: Dict[str, Any],
        run_manager: AsyncCallbackManagerForChainRun,
        config: RunnableConfig,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        assert isinstance(
            input, dict
        ), "The input to RunnablePassthrough.assign() must be a dict."
        return {
            **input,
            **await self.mapper.ainvoke(
                input,
                patch_config(config, callbacks=run_manager.get_child()),
                **kwargs,
            ),
        }
[docs]    async def ainvoke(
        self,
        input: Dict[str, Any],
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        return await self._acall_with_config(self._ainvoke, input, config, **kwargs)
    def _transform(
        self,
        input: Iterator[Dict[str, Any]],
        run_manager: CallbackManagerForChainRun,