语言模型

langchain_core.language_models

语言模型。

LangChain 有两个主要的类来处理语言模型：聊天模型和“老式”LLM。

聊天模型

将消息序列作为输入并返回聊天消息作为输出（而不是使用纯文本）的语言模型。

聊天模型支持为对话消息分配不同的角色，有助于区分来自 AI、用户以及系统消息等指令的消息。

聊天模型的关键抽象是 BaseChatModel。实现应继承自该类。

查看现有的聊天模型集成。

LLM

接收字符串作为输入并返回字符串的语言模型。这些通常是较旧的模型（较新的模型通常是聊天模型）。

尽管底层的模型是字符串输入、字符串输出，但 LangChain 包装器也允许这些模型接收消息作为输入。这使得它们拥有与聊天模型相同的接口。当消息作为输入传递时，它们会在内部被格式化为字符串，然后再传递给底层模型。

BaseChatModel

基类：BaseLanguageModel[AIMessage], ABC

              flowchart TD
              langchain_core.language_models.BaseChatModel[BaseChatModel]
              langchain_core.language_models.base.BaseLanguageModel[BaseLanguageModel]
              langchain_core.runnables.base.RunnableSerializable[RunnableSerializable]
              langchain_core.load.serializable.Serializable[Serializable]
              langchain_core.runnables.base.Runnable[Runnable]

                              langchain_core.language_models.base.BaseLanguageModel --> langchain_core.language_models.BaseChatModel
                                langchain_core.runnables.base.RunnableSerializable --> langchain_core.language_models.base.BaseLanguageModel
                                langchain_core.load.serializable.Serializable --> langchain_core.runnables.base.RunnableSerializable
                
                langchain_core.runnables.base.Runnable --> langchain_core.runnables.base.RunnableSerializable
                




              click langchain_core.language_models.BaseChatModel href "" "langchain_core.language_models.BaseChatModel"
              click langchain_core.language_models.base.BaseLanguageModel href "" "langchain_core.language_models.base.BaseLanguageModel"
              click langchain_core.runnables.base.RunnableSerializable href "" "langchain_core.runnables.base.RunnableSerializable"
              click langchain_core.load.serializable.Serializable href "" "langchain_core.load.serializable.Serializable"
              click langchain_core.runnables.base.Runnable href "" "langchain_core.runnables.base.Runnable"
            

聊天模型的基类。

关键的命令式方法

实际调用底层模型的方法。

此表简要概述了主要的命令式方法。请参阅基本的 Runnable 参考文档以获取完整文档。

方法	输入	输出	描述
invoke	str | list[dict | tuple | BaseMessage] | PromptValue	BaseMessage	单次聊天模型调用。
ainvoke	'''	BaseMessage	默认在异步执行器中运行 invoke。
stream	'''	Iterator[BaseMessageChunk]	默认生成 invoke 的输出。
astream	'''	AsyncIterator[BaseMessageChunk]	默认生成 ainvoke 的输出。
astream_events	'''	AsyncIterator[StreamEvent]	事件类型：on_chat_model_start、on_chat_model_stream、on_chat_model_end。
batch	list[''']	list[BaseMessage]	默认在并发线程中运行 invoke。
abatch	list[''']	list[BaseMessage]	默认在并发线程中运行 ainvoke。
batch_as_completed	list[''']	Iterator[tuple[int, Union[BaseMessage, Exception]]]	默认在并发线程中运行 invoke。
abatch_as_completed	list[''']	AsyncIterator[tuple[int, Union[BaseMessage, Exception]]]	默认在并发线程中运行 ainvoke。

关键的声明式方法

使用聊天模型创建另一个 Runnable 的方法。

此表简要概述了主要的声明式方法。请参阅每种方法的参考文档以获取完整文档。

方法	描述
bind_tools	创建可以调用工具的聊天模型。
with_structured_output	创建使用模式来结构化模型输出的包装器。
with_retry	创建在失败时重试模型调用的包装器。
with_fallbacks	创建在失败时回退到其他模型的包装器。
configurable_fields	指定可以在运行时通过 RunnableConfig 配置的模型的初始化参数。
configurable_alternatives	指定可以在运行时通过 RunnableConfig 替换的备用模型。

创建自定义聊天模型

自定义聊天模型的实现应继承自该类。请参考下表，了解实现中哪些方法和属性是必需的或可选的。

方法/属性	描述	必需
_generate	用于从提示生成聊天结果	必需
_llm_type（属性）	用于唯一标识模型类型。用于日志记录。	必需
_identifying_params（属性）	表示用于追踪目的的模型参数化。	可选
_stream	用于实现流式传输	可选
_agenerate	用于实现原生异步方法	可选
_astream	用于实现 _stream 的异步版本	可选

方法	描述
get_name	获取 Runnable 的名称。
get_input_schema	获取可用于验证 Runnable 输入的 Pydantic 模型。
get_input_jsonschema	获取表示 Runnable 输入的 JSON 模式。
get_output_schema	获取可用于验证 Runnable 输出的 Pydantic 模型。
get_output_jsonschema	获取表示 Runnable 输出的 JSON 模式。
config_schema	此 Runnable 接受的配置类型，指定为 Pydantic 模型。
get_config_jsonschema	获取表示 Runnable 配置的 JSON 模式。
get_graph	返回此 Runnable 的图形表示。
get_prompts	返回此 Runnable 使用的提示列表。
__or__	Runnable "or" 运算符。
__ror__	Runnable "reverse-or" 运算符。
pipe	管道连接 Runnable 对象。
pick	从此 Runnable 的输出 dict 中选择键。
assign	向此 Runnable 的 dict 输出分配新字段。
batch	默认实现使用线程池执行器并行运行 invoke。
batch_as_completed	在输入列表上并行运行 invoke。
abatch	默认实现使用 asyncio.gather 并行运行 ainvoke。
abatch_as_completed	在输入列表上并行运行 ainvoke。
astream_log	流式传输 Runnable 的所有输出，如回调系统所报告。
astream_events	生成事件流。
transform	将输入转换为输出。
atransform	将输入转换为输出。
bind	将参数绑定到 Runnable，返回一个新的 Runnable。
with_config	将配置绑定到 Runnable，返回一个新的 Runnable。
with_listeners	将生命周期侦听器绑定到 Runnable，返回一个新的 Runnable。
with_alisteners	将异步生命周期侦听器绑定到 Runnable。
with_types	将输入和输出类型绑定到 Runnable，返回一个新的 Runnable。
with_retry	创建一个新的 Runnable，它在发生异常时重试原始的 Runnable。
map	返回一个新的 Runnable，它将输入列表映射到输出列表。
with_fallbacks	向 Runnable 添加回退机制，返回一个新的 Runnable。
as_tool	从 Runnable 创建一个 BaseTool。
__init__
is_lc_serializable	这个类是否可序列化？
get_lc_namespace	获取 LangChain 对象的命名空间。
lc_id	为此类返回一个用于序列化目的的唯一标识符。
to_json	将 Runnable 序列化为 JSON。
to_json_not_implemented	序列化一个“未实现”的对象。
configurable_fields	在运行时配置特定的 Runnable 字段。
configurable_alternatives	为可在运行时设置的 Runnable 对象配置备选项。
set_verbose	如果 verbose 是 None，则设置它。
get_token_ids	返回文本中 token 的有序 ID。
get_num_tokens	获取文本中存在的 token 数量。
get_num_tokens_from_messages	获取消息中的 token 数量。
invoke	将单个输入转换为输出。
ainvoke	将单个输入转换为输出。
stream	stream 的默认实现，它调用 invoke。
astream	astream 的默认实现，它调用 ainvoke。
generate	将一系列提示传递给模型并返回模型生成的内容。
agenerate	异步地将一系列提示传递给模型并返回生成的内容。
generate_prompt	将一系列提示传递给模型并返回模型生成的内容。
agenerate_prompt	异步地将一系列提示传递并返回模型生成的内容。
dict	返回 LLM 的字典。
bind_tools	将工具绑定到模型。
with_structured_output	返回与给定模式匹配的格式化输出的模型包装器。

name class-attribute instance-attribute

name: str | None = None

Runnable 的名称。用于调试和追踪。

InputType property

InputType: TypeAlias

获取此 Runnable 的输入类型。

input_schema property

input_schema: type[BaseModel]

此 Runnable 接受的输入类型，指定为 Pydantic 模型。

output_schema property

output_schema: type[BaseModel]

输出模式。

此 Runnable 产生的输出类型，指定为 Pydantic 模型。

config_specs property

config_specs: list[ConfigurableFieldSpec]

列出此 Runnable 的可配置字段。

lc_secrets property

lc_secrets: dict[str, str]

构造函数参数名称到密钥 ID 的映射。

例如，{"openai_api_key": "OPENAI_API_KEY"}

lc_attributes property

lc_attributes: dict

应包含在序列化 kwargs 中的属性名称列表。

这些属性必须被构造函数接受。

默认为空字典。

cache class-attribute instance-attribute

cache: BaseCache | bool | None = Field(default=None, exclude=True)

是否缓存响应。

• 如果为 True，将使用全局缓存。
• 如果为 False，将不使用缓存
• 如果为 None，如果设置了全局缓存，则使用全局缓存，否则不使用缓存。
• 如果是 BaseCache 的实例，将使用提供的缓存。

目前不支持模型的流式方法的缓存。

verbose class-attribute instance-attribute

verbose: bool = Field(default_factory=_get_verbosity, exclude=True, repr=False)

是否打印响应文本。

callbacks class-attribute instance-attribute

callbacks: Callbacks = Field(default=None, exclude=True)

添加到运行跟踪中的回调。

tags class-attribute instance-attribute

tags: list[str] | None = Field(default=None, exclude=True)

添加到运行跟踪中的标签。

metadata class-attribute instance-attribute

metadata: dict[str, Any] | None = Field(default=None, exclude=True)

添加到运行跟踪中的元数据。

custom_get_token_ids class-attribute instance-attribute

custom_get_token_ids: Callable[[str], list[int]] | None = Field(
    default=None, exclude=True
)

用于计算 token 的可选编码器。

rate_limiter class-attribute instance-attribute

rate_limiter: BaseRateLimiter | None = Field(default=None, exclude=True)

一个可选的速率限制器，用于限制请求数量。

disable_streaming class-attribute instance-attribute

disable_streaming: bool | Literal['tool_calling'] = False

是否为此模型禁用流式传输。

如果绕过流式传输，则 stream/astream/astream_events 将转而调用 invoke/ainvoke。

• 如果为 True，将始终绕过流式传输情况。
• 如果为 'tool_calling'，仅当模型被调用时带有 tools 关键字参数时，才会绕过流式传输情况。换句话说，仅当提供 tools 参数时，LangChain 才会自动切换到非流式行为 (invoke)。这提供了两全其美的方案。
• 如果为 False（默认值），则在可用时始终使用流式传输情况。

此标志的主要原因是，代码可能是使用 stream 编写的，而用户可能希望将给定模型换成另一个实现不完全支持流式传输的模型。

output_version class-attribute instance-attribute

output_version: str | None = Field(
    default_factory=from_env("LC_OUTPUT_VERSION", default=None)
)

要存储在消息内容中的 AIMessage 输出格式的版本。

AIMessage.content_blocks 将懒解析 content 的内容为标准格式。此标志可用于额外将标准格式存储在消息内容中，例如，用于序列化目的。

支持的值

• 'v0'：内容中特定于提供商的格式（可以使用 content_blocks 进行懒解析）
• 'v1'：内容中的标准化格式（与 content_blocks 一致）

合作伙伴包（例如 langchain-openai）也可以使用此字段以向后兼容的方式推出新的内容格式。

在版本 1.0 中添加

OutputType property

OutputType: Any

获取此 Runnable 的输出类型。

profile property

profile: ModelProfile

返回模型的性能分析信息。

此属性将依赖于 langchain-model-profiles 包来检索聊天模型的能力，例如上下文窗口大小和支持的功能。

引发	描述
ImportError	如果未安装 langchain-model-profiles。

返回	描述
ModelProfile	一个 ModelProfile 对象，包含模型的性能分析信息。

get_name

get_name(suffix: str | None = None, *, name: str | None = None) -> str

获取 Runnable 的名称。

参数	描述
suffix	一个可选的后缀，附加到名称上。 类型: str | None 默认值: None
name	一个可选的名称，用于代替 Runnable 的名称。 类型: str | None 默认值: None

返回	描述
str	Runnable 的名称。

get_input_schema

get_input_schema(config: RunnableConfig | None = None) -> type[BaseModel]

获取可用于验证 Runnable 输入的 Pydantic 模型。

利用 configurable_fields 和 configurable_alternatives 方法的 Runnable 对象将具有一个动态输入模式，该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输入模式。

参数	描述
config	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证输入的 Pydantic 模型。

get_input_jsonschema

get_input_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

获取表示 Runnable 输入的 JSON 模式。

参数	描述
config	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 输入的 JSON 模式。

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_input_jsonschema())

在 0.3.0 版本中新增。

get_output_schema

get_output_schema(config: RunnableConfig | None = None) -> type[BaseModel]

获取可用于验证 Runnable 输出的 Pydantic 模型。

利用 configurable_fields 和 configurable_alternatives 方法的 Runnable 对象将具有一个动态输出模式，该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输出模式。

参数	描述
config	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证输出的 Pydantic 模型。

get_output_jsonschema

get_output_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

获取表示 Runnable 输出的 JSON 模式。

参数	描述
config	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 输出的 JSON 模式。

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_output_jsonschema())

在 0.3.0 版本中新增。

config_schema

config_schema(*, include: Sequence[str] | None = None) -> type[BaseModel]

此 Runnable 接受的配置类型，指定为 Pydantic 模型。

要将字段标记为可配置，请参阅 configurable_fields 和 configurable_alternatives 方法。

参数	描述
include	要包含在配置模式中的字段列表。 类型: Sequence[str] | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证配置的 Pydantic 模型。

get_config_jsonschema

get_config_jsonschema(*, include: Sequence[str] | None = None) -> dict[str, Any]

获取表示 Runnable 配置的 JSON 模式。

参数	描述
include	要包含在配置模式中的字段列表。 类型: Sequence[str] | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 配置的 JSON 模式。

在 0.3.0 版本中新增。

get_graph

get_graph(config: RunnableConfig | None = None) -> Graph

返回此 Runnable 的图形表示。

get_prompts

get_prompts(config: RunnableConfig | None = None) -> list[BasePromptTemplate]

返回此 Runnable 使用的提示列表。

__or__

__or__(
    other: Runnable[Any, Other]
    | Callable[[Iterator[Any]], Iterator[Other]]
    | Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
    | Callable[[Any], Other]
    | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]

Runnable "or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence。

参数	描述
other	另一个 Runnable 或类 Runnable 对象。 类型: Runnable[Any, Other] | Callable[[Iterator[Any]], Iterator[Other]] | Callable[[AsyncIterator[Any]], AsyncIterator[Other]] | Callable[[Any], Other] | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any]

返回	描述
RunnableSerializable[Input, Other]	一个新的 Runnable。

__ror__

__ror__(
    other: Runnable[Other, Any]
    | Callable[[Iterator[Other]], Iterator[Any]]
    | Callable[[AsyncIterator[Other]], AsyncIterator[Any]]
    | Callable[[Other], Any]
    | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any],
) -> RunnableSerializable[Other, Output]

Runnable "reverse-or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence。

参数	描述
other	另一个 Runnable 或类 Runnable 对象。 类型: Runnable[Other, Any] | Callable[[Iterator[Other]], Iterator[Any]] | Callable[[AsyncIterator[Other]], AsyncIterator[Any]] | Callable[[Other], Any] | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any]

返回	描述
RunnableSerializable[Other, Output]	一个新的 Runnable。

pipe

pipe(
    *others: Runnable[Any, Other] | Callable[[Any], Other], name: str | None = None
) -> RunnableSerializable[Input, Other]

管道连接 Runnable 对象。

将此 Runnable 与类 Runnable 对象组合以构成一个 RunnableSequence。

等同于 RunnableSequence(self, *others) 或 self | others[0] | ...

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


def mul_two(x: int) -> int:
    return x * 2


runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1.pipe(runnable_2)
# Or equivalently:
# sequence = runnable_1 | runnable_2
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)
# -> 4

sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
# -> [4, 6, 8]

参数	描述
*others	其他要组合的 Runnable 或类 Runnable 对象 类型: Runnable[Any, Other] | Callable[[Any], Other] 默认值: ()
name	生成的 RunnableSequence 的一个可选名称。 类型: str | None 默认值: None

返回	描述
RunnableSerializable[Input, Other]	一个新的 Runnable。

pick

pick(keys: str | list[str]) -> RunnableSerializable[Any, Any]

从此 Runnable 的输出 dict 中选择键。

选择单个键

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
chain = RunnableMap(str=as_str, json=as_json)

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3]}

json_only_chain = chain.pick("json")
json_only_chain.invoke("[1, 2, 3]")
# -> [1, 2, 3]

选择键列表

from typing import Any

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)


def as_bytes(x: Any) -> bytes:
    return bytes(x, "utf-8")


chain = RunnableMap(str=as_str, json=as_json, bytes=RunnableLambda(as_bytes))

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

json_and_bytes_chain = chain.pick(["json", "bytes"])
json_and_bytes_chain.invoke("[1, 2, 3]")
# -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

参数	描述
keys	从输出字典中选择的一个键或键列表。 类型: str | list[str]

返回	描述
RunnableSerializable[Any, Any]	一个新的 Runnable。

assign

assign(
    **kwargs: Runnable[dict[str, Any], Any]
    | Callable[[dict[str, Any]], Any]
    | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]],
) -> RunnableSerializable[Any, Any]

向此 Runnable 的 dict 输出分配新字段。

from langchain_community.llms.fake import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.runnables import Runnable
from operator import itemgetter

prompt = (
    SystemMessagePromptTemplate.from_template("You are a nice assistant.")
    + "{question}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])

chain: Runnable = prompt | model | {"str": StrOutputParser()}

chain_with_assign = chain.assign(hello=itemgetter("str") | model)

print(chain_with_assign.input_schema.model_json_schema())
# {'title': 'PromptInput', 'type': 'object', 'properties':
{'question': {'title': 'Question', 'type': 'string'}}}
print(chain_with_assign.output_schema.model_json_schema())
# {'title': 'RunnableSequenceOutput', 'type': 'object', 'properties':
{'str': {'title': 'Str',
'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}}

参数	描述
**kwargs	一个键到 Runnable 或类 Runnable 对象的映射，这些对象将使用此 Runnable 的整个输出字典来调用。 类型: Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any] | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]] 默认值: {}

返回	描述
RunnableSerializable[Any, Any]	一个新的 Runnable。

batch

batch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

默认实现使用线程池执行器并行运行 invoke。

批处理的默认实现对于 IO 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理，则必须重写此方法；例如，如果底层的 Runnable 使用支持批处理模式的 API。

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
config	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | list[RunnableConfig] | None 默认值: None
return_exceptions	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

batch_as_completed

batch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> Iterator[tuple[int, Output | Exception]]

在输入列表上并行运行 invoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
config	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | Sequence[RunnableConfig] | None 默认值: None
return_exceptions	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
tuple[int, Output | Exception]	由输入索引和 Runnable 输出组成的元组。

abatch async

abatch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

默认实现使用 asyncio.gather 并行运行 ainvoke。

batch 的默认实现对于 IO 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理，则必须重写此方法；例如，如果底层的 Runnable 使用支持批处理模式的 API。

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
config	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | list[RunnableConfig] | None 默认值: None
return_exceptions	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

abatch_as_completed async

abatch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> AsyncIterator[tuple[int, Output | Exception]]

在输入列表上并行运行 ainvoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
config	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | Sequence[RunnableConfig] | None 默认值: None
return_exceptions	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[tuple[int, Output | Exception]]	一个由输入索引和 Runnable 输出组成的元组。

astream_log async

astream_log(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    diff: bool = True,
    with_streamed_output_list: bool = True,
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]

流式传输 Runnable 的所有输出，如回调系统所报告。

这包括 LLM、检索器、工具等的所有内部运行。

输出以 Log 对象的形式流式传输，其中包括一个 Jsonpatch 操作列表，描述了运行状态在每一步中如何变化，以及运行的最终状态。

可以按顺序应用 Jsonpatch 操作来构造状态。

参数	描述
input	Runnable 的输入。 类型： Any
config	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
diff	是生成每一步之间的差异还是当前状态。 类型: bool 默认值: True
with_streamed_output_list	是否生成 streamed_output 列表。 类型: bool 默认值: True
include_names	仅包含具有这些名称的日志。 类型: Sequence[str] | None 默认值: None
include_types	仅包含具有这些类型的日志。 类型: Sequence[str] | None 默认值: None
include_tags	仅包含具有这些标签的日志。 类型: Sequence[str] | None 默认值: None
exclude_names	排除具有这些名称的日志。 类型: Sequence[str] | None 默认值: None
exclude_types	排除具有这些类型的日志。 类型: Sequence[str] | None 默认值: None
exclude_tags	排除具有这些标签的日志。 类型: Sequence[str] | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型： Any 默认值： {}

YIELDS	描述
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]	一个 RunLogPatch 或 RunLog 对象。

astream_events async

astream_events(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    version: Literal["v1", "v2"] = "v2",
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]

生成事件流。

用于创建一个 StreamEvent 的迭代器，提供有关 Runnable 进度的实时信息，包括来自中间结果的 StreamEvent。

一个 StreamEvent 是一个具有以下模式的字典

• event：事件名称的格式为：on_[runnable_type]_(start|stream|end)。
• name：生成事件的 Runnable 的名称。
• run_id：与发出事件的 Runnable 的给定执行相关联的随机生成的 ID。作为父 Runnable 执行的一部分被调用的子 Runnable 被分配其自己的唯一 ID。
• parent_ids：生成事件的父可运行对象的 ID。根 Runnable 将有一个空列表。父 ID 的顺序是从根到直接父级。仅适用于 API 的 v2 版本。API 的 v1 版本将返回一个空列表。
• tags：生成事件的 Runnable 的标签。
• metadata：生成事件的 Runnable 的元数据。
• data：与事件关联的数据。此字段的内容取决于事件的类型。有关更多详细信息，请参见下表。

下表说明了各种链可能发出的某些事件。为简洁起见，已从表中省略了元数据字段。链定义已包含在表之后。

注意

此参考表适用于模式的 v2 版本。

事件	name	chunk	输入	output
on_chat_model_start	'[model name]'		{"messages": [[SystemMessage, HumanMessage]]}
on_chat_model_stream	'[model name]'	AIMessageChunk(content="hello")
on_chat_model_end	'[model name]'		{"messages": [[SystemMessage, HumanMessage]]}	AIMessageChunk(content="hello world")
on_llm_start	'[model name]'		{'input': 'hello'}
on_llm_stream	'[model name]'	'你好'
on_llm_end	'[model name]'		'你好，人类！'
on_chain_start	'format_docs'
on_chain_stream	'format_docs'	'hello world!, goodbye world!'
on_chain_end	'format_docs'		[Document(...)]	'hello world!, goodbye world!'
on_tool_start	'some_tool'		{"x": 1, "y": "2"}
on_tool_end	'some_tool'			{"x": 1, "y": "2"}
on_retriever_start	'[retriever name]'		{"query": "hello"}
on_retriever_end	'[retriever name]'		{"query": "hello"}	[Document(...), ..]
on_prompt_start	'[template_name]'		{"question": "hello"}
on_prompt_end	'[template_name]'		{"question": "hello"}	ChatPromptValue(messages: [SystemMessage, ...])

