`langchain_experimental.llm_bash.base`.LLMBashChain¶

class langchain_experimental.llm_bash.base.LLMBashChain[source]¶

Bases: Chain

解释并执行bash操作的链。

Example

from langchain.chains import LLMBashChain
from langchain_community.llms import OpenAI
llm_bash = LLMBashChain.from_llm(OpenAI())

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 llm: Optional[BaseLanguageModel] = None¶: [已弃用] 用于使用的LLM包装器。

param llm_chain: LLMChain [Required]¶

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

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

param prompt: BasePromptTemplate = PromptTemplate(input_variables=['question'], output_parser=BashOutputParser(), template='If someone asks you to perform a task, your job is to come up with a series of bash commands that will perform the task. There is no need to put "#!/bin/bash" in your answer. Make sure to reason step by step, using this format:\n\nQuestion: "copy the files in the directory named \'target\' into a new directory at the same level as target called \'myNewDirectory\'"\n\nI need to take the following actions:\n- List all files in the directory\n- Create a new directory\n- Copy the files from the first directory into the second directory\n```bash\nls\nmkdir myNewDirectory\ncp -r target/* myNewDirectory\n```\n\nThat is the format. Begin!\n\nQuestion: {question}')¶: [已弃用]

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_llm(llm: BaseLanguageModel, prompt: BasePromptTemplate = PromptTemplate(input_variables=['question'], output_parser=BashOutputParser(), template='If someone asks you to perform a task, your job is to come up with a series of bash commands that will perform the task. There is no need to put "#!/bin/bash" in your answer. Make sure to reason step by step, using this format:\n\nQuestion: "copy the files in the directory named \'target\' into a new directory at the same level as target called \'myNewDirectory\'"\n\nI need to take the following actions:\n- List all files in the directory\n- Create a new directory\n- Copy the files from the first directory into the second directory\n```bash\nls\nmkdir myNewDirectory\ncp -r target/* myNewDirectory\n```\n\nThat is the format. Begin!\n\nQuestion: {question}'), **kwargs: Any) → LLMBashChain[source]¶

Parameters

llm (BaseLanguageModel) –
prompt (BasePromptTemplate) –
kwargs (Any) –

Return type

LLMBashChain

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模型。

langchain_experimental.llm_bash.base.LLMBashChain¶

`langchain_experimental.llm_bash.base`.LLMBashChain¶