langchain_community.tools.zapier.tool.ZapierNLAListActions

备注

“ZapierNLAListActions”实现了标准Runnable Interface。🏃

Runnable Interface提供了额外的runnables方法，例如with_types、with_retry、assign、bind、get_graph等。

class langchain_community.tools.zapier.tool.ZapierNLAListActions

基础类：BaseTool

返回与当前用户（与设置的api_key关联）关联的所有公开（启用）操作的列表。在此处更改您的公开操作：https://nla.zapier.com/demo/start/

如果未公开操作，返回列表可能为空。否则将包含操作对象的列表。

[{

“id”：字符串， “description”：字符串， “params”：字典[str, str]

}]

params将始终包含一个instructions键，这是唯一必需的参数。所有其他参数都是可选的，如果提供，将覆盖任何AI猜测（请参阅以下“理解AI猜测流程”：https://nla.zapier.com/docs/using-the-api#ai-guessing）。列出用户所有公开操作的工具。

初始化工具。

参数api_wrapper: ZapierNLAWrapper [Optional]

参数args_schema: Optional[TypeBaseModel] = None

Pydantic模型类用于验证和解析工具的输入参数。

Args schema应该是以下之一：

• pydantic.BaseModel的子类。

或者 - pydantic.v1.BaseModel的子类，如果您在pydantic 2中访问v1命名空间。

参数 callback_manager: Optional[BaseCallbackManager] = None

弃用。请使用callbacks代替。

参数 callbacks: Callbacks = None

工具执行期间要调用的回调。

param description: str = 'A wrapper around Zapier NLA actions. The input to this tool is a natural language instruction, for example "get the latest email from my bank" or "send a slack message to the #general channel". Each tool will have params associated with it that are specified as a list. You MUST take into account the params when creating the instruction. For example, if the params are [\'Message_Text\', \'Channel\'], your instruction should be something like \'send a slack message to the #general channel with the text hello world\'. Another example: if the params are [\'Calendar\', \'Search_Term\'], your instruction should be something like \'find the meeting in my personal calendar at 3pm\'. Do not make up params, they will be explicitly specified in the tool description. If you do not have enough information to fill in the params, just say \'not enough information provided in the instruction, missing <param>\'. If you get a none or null response, STOP EXECUTION, do not try to another tool!This tool specifically used for: {zapier_description}, and has params: {params}This tool returns a list of the user\'s exposed actions.'

用于告知模型何时/如何/为什么使用工具。

您可以在描述中提供少量示例。

