Source code for langchain_community.tools.zapier.tool
"""[已弃用]
## Zapier自然语言操作API
完整文档请参阅：https://nla.zapier.com/start/
**Zapier自然语言操作** 通过自然语言API接口，让您可以访问Zapier平台上的5000多个应用和20000多个操作。
NLA支持诸如Gmail、Salesforce、Trello、Slack、Asana、HubSpot、Google表格、Microsoft Teams等应用：https://zapier.com/apps
Zapier NLA处理所有底层API授权和自然语言 --> 底层API调用 --> 返回简化输出给LLMs的翻译。关键思想是您或您的用户通过类似于OAuth的设置窗口公开一组操作，然后可以通过REST API进行查询和执行。
NLA提供API密钥和OAuth来签署NLA API请求。
1. 服务器端（API密钥）：用于快速入门、测试和生产场景，LangChain将仅使用开发者Zapier帐户中公开的操作（并将使用Zapier.com上开发者连接的帐户）
2. 用户界面（OAuth）：用于生产场景，您正在部署面向最终用户的应用程序，LangChain需要访问最终用户公开的操作和Zapier.com上连接的帐户
为简洁起见，本快速入门将专注于服务器端用例。
请查看[完整文档](https://nla.zapier.com/start/)以获取面向用户的OAuth开发者支持。
通常，您会使用SequentialChain，以下是一个基本示例：
1. 使用NLA在Gmail中查找电子邮件
2. 使用LLMChain为（1）生成一份草稿回复
3. 使用NLA发送草稿回复（2）给Slack中的某人通过直接消息
在下面的代码中：
```python
import os
# 从https://platform.openai.com/获取
os.environ["OPENAI_API_KEY"] = os.environ.get("OPENAI_API_KEY", "")
# 从https://nla.zapier.com/docs/authentication/获取
os.environ["ZAPIER_NLA_API_KEY"] = os.environ.get("ZAPIER_NLA_API_KEY", "")
from langchain_community.agent_toolkits import ZapierToolkit
from langchain_community.utilities.zapier import ZapierNLAWrapper
## 步骤0. 公开Gmail“查找电子邮件”和Slack“发送频道消息”操作
# 首先转到这里，登录，公开（启用）这两个操作：
# https://nla.zapier.com/demo/start
# -- 对于此示例，可以将所有字段保留为“让AI猜测”
# 在OAuth场景中，您将获得自己的<provider> id（而不是'demo'），您首先将用户通过该id路由
zapier = ZapierNLAWrapper()
## 要利用OAuth，您可以将值`nla_oauth_access_token`传递给ZapierNLAWrapper。如果这样做，就不需要初始化ZAPIER_NLA_API_KEY环境变量
# zapier = ZapierNLAWrapper(zapier_nla_oauth_access_token="TOKEN_HERE")
toolkit = ZapierToolkit.from_zapier_nla_wrapper(zapier)
```
"""
from typing import Any, Dict, Optional
from langchain_core._api import warn_deprecated
from langchain_core.callbacks import (
AsyncCallbackManagerForToolRun,
CallbackManagerForToolRun,
)
from langchain_core.pydantic_v1 import Field, root_validator
from langchain_core.tools import BaseTool
from langchain_community.tools.zapier.prompt import BASE_ZAPIER_TOOL_PROMPT
from langchain_community.utilities.zapier import ZapierNLAWrapper
[docs]class ZapierNLARunAction(BaseTool):
"""工具，用于执行用户暴露的特定操作。
参数:
action_id: 要执行的特定操作的操作ID（来自操作列表）
（必须将api_key设置与操作所有者关联）
instructions: 用于使用该操作的自然语言指令字符串
（例如，“获取Mike Knoop的最新电子邮件”对应于“Gmail: 查找电子邮件”操作）
params: 字典，可选。提供的任何参数将*覆盖*来自`instructions`的AI猜测
（请参阅此处的“理解AI猜测流程”：
https://nla.zapier.com/docs/using-the-api#ai-guessing）"""
api_wrapper: ZapierNLAWrapper = Field(default_factory=ZapierNLAWrapper) # type: ignore[arg-type]
action_id: str
params: Optional[dict] = None
base_prompt: str = BASE_ZAPIER_TOOL_PROMPT
zapier_description: str
params_schema: Dict[str, str] = Field(default_factory=dict)
name: str = ""
description: str = ""
@root_validator
def set_name_description(cls, values: Dict[str, Any]) -> Dict[str, Any]:
zapier_description = values["zapier_description"]
params_schema = values["params_schema"]
if "instructions" in params_schema:
del params_schema["instructions"]
# Ensure base prompt (if overridden) contains necessary input fields
necessary_fields = {"{zapier_description}", "{params}"}
if not all(field in values["base_prompt"] for field in necessary_fields):
raise ValueError(
"Your custom base Zapier prompt must contain input fields for "
"{zapier_description} and {params}."
)
values["name"] = zapier_description
values["description"] = values["base_prompt"].format(
zapier_description=zapier_description,
params=str(list(params_schema.keys())),
)
return values
def _run(
self, instructions: str, run_manager: Optional[CallbackManagerForToolRun] = None
) -> str:
"""使用Zapier NLA工具返回所有暴露的用户操作列表。"""
warn_deprecated(
since="0.0.319",
message=(
"This tool will be deprecated on 2023-11-17. See "
"https://nla.zapier.com/sunset/ for details"
),
)
return self.api_wrapper.run_as_str(self.action_id, instructions, self.params)
async def _arun(
self,
instructions: str,
run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
) -> str:
"""使用Zapier NLA工具返回所有暴露的用户操作列表。"""
warn_deprecated(
since="0.0.319",
message=(
"This tool will be deprecated on 2023-11-17. See "
"https://nla.zapier.com/sunset/ for details"
),
)
return await self.api_wrapper.arun_as_str(
self.action_id,
instructions,
self.params,
)
ZapierNLARunAction.__doc__ = (
ZapierNLAWrapper.run.__doc__ + ZapierNLARunAction.__doc__ # type: ignore
)
# other useful actions
[docs]class ZapierNLAListActions(BaseTool):
"""用户操作列表工具。"""
name: str = "ZapierNLA_list_actions"
description: str = BASE_ZAPIER_TOOL_PROMPT + (
"This tool returns a list of the user's exposed actions."
)
api_wrapper: ZapierNLAWrapper = Field(default_factory=ZapierNLAWrapper) # type: ignore[arg-type]
def _run(
self,
_: str = "",
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> str:
"""使用Zapier NLA工具返回所有暴露的用户操作列表。"""
warn_deprecated(
since="0.0.319",
message=(
"This tool will be deprecated on 2023-11-17. See "
"https://nla.zapier.com/sunset/ for details"
),
)
return self.api_wrapper.list_as_str()
async def _arun(
self,
_: str = "",
run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
) -> str:
"""使用Zapier NLA工具返回所有暴露的用户操作列表。"""
warn_deprecated(
since="0.0.319",
message=(
"This tool will be deprecated on 2023-11-17. See "
"https://nla.zapier.com/sunset/ for details"
),
)
return await self.api_wrapper.alist_as_str()
ZapierNLAListActions.__doc__ = (
ZapierNLAWrapper.list.__doc__ + ZapierNLAListActions.__doc__ # type: ignore
)