当前位置：首页 > news >正文

20250913-01: Langchain概念：Runnable可运行接口

news 2025/9/14 17:58:32

20250913-01: Langchain概念：Runnable可运行接口

任务

🎯 学习目标

🔗 核心概念

可运行接口（Runnable）

Runnable 接口概述

优化的并行执行（批量）

异步支持

流式 API

输入和输出类型

检查 Schema

With_types

RunnableConfig

RunnableConfig 的传播
设置自定义运行名称、标签和元数据
设置运行 ID
设置递归限制
设置最大并发
设置可配置项
设置回调

从函数创建 Runnable

可配置的 Runnable

任务

阅读 LCEL 官方文档
理解 | 操作符的底层是 RunnableSequence

‍

🎯 学习目标

理解 LangChain 的“运行时引擎”——所有组件如何被统一调度和编排。

🔗 核心概念

可运行接口（Runnable）
LangChain 表达式语言（LCEL）
回调（Callbacks）
追踪（Tracing）
流式传输（Streaming）
异步编程（Async）

`可运行接口`（Runnable）

Runnable 接口 | 🦜️🔗 LangChain 框架

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

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

Runnable 接口概述

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

调用 invoked：将单个输入转换为输出。
批量处理 Batched：将多个输入高效地转换为输出。
流式传输 Streamed：输出在其生成时进行流式传输。
检查 Inspected：可以访问有关 Runnable 输入、输出和配置的示意信息。
组合 Composed：可以使用LangChain 表达式语言 (LCEL) 将多个 Runnable 组合在一起以创建复杂的管道。

请查看LCEL 速查表以了解涉及 Runnable 接口和 LCEL 表达式的一些常见模式。

优化的并行执行（批量）

LangChain Runnable 提供内置的 batch（和 batch_as_completed）API，允许您并行处理多个输入。

两种批量处理选项是

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

‍

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

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

‍

注意

当使用 batch 或 batch_as_completed 处理大量输入时，用户可能希望控制最大并行调用数量。这可以通过在 RunnableConfig 字典中设置 max_concurrency 属性来完成。有关更多信息，请参阅RunnableConfig。

聊天模型 (Chat Models) 还具有内置的速率限制器，可用于控制请求的速率。

‍

异步支持

Runnable 暴露了一个异步 API，允许它们在 Python 中使用 await 语法进行调用。异步方法可以通过“a”前缀来识别（例如，ainvoke、abatch、astream、abatch_as_completed）。

请参阅LangChain 异步编程指南以获取更多详细信息。

流式 API

流式传输对于使基于 LLM 的应用程序对最终用户响应迅速至关重要。

Runnable 暴露了以下三个流式 API

同步 stream 和异步 astream：在生成时产生 Runnable 的输出。
异步 astream_events：一个更高级的流式 API，允许流式传输中间步骤和最终输出
遗留异步 astream_log：一个遗留流式 API，用于流式传输中间步骤和最终输出

请参阅流式传输概念指南以获取有关如何在 LangChain 中进行流式传输的更多详细信息。

输入和输出类型

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

导致 Runnable 执行的 Runnable 方法（例如， invoke 、 batch 、 stream 、 astream_events ）都使用这些输入和输出类型。

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

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

组件	输入类型	输出类型
提示词	字典	PromptValue
聊天模型	字符串、聊天消息列表或 PromptValue	ChatMessage
LLM	字符串、聊天消息列表或 PromptValue	字符串
输出解析器	LLM 或聊天模型的输出	取决于解析器
检索器	字符串	文档列表
工具	字符串或字典，取决于工具	取决于工具

检查 Schema

注意

这是一个高级功能，对于大多数用户来说没有必要。除非您有特定的需求来检查 Runnable 的 schema，否则您应该跳过此部分。

Runnable 接口提供了获取 JSON Schema（Runnable 输入和输出类型的模式）以及输入和输出类型的 Pydantic schema 的方法。