除了标准事件外，用户还可以分派自定义事件（见下例）。

自定义事件将仅在 API 的 v2 版本中出现！

自定义事件具有以下格式

属性	类型	描述
name	str	用户为事件定义的名称。
data	任意	与事件关联的数据。这可以是任何东西，但我们建议使其可 JSON 序列化。

以下是与上面显示的标准事件相关的声明

format_docs:

def format_docs(docs: list[Document]) -> str:
    '''Format the docs.'''
    return ", ".join([doc.page_content for doc in docs])


format_docs = RunnableLambda(format_docs)

some_tool:

@tool
def some_tool(x: int, y: str) -> dict:
    '''Some_tool.'''
    return {"x": x, "y": y}

prompt:

template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are Cat Agent 007"),
        ("human", "{question}"),
    ]
).with_config({"run_name": "my_template", "tags": ["my_template"]})

例如

from langchain_core.runnables import RunnableLambda


async def reverse(s: str) -> str:
    return s[::-1]


chain = RunnableLambda(func=reverse)

events = [event async for event in chain.astream_events("hello", version="v2")]

# Will produce the following events
# (run_id, and parent_ids has been omitted for brevity):
[
    {
        "data": {"input": "hello"},
        "event": "on_chain_start",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"chunk": "olleh"},
        "event": "on_chain_stream",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"output": "olleh"},
        "event": "on_chain_end",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
]

示例：分派自定义事件

from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio


async def slow_thing(some_input: str, config: RunnableConfig) -> str:
    """Do something that takes a long time."""
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 1 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 2 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    return "Done"

slow_thing = RunnableLambda(slow_thing)

async for event in slow_thing.astream_events("some_input", version="v2"):
    print(event)

参数	描述
input	Runnable 的输入。 类型： Any
config	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
version	要使用的模式版本，可以是 'v2' 或 'v1'。用户应使用 'v2'。'v1' 是为了向后兼容，并将在 0.4.0 中弃用。在 API 稳定之前不会分配默认值。自定义事件将仅在 'v2' 中出现。 类型: Literal['v1', 'v2'] 默认值: 'v2'
include_names	仅包括来自具有匹配名称的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
include_types	仅包括来自具有匹配类型的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
include_tags	仅包括来自具有匹配标签的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
exclude_names	排除来自具有匹配名称的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
exclude_types	排除来自具有匹配类型的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
exclude_tags	排除来自具有匹配标签的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。这些将传递给 astream_log，因为 astream_events 的此实现是建立在 astream_log 之上的。 类型： Any 默认值： {}

YIELDS	描述
AsyncIterator[StreamEvent]	StreamEvent 的异步流。

引发	描述
NotImplementedError	如果版本不是 'v1' 或 'v2'。

transform

transform(
    input: Iterator[Input], config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]

将输入转换为输出。

transform 的默认实现，它缓冲输入并调用 astream。

如果子类可以在输入仍在生成时开始产生输出，则必须重写此方法。

参数	描述
input	Runnable 输入的迭代器。 类型: Iterator[Input]
config	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
输出	Runnable 的输出。

atransform async

atransform(
    input: AsyncIterator[Input],
    config: RunnableConfig | None = None,
    **kwargs: Any | None,
) -> AsyncIterator[Output]

将输入转换为输出。

atransform 的默认实现，它缓冲输入并调用 astream。

如果子类可以在输入仍在生成时开始产生输出，则必须重写此方法。

参数	描述
input	Runnable 输入的异步迭代器。 类型： AsyncIterator[Input]
config	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[Output]	Runnable 的输出。

bind

bind(**kwargs: Any) -> Runnable[Input, Output]

将参数绑定到 Runnable，返回一个新的 Runnable。

当链中的 Runnable 需要一个不在前一个 Runnable 的输出中或未包含在用户输入中的参数时非常有用。

参数	描述
**kwargs	要绑定到 Runnable 的参数。 类型： Any 默认值： {}

返回	描述
Runnable[Input, Output]	一个绑定了参数的新 Runnable。

示例

from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser

model = ChatOllama(model="llama3.1")

# Without bind
chain = model | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'

# With bind
chain = model.bind(stop=["three"]) | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'

with_config

with_config(
    config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]

将配置绑定到 Runnable，返回一个新的 Runnable。

参数	描述
config	要绑定到 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型： Any 默认值： {}

返回	描述
Runnable[Input, Output]	一个绑定了配置的新 Runnable。

with_listeners

with_listeners(
    *,
    on_start: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
    on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
    on_error: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
) -> Runnable[Input, Output]

将生命周期侦听器绑定到 Runnable，返回一个新的 Runnable。

Run 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_time 以及添加到运行中的任何标签或元数据。

参数	描述
on_start	在 Runnable 开始运行之前调用，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None
on_end	在 Runnable 完成运行后调用，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None
on_error	如果 Runnable 抛出错误，则调用此函数，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了监听器的新 Runnable。

示例

from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run

import time


def test_runnable(time_to_sleep: int):
    time.sleep(time_to_sleep)


def fn_start(run_obj: Run):
    print("start_time:", run_obj.start_time)


def fn_end(run_obj: Run):
    print("end_time:", run_obj.end_time)


chain = RunnableLambda(test_runnable).with_listeners(
    on_start=fn_start, on_end=fn_end
)
chain.invoke(2)

with_alisteners

with_alisteners(
    *,
    on_start: AsyncListener | None = None,
    on_end: AsyncListener | None = None,
    on_error: AsyncListener | None = None,
) -> Runnable[Input, Output]

将异步生命周期侦听器绑定到 Runnable。

返回一个新的 Runnable。

Run 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_time 以及添加到运行中的任何标签或元数据。

参数	描述
on_start	在 Runnable 开始运行之前异步调用，并传入 Run 对象。 类型： AsyncListener | None 默认值： None
on_end	在 Runnable 完成运行后异步调用，并传入 Run 对象。 类型： AsyncListener | None 默认值： None
on_error	如果 Runnable 抛出错误，则异步调用此函数，并传入 Run 对象。 类型： AsyncListener | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了监听器的新 Runnable。

示例

from langchain_core.runnables import RunnableLambda, Runnable
from datetime import datetime, timezone
import time
import asyncio

def format_t(timestamp: float) -> str:
    return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()

async def test_runnable(time_to_sleep: int):
    print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}")
    await asyncio.sleep(time_to_sleep)
    print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}")

async def fn_start(run_obj: Runnable):
    print(f"on start callback starts at {format_t(time.time())}")
    await asyncio.sleep(3)
    print(f"on start callback ends at {format_t(time.time())}")

async def fn_end(run_obj: Runnable):
    print(f"on end callback starts at {format_t(time.time())}")
    await asyncio.sleep(2)
    print(f"on end callback ends at {format_t(time.time())}")

runnable = RunnableLambda(test_runnable).with_alisteners(
    on_start=fn_start,
    on_end=fn_end
)
async def concurrent_runs():
    await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))

asyncio.run(concurrent_runs())
Result:
on start callback starts at 2025-03-01T07:05:22.875378+00:00
on start callback starts at 2025-03-01T07:05:22.875495+00:00
on start callback ends at 2025-03-01T07:05:25.878862+00:00
on start callback ends at 2025-03-01T07:05:25.878947+00:00
Runnable[2s]: starts at 2025-03-01T07:05:25.879392+00:00
Runnable[3s]: starts at 2025-03-01T07:05:25.879804+00:00
Runnable[2s]: ends at 2025-03-01T07:05:27.881998+00:00
on end callback starts at 2025-03-01T07:05:27.882360+00:00
Runnable[3s]: ends at 2025-03-01T07:05:28.881737+00:00
on end callback starts at 2025-03-01T07:05:28.882428+00:00
on end callback ends at 2025-03-01T07:05:29.883893+00:00
on end callback ends at 2025-03-01T07:05:30.884831+00:00

with_types

with_types(
    *, input_type: type[Input] | None = None, output_type: type[Output] | None = None
) -> Runnable[Input, Output]

将输入和输出类型绑定到 Runnable，返回一个新的 Runnable。

参数	描述
input_type	要绑定到 Runnable 的输入类型。 类型： type[Input] | None 默认值： None
output_type	要绑定到 Runnable 的输出类型。 类型： type[Output] | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了类型的新 Runnable。

with_retry

with_retry(
    *,
    retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
    wait_exponential_jitter: bool = True,
    exponential_jitter_params: ExponentialJitterParams | None = None,
    stop_after_attempt: int = 3,
) -> Runnable[Input, Output]

创建一个新的 Runnable，它在发生异常时重试原始的 Runnable。

参数	描述
retry_if_exception_type	一个用于重试的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
wait_exponential_jitter	是否在两次重试之间的等待时间中添加抖动。 类型: bool 默认值: True
stop_after_attempt	放弃前尝试的最大次数。 类型： int 默认值： 3
exponential_jitter_params	tenacity.wait_exponential_jitter 的参数。即：initial、max、exp_base 和 jitter（均为 float 值）。 类型： ExponentialJitterParams | None 默认值： None

返回	描述
Runnable[Input, Output]	一个新的 Runnable，它会在发生异常时重试原始的 Runnable。

示例

from langchain_core.runnables import RunnableLambda

count = 0


def _lambda(x: int) -> None:
    global count
    count = count + 1
    if x == 1:
        raise ValueError("x is 1")
    else:
        pass


runnable = RunnableLambda(_lambda)
try:
    runnable.with_retry(
        stop_after_attempt=2,
        retry_if_exception_type=(ValueError,),
    ).invoke(1)
except ValueError:
    pass

assert count == 2

map

map() -> Runnable[list[Input], list[Output]]

返回一个新的 Runnable，它将输入列表映射到输出列表。

使用每个输入调用 invoke。

返回	描述
Runnable[list[Input], list[Output]]	一个新的 Runnable，它将输入列表映射到输出列表。

示例

from langchain_core.runnables import RunnableLambda


def _lambda(x: int) -> int:
    return x + 1


runnable = RunnableLambda(_lambda)
print(runnable.map().invoke([1, 2, 3]))  # [2, 3, 4]

with_fallbacks

with_fallbacks(
    fallbacks: Sequence[Runnable[Input, Output]],
    *,
    exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
    exception_key: str | None = None,
) -> RunnableWithFallbacks[Input, Output]

向 Runnable 添加回退机制，返回一个新的 Runnable。

新的 Runnable 将在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

参数	描述
fallbacks	如果原始 Runnable 失败，将尝试的一系列 runnable。 类型： Sequence[Runnable[Input, Output]]
exceptions_to_handle	一个要处理的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
exception_key	如果指定了 string，则已处理的异常将作为输入的一部分，在指定的键下传递给备选方案。如果为 None，异常将不会传递给备选方案。如果使用，基础 Runnable 及其备选方案必须接受一个字典作为输入。 类型: str | None 默认值: None

返回	描述
RunnableWithFallbacks[Input, Output]	一个新的 Runnable，它会在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

示例

from typing import Iterator

from langchain_core.runnables import RunnableGenerator


def _generate_immediate_error(input: Iterator) -> Iterator[str]:
    raise ValueError()
    yield ""


def _generate(input: Iterator) -> Iterator[str]:
    yield from "foo bar"


runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
    [RunnableGenerator(_generate)]
)
print("".join(runnable.stream({})))  # foo bar

参数	描述
fallbacks	如果原始 Runnable 失败，将尝试的一系列 runnable。 类型： Sequence[Runnable[Input, Output]]
exceptions_to_handle	一个要处理的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
exception_key	如果指定了 string，则已处理的异常将作为输入的一部分，在指定的键下传递给备选方案。如果为 None，异常将不会传递给备选方案。如果使用，基础 Runnable 及其备选方案必须接受一个字典作为输入。 类型: str | None 默认值: None

返回	描述
RunnableWithFallbacks[Input, Output]	一个新的 Runnable，它会在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

as_tool

as_tool(
    args_schema: type[BaseModel] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    arg_types: dict[str, type] | None = None,
) -> BaseTool

从 Runnable 创建一个 BaseTool。

as_tool 将从一个 Runnable 实例化一个 BaseTool，该工具具有名称、描述和 args_schema。在可能的情况下，模式会从 runnable.get_input_schema 中推断。或者（例如，如果 Runnable 接受一个字典作为输入，并且特定的字典键没有类型），模式可以通过 args_schema 直接指定。你也可以传递 arg_types 来仅指定必需的参数及其类型。

参数	描述
args_schema	工具的模式。 类型： type[BaseModel] | None 默认值： None
name	工具的名称。 类型: str | None 默认值: None
description	工具的描述。 类型: str | None 默认值: None
arg_types	一个从参数名称到类型的字典。 类型： dict[str, type] | None 默认值： None

返回	描述
BaseTool	一个 BaseTool 实例。

类型化字典输入

from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda


class Args(TypedDict):
    a: int
    b: list[int]


def f(x: Args) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool()
as_tool.invoke({"a": 3, "b": [1, 2]})

dict 输入，通过 args_schema 指定模式

from typing import Any
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableLambda

def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))

class FSchema(BaseModel):
    """Apply a function to an integer and list of integers."""

    a: int = Field(..., description="Integer")
    b: list[int] = Field(..., description="List of ints")

runnable = RunnableLambda(f)
as_tool = runnable.as_tool(FSchema)
as_tool.invoke({"a": 3, "b": [1, 2]})

dict 输入，通过 arg_types 指定模式

from typing import Any
from langchain_core.runnables import RunnableLambda


def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool(arg_types={"a": int, "b": list[int]})
as_tool.invoke({"a": 3, "b": [1, 2]})

字符串输入

from langchain_core.runnables import RunnableLambda


def f(x: str) -> str:
    return x + "a"


def g(x: str) -> str:
    return x + "z"


runnable = RunnableLambda(f) | g
as_tool = runnable.as_tool()
as_tool.invoke("b")

__init__

__init__(*args: Any, **kwargs: Any) -> None

is_lc_serializable classmethod

is_lc_serializable() -> bool

这个类是否可序列化？

根据设计，即使一个类继承自 Serializable，它默认也是不可序列化的。这是为了防止意外序列化不应被序列化的对象。

返回	描述
bool	类是否可序列化。默认为 False。

get_lc_namespace classmethod

get_lc_namespace() -> list[str]

获取 LangChain 对象的命名空间。

例如，如果类是 langchain.llms.openai.OpenAI，那么命名空间是 ["langchain", "llms", "openai"]

返回	描述
list[str]	命名空间。

lc_id classmethod

lc_id() -> list[str]

为此类返回一个用于序列化目的的唯一标识符。

唯一标识符是一个描述对象路径的字符串列表。

例如，对于类 langchain.llms.openai.OpenAI，id 是 ["langchain", "llms", "openai", "OpenAI"]。

to_json

to_json() -> SerializedConstructor | SerializedNotImplemented

将 Runnable 序列化为 JSON。

返回	描述
SerializedConstructor | SerializedNotImplemented	一个 Runnable 的 JSON 可序列化表示。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

序列化一个“未实现”的对象。

返回	描述
SerializedNotImplemented	SerializedNotImplemented.

configurable_fields

configurable_fields(
    **kwargs: AnyConfigurableField,
) -> RunnableSerializable[Input, Output]

在运行时配置特定的 Runnable 字段。

参数	描述
**kwargs	一个要配置的 ConfigurableField 实例的字典。 类型： AnyConfigurableField 默认值： {}

引发	描述
ValueError	如果在 Runnable 中找不到配置键。

返回	描述
RunnableSerializable[Input, Output]	一个配置了字段的新 Runnable。

from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatOpenAI(max_tokens=20).configurable_fields(
    max_tokens=ConfigurableField(
        id="output_token_number",
        name="Max tokens in the output",
        description="The maximum number of tokens in the output",
    )
)

# max_tokens = 20
print("max_tokens_20: ", model.invoke("tell me something about chess").content)

# max_tokens = 200
print(
    "max_tokens_200: ",
    model.with_config(configurable={"output_token_number": 200})
    .invoke("tell me something about chess")
    .content,
)

configurable_alternatives

configurable_alternatives(
    which: ConfigurableField,
    *,
    default_key: str = "default",
    prefix_keys: bool = False,
    **kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]],
) -> RunnableSerializable[Input, Output]

为可在运行时设置的 Runnable 对象配置备选项。

参数	描述
which	将用于选择备选项的 ConfigurableField 实例。 类型： ConfigurableField
default_key	如果未选择备选项，则使用的默认键。 类型： str 默认值： 'default'
prefix_keys	是否用 ConfigurableField id 作为键的前缀。 类型： bool 默认值： False
**kwargs	一个从键到 Runnable 实例或返回 Runnable 实例的可调用对象的字典。 类型： Runnable[Input, Output] | Callable[[], Runnable[Input, Output]] 默认值： {}

返回	描述
RunnableSerializable[Input, Output]	一个配置了备选项的新 Runnable。

from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatAnthropic(
    model_name="claude-sonnet-4-5-20250929"
).configurable_alternatives(
    ConfigurableField(id="llm"),
    default_key="anthropic",
    openai=ChatOpenAI(),
)

# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)

# uses ChatOpenAI
print(
    model.with_config(configurable={"llm": "openai"})
    .invoke("which organization created you?")
    .content
)

set_verbose

set_verbose(verbose: bool | None) -> bool

如果 verbose 是 None，则设置它。

这允许用户传入 None 作为 verbose 来访问全局设置。

参数	描述
verbose	要使用的详细程度设置。 类型： bool | None

返回	描述
bool	要使用的详细程度设置。

get_token_ids

get_token_ids(text: str) -> list[int]

返回文本中 token 的有序 ID。

参数	描述
text	要进行分词的字符串输入。 类型： str

返回	描述
list[int]	与文本中的词元相对应的 ID 列表，按它们在文本中出现的顺序列出。

get_num_tokens

get_num_tokens(text: str) -> int

获取文本中存在的 token 数量。

用于检查输入是否适合模型的上下文窗口。

参数	描述
text	要进行分词的字符串输入。 类型： str

返回	描述
int	文本中的词元整数数量。

get_num_tokens_from_messages

get_num_tokens_from_messages(
    messages: list[BaseMessage], tools: Sequence | None = None
) -> int

获取消息中的 token 数量。

用于检查输入是否适合模型的上下文窗口。

注意

get_num_tokens_from_messages 的基本实现忽略了工具模式。

参数	描述
messages	要进行分词的消息输入。 类型： list[BaseMessage]
tools	如果提供，则为要转换为工具模式的 dict、BaseModel、函数或 BaseTool 对象的序列。 类型： Sequence | None 默认值： None

返回	描述
int	所有消息中词元数量的总和。

invoke

invoke(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AIMessage

将单个输入转换为输出。

参数	描述
input	Runnable 的输入。 类型： Input
config	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | None 默认值: None

返回	描述
输出	Runnable 的输出。

ainvoke async

ainvoke(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AIMessage

将单个输入转换为输出。

参数	描述
input	Runnable 的输入。 类型： Input
config	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | None 默认值: None

返回	描述
输出	Runnable 的输出。

stream

stream(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> Iterator[AIMessageChunk]

stream 的默认实现，它调用 invoke。

如果子类支持流式输出，则必须重写此方法。

参数	描述
input	Runnable 的输入。 类型： Input
config	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
输出	Runnable 的输出。

astream async

astream(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[AIMessageChunk]

astream 的默认实现，它调用 ainvoke。

如果子类支持流式输出，则必须重写此方法。

参数	描述
input	Runnable 的输入。 类型： Input
config	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[Output]	Runnable 的输出。

generate

generate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: UUID | None = None,
    **kwargs: Any,
) -> LLMResult

将一系列提示传递给模型并返回模型生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
messages	消息列表的列表。 类型： list[list[BaseMessage]]
stop	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
callbacks	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
tags	要应用的标签。 类型: list[str] | None 默认值: None
metadata	要应用的元数据。 类型： dict[str, Any] | None 默认值： None
run_name	运行的名称。 类型: str | None 默认值: None
run_id	运行的 ID。 类型: UUID | None 默认值: None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generations 列表以及额外的模型提供商特定的输出。

agenerate async

agenerate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: UUID | None = None,
    **kwargs: Any,
) -> LLMResult

异步地将一系列提示传递给模型并返回生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
messages	消息列表的列表。 类型： list[list[BaseMessage]]
stop	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
callbacks	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
tags	要应用的标签。 类型: list[str] | None 默认值: None
metadata	要应用的元数据。 类型： dict[str, Any] | None 默认值： None
run_name	运行的名称。 类型: str | None 默认值: None
run_id	运行的 ID。 类型: UUID | None 默认值: None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generations 列表以及额外的模型提供商特定的输出。

generate_prompt

generate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult

将一系列提示传递给模型并返回模型生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
prompts	PromptValue 对象列表。PromptValue 是一个可以转换为匹配任何语言模型格式的对象（对于纯文本生成模型是字符串，对于聊天模型是 BaseMessage 对象）。 类型： list[PromptValue]
stop	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
callbacks	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generation 对象列表以及额外的模型提供商特定的输出。

agenerate_prompt async

agenerate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult

异步地将一系列提示传递并返回模型生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
prompts	PromptValue 对象列表。PromptValue 是一个可以转换为匹配任何语言模型格式的对象（对于纯文本生成模型是字符串，对于聊天模型是 BaseMessage 对象）。 类型： list[PromptValue]
stop	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
callbacks	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generation 对象列表以及额外的模型提供商特定的输出。

dict

dict(**kwargs: Any) -> dict

返回 LLM 的字典。

bind_tools

bind_tools(
    tools: Sequence[Dict[str, Any] | type | Callable | BaseTool],
    *,
    tool_choice: str | None = None,
    **kwargs: Any,
) -> Runnable[LanguageModelInput, AIMessage]

将工具绑定到模型。

参数	描述
tools	要绑定到模型的工具序列。 类型： Sequence[Dict[str, Any] | type | Callable | BaseTool]
tool_choice	要使用的工具。如果为 "any"，则可以使用任何工具。 类型: str | None 默认值: None

返回	描述
Runnable[LanguageModelInput, AIMessage]	一个返回消息的 Runnable。

with_structured_output

with_structured_output(
    schema: Dict | type, *, include_raw: bool = False, **kwargs: Any
) -> Runnable[LanguageModelInput, Dict | BaseModel]

返回与给定模式匹配的格式化输出的模型包装器。

参数	描述
schema	输出模式。可以作为以下形式传入： 一个 OpenAI 函数/工具模式， 一个 JSON Schema， 一个 TypedDict 类， 或一个 Pydantic 类。 如果 schema 是一个 Pydantic 类，那么模型输出将是该类的 Pydantic 实例，并且模型生成的字段将由 Pydantic 类进行验证。否则，模型输出将是一个字典，并且不会被验证。 有关在指定 Pydantic 或 TypedDict 类时如何正确指定模式字段的类型和描述的更多信息，请参阅 langchain_core.utils.function_calling.convert_to_openai_tool。 类型： Dict | type
include_raw	如果为 False，则仅返回解析后的结构化输出。如果在模型输出解析期间发生错误，它将被引发。如果为 True，则将返回原始模型响应（一个 BaseMessage）和解析后的模型响应。如果在输出解析期间发生错误，它将被捕获并一并返回。 最终输出总是一个带有键 'raw'、'parsed' 和 'parsing_error' 的 dict。 类型： bool 默认值： False

引发	描述
ValueError	如果有任何不支持的 kwargs。
NotImplementedError	如果模型没有实现 with_structured_output()。

返回	描述
Runnable[LanguageModelInput, Dict | BaseModel]	一个接受与 langchain_core.language_models.chat.BaseChatModel 相同输入的 Runnable。如果 include_raw 为 False 且 schema 是一个 Pydantic 类，则 Runnable 输出一个 schema 的实例（即一个 Pydantic 对象）。否则，如果 include_raw 为 False，则 Runnable 输出一个 dict。 如果 include_raw 为 True，则 Runnable 输出一个带有以下键的 dict 'raw': BaseMessage 'parsed': 如果出现解析错误则为 None，否则类型取决于上面描述的 schema。 'parsing_error': BaseException | None

示例：Pydantic 模式 (include_raw=False)

from pydantic import BaseModel


class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''

    answer: str
    justification: str


model = ChatModel(model="model-name", temperature=0)
structured_model = model.with_structured_output(AnswerWithJustification)

structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)

# -> AnswerWithJustification(
#     answer='They weigh the same',
#     justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'
# )

示例：Pydantic 模式 (include_raw=True)

from pydantic import BaseModel


class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''

    answer: str
    justification: str


model = ChatModel(model="model-name", temperature=0)
structured_model = model.with_structured_output(
    AnswerWithJustification, include_raw=True
)

structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)
# -> {
#     'raw': AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_Ao02pnFYXD6GN1yzc0uXPsvF', 'function': {'arguments': '{"answer":"They weigh the same.","justification":"Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ."}', 'name': 'AnswerWithJustification'}, 'type': 'function'}]}),
#     'parsed': AnswerWithJustification(answer='They weigh the same.', justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'),
#     'parsing_error': None
# }

