提示模板#

class langchain_core.prompts.prompt.PromptTemplate[来源]#

基础类：StringPromptTemplate

语言模型的提示模板。

提示模板由一个字符串模板组成。它接受用户提供的一组参数，这些参数可用于生成语言模型的提示。

模板可以使用f-strings（默认）、jinja2或mustache语法进行格式化。

Security warning:

建议使用template_format=”f-string”而不是template_format=”jinja2”，或者确保永远不要接受来自不可信来源的jinja2模板，因为它们可能导致任意的Python代码执行。

截至LangChain 0.0.329版本，Jinja2模板将默认使用Jinja2的SandboxedEnvironment进行渲染。这种沙盒化应被视为一种尽力而为的方法，而不是安全性的保证，因为它是一种选择退出的方法，而不是选择加入的方法。

尽管有沙盒机制，我们建议永远不要使用来自不可信来源的jinja2模板。

示例

from langchain_core.prompts import PromptTemplate

# Instantiation using from_template (recommended)
prompt = PromptTemplate.from_template("Say {foo}")
prompt.format(foo="bar")

# Instantiation using initializer
prompt = PromptTemplate(template="Say {foo}")

注意

PromptTemplate 实现了标准的 Runnable Interface。🏃

Runnable Interface 接口在可运行对象上提供了额外的方法，例如 with_types, with_retry, assign, bind, get_graph, 等等。

param input_types: Dict[str, Any] [Optional]#: 提示模板期望的变量类型的字典。如果未提供，则假定所有变量均为字符串。

param input_variables: list[str] [Required]#: 需要作为提示输入的变量名称列表。

param metadata: Dict[str, Any] | None = None#: 用于追踪的元数据。

param optional_variables: list[str] = []#: optional_variables: 占位符或MessagePlaceholder的可选变量名称列表。这些变量是从提示中自动推断出来的，用户无需提供它们。

param output_parser: BaseOutputParser | None = None#: 如何解析在此格式化提示上调用LLM的输出。

param partial_variables: Mapping[str, Any] [Optional]#

提示模板携带的部分变量的字典。

部分变量填充模板，这样您就不需要在每次调用提示时传递它们。

param tags: list[str] | None = None#: 用于追踪的标签。

param template: str [Required]#: 提示模板。

param template_format: PromptTemplateFormat = 'f-string'#: 提示模板的格式。选项有：'f-string'、'mustache'、'jinja2'。

param validate_template: bool = False#: 是否尝试验证模板。

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绑定的runnables效果很好。

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

Parameters:

inputs (list[Input]) – Runnable 的输入列表。
config (RunnableConfig | list[RunnableConfig] | None) – 调用Runnable时使用的配置。该配置支持标准键，如用于跟踪目的的'tags'、'metadata'，用于控制并行工作量的'max_concurrency'，以及其他键。有关更多详细信息，请参阅RunnableConfig。默认为None。
return_exceptions (bool) – 是否返回异常而不是抛出它们。默认为 False。
kwargs (Any | None) – 传递给Runnable的额外关键字参数。

Returns:

Runnable 的输出列表。

Return type:

列表[输出]

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，在它们完成时产生结果。

Parameters:

inputs (Sequence[Input]) – Runnable 的输入列表。
config (RunnableConfig | Sequence[RunnableConfig] | None) – 调用Runnable时使用的配置。该配置支持标准键，如用于跟踪目的的'tags'、'metadata'，用于控制并行工作量的'max_concurrency'，以及其他键。有关更多详细信息，请参阅RunnableConfig。默认为None。默认为None。
return_exceptions (bool) – 是否返回异常而不是抛出它们。默认为 False。
kwargs (Any | None) – 传递给Runnable的额外关键字参数。

Yields:

输入索引和Runnable输出的元组。

Return type:

AsyncIterator[元组[int, Output | 异常]]

async aformat(**kwargs: Any) → FormatOutputType#

使用输入异步格式化提示。

Parameters:: kwargs (Any) – 传递给提示模板的任何参数。
Returns:: 一个格式化的字符串。
Return type:: FormatOutputType

示例：

await prompt.aformat(variable1="foo")

