langchain_core.runnables.base.RunnableGenerator

class langchain_core.runnables.base.RunnableGenerator(transform: Union[Callable[[Iterator[Input]], Iterator[Output]], Callable[[AsyncIterator[Input]], AsyncIterator[Output]]], atransform: Optional[Callable[[AsyncIterator[Input]], AsyncIterator[Output]]] = None)

可运行的生成器函数。

RunnableGenerators可以直接实例化，也可以通过在序列中使用生成器来实例化。

RunnableGenerators可用于实现自定义行为，例如自定义输出解析器，同时保留流式处理能力。给定一个具有签名Iterator[A] -> Iterator[B]的生成器函数，将其包装在RunnableGenerator中允许它在从上一步流入时立即发出输出块。

请注意，如果生成器函数具有签名A -> Iterator[B]，例如它需要在发出块之前完成来自上一步的输入（例如，大多数LLM需要完整的提示才能开始生成），则可以将其包装在RunnableLambda中。

以下是一个示例，展示了RunnableGenerator的基本机制：

from typing import Any, AsyncIterator, Iterator

from langchain_core.runnables import RunnableGenerator


def gen(input: Iterator[Any]) -> Iterator[str]:
    for token in ["Have", " a", " nice", " day"]:
        yield token


runnable = RunnableGenerator(gen)
runnable.invoke(None)  # "Have a nice day"
list(runnable.stream(None))  # ["Have", " a", " nice", " day"]
runnable.batch([None, None])  # ["Have a nice day", "Have a nice day"]


# 异步版本：
async def agen(input: AsyncIterator[Any]) -> AsyncIterator[str]:
    for token in ["Have", " a", " nice", " day"]:
        yield token

runnable = RunnableGenerator(agen)
await runnable.ainvoke(None)  # "Have a nice day"
[p async for p in runnable.astream(None)] # ["Have", " a", " nice", " day"]

RunnableGenerator使得在流式上下文中实现自定义行为变得容易。下面我们展示一个示例：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableGenerator, RunnableLambda
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser


model = ChatOpenAI()
chant_chain = (
    ChatPromptTemplate.from_template("Give me a 3 word chant about {topic}")
    | model
    | StrOutputParser()
)

def character_generator(input: Iterator[str]) -> Iterator[str]:
    for token in input:
        if "," in token or "." in token:
            yield "👏" + token
        else:
            yield token


runnable = chant_chain | character_generator
assert type(runnable.last) is RunnableGenerator
"".join(runnable.stream({"topic": "waste"})) # Reduce👏, Reuse👏, Recycle👏.

# 请注意，RunnableLambda可用于延迟序列中一步的流式处理，直到上一步完成：
def reverse_generator(input: str) -> Iterator[str]:
    # 以相反的顺序产生输入的字符。
    for character in input[::-1]:
        yield character

runnable = chant_chain | RunnableLambda(reverse_generator)
"".join(runnable.stream({"topic": "waste"}))  # ".elcycer ,esuer ,ecudeR"

Attributes

InputType	这个可运行程序接受的输入类型，以类型注释的形式指定。
OutputType	此可运行程序生成的输出类型，指定为类型注释。
config_specs	列出此可运行程序的可配置字段。
input_schema	此可运行程序接受的输入类型，由pydantic模型指定。
name	可运行对象的名称。用于调试和跟踪。
output_schema	这个可运行程序产生的输出类型被指定为一个pydantic模型。

Methods