示例：dict 模式 (include_raw=False)

from pydantic import BaseModel
from langchain_core.utils.function_calling import convert_to_openai_tool


class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''

    answer: str
    justification: str


dict_schema = convert_to_openai_tool(AnswerWithJustification)
model = ChatModel(model="model-name", temperature=0)
structured_model = model.with_structured_output(dict_schema)

structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)
# -> {
#     'answer': 'They weigh the same',
#     'justification': 'Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume and density of the two substances differ.'
# }

行为在 0.2.26 版本中已更改

添加了对 TypedDict 类的支持。

langchain_core.messages

消息 是在提示和聊天对话中使用的对象。

BaseMessage

基类：Serializable

基础抽象消息类。

消息是聊天模型的输入和输出。

示例包括 HumanMessage、AIMessage 和 SystemMessage。

方法	描述
lc_id	为此类返回一个用于序列化目的的唯一标识符。
to_json	将对象序列化为 JSON。
to_json_not_implemented	序列化一个“未实现”的对象。
__init__	初始化一个 BaseMessage。
is_lc_serializable	BaseMessage 是可序列化的。
get_lc_namespace	获取 LangChain 对象的命名空间。
__add__	将此消息与另一条消息连接。
pretty_repr	获取消息的精美表示。
pretty_print	打印消息的精美表示。

lc_secrets property

lc_secrets: dict[str, str]

构造函数参数名称到密钥 ID 的映射。

例如，{"openai_api_key": "OPENAI_API_KEY"}

lc_attributes property

lc_attributes: dict

应包含在序列化 kwargs 中的属性名称列表。

这些属性必须被构造函数接受。

默认为空字典。

content instance-attribute

content: str | list[str | dict]

消息的内容。

additional_kwargs class-attribute instance-attribute

additional_kwargs: dict = Field(default_factory=dict)

保留用于与消息关联的额外负载数据。

例如，对于来自 AI 的消息，这可以包括由模型提供商编码的工具调用。

response_metadata class-attribute instance-attribute

response_metadata: dict = Field(default_factory=dict)

示例：响应头、logprobs、词元计数、模型名称。

type instance-attribute

type: str

消息的类型。必须是该消息类型唯一的字符串。

此字段的目的是在反序列化消息时轻松识别消息类型。

name class-attribute instance-attribute

name: str | None = None

消息的可选名称。

这可用于为消息提供一个人类可读的名称。

此字段的使用是可选的，是否使用取决于模型实现。

id class-attribute instance-attribute

id: str | None = Field(default=None, coerce_numbers_to_str=True)

消息的可选唯一标识符。

理想情况下，这应该由创建消息的提供商/模型提供。

content_blocks property

content_blocks: list[ContentBlock]

从消息内容中加载内容块。

在 1.0.0 版本中添加

text property

text: TextAccessor

以字符串形式获取消息的文本内容。

既可以作为属性（message.text）使用，也可以作为方法（message.text()）使用。

已弃用

从 langchain-core 1.0.0 开始，将 .text() 作为方法调用已被弃用。请改用 .text 作为属性。此方法将在 2.0.0 版本中移除。

返回	描述
TextAccessor	消息的文本内容。

lc_id classmethod

lc_id() -> list[str]

为此类返回一个用于序列化目的的唯一标识符。

唯一标识符是一个描述对象路径的字符串列表。

例如，对于类 langchain.llms.openai.OpenAI，id 是 ["langchain", "llms", "openai", "OpenAI"]。

to_json

to_json() -> SerializedConstructor | SerializedNotImplemented

将对象序列化为 JSON。

引发	描述
ValueError	如果类有已弃用的属性。

返回	描述
SerializedConstructor | SerializedNotImplemented	一个可 JSON 序列化的对象或一个 SerializedNotImplemented 对象。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

序列化一个“未实现”的对象。

返回	描述
SerializedNotImplemented	SerializedNotImplemented.

__init__

__init__(
    content: str | list[str | dict] | None = None,
    content_blocks: list[ContentBlock] | None = None,
    **kwargs: Any,
) -> None

初始化一个 BaseMessage。

将 content 指定为位置参数，或为了类型提示指定 content_blocks。

参数	描述
content	消息的内容。 类型： str | list[str | dict] | None 默认值： None
content_blocks	类型化的标准内容。 类型： list[ContentBlock] | None 默认值： None
**kwargs	传递给父类的额外参数。 类型： Any 默认值： {}

is_lc_serializable classmethod

is_lc_serializable() -> bool

BaseMessage 是可序列化的。

返回	描述
bool	True

get_lc_namespace classmethod

get_lc_namespace() -> list[str]

获取 LangChain 对象的命名空间。

返回	描述
list[str]	["langchain", "schema", "messages"]

__add__

__add__(other: Any) -> ChatPromptTemplate

将此消息与另一条消息连接。

参数	描述
other	另一条要与此消息连接的消息。 类型： Any

返回	描述
ChatPromptTemplate	一个包含两条消息的 ChatPromptTemplate。

pretty_repr

pretty_repr(html: bool = False) -> str

获取消息的精美表示。

参数	描述
html	是否将消息格式化为 HTML。如果为 True，消息将使用 HTML 标签进行格式化。 类型： bool 默认值： False

返回	描述
str	消息的精美表示。

pretty_print

pretty_print() -> None

打印消息的精美表示。

BaseMessageChunk

基类：BaseMessage

消息块，可以与其他消息块连接。

方法	描述
__init__	初始化一个 BaseMessage。
is_lc_serializable	BaseMessage 是可序列化的。
get_lc_namespace	获取 LangChain 对象的命名空间。
lc_id	为此类返回一个用于序列化目的的唯一标识符。
to_json	将对象序列化为 JSON。
to_json_not_implemented	序列化一个“未实现”的对象。
pretty_repr	获取消息的精美表示。
pretty_print	打印消息的精美表示。
__add__	消息块支持与其他消息块的连接。

lc_secrets property

lc_secrets: dict[str, str]

构造函数参数名称到密钥 ID 的映射。

例如，{"openai_api_key": "OPENAI_API_KEY"}

lc_attributes property

lc_attributes: dict

应包含在序列化 kwargs 中的属性名称列表。

这些属性必须被构造函数接受。

默认为空字典。

content instance-attribute

content: str | list[str | dict]

消息的内容。

additional_kwargs class-attribute instance-attribute

additional_kwargs: dict = Field(default_factory=dict)

保留用于与消息关联的额外负载数据。

例如，对于来自 AI 的消息，这可以包括由模型提供商编码的工具调用。

response_metadata class-attribute instance-attribute

response_metadata: dict = Field(default_factory=dict)

示例：响应头、logprobs、词元计数、模型名称。

type instance-attribute

type: str

消息的类型。必须是该消息类型唯一的字符串。

此字段的目的是在反序列化消息时轻松识别消息类型。

name class-attribute instance-attribute

name: str | None = None

消息的可选名称。

这可用于为消息提供一个人类可读的名称。

此字段的使用是可选的，是否使用取决于模型实现。

id class-attribute instance-attribute

id: str | None = Field(default=None, coerce_numbers_to_str=True)

消息的可选唯一标识符。

理想情况下，这应该由创建消息的提供商/模型提供。

content_blocks property

content_blocks: list[ContentBlock]

从消息内容中加载内容块。

在 1.0.0 版本中添加

text property

text: TextAccessor

以字符串形式获取消息的文本内容。

既可以作为属性（message.text）使用，也可以作为方法（message.text()）使用。

已弃用

从 langchain-core 1.0.0 开始，将 .text() 作为方法调用已被弃用。请改用 .text 作为属性。此方法将在 2.0.0 版本中移除。

返回	描述
TextAccessor	消息的文本内容。

__init__

__init__(
    content: str | list[str | dict] | None = None,
    content_blocks: list[ContentBlock] | None = None,
    **kwargs: Any,
) -> None

初始化一个 BaseMessage。

将 content 指定为位置参数，或为了类型提示指定 content_blocks。

参数	描述
content	消息的内容。 类型： str | list[str | dict] | None 默认值： None
content_blocks	类型化的标准内容。 类型： list[ContentBlock] | None 默认值： None
**kwargs	传递给父类的额外参数。 类型： Any 默认值： {}

is_lc_serializable classmethod

is_lc_serializable() -> bool

BaseMessage 是可序列化的。

返回	描述
bool	True

get_lc_namespace classmethod

get_lc_namespace() -> list[str]

获取 LangChain 对象的命名空间。

返回	描述
list[str]	["langchain", "schema", "messages"]

lc_id classmethod

lc_id() -> list[str]

为此类返回一个用于序列化目的的唯一标识符。

唯一标识符是一个描述对象路径的字符串列表。

例如，对于类 langchain.llms.openai.OpenAI，id 是 ["langchain", "llms", "openai", "OpenAI"]。

to_json

to_json() -> SerializedConstructor | SerializedNotImplemented

将对象序列化为 JSON。

引发	描述
ValueError	如果类有已弃用的属性。

返回	描述
SerializedConstructor | SerializedNotImplemented	一个可 JSON 序列化的对象或一个 SerializedNotImplemented 对象。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

序列化一个“未实现”的对象。

返回	描述
SerializedNotImplemented	SerializedNotImplemented.

pretty_repr

pretty_repr(html: bool = False) -> str

获取消息的精美表示。

参数	描述
html	是否将消息格式化为 HTML。如果为 True，消息将使用 HTML 标签进行格式化。 类型： bool 默认值： False

返回	描述
str	消息的精美表示。

pretty_print

pretty_print() -> None

打印消息的精美表示。

__add__

__add__(other: Any) -> BaseMessageChunk

消息块支持与其他消息块的连接。

此功能对于将流式模型产生的消息块组合成一条完整的消息非常有用。

参数	描述
other	另一个要与此消息块连接的消息块。 类型： Any

返回	描述
BaseMessageChunk	一个新消息块，是此消息块
BaseMessageChunk	和另一个消息块的串联。

引发	描述
TypeError	如果另一个对象不是消息块。

例如，

AIMessageChunk(content="Hello") + AIMessageChunk(content=" World")

将得到 AIMessageChunk(content="Hello World")

langchain_core.language_models.fake_chat_models

用于测试目的的伪聊天模型。

GenericFakeChatModel

基类：BaseChatModel

通用的伪聊天模型，可用于测试聊天模型接口。

• 聊天模型应在同步和异步测试中都可用
• 调用 on_llm_new_token 以便测试与新词元相关的回调代码。
• 包含将消息分解为消息块的逻辑，以方便测试流式处理。

方法	描述
get_name	获取 Runnable 的名称。
get_input_schema	获取可用于验证 Runnable 输入的 Pydantic 模型。
get_input_jsonschema	获取表示 Runnable 输入的 JSON 模式。
get_output_schema	获取可用于验证 Runnable 输出的 Pydantic 模型。
get_output_jsonschema	获取表示 Runnable 输出的 JSON 模式。
config_schema	此 Runnable 接受的配置类型，指定为 Pydantic 模型。
get_config_jsonschema	获取表示 Runnable 配置的 JSON 模式。
get_graph	返回此 Runnable 的图形表示。
get_prompts	返回此 Runnable 使用的提示列表。
__or__	Runnable "or" 运算符。
__ror__	Runnable "reverse-or" 运算符。
pipe	管道连接 Runnable 对象。
pick	从此 Runnable 的输出 dict 中选择键。
assign	向此 Runnable 的 dict 输出分配新字段。
invoke	将单个输入转换为输出。
ainvoke	将单个输入转换为输出。
batch	默认实现使用线程池执行器并行运行 invoke。
batch_as_completed	在输入列表上并行运行 invoke。
abatch	默认实现使用 asyncio.gather 并行运行 ainvoke。
abatch_as_completed	在输入列表上并行运行 ainvoke。
stream	stream 的默认实现，它调用 invoke。
astream	astream 的默认实现，它调用 ainvoke。
astream_log	流式传输 Runnable 的所有输出，如回调系统所报告。
astream_events	生成事件流。
transform	将输入转换为输出。
atransform	将输入转换为输出。
bind	将参数绑定到 Runnable，返回一个新的 Runnable。
with_config	将配置绑定到 Runnable，返回一个新的 Runnable。
with_listeners	将生命周期侦听器绑定到 Runnable，返回一个新的 Runnable。
with_alisteners	将异步生命周期侦听器绑定到 Runnable。
with_types	将输入和输出类型绑定到 Runnable，返回一个新的 Runnable。
with_retry	创建一个新的 Runnable，它在发生异常时重试原始的 Runnable。
map	返回一个新的 Runnable，它将输入列表映射到输出列表。
with_fallbacks	向 Runnable 添加回退机制，返回一个新的 Runnable。
as_tool	从 Runnable 创建一个 BaseTool。
__init__
is_lc_serializable	这个类是否可序列化？
get_lc_namespace	获取 LangChain 对象的命名空间。
lc_id	为此类返回一个用于序列化目的的唯一标识符。
to_json	将 Runnable 序列化为 JSON。
to_json_not_implemented	序列化一个“未实现”的对象。
configurable_fields	在运行时配置特定的 Runnable 字段。
configurable_alternatives	为可在运行时设置的 Runnable 对象配置备选项。
set_verbose	如果 verbose 是 None，则设置它。
generate_prompt	将一系列提示传递给模型并返回模型生成的内容。
agenerate_prompt	异步地将一系列提示传递并返回模型生成的内容。
with_structured_output	返回与给定模式匹配的格式化输出的模型包装器。
get_token_ids	返回文本中 token 的有序 ID。
get_num_tokens	获取文本中存在的 token 数量。
get_num_tokens_from_messages	获取消息中的 token 数量。
generate	将一系列提示传递给模型并返回模型生成的内容。
agenerate	异步地将一系列提示传递给模型并返回生成的内容。
dict	返回 LLM 的字典。
bind_tools	将工具绑定到模型。

messages 实例属性

messages: Iterator[AIMessage | str]

获取一个消息的迭代器。

如果需要，可以扩展此接口以接受其他类型，如可调用对象/字典/字符串，使其更通用。

注意

如果你想传递一个列表，可以使用 iter 将其转换为迭代器。

警告

流式传输尚未实现。我们应该在未来尝试通过委托给 invoke 然后将结果输出分解成消息块来实现它。

name 类属性 实例属性

name: str | None = None

Runnable 的名称。用于调试和追踪。

InputType 属性

InputType: TypeAlias

获取此 Runnable 的输入类型。

OutputType 属性

OutputType: Any

获取此 Runnable 的输出类型。

input_schema 属性

input_schema: type[BaseModel]

此 Runnable 接受的输入类型，指定为 Pydantic 模型。

output_schema 属性

output_schema: type[BaseModel]

输出模式。

此 Runnable 产生的输出类型，指定为 Pydantic 模型。

config_specs 属性

config_specs: list[ConfigurableFieldSpec]

列出此 Runnable 的可配置字段。

lc_secrets 属性

lc_secrets: dict[str, str]

构造函数参数名称到密钥 ID 的映射。

例如，{"openai_api_key": "OPENAI_API_KEY"}

lc_attributes 属性

lc_attributes: dict

应包含在序列化 kwargs 中的属性名称列表。

这些属性必须被构造函数接受。

默认为空字典。

cache 类属性 实例属性

cache: BaseCache | bool | None = Field(default=None, exclude=True)

是否缓存响应。

• 如果为 True，将使用全局缓存。
• 如果为 False，将不使用缓存
• 如果为 None，如果设置了全局缓存，则使用全局缓存，否则不使用缓存。
• 如果是 BaseCache 的实例，将使用提供的缓存。

目前不支持模型的流式方法的缓存。

verbose 类属性 实例属性

verbose: bool = Field(default_factory=_get_verbosity, exclude=True, repr=False)

是否打印响应文本。

callbacks 类属性 实例属性

callbacks: Callbacks = Field(default=None, exclude=True)

添加到运行跟踪中的回调。

tags 类属性 实例属性

tags: list[str] | None = Field(default=None, exclude=True)

添加到运行跟踪中的标签。

metadata 类属性 实例属性

metadata: dict[str, Any] | None = Field(default=None, exclude=True)

添加到运行跟踪中的元数据。

custom_get_token_ids 类属性 实例属性

custom_get_token_ids: Callable[[str], list[int]] | None = Field(
    default=None, exclude=True
)

用于计算 token 的可选编码器。

rate_limiter 类属性 实例属性

rate_limiter: BaseRateLimiter | None = Field(default=None, exclude=True)

一个可选的速率限制器，用于限制请求数量。

disable_streaming 类属性 实例属性

disable_streaming: bool | Literal['tool_calling'] = False

是否为此模型禁用流式传输。

如果绕过流式传输，则 stream/astream/astream_events 将转而调用 invoke/ainvoke。

• 如果为 True，将始终绕过流式传输情况。
• 如果为 'tool_calling'，仅当模型被调用时带有 tools 关键字参数时，才会绕过流式传输情况。换句话说，仅当提供 tools 参数时，LangChain 才会自动切换到非流式行为 (invoke)。这提供了两全其美的方案。
• 如果为 False（默认值），则在可用时始终使用流式传输情况。

此标志的主要原因是，代码可能是使用 stream 编写的，而用户可能希望将给定模型换成另一个实现不完全支持流式传输的模型。

output_version 类属性 实例属性

output_version: str | None = Field(
    default_factory=from_env("LC_OUTPUT_VERSION", default=None)
)

要存储在消息内容中的 AIMessage 输出格式的版本。

AIMessage.content_blocks 将懒解析 content 的内容为标准格式。此标志可用于额外将标准格式存储在消息内容中，例如，用于序列化目的。

支持的值

• 'v0'：内容中特定于提供商的格式（可以使用 content_blocks 进行懒解析）
• 'v1'：内容中的标准化格式（与 content_blocks 一致）

合作伙伴包（例如 langchain-openai）也可以使用此字段以向后兼容的方式推出新的内容格式。

在版本 1.0 中添加

profile 属性

profile: ModelProfile

返回模型的性能分析信息。

此属性将依赖于 langchain-model-profiles 包来检索聊天模型的能力，例如上下文窗口大小和支持的功能。

引发	描述
ImportError	如果未安装 langchain-model-profiles。

返回	描述
ModelProfile	一个 ModelProfile 对象，包含模型的性能分析信息。

get_name

get_name(suffix: str | None = None, *, name: str | None = None) -> str

获取 Runnable 的名称。

参数	描述
后缀	一个可选的后缀，附加到名称上。 类型: str | None 默认值: None
name	一个可选的名称，用于代替 Runnable 的名称。 类型: str | None 默认值: None

返回	描述
str	Runnable 的名称。

get_input_schema

get_input_schema(config: RunnableConfig | None = None) -> type[BaseModel]

获取可用于验证 Runnable 输入的 Pydantic 模型。

利用 configurable_fields 和 configurable_alternatives 方法的 Runnable 对象将具有一个动态输入模式，该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输入模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证输入的 Pydantic 模型。

get_input_jsonschema

get_input_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

获取表示 Runnable 输入的 JSON 模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 输入的 JSON 模式。

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_input_jsonschema())

在 0.3.0 版本中新增。

get_output_schema

get_output_schema(config: RunnableConfig | None = None) -> type[BaseModel]

获取可用于验证 Runnable 输出的 Pydantic 模型。

利用 configurable_fields 和 configurable_alternatives 方法的 Runnable 对象将具有一个动态输出模式，该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输出模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证输出的 Pydantic 模型。

get_output_jsonschema

get_output_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

获取表示 Runnable 输出的 JSON 模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 输出的 JSON 模式。

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_output_jsonschema())

在 0.3.0 版本中新增。

config_schema

config_schema(*, include: Sequence[str] | None = None) -> type[BaseModel]

此 Runnable 接受的配置类型，指定为 Pydantic 模型。

要将字段标记为可配置，请参阅 configurable_fields 和 configurable_alternatives 方法。

参数	描述
包含	要包含在配置模式中的字段列表。 类型: Sequence[str] | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证配置的 Pydantic 模型。

get_config_jsonschema

get_config_jsonschema(*, include: Sequence[str] | None = None) -> dict[str, Any]

获取表示 Runnable 配置的 JSON 模式。

参数	描述
包含	要包含在配置模式中的字段列表。 类型: Sequence[str] | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 配置的 JSON 模式。

在 0.3.0 版本中新增。

get_graph

get_graph(config: RunnableConfig | None = None) -> Graph

返回此 Runnable 的图形表示。

get_prompts

get_prompts(config: RunnableConfig | None = None) -> list[BasePromptTemplate]

返回此 Runnable 使用的提示列表。

__or__

__or__(
    other: Runnable[Any, Other]
    | Callable[[Iterator[Any]], Iterator[Other]]
    | Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
    | Callable[[Any], Other]
    | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]

Runnable "or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence。

参数	描述
other	另一个 Runnable 或类 Runnable 对象。 类型: Runnable[Any, Other] | Callable[[Iterator[Any]], Iterator[Other]] | Callable[[AsyncIterator[Any]], AsyncIterator[Other]] | Callable[[Any], Other] | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any]

返回	描述
RunnableSerializable[Input, Other]	一个新的 Runnable。

__ror__

__ror__(
    other: Runnable[Other, Any]
    | Callable[[Iterator[Other]], Iterator[Any]]
    | Callable[[AsyncIterator[Other]], AsyncIterator[Any]]
    | Callable[[Other], Any]
    | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any],
) -> RunnableSerializable[Other, Output]

