Runnable 接口

Runnable 接口是使用 LangChain 组件的基础，并且被许多组件实现，例如语言模型、输出解析器、检索器、已编译的 LangGraph 图等等。

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

Runnable 接口概述

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

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

请查阅 LCEL Cheatsheet 来了解涉及 Runnable 接口和 LCEL 表达式的一些常见模式。

优化的并行执行（批量）

LangChain Runnables 提供内置的 batch（和 batch_as_completed）API，允许您并行处理多个输入。

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

两种批量处理选项是：

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

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

某些 Runnables 可能会提供自己的 batch 和 batch_as_completed 实现，这些实现已针对其特定用例进行了优化（例如，依赖模型提供商提供的 batch API）。

note

异步版本的 abatch 和 abatch_as_completed 依赖于 asyncio 的 gather 和 as_completed 函数来并行运行 ainvoke 方法。

tip

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

聊天模型还内置了速率限制器，可用于控制请求的速率。

异步支持

Runnables 暴露了一个异步 API，允许使用 Python 中的 await 语法进行调用。异步方法可以通过 "a" 前缀进行识别（例如 ainvoke、abatch、astream、abatch_as_completed）。

有关更多详细信息，请参阅 LangChain 的异步编程指南。

流式传输 API

流式传输对于使基于 LLM 的应用程序在用户感觉上响应迅速至关重要。

Runnables 暴露了以下三个流式传输 API：

同步 stream 和异步 astream：在生成时产生 Runnable 的输出。
异步 astream_events：一个更高级的流式传输 API，允许流式传输中间步骤和最终输出。
旧版本的异步 astream_log：一个旧版流式传输 API，用于流式传输中间步骤和最终输出。

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

输入和输出类型

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

导致执行 Runnable 的 Runnable 方法（例如 invoke、batch、stream、astream_events）使用这些输入和输出类型。

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

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

组件	输入类型	输出类型
Prompt	字典	PromptValue
ChatModel	字符串、聊天消息列表或 PromptValue	ChatMessage
LLM	字符串、聊天消息列表或 PromptValue	字符串
OutputParser	LLM 或 ChatModel 的输出	取决于解析器
Retriever	字符串	文档列表
Tool	字符串或字典，取决于工具	取决于工具

有关输入和输出类型及其使用方法的更多信息，请参阅各个组件的文档。

自检模式

note

这是一个高级功能，对大多数用户来说是不必要的。您可能应该跳过此部分，除非您有特定的需要来检查 Runnable 的模式。

在更高级的用例中，您可能希望以编程方式自检 Runnable，并确定 Runnable 预期的输入和输出类型，以及它生成的输入和输出类型。

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

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

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

方法	描述
`get_input_schema`	提供 Runnable 的输入模式的 Pydantic Schema。
`get_output_schema`	提供 Runnable 的输出模式的 Pydantic Schema。
`config_schema`	提供 Runnable 的配置模式的 Pydantic Schema。
`get_input_jsonschema`	提供 Runnable 的输入模式的 JSONSchema。
`get_output_jsonschema`	提供 Runnable 的输出模式的 JSONSchema。
`get_config_jsonschema`	提供 Runnable 的配置模式的 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	此调用的唯一标识符。子调用将获得自己的唯一 run id。
tags	此调用及任何子调用的标签。
metadata	此调用及任何子调用的元数据。
callbacks	此调用及任何子调用的回调。
max_concurrency	要发出的并行调用最大数量（例如，由 batch 使用）。
recursion_limit	调用可以递归的最大次数（例如，由返回 Runnables 的 Runnables 使用）
configurable	可配置属性的运行时值。

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

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

   }
)

RunnableConfig 的传播

许多 Runnables 由其他 Runnables 组成，并且 RunnableConfig 能够传播到 Runnable 所做的所有子调用非常重要。这允许为父 Runnable 提供运行时配置值，这些值将被所有子调用继承。

如果不是这样，将无法设置和传播回调或其他配置值，如 tags 和 metadata，这些值是所有子调用所期望继承的。

创建新 Runnables 的模式主要有两种：

使用 LangChain 表达式语言 (LCEL) 声明式创建：
```
chain = prompt | chat_model | output_parser
```

使用自定义 Runnable（例如 RunnableLambda）或使用 @tool 装饰器：

def foo(input):
    # 注意这里直接使用了 .invoke()
    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): # <-- 注意 config 参数
    return await bar_runnable.ainvoke(input, config=config)

foo_runnable = RunnableLambda(foo)

caution

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

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

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

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

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

使用这些属性对于跟踪和调试运行很有用，因为它们将在 LangSmith 中作为跟踪属性显示，您可以对其进行过滤和搜索。

这些属性也将传播到回调，并且将作为流中每个事件的一部分出现在 astream_events 等流式传输 API 中。

设置运行 ID

note

这是一个高级功能，对大多数用户来说是不必要的。

您可能需要为给定的运行设置自定义 run_id，以防您想稍后引用它或将其与其他系统关联。

run_id 必须是有效的 UUID 字符串，并且对于每次运行都唯一。它用于标识父运行，子类将自动获得自己的唯一 run id。

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

import uuid

run_id = uuid.uuid4()

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

# 使用 run_id 进行某些操作

设置递归限制

note

这是一个高级功能，对大多数用户来说是不必要的。

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

设置最大并发数

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

tip

如果您尝试限制聊天模型发出的请求数量，您可以使用内置的速率限制器来代替设置 max_concurrency，这样会更有效。

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

设置可配置项

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

它经常在 LangGraph 中与LangGraph 持久化和内存一起使用。

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

此外，您还可以使用它为任何可配置 Runnable 指定任何自定义配置选项。

设置回调

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

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

有关如何在 LangChain 中使用回调的更多信息，请阅读回调概念指南。

important

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

从函数创建 Runnable

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

有两种从函数创建自定义 Runnable 的方法：

RunnableLambda：适用于不需要流式传输的简单转换。
RunnableGenerator：适用于需要流式传输的更复杂的转换。

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

important

用户不应尝试子类化 Runnables 来创建新的自定义 Runnable。这比仅使用 RunnableLambda 或 RunnableGenerator 要复杂且容易出错得多。

可配置的 Runnables

note

这是一个高级功能，对大多数用户来说是不必要的。

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

有时，您可能希望尝试，甚至向最终用户公开多种不同的方法来处理您的 Runnable。这可能涉及调整聊天模型的温度等参数，甚至在不同的聊天模型之间进行切换。

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

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

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

Runnable 接口概述​

优化的并行执行（批量）​

异步支持​

流式传输 API​

输入和输出类型​

自检模式​

With_types​

RunnableConfig​

RunnableConfig 的传播​

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

设置运行 ID​

设置递归限制​

设置最大并发数​

设置可配置项​

设置回调​

从函数创建 Runnable​

可配置的 Runnables​

Runnable 接口概述

优化的并行执行（批量）

异步支持

流式传输 API

输入和输出类型

自检模式

With_types

RunnableConfig

RunnableConfig 的传播

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

设置运行 ID

设置递归限制

设置最大并发数

设置可配置项

设置回调

从函数创建 Runnable

可配置的 Runnables