RunnableLambda Class — langchain Architecture

Source Code

libs/core/langchain_core/runnables/base.py lines 4399–5269

class RunnableLambda(Runnable[Input, Output]):
    """`RunnableLambda` converts a python callable into a `Runnable`.

    Wrapping a callable in a `RunnableLambda` makes the callable usable
    within either a sync or async context.

    `RunnableLambda` can be composed as any other `Runnable` and provides
    seamless integration with LangChain tracing.

    `RunnableLambda` is best suited for code that does not need to support
    streaming. If you need to support streaming (i.e., be able to operate
    on chunks of inputs and yield chunks of outputs), use `RunnableGenerator`
    instead.

    Note that if a `RunnableLambda` returns an instance of `Runnable`, that
    instance is invoked (or streamed) during execution.

    Examples:
        ```python
        # This is a RunnableLambda
        from langchain_core.runnables import RunnableLambda


        def add_one(x: int) -> int:
            return x + 1


        runnable = RunnableLambda(add_one)

        runnable.invoke(1)  # returns 2
        runnable.batch([1, 2, 3])  # returns [2, 3, 4]

        # Async is supported by default by delegating to the sync implementation
        await runnable.ainvoke(1)  # returns 2
        await runnable.abatch([1, 2, 3])  # returns [2, 3, 4]


        # Alternatively, can provide both synd and sync implementations
        async def add_one_async(x: int) -> int:
            return x + 1


        runnable = RunnableLambda(add_one, afunc=add_one_async)
        runnable.invoke(1)  # Uses add_one
        await runnable.ainvoke(1)  # Uses add_one_async
        ```
    """

    @overload
    def __init__(
        self,
        func: Callable[[Input, RunnableConfig], Awaitable[Output]],
        afunc: None = None,
        name: str | None = None,
    ) -> None: ...

    @overload
    def __init__(
        self,
        func: Callable[[Input], Awaitable[Output]],
        afunc: None = None,
        name: str | None = None,
    ) -> None: ...

    @overload
    def __init__(
        self,
        func: Callable[[Input], AsyncIterator[Output]],
        afunc: None = None,
        name: str | None = None,
    ) -> None: ...

    @overload
    def __init__(
        self,
        func: Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]],
        afunc: None = None,
        name: str | None = None,
    ) -> None: ...

    @overload