方法	描述
`get_input_schema`	提供 Runnable 输入 schema 的 Pydantic Schema。
`get_output_schema`	提供 Runnable 输出 schema 的 Pydantic Schema。
`config_schema`	提供 Runnable 配置 schema 的 Pydantic Schema。
`get_input_jsonschema`	提供 Runnable 输入 schema 的 JSONSchema。
`get_output_jsonschema`	提供 Runnable 输出 schema 的 JSONSchema。
`get_config_jsonschema`	提供 Runnable 配置 schema 的 JSONSchema。

With_types

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

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

‍

RunnableConfig

任何用于执行 Runnable 的方法（例如，invoke、batch、stream、astream_events）都接受一个名为 RunnableConfig（API 参考）的第二个参数。

此参数是一个** 字典  ，包含将在 Runnable 执行期间在 运行时 使用的 Runnable 配置。**

RunnableConfig 可以定义以下任何属性

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

将 config 传递给 invoke 方法的方式如下

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

RunnableConfig 的传播

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

如果不是这样，将无法设置和传播回调或诸如 tags 和 metadata 等其他配置值，而这些值是期望由所有子调用继承的。

创建新 Runnable 的两种主要模式是

使用LangChain 表达式语言 (LCEL) 声明式创建

创建新 Runnable 的两种主要模式是使用LangChain 表达式语言 (LCEL) 声明式创建

使用自定义 Runnable（例如，RunnableLambda）或使用 @tool 装饰器

 def foo(input):# Note that .invoke() is used directly herereturn 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): # <-- Note the config argumentreturn 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 中。

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

如何使用 LangChain 进行追踪

设置运行 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}
)# Do something with the run_id

设置递归限制

某些 Runnable 可能会返回其他 Runnable，如果处理不当，可能导致** 无限递归 **。为了防止这种情况，您可以在 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) 来组合多个 Runnable，并且您需要在其中一个步骤中添加自定义处理逻辑，这将特别有用。

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

RunnableLambda：适用于 不需要流式传输 的简单转换。
RunnableGenerator：当需要 流式传输 时，适用于更复杂的转换。

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

‍

重要

用户不应尝试通过继承 Runnable 来创建新的自定义 Runnable。这比简单地使用 RunnableLambda 或 RunnableGenerator 要复杂得多且更容易出错·。

可配置的 Runnable

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

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

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

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

理解

问题：为什么需要“可配置的” Runnable？

现在问题来了。当你用 LCEL 创建好一个复杂的大链（比如一个客服机器人）后，你可能想：

进行实验（Try） ：在开发阶段，我想快速尝试用不同的模型（比如从 GPT-3.5 切换到 Claude）或者调整参数（比如把 temperature 从 0 调到 1，让回答更有创意），看看哪个效果更好。
面向用户（Expose） ：在产品阶段，我想让用户自己选择他们喜欢的模型或 creativity 级别，甚至为高级用户提供更强大的模型选项。

如果每次修改都要重新定义和创建整个链，会非常麻烦和低效。 “可配置的 Runnable” 就是为了解决这个问题而设计的。

“可配置的 Runnable” 是什么？

它是一种特殊的 Runnable，允许你在不重新构建整个链的情况下，动态地修改链中特定组件的配置。

它通过 RunnableConfig 参数来实现这一点。当你调用链时（如 chain.invoke(input, config=my_config)），你可以传入一个配置对象，这个配置对象可以告诉链：“嘿，这次运行，请把链里那个叫 ‘model’ 的组件换成我指定的这个，或者把它的温度参数调高。”

这就实现了链的“动态配置” 。

这段话可以这样理解：

“可配置的 Runnable” 是 LangChain 中的一个高级功能，它让你能轻松地动态调整那些由 LCEL 创建的复杂AI应用链（如客服机器人、摘要工具等）的内部设置（如切换模型、调整参数），而无需重写代码。这个功能极大地便利了开发阶段的实验和产品阶段的用户自定义。同时，它是 LangServe 部署工具能够提供灵活 API 的基础，使得同一个部署好的AI链可以根据不同请求的配置，表现出不同的行为。

简单来说，它就是让你已经搭好的“AI乐高”（链），不用拆开就能随时换掉里面的某个“积木”（组件）。