__init__(transform[, atransform])
abatch(inputs[, config, return_exceptions])	默认实现使用asyncio.gather并行运行调用。
abatch_as_completed(inputs[, config, ...])	在并行上运行一组输入的 ainvoke，完成后产生结果。
ainvoke(input[, config])	默认的ainvoke实现，从一个线程中调用invoke。
assign(**kwargs)	为这个可运行程序的字典输出分配新的字段。 返回一个新的可运行程序。
astream(input[, config])	默认实现astream，调用ainvoke。 如果子类支持流式输出，则应该重写此方法。
astream_events(input[, config, ...])	[Beta] 生成事件流。
astream_log(input[, config, diff, ...])	从可运行对象中流式传输所有输出，如回调系统所报告的。 这包括LLMs、检索器、工具等的所有内部运行。
atransform(input[, config])	默认实现一个转换的方法，它会缓冲输入并调用流。 如果子类可以在生成输入时开始产生输出，应该重写这个方法。
batch(inputs[, config, return_exceptions])	默认实现使用线程池执行程序并行运行invoke。
batch_as_completed(inputs[, config, ...])	在输入列表上并行运行invoke，当它们完成时产生结果。
bind(**kwargs)	将参数绑定到一个Runnable，返回一个新的Runnable。
config_schema(*[, include])	此可运行程序接受的配置类型，指定为一个pydantic模型。
get_graph([config])	返回此可运行对象的图形表示。
get_input_schema([config])	获取一个Pydantic模型，用于验证可运行对象的输入。
get_name([suffix, name])	获取可运行对象的名称。
get_output_schema([config])	获取一个Pydantic模型，用于验证可运行代码的输出。
get_prompts([config])
invoke(input[, config])	将单个输入转换为输出。覆盖以实现。
map()	返回一个新的Runnable，它将一个输入列表映射到一个输出列表，通过对每个输入调用invoke()。
pick(keys)	从可运行的字典输出中选择键。
pipe(*others[, name])	将这个Runnable与类似Runnable的对象组合起来，形成一个RunnableSequence。
stream(input[, config])	默认的流实现，调用invoke方法。 如果子类支持流式输出，应该重写这个方法。
transform(input[, config])	transform的默认实现，它会缓冲输入然后调用stream。如果子类能够在生成输入的同时开始生成输出，那么应该重写这个方法。
with_config([config])	将配置绑定到一个可运行对象，返回一个新的可运行对象。
with_fallbacks(fallbacks, *[, ...])	为可运行对象添加回退，返回一个新的可运行对象。
with_listeners(*[, on_start, on_end, on_error])	将生命周期监听器绑定到一个Runnable，返回一个新的Runnable。
with_retry(*[, retry_if_exception_type, ...])	创建一个新的Runnable，在发生异常时重试原始的runnable。
with_types(*[, input_type, output_type])	将输入和输出类型绑定到一个Runnable，返回一个新的Runnable。

Parameters

• transform (Union[Callable[[Iterator[Input]], Iterator[Output]], Callable[[AsyncIterator[Input]], AsyncIterator[Output]]]) –

• atransform (Optional[Callable[[AsyncIterator[Input]], AsyncIterator[Output]]]) –

Return type

None

__init__(transform: Union[Callable[[Iterator[Input]], Iterator[Output]], Callable[[AsyncIterator[Input]], AsyncIterator[Output]]], atransform: Optional[Callable[[AsyncIterator[Input]], AsyncIterator[Output]]] = None) → None

Parameters

• transform (Union[Callable[[Iterator[Input]], Iterator[Output]], Callable[[AsyncIterator[Input]], AsyncIterator[Output]]]) –

• atransform (Optional[Callable[[AsyncIterator[Input]], AsyncIterator[Output]]]) –

Return type

None

async abatch(inputs: List[Input], config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) → List[Output]

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

对于IO绑定的可运行对象，默认的批处理实现效果很好。

如果子类可以更有效地批处理，应该重写这个方法； 例如，如果底层可运行对象使用支持批处理模式的API。

Parameters

• inputs (List[Input]) –

• config (Optional[Union[RunnableConfig, List[RunnableConfig]]]) –

• return_exceptions (bool) –

• kwargs (Optional[Any]) –

Return type

List[Output]

async abatch_as_completed(inputs: List[Input], config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) → AsyncIterator[Tuple[int, Union[Output, Exception]]]

在并行上运行一组输入的 ainvoke，完成后产生结果。

Parameters

• inputs (List[Input]) –

• config (Optional[Union[RunnableConfig, List[RunnableConfig]]]) –

• return_exceptions (bool) –

• kwargs (Optional[Any]) –

Return type

AsyncIterator[Tuple[int, Union[Output, Exception]]]

async ainvoke(input: Input, config: Optional[RunnableConfig] = None, **kwargs: Any) → Output

默认的ainvoke实现，从一个线程中调用invoke。

默认实现允许即使runnable没有实现invoke的本机异步版本，也可以使用异步代码。

如果子类可以异步运行，应该重写这个方法。

Parameters

• input (Input) –

• config (Optional[RunnableConfig]) –

• kwargs (Any) –

Return type

Output

assign(**kwargs: Union[Runnable[Dict[str, Any], Any], Callable[[Dict[str, Any]], Any], Mapping[str, Union[Runnable[Dict[str, Any], Any], Callable[[Dict[str, Any]], Any]]]]) → RunnableSerializable[Any, Any]

为这个可运行程序的字典输出分配新的字段。 返回一个新的可运行程序。

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}"
)
llm = FakeStreamingListLLM(responses=["foo-lish"])

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

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

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

