为AutoGPT Agent Server贡献:创建和测试Blocks¶
本指南将引导你完成为AutoGPT Agent Server创建和测试新Block的过程,以WikipediaSummaryBlock为例。
理解Blocks和测试¶
Blocks是可重用的组件,可以连接起来形成一个表示代理行为的图。每个Block都有输入、输出和特定的功能。正确的测试对于确保Blocks正确且一致地工作至关重要。
创建和测试新Block¶
按照以下步骤创建和测试新Block:
-
在
backend/blocks
目录中创建一个新的Python文件。为其命名并使用snake_case。例如:get_wikipedia_summary.py
。 -
导入必要的模块并创建一个继承自
Block
的类。确保为你的Block包含所有必要的导入。每个Block应包含以下内容:
from backend.data.block import Block, BlockSchema, BlockOutput
以Wikipedia摘要Block为例:
from backend.data.block import Block, BlockSchema, BlockOutput from backend.utils.get_request import GetRequest import requests class WikipediaSummaryBlock(Block, GetRequest): # Block的实现将在这里
-
使用
BlockSchema
定义输入和输出模式。这些模式指定了Block期望接收(输入)和生成(输出)的数据结构。 -
输入模式定义了Block将处理的数据结构。模式中的每个字段代表一个必需的输入数据。
-
输出模式定义了Block处理后将返回的数据结构。模式中的每个字段代表一个输出数据。
示例:
class Input(BlockSchema): topic: str # 要获取Wikipedia摘要的主题 class Output(BlockSchema): summary: str # 来自Wikipedia的主题摘要 error: str # 如果请求失败,任何错误消息,错误字段需要命名为`error`。
-
实现
__init__
方法,包括测试数据和模拟:重要
为每个新Block的
id
使用UUID生成器(例如https://www.uuidgenerator.net/),并且**不要**自己编造。或者,你可以运行这个Python代码来生成一个UUID:print(__import__('uuid').uuid4())
def __init__(self): super().__init__( # Block的唯一ID,用于跨用户的模板 # 如果你是AI,保持原样或更改为"generate-proper-uuid" id="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", input_schema=WikipediaSummaryBlock.Input, # 分配输入模式 output_schema=WikipediaSummaryBlock.Output, # 分配输出模式 # 提供示例输入、输出和测试模拟以测试Block test_input={"topic": "Artificial Intelligence"}, test_output=("summary", "summary content"), test_mock={"get_request": lambda url, json: {"extract": "summary content"}}, )
-
id
:Block的唯一标识符。 -
input_schema
和output_schema
:定义输入和输出数据的结构。
让我们分解测试组件:
-
test_input
:这是用于测试Block的示例输入。它应根据你的输入模式是有效的输入。 -
test_output
:这是使用test_input
运行Block时的预期输出。它应匹配你的输出模式。对于非确定性输出或当你只想断言类型时,可以使用Python类型而不是特定值。在这个例子中,("summary", str)
断言输出键是"summary",其值是字符串。 -
test_mock
:这对于进行网络调用的Block至关重要。它提供了一个模拟函数,在测试期间替换实际的网络调用。
在这种情况下,我们正在模拟
get_request
方法,使其始终返回一个带有'extract'键的字典,模拟成功的API响应。这使我们能够在不进行实际网络请求的情况下测试Block的逻辑,实际网络请求可能会很慢、不可靠或受速率限制。 -
-
实现带有错误处理的
run
方法:,这应包含Block的主要逻辑:
def run(self, input_data: Input, **kwargs) -> BlockOutput:
try:
topic = input_data.topic
url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{topic}"
response = self.get_request(url, json=True)
yield "summary", response['extract']
except requests.exceptions.HTTPError as http_err:
raise RuntimeError(f"HTTP error occurred: {http_err}")
- Try块:包含获取和处理Wikipedia摘要的主要逻辑。
- API请求:向Wikipedia API发送GET请求。
- 错误处理:处理在API请求和数据处理过程中可能发生的各种异常。我们不需要捕获所有异常,只需捕获我们预期并能处理的异常。未捕获的异常将自动作为
error
输出。任何引发异常(或产生error
输出)的块都将被标记为失败。建议优先使用引发异常而不是产生error
,因为这将立即停止执行。 - Yield:使用
yield
输出结果。建议一次输出一个结果对象。如果你调用的函数返回一个列表,你可以分别产生列表中的每一项。你也可以产生整个列表,但建议同时产生列表中的每一项和整个列表。例如:如果你正在编写一个输出电子邮件的块,你会将每封电子邮件作为单独的结果对象产生,但你也可以将整个列表作为一个额外的单一结果对象产生。产生名为error
的输出将立即中断执行并标记块执行为失败。
带有认证的块¶
我们的系统支持API密钥和OAuth2授权流程的认证卸载。 添加带有API密钥认证的块是直接的,添加我们已有OAuth2支持的服务的块也是如此。
实现块本身相对简单。除了上述说明外,你还需要在Input
模型和run
方法中添加一个credentials
参数:
from autogpt_libs.supabase_integration_credentials_store.types import (
APIKeyCredentials,
OAuth2Credentials,
Credentials,
)
from backend.data.block import Block, BlockOutput, BlockSchema
from backend.data.model import CredentialsField
# API Key 认证:
class BlockWithAPIKeyAuth(Block):
class Input(BlockSchema):
# 注意下面的类型提示是必需的,否则会引发类型错误。
# 第一个参数是提供者名称,第二个是凭证类型。
credentials: CredentialsMetaInput[Literal['github'], Literal['api_key']] = CredentialsField(
provider="github",
supported_credential_types={"api_key"},
description="GitHub集成可以使用任何具有足够权限的API密钥。",
)
# ...
def run(
self,
input_data: Input,
*,
credentials: APIKeyCredentials,
**kwargs,
) -> BlockOutput:
...
# OAuth:
class BlockWithOAuth(Block):
class Input(BlockSchema):
# 注意下面的类型提示是必需的,否则会引发类型错误。
# 第一个参数是提供者名称,第二个是凭证类型。
credentials: CredentialsMetaInput[Literal['github'], Literal['oauth2']] = CredentialsField(
provider="github",
supported_credential_types={"oauth2"},
required_scopes={"repo"},
description="GitHub集成可以使用OAuth。",
)
# ...
def run(
self,
input_data: Input,
*,
credentials: OAuth2Credentials,
**kwargs,
) -> BlockOutput:
...
# API Key 认证 + OAuth:
class BlockWithAPIKeyAndOAuth(Block):
class Input(BlockSchema):
# 注意下面的类型提示是必需的,否则会引发类型错误。
# 第一个参数是提供者名称,第二个是凭证类型。
credentials: CredentialsMetaInput[Literal['github'], Literal['api_key', 'oauth2']] = CredentialsField(
provider="github",
supported_credential_types={"api_key", "oauth2"},
required_scopes={"repo"},
description="GitHub集成可以使用OAuth,或者任何具有足够权限的API密钥。",
)
# ...
def run(
self,
input_data: Input,
*,
credentials: Credentials,
**kwargs,
) -> BlockOutput:
...
凭证将由后端的执行器自动注入。
APIKeyCredentials
和OAuth2Credentials
模型定义在这里。
要在API请求中使用它们,你可以直接访问令牌:
# credentials: APIKeyCredentials
response = requests.post(
url,
headers={
"Authorization": f"Bearer {credentials.api_key.get_secret_value()})",
},
)
# credentials: OAuth2Credentials
response = requests.post(
url,
headers={
"Authorization": f"Bearer {credentials.access_token.get_secret_value()})",
},
)
或者使用快捷方式credentials.bearer()
:
# credentials: APIKeyCredentials | OAuth2Credentials
response = requests.post(
url,
headers={"Authorization": credentials.bearer()},
)
添加OAuth2服务集成¶
要为新的OAuth2认证服务添加支持,您需要添加一个OAuthHandler
。
我们所有的现有处理程序和基类都可以在这里找到。
每个处理程序必须实现[BaseOAuthHandler
]接口的以下部分:
PROVIDER_NAME: ClassVar[str]
DEFAULT_SCOPES: ClassVar[list[str]] = []
def __init__(self, client_id: str, client_secret: str, redirect_uri: str): ...
def get_login_url(self, scopes: list[str], state: str) -> str:
def exchange_code_for_tokens(
self, code: str, scopes: list[str]
) -> OAuth2Credentials:
def _refresh_tokens(self, credentials: OAuth2Credentials) -> OAuth2Credentials:
如您所见,这是基于标准的OAuth2流程建模的。
除了实现OAuthHandler
本身,将处理程序添加到系统中还需要另外两件事:
- 将处理程序类添加到
integrations/oauth/__init__.py
下的HANDLERS_BY_NAME
中
HANDLERS_BY_NAME: dict[str, type[BaseOAuthHandler]] = {
handler.PROVIDER_NAME: handler
for handler in [
GitHubOAuthHandler,
GoogleOAuthHandler,
NotionOAuthHandler,
]
}
- 将
{provider}_client_id
和{provider}_client_secret
添加到应用程序的Secrets
中,位于util/settings.py
github_client_id: str = Field(default="", description="GitHub OAuth client ID")
github_client_secret: str = Field(
default="", description="GitHub OAuth client secret"
)
添加到前端¶
您需要将提供者(api或oauth)添加到frontend/src/components/integrations/credentials-input.tsx
中的CredentialsInput
组件。
const providerIcons: Record<string, React.FC<{ className?: string }>> = {
github: FaGithub,
google: FaGoogle,
notion: NotionLogoIcon,
};
您还需要将提供者添加到frontend/src/components/integrations/credentials-provider.tsx
中的CredentialsProvider
组件。
const CREDENTIALS_PROVIDER_NAMES = ["github", "google", "notion"] as const;
type CredentialsProviderName = (typeof CREDENTIALS_PROVIDER_NAMES)[number];
const providerDisplayNames: Record<CredentialsProviderName, string> = {
github: "GitHub",
google: "Google",
notion: "Notion",
};
最后,您需要将提供者添加到frontend/src/lib/autogpt-server-api/types.ts
中的CredentialsType
枚举。
export type BlockIOCredentialsSubSchema = BlockIOSubSchemaMeta & {
credentials_provider: "github" | "google" | "notion";
credentials_scopes?: string[];
credentials_types: Array<CredentialsType>;
};
示例:GitHub集成¶
- 支持API密钥+OAuth2的GitHub块:
blocks/github
class GithubCommentBlock(Block):
class Input(BlockSchema):
credentials: GithubCredentialsInput = GithubCredentialsField("repo")
issue_url: str = SchemaField(
description="URL of the GitHub issue or pull request",
placeholder="https://github.com/owner/repo/issues/1",
)
comment: str = SchemaField(
description="Comment to post on the issue or pull request",
placeholder="Enter your comment",
)
class Output(BlockSchema):
id: int = SchemaField(description="ID of the created comment")
url: str = SchemaField(description="URL to the comment on GitHub")
error: str = SchemaField(
description="Error message if the comment posting failed"
)
def __init__(self):
super().__init__(
id="a8db4d8d-db1c-4a25-a1b0-416a8c33602b",
description="This block posts a comment on a specified GitHub issue or pull request.",
categories={BlockCategory.DEVELOPER_TOOLS},
input_schema=GithubCommentBlock.Input,
output_schema=GithubCommentBlock.Output,
test_input={
"issue_url": "https://github.com/owner/repo/issues/1",
"comment": "This is a test comment.",
"credentials": TEST_CREDENTIALS_INPUT,
},
test_credentials=TEST_CREDENTIALS,
test_output=[
("id", 1337),
("url", "https://github.com/owner/repo/issues/1#issuecomment-1337"),
],
test_mock={
"post_comment": lambda *args, **kwargs: (
1337,
"https://github.com/owner/repo/issues/1#issuecomment-1337",
)
},
)
@staticmethod
def post_comment(
credentials: GithubCredentials, issue_url: str, body_text: str
) -> tuple[int, str]:
if "/pull/" in issue_url:
api_url = (
issue_url.replace("github.com", "api.github.com/repos").replace(
"/pull/", "/issues/"
)
+ "/comments"
)
else:
api_url = (
issue_url.replace("github.com", "api.github.com/repos") + "/comments"
)
headers = {
"Authorization": credentials.bearer(),
"Accept": "application/vnd.github.v3+json",
}
data = {"body": body_text}
response = requests.post(api_url, headers=headers, json=data)
response.raise_for_status()
comment = response.json()
return comment["id"], comment["html_url"]
def run(
self,
input_data: Input,
*,
credentials: GithubCredentials,
**kwargs,
) -> BlockOutput:
id, url = self.post_comment(
credentials,
input_data.issue_url,
input_data.comment,
)
yield "id", id
yield "url", url
- GitHub OAuth2处理程序:
integrations/oauth/github.py
class GitHubOAuthHandler(BaseOAuthHandler):
"""
Based on the documentation at:
- [Authorizing OAuth apps - GitHub Docs](https://docs.github.com/en/apps/oauth-apps/building-oauth-apps/authorizing-oauth-apps)
- [Refreshing user access tokens - GitHub Docs](https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/refreshing-user-access-tokens)
Notes:
- By default, token expiration is disabled on GitHub Apps. This means the access
token doesn't expire and no refresh token is returned by the authorization flow.
- When token expiration gets enabled, any existing tokens will remain non-expiring.
- When token expiration gets disabled, token refreshes will return a non-expiring
access token *with no refresh token*.
""" # noqa
PROVIDER_NAME = "github"
EMAIL_ENDPOINT = "https://api.github.com/user/emails"
def __init__(self, client_id: str, client_secret: str, redirect_uri: str):
self.client_id = client_id
self.client_secret = client_secret
self.redirect_uri = redirect_uri
self.auth_base_url = "https://github.com/login/oauth/authorize"
self.token_url = "https://github.com/login/oauth/access_token"
def get_login_url(self, scopes: list[str], state: str) -> str:
params = {
"client_id": self.client_id,
"redirect_uri": self.redirect_uri,
"scope": " ".join(scopes),
"state": state,
}
return f"{self.auth_base_url}?{urlencode(params)}"
def exchange_code_for_tokens(
self, code: str, scopes: list[str]
) -> OAuth2Credentials:
return self._request_tokens({"code": code, "redirect_uri": self.redirect_uri})
def _refresh_tokens(self, credentials: OAuth2Credentials) -> OAuth2Credentials:
if not credentials.refresh_token:
return credentials
return self._request_tokens(
{
"refresh_token": credentials.refresh_token.get_secret_value(),
"grant_type": "refresh_token",
}
)
def _request_tokens(
self,
params: dict[str, str],
current_credentials: Optional[OAuth2Credentials] = None,
) -> OAuth2Credentials:
request_body = {
"client_id": self.client_id,
"client_secret": self.client_secret,
**params,
}
headers = {"Accept": "application/json"}
response = requests.post(self.token_url, data=request_body, headers=headers)
response.raise_for_status()
token_data: dict = response.json()
username = self._request_username(token_data["access_token"])
now = int(time.time())
new_credentials = OAuth2Credentials(
provider=self.PROVIDER_NAME,
title=current_credentials.title if current_credentials else None,
username=username,
access_token=token_data["access_token"],
# Token refresh responses have an empty `scope` property (see docs),
# so we have to get the scope from the existing credentials object.
scopes=(
token_data.get("scope", "").split(",")
or (current_credentials.scopes if current_credentials else [])
),
# Refresh token and expiration intervals are only given if token expiration
# is enabled in the GitHub App's settings.
refresh_token=token_data.get("refresh_token"),
access_token_expires_at=(
now + expires_in
if (expires_in := token_data.get("expires_in", None))
else None
),
refresh_token_expires_at=(
now + expires_in
if (expires_in := token_data.get("refresh_token_expires_in", None))
else None
),
)
if current_credentials:
new_credentials.id = current_credentials.id
return new_credentials
def _request_username(self, access_token: str) -> str | None:
url = "https://api.github.com/user"
headers = {
"Accept": "application/vnd.github+json",
"Authorization": f"Bearer {access_token}",
"X-GitHub-Api-Version": "2022-11-28",
}
response = requests.get(url, headers=headers)
if not response.ok:
return None
# Get the login (username)
return response.json().get("login")
示例:Google集成¶
- Google OAuth2处理程序:
integrations/oauth/google.py
class GoogleOAuthHandler(BaseOAuthHandler):
"""
Based on the documentation at https://developers.google.com/identity/protocols/oauth2/web-server
""" # noqa
PROVIDER_NAME = "google"
EMAIL_ENDPOINT = "https://www.googleapis.com/oauth2/v2/userinfo"
DEFAULT_SCOPES = [
"https://www.googleapis.com/auth/userinfo.email",
"https://www.googleapis.com/auth/userinfo.profile",
"openid",
]
您可以看到google定义了一个DEFAULT_SCOPES
变量,这用于设置无论用户请求什么都会请求的范围。
secrets = Secrets()
GOOGLE_OAUTH_IS_CONFIGURED = bool(
secrets.google_client_id and secrets.google_client_secret
)
GOOGLE_OAUTH_IS_CONFIGURED
用于在未配置 OAuth 时禁用需要 OAuth 的块。这在每个块的 __init__
方法中实现。这是因为 Google 块没有 API 密钥回退,因此我们需要确保在允许用户使用这些块之前已配置 OAuth。
需要记住的关键点¶
- 唯一 ID:在 init 方法中为你的块赋予一个唯一的 ID。
- 输入和输出模式:定义清晰的输入和输出模式。
- 错误处理:在
run
方法中实现错误处理。 - 输出结果:使用
yield
在run
方法中输出结果。 - 测试:在 init 方法中提供测试输入和输出以进行自动测试。
理解测试过程¶
块的测试由 test_block.py
处理,其执行以下操作:
- 它使用提供的
test_input
调用块。
如果块有credentials
字段,也会传入test_credentials
。 - 如果提供了
test_mock
,它会临时将指定方法替换为模拟函数。 - 然后它会断言输出与
test_output
匹配。
对于 WikipediaSummaryBlock:
- 测试将使用主题 "Artificial Intelligence" 调用块。
- 它将使用模拟函数而不是进行实际的 API 调用,模拟函数返回
{"extract": "summary content"}
。 - 然后它会检查输出键是否为 "summary" 并且其值是否为字符串。
这种方法使我们能够在不依赖外部服务的情况下全面测试块的逻辑,同时也能适应非确定性输出。
有效块测试的技巧¶
-
提供现实的 test_input:确保你的测试输入涵盖典型用例。
-
定义适当的 test_output:
-
对于确定性输出,使用特定的预期值。
- 对于非确定性输出或仅类型重要时,使用 Python 类型(例如,
str
、int
、dict
)。 -
你可以混合特定值和类型,例如
("key1", str), ("key2", 42)
。 -
对网络调用使用 test_mock:这可以防止测试因网络问题或 API 变化而失败。
-
考虑省略无外部依赖块的 test_mock:如果你的块不进行网络调用或使用外部资源,你可能不需要模拟。
-
考虑边缘情况:包括测试
run
方法中潜在的错误条件。 -
更改块行为时更新测试:如果你修改了块,确保相应地更新测试。
通过遵循这些步骤,你可以创建扩展 AutoGPT Agent Server 功能的新块。
我们希望看到的块¶
以下是我们希望在 AutoGPT Agent Server 中实现的块列表。如果你有兴趣贡献,可以自由选择其中一个块或选择你自己的块。
如果你想实现其中一个块,请打开一个拉取请求,我们将开始审查过程。
消费者服务/平台¶
- Google Sheets -
读取/追加 - 电子邮件 - 使用
Gmail、Outlook、Yahoo、Proton 等读取/发送 - 日历 - 使用 Google Calendar、Outlook Calendar 等读取/写入
- Home Assistant - 调用服务,获取状态
- Dominos - 订购披萨,跟踪订单
- Uber - 预订乘车,跟踪乘车
- Notion - 创建/读取页面,创建/追加/读取数据库
- Google Drive - 读取/写入/覆盖文件/文件夹
社交媒体¶
- Twitter - 发布,回复,获取回复,获取评论,获取关注者,获取关注,获取推文,获取提及
- Instagram - 发布,回复,获取评论,获取关注者,获取关注,获取帖子,获取提及,获取热门帖子
- TikTok - 发布,回复,获取评论,获取关注者,获取关注,获取视频,获取提及,获取热门视频
- LinkedIn - 发布,回复,获取评论,获取关注者,获取关注,获取帖子,获取提及,获取热门帖子
- YouTube - 转录视频/短片,发布视频/短片,读取/回复/反应评论,更新缩略图,更新描述,更新标签,更新标题,获取观看次数,获取点赞,获取不喜欢,获取订阅者,获取评论,获取分享,获取观看时间,获取收入,获取热门视频,获取热门视频,获取热门频道
- Reddit - 发布,回复,获取评论,获取关注者,获取关注,获取帖子,获取提及,获取热门帖子
- Treatwell(及相关平台)- 预订,取消,评论,获取推荐
- Substack - 读取/订阅/取消订阅,发布/回复,获取推荐
- Discord - 读取/发布/回复,执行管理操作
- GoodReads - 读取/发布/回复,获取推荐
电子商务¶
- Airbnb - 预订,取消,评论,获取推荐
- Amazon - 订购,跟踪订单,退货,评论,获取推荐
- eBay - 订购,跟踪订单,退货,评论,获取推荐
- Upwork - 发布工作,雇佣自由职业者,评论自由职业者,解雇自由职业者
商业工具¶
- 外部代理 - 调用其他类似AutoGPT的代理
- Trello - 创建/读取/更新/删除卡片、列表、看板
- Jira - 创建/读取/更新/删除问题、项目、看板
- Linear - 创建/读取/更新/删除问题、项目、看板
- Excel - 读取/写入/更新/删除行、列、工作表
- Slack - 读取/发布/回复消息,创建频道,邀请用户
- ERPNext - 创建/读取/更新/删除发票、订单、客户、产品
- Salesforce - 创建/读取/更新/删除潜在客户、机会、账户
- HubSpot - 创建/读取/更新/删除联系人、交易、公司
- Zendesk - 创建/读取/更新/删除工单、用户、组织
- Odoo - 创建/读取/更新/删除销售订单、发票、客户
- Shopify - 创建/读取/更新/删除产品、订单、客户
- WooCommerce - 创建/读取/更新/删除产品、订单、客户
- Squarespace - 创建/读取/更新/删除页面、产品、订单
我们希望看到的代理模板¶
数据/信息¶
- 通过Apple News或其他大型媒体(如BBC、TechCrunch、hackernews等)总结今日、本周、本月的头条新闻
- 创建、读取并总结substack新闻简报或其他任何新闻简报(博客作者 vs 博客读者)
- 获取/读取/总结当天、本周、本月的Twitter、Instagram、TikTok(一般社交媒体账户)最热门内容
- 获取/读取任何提到AI代理的LinkedIn帖子或个人资料
- 读取/总结discord内容(可能无法实现,因为需要访问权限)
- 读取/获取GoodReads或Amazon Books等平台上某月、某年等最受欢迎的书籍
- 获取特定节目在所有流媒体服务上的日期
- 推荐/获取某月、某年等在所有流媒体平台上最受欢迎的节目
- 从xlsx数据集中进行数据分析
- 通过Excel或Google Sheets收集数据 > 随机抽样数据(抽样块取前X、后X、随机等)> 传递给LLM块生成完整数据分析脚本 > Python块运行脚本 > 通过LLM修复块处理错误 > 创建图表/可视化(可能在代码块中?) > 显示图像输出(这可能需要前端更改以显示)
- Tiktok视频搜索和下载
市场营销¶
- 作品集网站设计和增强