这两个功能都来自于 LangChain 的 Runnable 接口，它们允许你在运行时（Runtime） 动态地改变链（Chain）的行为，而无需重新定义或重新部署整个链。这在需要 A/B 测试、根据不同用户调整参数或快速故障转移等场景下非常强大。

1. 动态配置 (`configurable_fields`)

这个功能允许你为一个 Runnable 的特定字段（如 LLM 的 temperature）设置可在运行时覆盖的配置。

核心方法： .configurable_fields(<field_name>=ConfigurableField(...))

示例场景： 创建一个翻译链，但其严谨性（由 temperature 控制）可以在调用时动态决定。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import ConfigurableField
# 外部配置好的模型
from src.langchain_base import model_glm45,model_qwen3_8b# # 1. 定义一个基础的 LLM，并为其 temperature 字段设置为可配置
# 这里的 temperature=0.7 是一个默认值，将在运行时被覆盖
configurable_llm = model_glm45.configurable_fields(temperature=ConfigurableField(id="llm_temperature",# 这个配置的唯一 IDname="llm's temperature", # 可读的名称description="控制模型输出的随机性（0.0-1.0）"))# 描述信息# 2. 创建一个简单的提示模板
prompt = ChatPromptTemplate.from_messages([("system", "你是一个专业的翻译家。将用户输入的内容翻译成英文。"),("human", "{text}")
])# 3. 将提示模板和可配置的 LLM 组合成链
chain = prompt | configurable_llm# 4. 调用链时，通过 `with_config` 动态传入配置
# 示例 1: 使用高创造性翻译
creative_translation = chain.with_config(config={"llm_temperature": 0.9})for chunk in creative_translation.stream(({"text": "今天天气真好，阳光明媚。"})):print(chunk.content, end="", flush=True)# 示例 2: 使用高确定性翻译（更字面、更准确）
precise_translation = chain.with_config(configurable={"llm_temperature": 0.1}
).invoke({"text": "今天天气真好，阳光明媚。"})
print("高确定性翻译:", precise_translation.content)# 示例 3: 不提供配置，则使用默认的 temperature=0.7
default_translation = chain.invoke({"text": "今天天气真好，阳光明媚。"})
print("默认翻译:", default_translation.content)

关键点：

ConfigurableField 定义了哪个字段可以被配置以及如何描述它。
with_config(configurable={"field_id": value}) 是在调用时覆盖配置的方法。
你可以在同一个 Runnable 上为多个字段调用 .configurable_fields()。

2. 动态切换 (`configurable_alternatives`)

这个功能允许你在一组不同的 Runnable 之间进行运行时切换。这是实现模型故障转移、A/B 测试或根据输入路由到不同专家的完美工具。

核心方法： .configurable_alternatives(ConfigurableField(...), <key>=<runnable>)

示例场景： 创建一个链，可以根据需要在 OpenAI、Anthropic 和一个快速的本地模型之间切换。

# 1. 定义多个可选的 LLM
from src.langchain_base import model_glm45,model_qwen3_8b
# 2. 创建一个基础 LLM（通常是默认选择）
base_llm = model_glm45# 3. 使用 configurable_alternatives 将其变为一个“开关”
switchable_llm = base_llm.configurable_alternatives(# 定义配置字段ConfigurableField(id="llm_choice", name="llm_choice", description="llm_choice"),# 定义备选项键值对# 键：在运行时传入的字符串# 值：对应的 Runnableqwen3_8b=model_qwen3_8b,glm45=model_glm45# 如果传入了未定义的键，或者不传，则使用默认的 base_llm (即 openai_llm)
)# 4. 创建提示模板和链
prompt = ChatPromptTemplate.from_messages([("system", "你是一个乐于助人的助手。"),("human", "{question}")
])chain = prompt | switchable_llm
# 5. 在运行时动态切换模型！
# 使用 qwen3_8b
# response_openai = chain.with_config(
#     config={"llm_choice": "qwen3_8b"}
# ).invoke({"question": "量子计算的主要挑战是什么？"})
# print(f"qwen3_8b 回答: {response_openai.content}\n")# 使用 glm45
response_anthropic = chain.with_config(configurable={"llm_choice": "glm45"}
).invoke({"question": "量子计算的主要挑战是什么？"})
print(f"glm45 回答: {response_anthropic.content}\n")# 使用 glm45
response_anthropic = chain.invoke({"question": "量子计算的主要挑战是什么？"})
print(f"default 回答: {response_anthropic.content}\n")

更复杂的示例：切换整个子链

你不仅可以切换 LLM，还可以切换任何 Runnable，例如整个检索QA链。

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain# 假设我们有两个不同的检索器：一个用于科学文档，一个用于法律文档
science_retriever = ...
law_retriever = ...# 为每个检索器创建各自的 QA 子链
prompt = ChatPromptTemplate.from_template(...)
llm = ChatOpenAI()science_qa_chain = create_stuff_documents_chain(llm, prompt)
law_qa_chain = create_stuff_documents_chain(llm, prompt)science_chain = create_retrieval_chain(science_retriever, science_qa_chain)
law_chain = create_retrieval_chain(law_retriever, law_qa_chain)# 创建一个基础链（例如默认科学链）
base_chain = science_chain# 使其可切换
router_chain = base_chain.configurable_alternatives(ConfigurableField(id="retriever_domain",name="Retriever Domain",description="根据问题领域选择检索器",),science=science_chain,law=law_chain,
)# 现在，你可以根据用户问题的领域动态选择最合适的检索链！
science_answer = router_chain.with_config(configurable={"retriever_domain": "science"}
).invoke({"input": "解释一下光合作用。"})law_answer = router_chain.with_config(configurable={"retriever_domain": "law"}
).invoke({"input": "什么是著作权？"})

总结

功能	用途	核心方法
`configurable_fields`	微调一个 Runnable 的内部参数	`.configurable_fields(field=ConfigurableField(...))`
`configurable_alternatives`	在多个完整的 Runnable 之间切换	`.configurable_alternatives(ConfigurableField(...), key=runnable)`

复习知识点

Runnable 接口 - 核心概念

基础定义: Runnable 接口是 LangChain 组件的统一基础，被语言模型、输出解析器、检索器等广泛实现。
核心价值: 它提供了一种一致且可预测的方式与所有 LangChain 组件进行交互。

Runnable 接口概述 (核心能力)

核心方法: 定义了 invoke（单输入转输出）、batch（批量处理）、stream（流式传输）等标准方法。
关键能力: 允许组件被检查（获取输入/输出规范）和组合（通过 LCEL 创建复杂管道）。
优化的批量处理: batch 和 batch_as_completed API 提供内置的并行处理能力，显著提升多输入处理性能。
- 关键说明: 默认使用线程池，优化 I/O 密集型操作，但对 CPU 密集型任务因 GIL 限制效果不佳。
异步支持: 所有执行方法都有对应的异步版本（如 ainvoke, abatch），通过 a前缀标识。
流式传输 API: 提供 stream/astream（流式输出）和更高级的 astream_events（流式中间步骤和输出）。

输入和输出类型

类型化接口: 每个 Runnable 都由输入和输出类型表征，这些类型因组件而异（如提示词输入字典，聊天模型输出 ChatMessage）。
执行方法约束: invoke, batch, stream 等方法都严格使用这些预定义的输入和输出类型。

RunnableConfig (高级但关键)

配置载体: 一个字典，作为第二个参数传递给所有执行方法（如 invoke(input, config={})），用于传递运行时配置。
核心配置属性:
- callbacks: 设置回调处理器，用于追踪和日志。
- tags / metadata: 为运行设置标签和元数据，便于在 LangSmith 中过滤和搜索。
- run_name: 为运行设置自定义名称（不可继承）。
- max_concurrency: 控制 batch 等方法的最大并行调用数。
配置传播: RunnableConfig 会自动传播到所有子调用，这对于确保回调、标签等配置在整个链中生效至关重要。
Python 版本警告: 在 Python 3.9 和 3.10 的异步代码中，RunnableConfig 无法自动传播，必须手动传递（ainvoke(input, config=config)）。