Runnable "reverse-or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence。

参数	描述
other	另一个 Runnable 或类 Runnable 对象。 类型: Runnable[Other, Any] | Callable[[Iterator[Other]], Iterator[Any]] | Callable[[AsyncIterator[Other]], AsyncIterator[Any]] | Callable[[Other], Any] | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any]

返回	描述
RunnableSerializable[Other, Output]	一个新的 Runnable。

pipe

pipe(
    *others: Runnable[Any, Other] | Callable[[Any], Other], name: str | None = None
) -> RunnableSerializable[Input, Other]

管道连接 Runnable 对象。

将此 Runnable 与类 Runnable 对象组合以构成一个 RunnableSequence。

等同于 RunnableSequence(self, *others) 或 self | others[0] | ...

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


def mul_two(x: int) -> int:
    return x * 2


runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1.pipe(runnable_2)
# Or equivalently:
# sequence = runnable_1 | runnable_2
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)
# -> 4

sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
# -> [4, 6, 8]

参数	描述
*其他	其他要组合的 Runnable 或类 Runnable 对象 类型: Runnable[Any, Other] | Callable[[Any], Other] 默认值: ()
name	生成的 RunnableSequence 的一个可选名称。 类型: str | None 默认值: None

返回	描述
RunnableSerializable[Input, Other]	一个新的 Runnable。

pick

pick(keys: str | list[str]) -> RunnableSerializable[Any, Any]

从此 Runnable 的输出 dict 中选择键。

选择单个键

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
chain = RunnableMap(str=as_str, json=as_json)

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3]}

json_only_chain = chain.pick("json")
json_only_chain.invoke("[1, 2, 3]")
# -> [1, 2, 3]

选择键列表

from typing import Any

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)


def as_bytes(x: Any) -> bytes:
    return bytes(x, "utf-8")


chain = RunnableMap(str=as_str, json=as_json, bytes=RunnableLambda(as_bytes))

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

json_and_bytes_chain = chain.pick(["json", "bytes"])
json_and_bytes_chain.invoke("[1, 2, 3]")
# -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

参数	描述
keys	从输出字典中选择的一个键或键列表。 类型: str | list[str]

返回	描述
RunnableSerializable[Any, Any]	一个新的 Runnable。

assign

assign(
    **kwargs: Runnable[dict[str, Any], Any]
    | Callable[[dict[str, Any]], Any]
    | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]],
) -> RunnableSerializable[Any, Any]

向此 Runnable 的 dict 输出分配新字段。

from langchain_community.llms.fake import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.runnables import Runnable
from operator import itemgetter

prompt = (
    SystemMessagePromptTemplate.from_template("You are a nice assistant.")
    + "{question}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])

chain: Runnable = prompt | model | {"str": StrOutputParser()}

chain_with_assign = chain.assign(hello=itemgetter("str") | model)

print(chain_with_assign.input_schema.model_json_schema())
# {'title': 'PromptInput', 'type': 'object', 'properties':
{'question': {'title': 'Question', 'type': 'string'}}}
print(chain_with_assign.output_schema.model_json_schema())
# {'title': 'RunnableSequenceOutput', 'type': 'object', 'properties':
{'str': {'title': 'Str',
'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}}

参数	描述
**kwargs	一个键到 Runnable 或类 Runnable 对象的映射，这些对象将使用此 Runnable 的整个输出字典来调用。 类型: Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any] | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]] 默认值: {}

返回	描述
RunnableSerializable[Any, Any]	一个新的 Runnable。

invoke

invoke(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AIMessage

将单个输入转换为输出。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | None 默认值: None

返回	描述
输出	Runnable 的输出。

ainvoke 异步

ainvoke(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AIMessage

将单个输入转换为输出。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | None 默认值: None

返回	描述
输出	Runnable 的输出。

batch

batch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

默认实现使用线程池执行器并行运行 invoke。

批处理的默认实现对于 IO 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理，则必须重写此方法；例如，如果底层的 Runnable 使用支持批处理模式的 API。

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | list[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

batch_as_completed

batch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> Iterator[tuple[int, Output | Exception]]

在输入列表上并行运行 invoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | Sequence[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
tuple[int, Output | Exception]	由输入索引和 Runnable 输出组成的元组。

abatch 异步

abatch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

默认实现使用 asyncio.gather 并行运行 ainvoke。

batch 的默认实现对于 IO 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理，则必须重写此方法；例如，如果底层的 Runnable 使用支持批处理模式的 API。

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | list[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

abatch_as_completed 异步

abatch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> AsyncIterator[tuple[int, Output | Exception]]

在输入列表上并行运行 ainvoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | Sequence[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[tuple[int, Output | Exception]]	一个由输入索引和 Runnable 输出组成的元组。

stream

stream(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> Iterator[AIMessageChunk]

stream 的默认实现，它调用 invoke。

如果子类支持流式输出，则必须重写此方法。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
输出	Runnable 的输出。

astream 异步

astream(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[AIMessageChunk]

astream 的默认实现，它调用 ainvoke。

如果子类支持流式输出，则必须重写此方法。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[Output]	Runnable 的输出。

astream_log 异步

astream_log(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    diff: bool = True,
    with_streamed_output_list: bool = True,
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]

流式传输 Runnable 的所有输出，如回调系统所报告。

这包括 LLM、检索器、工具等的所有内部运行。

输出以 Log 对象的形式流式传输，其中包括一个 Jsonpatch 操作列表，描述了运行状态在每一步中如何变化，以及运行的最终状态。

可以按顺序应用 Jsonpatch 操作来构造状态。

参数	描述
输入	Runnable 的输入。 类型： Any
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
差异	是生成每一步之间的差异还是当前状态。 类型: bool 默认值: True
带流式输出列表	是否生成 streamed_output 列表。 类型: bool 默认值: True
包含名称	仅包含具有这些名称的日志。 类型: Sequence[str] | None 默认值: None
包含类型	仅包含具有这些类型的日志。 类型: Sequence[str] | None 默认值: None
包含标签	仅包含具有这些标签的日志。 类型: Sequence[str] | None 默认值: None
排除名称	排除具有这些名称的日志。 类型: Sequence[str] | None 默认值: None
排除类型	排除具有这些类型的日志。 类型: Sequence[str] | None 默认值: None
排除标签	排除具有这些标签的日志。 类型: Sequence[str] | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型： Any 默认值： {}

YIELDS	描述
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]	一个 RunLogPatch 或 RunLog 对象。

astream_events 异步

astream_events(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    version: Literal["v1", "v2"] = "v2",
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]

生成事件流。

用于创建一个 StreamEvent 的迭代器，提供有关 Runnable 进度的实时信息，包括来自中间结果的 StreamEvent。

一个 StreamEvent 是一个具有以下模式的字典

• event：事件名称的格式为：on_[runnable_type]_(start|stream|end)。
• name：生成事件的 Runnable 的名称。
• run_id：与发出事件的 Runnable 的给定执行相关联的随机生成的 ID。作为父 Runnable 执行的一部分被调用的子 Runnable 被分配其自己的唯一 ID。
• parent_ids：生成事件的父可运行对象的 ID。根 Runnable 将有一个空列表。父 ID 的顺序是从根到直接父级。仅适用于 API 的 v2 版本。API 的 v1 版本将返回一个空列表。
• tags：生成事件的 Runnable 的标签。
• metadata：生成事件的 Runnable 的元数据。
• data：与事件关联的数据。此字段的内容取决于事件的类型。有关更多详细信息，请参见下表。

下表说明了各种链可能发出的某些事件。为简洁起见，已从表中省略了元数据字段。链定义已包含在表之后。

注意

此参考表适用于模式的 v2 版本。

事件	name	chunk	输入	output
on_chat_model_start	'[model name]'		{"messages": [[SystemMessage, HumanMessage]]}
on_chat_model_stream	'[model name]'	AIMessageChunk(content="hello")
on_chat_model_end	'[model name]'		{"messages": [[SystemMessage, HumanMessage]]}	AIMessageChunk(content="hello world")
on_llm_start	'[model name]'		{'input': 'hello'}
on_llm_stream	'[model name]'	'你好'
on_llm_end	'[model name]'		'你好，人类！'
on_chain_start	'format_docs'
on_chain_stream	'format_docs'	'hello world!, goodbye world!'
on_chain_end	'format_docs'		[Document(...)]	'hello world!, goodbye world!'
on_tool_start	'some_tool'		{"x": 1, "y": "2"}
on_tool_end	'some_tool'			{"x": 1, "y": "2"}
on_retriever_start	'[retriever name]'		{"query": "hello"}
on_retriever_end	'[retriever name]'		{"query": "hello"}	[Document(...), ..]
on_prompt_start	'[template_name]'		{"question": "hello"}
on_prompt_end	'[template_name]'		{"question": "hello"}	ChatPromptValue(messages: [SystemMessage, ...])

除了标准事件外，用户还可以分派自定义事件（见下例）。

自定义事件将仅在 API 的 v2 版本中出现！

自定义事件具有以下格式

属性	类型	描述
name	str	用户为事件定义的名称。
data	任意	与事件关联的数据。这可以是任何东西，但我们建议使其可 JSON 序列化。

以下是与上面显示的标准事件相关的声明

format_docs:

def format_docs(docs: list[Document]) -> str:
    '''Format the docs.'''
    return ", ".join([doc.page_content for doc in docs])


format_docs = RunnableLambda(format_docs)

some_tool:

@tool
def some_tool(x: int, y: str) -> dict:
    '''Some_tool.'''
    return {"x": x, "y": y}

prompt:

template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are Cat Agent 007"),
        ("human", "{question}"),
    ]
).with_config({"run_name": "my_template", "tags": ["my_template"]})

例如

from langchain_core.runnables import RunnableLambda


async def reverse(s: str) -> str:
    return s[::-1]


chain = RunnableLambda(func=reverse)

events = [event async for event in chain.astream_events("hello", version="v2")]

# Will produce the following events
# (run_id, and parent_ids has been omitted for brevity):
[
    {
        "data": {"input": "hello"},
        "event": "on_chain_start",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"chunk": "olleh"},
        "event": "on_chain_stream",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"output": "olleh"},
        "event": "on_chain_end",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
]

示例：分派自定义事件

from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio


async def slow_thing(some_input: str, config: RunnableConfig) -> str:
    """Do something that takes a long time."""
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 1 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 2 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    return "Done"

slow_thing = RunnableLambda(slow_thing)

async for event in slow_thing.astream_events("some_input", version="v2"):
    print(event)

参数	描述
输入	Runnable 的输入。 类型： Any
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
版本	要使用的模式版本，可以是 'v2' 或 'v1'。用户应使用 'v2'。'v1' 是为了向后兼容，并将在 0.4.0 中弃用。在 API 稳定之前不会分配默认值。自定义事件将仅在 'v2' 中出现。 类型: Literal['v1', 'v2'] 默认值: 'v2'
包含名称	仅包括来自具有匹配名称的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
包含类型	仅包括来自具有匹配类型的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
包含标签	仅包括来自具有匹配标签的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
排除名称	排除来自具有匹配名称的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
排除类型	排除来自具有匹配类型的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
排除标签	排除来自具有匹配标签的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。这些将传递给 astream_log，因为 astream_events 的此实现是建立在 astream_log 之上的。 类型： Any 默认值： {}

YIELDS	描述
AsyncIterator[StreamEvent]	StreamEvent 的异步流。

引发	描述
NotImplementedError	如果版本不是 'v1' 或 'v2'。

transform

transform(
    input: Iterator[Input], config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]

将输入转换为输出。

transform 的默认实现，它缓冲输入并调用 astream。

如果子类可以在输入仍在生成时开始产生输出，则必须重写此方法。

参数	描述
输入	Runnable 输入的迭代器。 类型: Iterator[Input]
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
输出	Runnable 的输出。

atransform 异步

atransform(
    input: AsyncIterator[Input],
    config: RunnableConfig | None = None,
    **kwargs: Any | None,
) -> AsyncIterator[Output]

将输入转换为输出。

atransform 的默认实现，它缓冲输入并调用 astream。

如果子类可以在输入仍在生成时开始产生输出，则必须重写此方法。

参数	描述
输入	Runnable 输入的异步迭代器。 类型： AsyncIterator[Input]
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[Output]	Runnable 的输出。

bind

bind(**kwargs: Any) -> Runnable[Input, Output]

将参数绑定到 Runnable，返回一个新的 Runnable。

当链中的 Runnable 需要一个不在前一个 Runnable 的输出中或未包含在用户输入中的参数时非常有用。

参数	描述
**kwargs	要绑定到 Runnable 的参数。 类型： Any 默认值： {}

返回	描述
Runnable[Input, Output]	一个绑定了参数的新 Runnable。

示例

from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser

model = ChatOllama(model="llama3.1")

# Without bind
chain = model | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'

# With bind
chain = model.bind(stop=["three"]) | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'

with_config

with_config(
    config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]

将配置绑定到 Runnable，返回一个新的 Runnable。

参数	描述
配置	要绑定到 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型： Any 默认值： {}

返回	描述
Runnable[Input, Output]	一个绑定了配置的新 Runnable。

with_listeners

with_listeners(
    *,
    on_start: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
    on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
    on_error: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
) -> Runnable[Input, Output]

将生命周期侦听器绑定到 Runnable，返回一个新的 Runnable。

Run 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_time 以及添加到运行中的任何标签或元数据。

参数	描述
开始时	在 Runnable 开始运行之前调用，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None
结束时	在 Runnable 完成运行后调用，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None
出错时	如果 Runnable 抛出错误，则调用此函数，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了监听器的新 Runnable。

示例

from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run

import time


def test_runnable(time_to_sleep: int):
    time.sleep(time_to_sleep)


def fn_start(run_obj: Run):
    print("start_time:", run_obj.start_time)


def fn_end(run_obj: Run):
    print("end_time:", run_obj.end_time)


chain = RunnableLambda(test_runnable).with_listeners(
    on_start=fn_start, on_end=fn_end
)
chain.invoke(2)

with_alisteners

with_alisteners(
    *,
    on_start: AsyncListener | None = None,
    on_end: AsyncListener | None = None,
    on_error: AsyncListener | None = None,
) -> Runnable[Input, Output]

将异步生命周期侦听器绑定到 Runnable。

返回一个新的 Runnable。

Run 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_time 以及添加到运行中的任何标签或元数据。

参数	描述
开始时	在 Runnable 开始运行之前异步调用，并传入 Run 对象。 类型： AsyncListener | None 默认值： None
结束时	在 Runnable 完成运行后异步调用，并传入 Run 对象。 类型： AsyncListener | None 默认值： None
出错时	如果 Runnable 抛出错误，则异步调用此函数，并传入 Run 对象。 类型： AsyncListener | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了监听器的新 Runnable。

示例

from langchain_core.runnables import RunnableLambda, Runnable
from datetime import datetime, timezone
import time
import asyncio

def format_t(timestamp: float) -> str:
    return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()

async def test_runnable(time_to_sleep: int):
    print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}")
    await asyncio.sleep(time_to_sleep)
    print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}")

async def fn_start(run_obj: Runnable):
    print(f"on start callback starts at {format_t(time.time())}")
    await asyncio.sleep(3)
    print(f"on start callback ends at {format_t(time.time())}")

async def fn_end(run_obj: Runnable):
    print(f"on end callback starts at {format_t(time.time())}")
    await asyncio.sleep(2)
    print(f"on end callback ends at {format_t(time.time())}")

runnable = RunnableLambda(test_runnable).with_alisteners(
    on_start=fn_start,
    on_end=fn_end
)
async def concurrent_runs():
    await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))

asyncio.run(concurrent_runs())
Result:
on start callback starts at 2025-03-01T07:05:22.875378+00:00
on start callback starts at 2025-03-01T07:05:22.875495+00:00
on start callback ends at 2025-03-01T07:05:25.878862+00:00
on start callback ends at 2025-03-01T07:05:25.878947+00:00
Runnable[2s]: starts at 2025-03-01T07:05:25.879392+00:00
Runnable[3s]: starts at 2025-03-01T07:05:25.879804+00:00
Runnable[2s]: ends at 2025-03-01T07:05:27.881998+00:00
on end callback starts at 2025-03-01T07:05:27.882360+00:00
Runnable[3s]: ends at 2025-03-01T07:05:28.881737+00:00
on end callback starts at 2025-03-01T07:05:28.882428+00:00
on end callback ends at 2025-03-01T07:05:29.883893+00:00
on end callback ends at 2025-03-01T07:05:30.884831+00:00

with_types

with_types(
    *, input_type: type[Input] | None = None, output_type: type[Output] | None = None
) -> Runnable[Input, Output]

将输入和输出类型绑定到 Runnable，返回一个新的 Runnable。

参数	描述
输入类型	要绑定到 Runnable 的输入类型。 类型： type[Input] | None 默认值： None
输出类型	要绑定到 Runnable 的输出类型。 类型： type[Output] | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了类型的新 Runnable。

with_retry

with_retry(
    *,
    retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
    wait_exponential_jitter: bool = True,
    exponential_jitter_params: ExponentialJitterParams | None = None,
    stop_after_attempt: int = 3,
) -> Runnable[Input, Output]

创建一个新的 Runnable，它在发生异常时重试原始的 Runnable。

参数	描述
如果异常类型则重试	一个用于重试的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
指数等待抖动	是否在两次重试之间的等待时间中添加抖动。 类型: bool 默认值: True
尝试后停止	放弃前尝试的最大次数。 类型： int 默认值： 3
指数抖动参数	tenacity.wait_exponential_jitter 的参数。即：initial、max、exp_base 和 jitter（均为 float 值）。 类型： ExponentialJitterParams | None 默认值： None

返回	描述
Runnable[Input, Output]	一个新的 Runnable，它会在发生异常时重试原始的 Runnable。

示例

from langchain_core.runnables import RunnableLambda

count = 0


def _lambda(x: int) -> None:
    global count
    count = count + 1
    if x == 1:
        raise ValueError("x is 1")
    else:
        pass


runnable = RunnableLambda(_lambda)
try:
    runnable.with_retry(
        stop_after_attempt=2,
        retry_if_exception_type=(ValueError,),
    ).invoke(1)
except ValueError:
    pass

assert count == 2

map

map() -> Runnable[list[Input], list[Output]]

返回一个新的 Runnable，它将输入列表映射到输出列表。

使用每个输入调用 invoke。

返回	描述
Runnable[list[Input], list[Output]]	一个新的 Runnable，它将输入列表映射到输出列表。

示例

from langchain_core.runnables import RunnableLambda


def _lambda(x: int) -> int:
    return x + 1


runnable = RunnableLambda(_lambda)
print(runnable.map().invoke([1, 2, 3]))  # [2, 3, 4]

with_fallbacks

with_fallbacks(
    fallbacks: Sequence[Runnable[Input, Output]],
    *,
    exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
    exception_key: str | None = None,
) -> RunnableWithFallbacks[Input, Output]

向 Runnable 添加回退机制，返回一个新的 Runnable。

新的 Runnable 将在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

参数	描述
备用方案	如果原始 Runnable 失败，将尝试的一系列 runnable。 类型： Sequence[Runnable[Input, Output]]
要处理的异常	一个要处理的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
异常键	如果指定了 string，则已处理的异常将作为输入的一部分，在指定的键下传递给备选方案。如果为 None，异常将不会传递给备选方案。如果使用，基础 Runnable 及其备选方案必须接受一个字典作为输入。 类型: str | None 默认值: None

返回	描述
RunnableWithFallbacks[Input, Output]	一个新的 Runnable，它会在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

示例

from typing import Iterator

from langchain_core.runnables import RunnableGenerator


def _generate_immediate_error(input: Iterator) -> Iterator[str]:
    raise ValueError()
    yield ""


def _generate(input: Iterator) -> Iterator[str]:
    yield from "foo bar"


runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
    [RunnableGenerator(_generate)]
)
print("".join(runnable.stream({})))  # foo bar

参数	描述
备用方案	如果原始 Runnable 失败，将尝试的一系列 runnable。 类型： Sequence[Runnable[Input, Output]]
要处理的异常	一个要处理的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
异常键	如果指定了 string，则已处理的异常将作为输入的一部分，在指定的键下传递给备选方案。如果为 None，异常将不会传递给备选方案。如果使用，基础 Runnable 及其备选方案必须接受一个字典作为输入。 类型: str | None 默认值: None

返回	描述
RunnableWithFallbacks[Input, Output]	一个新的 Runnable，它会在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

as_tool

as_tool(
    args_schema: type[BaseModel] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    arg_types: dict[str, type] | None = None,
) -> BaseTool

从 Runnable 创建一个 BaseTool。

as_tool 将从一个 Runnable 实例化一个 BaseTool，该工具具有名称、描述和 args_schema。在可能的情况下，模式会从 runnable.get_input_schema 中推断。或者（例如，如果 Runnable 接受一个字典作为输入，并且特定的字典键没有类型），模式可以通过 args_schema 直接指定。你也可以传递 arg_types 来仅指定必需的参数及其类型。

参数	描述
参数模式	工具的模式。 类型： type[BaseModel] | None 默认值： None
name	工具的名称。 类型: str | None 默认值: None
描述	工具的描述。 类型: str | None 默认值: None
参数类型	一个从参数名称到类型的字典。 类型： dict[str, type] | None 默认值： None

返回	描述
BaseTool	一个 BaseTool 实例。

类型化字典输入

from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda


class Args(TypedDict):
    a: int
    b: list[int]


def f(x: Args) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool()
as_tool.invoke({"a": 3, "b": [1, 2]})

dict 输入，通过 args_schema 指定模式

from typing import Any
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableLambda

def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))

class FSchema(BaseModel):
    """Apply a function to an integer and list of integers."""

    a: int = Field(..., description="Integer")
    b: list[int] = Field(..., description="List of ints")

runnable = RunnableLambda(f)
as_tool = runnable.as_tool(FSchema)
as_tool.invoke({"a": 3, "b": [1, 2]})

dict 输入，通过 arg_types 指定模式

from typing import Any
from langchain_core.runnables import RunnableLambda


def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool(arg_types={"a": int, "b": list[int]})
as_tool.invoke({"a": 3, "b": [1, 2]})

字符串输入

from langchain_core.runnables import RunnableLambda


def f(x: str) -> str:
    return x + "a"


def g(x: str) -> str:
    return x + "z"


runnable = RunnableLambda(f) | g
as_tool = runnable.as_tool()
as_tool.invoke("b")

__init__

__init__(*args: Any, **kwargs: Any) -> None

is_lc_serializable 类方法

is_lc_serializable() -> bool

这个类是否可序列化？

根据设计，即使一个类继承自 Serializable，它默认也是不可序列化的。这是为了防止意外序列化不应被序列化的对象。

返回	描述
bool	类是否可序列化。默认为 False。

get_lc_namespace 类方法

get_lc_namespace() -> list[str]

获取 LangChain 对象的命名空间。

例如，如果类是 langchain.llms.openai.OpenAI，那么命名空间是 ["langchain", "llms", "openai"]

返回	描述
list[str]	命名空间。

lc_id 类方法

lc_id() -> list[str]

为此类返回一个用于序列化目的的唯一标识符。

唯一标识符是一个描述对象路径的字符串列表。

例如，对于类 langchain.llms.openai.OpenAI，id 是 ["langchain", "llms", "openai", "OpenAI"]。