参数 handle_tool_error: Optional[Union[bool, str, Callable[[ToolException], str]] = False

处理抛出的ToolException内容。

参数 handle_validation_error: Optional[Union[bool, str, Callable[[ValidationError], str]] = False

处理抛出的ValidationError内容。

参数 metadata: Optional[Dict[str, Any]] = None

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

param response_format: Literal['content', 'content_and_artifact'] = 'content'

工具响应格式。默认为 'content'。

如果为 "content"，则工具体现在 ToolMessage 的内容上。如果为 "content_and_artifact"，则预期输出是包含 (内容，工件) 的两个元素的元组。

param return_direct: bool = False

是否直接返回工具的输出。

将此设置为 True 意味着在调用工具后，AgentExecutor 将停止循环。

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

与工具相关的可选标签列表。默认为 None。这些标签将与调用此工具的每个调用关联，并作为在 callbacks 中定义的处理程序的参数传递。您可以使用这些标签识别具有用例的工具的特定实例。

param verbose: bool = False

是否记录工具的进度。

__call__(tool_input: str, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None) → str

自 langchain-core==0.1.47 版本以来已弃用: 请使用 invoke 替代。

使工具可调用。

参数

• tool_input (str) –

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

返回类型

str

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

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

默认批处理实现适用于IO密集型可运行对象。

子类应该重写此方法，如果它们可以更有效地批量执行；例如，如果底层的Runnable使用一个支持批处理模式的API。

参数

• <强>inputs （List[输入]） - Runnable的输入列表。

• <强>config （Optional[Union[RunnableConfig, List[RunnableConfig]]]） - 调用Runnable时要使用的配置。配置支持诸如'标签'、'元数据'等标准密钥，用于跟踪目的，'max_concurrency'用于控制并行执行的工作量，以及其他密钥。有关更多详细信息，请参阅RunnableConfig。默认为None。

• <强>return_exceptions （布尔值） - 是否返回异常而不是抛出它们。默认为False。

• <强>kwargs （Optional[任意]） - 要传递给Runnable的附加关键字参数。

返回：

Runnable的输出列表。

返回类型

List[输出]

异步 abatch_as_completed(inputs: 顺序[输入], config: Optional[Union[RunnableConfig, 顺序[RunnableConfig]]] = None, *, return_exceptions: 布尔值 = False, **kwargs: Optional[任意]) → 异步迭代器[元组[int, Union[Output, Exception]]]¶

并行在输入列表上运行ainvoke，随着它们的完成依次产生结果。

参数

• <强>inputs （顺序[输入]） - Runnable的输入列表。

• config (可选[联合[RunnableConfig, Sequence[RunnableConfig]]]) – 调用Runnable时的配置。该配置支持标准键，如用于追踪的“标签”、“元数据”，用于控制并行工作多少的“最大并发数”和其他键。请参阅RunnableConfig以获取更多详细信息。默认值为None。默认值为None。

• <强>return_exceptions （布尔值） - 是否返回异常而不是抛出它们。默认为False。

• <强>kwargs （Optional[任意]） - 要传递给Runnable的附加关键字参数。

产出

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

返回类型

AsyncIterator[Tuple[int, 联合[Output, Exception]]]

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

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

默认实现允许即使Runnable未实现invoke的本地异步版本也可以使用异步代码。

子类应覆盖此方法，如果它们可以异步运行。

参数

• input (联合[str, Dict, ToolCall]) –

• config (可选[RunnableConfig]) –

• kwargs (Any) –

返回类型

Any

async arun(tool_input: Union[str, Dict], verbose: Optional[bool] = None, start_color: Optional[str] = 'green', color: Optional[str] = 'green', callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, run_id: Optional[UUID] = None, config: Optional[RunnableConfig] = None, tool_call_id: Optional[str] = None, **kwargs: Any) → Any

异步运行工具。

参数

• tool_input (联合[str, Dict]) – 工具的输入。

• verbose (可选[bool]) – 是否记录工具的进度。默认值为None。

• start_color (可选[str]) – 启动工具时要使用的颜色。默认值为‘green’。

• color (可选[str]) – 结束工具时要使用的颜色。默认值为‘green’。

• callbacks (可选[联合[List[BaseCallbackHandler], BaseCallbackManager]]) – 工具执行期间要调用的回调。默认值为None。

• tags (可选[List[str]]) - 与工具相关联的可选标签列表。默认为 None。

• metadata (可选[Dict[str, Any]]) - 与工具相关联的可选元数据。默认为 None。

• run_name (可选[str]) - 运行的名称。默认为 None。

• run_id (可选[UUID]) - 运行的 id。默认为 None。

• config (可选[RunnableConfig]) - 工具的配置。默认为 None。

• tool_call_id (可选[str]) - 工具调用的 id。默认为 None。

• kwargs (Any) - 传递给工具的额外参数

返回：

工具的输出。

引发

ToolException - 如果工具执行期间发生错误。

返回类型

Any

as_tool(args_schema: Optional[Type[BaseModel]] = None, *, name: Optional[str] = None, description: Optional[str] = None, arg_types: Optional[Dict[str, Type]] = None) → BaseTool

测试版

此 API 为测试版，未来可能会有所变动。

从一个 Runnable 创建 BaseTool。

as_tool 将从一个 Runnable 中实例化一个具有名称、描述和 args_schema 的 BaseTool。在可能的情况下，架构是从 runnable.get_input_schema 推断得出的。作为替代（例如，如果 Runnable 以字典作为输入并且特定的字典键未进行类型化），可以通过 args_schema 直接指定架构。您还可以传递 arg_types 以仅指定所需的参数及其类型。

参数

• args_schema (可选[类型[BaseModel]]) – 工具的架构。默认为 None。

• name (可选[str]) – 工具的名称。默认为 None。

• description (可选[str]) – 工具的描述。默认为 None。

• arg_types (可选[Dict[str, 类型]]) – 参数名称到类型的字典。默认为 None。

返回：

BaseTool 实例。

返回类型

BaseTool

类型化字典输入

from typing import List
from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda

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

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

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

dict 输入，通过 args_schema 指定架构

from typing import Any, Dict, List
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import RunnableLambda

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

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

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

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

dict 输入，通过 arg_types 指定架构

from typing import Any, Dict, List
from langchain_core.runnables import RunnableLambda

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

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

字符串输入

from langchain_core.runnables import RunnableLambda

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

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

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

新版本 0.2.14。

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

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

参数

• input (输入) – Runnable 的输入。

• config (可选[RunnableConfig]) – 可用于 Runnable 的配置。默认为 None。

• <强>kwargs （Optional[任意]） - 要传递给Runnable的附加关键字参数。

产出

Runnable 的输出。

返回类型

AsyncIterator[输出]

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[Union[StandardStreamEvent, CustomStreamEvent]]

测试版

此 API 为测试版，未来可能会有所变动。

生成事件流。

用于创建 StreamEvents 迭代器，提供关于 Runnable 进度的实时信息，包括中间结果的事件。

StreamEvent 是一个字典，其架构如下

• event: str - 事件名称具有以下格式

  on_[runnable_type]_(start|stream|end).

• name: str - 生成事件的Runnable的名称。

• run_id: str - 与给定的Runnable执行相关联的随机生成的ID。

  该事件。作为父Runnable执行一部分任务的子Runnable将分配自己的唯一ID。

• parent_ids: List[str] - 生成事件的父Runnable的ID。根Runnable将有一个空列表。

  父ID的顺序是从根到直接父级。仅适用于API的第2版。API的第1版将返回一个空列表。

• tags: Optional[List[str]] - 生成事件的Runnable的标签。

• metadata: Optional[Dict[str, Any]] - 生成事件的Runnable的元数据。

• data: Dict[str, Any]

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

注意 此参考表是为第2版本的架构。

事件	名称	块	输入	输出
on_chat_model_start	[模型名称]		{“messages”: [[系统消息，人类消息]]}
on_chat_model_stream	[模型名称]	AIMessageChunk(content="hello")
on_chat_model_end	[模型名称]		{“messages”: [[系统消息，人类消息]]}	AIMessageChunk(content="hello world")
on_llm_start	[模型名称]		{‘input’: 'hello’}
on_llm_stream	[模型名称]	“Hello”
on_llm_end	[模型名称]		“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	[检索器名称]		{“query”: “hello”}
on_retriever_end	[检索器名称]		{“query”: “hello”}	[Document(…), ..]
on_prompt_start	[模板名称]		{“question”: “hello”}
on_prompt_end	[模板名称]		{“question”: “hello”}	ChatPromptValue(messages: [系统消息，…])

除了标准事件外，用户还可以分发自定义事件（请参阅以下示例）。

自定义事件仅在第2版的API中呈现！

自定义事件的格式如下

属性	类型	描述
名称	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)

参数

• input (Any) – Runnable的输入。

• config (Optional[RunnableConfig]) – 用于Runnable的配置。

• version (Literal['v1', 'v2']) – 要使用的架构版本，要么是v2，要么是v1。用户应使用v2。v1用于向后兼容，并将从0.4.0版本中弃用。直到API稳定，不会分配默认值。自定义事件仅在第2版中呈现。

• include_names (Optional[Sequence[str]]) – 仅包括具有匹配名称的runnables的事件。

• include_types (Optional[Sequence[str]]) – 仅包括具有匹配类型的runnables的事件。

• include_tags (Optional[Sequence[str]]) – 仅包括具有匹配标签的runnables的事件。

• exclude_names (可选[字符串序列]) – 排除具有匹配名称的可运行对象中的事件。

• exclude_types (可选[字符串序列]) – 排除具有匹配类型的可运行对象中的事件。

• exclude_tags (可选[字符串序列]) – 排除具有匹配标签的可运行对象中的事件。

• kwargs (任意) – 将传递给可运行对象的附加关键字参数。这些参数将传递给astream_log，因为astream_events的实施是基于astream_log。

产出

异步流事件流。

引发

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

返回类型

异步迭代器 Union [[标准流事件], 自定义流事件]]

batch(inputs: 输入列表), config: 可选: RunnableConfig 或 RunnableConfig 列表], **kwargs), return_exceptions: bool)→ 输出列表

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

