可运行接口

可运行接口是使用 LangChain 组件的基础，它在许多组件中实现，例如 语言模型、输出解析器、检索器、已编译的 LangGraph 图 等。

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

相关资源

• “可运行”接口 API 参考 提供了可运行接口及其方法的详细概述。
• 内置 Runnables 的列表可以在 LangChain Core API 参考中找到。当使用 LangChain 表达式语言 (LCEL) 在 LangChain 中组合自定义“链”时，这些 Runnables 中的许多都很有用。

可运行接口概述

可运行方式定义了一个标准接口，允许可运行组件被

• 调用：单个输入转换为输出。
• 批处理：多个输入有效地转换为输出。
• 流式传输：输出在生成时流式传输。
• 检查：可以访问有关可运行对象的输入、输出和配置的示意性信息。
• 组合：可以使用LangChain 表达式语言 (LCEL)组合多个 Runnable 以协同工作，从而创建复杂的管道。

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

优化的并行执行（批量）

LangChain Runnables 提供了一个内置的 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。

聊天模型还具有一个内置的速率限制器，可用于控制发出请求的速率。

异步支持

Runnables 公开了异步 API，允许在 Python 中使用 await 语法进行调用。异步方法可以通过“a”前缀来识别（例如，ainvoke、abatch、astream、abatch_as_completed）。

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

流式 API

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

Runnables 公开了以下三个流式 API：

1. 同步 stream 和异步 astream：在生成时产生 Runnable 的输出。
2. 异步 astream_events：一种更高级的流式 API，允许流式传输中间步骤和最终输出。
3. 旧版异步 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	字符串	文档列表
工具	字符串或字典，取决于工具	取决于工具

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

检查模式

注意

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

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

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

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

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

方法	描述
get_input_schema	提供 Runnable 输入模式的 Pydantic 模式。
get_output_chema	提供 Runnable 输出模式的 Pydantic 模式。
config_schema	提供 Runnable 配置模式的 Pydantic 模式。
get_input_jsonschema	提供 Runnable 输入模式的 JSONSchema。
get_output_jsonschema	提供 Runnable 输出模式的 JSONSchema。
get_config_jsonschema	提供 Runnable 配置模式的 JSONSchema。

With_types

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

目前，这种推断对于使用 LCEL 组合构建的更复杂的 Runnables 效果不佳，并且推断的输入和/或输出类型可能不正确。在这些情况下，我们建议用户使用 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	要进行的最大并行调用数（例如，由 batch 使用）。
recursion_limit	调用可以递归的最大次数（例如，由返回 Runnables 的 Runnables 使用）。
configurable	Runnable 的可配置属性的运行时值。

将 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 主要有两种模式：

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_events 和 astream_log 流式传输数据时，这是一个常见的陷阱，因为这些方法依赖于正确传播在 RunnableConfig 内部定义的回调。

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

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

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

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

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

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

相关

• 如何使用 LangChain 进行跟踪

设置运行 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 过载时，这非常有用。

提示

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

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

设置可配置项

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

它在 LangGraph 中与 LangGraph Persistence 和 内存一起频繁使用。

它在 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 的更多信息，请参阅 如何运行自定义函数 指南。

重要提示

用户不应尝试子类化 Runnables 来创建新的自定义 Runnable。 它比简单地使用 RunnableLambda 或 RunnableGenerator 更复杂且更容易出错。

可配置 runnable

注意

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

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

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

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

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

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