跳到主要内容
Open on GitHub

可运行接口

Runnable 接口是使用 LangChain 组件的基础,它在许多组件中都有实现,例如语言模型输出解析器检索器编译后的 LangGraph 图等等。

本指南涵盖了 Runnable 接口的主要概念和方法,它允许开发者以一致且可预测的方式与各种 LangChain 组件进行交互。

相关资源

Runnable 接口概述

Runnable 方式定义了一个标准接口,允许 Runnable 组件

  • 调用:将单个输入转换为输出。
  • 批量处理:将多个输入高效地转换为输出。
  • 流式传输:输出在其生成时进行流式传输。
  • 检查:可以访问有关 Runnable 输入、输出和配置的示意信息。
  • 组合:可以使用LangChain 表达式语言 (LCEL) 将多个 Runnable 组合在一起以创建复杂的管道。

请查看LCEL 速查表以了解涉及 Runnable 接口和 LCEL 表达式的一些常见模式。

优化的并行执行(批量)

LangChain Runnable 提供内置的 batch(和 batch_as_completed)API,允许您并行处理多个输入。

当需要处理多个独立输入时,使用这些方法可以显著提高性能,因为处理可以并行而不是顺序进行。

两种批量处理选项是

  • batch:并行处理多个输入,并按与输入相同的顺序返回结果。
  • batch_as_completed:并行处理多个输入,并在完成时返回结果。结果可能无序到达,但每个结果都包含输入索引以进行匹配。

batchbatch_as_completed 的默认实现使用线程池执行器来并行运行 invoke 方法。这允许高效的并行执行,而无需用户管理线程,并加速 I/O 密集型代码(例如,发出 API 请求、读取文件等)。对于 CPU 密集型操作,它不会那么有效,因为 Python 中的 GIL(全局解释器锁)会阻止真正的并行执行。

某些 Runnable 可能会提供其自己的 batchbatch_as_completed 实现,这些实现针对其特定用例进行了优化(例如,依赖于模型提供商提供的 batch API)。

注意

abatchabatch_as_completed 的异步版本依赖于 asyncio 的gatheras_completed 函数来并行运行 ainvoke 方法。

提示

当使用 batchbatch_as_completed 处理大量输入时,用户可能希望控制最大并行调用数量。这可以通过在 RunnableConfig 字典中设置 max_concurrency 属性来完成。有关更多信息,请参阅RunnableConfig

聊天模型 (Chat Models) 还具有内置的速率限制器,可用于控制请求的速率。

异步支持

Runnable 暴露了一个异步 API,允许它们在 Python 中使用 await 语法进行调用。异步方法可以通过“a”前缀来识别(例如,ainvokeabatchastreamabatch_as_completed)。

请参阅LangChain 异步编程指南以获取更多详细信息。

流式 API

流式传输对于使基于 LLM 的应用程序对最终用户响应迅速至关重要。

Runnable 暴露了以下三个流式 API

  1. 同步 stream 和异步 astream:在生成时产生 Runnable 的输出。
  2. 异步 astream_events:一个更高级的流式 API,允许流式传输中间步骤和最终输出
  3. 遗留异步 astream_log:一个遗留流式 API,用于流式传输中间步骤和最终输出

请参阅流式传输概念指南以获取有关如何在 LangChain 中进行流式传输的更多详细信息。

输入和输出类型

每个 Runnable 都由一个输入类型和一个输出类型来表征。这些输入和输出类型可以是任何 Python 对象,并由 Runnable 本身定义。

导致 Runnable 执行的 Runnable 方法(例如,invokebatchstreamastream_events)都使用这些输入和输出类型。

  • invoke:接受一个输入并返回一个输出。
  • batch:接受一个输入列表并返回一个输出列表。
  • stream:接受一个输入并返回一个产生输出的生成器。

输入类型输出类型因组件而异

组件输入类型输出类型
提示词字典PromptValue
聊天模型字符串、聊天消息列表或 PromptValueChatMessage
LLM字符串、聊天消息列表或 PromptValue字符串
输出解析器LLM 或聊天模型的输出取决于解析器
检索器字符串文档列表
工具字符串或字典,取决于工具取决于工具

请参阅各个组件的文档,了解更多关于输入和输出类型以及如何使用它们的信息。

检查 Schema

注意

这是一个高级功能,对于大多数用户来说没有必要。除非您有特定的需求来检查 Runnable 的 schema,否则您应该跳过此部分。

在更高级的用例中,您可能希望通过编程方式检查 Runnable,并确定 Runnable 期望和产生的输入和输出类型。

