],
    input: Input,
    run_manager: CallbackManagerForChainRun,
    config: RunnableConfig,
    **kwargs: Any,
) -> Output:
    """Call function that may optionally accept a run_manager and/or config."""
    if accepts_config(func):
        kwargs["config"] = patch_config(config, callbacks=run_manager.get_child())
    if accepts_run_manager(func):
        kwargs["run_manager"] = run_manager
    return func(input, **kwargs)  # type: ignore[call-arg]
[docs]async def acall_func_with_variable_args(
    func: Union[
        Callable[[Input], Awaitable[Output]],
        Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]],
        Callable[
            [Input, AsyncCallbackManagerForChainRun, RunnableConfig],
            Awaitable[Output],
        ],
    ],
    input: Input,
    run_manager: AsyncCallbackManagerForChainRun,
    config: RunnableConfig,
    **kwargs: Any,
) -> Output:
    """Call function that may optionally accept a run_manager and/or config."""
    if accepts_config(func):
        kwargs["config"] = patch_config(config, callbacks=run_manager.get_child())
    if accepts_run_manager(func):
        kwargs["run_manager"] = run_manager
    return await func(input, **kwargs)  # type: ignore[call-arg]
[docs]def get_callback_manager_for_config(config: RunnableConfig) -> CallbackManager:
    from langchain.callbacks.manager import CallbackManager
    return CallbackManager.configure(
        inheritable_callbacks=config.get("callbacks"),
        inheritable_tags=config.get("tags"),
        inheritable_metadata=config.get("metadata"),
    )
[docs]def get_async_callback_manager_for_config(
    config: RunnableConfig,