langchain runnables 概念指南

LangChain Runnables 概念指南

概述

Runnable接口是LangChain中的核心抽象，它为大多数组件提供了统一的调用方式。这个接口使得创建自定义链变得简单，并且以标准方式调用它们。标准接口包括：

• stream: 流式返回响应块
• invoke: 对输入调用链
• batch: 对输入列表调用链
• astream: 异步流式返回响应块
• ainvoke: 异步调用链
• abatch: 异步对输入列表调用链
• astream_log: 流式返回中间步骤，随着它们的发生
• astream_events: 流式返回事件，随着它们在链中发生（在v0.1.14中引入）

Runnable接口概述

优化的并行执行（批处理）

Runnable接口支持批处理操作，可以并行处理多个输入，这比逐个处理输入要高效得多。

异步支持

所有Runnable都支持同步和异步调用。这使得在异步环境中使用相同的代码变得容易，例如在Jupyter笔记本中进行原型设计，然后在异步服务器中使用。

流式API

Runnable接口提供了多种流式API：

• stream: 流式返回最终输出的块
• astream: 异步版本的stream
• astream_log: 流式返回中间步骤和最终输出
• astream_events: 流式返回链中发生的事件

这些流式API对于创建响应式用户界面特别有用，可以在处理过程中向用户显示进度。

输入和输出类型

每个Runnable都公开了输入和输出模式，提供了关于输入和输出结构的信息。

检查模式

您可以使用以下方法检查Runnable的输入和输出模式：

# 检查输入模式
runnable.input_schema# 检查输出模式  
runnable.output_schema

RunnableConfig

RunnableConfig是一个配置对象，可以传递给Runnable以控制其执行方式。它包含以下可选字段：

• tags: 用于过滤和搜索的标签列表
• metadata: 与运行相关的元数据字典
• callbacks: 要使用的回调处理程序列表
• run_name: 运行的自定义名称
• max_concurrency: 批处理操作的最大并发数
• recursion_limit: 防止无限递归的递归限制
• configurable: 可配置属性的运行时值
• run_id: 运行的唯一标识符

RunnableConfig的传播

许多Runnables由其他Runnables组成，重要的是RunnableConfig要传播到Runnable进行的所有子调用。这允许向父Runnable提供运行时配置值，这些值由所有子调用继承。

如果不是这种情况，就不可能设置和传播回调或其他配置值，如tags和metadata，这些值预期由所有子调用继承。

创建新Runnables有两种主要模式：

1. 使用LangChain表达式语言（LCEL）声明式创建：

chain = prompt | chat_model | output_parser

2. 使用自定义Runnable或@tool装饰器：

def foo(input):# 注意这里直接使用.invoke()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): # <-- 注意config参数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）中作为流中每个事件的一部分出现。

设置运行ID

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

您可能需要为给定的运行设置自定义run_id，以防您想稍后引用它或将其与其他系统关联。

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

要设置自定义run_id，您可以在调用Runnable时将其作为键值对传递到config字典中：

import uuidrun_id = uuid.uuid4()some_runnable.invoke(some_input, config={'run_id': run_id}
)# 对run_id做一些操作

设置递归限制

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

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

设置最大并发数

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

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

设置可配置项

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

它在LangGraph中与LangGraph持久化和内存一起频繁使用。

它在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）来组合多个Runnables，并且您需要在其中一个步骤中添加自定义处理逻辑，这特别有用。

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

• RunnableLambda：用于不需要流式处理的简单转换。
• RunnableGenerator：用于需要流式处理的更复杂转换。

⚠️ 重要：用户不应尝试子类化Runnables来创建新的自定义Runnable。这比简单地使用RunnableLambda或RunnableGenerator要复杂得多且容易出错。

可配置的Runnables

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

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

有时您可能想要实验，甚至向最终用户公开，使用Runnable做事情的多种不同方式。这可能涉及调整参数，如聊天模型中的温度，甚至在不同的聊天模型之间切换。

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

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

总结

Runnable接口是LangChain的核心抽象，它提供了：

1. 统一的调用方式：所有组件都遵循相同的接口
2. 流式支持：实时响应和进度显示
3. 异步支持：适用于现代异步应用
4. 批处理优化：高效的并行处理
5. 灵活配置：运行时配置和自定义
6. 可组合性：通过LCEL轻松组合复杂链

这个接口使得LangChain应用的开发更加标准化和高效，无论是简单的单步操作还是复杂的多步骤链。