RunnableLambda#

class langchain_core.runnables.base.RunnableLambda(
func: Callable[[Input], Iterator[Output]] | Callable[[Input], Runnable[Input, Output]] | Callable[[Input], Output] | Callable[[Input, RunnableConfig], Output] | Callable[[Input, CallbackManagerForChainRun], Output] | Callable[[Input, CallbackManagerForChainRun, RunnableConfig], Output] | Callable[[Input], Awaitable[Output]] | Callable[[Input], AsyncIterator[Output]] | Callable[[Input, RunnableConfig], Awaitable[Output]] | Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]] | Callable[[Input, AsyncCallbackManagerForChainRun, RunnableConfig], Awaitable[Output]],
afunc: Callable[[Input], Awaitable[Output]] | Callable[[Input], AsyncIterator[Output]] | Callable[[Input, RunnableConfig], Awaitable[Output]] | Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]] | Callable[[Input, AsyncCallbackManagerForChainRun, RunnableConfig], Awaitable[Output]] | None = None,
name: str | None = None,
)[source]#

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

# 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

Create a RunnableLambda from a callable, and async callable or both.

Accepts both sync and async variants to allow providing efficient implementations for sync and async execution.

Parameters:
Raises:

  • TypeError – If the func is not a callable type.

  • TypeError – If both func and afunc are provided.

Note

RunnableLambda implements the standard Runnable Interface. 🏃

The Runnable Interface has additional methods that are available on runnables, such as with_config, with_types, with_retry, assign, bind, get_graph, and more.

Attributes

Methods

__init__(*args, **kwargs)

Initialize self.

Examples using RunnableLambda

On this page