langchain.chains.qa_with_sources.retrieval.RetrievalQAWithSourcesChain

class langchain.chains.qa_with_sources.retrieval.RetrievalQAWithSourcesChain

Bases: BaseQAWithSourcesChain

在索引上使用来源进行问答。

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

param callback_manager: Optional[BaseCallbackManager] = None

[已弃用] 使用`callbacks`代替。

param callbacks: Callbacks = None

可选的回调处理程序列表（或回调管理器）。默认为None。 回调处理程序在调用链的整个生命周期中被调用，从on_chain_start开始，到on_chain_end或on_chain_error结束。 每个自定义链可以选择性地调用其他回调方法，请参阅回调文档以获取完整详情。

param combine_documents_chain: BaseCombineDocumentsChain [Required]

用于合并文档的链。

param max_tokens_limit: int = 3375

根据令牌限制从存储返回的文档， 仅对StuffDocumentChain强制执行，如果reduce_k_below_max_tokens设置为true。

param memory: Optional[BaseMemory] = None

可选的内存对象。默认值为None。 Memory是一个在每个链的开始和结束时被调用的类。在开始时，内存加载变量并在链中传递它们。在结束时，它保存任何返回的变量。 有许多不同类型的内存 - 请参阅内存文档以获取完整目录。

param metadata: Optional[Dict[str, Any]] = None

与链相关的可选元数据。默认为None。 此元数据将与对该链的每次调用相关联， 并作为参数传递给在`callbacks`中定义的处理程序。 您可以使用这些来识别链的特定实例及其用例。

param reduce_k_below_max_tokens: bool = False

根据令牌限制减少从存储中返回的结果数量

param retriever: langchain_core.retrievers.BaseRetriever [Required]

连接的索引。

param return_source_documents: bool = False

返回源文件。

param tags: Optional[List[str]] = None

可选的与链相关联的标签列表。默认为None。 这些标签将与对该链的每次调用相关联， 并作为参数传递给在`callbacks`中定义的处理程序。 您可以使用这些标签来识别链的特定实例及其用例。

param verbose: bool [Optional]

是否在详细模式下运行。在详细模式下，一些中间日志将被打印到控制台。默认为全局`verbose`值，可通过`langchain.globals.get_verbose()`访问。

__call__(inputs: Union[Dict[str, Any], Any], return_only_outputs: bool = False, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, include_run_info: bool = False) → Dict[str, Any]

[Deprecated] 执行链。

参数:

inputs: 输入的字典，或者如果链只需要一个参数，则为单个输入。应包含在`Chain.input_keys`中指定的所有输入，除了将由链的内存设置的输入。 return_only_outputs: 是否仅在响应中返回输出。如果为True，则仅返回此链生成的新键。如果为False，则返回输入键和此链生成的新键。默认为False。 callbacks: 用于此链运行的回调。除了在构造期间传递给链的回调之外，还将调用这些回调，但是只有这些运行时回调将传播到对其他对象的调用。 tags: 要传递给所有回调的字符串标签列表。这些将被传递给链在构造期间传递的标签之外，但是只有这些运行时标签将传播到对其他对象的调用。 metadata: 与链关联的可选元数据。默认为None include_run_info: 是否在响应中包含运行信息。默认为False。

返回:

一个命名输出的字典。应包含在`Chain.output_keys`中指定的所有输出。

Notes

Deprecated since version langchain==0.1.0: Use invoke instead.

Parameters

• inputs (Union[Dict[str, Any], Any]) –

• return_only_outputs (bool) –

• callbacks (Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]]) –

• tags (Optional[List[str]]) –

• metadata (Optional[Dict[str, Any]]) –

• run_name (Optional[str]) –

• include_run_info (bool) –

Return type

Dict[str, Any]

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 acall(inputs: Union[Dict[str, Any], Any], return_only_outputs: bool = False, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, include_run_info: bool = False) → Dict[str, Any]

[Deprecated] 异步执行链。

参数：