async aformat_prompt(**kwargs: Any) → PromptValue#

使用输入异步格式化提示。

Parameters:: kwargs (Any) – 传递给提示模板的任何参数。
Returns:: 一个格式化的字符串。
Return type:: PromptValue

async ainvoke(input: dict, config: RunnableConfig | None = None, **kwargs: Any) → PromptValue#

异步调用提示。

Parameters:

input (dict) – 字典，提示的输入。
config (RunnableConfig | None) – RunnableConfig，用于配置提示的配置。
kwargs (Any)

Returns:

提示的输出。

Return type:

PromptValue

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

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

Parameters:

input (Input) – Runnable 的输入。
config (RunnableConfig | None) – 用于Runnable的配置。默认为None。
kwargs (Any | None) – 传递给Runnable的额外关键字参数。

Yields:

Runnable 的输出。

Return type:

AsyncIterator[Output]

async astream_events(input: Any, config: RunnableConfig | None = None, *, version: Literal['v1', '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[StandardStreamEvent | CustomStreamEvent]#

生成事件流。

用于创建一个迭代器，遍历提供实时信息的StreamEvents，包括来自中间结果的StreamEvents。

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

event: str - 事件名称的格式为
格式: on_[runnable_type]_(start|stream|end).
name: str - 生成事件的 Runnable 的名称。
run_id: str - 与给定执行相关联的随机生成的ID
发出事件的Runnable。作为父Runnable执行的一部分被调用的子Runnable会被分配其自己唯一的ID。
parent_ids: List[str] - 生成事件的父可运行对象的ID。
根可运行对象将有一个空列表。父ID的顺序是从根到直接父对象。仅适用于API的v2版本。API的v1版本将返回一个空列表。
tags: Optional[List[str]] - 生成事件的Runnable的标签
事件。
metadata: Optional[Dict[str, Any]] - Runnable的元数据
生成事件的元数据。
data: Dict[str, Any]

下表展示了一些可能由不同链发出的事件。为了简洁起见，表中省略了元数据字段。链定义已包含在表后。

注意此参考表适用于V2版本的架构。

事件	名称	块	输入	输出
on_chat_model_start	[模型名称]		{“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	[模型名称]	‘Hello’
on_llm_end	[model name]		‘你好，人类！’
链上开始	格式化文档
on_chain_stream	format_docs	“你好世界！，再见世界！”
on_chain_end	format_docs		[Document(…)]	“你好世界！，再见世界！”
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	Any	与事件相关的数据。这可以是任何内容，但我们建议使其可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}

提示:

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)

Parameters:

input (Any) – Runnable 的输入。
config (RunnableConfig | None) – 用于Runnable的配置。
version (Literal['v1', 'v2']) – 使用的模式版本，可以是 v2 或 v1。用户应使用 v2。 v1 是为了向后兼容，将在 0.4.0 版本中弃用。在 API 稳定之前不会分配默认值。自定义事件仅在 v2 中显示。
include_names (Sequence[str] | None) – 仅包含来自具有匹配名称的可运行对象的事件。
include_types (Sequence[str] | None) – 仅包含来自具有匹配类型的可运行对象的事件。
include_tags (Sequence[str] | None) – 仅包含具有匹配标签的可运行对象的事件。
exclude_names (Sequence[str] | None) – 排除具有匹配名称的可运行对象的事件。
exclude_types (Sequence[str] | None) – 排除具有匹配类型的可运行对象的事件。
exclude_tags (Sequence[str] | None) – 排除具有匹配标签的可运行对象的事件。
kwargs (Any) – 传递给 Runnable 的额外关键字参数。这些参数将传递给 astream_log，因为 astream_events 的实现是基于 astream_log 的。

Yields:

一个异步的StreamEvents流。

Raises:

NotImplementedError – 如果版本不是v1或v2。

Return type:

AsyncIterator[StandardStreamEvent | CustomStreamEvent]

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

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

batch的默认实现对于IO绑定的runnables效果很好。

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

Parameters:

inputs (列表[Input])
config (RunnableConfig | list[RunnableConfig] | None)
return_exceptions (bool)
kwargs (任意 | 无)

Return type:

列表[输出]

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，在它们完成时产生结果。

Parameters:

inputs (Sequence[Input])
config (RunnableConfig | Sequence[RunnableConfig] | None)
return_exceptions (bool)
kwargs (任意 | 无)

Return type:

Iterator[元组[int, Output | 异常]]

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

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

当链中的Runnable需要一个不在前一个Runnable输出中或用户输入中的参数时，这很有用。

Parameters:: kwargs (Any) – 绑定到Runnable的参数。
Returns:: 一个新的Runnable，参数已绑定。
Return type:: Runnable[Input, Output]

示例：

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

llm = ChatOllama(model='llama2')

# Without bind.
chain = (
    llm
    | StrOutputParser()
)

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

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

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

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

配置可以在运行时设置的Runnables的替代方案。

Parameters:

which (ConfigurableField) – 将用于选择替代方案的ConfigurableField实例。
default_key (str) – 如果没有选择其他选项，则使用的默认键。默认为“default”。
prefix_keys (bool) – 是否在键前加上 ConfigurableField 的 id。默认为 False。
**kwargs (Runnable[Input, Output] | Callable[[], Runnable[Input, Output]]) – 一个键值对字典，键对应的是Runnable实例或返回Runnable实例的可调用对象。

Returns:

一个新的Runnable，配置了替代方案。

Return type:

RunnableSerializable

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

model = ChatAnthropic(
    model_name="claude-3-sonnet-20240229"
).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
)

