Runnable 接口

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

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

Runnable接口概述​

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

• Invoked: 单个输入被转换为输出。
• Batched: 多个输入被高效地转换为输出。
• Streamed: 输出在生成时进行流式传输。
• 已检查：可以访问有关Runnable的输入、输出和配置的示意图信息。
• 组合：多个Runnables可以使用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）。

异步支持​

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执行的可运行方法（例如，invoke，batch，stream，astream_events）适用于这些输入和输出类型。

• invoke: 接受输入并返回输出。
• batch: 接受一个输入列表并返回一个输出列表。
• stream: 接受一个输入并返回一个生成器，该生成器产生输出。

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

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

检查模式​

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

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

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

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

带类型​

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

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

RunnableConfig​

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

一个 RunnableConfig 可以定义以下任何属性：

将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. Using a custom Runnable (e.g., RunnableLambda) or using the @tool decorator:

   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)

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

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过载时非常有用。

设置可配置​

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

它在LangGraph中经常与 LangGraph Persistence 和memory一起使用。

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

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

设置回调​

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

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

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

从函数创建可运行对象​

您可能需要创建一个运行任意逻辑的自定义Runnable。这在以下情况下特别有用：使用LangChain表达式语言（LCEL）来组合多个Runnable，并且您需要在其中一个步骤中添加自定义处理逻辑。

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

• RunnableLambda: 用于不需要流式处理的简单转换。
• RunnableGenerator: 当需要流式处理时，使用此工具进行更复杂的转换。

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

可配置的Runnables​

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

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

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

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