inputs：输入的字典，或者如果链只需要一个参数，则为单个输入。应包含在`Chain.input_keys`中指定的所有输入，除了将由链的内存设置的输入。 return_only_outputs：是否仅在响应中返回输出。如果为True，则仅返回此链生成的新键。如果为False，则返回输入键和此链生成的新键。默认为False。 callbacks：用于此链运行的回调。除了在构造期间传递给链的回调之外，还将调用这些回调，但是只有这些运行时回调将传播到对其他对象的调用。 tags：要传递给所有回调的字符串标签列表。这些将被传递给链在构造期间传递的标签之外，但是只有这些运行时标签将传播到对其他对象的调用。 metadata：与链关联的可选元数据。默认为None。 include_run_info：是否在响应中包含运行信息。默认为False。

返回：

命名输出的字典。应包含在`Chain.output_keys`中指定的所有输出。

Notes

Deprecated since version langchain==0.1.0: Use ainvoke instead.

Parameters

• inputs (Union[Dict[str, Any], Any]) –

• return_only_outputs (bool) –

• callbacks (Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]]) –

• tags (Optional[List[str]]) –

• metadata (Optional[Dict[str, Any]]) –

• run_name (Optional[str]) –

• include_run_info (bool) –

Return type

Dict[str, Any]

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

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

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

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

Parameters

• input (Dict[str, Any]) –

• config (Optional[RunnableConfig]) –

• kwargs (Any) –

Return type

Dict[str, Any]

apply(input_list: List[Dict[str, Any]], callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None) → List[Dict[str, str]]

[Deprecated] 对列表中的所有输入调用链。

Notes

Deprecated since version langchain==0.1.0: Use batch instead.

Parameters

• input_list (List[Dict[str, Any]]) –

• callbacks (Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]]) –

Return type

List[Dict[str, str]]

async aprep_inputs(inputs: Union[Dict[str, Any], Any]) → Dict[str, str]

准备链的输入，包括从内存中添加输入。

参数:

inputs: 原始输入的字典，或者如果链只需要一个参数，则为单个输入。应包含`Chain.input_keys`中指定的所有输入，除了将由链的内存设置的输入。

返回:

包括链的内存添加的所有输入的字典。

Parameters

inputs (Union[Dict[str, Any], Any]) –

Return type

Dict[str, str]

async aprep_outputs(inputs: Dict[str, str], outputs: Dict[str, str], return_only_outputs: bool = False) → Dict[str, str]

验证和准备链输出，并将此运行的信息保存到内存中。

参数：

inputs：包括链内存中添加的任何输入的链输入字典。 outputs：初始链输出字典。 return_only_outputs：是否仅返回链输出。如果为False，则还将输入添加到最终输出中。

返回：

最终链输出的字典。

Parameters

• inputs (Dict[str, str]) –

• outputs (Dict[str, str]) –

• return_only_outputs (bool) –

Return type

Dict[str, str]

async arun(*args: Any, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) → Any

[Deprecated] 方便执行链的方法。

这个方法和`Chain.__call__`方法的主要区别在于，这个方法期望直接将输入作为位置参数或关键字参数传递，而`Chain.__call__`方法期望一个包含所有输入的单个输入字典。

参数：

*args: 如果链期望单个输入，可以作为唯一的位置参数传入。 callbacks: 用于此链运行的回调。这些回调将会被调用，除了在构建期间传递给链的回调之外，但是只有这些运行时回调会传播到对其他对象的调用。 tags: 要传递给所有回调的字符串标签列表。这些标签将会被传递，除了在构建期间传递给链的标签之外，但是只有这些运行时标签会传播到对其他对象的调用。 **kwargs: 如果链期望多个输入，它们可以直接作为关键字参数传入。

返回值：

链的输出。

示例：

# 假设我们有一个单输入链，接受一个'question'字符串：
await chain.arun("What's the temperature in Boise, Idaho?")
# -> "The temperature in Boise is..."

# 假设我们有一个多输入链，接受一个'question'字符串和一个'context'字符串：
question = "What's the temperature in Boise, Idaho?"
context = "Weather report for Boise, Idaho on 07/03/23..."
await chain.arun(question=question, context=context)
# -> "The temperature in Boise is..."

Notes

Deprecated since version langchain==0.1.0: Use ainvoke instead.

Parameters

• args (Any) –

• callbacks (Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]]) –

• tags (Optional[List[str]]) –

• metadata (Optional[Dict[str, Any]]) –

• kwargs (Any) –

Return type

Any

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]

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

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

Parameters

• input (Input) –

• config (Optional[RunnableConfig]) –

• kwargs (Optional[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]]

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

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