configurable_fields(**kwargs: ConfigurableField | ConfigurableFieldSingleOption | ConfigurableFieldMultiOption) → RunnableSerializable#

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

Parameters:: **kwargs (ConfigurableField | ConfigurableFieldSingleOption | ConfigurableFieldMultiOption) – 一个包含 ConfigurableField 实例的字典，用于配置。
Returns:: 一个新的Runnable，其字段已配置。
Return type:: RunnableSerializable

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
)

format(**kwargs: Any) → str[来源]#

使用输入格式化提示。

Parameters:: kwargs (Any) – 传递给提示模板的任何参数。
Returns:: 一个格式化的字符串。
Return type:: 字符串

format_prompt(**kwargs: Any) → PromptValue#

使用输入格式化提示。

Parameters:: kwargs (Any) – 传递给提示模板的任何参数。
Returns:: 一个格式化的字符串。
Return type:: PromptValue

classmethod from_examples(examples: list[str], suffix: str, input_variables: list[str], example_separator: str = '\n\n', prefix: str = '', **kwargs: Any) → PromptTemplate[source]#

以列表格式的示例，带有前缀和后缀来创建提示。

旨在用作从示例动态创建提示的一种方式。

Parameters:

示例 (列表[字符串]) – 用于提示中的示例列表。
suffix (str) – 字符串，用于放在示例列表之后。通常应设置用户的输入。
input_variables (list[str]) – 最终提示模板将期望的变量名称列表。
example_separator (str) – 用于在示例之间使用的分隔符。默认为两个换行符。
prefix (str) – 应该放在任何示例之前的字符串。通常包括示例。默认为空字符串。
kwargs (Any)

Returns:

生成的最终提示。

Return type:

PromptTemplate

classmethod from_file(template_file: str | Path, input_variables: list[str] | None = None, encoding: str | None = None, **kwargs: Any) → PromptTemplate[source]#

从文件加载提示。

Parameters:

template_file (str | Path) – 包含提示模板的文件的路径。
input_variables (list[str] | None) – [已弃用] 最终提示模板将期望的变量名称列表。默认为 None。
encoding (str | None) – 用于打开模板文件的编码系统。如果未提供，将使用操作系统的默认编码。
kwargs (Any)

Return type:

PromptTemplate

input_variables 被忽略，因为 from_file 现在委托给 from_template()。

Returns:

从文件加载的提示。

Parameters:

template_file (str | Path)
input_variables (列表[字符串] | 无)
encoding (str | None)
kwargs (Any)

Return type:

PromptTemplate

classmethod from_template(template: str, *, template_format: Literal['f-string', 'mustache', 'jinja2'] = 'f-string', partial_variables: dict[str, Any] | None = None, **kwargs: Any) → PromptTemplate[源代码]#