Parameters

kwargs (Union[Runnable[Dict[str, Any], Any], Callable[[Dict[str, Any]], Any], Mapping[str, Union[Runnable[Dict[str, Any], Any], Callable[[Dict[str, Any]], Any]]]]) –

Return type

RunnableSerializable[Any, Any]

astream(input: Input, config: Optional[RunnableConfig] = None, **kwargs: Any) → AsyncIterator[Output]

默认实现astream，调用ainvoke。 如果子类支持流式输出，则应该重写此方法。

Parameters

• input (Input) –

• config (Optional[RunnableConfig]) –

• kwargs (Any) –

Return type

AsyncIterator[Output]

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

[Beta] 生成事件流。

用于创建一个迭代器，遍历StreamEvents，提供有关可运行对象进展的实时信息，包括来自中间结果的StreamEvents。

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

• event: str - 事件名称的格式为：on_[runnable_type]_(start|stream|end)。

• name: str - 生成事件的可运行对象的名称。

• run_id: str - 与生成事件的可运行对象的执行关联的随机生成的ID。作为父可运行对象执行的一部分调用的子可运行对象被分配其自己的唯一ID。

• tags: Optional[List[str]] - 生成事件的可运行对象的标签。

• metadata: Optional[Dict[str, Any]] - 生成事件的可运行对象的元数据。

• data: Dict[str, Any]

下面是一个表格，展示了各种链可能发出的一些事件。为简洁起见，表格中省略了元数据字段。表格后包含了链定义。

注意 此参考表格适用于模式的V2版本。

event	name	chunk	input	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]	‘Hello’
on_llm_end	[model name]		‘Hello human!’
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, …])

以下是与上述事件相关的声明：

format_docs:

def format_docs(docs: List[Document]) -> str:
    '''格式化文档。'''
    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")
]

# 将产生以下事件（为简洁起见，省略了run_id）：
[
    {
        "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": [],
    },
]

参数:

input: 可运行对象的输入。 config: 可运行对象的配置。 version: 要使用的模式版本，可以是`v2`或`v1`。用户应该使用`v2`。`v1`用于向后兼容，将在0.4.0中弃用。在API稳定之前不会分配默认值。 include_names: 仅包括与名称匹配的可运行对象的事件。 include_types: 仅包括与类型匹配的可运行对象的事件。 include_tags: 仅包括与标签匹配的可运行对象的事件。 exclude_names: 排除与名称匹配的可运行对象的事件。 exclude_types: 排除与类型匹配的可运行对象的事件。 exclude_tags: 排除与标签匹配的可运行对象的事件。 kwargs: 要传递给可运行对象的其他关键字参数。这些参数将传递给astream_log，因为此astream_events的实现是基于astream_log构建的。

返回:

一个StreamEvents的异步流。

Notes

Parameters

• input (Any) –

• config (Optional[RunnableConfig]) –

• version (Literal['v1', 'v2']) –

• include_names (Optional[Sequence[str]]) –

• include_types (Optional[Sequence[str]]) –

• include_tags (Optional[Sequence[str]]) –

• exclude_names (Optional[Sequence[str]]) –

• exclude_types (Optional[Sequence[str]]) –

• exclude_tags (Optional[Sequence[str]]) –

• kwargs (Any) –

Return type

AsyncIterator[StreamEvent]

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

从可运行对象中流式传输所有输出，如回调系统所报告的。 这包括LLMs、检索器、工具等的所有内部运行。

输出被流式传输为Log对象，其中包括描述每个步骤中运行状态如何改变的jsonpatch操作列表，以及运行的最终状态。

jsonpatch操作可以按顺序应用以构建状态。

参数：

input：可运行对象的输入。 config：用于可运行对象的配置。 diff：是否在每个步骤之间生成差异，或者生成当前状态。 with_streamed_output_list：是否生成流式输出列表。 include_names：仅包括具有这些名称的日志。 include_types：仅包括具有这些类型的日志。 include_tags：仅包括具有这些标签的日志。 exclude_names：排除具有这些名称的日志。 exclude_types：排除具有这些类型的日志。 exclude_tags：排除具有这些标签的日志。

Parameters

• input (Any) –

• config (Optional[RunnableConfig]) –

• diff (bool) –

• with_streamed_output_list (bool) –

• include_names (Optional[Sequence[str]]) –

• include_types (Optional[Sequence[str]]) –

• include_tags (Optional[Sequence[str]]) –

• exclude_names (Optional[Sequence[str]]) –