Parameters

• input (AsyncIterator[Input]) –

• config (Optional[RunnableConfig]) –

• kwargs (Optional[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]

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

为可以在运行时设置的可运行项配置备选项。

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()
)

# 使用默认模型ChatAnthropic
print(model.invoke("which organization created you?").content)

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

Parameters

• which (ConfigurableField) –

• default_key (str) –

• prefix_keys (bool) –

• kwargs (Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]]) –

Return type

RunnableSerializable[Input, Output]

configurable_fields(**kwargs: Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) → RunnableSerializable[Input, Output]

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

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="输出中的最大标记数",
        description="输出中的最大标记数",
    )
)

# max_tokens = 20
print(
    "max_tokens_20: ",
    model.invoke("告诉我一些关于国际象棋的事情").content
)

# max_tokens = 200
print("max_tokens_200: ", model.with_config(
    configurable={"output_token_number": 200}
    ).invoke("告诉我一些关于国际象棋的事情").content
)

Parameters

kwargs (Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) –

Return type

RunnableSerializable[Input, Output]

classmethod construct(_fields_set: Optional[SetStr] = None, **values: Any) → Model

Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values

Parameters

• _fields_set (Optional[SetStr]) –

• values (Any) –

Return type

Model

copy(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, update: Optional[DictStrAny] = None, deep: bool = False) → Model

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters

• include (Optional[Union[AbstractSetIntStr, MappingIntStrAny]]) – fields to include in new model

• exclude (Optional[Union[AbstractSetIntStr, MappingIntStrAny]]) – fields to exclude from new model, as with values this takes precedence over include

• update (Optional[DictStrAny]) – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data

• deep (bool) – set to True to make a deep copy of the model

• self (Model) –

Returns

new model instance

Return type

Model

dict(**kwargs: Any) → Dict

链的字典表示。

期望`Chain._chain_type`属性被实现，并且内存为空。

参数：

**kwargs: 传递给默认`pydantic.BaseModel.dict`方法的关键字参数。

返回：

链的字典表示。

示例：

chain.dict(exclude_unset=True)
# -> {"_type": "foo", "verbose": False, ...}

Parameters

kwargs (Any) –

Return type

Dict

classmethod from_chain_type(llm: BaseLanguageModel, chain_type: str = 'stuff', chain_type_kwargs: Optional[dict] = None, **kwargs: Any) → BaseQAWithSourcesChain

从链类型加载链。

Parameters

• llm (BaseLanguageModel) –

• chain_type (str) –

• chain_type_kwargs (Optional[dict]) –

• kwargs (Any) –

Return type

BaseQAWithSourcesChain