从模板加载提示模板。

Security warning:

建议使用template_format=”f-string”而不是template_format=”jinja2”，或者确保永远不要接受来自不可信来源的jinja2模板，因为它们可能导致任意的Python代码执行。

截至LangChain 0.0.329版本，Jinja2模板将默认使用Jinja2的SandboxedEnvironment进行渲染。这种沙盒化应被视为一种尽力而为的方法，而不是安全性的保证，因为它是一种选择退出的方法，而不是选择加入的方法。

尽管有沙盒机制，我们建议永远不要使用来自不可信来源的jinja2模板。

Parameters:

template (str) – 要加载的模板。
template_format (Literal['f-string', 'mustache', 'jinja2']) – 模板的格式。使用 jinja2 表示 jinja2， mustache 表示 mustache，f-string 表示 f-strings。默认为 f-string。
partial_variables (dict[str, Any] | None) –

一个可以用于部分填充模板的变量字典。
例如，如果模板是

”{variable1} {variable2}”，并且 partial_variables 是 {“variable1”: “foo”}，那么最终的提示将是 “foo {variable2}”。默认为 None。
kwargs (Any) – 传递给提示模板的任何其他参数。

Returns:

从模板加载的提示模板。

Return type:

PromptTemplate

invoke(input: dict, config: RunnableConfig | None = None, **kwargs: Any) → PromptValue#

调用提示。

Parameters:

input (dict) – 字典，提示的输入。
config (RunnableConfig | None) – RunnableConfig，用于配置提示的配置。
kwargs (Any)

Returns:

提示的输出。

Return type:

PromptValue

partial(**kwargs: str | Callable[[], str]) → BasePromptTemplate#

返回提示模板的一部分。

Parameters:: kwargs (str | Callable[[], str]) – Union[str, Callable[[], str]], 设置的部分变量。
Returns:: 提示模板的一部分。
Return type:: BasePromptTemplate

pretty_print() → None#

打印提示的漂亮表示。

Return type:: 无

pretty_repr(html: bool = False) → str#

获取提示的漂亮表示。

Parameters:: html (bool) – 是否返回HTML格式的字符串。
Returns:: 提示的漂亮表示。
Return type:: 字符串

save(file_path: Path | str) → None#

保存提示。

Parameters:

file_path (Path | str) – 保存提示的目录路径。

Raises:

ValueError – 如果提示包含部分变量。
ValueError – 如果文件路径不是json或yaml。
NotImplementedError – 如果提示类型未实现。

Return type:

无

示例： .. code-block:: python

prompt.save(file_path=”path/prompt.yaml”)

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

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

Parameters:

input (Input) – Runnable 的输入。
config (RunnableConfig | None) – 用于Runnable的配置。默认为None。
kwargs (Any | None) – 传递给Runnable的额外关键字参数。

Yields:

Runnable 的输出。

Return type:

迭代器[输出]

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

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

on_start: 在Runnable开始运行之前异步调用。 on_end: 在Runnable完成运行之后异步调用。 on_error: 如果Runnable抛出错误，则异步调用。

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

Parameters:

on_start (Optional[AsyncListener]) – 在Runnable开始运行之前异步调用。默认为None。
on_end (Optional[AsyncListener]) – 在Runnable运行结束后异步调用。默认为None。
on_error (可选[AsyncListener]) – 如果Runnable抛出错误，则异步调用。默认为None。

Returns:

一个新的Runnable，绑定了监听器。

Return type:

Runnable[Input, Output]

示例：

from langchain_core.runnables import RunnableLambda
import time

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 2024-05-16T14:20:29.637053+00:00
on start callback starts at 2024-05-16T14:20:29.637150+00:00
on start callback ends at 2024-05-16T14:20:32.638305+00:00
on start callback ends at 2024-05-16T14:20:32.638383+00:00
Runnable[3s]: starts at 2024-05-16T14:20:32.638849+00:00
Runnable[5s]: starts at 2024-05-16T14:20:32.638999+00:00
Runnable[3s]: ends at 2024-05-16T14:20:35.640016+00:00
on end callback starts at 2024-05-16T14:20:35.640534+00:00
Runnable[5s]: ends at 2024-05-16T14:20:37.640169+00:00
on end callback starts at 2024-05-16T14:20:37.640574+00:00
on end callback ends at 2024-05-16T14:20:37.640654+00:00
on end callback ends at 2024-05-16T14:20:39.641751+00:00

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

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