to_json

to_json() -> SerializedConstructor | SerializedNotImplemented

将 Runnable 序列化为 JSON。

返回	描述
SerializedConstructor | SerializedNotImplemented	一个 Runnable 的 JSON 可序列化表示。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

序列化一个“未实现”的对象。

返回	描述
SerializedNotImplemented	SerializedNotImplemented.

configurable_fields

configurable_fields(
    **kwargs: AnyConfigurableField,
) -> RunnableSerializable[Input, Output]

在运行时配置特定的 Runnable 字段。

参数	描述
**kwargs	一个要配置的 ConfigurableField 实例的字典。 类型： AnyConfigurableField 默认值： {}

引发	描述
ValueError	如果在 Runnable 中找不到配置键。

返回	描述
RunnableSerializable[Input, Output]	一个配置了字段的新 Runnable。

from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatOpenAI(max_tokens=20).configurable_fields(
    max_tokens=ConfigurableField(
        id="output_token_number",
        name="Max tokens in the output",
        description="The maximum number of tokens in the output",
    )
)

# max_tokens = 20
print("max_tokens_20: ", model.invoke("tell me something about chess").content)

# max_tokens = 200
print(
    "max_tokens_200: ",
    model.with_config(configurable={"output_token_number": 200})
    .invoke("tell me something about chess")
    .content,
)

configurable_alternatives

configurable_alternatives(
    which: ConfigurableField,
    *,
    default_key: str = "default",
    prefix_keys: bool = False,
    **kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]],
) -> RunnableSerializable[Input, Output]

为可在运行时设置的 Runnable 对象配置备选项。

参数	描述
哪个	将用于选择备选项的 ConfigurableField 实例。 类型： ConfigurableField
默认键	如果未选择备选项，则使用的默认键。 类型： str 默认值： 'default'
前缀键	是否用 ConfigurableField id 作为键的前缀。 类型： bool 默认值： False
**kwargs	一个从键到 Runnable 实例或返回 Runnable 实例的可调用对象的字典。 类型： Runnable[Input, Output] | Callable[[], Runnable[Input, Output]] 默认值： {}

返回	描述
RunnableSerializable[Input, Output]	一个配置了备选项的新 Runnable。

from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatAnthropic(
    model_name="claude-sonnet-4-5-20250929"
).configurable_alternatives(
    ConfigurableField(id="llm"),
    default_key="anthropic",
    openai=ChatOpenAI(),
)

# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)

# uses ChatOpenAI
print(
    model.with_config(configurable={"llm": "openai"})
    .invoke("which organization created you?")
    .content
)

set_verbose

set_verbose(verbose: bool | None) -> bool

如果 verbose 是 None，则设置它。

这允许用户传入 None 作为 verbose 来访问全局设置。

参数	描述
verbose	要使用的详细程度设置。 类型： bool | None

返回	描述
bool	要使用的详细程度设置。

generate_prompt

generate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult

将一系列提示传递给模型并返回模型生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
prompts	PromptValue 对象列表。PromptValue 是一个可以转换为匹配任何语言模型格式的对象（对于纯文本生成模型是字符串，对于聊天模型是 BaseMessage 对象）。 类型： list[PromptValue]
停止	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
回调	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generation 对象列表以及额外的模型提供商特定的输出。

agenerate_prompt 异步

agenerate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult

异步地将一系列提示传递并返回模型生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
prompts	PromptValue 对象列表。PromptValue 是一个可以转换为匹配任何语言模型格式的对象（对于纯文本生成模型是字符串，对于聊天模型是 BaseMessage 对象）。 类型： list[PromptValue]
停止	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
回调	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generation 对象列表以及额外的模型提供商特定的输出。

with_structured_output

with_structured_output(
    schema: Dict | type, *, include_raw: bool = False, **kwargs: Any
) -> Runnable[LanguageModelInput, Dict | BaseModel]

返回与给定模式匹配的格式化输出的模型包装器。

参数	描述
模式	输出模式。可以作为以下形式传入： 一个 OpenAI 函数/工具模式， 一个 JSON Schema， 一个 TypedDict 类， 或一个 Pydantic 类。 如果 schema 是一个 Pydantic 类，那么模型输出将是该类的 Pydantic 实例，并且模型生成的字段将由 Pydantic 类进行验证。否则，模型输出将是一个字典，并且不会被验证。 有关在指定 Pydantic 或 TypedDict 类时如何正确指定模式字段的类型和描述的更多信息，请参阅 langchain_core.utils.function_calling.convert_to_openai_tool。 类型： Dict | type
包含原始数据	如果为 False，则仅返回解析后的结构化输出。如果在模型输出解析期间发生错误，它将被引发。如果为 True，则将返回原始模型响应（一个 BaseMessage）和解析后的模型响应。如果在输出解析期间发生错误，它将被捕获并一并返回。 最终输出总是一个带有键 'raw'、'parsed' 和 'parsing_error' 的 dict。 类型： bool 默认值： False

引发	描述
ValueError	如果有任何不支持的 kwargs。
NotImplementedError	如果模型没有实现 with_structured_output()。

返回	描述
Runnable[LanguageModelInput, Dict | BaseModel]	一个接受与 langchain_core.language_models.chat.BaseChatModel 相同输入的 Runnable。如果 include_raw 为 False 且 schema 是一个 Pydantic 类，则 Runnable 输出一个 schema 的实例（即一个 Pydantic 对象）。否则，如果 include_raw 为 False，则 Runnable 输出一个 dict。 如果 include_raw 为 True，则 Runnable 输出一个带有以下键的 dict 'raw': BaseMessage 'parsed': 如果出现解析错误则为 None，否则类型取决于上面描述的 schema。 'parsing_error': BaseException | None

示例：Pydantic 模式 (include_raw=False)

from pydantic import BaseModel


class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''

    answer: str
    justification: str


model = ChatModel(model="model-name", temperature=0)
structured_model = model.with_structured_output(AnswerWithJustification)

structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)

# -> AnswerWithJustification(
#     answer='They weigh the same',
#     justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'
# )

示例：Pydantic 模式 (include_raw=True)

from pydantic import BaseModel


class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''

    answer: str
    justification: str


model = ChatModel(model="model-name", temperature=0)
structured_model = model.with_structured_output(
    AnswerWithJustification, include_raw=True
)

structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)
# -> {
#     'raw': AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_Ao02pnFYXD6GN1yzc0uXPsvF', 'function': {'arguments': '{"answer":"They weigh the same.","justification":"Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ."}', 'name': 'AnswerWithJustification'}, 'type': 'function'}]}),
#     'parsed': AnswerWithJustification(answer='They weigh the same.', justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'),
#     'parsing_error': None
# }

示例：dict 模式 (include_raw=False)

from pydantic import BaseModel
from langchain_core.utils.function_calling import convert_to_openai_tool


class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''

    answer: str
    justification: str


dict_schema = convert_to_openai_tool(AnswerWithJustification)
model = ChatModel(model="model-name", temperature=0)
structured_model = model.with_structured_output(dict_schema)

structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)
# -> {
#     'answer': 'They weigh the same',
#     'justification': 'Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume and density of the two substances differ.'
# }

行为在 0.2.26 版本中已更改

添加了对 TypedDict 类的支持。

get_token_ids

get_token_ids(text: str) -> list[int]

返回文本中 token 的有序 ID。

参数	描述
text	要进行分词的字符串输入。 类型： str

返回	描述
list[int]	与文本中的词元相对应的 ID 列表，按它们在文本中出现的顺序列出。

get_num_tokens

get_num_tokens(text: str) -> int

获取文本中存在的 token 数量。

用于检查输入是否适合模型的上下文窗口。

参数	描述
text	要进行分词的字符串输入。 类型： str

返回	描述
int	文本中的词元整数数量。

get_num_tokens_from_messages

get_num_tokens_from_messages(
    messages: list[BaseMessage], tools: Sequence | None = None
) -> int

获取消息中的 token 数量。

用于检查输入是否适合模型的上下文窗口。

注意

get_num_tokens_from_messages 的基本实现忽略了工具模式。

参数	描述
messages	要进行分词的消息输入。 类型： list[BaseMessage]
工具	如果提供，则为要转换为工具模式的 dict、BaseModel、函数或 BaseTool 对象的序列。 类型： Sequence | None 默认值： None

返回	描述
int	所有消息中词元数量的总和。

generate

generate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: UUID | None = None,
    **kwargs: Any,
) -> LLMResult

将一系列提示传递给模型并返回模型生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
messages	消息列表的列表。 类型： list[list[BaseMessage]]
停止	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
回调	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
tags	要应用的标签。 类型: list[str] | None 默认值: None
metadata	要应用的元数据。 类型： dict[str, Any] | None 默认值： None
运行名称	运行的名称。 类型: str | None 默认值: None
run_id	运行的 ID。 类型: UUID | None 默认值: None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generations 列表以及额外的模型提供商特定的输出。

agenerate 异步

agenerate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: UUID | None = None,
    **kwargs: Any,
) -> LLMResult

异步地将一系列提示传递给模型并返回生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
messages	消息列表的列表。 类型： list[list[BaseMessage]]
停止	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
回调	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
tags	要应用的标签。 类型: list[str] | None 默认值: None
metadata	要应用的元数据。 类型： dict[str, Any] | None 默认值： None
运行名称	运行的名称。 类型: str | None 默认值: None
run_id	运行的 ID。 类型: UUID | None 默认值: None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generations 列表以及额外的模型提供商特定的输出。

dict

dict(**kwargs: Any) -> dict

返回 LLM 的字典。

bind_tools

bind_tools(
    tools: Sequence[Dict[str, Any] | type | Callable | BaseTool],
    *,
    tool_choice: str | None = None,
    **kwargs: Any,
) -> Runnable[LanguageModelInput, AIMessage]

将工具绑定到模型。

参数	描述
工具	要绑定到模型的工具序列。 类型： Sequence[Dict[str, Any] | type | Callable | BaseTool]
工具选择	要使用的工具。如果为 "any"，则可以使用任何工具。 类型: str | None 默认值: None

返回	描述
Runnable[LanguageModelInput, AIMessage]	一个返回消息的 Runnable。

ParrotFakeChatModel

基类：BaseChatModel

通用的伪聊天模型，可用于测试聊天模型接口。

• 聊天模型应在同步和异步测试中都可用

方法	描述
get_name	获取 Runnable 的名称。
get_input_schema	获取可用于验证 Runnable 输入的 Pydantic 模型。
get_input_jsonschema	获取表示 Runnable 输入的 JSON 模式。
get_output_schema	获取可用于验证 Runnable 输出的 Pydantic 模型。
get_output_jsonschema	获取表示 Runnable 输出的 JSON 模式。
config_schema	此 Runnable 接受的配置类型，指定为 Pydantic 模型。
get_config_jsonschema	获取表示 Runnable 配置的 JSON 模式。
get_graph	返回此 Runnable 的图形表示。
get_prompts	返回此 Runnable 使用的提示列表。
__or__	Runnable "or" 运算符。
__ror__	Runnable "reverse-or" 运算符。
pipe	管道连接 Runnable 对象。
pick	从此 Runnable 的输出 dict 中选择键。
assign	向此 Runnable 的 dict 输出分配新字段。
invoke	将单个输入转换为输出。
ainvoke	将单个输入转换为输出。
batch	默认实现使用线程池执行器并行运行 invoke。
batch_as_completed	在输入列表上并行运行 invoke。
abatch	默认实现使用 asyncio.gather 并行运行 ainvoke。
abatch_as_completed	在输入列表上并行运行 ainvoke。
stream	stream 的默认实现，它调用 invoke。
astream	astream 的默认实现，它调用 ainvoke。
astream_log	流式传输 Runnable 的所有输出，如回调系统所报告。
astream_events	生成事件流。
transform	将输入转换为输出。
atransform	将输入转换为输出。
bind	将参数绑定到 Runnable，返回一个新的 Runnable。
with_config	将配置绑定到 Runnable，返回一个新的 Runnable。
with_listeners	将生命周期侦听器绑定到 Runnable，返回一个新的 Runnable。
with_alisteners	将异步生命周期侦听器绑定到 Runnable。
with_types	将输入和输出类型绑定到 Runnable，返回一个新的 Runnable。
with_retry	创建一个新的 Runnable，它在发生异常时重试原始的 Runnable。
map	返回一个新的 Runnable，它将输入列表映射到输出列表。
with_fallbacks	向 Runnable 添加回退机制，返回一个新的 Runnable。
as_tool	从 Runnable 创建一个 BaseTool。
__init__
is_lc_serializable	这个类是否可序列化？
get_lc_namespace	获取 LangChain 对象的命名空间。
lc_id	为此类返回一个用于序列化目的的唯一标识符。
to_json	将 Runnable 序列化为 JSON。
to_json_not_implemented	序列化一个“未实现”的对象。
configurable_fields	在运行时配置特定的 Runnable 字段。
configurable_alternatives	为可在运行时设置的 Runnable 对象配置备选项。
set_verbose	如果 verbose 是 None，则设置它。
generate_prompt	将一系列提示传递给模型并返回模型生成的内容。
agenerate_prompt	异步地将一系列提示传递并返回模型生成的内容。
with_structured_output	返回与给定模式匹配的格式化输出的模型包装器。
get_token_ids	返回文本中 token 的有序 ID。
get_num_tokens	获取文本中存在的 token 数量。
get_num_tokens_from_messages	获取消息中的 token 数量。
generate	将一系列提示传递给模型并返回模型生成的内容。
agenerate	异步地将一系列提示传递给模型并返回生成的内容。
dict	返回 LLM 的字典。
bind_tools	将工具绑定到模型。

name 类属性 实例属性

name: str | None = None

Runnable 的名称。用于调试和追踪。

InputType 属性

InputType: TypeAlias

获取此 Runnable 的输入类型。

OutputType 属性

OutputType: Any

获取此 Runnable 的输出类型。

input_schema 属性

input_schema: type[BaseModel]

此 Runnable 接受的输入类型，指定为 Pydantic 模型。

output_schema 属性

output_schema: type[BaseModel]

输出模式。

此 Runnable 产生的输出类型，指定为 Pydantic 模型。

config_specs 属性

config_specs: list[ConfigurableFieldSpec]

列出此 Runnable 的可配置字段。

lc_secrets 属性

lc_secrets: dict[str, str]

构造函数参数名称到密钥 ID 的映射。

例如，{"openai_api_key": "OPENAI_API_KEY"}

lc_attributes 属性

lc_attributes: dict

应包含在序列化 kwargs 中的属性名称列表。

这些属性必须被构造函数接受。

默认为空字典。

cache 类属性 实例属性

cache: BaseCache | bool | None = Field(default=None, exclude=True)

是否缓存响应。

• 如果为 True，将使用全局缓存。
• 如果为 False，将不使用缓存
• 如果为 None，如果设置了全局缓存，则使用全局缓存，否则不使用缓存。
• 如果是 BaseCache 的实例，将使用提供的缓存。

目前不支持模型的流式方法的缓存。

verbose 类属性 实例属性

verbose: bool = Field(default_factory=_get_verbosity, exclude=True, repr=False)

是否打印响应文本。

callbacks 类属性 实例属性

callbacks: Callbacks = Field(default=None, exclude=True)

添加到运行跟踪中的回调。

tags 类属性 实例属性

tags: list[str] | None = Field(default=None, exclude=True)

添加到运行跟踪中的标签。

metadata 类属性 实例属性

metadata: dict[str, Any] | None = Field(default=None, exclude=True)

添加到运行跟踪中的元数据。

custom_get_token_ids 类属性 实例属性

custom_get_token_ids: Callable[[str], list[int]] | None = Field(
    default=None, exclude=True
)

用于计算 token 的可选编码器。

rate_limiter 类属性 实例属性

rate_limiter: BaseRateLimiter | None = Field(default=None, exclude=True)

一个可选的速率限制器，用于限制请求数量。

disable_streaming 类属性 实例属性

disable_streaming: bool | Literal['tool_calling'] = False

是否为此模型禁用流式传输。

如果绕过流式传输，则 stream/astream/astream_events 将转而调用 invoke/ainvoke。

• 如果为 True，将始终绕过流式传输情况。
• 如果为 'tool_calling'，仅当模型被调用时带有 tools 关键字参数时，才会绕过流式传输情况。换句话说，仅当提供 tools 参数时，LangChain 才会自动切换到非流式行为 (invoke)。这提供了两全其美的方案。
• 如果为 False（默认值），则在可用时始终使用流式传输情况。

此标志的主要原因是，代码可能是使用 stream 编写的，而用户可能希望将给定模型换成另一个实现不完全支持流式传输的模型。

output_version 类属性 实例属性

output_version: str | None = Field(
    default_factory=from_env("LC_OUTPUT_VERSION", default=None)
)

要存储在消息内容中的 AIMessage 输出格式的版本。

AIMessage.content_blocks 将懒解析 content 的内容为标准格式。此标志可用于额外将标准格式存储在消息内容中，例如，用于序列化目的。

支持的值

• 'v0'：内容中特定于提供商的格式（可以使用 content_blocks 进行懒解析）
• 'v1'：内容中的标准化格式（与 content_blocks 一致）

合作伙伴包（例如 langchain-openai）也可以使用此字段以向后兼容的方式推出新的内容格式。

在版本 1.0 中添加

profile 属性

profile: ModelProfile

返回模型的性能分析信息。

此属性将依赖于 langchain-model-profiles 包来检索聊天模型的能力，例如上下文窗口大小和支持的功能。

引发	描述
ImportError	如果未安装 langchain-model-profiles。

返回	描述
ModelProfile	一个 ModelProfile 对象，包含模型的性能分析信息。

get_name

get_name(suffix: str | None = None, *, name: str | None = None) -> str

获取 Runnable 的名称。

参数	描述
后缀	一个可选的后缀，附加到名称上。 类型: str | None 默认值: None
name	一个可选的名称，用于代替 Runnable 的名称。 类型: str | None 默认值: None

返回	描述
str	Runnable 的名称。

get_input_schema

get_input_schema(config: RunnableConfig | None = None) -> type[BaseModel]

获取可用于验证 Runnable 输入的 Pydantic 模型。

利用 configurable_fields 和 configurable_alternatives 方法的 Runnable 对象将具有一个动态输入模式，该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输入模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证输入的 Pydantic 模型。

get_input_jsonschema

get_input_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

获取表示 Runnable 输入的 JSON 模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 输入的 JSON 模式。

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_input_jsonschema())

在 0.3.0 版本中新增。

get_output_schema

get_output_schema(config: RunnableConfig | None = None) -> type[BaseModel]

获取可用于验证 Runnable 输出的 Pydantic 模型。

利用 configurable_fields 和 configurable_alternatives 方法的 Runnable 对象将具有一个动态输出模式，该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输出模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证输出的 Pydantic 模型。

get_output_jsonschema

get_output_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

获取表示 Runnable 输出的 JSON 模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 输出的 JSON 模式。

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_output_jsonschema())

在 0.3.0 版本中新增。

config_schema

config_schema(*, include: Sequence[str] | None = None) -> type[BaseModel]

此 Runnable 接受的配置类型，指定为 Pydantic 模型。

要将字段标记为可配置，请参阅 configurable_fields 和 configurable_alternatives 方法。

参数	描述
包含	要包含在配置模式中的字段列表。 类型: Sequence[str] | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证配置的 Pydantic 模型。

get_config_jsonschema

get_config_jsonschema(*, include: Sequence[str] | None = None) -> dict[str, Any]

获取表示 Runnable 配置的 JSON 模式。

参数	描述
包含	要包含在配置模式中的字段列表。 类型: Sequence[str] | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 配置的 JSON 模式。

在 0.3.0 版本中新增。

get_graph

get_graph(config: RunnableConfig | None = None) -> Graph

返回此 Runnable 的图形表示。

get_prompts

get_prompts(config: RunnableConfig | None = None) -> list[BasePromptTemplate]

返回此 Runnable 使用的提示列表。

__or__

__or__(
    other: Runnable[Any, Other]
    | Callable[[Iterator[Any]], Iterator[Other]]
    | Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
    | Callable[[Any], Other]
    | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]

Runnable "or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence。

参数	描述
other	另一个 Runnable 或类 Runnable 对象。 类型: Runnable[Any, Other] | Callable[[Iterator[Any]], Iterator[Other]] | Callable[[AsyncIterator[Any]], AsyncIterator[Other]] | Callable[[Any], Other] | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any]

返回	描述
RunnableSerializable[Input, Other]	一个新的 Runnable。

__ror__

__ror__(
    other: Runnable[Other, Any]
    | Callable[[Iterator[Other]], Iterator[Any]]
    | Callable[[AsyncIterator[Other]], AsyncIterator[Any]]
    | Callable[[Other], Any]
    | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any],
) -> RunnableSerializable[Other, Output]

Runnable "reverse-or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence。

参数	描述
other	另一个 Runnable 或类 Runnable 对象。 类型: Runnable[Other, Any] | Callable[[Iterator[Other]], Iterator[Any]] | Callable[[AsyncIterator[Other]], AsyncIterator[Any]] | Callable[[Other], Any] | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any]

返回	描述
RunnableSerializable[Other, Output]	一个新的 Runnable。

pipe

pipe(
    *others: Runnable[Any, Other] | Callable[[Any], Other], name: str | None = None
) -> RunnableSerializable[Input, Other]

管道连接 Runnable 对象。

将此 Runnable 与类 Runnable 对象组合以构成一个 RunnableSequence。

等同于 RunnableSequence(self, *others) 或 self | others[0] | ...

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


def mul_two(x: int) -> int:
    return x * 2


runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1.pipe(runnable_2)
# Or equivalently:
# sequence = runnable_1 | runnable_2
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)
# -> 4

sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
# -> [4, 6, 8]

参数	描述
*其他	其他要组合的 Runnable 或类 Runnable 对象 类型: Runnable[Any, Other] | Callable[[Any], Other] 默认值: ()
name	生成的 RunnableSequence 的一个可选名称。 类型: str | None 默认值: None

返回	描述
RunnableSerializable[Input, Other]	一个新的 Runnable。

pick

pick(keys: str | list[str]) -> RunnableSerializable[Any, Any]

从此 Runnable 的输出 dict 中选择键。

选择单个键

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
chain = RunnableMap(str=as_str, json=as_json)

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3]}

json_only_chain = chain.pick("json")
json_only_chain.invoke("[1, 2, 3]")
# -> [1, 2, 3]

选择键列表

from typing import Any

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)


def as_bytes(x: Any) -> bytes:
    return bytes(x, "utf-8")


chain = RunnableMap(str=as_str, json=as_json, bytes=RunnableLambda(as_bytes))

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

json_and_bytes_chain = chain.pick(["json", "bytes"])
json_and_bytes_chain.invoke("[1, 2, 3]")
# -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

参数	描述
keys	从输出字典中选择的一个键或键列表。 类型: str | list[str]

返回	描述
RunnableSerializable[Any, Any]	一个新的 Runnable。

assign

assign(
    **kwargs: Runnable[dict[str, Any], Any]
    | Callable[[dict[str, Any]], Any]
    | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]],
) -> RunnableSerializable[Any, Any]

向此 Runnable 的 dict 输出分配新字段。

from langchain_community.llms.fake import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.runnables import Runnable
from operator import itemgetter

