可运行接口
Runnable 接口是操作 LangChain 组件的基础，许多组件都实现了该接口，例如 语言模型、输出解析器、检索器、编译后的 LangGraph 图 等。
本指南涵盖了 Runnable 接口的主要概念和方法，该接口允许开发者以一致且可预测的方式与各种 LangChain 组件进行交互。
- “可运行”接口 API 参考提供了可运行接口及其方法的详细概述。
- 一组内置的
Runnables可以在LangChain Core API 参考文档中找到。在使用LangChain 表达式语言 (LCEL)通过 LangChain 组合自定义“链”时，其中许多可运行对象非常有用。
可运行接口概览
可运行方式定义了一个标准接口，允许可运行组件能够：
- 已调用: 单个输入被转换为输出。
- 批处理: 多个输入被高效地转换为输出。
- 流式输出: 输出在生成时即进行流式传输。
- 检查：可以访问 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。
Chat Models 还内置了一个速率限制器，可用于控制请求的发出频率。
异步支持
Runnables 暴露了一个异步 API，允许它们在 Python 中使用 await 语法进行调用。异步方法可以通过前缀 "a" 来识别（例如：ainvoke, abatch, astream, abatch_as_completed）。
有关更多详细信息，请参阅 使用 LangChain 进行异步编程 指南。
流式API
流式处理对于使基于大语言模型的应用程序对最终用户具有响应性至关重要。
可运行对象暴露以下三种流式 API：
- sync 流式输出 和 async 异步流式输出：在生成时逐个产出 Runnable 的输出结果。
- 异步
astream_events：一种更高级的流式 API，支持流式传输中间步骤和最终输出 - The legacy async
astream_log: 一个传统的流式 API，用于流式传输中间步骤和最终输出
有关如何在 LangChain 中流式传输的更多详细信息，请参阅 流式处理概念指南。
输入和输出类型
每个 Runnable 都由输入和输出类型定义。这些输入和输出类型可以是任何 Python 对象，并由 Runnable 本身进行定义。
可运行方法会触发 Runnable 的执行（例如 invoke、batch、stream、astream_events），这些方法支持以下输入和输出类型。
- invoke: 接受一个输入并返回一个输出。
- batch: 接受一个输入列表并返回一个输出列表。
- 流式处理：接受一个输入并返回一个生成器，该生成器产生输出。
输入类型和输出类型因组件而异：
| 组件 | 输入类型 | 输出类型 |
|---|---|---|
| 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 输入和输出类型的 JSON Schema 的方法，以及用于输入和输出类型的 Pydantic schemas。
这些 API 主要用于内部单元测试，以及被 LangServe 使用，后者利用这些 API 进行输入验证和生成 OpenAPI 文档。
此外，除了输入和输出类型外，一些 Runnables 还配置了额外的运行时配置选项。 对于 Runnable 的配置选项，有相应的 API 可以获取其 Pydantic Schema 和 JSON Schema。 有关更多信息，请参阅 可配置的 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 组合构建的更复杂的 Runnables，此推理功能表现不佳，且推断出的输入和/或输出类型可能不正确。在这些情况下，我们建议用户使用 with_types 方法覆盖推断出的输入和输出类型（API 参考）。
RunnableConfig
任何用于执行可运行对象（runnable）的方法（例如，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 由其他可运行对象（Runnables）组成，重要的是将 RunnableConfig 传播给该可运行对象执行的所有子调用。这允许向父级可运行对象提供运行时配置值，这些值会被所有子调用继承。
如果不是这样，就不可能设置和传播 回调 或其他配置值，例如 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。
这是由于 asyncio 的任务 在 Python 3.9 和 3.10 中的限制，它们不接受 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，以便稍后引用它或将其与其他系统关联。
The 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）可能会返回其他可运行对象，如果处理不当可能导致无限递归。为防止这种情况，您可以在 RunnableConfig 字典中设置一个 recursion_limit。这将限制 Runnable 的递归次数。
设置最大并发
如果使用 batch 或 batch_as_completed 方法，您可以在 RunnableConfig 字典中设置 max_concurrency 属性以控制可发起的最大并行调用数。当您需要限制并行调用数量以防止服务器或 API 过载时，这非常有用。
设置可配置
configurable 字段用于传递可配置 Runnable 属性的运行时值。
它经常与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 Expression Language (LCEL) 来组合多个 Runnable，并且需要在其中一个步骤中添加自定义处理逻辑，这将特别有用。
有兩種方法可以從函數創建自定義 Runnable：
RunnableLambda: 用于不需要流式处理的简单转换。RunnableGenerator: 当需要流式传输时，使用此选项进行更复杂的转换。
有关如何使用RunnableLambda和RunnableGenerator的更多信息，请参阅如何运行自定义函数指南。
用户不应尝试子类化 Runnables 来创建新的自定义 Runnable。这比简单地使用 RunnableLambda 或 RunnableGenerator 要复杂得多且更容易出错。
可配置的可运行对象
这是大多数用户都不需要的一个高级功能。
它有助于配置使用 LangChain 表达式语言 (LCEL) 创建的大型“链”，并被 LangServe 用于已部署的 Runnable。
有时您可能希望尝试多种不同的方式来实现您的 Runnable，甚至向最终用户暴露这些方式。这可能涉及调整参数（例如聊天模型中的温度），或者在不同聊天模型之间进行切换。
为了简化这一过程，Runnable 接口提供了两种方法，用于在运行时创建可配置的 Runnables：
configurable_fields: 此方法允许您配置 Runnable 中的特定 属性。例如，聊天模型的temperature属性。configurable_alternatives: 此方法使您能够指定在运行时可以运行的替代可运行对象。例如，您可以指定一个不同的聊天模型列表以供使用。
有关如何配置运行时链内部结构的更多信息，请参阅如何配置运行时链内部结构指南。