RunnableLambda#

class langchain_core.runnables.base.RunnableLambda(func: Callable[[Input], Output] | Callable[[Input], Iterator[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 将一个可调用的 Python 对象转换为 Runnable。

将可调用对象包装在RunnableLambda中，使得该可调用对象可以在同步或异步上下文中使用。

RunnableLambda 可以像任何其他 Runnable 一样组合，并提供与 LangChain 跟踪的无缝集成。

RunnableLambda 最适合不需要支持流式处理的代码。如果你需要支持流式处理（即能够操作输入块并生成输出块），请改用 RunnableGenerator。

请注意，如果RunnableLambda返回一个Runnable的实例，该实例在执行期间会被调用（或流式处理）。

示例

# 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

从一个可调用对象、异步可调用对象或两者创建一个RunnableLambda。

接受同步和异步变体，以便为同步和异步执行提供高效的实现。

Parameters:

func (Union[Union[Callable[[Input], Output], Callable[[Input], Iterator[Output]], Callable[[Input, RunnableConfig], Output], Callable[[Input, CallbackManagerForChainRun], Output], Callable[[Input, CallbackManagerForChainRun, RunnableConfig], Output]], Union[Callable[[Input], Awaitable[Output]], Callable[[Input], AsyncIterator[Output]], Callable[[Input, RunnableConfig], Awaitable[Output]], Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]], Callable[[Input, AsyncCallbackManagerForChainRun, RunnableConfig], Awaitable[Output]]]]) – Either sync or async callable
afunc (可选[联合[可调用[[输入], 可等待[输出]], 可调用[[输入], 异步迭代器[输出]], 可调用[[输入, RunnableConfig], 可等待[输出]], 可调用[[输入, AsyncCallbackManagerForChainRun], 可等待[输出]], 可调用[[输入, AsyncCallbackManagerForChainRun, RunnableConfig], 可等待[输出]]]]) – 一个异步可调用函数，接受输入并返回输出。默认为 None。
name (str | None) – Runnable 的名称。默认为 None。

Raises:

TypeError – 如果 func 不是一个可调用类型。
TypeError – 如果同时提供了 func 和 afunc。

注意

RunnableLambda 实现了标准的 Runnable Interface。🏃

Runnable Interface 接口在可运行对象上提供了额外的方法，例如 with_types, with_retry, assign, bind, get_graph, 以及更多。

属性

方法

__init__(*args, **kwargs)

初始化自身。

使用 RunnableLambda 的示例