classmethod from_llm(llm: BaseLanguageModel, document_prompt: BasePromptTemplate = PromptTemplate(input_variables=['page_content', 'source'], template='Content: {page_content}\nSource: {source}'), question_prompt: BasePromptTemplate = PromptTemplate(input_variables=['context', 'question'], template='Use the following portion of a long document to see if any of the text is relevant to answer the question. \nReturn any relevant text verbatim.\n{context}\nQuestion: {question}\nRelevant text, if any:'), combine_prompt: BasePromptTemplate = PromptTemplate(input_variables=['question', 'summaries'], template='Given the following extracted parts of a long document and a question, create a final answer with references ("SOURCES"). \nIf you don\'t know the answer, just say that you don\'t know. Don\'t try to make up an answer.\nALWAYS return a "SOURCES" part in your answer.\n\nQUESTION: Which state/country\'s law governs the interpretation of the contract?\n=========\nContent: This Agreement is governed by English law and the parties submit to the exclusive jurisdiction of the English courts in  relation to any dispute (contractual or non-contractual) concerning this Agreement save that either party may apply to any court for an  injunction or other relief to protect its Intellectual Property Rights.\nSource: 28-pl\nContent: No Waiver. Failure or delay in exercising any right or remedy under this Agreement shall not constitute a waiver of such (or any other)  right or remedy.\n\n11.7 Severability. The invalidity, illegality or unenforceability of any term (or part of a term) of this Agreement shall not affect the continuation  in force of the remainder of the term (if any) and this Agreement.\n\n11.8 No Agency. Except as expressly stated otherwise, nothing in this Agreement shall create an agency, partnership or joint venture of any  kind between the parties.\n\n11.9 No Third-Party Beneficiaries.\nSource: 30-pl\nContent: (b) if Google believes, in good faith, that the Distributor has violated or caused Google to violate any Anti-Bribery Laws (as  defined in Clause 8.5) or that such a violation is reasonably likely to occur,\nSource: 4-pl\n=========\nFINAL ANSWER: This Agreement is governed by English law.\nSOURCES: 28-pl\n\nQUESTION: What did the president say about Michael Jackson?\n=========\nContent: Madam Speaker, Madam Vice President, our First Lady and Second Gentleman. Members of Congress and the Cabinet. Justices of the Supreme Court. My fellow Americans.  \n\nLast year COVID-19 kept us apart. This year we are finally together again. \n\nTonight, we meet as Democrats Republicans and Independents. But most importantly as Americans. \n\nWith a duty to one another to the American people to the Constitution. \n\nAnd with an unwavering resolve that freedom will always triumph over tyranny. \n\nSix days ago, Russia’s Vladimir Putin sought to shake the foundations of the free world thinking he could make it bend to his menacing ways. But he badly miscalculated. \n\nHe thought he could roll into Ukraine and the world would roll over. Instead he met a wall of strength he never imagined. \n\nHe met the Ukrainian people. \n\nFrom President Zelenskyy to every Ukrainian, their fearlessness, their courage, their determination, inspires the world. \n\nGroups of citizens blocking tanks with their bodies. Everyone from students to retirees teachers turned soldiers defending their homeland.\nSource: 0-pl\nContent: And we won’t stop. \n\nWe have lost so much to COVID-19. Time with one another. And worst of all, so much loss of life. \n\nLet’s use this moment to reset. Let’s stop looking at COVID-19 as a partisan dividing line and see it for what it is: A God-awful disease.  \n\nLet’s stop seeing each other as enemies, and start seeing each other for who we really are: Fellow Americans.  \n\nWe can’t change how divided we’ve been. But we can change how we move forward—on COVID-19 and other issues we must face together. \n\nI recently visited the New York City Police Department days after the funerals of Officer Wilbert Mora and his partner, Officer Jason Rivera. \n\nThey were responding to a 9-1-1 call when a man shot and killed them with a stolen gun. \n\nOfficer Mora was 27 years old. \n\nOfficer Rivera was 22. \n\nBoth Dominican Americans who’d grown up on the same streets they later chose to patrol as police officers. \n\nI spoke with their families and told them that we are forever in debt for their sacrifice, and we will carry on their mission to restore the trust and safety every community deserves.\nSource: 24-pl\nContent: And a proud Ukrainian people, who have known 30 years  of independence, have repeatedly shown that they will not tolerate anyone who tries to take their country backwards.  \n\nTo all Americans, I will be honest with you, as I’ve always promised. A Russian dictator, invading a foreign country, has costs around the world. \n\nAnd I’m taking robust action to make sure the pain of our sanctions  is targeted at Russia’s economy. And I will use every tool at our disposal to protect American businesses and consumers. \n\nTonight, I can announce that the United States has worked with 30 other countries to release 60 Million barrels of oil from reserves around the world.  \n\nAmerica will lead that effort, releasing 30 Million barrels from our own Strategic Petroleum Reserve. And we stand ready to do more if necessary, unified with our allies.  \n\nThese steps will help blunt gas prices here at home. And I know the news about what’s happening can seem alarming. \n\nBut I want you to know that we are going to be okay.\nSource: 5-pl\nContent: More support for patients and families. \n\nTo get there, I call on Congress to fund ARPA-H, the Advanced Research Projects Agency for Health. \n\nIt’s based on DARPA—the Defense Department project that led to the Internet, GPS, and so much more.  \n\nARPA-H will have a singular purpose—to drive breakthroughs in cancer, Alzheimer’s, diabetes, and more. \n\nA unity agenda for the nation. \n\nWe can do this. \n\nMy fellow Americans—tonight , we have gathered in a sacred space—the citadel of our democracy. \n\nIn this Capitol, generation after generation, Americans have debated great questions amid great strife, and have done great things. \n\nWe have fought for freedom, expanded liberty, defeated totalitarianism and terror. \n\nAnd built the strongest, freest, and most prosperous nation the world has ever known. \n\nNow is the hour. \n\nOur moment of responsibility. \n\nOur test of resolve and conscience, of history itself. \n\nIt is in this moment that our character is formed. Our purpose is found. Our future is forged. \n\nWell I know this nation.\nSource: 34-pl\n=========\nFINAL ANSWER: The president did not mention Michael Jackson.\nSOURCES:\n\nQUESTION: {question}\n=========\n{summaries}\n=========\nFINAL ANSWER:'), **kwargs: Any) → BaseQAWithSourcesChain