• exclude_types (Optional[Sequence[str]]) –

• exclude_tags (Optional[Sequence[str]]) –

• kwargs (Any) –

Return type

Union[AsyncIterator[RunLogPatch], AsyncIterator[RunLog]]

atransform(input: AsyncIterator[Input], config: Optional[RunnableConfig] = None, **kwargs: Any) → AsyncIterator[Output]

默认实现一个转换的方法，它会缓冲输入并调用流。 如果子类可以在生成输入时开始产生输出，应该重写这个方法。

Parameters

• input (AsyncIterator[Input]) –

• config (Optional[RunnableConfig]) –

• kwargs (Any) –

Return type

AsyncIterator[Output]

batch(inputs: List[Input], config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) → List[Output]

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

默认的batch实现适用于IO绑定的可运行程序。

如果子类可以更有效地批处理，则应该重写此方法； 例如，如果底层可运行程序使用支持批处理模式的API。

Parameters

• inputs (List[Input]) –

• config (Optional[Union[RunnableConfig, List[RunnableConfig]]]) –

• return_exceptions (bool) –

• kwargs (Optional[Any]) –

Return type

List[Output]

batch_as_completed(inputs: List[Input], config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) → Iterator[Tuple[int, Union[Output, Exception]]]

在输入列表上并行运行invoke，当它们完成时产生结果。

Parameters

• inputs (List[Input]) –

• config (Optional[Union[RunnableConfig, List[RunnableConfig]]]) –

• return_exceptions (bool) –

• kwargs (Optional[Any]) –

Return type

Iterator[Tuple[int, Union[Output, Exception]]]

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

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

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

示例：

from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import StrOutputParser

llm = ChatOllama(model='llama2')

# 没有绑定。
chain = (
    llm
    | StrOutputParser()
)

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# 输出为 'One two three four five.'

# 有绑定。
chain = (
    llm.bind(stop=["three"])
    | StrOutputParser()
)

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# 输出为 'One two'

Parameters

kwargs (Any) –

Return type

Runnable[Input, Output]

config_schema(*, include: Optional[Sequence[str]] = None) → Type[BaseModel]

此可运行程序接受的配置类型，指定为一个pydantic模型。

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

参数：

include: 要包含在配置模式中的字段列表。

返回：

一个可以用于验证配置的pydantic模型。

Parameters

include (Optional[Sequence[str]]) –

Return type

Type[BaseModel]

get_graph(config: Optional[RunnableConfig] = None) → Graph

返回此可运行对象的图形表示。

Parameters

config (Optional[RunnableConfig]) –

Return type

Graph

get_input_schema(config: Optional[RunnableConfig] = None) → Type[BaseModel]

获取一个Pydantic模型，用于验证可运行对象的输入。

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

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

参数：

config：生成模式时要使用的配置。

返回：

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

Parameters

config (Optional[RunnableConfig]) –

Return type

Type[BaseModel]

get_name(suffix: Optional[str] = None, *, name: Optional[str] = None) → str

获取可运行对象的名称。

Parameters

• suffix (Optional[str]) –

• name (Optional[str]) –

Return type

str

get_output_schema(config: Optional[RunnableConfig] = None) → Type[BaseModel]

获取一个Pydantic模型，用于验证可运行代码的输出。

利用configurable_fields和configurable_alternatives方法的可运行代码将具有动态输出模式，取决于调用可运行代码时使用的配置。

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

参数：

config: 生成模式时要使用的配置。

返回：

一个Pydantic模型，可用于验证输出。

Parameters

config (Optional[RunnableConfig]) –

Return type

Type[BaseModel]

get_prompts(config: Optional[RunnableConfig] = None) → List[BasePromptTemplate]

Parameters

config (Optional[RunnableConfig]) –

Return type

List[BasePromptTemplate]

invoke(input: Input, config: Optional[RunnableConfig] = None, **kwargs: Any) → Output

将单个输入转换为输出。覆盖以实现。

参数：

input: 可运行的输入。 config: 在调用可运行对象时要使用的配置。

该配置支持标准键，如’tags’，’metadata’用于跟踪目的， ‘max_concurrency’用于控制并行执行的工作量， 以及其他键。请参考RunnableConfig获取更多详细信息。

返回：

可运行对象的输出。

Parameters

• input (Input) –

• config (Optional[RunnableConfig]) –

• kwargs (Any) –

Return type

Output

map() → Runnable[List[Input], List[Output]]

返回一个新的Runnable，它将一个输入列表映射到一个输出列表，通过对每个输入调用invoke()。

