可运行接口

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

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

可运行接口概览

Runnable 方式定义了一个标准接口，该接口使得 Runnable 组件能够：

已调用: 输入被转换为输出。
批量处理: 多个输入被高效地转换为输出。
流式传输: 输出会随着生成而实时传输。
已检查：可以访问 Runnable 的输入、输出和配置的图示信息。
组合：多个可运行对象可以使用LangChain表达式语言（LCEL）组合在一起，以创建复杂的管道。

请查看 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）会阻止真正的并行执行。

某些 Runnables 可能会提供其自己的 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

流式传输在使基于大语言模型的应用对最终用户感觉响应迅速方面至关重要。

可运行对象公开以下三个流式传输API：

同步流和异步异步流：在生成时输出 Runnable 的结果。
异步 astream_events：一种更高级的流式 API，允许流式传输中间步骤和最终输出
astream_log 的遗留异步：一种遗留的流式 API，可流式传输中间步骤和最终输出

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

输入和输出类型

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

导致 Runnable 执行的可运行方法（例如 invoke、batch、stream、astream_events）会与这些输入和输出类型一起工作。

invoke：接受一个输入并返回一个输出。
batch：接受一个输入列表，并返回一个输出列表。
stream：接受一个输入，并返回一个生成器，该生成器会逐个输出结果。

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

组件	输入类型	输出类型
Prompt	dictionary	PromptValue
ChatModel	a string, list of chat messages or a PromptValue	ChatMessage
LLM	a string, list of chat messages or a PromptValue	String
OutputParser	the output of an LLM or ChatModel	Depends on the parser
Retriever	a string	List of Documents
Tool	a string or dictionary, depending on the tool	Depends on the tool

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

检查模式

笔记

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

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

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

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

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

方法	描述
`get_input_schema`	Gives the Pydantic Schema of the input schema for the Runnable.
`get_output_schema`	Gives the Pydantic Schema of the output schema for the Runnable.
`config_schema`	Gives the Pydantic Schema of the config schema for the Runnable.
`get_input_jsonschema`	Gives the JSONSchema of the input schema for the Runnable.
`get_output_jsonschema`	Gives the JSONSchema of the output schema for the Runnable.
`get_config_jsonschema`	Gives the JSONSchema of the config schema for the Runnable.

With_types

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

目前，这种推理对于使用 LCEL 组合构建的更复杂的 Runnables 无法很好地工作，推断的输入和/或输出类型可能不正确。在这些情况下，我们建议用户使用 with_types 方法覆盖推断的输入和输出类型 (API 参考)。

RunnableConfig

用于执行可运行对象的方法（例如 invoke、batch、stream、astream_events）都接受一个名为 RunnableConfig 的第二个参数（API 参考）。此参数是一个字典，其中包含在可运行对象执行期间运行时使用的配置。

RunnableConfig 可以具有以下任何属性：

属性	描述
run_name	Name used for the given Runnable (not inherited).
run_id	Unique identifier for this call. sub-calls will get their own unique run ids.
tags	Tags for this call and any sub-calls.
metadata	Metadata for this call and any sub-calls.
callbacks	Callbacks for this call and any sub-calls.
max_concurrency	Maximum number of parallel calls to make (e.g., used by batch).
recursion_limit	Maximum number of times a call can recurse (e.g., used by Runnables that return Runnables)
configurable	Runtime values for configurable attributes of the Runnable.

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

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

RunnableConfig的传播

许多 Runnables 由其他 Runnable 组成，重要的是 RunnableConfig 必须传递给 Runnable 所做的所有子调用。这允许为父 Runnable 提供运行时配置值，这些值将被所有子调用继承。

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

新的 Runnables 主要通过两种方式创建：

声明式使用 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 中，如果你使用 异步代码，在调用时需要手动将 RunnableConfig 传递给 Runnable。

这是由于 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 内定义的回调函数的正确传播。

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

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

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

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

使用这些属性有助于跟踪和调试运行，因为它们将在 LangSmith 中作为可筛选和搜索的追踪属性显示。

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

设置运行 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

设置递归限制

笔记

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

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

设置最大并发数

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

提示

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

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

设置可配置的

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

它经常与 LangGraph、 LangGraph 持久化和存储一起使用。

它在 RunnableWithMessageHistory 中用于类似的目的，以指定保留对话历史记录的 session_id / conversation_id。

此外，您还可以使用它来指定任何自定义配置选项，传递给它们创建的任何可配置运行对象。

设置回调

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

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

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

重要

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

从函数创建可运行的实例

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

有两种方式可以从函数创建自定义的 Runnable：

RunnableLambda: 用于不需要流式处理的简单转换。
RunnableGenerator: 在需要流式传输时，用于更复杂的转换。

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

重要

用户不应尝试通过继承 Runnables 来创建新的自定义 Runnable。这比直接使用 RunnableLambda 或 RunnableGenerator 要复杂得多，且更容易出错。

可配置的可运行对象

笔记

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

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

有时，你可能希望尝试多种不同的方法来使用你的 Runnable，甚至向最终用户暴露这些选项。这可能涉及调整参数，例如聊天模型中的温度值，甚至在不同的聊天模型之间切换。

为了简化这一过程，Runnable 接口提供了两种方法，用于在运行时创建可配置的 Runnable。

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

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

可运行接口概览​

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

异步支持​

流式API​

输入和输出类型​

检查模式​

With_types​

RunnableConfig​

RunnableConfig的传播​

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

设置运行 ID​

设置递归限制​

设置最大并发数​

设置可配置的​

设置回调​

从函数创建可运行的实例​

可配置的可运行对象​

可运行接口概览

优化的并行执行（批量）

异步支持

流式API

输入和输出类型

检查模式

With_types

RunnableConfig

RunnableConfig的传播

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

设置运行 ID

设置递归限制

设置最大并发数

设置可配置的

设置回调

从函数创建可运行的实例

可配置的可运行对象