从LLM构建链条。

Parameters

• llm (BaseLanguageModel) –

• document_prompt (BasePromptTemplate) –

• question_prompt (BasePromptTemplate) –

• combine_prompt (BasePromptTemplate) –

• kwargs (Any) –

Return type

BaseQAWithSourcesChain

classmethod from_orm(obj: Any) → Model

Parameters

obj (Any) –

Return type

Model

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]

classmethod get_lc_namespace() → List[str]

获取langchain对象的命名空间。

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

Return type

List[str]

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: Dict[str, Any], config: Optional[RunnableConfig] = None, **kwargs: Any) → Dict[str, Any]

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

参数：

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

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

返回：

可运行对象的输出。

Parameters

• input (Dict[str, Any]) –

• config (Optional[RunnableConfig]) –

• kwargs (Any) –

Return type

Dict[str, Any]

classmethod is_lc_serializable() → bool

这个类是否可序列化？

Return type

bool

json(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Optional[Callable[[Any], Any]] = None, models_as_dict: bool = True, **dumps_kwargs: Any) → unicode

Generate a JSON representation of the model, include and exclude arguments as per dict().

encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().

Parameters

• include (Optional[Union[AbstractSetIntStr, MappingIntStrAny]]) –

• exclude (Optional[Union[AbstractSetIntStr, MappingIntStrAny]]) –

• by_alias (bool) –

• skip_defaults (Optional[bool]) –

• exclude_unset (bool) –

• exclude_defaults (bool) –

• exclude_none (bool) –

• encoder (Optional[Callable[[Any], Any]]) –

• models_as_dict (bool) –

• dumps_kwargs (Any) –

Return type

unicode

classmethod lc_id() → List[str]

用于序列化目的的此类的唯一标识符。

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

Return type

List[str]

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

classmethod parse_file(path: Union[str, Path], *, content_type: unicode = None, encoding: unicode = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

Parameters

• path (Union[str, Path]) –

• content_type (unicode) –

• encoding (unicode) –

• proto (Protocol) –

• allow_pickle (bool) –

Return type

Model

classmethod parse_obj(obj: Any) → Model

Parameters

obj (Any) –

Return type

Model

classmethod parse_raw(b: Union[str, bytes], *, content_type: unicode = None, encoding: unicode = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

Parameters

• b (Union[str, bytes]) –

• content_type (unicode) –

• encoding (unicode) –

• proto (Protocol) –

• allow_pickle (bool) –

Return type

Model

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]

prep_inputs(inputs: Union[Dict[str, Any], Any]) → Dict[str, str]

准备链的输入，包括从内存中添加输入。

参数:

inputs: 原始输入的字典，或者如果链只需要一个参数，则为单个输入。应包含`Chain.input_keys`中指定的所有输入，除了将由链的内存设置的输入。

返回:

包括链的内存添加的所有输入的字典。

Parameters

inputs (Union[Dict[str, Any], Any]) –

Return type

Dict[str, str]

prep_outputs(inputs: Dict[str, str], outputs: Dict[str, str], return_only_outputs: bool = False) → Dict[str, str]

验证和准备链输出，并将此运行的信息保存到内存中。

参数：

inputs：包括链内存中添加的任何输入的链输入字典。 outputs：初始链输出字典。 return_only_outputs：是否仅返回链输出。如果为False，则还将输入添加到最终输出中。

返回：

最终链输出的字典。

Parameters

• inputs (Dict[str, str]) –

• outputs (Dict[str, str]) –

• return_only_outputs (bool) –

Return type

Dict[str, str]

run(*args: Any, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) → Any

[Deprecated] 方便执行链的方法。

这个方法与`Chain.__call__`方法的主要区别在于，这个方法期望直接将输入作为位置参数或关键字参数传递，而`Chain.__call__`方法期望一个包含所有输入的单个输入字典。

参数:

*args: 如果链期望单个输入，则可以将其作为唯一的位置参数传递。 callbacks: 用于此链运行的回调。这些回调将被调用，除了在构建过程中传递给链的回调之外，但只有这些运行时回调会传播到对其他对象的调用。 tags: 要传递给所有回调的字符串标签列表。这些标签将被传递，除了在构建过程中传递给链的标签之外，但只有这些运行时标签会传播到对其他对象的调用。 **kwargs: 如果链期望多个输入，则可以直接将它们作为关键字参数传递。

返回:

链的输出。

示例:

# 假设我们有一个单输入链，接受一个'question'字符串:
chain.run("What's the temperature in Boise, Idaho?")
# -> "The temperature in Boise is..."

# 假设我们有一个多输入链，接受一个'question'字符串和一个'context'字符串:
question = "What's the temperature in Boise, Idaho?"
context = "Weather report for Boise, Idaho on 07/03/23..."
chain.run(question=question, context=context)
# -> "The temperature in Boise is..."

Notes

Deprecated since version langchain==0.1.0: Use invoke instead.

Parameters

• args (Any) –

• callbacks (Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]]) –