Runnable 接口提供了获取 JSON Schema(Runnable 输入和输出类型的模式)以及输入和输出类型的 Pydantic schema 的方法。

这些 API 主要用于内部单元测试以及LangServe,它使用这些 API 进行输入验证和生成 OpenAPI 文档

除了输入和输出类型之外,一些 Runnable 还设置了额外的运行时配置选项。有相应的 API 来获取 Runnable 配置选项的 Pydantic Schema 和 JSON Schema。请参阅可配置的 Runnable 部分以获取更多信息。

方法描述
get_input_schema提供 Runnable 输入 schema 的 Pydantic Schema。
get_output_schema提供 Runnable 输出 schema 的 Pydantic Schema。
config_schema提供 Runnable 配置 schema 的 Pydantic Schema。
get_input_jsonschema提供 Runnable 输入 schema 的 JSONSchema。
get_output_jsonschema提供 Runnable 输出 schema 的 JSONSchema。
get_config_jsonschema提供 Runnable 配置 schema 的 JSONSchema。

With_types

LangChain 将根据可用信息自动尝试推断 Runnable 的输入和输出类型。

目前,这种推断对于使用LCEL组合构建的更复杂的 Runnable 效果不佳,推断的输入和/或输出类型可能不正确。在这些情况下,我们建议用户使用 with_types 方法(API 参考)覆盖推断的输入和输出类型。

RunnableConfig

任何用于执行 Runnable 的方法(例如,invokebatchstreamastream_events)都接受一个名为 RunnableConfigAPI 参考)的第二个参数。此参数是一个字典,包含将在 Runnable 执行期间在运行时使用的 Runnable 配置。

RunnableConfig 可以定义以下任何属性

属性描述
run_name用于给定 Runnable 的名称(不可继承)。
run_id此调用的唯一标识符。子调用将获得自己的唯一运行 ID。
tags此调用和任何子调用的标签。
metadata此调用和任何子调用的元数据。
callbacks此调用和任何子调用的回调。
max_concurrency最大并行调用数量(例如,由批量处理使用)。
recursion_limit调用可以递归的最大次数(例如,由返回 Runnable 的 Runnable 使用)
configurableRunnable 可配置属性的运行时值。

config 传递给 invoke 方法的方式如下

some_runnable.invoke(
some_input,
config={
'run_name': 'my_run',
'tags': ['tag1', 'tag2'],
'metadata': {'key': 'value'}

}
)

RunnableConfig 的传播

许多 Runnable 由其他 Runnable 组成,重要的是 RunnableConfig 要传播到 Runnable 进行的所有子调用。这允许向父 Runnable 提供运行时配置值,这些值将由所有子调用继承。

如果不是这样,将无法设置和传播回调或诸如 tagsmetadata 等其他配置值,而这些值是期望由所有子调用继承的。

创建新 Runnable 的两种主要模式是

  1. 使用LangChain 表达式语言 (LCEL) 声明式创建

    chain = prompt | chat_model | output_parser
  2. 使用自定义 Runnable(例如,RunnableLambda)或使用 @tool 装饰器

    def foo(input):
    # Note that .invoke() is used directly here
    return bar_runnable.invoke(input)
    foo_runnable = RunnableLambda(foo)

LangChain 将尝试自动为这两种模式传播 RunnableConfig

为了处理第二种模式,LangChain 依赖于 Python 的 contextvars

在 Python 3.11 及更高版本中,这开箱即用,您无需进行任何特殊操作即可将 RunnableConfig 传播到子调用。

在 Python 3.9 和 3.10 中,如果您正在使用异步代码,您需要在调用 Runnable 时手动将 RunnableConfig 传递给它。

这是由于 Python 3.9 和 3.10 中 asyncio 任务的一个限制,它们不接受 context 参数。

手动传播 RunnableConfig 的方式如下

async def foo(input, config): # <-- Note the config argument
return await bar_runnable.ainvoke(input, config=config)

foo_runnable = RunnableLambda(foo)
警告

当使用 Python 3.10 或更低版本并编写异步代码时,RunnableConfig 无法自动传播,您需要手动进行!这是尝试使用 astream_eventsastream_log 流式传输数据时的一个常见陷阱,因为这些方法依赖于 RunnableConfig 中定义回调的正确传播。

设置自定义运行名称、标签和元数据

RunnableConfig 字典的 run_nametagsmetadata 属性可用于为给定 Runnable 设置运行名称、标签和元数据的自定义值。