默认批处理实现适用于IO密集型可运行对象。

子类应该重写此方法，如果它们可以更有效地批量执行；例如，如果底层的Runnable使用一个支持批处理模式的API。

参数

• inputs (输入列表) –

• config (可选RunnableConfig 或 RunnableConfig 列表) –

• return_exceptions (bool) –

• kwargs (可选任意) –

返回类型

List[输出]

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

并行在输入列表上运行调用，按完成顺序产生结果。

参数

• inputs (Sequence[Input]) –

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

• return_exceptions (bool) –

• kwargs (可选任意) –

返回类型

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

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

配置可以运行时设置的Runnables的备选方案。

参数

• which (ConfigurableField) – 用于选择备选方案的ConfigurableField实例。

• default_key (str) – 如果未选择备选方案，将使用默认键。默认为“default”。

• prefix_keys (bool) – 是否以ConfigurableField id作为键的前缀。默认为False。

• **kwargs (Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]]) – 键到Runnable实例或返回Runnable实例的可调用对象的字典。

返回：

配置备选方案的新的Runnable。

返回类型

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

# 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: Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) → RunnableSerializable[Input, Output]¶

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

参数

**kwargs (Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) – 要配置的 ConfigurableField 实例的字典。

返回：

配置字段后的新可执行对象。

返回类型

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="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
)