• tags (Optional[List[str]]) –

• metadata (Optional[Dict[str, Any]]) –

• kwargs (Any) –

Return type

Any

save(file_path: Union[Path, str]) → None

保存链条。

期望`Chain._chain_type`属性已实现，并且内存为空。

参数：

file_path：保存链条的文件路径。

示例：

chain.save(file_path="path/chain.yaml")

Parameters

file_path (Union[Path, str]) –

Return type

None

classmethod schema(by_alias: bool = True, ref_template: unicode = '#/definitions/{model}') → DictStrAny

Parameters

• by_alias (bool) –

• ref_template (unicode) –

Return type

DictStrAny

classmethod schema_json(*, by_alias: bool = True, ref_template: unicode = '#/definitions/{model}', **dumps_kwargs: Any) → unicode

Parameters

• by_alias (bool) –

• ref_template (unicode) –

• dumps_kwargs (Any) –

Return type

unicode

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

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

Parameters

• input (Input) –

• config (Optional[RunnableConfig]) –

• kwargs (Optional[Any]) –

Return type

Iterator[Output]

to_json() → Union[SerializedConstructor, SerializedNotImplemented]

将可运行对象序列化为JSON。

Return type

Union[SerializedConstructor, SerializedNotImplemented]

to_json_not_implemented() → SerializedNotImplemented

Return type

SerializedNotImplemented

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

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

Parameters

• input (Iterator[Input]) –

• config (Optional[RunnableConfig]) –

• kwargs (Optional[Any]) –

Return type

Iterator[Output]

classmethod update_forward_refs(**localns: Any) → None

Try to update ForwardRefs on fields based on this Model, globalns and localns.

Parameters

localns (Any) –

Return type

None

classmethod validate(value: Any) → Model

Parameters

value (Any) –

Return type

Model

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]

property InputType: Type[langchain_core.runnables.utils.Input]

这个可运行程序接受的输入类型，以类型注释的形式指定。

property OutputType: Type[langchain_core.runnables.utils.Output]

此可运行程序生成的输出类型，指定为类型注释。

property config_specs: List[langchain_core.runnables.utils.ConfigurableFieldSpec]

列出此可运行程序的可配置字段。

property input_keys: List[str]

期望输入键。

元数据 私密

property input_schema: Type[pydantic.main.BaseModel]

此可运行程序接受的输入类型，由pydantic模型指定。

property lc_attributes: Dict

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

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

property lc_secrets: Dict[str, str]

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

例如，

{“openai_api_key”: “OPENAI_API_KEY”}

name: Optional[str] = None

可运行对象的名称。用于调试和跟踪。

property output_keys: List[str]

返回输出键。

元数据 私有

property output_schema: Type[pydantic.main.BaseModel]

这个可运行程序产生的输出类型被指定为一个pydantic模型。

Examples using RetrievalQAWithSourcesChain

• Install all needed libraries

• Pip install necessary package

• Set env var OPENAI_API_KEY or load from a .env file:

• Uncomment this to install psychicapi if you don’t already have it installed

• cosine: distance metric

• initialize marqo

• install package