prompt = (
    SystemMessagePromptTemplate.from_template("You are a nice assistant.")
    + "{question}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])

chain: Runnable = prompt | model | {"str": StrOutputParser()}

chain_with_assign = chain.assign(hello=itemgetter("str") | model)

print(chain_with_assign.input_schema.model_json_schema())
# {'title': 'PromptInput', 'type': 'object', 'properties':
{'question': {'title': 'Question', 'type': 'string'}}}
print(chain_with_assign.output_schema.model_json_schema())
# {'title': 'RunnableSequenceOutput', 'type': 'object', 'properties':
{'str': {'title': 'Str',
'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}}

参数	描述
**kwargs	一个键到 Runnable 或类 Runnable 对象的映射，这些对象将使用此 Runnable 的整个输出字典来调用。 类型: Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any] | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]] 默认值: {}

返回	描述
RunnableSerializable[Any, Any]	一个新的 Runnable。

invoke

invoke(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AIMessage

将单个输入转换为输出。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | None 默认值: None

返回	描述
输出	Runnable 的输出。

ainvoke 异步

ainvoke(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AIMessage

将单个输入转换为输出。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | None 默认值: None

返回	描述
输出	Runnable 的输出。

batch

batch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

默认实现使用线程池执行器并行运行 invoke。

批处理的默认实现对于 IO 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理，则必须重写此方法；例如，如果底层的 Runnable 使用支持批处理模式的 API。

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | list[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

batch_as_completed

batch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> Iterator[tuple[int, Output | Exception]]

在输入列表上并行运行 invoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | Sequence[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
tuple[int, Output | Exception]	由输入索引和 Runnable 输出组成的元组。

abatch 异步

abatch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

默认实现使用 asyncio.gather 并行运行 ainvoke。

batch 的默认实现对于 IO 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理，则必须重写此方法；例如，如果底层的 Runnable 使用支持批处理模式的 API。

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | list[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

abatch_as_completed 异步

abatch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> AsyncIterator[tuple[int, Output | Exception]]

在输入列表上并行运行 ainvoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | Sequence[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[tuple[int, Output | Exception]]	一个由输入索引和 Runnable 输出组成的元组。

stream

stream(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> Iterator[AIMessageChunk]

stream 的默认实现，它调用 invoke。

如果子类支持流式输出，则必须重写此方法。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
输出	Runnable 的输出。

astream 异步

astream(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[AIMessageChunk]

astream 的默认实现，它调用 ainvoke。

如果子类支持流式输出，则必须重写此方法。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[Output]	Runnable 的输出。

astream_log 异步

astream_log(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    diff: bool = True,
    with_streamed_output_list: bool = True,
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]

流式传输 Runnable 的所有输出，如回调系统所报告。

这包括 LLM、检索器、工具等的所有内部运行。

输出以 Log 对象的形式流式传输，其中包括一个 Jsonpatch 操作列表，描述了运行状态在每一步中如何变化，以及运行的最终状态。

可以按顺序应用 Jsonpatch 操作来构造状态。

参数	描述
输入	Runnable 的输入。 类型： Any
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
差异	是生成每一步之间的差异还是当前状态。 类型: bool 默认值: True
带流式输出列表	是否生成 streamed_output 列表。 类型: bool 默认值: True
包含名称	仅包含具有这些名称的日志。 类型: Sequence[str] | None 默认值: None
包含类型	仅包含具有这些类型的日志。 类型: Sequence[str] | None 默认值: None
包含标签	仅包含具有这些标签的日志。 类型: Sequence[str] | None 默认值: None
排除名称	排除具有这些名称的日志。 类型: Sequence[str] | None 默认值: None
排除类型	排除具有这些类型的日志。 类型: Sequence[str] | None 默认值: None
排除标签	排除具有这些标签的日志。 类型: Sequence[str] | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型： Any 默认值： {}

YIELDS	描述
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]	一个 RunLogPatch 或 RunLog 对象。

astream_events 异步

astream_events(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    version: Literal["v1", "v2"] = "v2",
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]

生成事件流。

用于创建一个 StreamEvent 的迭代器，提供有关 Runnable 进度的实时信息，包括来自中间结果的 StreamEvent。

一个 StreamEvent 是一个具有以下模式的字典

• event：事件名称的格式为：on_[runnable_type]_(start|stream|end)。
• name：生成事件的 Runnable 的名称。
• run_id：与发出事件的 Runnable 的给定执行相关联的随机生成的 ID。作为父 Runnable 执行的一部分被调用的子 Runnable 被分配其自己的唯一 ID。
• parent_ids：生成事件的父可运行对象的 ID。根 Runnable 将有一个空列表。父 ID 的顺序是从根到直接父级。仅适用于 API 的 v2 版本。API 的 v1 版本将返回一个空列表。
• tags：生成事件的 Runnable 的标签。
• metadata：生成事件的 Runnable 的元数据。
• data：与事件关联的数据。此字段的内容取决于事件的类型。有关更多详细信息，请参见下表。

下表说明了各种链可能发出的某些事件。为简洁起见，已从表中省略了元数据字段。链定义已包含在表之后。

注意

此参考表适用于模式的 v2 版本。

事件	name	chunk	输入	output
on_chat_model_start	'[model name]'		{"messages": [[SystemMessage, HumanMessage]]}
on_chat_model_stream	'[model name]'	AIMessageChunk(content="hello")
on_chat_model_end	'[model name]'		{"messages": [[SystemMessage, HumanMessage]]}	AIMessageChunk(content="hello world")
on_llm_start	'[model name]'		{'input': 'hello'}
on_llm_stream	'[model name]'	'你好'
on_llm_end	'[model name]'		'你好，人类！'
on_chain_start	'format_docs'
on_chain_stream	'format_docs'	'hello world!, goodbye world!'
on_chain_end	'format_docs'		[Document(...)]	'hello world!, goodbye world!'
on_tool_start	'some_tool'		{"x": 1, "y": "2"}
on_tool_end	'some_tool'			{"x": 1, "y": "2"}
on_retriever_start	'[retriever name]'		{"query": "hello"}
on_retriever_end	'[retriever name]'		{"query": "hello"}	[Document(...), ..]
on_prompt_start	'[template_name]'		{"question": "hello"}
on_prompt_end	'[template_name]'		{"question": "hello"}	ChatPromptValue(messages: [SystemMessage, ...])

除了标准事件外，用户还可以分派自定义事件（见下例）。

自定义事件将仅在 API 的 v2 版本中出现！

自定义事件具有以下格式

属性	类型	描述
name	str	用户为事件定义的名称。
data	任意	与事件关联的数据。这可以是任何东西，但我们建议使其可 JSON 序列化。

以下是与上面显示的标准事件相关的声明

format_docs:

def format_docs(docs: list[Document]) -> str:
    '''Format the docs.'''
    return ", ".join([doc.page_content for doc in docs])


format_docs = RunnableLambda(format_docs)

some_tool:

@tool
def some_tool(x: int, y: str) -> dict:
    '''Some_tool.'''
    return {"x": x, "y": y}

prompt:

template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are Cat Agent 007"),
        ("human", "{question}"),
    ]
).with_config({"run_name": "my_template", "tags": ["my_template"]})

例如

from langchain_core.runnables import RunnableLambda


async def reverse(s: str) -> str:
    return s[::-1]


chain = RunnableLambda(func=reverse)

events = [event async for event in chain.astream_events("hello", version="v2")]

# Will produce the following events
# (run_id, and parent_ids has been omitted for brevity):
[
    {
        "data": {"input": "hello"},
        "event": "on_chain_start",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"chunk": "olleh"},
        "event": "on_chain_stream",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"output": "olleh"},
        "event": "on_chain_end",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
]

示例：分派自定义事件

from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio


async def slow_thing(some_input: str, config: RunnableConfig) -> str:
    """Do something that takes a long time."""
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 1 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 2 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    return "Done"

slow_thing = RunnableLambda(slow_thing)

async for event in slow_thing.astream_events("some_input", version="v2"):
    print(event)

参数	描述
输入	Runnable 的输入。 类型： Any
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
版本	要使用的模式版本，可以是 'v2' 或 'v1'。用户应使用 'v2'。'v1' 是为了向后兼容，并将在 0.4.0 中弃用。在 API 稳定之前不会分配默认值。自定义事件将仅在 'v2' 中出现。 类型: Literal['v1', 'v2'] 默认值: 'v2'
包含名称	仅包括来自具有匹配名称的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
包含类型	仅包括来自具有匹配类型的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
包含标签	仅包括来自具有匹配标签的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
排除名称	排除来自具有匹配名称的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
排除类型	排除来自具有匹配类型的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
排除标签	排除来自具有匹配标签的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。这些将传递给 astream_log，因为 astream_events 的此实现是建立在 astream_log 之上的。 类型： Any 默认值： {}

YIELDS	描述
AsyncIterator[StreamEvent]	StreamEvent 的异步流。

引发	描述
NotImplementedError	如果版本不是 'v1' 或 'v2'。

transform

transform(
    input: Iterator[Input], config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]

将输入转换为输出。

transform 的默认实现，它缓冲输入并调用 astream。

如果子类可以在输入仍在生成时开始产生输出，则必须重写此方法。

参数	描述
输入	Runnable 输入的迭代器。 类型: Iterator[Input]
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
输出	Runnable 的输出。

atransform 异步

atransform(
    input: AsyncIterator[Input],
    config: RunnableConfig | None = None,
    **kwargs: Any | None,
) -> AsyncIterator[Output]

将输入转换为输出。

atransform 的默认实现，它缓冲输入并调用 astream。

如果子类可以在输入仍在生成时开始产生输出，则必须重写此方法。

参数	描述
输入	Runnable 输入的异步迭代器。 类型： AsyncIterator[Input]
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[Output]	Runnable 的输出。

bind

bind(**kwargs: Any) -> Runnable[Input, Output]

将参数绑定到 Runnable，返回一个新的 Runnable。

当链中的 Runnable 需要一个不在前一个 Runnable 的输出中或未包含在用户输入中的参数时非常有用。

参数	描述
**kwargs	要绑定到 Runnable 的参数。 类型： Any 默认值： {}

返回	描述
Runnable[Input, Output]	一个绑定了参数的新 Runnable。

示例

from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser

model = ChatOllama(model="llama3.1")

# Without bind
chain = model | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'

# With bind
chain = model.bind(stop=["three"]) | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'

with_config

with_config(
    config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]

将配置绑定到 Runnable，返回一个新的 Runnable。

参数	描述
配置	要绑定到 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型： Any 默认值： {}

返回	描述
Runnable[Input, Output]	一个绑定了配置的新 Runnable。

with_listeners

with_listeners(
    *,
    on_start: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
    on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
    on_error: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
) -> Runnable[Input, Output]

将生命周期侦听器绑定到 Runnable，返回一个新的 Runnable。

Run 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_time 以及添加到运行中的任何标签或元数据。

参数	描述
开始时	在 Runnable 开始运行之前调用，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None
结束时	在 Runnable 完成运行后调用，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None
出错时	如果 Runnable 抛出错误，则调用此函数，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了监听器的新 Runnable。

示例

from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run

import time


def test_runnable(time_to_sleep: int):
    time.sleep(time_to_sleep)


def fn_start(run_obj: Run):
    print("start_time:", run_obj.start_time)


def fn_end(run_obj: Run):
    print("end_time:", run_obj.end_time)


chain = RunnableLambda(test_runnable).with_listeners(
    on_start=fn_start, on_end=fn_end
)
chain.invoke(2)

with_alisteners

with_alisteners(
    *,
    on_start: AsyncListener | None = None,
    on_end: AsyncListener | None = None,
    on_error: AsyncListener | None = None,
) -> Runnable[Input, Output]

将异步生命周期侦听器绑定到 Runnable。

返回一个新的 Runnable。

Run 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_time 以及添加到运行中的任何标签或元数据。

参数	描述
开始时	在 Runnable 开始运行之前异步调用，并传入 Run 对象。 类型： AsyncListener | None 默认值： None
结束时	在 Runnable 完成运行后异步调用，并传入 Run 对象。 类型： AsyncListener | None 默认值： None
出错时	如果 Runnable 抛出错误，则异步调用此函数，并传入 Run 对象。 类型： AsyncListener | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了监听器的新 Runnable。

示例

from langchain_core.runnables import RunnableLambda, Runnable
from datetime import datetime, timezone
import time
import asyncio

def format_t(timestamp: float) -> str:
    return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()

async def test_runnable(time_to_sleep: int):
    print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}")
    await asyncio.sleep(time_to_sleep)
    print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}")

async def fn_start(run_obj: Runnable):
    print(f"on start callback starts at {format_t(time.time())}")
    await asyncio.sleep(3)
    print(f"on start callback ends at {format_t(time.time())}")

async def fn_end(run_obj: Runnable):
    print(f"on end callback starts at {format_t(time.time())}")
    await asyncio.sleep(2)
    print(f"on end callback ends at {format_t(time.time())}")

runnable = RunnableLambda(test_runnable).with_alisteners(
    on_start=fn_start,
    on_end=fn_end
)
async def concurrent_runs():
    await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))

asyncio.run(concurrent_runs())
Result:
on start callback starts at 2025-03-01T07:05:22.875378+00:00
on start callback starts at 2025-03-01T07:05:22.875495+00:00
on start callback ends at 2025-03-01T07:05:25.878862+00:00
on start callback ends at 2025-03-01T07:05:25.878947+00:00
Runnable[2s]: starts at 2025-03-01T07:05:25.879392+00:00
Runnable[3s]: starts at 2025-03-01T07:05:25.879804+00:00
Runnable[2s]: ends at 2025-03-01T07:05:27.881998+00:00
on end callback starts at 2025-03-01T07:05:27.882360+00:00
Runnable[3s]: ends at 2025-03-01T07:05:28.881737+00:00
on end callback starts at 2025-03-01T07:05:28.882428+00:00
on end callback ends at 2025-03-01T07:05:29.883893+00:00
on end callback ends at 2025-03-01T07:05:30.884831+00:00

with_types

with_types(
    *, input_type: type[Input] | None = None, output_type: type[Output] | None = None
) -> Runnable[Input, Output]

将输入和输出类型绑定到 Runnable，返回一个新的 Runnable。

参数	描述
输入类型	要绑定到 Runnable 的输入类型。 类型： type[Input] | None 默认值： None
输出类型	要绑定到 Runnable 的输出类型。 类型： type[Output] | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了类型的新 Runnable。

with_retry

with_retry(
    *,
    retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
    wait_exponential_jitter: bool = True,
    exponential_jitter_params: ExponentialJitterParams | None = None,
    stop_after_attempt: int = 3,
) -> Runnable[Input, Output]

创建一个新的 Runnable，它在发生异常时重试原始的 Runnable。

参数	描述
如果异常类型则重试	一个用于重试的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
指数等待抖动	是否在两次重试之间的等待时间中添加抖动。 类型: bool 默认值: True
尝试后停止	放弃前尝试的最大次数。 类型： int 默认值： 3
指数抖动参数	tenacity.wait_exponential_jitter 的参数。即：initial、max、exp_base 和 jitter（均为 float 值）。 类型： ExponentialJitterParams | None 默认值： None

返回	描述
Runnable[Input, Output]	一个新的 Runnable，它会在发生异常时重试原始的 Runnable。

示例

from langchain_core.runnables import RunnableLambda

count = 0


def _lambda(x: int) -> None:
    global count
    count = count + 1
    if x == 1:
        raise ValueError("x is 1")
    else:
        pass


runnable = RunnableLambda(_lambda)
try:
    runnable.with_retry(
        stop_after_attempt=2,
        retry_if_exception_type=(ValueError,),
    ).invoke(1)
except ValueError:
    pass

assert count == 2

map

map() -> Runnable[list[Input], list[Output]]

返回一个新的 Runnable，它将输入列表映射到输出列表。

使用每个输入调用 invoke。

返回	描述
Runnable[list[Input], list[Output]]	一个新的 Runnable，它将输入列表映射到输出列表。

示例

from langchain_core.runnables import RunnableLambda


def _lambda(x: int) -> int:
    return x + 1


runnable = RunnableLambda(_lambda)
print(runnable.map().invoke([1, 2, 3]))  # [2, 3, 4]

with_fallbacks

with_fallbacks(
    fallbacks: Sequence[Runnable[Input, Output]],
    *,
    exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
    exception_key: str | None = None,
) -> RunnableWithFallbacks[Input, Output]

向 Runnable 添加回退机制，返回一个新的 Runnable。

新的 Runnable 将在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

参数	描述
备用方案	如果原始 Runnable 失败，将尝试的一系列 runnable。 类型： Sequence[Runnable[Input, Output]]
要处理的异常	一个要处理的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
异常键	如果指定了 string，则已处理的异常将作为输入的一部分，在指定的键下传递给备选方案。如果为 None，异常将不会传递给备选方案。如果使用，基础 Runnable 及其备选方案必须接受一个字典作为输入。 类型: str | None 默认值: None

返回	描述
RunnableWithFallbacks[Input, Output]	一个新的 Runnable，它会在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

示例

from typing import Iterator

from langchain_core.runnables import RunnableGenerator


def _generate_immediate_error(input: Iterator) -> Iterator[str]:
    raise ValueError()
    yield ""


def _generate(input: Iterator) -> Iterator[str]:
    yield from "foo bar"


runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
    [RunnableGenerator(_generate)]
)
print("".join(runnable.stream({})))  # foo bar

参数	描述
备用方案	如果原始 Runnable 失败，将尝试的一系列 runnable。 类型： Sequence[Runnable[Input, Output]]
要处理的异常	一个要处理的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
异常键	如果指定了 string，则已处理的异常将作为输入的一部分，在指定的键下传递给备选方案。如果为 None，异常将不会传递给备选方案。如果使用，基础 Runnable 及其备选方案必须接受一个字典作为输入。 类型: str | None 默认值: None

返回	描述
RunnableWithFallbacks[Input, Output]	一个新的 Runnable，它会在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

as_tool

as_tool(
    args_schema: type[BaseModel] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    arg_types: dict[str, type] | None = None,
) -> BaseTool

从 Runnable 创建一个 BaseTool。

as_tool 将从一个 Runnable 实例化一个 BaseTool，该工具具有名称、描述和 args_schema。在可能的情况下，模式会从 runnable.get_input_schema 中推断。或者（例如，如果 Runnable 接受一个字典作为输入，并且特定的字典键没有类型），模式可以通过 args_schema 直接指定。你也可以传递 arg_types 来仅指定必需的参数及其类型。

参数	描述
参数模式	工具的模式。 类型： type[BaseModel] | None 默认值： None
name	工具的名称。 类型: str | None 默认值: None
描述	工具的描述。 类型: str | None 默认值: None
参数类型	一个从参数名称到类型的字典。 类型： dict[str, type] | None 默认值： None

返回	描述
BaseTool	一个 BaseTool 实例。

类型化字典输入

from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda


class Args(TypedDict):
    a: int
    b: list[int]


def f(x: Args) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool()
as_tool.invoke({"a": 3, "b": [1, 2]})

dict 输入，通过 args_schema 指定模式

from typing import Any
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableLambda

def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))

class FSchema(BaseModel):
    """Apply a function to an integer and list of integers."""

    a: int = Field(..., description="Integer")
    b: list[int] = Field(..., description="List of ints")

runnable = RunnableLambda(f)
as_tool = runnable.as_tool(FSchema)
as_tool.invoke({"a": 3, "b": [1, 2]})

dict 输入，通过 arg_types 指定模式

from typing import Any
from langchain_core.runnables import RunnableLambda


def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool(arg_types={"a": int, "b": list[int]})
as_tool.invoke({"a": 3, "b": [1, 2]})

字符串输入

from langchain_core.runnables import RunnableLambda


def f(x: str) -> str:
    return x + "a"


def g(x: str) -> str:
    return x + "z"


runnable = RunnableLambda(f) | g
as_tool = runnable.as_tool()
as_tool.invoke("b")

__init__

__init__(*args: Any, **kwargs: Any) -> None

is_lc_serializable 类方法

is_lc_serializable() -> bool

这个类是否可序列化？

根据设计，即使一个类继承自 Serializable，它默认也是不可序列化的。这是为了防止意外序列化不应被序列化的对象。

返回	描述
bool	类是否可序列化。默认为 False。

get_lc_namespace 类方法

get_lc_namespace() -> list[str]

获取 LangChain 对象的命名空间。

例如，如果类是 langchain.llms.openai.OpenAI，那么命名空间是 ["langchain", "llms", "openai"]

返回	描述
list[str]	命名空间。

lc_id 类方法

lc_id() -> list[str]

为此类返回一个用于序列化目的的唯一标识符。

唯一标识符是一个描述对象路径的字符串列表。

例如，对于类 langchain.llms.openai.OpenAI，id 是 ["langchain", "llms", "openai", "OpenAI"]。

to_json

to_json() -> SerializedConstructor | SerializedNotImplemented

将 Runnable 序列化为 JSON。

返回	描述
SerializedConstructor | SerializedNotImplemented	一个 Runnable 的 JSON 可序列化表示。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

序列化一个“未实现”的对象。

返回	描述
SerializedNotImplemented	SerializedNotImplemented.

configurable_fields

configurable_fields(
    **kwargs: AnyConfigurableField,
) -> RunnableSerializable[Input, Output]

在运行时配置特定的 Runnable 字段。

参数	描述
**kwargs	一个要配置的 ConfigurableField 实例的字典。 类型： AnyConfigurableField 默认值： {}

引发	描述
ValueError	如果在 Runnable 中找不到配置键。

返回	描述
RunnableSerializable[Input, Output]	一个配置了字段的新 Runnable。

from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatOpenAI(max_tokens=20).configurable_fields(
    max_tokens=ConfigurableField(
        id="output_token_number",
        name="Max tokens in the output",
        description="The maximum number of tokens in the output",
    )
)

# max_tokens = 20
print("max_tokens_20: ", model.invoke("tell me something about chess").content)

# max_tokens = 200
print(
    "max_tokens_200: ",
    model.with_config(configurable={"output_token_number": 200})
    .invoke("tell me something about chess")
    .content,
)

configurable_alternatives

configurable_alternatives(
    which: ConfigurableField,
    *,
    default_key: str = "default",
    prefix_keys: bool = False,
    **kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]],
) -> RunnableSerializable[Input, Output]

为可在运行时设置的 Runnable 对象配置备选项。

参数	描述
哪个	将用于选择备选项的 ConfigurableField 实例。 类型： ConfigurableField
默认键	如果未选择备选项，则使用的默认键。 类型： str 默认值： 'default'
前缀键	是否用 ConfigurableField id 作为键的前缀。 类型： bool 默认值： False
**kwargs	一个从键到 Runnable 实例或返回 Runnable 实例的可调用对象的字典。 类型： Runnable[Input, Output] | Callable[[], Runnable[Input, Output]] 默认值： {}

返回	描述
RunnableSerializable[Input, Output]	一个配置了备选项的新 Runnable。

from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatAnthropic(
    model_name="claude-sonnet-4-5-20250929"
).configurable_alternatives(
    ConfigurableField(id="llm"),
    default_key="anthropic",
    openai=ChatOpenAI(),
)

# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)

# uses ChatOpenAI
print(
    model.with_config(configurable={"llm": "openai"})
    .invoke("which organization created you?")
    .content
)

set_verbose

set_verbose(verbose: bool | None) -> bool