示例：

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]

Return type

Runnable[List[Input], List[Output]]

pick(keys: Union[str, List[str]]) → RunnableSerializable[Any, Any]

从可运行的字典输出中选择键。

选择单个键：

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]"}

Parameters

keys (Union[str, List[str]]) –

Return type

RunnableSerializable[Any, Any]

pipe(*others: Union[Runnable[Any, Other], Callable[[Any], Other]], name: Optional[str] = None) → RunnableSerializable[Input, Other]

将这个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)
# 或者等价地：
# 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]

Parameters

• others (Union[Runnable[Any, Other], Callable[[Any], Other]]) –

• name (Optional[str]) –

Return type

RunnableSerializable[Input, Other]

stream(input: Input, config: Optional[RunnableConfig] = None, **kwargs: Any) → Iterator[Output]

默认的流实现，调用invoke方法。 如果子类支持流式输出，应该重写这个方法。

Parameters

• input (Input) –

• config (Optional[RunnableConfig]) –

• kwargs (Any) –

Return type

Iterator[Output]

transform(input: Iterator[Input], config: Optional[RunnableConfig] = None, **kwargs: Any) → Iterator[Output]

transform的默认实现，它会缓冲输入然后调用stream。如果子类能够在生成输入的同时开始生成输出，那么应该重写这个方法。

Parameters

• input (Iterator[Input]) –

• config (Optional[RunnableConfig]) –

• kwargs (Any) –

Return type

Iterator[Output]

with_config(config: Optional[RunnableConfig] = None, **kwargs: Any) → Runnable[Input, Output]

将配置绑定到一个可运行对象，返回一个新的可运行对象。

Parameters

• config (Optional[RunnableConfig]) –

• kwargs (Any) –

Return type

Runnable[Input, Output]

with_fallbacks(fallbacks: Sequence[Runnable[Input, Output]], *, exceptions_to_handle: Tuple[Type[BaseException], ...] = (<class 'Exception'>,), exception_key: Optional[str] = None) → RunnableWithFallbacksT[Input, Output]

为可运行对象添加回退，返回一个新的可运行对象。

示例：

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: 一系列可运行对象，如果原始可运行对象失败，则尝试。 exceptions_to_handle: 要处理的异常类型元组。 exception_key: 如果指定了字符串，则处理的异常将作为输入的一部分传递给回退，指定的键下。如果为None，则异常将不会传递给回退。如果使用，基本可运行对象及其回退必须接受字典作为输入。

返回：

一个新的可运行对象，将尝试原始可运行对象，然后按顺序尝试每个回退，以处理失败情况。

Parameters

• fallbacks (Sequence[Runnable[Input, Output]]) –

• exceptions_to_handle (Tuple[Type[BaseException], ...]) –

• exception_key (Optional[str]) –

Return type

RunnableWithFallbacksT[Input, Output]

with_listeners(*, on_start: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None, on_end: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None, on_error: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None) → Runnable[Input, Output]

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

on_start: 在runnable开始运行之前调用，使用Run对象。 on_end: 在runnable完成运行后调用，使用Run对象。 on_error: 如果runnable抛出错误，则调用，使用Run对象。

Run对象包含有关运行的信息，包括其id、类型、输入、输出、错误、开始时间、结束时间以及添加到运行中的任何标签或元数据。

示例：

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)

Parameters

• on_start (Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]]) –

• on_end (Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]]) –

• on_error (Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]]) –

Return type

Runnable[Input, Output]

with_retry(*, retry_if_exception_type: ~typing.Tuple[~typing.Type[BaseException], ...] = (<class 'Exception'>,), wait_exponential_jitter: bool = True, stop_after_attempt: int = 3) → 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)

参数:

retry_if_exception_type: 重试的异常类型元组 wait_exponential_jitter: 是否在重试之间的等待时间中添加抖动 stop_after_attempt: 放弃之前尝试的最大次数

返回:

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

Parameters

• retry_if_exception_type (Tuple[Type[BaseException], ...]) –

• wait_exponential_jitter (bool) –

• stop_after_attempt (int) –

Return type

Runnable[Input, Output]

with_types(*, input_type: Optional[Type[Input]] = None, output_type: Optional[Type[Output]] = None) → Runnable[Input, Output]

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

Parameters

• input_type (Optional[Type[Input]]) –

• output_type (Optional[Type[Output]]) –

Return type

Runnable[Input, Output]

Examples using RunnableGenerator

• The [bool] desribes a parameterization of a generic.