LangChain 专题6 可运行接口

可运行接口（Runnable interface）是处理 LangChain 组件的基础，它在许多组件中实现，如语言模型、输出解析器、检索器、编译的 LangGraph 图等。

本文介绍了 Runnable 接口的主要概念和方法，它允许开发者在一致且可预测的方式下与 LangChain 组件交互。

可运行接口概览

Runnable 接口定义了一个标准接口，它允许 Runnable 组件能够被：

  • 调用：单个输入会被转化为输出。
  • 批处理：多个输入可以高效地转化为输出。
  • 流处理：输出随着产生的即时流处理。
  • 检查：可以访问关于 Runnable 输入、输出和配置的概况信息。
  • 组合：多个 Runnable 可以通过 LangChain 表达式语言（LangChain Expression Language，LCEL）组合在一起，以创建复杂的流水线。

优化并行执行（批处理）

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

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

两种批处理选项是：

  • batch：并行处理多个输入，并以与输入相同的顺序返回结果。
  • batch_as_completed：并行处理多个输入，当它们完成时返回结果。返回的结果可能按顺序排列，也可能不按顺序排列，但每个结果都包括匹配的输入索引。

batch 和 batch_as_completed 的默认实现使用线程池执行器来并行运行调用方法。 这允许高效的并行执行，无需用户管理线程，并且加快了 I/O 绑定的代码（如 API 请求、读取文件等）。 对于 CPU 绑定的操作，这不会那么有效，因为 Python 的 GIL（全局解释器锁）会阻止真正的并行执行。

一些 Runnable 接口可能提供自己的 batch 和 batch_as_completed 实现， 这些实现是为了特定用例而优化的（例如，依赖于模型提供商提供的 batch API）。

异步支持

Runnable 接口暴露一个异步 API，可通过 Python 中的 await 语法调用异步方法。 可以通过以 “a” 开头的名称来识别异步方法（例如：ainvoke、abatch、astream、abatch_as_completed）。

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

技巧：在使用 batch 或 batch_as_completed 处理大量输入时，用户可能希望控制并行调用的最大数量。 这可以通过在 RunnableConfig 字典中设置 max_concurrency 属性来实现。 参考：RunnableConfig

聊天模型还内置了一个速率限制器，可以用来控制请求的频率。 参考：Rate-limiting

流式 API

流是使基于 LLM 的应用对终端用户感觉实时的重要因素。

Runnable 接口暴露了以下三个流式 API：

  • 同步流和异步流：生成时输出一个可运行对象。
  • 异步 astream_events：一个更高级的流 API，允许流中间步骤和最终输出
  • 传统异步 astream_log：一个传统的流 API，流中间步骤和最终输出

输入输出类型

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

具有运行效果的方法（例如，invoke、batch、stream、astream）与这些输入和输出类型一起工作。

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

输入类型和输出类型因组件而异，以下是常用组件的输入输出类型列表：

Component Input Type Output Type
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

RunnableConfig

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

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

属性 描述
run_name 用于给定Runnable的名称（不继承）。
run_id 此呼叫的唯一标识符。子调用将获得自己的唯一run id。
tags 此调用和任何子调用的标签。
metadata 此调用和任何子调用元数据的元数据。
callbacks 此调用和任何子调用中的回调。
max_concurrency 最多可并行调用数量（例如，批量使用）。
recursion_limit 调用可回溯的最大次数（例如，批量调用使用）。
configurable Runnable可配置属性的运行时值。

将配置传递给 invoke 方法如下：

1
2
3
4
5
6
7
8
9

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

参考