Parameters:

config (RunnableConfig | None) – 绑定到Runnable的配置。
kwargs (Any) – 传递给Runnable的额外关键字参数。

Returns:

一个新的Runnable，带有绑定的配置。

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]#

为Runnable添加回退，返回一个新的Runnable。

新的Runnable将尝试原始的Runnable，然后在失败时依次尝试每个回退。

Parameters:

fallbacks (Sequence[Runnable[Input, Output]]) – 如果原始 Runnable 失败，将尝试的一系列 Runnable。
exceptions_to_handle (tuple[type[BaseException], ...]) – 要处理的异常类型的元组。默认为 (Exception,)。
exception_key (Optional[str]) – 如果指定了字符串，则处理的异常将作为输入的一部分传递给后备函数，使用指定的键。如果为 None，异常将不会传递给后备函数。如果使用此参数，基础 Runnable 及其后备函数必须接受字典作为输入。默认为 None。

Returns:

一个新的Runnable，它将在失败时尝试原始的Runnable，然后依次尝试每个回退。

Return type:

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

Parameters:

fallbacks (Sequence[Runnable[Input, Output]]) – 如果原始 Runnable 失败，将尝试的一系列 Runnable。
exceptions_to_handle (tuple[type[BaseException], ...]) – 要处理的异常类型的元组。
exception_key (Optional[str]) – 如果指定了字符串，则处理的异常将作为输入的一部分传递给后备函数，使用指定的键。如果为 None，异常将不会传递给后备函数。如果使用此参数，基础 Runnable 及其后备函数必须接受字典作为输入。

Returns:

一个新的Runnable，它将在失败时尝试原始的Runnable，然后依次尝试每个回退。

Return type:

RunnableWithFallbacksT[Input, Output]

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。

on_start: 在Runnable开始运行之前调用，带有Run对象。 on_end: 在Runnable完成运行之后调用，带有Run对象。 on_error: 如果Runnable抛出错误时调用，带有Run对象。

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

Parameters:

on_start (Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]]) – 在Runnable开始运行之前调用。默认为None。
on_end (可选[联合[可调用[[运行], 无], 可调用[[运行, RunnableConfig], 无]]]) – 在Runnable运行结束后调用。默认为无。
on_error (可选[联合[可调用[[运行], 无], 可调用[[运行, RunnableConfig], 无]]]) – 如果Runnable抛出错误时调用。默认为无。

Returns:

一个新的Runnable，绑定了监听器。

Return type:

Runnable[Input, Output]

示例：

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_retry(*, retry_if_exception_type: tuple[type[BaseException], ...] = (<class 'Exception'>,), wait_exponential_jitter: bool = True, stop_after_attempt: int = 3) → Runnable[Input, Output]#

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

Parameters:

retry_if_exception_type (tuple[type[BaseException], ...]) – 一个异常类型的元组，用于重试。默认值为 (Exception,)。
wait_exponential_jitter (bool) – 是否在重试之间的等待时间中添加抖动。默认为 True。
stop_after_attempt (int) – 在放弃之前尝试的最大次数。默认为3。

Returns:

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

Return type:

Runnable[Input, Output]

示例：

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)

Parameters:

retry_if_exception_type (tuple[type[BaseException], ...]) – 一个异常类型的元组，用于在发生这些异常时重试
wait_exponential_jitter (bool) – 是否在重试之间为等待时间添加抖动
stop_after_attempt (int) – 在放弃之前尝试的最大次数

Returns:

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

Return type:

Runnable[Input, Output]

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

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

Parameters:

input_type (type[Input] | None) – 要绑定到Runnable的输入类型。默认为None。
output_type (type[Output] | None) – 要绑定到Runnable的输出类型。默认为None。

Returns:

一个带有类型绑定的新Runnable。

Return type:

Runnable[Input, Output]

使用 PromptTemplate 的示例