invoke(input: Union[str, Dict, ToolCall], config: Optional[RunnableConfig] = None, **kwargs: Any) → Any

将单个输入转换为输出。重写以实现。

参数

• input (Union[str, Dict, ToolCall]) – 可执行的输入。

• config (可选[RunnableConfig]) – 调用可执行时使用的配置。该配置支持标准键，如用于跟踪的“tags”、“metadata”，控制并行工作量的“max_concurrency”，以及其他键。请参阅RunnableConfig获取更多详细信息。

• kwargs (Any) –

返回：

Runnable 的输出。

返回类型

Any

run(tool_input: Union[str, Dict[str, Any]], verbose: Optional[bool] = None, start_color: Optional[str] = 'green', color: Optional[str] = 'green', callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, run_id: Optional[UUID] = None, config: Optional[RunnableConfig] = None, tool_call_id: Optional[str] = None, **kwargs: Any) → Any

运行工具。

参数

• tool_input (联合[str, Dict[str, Any]]) – 工具的输入。

• verbose (可选[bool]) – 是否记录工具的进度。默认值为None。

• start_color (可选[str]) – 启动工具时要使用的颜色。默认值为‘green’。

• color (可选[str]) – 结束工具时要使用的颜色。默认值为‘green’。

• callbacks (可选[联合[List[BaseCallbackHandler], BaseCallbackManager]]) – 工具执行期间要调用的回调。默认值为None。

• tags (可选[List[str]]) - 与工具相关联的可选标签列表。默认为 None。

• metadata (可选[Dict[str, Any]]) - 与工具相关联的可选元数据。默认为 None。

• run_name (可选[str]) - 运行的名称。默认为 None。

• run_id (可选[UUID]) - 运行的 id。默认为 None。

• config (可选[RunnableConfig]) - 工具的配置。默认为 None。

• tool_call_id (可选[str]) - 工具调用的 id。默认为 None。

• kwargs (Any) - 传递给工具的额外参数

返回：

工具的输出。

引发

ToolException - 如果工具执行期间发生错误。

返回类型

Any

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

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

参数

• input (输入) – Runnable 的输入。

• config (可选[RunnableConfig]) – 可用于 Runnable 的配置。默认为 None。

• <强>kwargs （Optional[任意]） - 要传递给Runnable的附加关键字参数。

产出

Runnable 的输出。

返回类型

Iterator[Output]

to_json() → Union[SerializedConstructor, SerializedNotImplemented]

将可执行序列化为JSON。

返回：

可序列化表示的可执行。

返回类型

Union[SerializedConstructor, SerializedNotImplemented]

属性 args: dict

属性 is_single_input: bool

工具是否只接受单个输入。

属性 tool_call_schema: Type[BaseModel]