如果 verbose 是 None，则设置它。

这允许用户传入 None 作为 verbose 来访问全局设置。

参数	描述
verbose	要使用的详细程度设置。 类型： bool | None

返回	描述
bool	要使用的详细程度设置。

generate_prompt

generate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult

将一系列提示传递给模型并返回模型生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
prompts	PromptValue 对象列表。PromptValue 是一个可以转换为匹配任何语言模型格式的对象（对于纯文本生成模型是字符串，对于聊天模型是 BaseMessage 对象）。 类型： list[PromptValue]
停止	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
回调	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generation 对象列表以及额外的模型提供商特定的输出。

agenerate_prompt 异步

agenerate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult

异步地将一系列提示传递并返回模型生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
prompts	PromptValue 对象列表。PromptValue 是一个可以转换为匹配任何语言模型格式的对象（对于纯文本生成模型是字符串，对于聊天模型是 BaseMessage 对象）。 类型： list[PromptValue]
停止	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
回调	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generation 对象列表以及额外的模型提供商特定的输出。

with_structured_output

with_structured_output(
    schema: Dict | type, *, include_raw: bool = False, **kwargs: Any
) -> Runnable[LanguageModelInput, Dict | BaseModel]

返回与给定模式匹配的格式化输出的模型包装器。

参数	描述
模式	输出模式。可以作为以下形式传入： 一个 OpenAI 函数/工具模式， 一个 JSON Schema， 一个 TypedDict 类， 或一个 Pydantic 类。 如果 schema 是一个 Pydantic 类，那么模型输出将是该类的 Pydantic 实例，并且模型生成的字段将由 Pydantic 类进行验证。否则，模型输出将是一个字典，并且不会被验证。 有关在指定 Pydantic 或 TypedDict 类时如何正确指定模式字段的类型和描述的更多信息，请参阅 langchain_core.utils.function_calling.convert_to_openai_tool。 类型： Dict | type
包含原始数据	如果为 False，则仅返回解析后的结构化输出。如果在模型输出解析期间发生错误，它将被引发。如果为 True，则将返回原始模型响应（一个 BaseMessage）和解析后的模型响应。如果在输出解析期间发生错误，它将被捕获并一并返回。 最终输出总是一个带有键 'raw'、'parsed' 和 'parsing_error' 的 dict。 类型： bool 默认值： False

引发	描述
ValueError	如果有任何不支持的 kwargs。
NotImplementedError	如果模型没有实现 with_structured_output()。

返回	描述
Runnable[LanguageModelInput, Dict | BaseModel]	一个接受与 langchain_core.language_models.chat.BaseChatModel 相同输入的 Runnable。如果 include_raw 为 False 且 schema 是一个 Pydantic 类，则 Runnable 输出一个 schema 的实例（即一个 Pydantic 对象）。否则，如果 include_raw 为 False，则 Runnable 输出一个 dict。 如果 include_raw 为 True，则 Runnable 输出一个带有以下键的 dict 'raw': BaseMessage 'parsed': 如果出现解析错误则为 None，否则类型取决于上面描述的 schema。 'parsing_error': BaseException | None

示例：Pydantic 模式 (include_raw=False)

from pydantic import BaseModel


class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''

    answer: str
    justification: str


model = ChatModel(model="model-name", temperature=0)
structured_model = model.with_structured_output(AnswerWithJustification)

structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)

# -> AnswerWithJustification(
#     answer='They weigh the same',
#     justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'
# )

示例：Pydantic 模式 (include_raw=True)

from pydantic import BaseModel


class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''

    answer: str
    justification: str


model = ChatModel(model="model-name", temperature=0)
structured_model = model.with_structured_output(
    AnswerWithJustification, include_raw=True
)

structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)
# -> {
#     'raw': AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_Ao02pnFYXD6GN1yzc0uXPsvF', 'function': {'arguments': '{"answer":"They weigh the same.","justification":"Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ."}', 'name': 'AnswerWithJustification'}, 'type': 'function'}]}),
#     'parsed': AnswerWithJustification(answer='They weigh the same.', justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'),
#     'parsing_error': None
# }

示例：dict 模式 (include_raw=False)

from pydantic import BaseModel
from langchain_core.utils.function_calling import convert_to_openai_tool


class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''

    answer: str
    justification: str


dict_schema = convert_to_openai_tool(AnswerWithJustification)
model = ChatModel(model="model-name", temperature=0)
structured_model = model.with_structured_output(dict_schema)

structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)
# -> {
#     'answer': 'They weigh the same',
#     'justification': 'Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume and density of the two substances differ.'
# }

行为在 0.2.26 版本中已更改

添加了对 TypedDict 类的支持。

get_token_ids

get_token_ids(text: str) -> list[int]

返回文本中 token 的有序 ID。

参数	描述
text	要进行分词的字符串输入。 类型： str

返回	描述
list[int]	与文本中的词元相对应的 ID 列表，按它们在文本中出现的顺序列出。

get_num_tokens

get_num_tokens(text: str) -> int

获取文本中存在的 token 数量。

用于检查输入是否适合模型的上下文窗口。

参数	描述
text	要进行分词的字符串输入。 类型： str

返回	描述
int	文本中的词元整数数量。

get_num_tokens_from_messages

get_num_tokens_from_messages(
    messages: list[BaseMessage], tools: Sequence | None = None
) -> int

获取消息中的 token 数量。

用于检查输入是否适合模型的上下文窗口。

注意

get_num_tokens_from_messages 的基本实现忽略了工具模式。

参数	描述
messages	要进行分词的消息输入。 类型： list[BaseMessage]
工具	如果提供，则为要转换为工具模式的 dict、BaseModel、函数或 BaseTool 对象的序列。 类型： Sequence | None 默认值： None

返回	描述
int	所有消息中词元数量的总和。

generate

generate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: UUID | None = None,
    **kwargs: Any,
) -> LLMResult

将一系列提示传递给模型并返回模型生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
messages	消息列表的列表。 类型： list[list[BaseMessage]]
停止	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
回调	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
tags	要应用的标签。 类型: list[str] | None 默认值: None
metadata	要应用的元数据。 类型： dict[str, Any] | None 默认值： None
运行名称	运行的名称。 类型: str | None 默认值: None
run_id	运行的 ID。 类型: UUID | None 默认值: None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generations 列表以及额外的模型提供商特定的输出。

agenerate 异步

agenerate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: UUID | None = None,
    **kwargs: Any,
) -> LLMResult

异步地将一系列提示传递给模型并返回生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
messages	消息列表的列表。 类型： list[list[BaseMessage]]
停止	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
回调	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
tags	要应用的标签。 类型: list[str] | None 默认值: None
metadata	要应用的元数据。 类型： dict[str, Any] | None 默认值： None
运行名称	运行的名称。 类型: str | None 默认值: None
run_id	运行的 ID。 类型: UUID | None 默认值: None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generations 列表以及额外的模型提供商特定的输出。

dict

dict(**kwargs: Any) -> dict

返回 LLM 的字典。

bind_tools

bind_tools(
    tools: Sequence[Dict[str, Any] | type | Callable | BaseTool],
    *,
    tool_choice: str | None = None,
    **kwargs: Any,
) -> Runnable[LanguageModelInput, AIMessage]

将工具绑定到模型。

参数	描述
工具	要绑定到模型的工具序列。 类型： Sequence[Dict[str, Any] | type | Callable | BaseTool]
工具选择	要使用的工具。如果为 "any"，则可以使用任何工具。 类型: str | None 默认值: None

返回	描述
Runnable[LanguageModelInput, AIMessage]	一个返回消息的 Runnable。

langchain_core.language_models.base

基础语言模型类。

LanguageModelInput 模块属性

LanguageModelInput = PromptValue | str | Sequence[MessageLikeRepresentation]

语言模型的输入。

LanguageModelOutput 模块属性

LanguageModelOutput = BaseMessage | str

语言模型的输出。

LanguageModelLike 模块属性

LanguageModelLike = Runnable[LanguageModelInput, LanguageModelOutput]

语言模型的输入/输出接口。

BaseLanguageModel

基类： RunnableSerializable[LanguageModelInput, LanguageModelOutputVar], ABC

用于与语言模型交互的抽象基类。

所有语言模型包装器都继承自 BaseLanguageModel。

方法	描述
set_verbose	如果 verbose 是 None，则设置它。
generate_prompt	将一系列提示传递给模型并返回模型生成的内容。
agenerate_prompt	异步地将一系列提示传递并返回模型生成的内容。
with_structured_output	此类未实现。
get_token_ids	返回文本中 token 的有序 ID。
get_num_tokens	获取文本中存在的 token 数量。
get_num_tokens_from_messages	获取消息中的 token 数量。
get_name	获取 Runnable 的名称。
get_input_schema	获取可用于验证 Runnable 输入的 Pydantic 模型。
get_input_jsonschema	获取表示 Runnable 输入的 JSON 模式。
get_output_schema	获取可用于验证 Runnable 输出的 Pydantic 模型。
get_output_jsonschema	获取表示 Runnable 输出的 JSON 模式。
config_schema	此 Runnable 接受的配置类型，指定为 Pydantic 模型。
get_config_jsonschema	获取表示 Runnable 配置的 JSON 模式。
get_graph	返回此 Runnable 的图形表示。
get_prompts	返回此 Runnable 使用的提示列表。
__or__	Runnable "or" 运算符。
__ror__	Runnable "reverse-or" 运算符。
pipe	管道连接 Runnable 对象。
pick	从此 Runnable 的输出 dict 中选择键。
assign	向此 Runnable 的 dict 输出分配新字段。
invoke	将单个输入转换为输出。
ainvoke	将单个输入转换为输出。
batch	默认实现使用线程池执行器并行运行 invoke。
batch_as_completed	在输入列表上并行运行 invoke。
abatch	默认实现使用 asyncio.gather 并行运行 ainvoke。
abatch_as_completed	在输入列表上并行运行 ainvoke。
stream	stream 的默认实现，它调用 invoke。
astream	astream 的默认实现，它调用 ainvoke。
astream_log	流式传输 Runnable 的所有输出，如回调系统所报告。
astream_events	生成事件流。
transform	将输入转换为输出。
atransform	将输入转换为输出。
bind	将参数绑定到 Runnable，返回一个新的 Runnable。
with_config	将配置绑定到 Runnable，返回一个新的 Runnable。
with_listeners	将生命周期侦听器绑定到 Runnable，返回一个新的 Runnable。
with_alisteners	将异步生命周期侦听器绑定到 Runnable。
with_types	将输入和输出类型绑定到 Runnable，返回一个新的 Runnable。
with_retry	创建一个新的 Runnable，它在发生异常时重试原始的 Runnable。
map	返回一个新的 Runnable，它将输入列表映射到输出列表。
with_fallbacks	向 Runnable 添加回退机制，返回一个新的 Runnable。
as_tool	从 Runnable 创建一个 BaseTool。
__init__
is_lc_serializable	这个类是否可序列化？
get_lc_namespace	获取 LangChain 对象的命名空间。
lc_id	为此类返回一个用于序列化目的的唯一标识符。
to_json	将 Runnable 序列化为 JSON。
to_json_not_implemented	序列化一个“未实现”的对象。
configurable_fields	在运行时配置特定的 Runnable 字段。
configurable_alternatives	为可在运行时设置的 Runnable 对象配置备选项。

cache 类属性 实例属性

cache: BaseCache | bool | None = Field(default=None, exclude=True)

是否缓存响应。

• 如果为 True，将使用全局缓存。
• 如果为 False，将不使用缓存
• 如果为 None，如果设置了全局缓存，则使用全局缓存，否则不使用缓存。
• 如果是 BaseCache 的实例，将使用提供的缓存。

目前不支持模型的流式方法的缓存。

verbose 类属性 实例属性

verbose: bool = Field(default_factory=_get_verbosity, exclude=True, repr=False)

是否打印响应文本。

callbacks 类属性 实例属性

callbacks: Callbacks = Field(default=None, exclude=True)

添加到运行跟踪中的回调。

tags 类属性 实例属性

tags: list[str] | None = Field(default=None, exclude=True)

添加到运行跟踪中的标签。

metadata 类属性 实例属性

metadata: dict[str, Any] | None = Field(default=None, exclude=True)

添加到运行跟踪中的元数据。

custom_get_token_ids 类属性 实例属性

custom_get_token_ids: Callable[[str], list[int]] | None = Field(
    default=None, exclude=True
)

用于计算 token 的可选编码器。

InputType 属性

InputType: TypeAlias

获取此 Runnable 的输入类型。

name 类属性 实例属性

name: str | None = None

Runnable 的名称。用于调试和追踪。

OutputType 属性

OutputType: type[Output]

输出类型。

此 Runnable 产生的输出类型，指定为类型注解。

引发	描述
TypeError	如果无法推断输出类型。

input_schema 属性

input_schema: type[BaseModel]

此 Runnable 接受的输入类型，指定为 Pydantic 模型。

output_schema 属性

output_schema: type[BaseModel]

输出模式。

此 Runnable 产生的输出类型，指定为 Pydantic 模型。

config_specs 属性

config_specs: list[ConfigurableFieldSpec]

列出此 Runnable 的可配置字段。

lc_secrets 属性

lc_secrets: dict[str, str]

构造函数参数名称到密钥 ID 的映射。

例如，{"openai_api_key": "OPENAI_API_KEY"}

lc_attributes 属性

lc_attributes: dict

应包含在序列化 kwargs 中的属性名称列表。

这些属性必须被构造函数接受。

默认为空字典。

set_verbose

set_verbose(verbose: bool | None) -> bool

如果 verbose 是 None，则设置它。

这允许用户传入 None 作为 verbose 来访问全局设置。

参数	描述
verbose	要使用的详细程度设置。 类型： bool | None

返回	描述
bool	要使用的详细程度设置。

generate_prompt 抽象方法

generate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult

将一系列提示传递给模型并返回模型生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
prompts	PromptValue 对象列表。PromptValue 是一个可以转换为匹配任何语言模型格式的对象（对于纯文本生成模型是字符串，对于聊天模型是 BaseMessage 对象）。 类型： list[PromptValue]
停止	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
回调	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generation 对象列表以及额外的模型提供商特定的输出。

agenerate_prompt 抽象方法 异步

agenerate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult

异步地将一系列提示传递并返回模型生成的内容。

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

1. 利用批量调用，
2. 需要从模型获得比最高生成值更多的输出，
3. 正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
prompts	PromptValue 对象列表。PromptValue 是一个可以转换为匹配任何语言模型格式的对象（对于纯文本生成模型是字符串，对于聊天模型是 BaseMessage 对象）。 类型： list[PromptValue]
停止	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。 类型: list[str] | None 默认值: None
回调	要传递的 Callbacks。用于在生成过程中执行额外的功能，如日志记录或流式处理。 类型： Callbacks 默认值： None
**kwargs	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。 类型： Any 默认值： {}

返回	描述
LLMResult	一个 LLMResult，其中包含每个输入提示的候选 Generation 对象列表以及额外的模型提供商特定的输出。

with_structured_output

with_structured_output(
    schema: dict | type, **kwargs: Any
) -> Runnable[LanguageModelInput, dict | BaseModel]

此类未实现。

get_token_ids

get_token_ids(text: str) -> list[int]

返回文本中 token 的有序 ID。

参数	描述
text	要进行分词的字符串输入。 类型： str

返回	描述
list[int]	与文本中的词元相对应的 ID 列表，按它们在文本中出现的顺序列出。

get_num_tokens

get_num_tokens(text: str) -> int

获取文本中存在的 token 数量。

用于检查输入是否适合模型的上下文窗口。

参数	描述
text	要进行分词的字符串输入。 类型： str

返回	描述
int	文本中的词元整数数量。

get_num_tokens_from_messages

get_num_tokens_from_messages(
    messages: list[BaseMessage], tools: Sequence | None = None
) -> int

获取消息中的 token 数量。

用于检查输入是否适合模型的上下文窗口。

注意

get_num_tokens_from_messages 的基本实现忽略了工具模式。

参数	描述
messages	要进行分词的消息输入。 类型： list[BaseMessage]
工具	如果提供，则为要转换为工具模式的 dict、BaseModel、函数或 BaseTool 对象的序列。 类型： Sequence | None 默认值： None

返回	描述
int	所有消息中词元数量的总和。

get_name

get_name(suffix: str | None = None, *, name: str | None = None) -> str

获取 Runnable 的名称。

参数	描述
后缀	一个可选的后缀，附加到名称上。 类型: str | None 默认值: None
name	一个可选的名称，用于代替 Runnable 的名称。 类型: str | None 默认值: None

返回	描述
str	Runnable 的名称。

get_input_schema

get_input_schema(config: RunnableConfig | None = None) -> type[BaseModel]

获取可用于验证 Runnable 输入的 Pydantic 模型。

利用 configurable_fields 和 configurable_alternatives 方法的 Runnable 对象将具有一个动态输入模式，该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输入模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证输入的 Pydantic 模型。

get_input_jsonschema

get_input_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

获取表示 Runnable 输入的 JSON 模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 输入的 JSON 模式。

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_input_jsonschema())

在 0.3.0 版本中新增。

get_output_schema

get_output_schema(config: RunnableConfig | None = None) -> type[BaseModel]

获取可用于验证 Runnable 输出的 Pydantic 模型。

利用 configurable_fields 和 configurable_alternatives 方法的 Runnable 对象将具有一个动态输出模式，该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输出模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证输出的 Pydantic 模型。

get_output_jsonschema

get_output_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

获取表示 Runnable 输出的 JSON 模式。

参数	描述
配置	生成模式时使用的配置。 类型: RunnableConfig | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 输出的 JSON 模式。

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_output_jsonschema())

在 0.3.0 版本中新增。

config_schema

config_schema(*, include: Sequence[str] | None = None) -> type[BaseModel]

此 Runnable 接受的配置类型，指定为 Pydantic 模型。

要将字段标记为可配置，请参阅 configurable_fields 和 configurable_alternatives 方法。

参数	描述
包含	要包含在配置模式中的字段列表。 类型: Sequence[str] | None 默认值: None

返回	描述
type[BaseModel]	一个可用于验证配置的 Pydantic 模型。

get_config_jsonschema

get_config_jsonschema(*, include: Sequence[str] | None = None) -> dict[str, Any]

获取表示 Runnable 配置的 JSON 模式。

参数	描述
包含	要包含在配置模式中的字段列表。 类型: Sequence[str] | None 默认值: None

返回	描述
dict[str, Any]	表示 Runnable 配置的 JSON 模式。

在 0.3.0 版本中新增。

get_graph

get_graph(config: RunnableConfig | None = None) -> Graph

返回此 Runnable 的图形表示。

get_prompts

get_prompts(config: RunnableConfig | None = None) -> list[BasePromptTemplate]

返回此 Runnable 使用的提示列表。

__or__

__or__(
    other: Runnable[Any, Other]
    | Callable[[Iterator[Any]], Iterator[Other]]
    | Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
    | Callable[[Any], Other]
    | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]

Runnable "or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence。

参数	描述
other	另一个 Runnable 或类 Runnable 对象。 类型: Runnable[Any, Other] | Callable[[Iterator[Any]], Iterator[Other]] | Callable[[AsyncIterator[Any]], AsyncIterator[Other]] | Callable[[Any], Other] | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any]

返回	描述
RunnableSerializable[Input, Other]	一个新的 Runnable。

__ror__

__ror__(
    other: Runnable[Other, Any]
    | Callable[[Iterator[Other]], Iterator[Any]]
    | Callable[[AsyncIterator[Other]], AsyncIterator[Any]]
    | Callable[[Other], Any]
    | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any],
) -> RunnableSerializable[Other, Output]

Runnable "reverse-or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence。

参数	描述
other	另一个 Runnable 或类 Runnable 对象。 类型: Runnable[Other, Any] | Callable[[Iterator[Other]], Iterator[Any]] | Callable[[AsyncIterator[Other]], AsyncIterator[Any]] | Callable[[Other], Any] | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any]

返回	描述
RunnableSerializable[Other, Output]	一个新的 Runnable。

pipe

pipe(
    *others: Runnable[Any, Other] | Callable[[Any], Other], name: str | None = None
) -> RunnableSerializable[Input, Other]

管道连接 Runnable 对象。

将此 Runnable 与类 Runnable 对象组合以构成一个 RunnableSequence。

等同于 RunnableSequence(self, *others) 或 self | others[0] | ...

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


def mul_two(x: int) -> int:
    return x * 2


runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1.pipe(runnable_2)
# Or equivalently:
# sequence = runnable_1 | runnable_2
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)
# -> 4

sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
# -> [4, 6, 8]

参数	描述
*其他	其他要组合的 Runnable 或类 Runnable 对象 类型: Runnable[Any, Other] | Callable[[Any], Other] 默认值: ()
name	生成的 RunnableSequence 的一个可选名称。 类型: str | None 默认值: None

返回	描述
RunnableSerializable[Input, Other]	一个新的 Runnable。

pick

pick(keys: str | list[str]) -> RunnableSerializable[Any, Any]

从此 Runnable 的输出 dict 中选择键。

选择单个键

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
chain = RunnableMap(str=as_str, json=as_json)

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3]}

json_only_chain = chain.pick("json")
json_only_chain.invoke("[1, 2, 3]")
# -> [1, 2, 3]

选择键列表

from typing import Any

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)


def as_bytes(x: Any) -> bytes:
    return bytes(x, "utf-8")


chain = RunnableMap(str=as_str, json=as_json, bytes=RunnableLambda(as_bytes))

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

json_and_bytes_chain = chain.pick(["json", "bytes"])
json_and_bytes_chain.invoke("[1, 2, 3]")
# -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

参数	描述
keys	从输出字典中选择的一个键或键列表。 类型: str | list[str]

返回	描述
RunnableSerializable[Any, Any]	一个新的 Runnable。

assign

assign(
    **kwargs: Runnable[dict[str, Any], Any]
    | Callable[[dict[str, Any]], Any]
    | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]],
) -> RunnableSerializable[Any, Any]

向此 Runnable 的 dict 输出分配新字段。

from langchain_community.llms.fake import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.runnables import Runnable
from operator import itemgetter

prompt = (
    SystemMessagePromptTemplate.from_template("You are a nice assistant.")
    + "{question}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])

chain: Runnable = prompt | model | {"str": StrOutputParser()}

chain_with_assign = chain.assign(hello=itemgetter("str") | model)

print(chain_with_assign.input_schema.model_json_schema())
# {'title': 'PromptInput', 'type': 'object', 'properties':
{'question': {'title': 'Question', 'type': 'string'}}}
print(chain_with_assign.output_schema.model_json_schema())
# {'title': 'RunnableSequenceOutput', 'type': 'object', 'properties':
{'str': {'title': 'Str',
'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}}

参数	描述
**kwargs	一个键到 Runnable 或类 Runnable 对象的映射，这些对象将使用此 Runnable 的整个输出字典来调用。 类型: Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any] | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]] 默认值: {}

返回	描述
RunnableSerializable[Any, Any]	一个新的 Runnable。

invoke 抽象方法

invoke(input: Input, config: RunnableConfig | None = None, **kwargs: Any) -> Output

将单个输入转换为输出。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | None 默认值: None

返回	描述
输出	Runnable 的输出。

ainvoke 异步

ainvoke(input: Input, config: RunnableConfig | None = None, **kwargs: Any) -> Output

