Runnable 接口

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

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

runnable 接口概述

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

Invoked：将单个输入转换为输出。
Batched：多个输入被有效地转换为输出。
streamed（流式）：输出在生成时进行流式处理。
已检查：可以访问有关 Runnable 的输入、输出和配置的原理图信息。
组合：可以使用 LangChain 表达式语言（LCEL）组合多个 Runnable 以协同工作，以创建复杂的管道。

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

优化的并行执行（batch）

LangChain Runnables 提供了一个内置的batch（以及batch_as_completed） API 的调用，该 API 允许您并行处理多个输入。

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

两个批处理选项是：

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

的默认实现batch和batch_as_completed使用线程池 Executor 运行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，用户可能希望控制并行调用的最大数量。这可以通过设置max_concurrency属性中的RunnableConfig字典。有关更多信息，请参阅 RunnableConfig。

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

异步支持

Runnables 公开了一个异步 API，允许使用await语法。异步方法可以通过 “a” 前缀来标识（例如ainvoke,abatch,astream,abatch_as_completed).

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

流式处理 API

流式处理对于使基于 LLM 的应用程序能够响应最终用户至关重要。

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

sync stream 和 async astream：在生成输出时生成 Runnable。
异步astream_events：一个更高级的流式处理 API，允许流式处理中间步骤和最终输出
旧版异步astream_log：流式处理中间步骤和最终输出的旧版流式处理 API

有关如何在 LangChain 中进行流式传输的更多详细信息，请参阅 Streaming Conceptual Guide。

输入和输出类型

每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

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

检查 schema

注意

这是一项高级功能，对于大多数用户来说是必需的。您可能应该跳过本节，除非您有特定需要检查 Runnable 的架构。

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

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

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

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

方法	描述
`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 组合构建的更复杂的 Runnable，并且推断的输入和/或输出类型可能不正确。在这些情况下，我们建议用户使用with_typesmethod（API 参考）。

RunnableConfig

用于执行 runnable 的任何方法（例如invoke,batch,stream,astream_events）接受第二个名为RunnableConfig (API 参考）。此参数是一个字典，其中包含将使用的 Runnable 的配置在执行 Runnable 期间的运行时。

一个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到invokemethod 的实现方式如下：

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论点。

传播RunnableConfigmanual 的完成方式如下：

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的属性RunnableConfigdictionary 可用于为给定 Runnable 的运行名称、标签和元数据设置自定义值。

这run_name是一个字符串，可用于为运行设置自定义名称。此名称将在日志和其他地方用于标识运行。它不被 sub-calls 继承。

这tags和metadataattributes 分别是 lists 和 dictionaries，可用于为运行设置自定义标签和元数据。这些值由 sub-calls 继承。

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

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

设置运行 ID

注意

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

您可能需要设置自定义run_id对于给定的运行，如果需要稍后引用它或将其与其他系统相关联。

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

设置自定义run_id中，您可以将其作为键值对传递到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，如果处理不当，这可能会导致无限递归。为了防止这种情况，您可以设置recursion_limit在RunnableConfig字典。这将限制 Runnable 可以递归的次数。

设置最大并发数

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

提示

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

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

设置 configurable

这configurablefield 用于传递 Runnable 的可配置属性的运行时值。

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

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

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

设置回调

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

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

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

重要

如果您在异步环境中使用 Python 3.9 或 3.10，则必须传播这RunnableConfigmanually 到 sub 调用。有关更多信息，请参阅传播 RunnableConfig 部分。

从函数创建 Runnable

你可能需要创建一个运行任意逻辑的自定义 Runnable。这一点尤其如果使用 LangChain 表达式语言（LCEL）进行组合，则很有用多个 Runnables，您需要在其中一个步骤中添加自定义处理逻辑。

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

RunnableLambda：用于不需要流式处理的简单转换。
RunnableGenerator：当需要流式处理时，将此函数用于更复杂的转换。

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

重要

用户不应尝试将 Runnables 子类化来创建新的自定义 Runnable。是的比简单地使用RunnableLambda或RunnableGenerator.

可配置的 runnables

注意

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

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

有时你可能想尝试，甚至向最终用户展示使用 Runnable 的多种不同方式。这可能涉及调整聊天模型中的温度等参数，甚至在不同的聊天模型之间切换。

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

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

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

runnable 接口概述

优化的并行执行 （batch）

异步支持

流式处理 API

输入和输出类型

检查 schema

With_types

RunnableConfig

RunnableConfig 的传播

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

设置运行 ID

设置递归限制

设置最大并发数

设置 configurable

设置回调

从函数创建 Runnable

可配置的 runnables

优化的并行执行（batch）