run_name 是一个字符串,可用于为运行设置自定义名称。此名称将用于日志和其他位置以识别运行。它不会被子调用继承。

tagsmetadata 属性分别是列表和字典,可用于为运行设置自定义标签和元数据。这些值将被子调用继承。

使用这些属性对于跟踪和调试运行非常有用,因为它们将作为可过滤和搜索的跟踪属性呈现在 LangSmith 中。

这些属性也将传播到回调,并将作为流中每个事件的一部分出现在诸如astream_events之类的流式 API 中。

设置运行 ID

注意

这是一个高级功能,对于大多数用户来说没有必要。

您可能需要为给定运行设置自定义 run_id,以备将来引用或与其他系统关联。

run_id 必须是有效的 UUID 字符串,并且每个运行都唯一。它用于识别父运行,子类将自动获得自己的唯一运行 ID。

要设置自定义 run_id,您可以在调用 Runnable 时将其作为键值对传递到 config 字典中

import uuid

run_id = uuid.uuid4()

some_runnable.invoke(
some_input,
config={
'run_id': run_id
}
)

# Do something with the run_id

设置递归限制

注意

这是一个高级功能,对于大多数用户来说没有必要。

某些 Runnable 可能会返回其他 Runnable,如果处理不当,可能导致无限递归。为了防止这种情况,您可以在 RunnableConfig 字典中设置 recursion_limit。这将限制 Runnable 可以递归的次数。

设置最大并发

如果使用 batchbatch_as_completed 方法,您可以在 RunnableConfig 字典中设置 max_concurrency 属性来控制最大并行调用数量。当您想限制并行调用数量以防止服务器或 API 过载时,这会很有用。

提示

如果您正在尝试限制聊天模型发出的请求数量,您可以使用内置的速率限制器,而不是设置 max_concurrency,这会更有效。

有关更多信息,请参阅如何处理速率限制指南。

设置可配置项

configurable 字段用于为 Runnable 的可配置属性传递运行时值。

它经常在LangGraphLangGraph 持久化内存一起使用。

它在RunnableWithMessageHistory中用于类似目的,以指定 session_id / conversation_id 来跟踪对话历史记录。

此外,您可以使用它来指定任何自定义配置选项,以便传递给他们创建的任何可配置的 Runnable

设置回调

使用此选项可在运行时为 Runnable 配置回调。回调将传递给 Runnable 进行的所有子调用。

some_runnable.invoke(
some_input,
{
"callbacks": [
SomeCallbackHandler(),
AnotherCallbackHandler(),
]
}
)

请阅读回调概念指南,了解更多关于如何在 LangChain 中使用回调的信息。

重要

如果您在异步环境中使用 Python 3.9 或 3.10,在某些情况下,您必须手动将 RunnableConfig 传播到子调用。请参阅RunnableConfig 的传播部分以获取更多信息。

从函数创建 Runnable

您可能需要创建一个运行任意逻辑的自定义 Runnable。如果使用LangChain 表达式语言 (LCEL) 来组合多个 Runnable,并且您需要在其中一个步骤中添加自定义处理逻辑,这将特别有用。

有两种方法可以从函数创建自定义 Runnable

  • RunnableLambda:适用于不需要流式传输的简单转换。
  • RunnableGenerator:当需要流式传输时,适用于更复杂的转换。

有关如何使用 RunnableLambdaRunnableGenerator 的更多信息,请参阅如何运行自定义函数指南。

重要

用户不应尝试通过继承 Runnable 来创建新的自定义 Runnable。这比简单地使用 RunnableLambdaRunnableGenerator 要复杂得多且更容易出错。

可配置的 Runnable

注意

这是一个高级功能,对于大多数用户来说没有必要。

它有助于配置使用LangChain 表达式语言 (LCEL) 创建的大型“链”,并被 LangServe 用于已部署的 Runnable。

有时您可能希望尝试,甚至向最终用户公开使用您的 Runnable 完成任务的多种不同方式。这可能涉及调整聊天模型中的温度等参数,甚至在不同的聊天模型之间切换。

为了简化此过程,Runnable 接口提供了两种在运行时创建可配置 Runnable 的方法

  • configurable_fields:此方法允许您配置 Runnable 中的特定属性。例如,聊天模型的 temperature 属性。
  • configurable_alternatives:此方法使您能够指定可在运行时运行的替代 Runnable。例如,您可以指定一个可以使用的不同聊天模型的列表。

有关如何配置运行时链内部的更多信息,请参阅如何配置运行时链内部指南。