将单个输入转换为输出。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | None 默认值: None

返回	描述
输出	Runnable 的输出。

batch

batch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

默认实现使用线程池执行器并行运行 invoke。

批处理的默认实现对于 IO 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理，则必须重写此方法；例如，如果底层的 Runnable 使用支持批处理模式的 API。

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | list[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

batch_as_completed

batch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> Iterator[tuple[int, Output | Exception]]

在输入列表上并行运行 invoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | Sequence[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
tuple[int, Output | Exception]	由输入索引和 Runnable 输出组成的元组。

abatch 异步

abatch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

默认实现使用 asyncio.gather 并行运行 ainvoke。

batch 的默认实现对于 IO 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理，则必须重写此方法；例如，如果底层的 Runnable 使用支持批处理模式的 API。

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | list[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

abatch_as_completed 异步

abatch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> AsyncIterator[tuple[int, Output | Exception]]

在输入列表上并行运行 ainvoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型: RunnableConfig | Sequence[RunnableConfig] | None 默认值: None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[tuple[int, Output | Exception]]	一个由输入索引和 Runnable 输出组成的元组。

stream

stream(
    input: Input, config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]

stream 的默认实现，它调用 invoke。

如果子类支持流式输出，则必须重写此方法。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
输出	Runnable 的输出。

astream 异步

astream(
    input: Input, config: RunnableConfig | None = None, **kwargs: Any | None
) -> AsyncIterator[Output]

astream 的默认实现，它调用 ainvoke。

如果子类支持流式输出，则必须重写此方法。

参数	描述
输入	Runnable 的输入。 类型： Input
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[Output]	Runnable 的输出。

astream_log 异步

astream_log(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    diff: bool = True,
    with_streamed_output_list: bool = True,
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]

流式传输 Runnable 的所有输出，如回调系统所报告。

这包括 LLM、检索器、工具等的所有内部运行。

输出以 Log 对象的形式流式传输，其中包括一个 Jsonpatch 操作列表，描述了运行状态在每一步中如何变化，以及运行的最终状态。

可以按顺序应用 Jsonpatch 操作来构造状态。

参数	描述
输入	Runnable 的输入。 类型： Any
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
差异	是生成每一步之间的差异还是当前状态。 类型: bool 默认值: True
带流式输出列表	是否生成 streamed_output 列表。 类型: bool 默认值: True
包含名称	仅包含具有这些名称的日志。 类型: Sequence[str] | None 默认值: None
包含类型	仅包含具有这些类型的日志。 类型: Sequence[str] | None 默认值: None
包含标签	仅包含具有这些标签的日志。 类型: Sequence[str] | None 默认值: None
排除名称	排除具有这些名称的日志。 类型: Sequence[str] | None 默认值: None
排除类型	排除具有这些类型的日志。 类型: Sequence[str] | None 默认值: None
排除标签	排除具有这些标签的日志。 类型: Sequence[str] | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型： Any 默认值： {}

YIELDS	描述
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]	一个 RunLogPatch 或 RunLog 对象。

astream_events 异步

astream_events(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    version: Literal["v1", "v2"] = "v2",
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]

生成事件流。

用于创建一个 StreamEvent 的迭代器，提供有关 Runnable 进度的实时信息，包括来自中间结果的 StreamEvent。

一个 StreamEvent 是一个具有以下模式的字典

• event：事件名称的格式为：on_[runnable_type]_(start|stream|end)。
• name：生成事件的 Runnable 的名称。
• run_id：与发出事件的 Runnable 的给定执行相关联的随机生成的 ID。作为父 Runnable 执行的一部分被调用的子 Runnable 被分配其自己的唯一 ID。
• parent_ids：生成事件的父可运行对象的 ID。根 Runnable 将有一个空列表。父 ID 的顺序是从根到直接父级。仅适用于 API 的 v2 版本。API 的 v1 版本将返回一个空列表。
• tags：生成事件的 Runnable 的标签。
• metadata：生成事件的 Runnable 的元数据。
• data：与事件关联的数据。此字段的内容取决于事件的类型。有关更多详细信息，请参见下表。

下表说明了各种链可能发出的某些事件。为简洁起见，已从表中省略了元数据字段。链定义已包含在表之后。

注意

此参考表适用于模式的 v2 版本。

事件	name	chunk	输入	output
on_chat_model_start	'[model name]'		{"messages": [[SystemMessage, HumanMessage]]}
on_chat_model_stream	'[model name]'	AIMessageChunk(content="hello")
on_chat_model_end	'[model name]'		{"messages": [[SystemMessage, HumanMessage]]}	AIMessageChunk(content="hello world")
on_llm_start	'[model name]'		{'input': 'hello'}
on_llm_stream	'[model name]'	'你好'
on_llm_end	'[model name]'		'你好，人类！'
on_chain_start	'format_docs'
on_chain_stream	'format_docs'	'hello world!, goodbye world!'
on_chain_end	'format_docs'		[Document(...)]	'hello world!, goodbye world!'
on_tool_start	'some_tool'		{"x": 1, "y": "2"}
on_tool_end	'some_tool'			{"x": 1, "y": "2"}
on_retriever_start	'[retriever name]'		{"query": "hello"}
on_retriever_end	'[retriever name]'		{"query": "hello"}	[Document(...), ..]
on_prompt_start	'[template_name]'		{"question": "hello"}
on_prompt_end	'[template_name]'		{"question": "hello"}	ChatPromptValue(messages: [SystemMessage, ...])

除了标准事件外，用户还可以分派自定义事件（见下例）。

自定义事件将仅在 API 的 v2 版本中出现！

自定义事件具有以下格式

属性	类型	描述
name	str	用户为事件定义的名称。
data	任意	与事件关联的数据。这可以是任何东西，但我们建议使其可 JSON 序列化。

以下是与上面显示的标准事件相关的声明

format_docs:

def format_docs(docs: list[Document]) -> str:
    '''Format the docs.'''
    return ", ".join([doc.page_content for doc in docs])


format_docs = RunnableLambda(format_docs)

some_tool:

@tool
def some_tool(x: int, y: str) -> dict:
    '''Some_tool.'''
    return {"x": x, "y": y}

prompt:

template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are Cat Agent 007"),
        ("human", "{question}"),
    ]
).with_config({"run_name": "my_template", "tags": ["my_template"]})

例如

from langchain_core.runnables import RunnableLambda


async def reverse(s: str) -> str:
    return s[::-1]


chain = RunnableLambda(func=reverse)

events = [event async for event in chain.astream_events("hello", version="v2")]

# Will produce the following events
# (run_id, and parent_ids has been omitted for brevity):
[
    {
        "data": {"input": "hello"},
        "event": "on_chain_start",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"chunk": "olleh"},
        "event": "on_chain_stream",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"output": "olleh"},
        "event": "on_chain_end",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
]

示例：分派自定义事件

from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio


async def slow_thing(some_input: str, config: RunnableConfig) -> str:
    """Do something that takes a long time."""
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 1 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 2 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    return "Done"

slow_thing = RunnableLambda(slow_thing)

async for event in slow_thing.astream_events("some_input", version="v2"):
    print(event)

参数	描述
输入	Runnable 的输入。 类型： Any
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
版本	要使用的模式版本，可以是 'v2' 或 'v1'。用户应使用 'v2'。'v1' 是为了向后兼容，并将在 0.4.0 中弃用。在 API 稳定之前不会分配默认值。自定义事件将仅在 'v2' 中出现。 类型: Literal['v1', 'v2'] 默认值: 'v2'
包含名称	仅包括来自具有匹配名称的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
包含类型	仅包括来自具有匹配类型的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
包含标签	仅包括来自具有匹配标签的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
排除名称	排除来自具有匹配名称的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
排除类型	排除来自具有匹配类型的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
排除标签	排除来自具有匹配标签的 Runnable 对象的事件。 类型: Sequence[str] | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。这些将传递给 astream_log，因为 astream_events 的此实现是建立在 astream_log 之上的。 类型： Any 默认值： {}

YIELDS	描述
AsyncIterator[StreamEvent]	StreamEvent 的异步流。

引发	描述
NotImplementedError	如果版本不是 'v1' 或 'v2'。

transform

transform(
    input: Iterator[Input], config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]

将输入转换为输出。

transform 的默认实现，它缓冲输入并调用 astream。

如果子类可以在输入仍在生成时开始产生输出，则必须重写此方法。

参数	描述
输入	Runnable 输入的迭代器。 类型: Iterator[Input]
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
输出	Runnable 的输出。

atransform 异步

atransform(
    input: AsyncIterator[Input],
    config: RunnableConfig | None = None,
    **kwargs: Any | None,
) -> AsyncIterator[Output]

将输入转换为输出。

atransform 的默认实现，它缓冲输入并调用 astream。

如果子类可以在输入仍在生成时开始产生输出，则必须重写此方法。

参数	描述
输入	Runnable 输入的异步迭代器。 类型： AsyncIterator[Input]
配置	用于 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[Output]	Runnable 的输出。

bind

bind(**kwargs: Any) -> Runnable[Input, Output]

将参数绑定到 Runnable，返回一个新的 Runnable。

当链中的 Runnable 需要一个不在前一个 Runnable 的输出中或未包含在用户输入中的参数时非常有用。

参数	描述
**kwargs	要绑定到 Runnable 的参数。 类型： Any 默认值： {}

返回	描述
Runnable[Input, Output]	一个绑定了参数的新 Runnable。

示例

from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser

model = ChatOllama(model="llama3.1")

# Without bind
chain = model | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'

# With bind
chain = model.bind(stop=["three"]) | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'

with_config

with_config(
    config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]

将配置绑定到 Runnable，返回一个新的 Runnable。

参数	描述
配置	要绑定到 Runnable 的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型： Any 默认值： {}

返回	描述
Runnable[Input, Output]	一个绑定了配置的新 Runnable。

with_listeners

with_listeners(
    *,
    on_start: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
    on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
    on_error: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
) -> Runnable[Input, Output]

将生命周期侦听器绑定到 Runnable，返回一个新的 Runnable。

Run 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_time 以及添加到运行中的任何标签或元数据。

参数	描述
开始时	在 Runnable 开始运行之前调用，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None
结束时	在 Runnable 完成运行后调用，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None
出错时	如果 Runnable 抛出错误，则调用此函数，并传入 Run 对象。 类型： Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了监听器的新 Runnable。

示例

from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run

import time


def test_runnable(time_to_sleep: int):
    time.sleep(time_to_sleep)


def fn_start(run_obj: Run):
    print("start_time:", run_obj.start_time)


def fn_end(run_obj: Run):
    print("end_time:", run_obj.end_time)


chain = RunnableLambda(test_runnable).with_listeners(
    on_start=fn_start, on_end=fn_end
)
chain.invoke(2)

with_alisteners

with_alisteners(
    *,
    on_start: AsyncListener | None = None,
    on_end: AsyncListener | None = None,
    on_error: AsyncListener | None = None,
) -> Runnable[Input, Output]

将异步生命周期侦听器绑定到 Runnable。

返回一个新的 Runnable。

Run 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_time 以及添加到运行中的任何标签或元数据。

参数	描述
开始时	在 Runnable 开始运行之前异步调用，并传入 Run 对象。 类型： AsyncListener | None 默认值： None
结束时	在 Runnable 完成运行后异步调用，并传入 Run 对象。 类型： AsyncListener | None 默认值： None
出错时	如果 Runnable 抛出错误，则异步调用此函数，并传入 Run 对象。 类型： AsyncListener | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了监听器的新 Runnable。

示例

from langchain_core.runnables import RunnableLambda, Runnable
from datetime import datetime, timezone
import time
import asyncio

def format_t(timestamp: float) -> str:
    return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()

async def test_runnable(time_to_sleep: int):
    print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}")
    await asyncio.sleep(time_to_sleep)
    print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}")

async def fn_start(run_obj: Runnable):
    print(f"on start callback starts at {format_t(time.time())}")
    await asyncio.sleep(3)
    print(f"on start callback ends at {format_t(time.time())}")

async def fn_end(run_obj: Runnable):
    print(f"on end callback starts at {format_t(time.time())}")
    await asyncio.sleep(2)
    print(f"on end callback ends at {format_t(time.time())}")

runnable = RunnableLambda(test_runnable).with_alisteners(
    on_start=fn_start,
    on_end=fn_end
)
async def concurrent_runs():
    await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))

asyncio.run(concurrent_runs())
Result:
on start callback starts at 2025-03-01T07:05:22.875378+00:00
on start callback starts at 2025-03-01T07:05:22.875495+00:00
on start callback ends at 2025-03-01T07:05:25.878862+00:00
on start callback ends at 2025-03-01T07:05:25.878947+00:00
Runnable[2s]: starts at 2025-03-01T07:05:25.879392+00:00
Runnable[3s]: starts at 2025-03-01T07:05:25.879804+00:00
Runnable[2s]: ends at 2025-03-01T07:05:27.881998+00:00
on end callback starts at 2025-03-01T07:05:27.882360+00:00
Runnable[3s]: ends at 2025-03-01T07:05:28.881737+00:00
on end callback starts at 2025-03-01T07:05:28.882428+00:00
on end callback ends at 2025-03-01T07:05:29.883893+00:00
on end callback ends at 2025-03-01T07:05:30.884831+00:00

with_types

with_types(
    *, input_type: type[Input] | None = None, output_type: type[Output] | None = None
) -> Runnable[Input, Output]

将输入和输出类型绑定到 Runnable，返回一个新的 Runnable。

参数	描述
输入类型	要绑定到 Runnable 的输入类型。 类型： type[Input] | None 默认值： None
输出类型	要绑定到 Runnable 的输出类型。 类型： type[Output] | None 默认值： None

返回	描述
Runnable[Input, Output]	一个绑定了类型的新 Runnable。

with_retry

with_retry(
    *,
    retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
    wait_exponential_jitter: bool = True,
    exponential_jitter_params: ExponentialJitterParams | None = None,
    stop_after_attempt: int = 3,
) -> Runnable[Input, Output]

创建一个新的 Runnable，它在发生异常时重试原始的 Runnable。

参数	描述
如果异常类型则重试	一个用于重试的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
指数等待抖动	是否在两次重试之间的等待时间中添加抖动。 类型: bool 默认值: True
尝试后停止	放弃前尝试的最大次数。 类型： int 默认值： 3
指数抖动参数	tenacity.wait_exponential_jitter 的参数。即：initial、max、exp_base 和 jitter（均为 float 值）。 类型： ExponentialJitterParams | None 默认值： None

返回	描述
Runnable[Input, Output]	一个新的 Runnable，它会在发生异常时重试原始的 Runnable。

示例

from langchain_core.runnables import RunnableLambda

count = 0


def _lambda(x: int) -> None:
    global count
    count = count + 1
    if x == 1:
        raise ValueError("x is 1")
    else:
        pass


runnable = RunnableLambda(_lambda)
try:
    runnable.with_retry(
        stop_after_attempt=2,
        retry_if_exception_type=(ValueError,),
    ).invoke(1)
except ValueError:
    pass

assert count == 2

map

map() -> Runnable[list[Input], list[Output]]

返回一个新的 Runnable，它将输入列表映射到输出列表。

使用每个输入调用 invoke。

返回	描述
Runnable[list[Input], list[Output]]	一个新的 Runnable，它将输入列表映射到输出列表。

示例

from langchain_core.runnables import RunnableLambda


def _lambda(x: int) -> int:
    return x + 1


runnable = RunnableLambda(_lambda)
print(runnable.map().invoke([1, 2, 3]))  # [2, 3, 4]

with_fallbacks

with_fallbacks(
    fallbacks: Sequence[Runnable[Input, Output]],
    *,
    exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
    exception_key: str | None = None,
) -> RunnableWithFallbacks[Input, Output]

向 Runnable 添加回退机制，返回一个新的 Runnable。

新的 Runnable 将在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

参数	描述
备用方案	如果原始 Runnable 失败，将尝试的一系列 runnable。 类型： Sequence[Runnable[Input, Output]]
要处理的异常	一个要处理的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
异常键	如果指定了 string，则已处理的异常将作为输入的一部分，在指定的键下传递给备选方案。如果为 None，异常将不会传递给备选方案。如果使用，基础 Runnable 及其备选方案必须接受一个字典作为输入。 类型: str | None 默认值: None

返回	描述
RunnableWithFallbacks[Input, Output]	一个新的 Runnable，它会在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

示例

from typing import Iterator

from langchain_core.runnables import RunnableGenerator


def _generate_immediate_error(input: Iterator) -> Iterator[str]:
    raise ValueError()
    yield ""


def _generate(input: Iterator) -> Iterator[str]:
    yield from "foo bar"


runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
    [RunnableGenerator(_generate)]
)
print("".join(runnable.stream({})))  # foo bar

参数	描述
备用方案	如果原始 Runnable 失败，将尝试的一系列 runnable。 类型： Sequence[Runnable[Input, Output]]
要处理的异常	一个要处理的异常类型元组。 类型： tuple[type[BaseException], ...] 默认值： (Exception,)
异常键	如果指定了 string，则已处理的异常将作为输入的一部分，在指定的键下传递给备选方案。如果为 None，异常将不会传递给备选方案。如果使用，基础 Runnable 及其备选方案必须接受一个字典作为输入。 类型: str | None 默认值: None

返回	描述
RunnableWithFallbacks[Input, Output]	一个新的 Runnable，它会在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

as_tool

as_tool(
    args_schema: type[BaseModel] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    arg_types: dict[str, type] | None = None,
) -> BaseTool

从 Runnable 创建一个 BaseTool。

as_tool 将从一个 Runnable 实例化一个 BaseTool，该工具具有名称、描述和 args_schema。在可能的情况下，模式会从 runnable.get_input_schema 中推断。或者（例如，如果 Runnable 接受一个字典作为输入，并且特定的字典键没有类型），模式可以通过 args_schema 直接指定。你也可以传递 arg_types 来仅指定必需的参数及其类型。

参数	描述
参数模式	工具的模式。 类型： type[BaseModel] | None 默认值： None
name	工具的名称。 类型: str | None 默认值: None
描述	工具的描述。 类型: str | None 默认值: None
参数类型	一个从参数名称到类型的字典。 类型： dict[str, type] | None 默认值： None

返回	描述
BaseTool	一个 BaseTool 实例。

类型化字典输入

from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda


class Args(TypedDict):
    a: int
    b: list[int]


def f(x: Args) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool()
as_tool.invoke({"a": 3, "b": [1, 2]})

dict 输入，通过 args_schema 指定模式

from typing import Any
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableLambda

def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))

class FSchema(BaseModel):
    """Apply a function to an integer and list of integers."""

    a: int = Field(..., description="Integer")
    b: list[int] = Field(..., description="List of ints")

runnable = RunnableLambda(f)
as_tool = runnable.as_tool(FSchema)
as_tool.invoke({"a": 3, "b": [1, 2]})

dict 输入，通过 arg_types 指定模式

from typing import Any
from langchain_core.runnables import RunnableLambda


def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool(arg_types={"a": int, "b": list[int]})
as_tool.invoke({"a": 3, "b": [1, 2]})

字符串输入

from langchain_core.runnables import RunnableLambda


def f(x: str) -> str:
    return x + "a"


def g(x: str) -> str:
    return x + "z"


runnable = RunnableLambda(f) | g
as_tool = runnable.as_tool()
as_tool.invoke("b")

__init__

__init__(*args: Any, **kwargs: Any) -> None

is_lc_serializable 类方法

is_lc_serializable() -> bool

这个类是否可序列化？

根据设计，即使一个类继承自 Serializable，它默认也是不可序列化的。这是为了防止意外序列化不应被序列化的对象。

返回	描述
bool	类是否可序列化。默认为 False。

get_lc_namespace 类方法

get_lc_namespace() -> list[str]

获取 LangChain 对象的命名空间。

例如，如果类是 langchain.llms.openai.OpenAI，那么命名空间是 ["langchain", "llms", "openai"]

返回	描述
list[str]	命名空间。

lc_id 类方法

lc_id() -> list[str]

为此类返回一个用于序列化目的的唯一标识符。

唯一标识符是一个描述对象路径的字符串列表。

例如，对于类 langchain.llms.openai.OpenAI，id 是 ["langchain", "llms", "openai", "OpenAI"]。

to_json

to_json() -> SerializedConstructor | SerializedNotImplemented

将 Runnable 序列化为 JSON。

返回	描述
SerializedConstructor | SerializedNotImplemented	一个 Runnable 的 JSON 可序列化表示。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

序列化一个“未实现”的对象。

返回	描述
SerializedNotImplemented	SerializedNotImplemented.

configurable_fields

configurable_fields(
    **kwargs: AnyConfigurableField,
) -> RunnableSerializable[Input, Output]

在运行时配置特定的 Runnable 字段。

参数	描述
**kwargs	一个要配置的 ConfigurableField 实例的字典。 类型： AnyConfigurableField 默认值： {}

引发	描述
ValueError	如果在 Runnable 中找不到配置键。

返回	描述
RunnableSerializable[Input, Output]	一个配置了字段的新 Runnable。

from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatOpenAI(max_tokens=20).configurable_fields(
    max_tokens=ConfigurableField(
        id="output_token_number",
        name="Max tokens in the output",
        description="The maximum number of tokens in the output",
    )
)

# max_tokens = 20
print("max_tokens_20: ", model.invoke("tell me something about chess").content)

# max_tokens = 200
print(
    "max_tokens_200: ",
    model.with_config(configurable={"output_token_number": 200})
    .invoke("tell me something about chess")
    .content,
)

configurable_alternatives

configurable_alternatives(
    which: ConfigurableField,
    *,
    default_key: str = "default",
    prefix_keys: bool = False,
    **kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]],
) -> RunnableSerializable[Input, Output]

为可在运行时设置的 Runnable 对象配置备选项。

参数	描述
哪个	将用于选择备选项的 ConfigurableField 实例。 类型： ConfigurableField
默认键	如果未选择备选项，则使用的默认键。 类型： str 默认值： 'default'
前缀键	是否用 ConfigurableField id 作为键的前缀。 类型： bool 默认值： False
**kwargs	一个从键到 Runnable 实例或返回 Runnable 实例的可调用对象的字典。 类型： Runnable[Input, Output] | Callable[[], Runnable[Input, Output]] 默认值： {}

返回	描述
RunnableSerializable[Input, Output]	一个配置了备选项的新 Runnable。

from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatAnthropic(
    model_name="claude-sonnet-4-5-20250929"
).configurable_alternatives(
    ConfigurableField(id="llm"),
    default_key="anthropic",
    openai=ChatOpenAI(),
)

# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)

# uses ChatOpenAI
print(
    model.with_config(configurable={"llm": "openai"})
    .invoke("which organization created you?")
    .content
)

LangSmithParams

基类：TypedDict

用于跟踪的 LangSmith 参数。

ls_provider 实例属性

ls_provider: str

模型的提供商。

ls_model_name 实例属性

ls_model_name: str

模型的名称。

ls_model_type 实例属性

ls_model_type: Literal['chat', 'llm']

模型的类型。应为 'chat' 或 'llm'。

ls_temperature 实例属性

ls_temperature: float | None

生成的温度。

ls_max_tokens 实例属性

ls_max_tokens: int | None

生成的最大令牌数。

ls_stop 实例属性

ls_stop: list[str] | None

生成的停止词。