可运行接口
Runnable 接口是使用 LangChain 组件的基础,它在许多组件中都有实现,例如语言模型、输出解析器、检索器、编译后的 LangGraph 图等等。
本指南涵盖了 Runnable 接口的主要概念和方法,它允许开发者以一致且可预测的方式与各种 LangChain 组件进行交互。
- 《"Runnable" 接口 API 参考》提供了 Runnable 接口及其方法的详细概述。
- 内置的
Runnables
列表可以在LangChain 核心 API 参考中找到。这些 Runnable 中的许多在您使用LangChain 表达式语言 (LCEL) 在 LangChain 中组合自定义“链”时非常有用。
Runnable 接口概述
Runnable 方式定义了一个标准接口,允许 Runnable 组件
- 调用:将单个输入转换为输出。
- 批量处理:将多个输入高效地转换为输出。
- 流式传输:输出在其生成时进行流式传输。
- 检查:可以访问有关 Runnable 输入、输出和配置的示意信息。
- 组合:可以使用LangChain 表达式语言 (LCEL) 将多个 Runnable 组合在一起以创建复杂的管道。
请查看LCEL 速查表以了解涉及 Runnable 接口和 LCEL 表达式的一些常见模式。
优化的并行执行(批量)
LangChain Runnable 提供内置的 batch
(和 batch_as_completed
)API,允许您并行处理多个输入。
当需要处理多个独立输入时,使用这些方法可以显著提高性能,因为处理可以并行而不是顺序进行。
两种批量处理选项是
batch
:并行处理多个输入,并按与输入相同的顺序返回结果。batch_as_completed
:并行处理多个输入,并在完成时返回结果。结果可能无序到达,但每个结果都包含输入索引以进行匹配。
batch
和 batch_as_completed
的默认实现使用线程池执行器来并行运行 invoke
方法。这允许高效的并行执行,而无需用户管理线程,并加速 I/O 密集型代码(例如,发出 API 请求、读取文件等)。对于 CPU 密集型操作,它不会那么有效,因为 Python 中的 GIL(全局解释器锁)会阻止真正的并行执行。
某些 Runnable 可能会提供其自己的 batch
和 batch_as_completed
实现,这些实现针对其特定用例进行了优化(例如,依赖于模型提供商提供的 batch
API)。
abatch
和 abatch_as_completed
的异步版本依赖于 asyncio 的gather 和 as_completed 函数来并行运行 ainvoke
方法。
当使用 batch
或 batch_as_completed
处理大量输入时,用户可能希望控制最大并行调用数量。这可以通过在 RunnableConfig
字典中设置 max_concurrency
属性来完成。有关更多信息,请参阅RunnableConfig。
聊天模型 (Chat Models) 还具有内置的速率限制器,可用于控制请求的速率。
异步支持
Runnable 暴露了一个异步 API,允许它们在 Python 中使用 await
语法进行调用。异步方法可以通过“a”前缀来识别(例如,ainvoke
、abatch
、astream
、abatch_as_completed
)。
请参阅LangChain 异步编程指南以获取更多详细信息。
流式 API
流式传输对于使基于 LLM 的应用程序对最终用户响应迅速至关重要。
Runnable 暴露了以下三个流式 API
- 同步 stream 和异步 astream:在生成时产生 Runnable 的输出。
- 异步
astream_events
:一个更高级的流式 API,允许流式传输中间步骤和最终输出 - 遗留异步
astream_log
:一个遗留流式 API,用于流式传输中间步骤和最终输出
请参阅流式传输概念指南以获取有关如何在 LangChain 中进行流式传输的更多详细信息。
输入和输出类型
每个 Runnable
都由一个输入类型和一个输出类型来表征。这些输入和输出类型可以是任何 Python 对象,并由 Runnable 本身定义。
导致 Runnable 执行的 Runnable 方法(例如,invoke
、batch
、stream
、astream_events
)都使用这些输入和输出类型。
- invoke:接受一个输入并返回一个输出。
- batch:接受一个输入列表并返回一个输出列表。
- stream:接受一个输入并返回一个产生输出的生成器。
输入类型和输出类型因组件而异
组件 | 输入类型 | 输出类型 |
---|---|---|
提示词 | 字典 | PromptValue |
聊天模型 | 字符串、聊天消息列表或 PromptValue | ChatMessage |
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 的方法(例如,invoke
、batch
、stream
、astream_events
)都接受一个名为 RunnableConfig
(API 参考)的第二个参数。此参数是一个字典,包含将在 Runnable 执行期间在运行时使用的 Runnable 配置。
RunnableConfig
可以定义以下任何属性
属性 | 描述 |
---|---|
run_name | 用于给定 Runnable 的名称(不可继承)。 |
run_id | 此调用的唯一标识符。子调用将获得自己的唯一运行 ID。 |
tags | 此调用和任何子调用的标签。 |
metadata | 此调用和任何子调用的元数据。 |
callbacks | 此调用和任何子调用的回调。 |
max_concurrency | 最大并行调用数量(例如,由批量处理使用)。 |
recursion_limit | 调用可以递归的最大次数(例如,由返回 Runnable 的 Runnable 使用) |
configurable | Runnable 可配置属性的运行时值。 |
将 config
传递给 invoke
方法的方式如下
some_runnable.invoke(
some_input,
config={
'run_name': 'my_run',
'tags': ['tag1', 'tag2'],
'metadata': {'key': 'value'}
}
)
RunnableConfig 的传播
许多 Runnable
由其他 Runnable 组成,重要的是 RunnableConfig
要传播到 Runnable 进行的所有子调用。这允许向父 Runnable 提供运行时配置值,这些值将由所有子调用继承。
如果不是这样,将无法设置和传播回调或诸如 tags
和 metadata
等其他配置值,而这些值是期望由所有子调用继承的。
创建新 Runnable
的两种主要模式是
-
使用LangChain 表达式语言 (LCEL) 声明式创建
chain = prompt | chat_model | output_parser
-
使用自定义 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_events
和 astream_log
流式传输数据时的一个常见陷阱,因为这些方法依赖于 RunnableConfig
中定义回调的正确传播。
设置自定义运行名称、标签和元数据
RunnableConfig
字典的 run_name
、tags
和 metadata
属性可用于为给定 Runnable 设置运行名称、标签和元数据的自定义值。
run_name
是一个字符串,可用于为运行设置自定义名称。此名称将用于日志和其他位置以识别运行。它不会被子调用继承。
tags
和 metadata
属性分别是列表和字典,可用于为运行设置自定义标签和元数据。这些值将被子调用继承。
使用这些属性对于跟踪和调试运行非常有用,因为它们将作为可过滤和搜索的跟踪属性呈现在 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 可以递归的次数。
设置最大并发
如果使用 batch
或 batch_as_completed
方法,您可以在 RunnableConfig
字典中设置 max_concurrency
属性来控制最大并行调用数量。当您想限制并行调用数量以防止服务器或 API 过载时,这会很有用。
设置可配置项
configurable
字段用于为 Runnable 的可配置属性传递运行时值。
它经常在LangGraph与LangGraph 持久化和内存一起使用。
它在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
:当需要流式传输时,适用于更复杂的转换。
有关如何使用 RunnableLambda
和 RunnableGenerator
的更多信息,请参阅如何运行自定义函数指南。
用户不应尝试通过继承 Runnable 来创建新的自定义 Runnable。这比简单地使用 RunnableLambda
或 RunnableGenerator
要复杂得多且更容易出错。
可配置的 Runnable
这是一个高级功能,对于大多数用户来说没有必要。
它有助于配置使用LangChain 表达式语言 (LCEL) 创建的大型“链”,并被 LangServe 用于已部署的 Runnable。
有时您可能希望尝试,甚至向最终用户公开使用您的 Runnable 完成任务的多种不同方式。这可能涉及调整聊天模型中的温度等参数,甚至在不同的聊天模型之间切换。
为了简化此过程,Runnable 接口提供了两种在运行时创建可配置 Runnable 的方法
configurable_fields
:此方法允许您配置 Runnable 中的特定属性。例如,聊天模型的temperature
属性。configurable_alternatives
:此方法使您能够指定可在运行时运行的替代 Runnable。例如,您可以指定一个可以使用的不同聊天模型的列表。
有关如何配置运行时链内部的更多信息,请参阅如何配置运行时链内部指南。