Skip to content

为AutoGPT Agent Server贡献:创建和测试Blocks

本指南将引导你完成为AutoGPT Agent Server创建和测试新Block的过程,以WikipediaSummaryBlock为例。

理解Blocks和测试

Blocks是可重用的组件,可以连接起来形成一个表示代理行为的图。每个Block都有输入、输出和特定的功能。正确的测试对于确保Blocks正确且一致地工作至关重要。

创建和测试新Block

按照以下步骤创建和测试新Block:

  1. backend/blocks目录中创建一个新的Python文件。为其命名并使用snake_case。例如:get_wikipedia_summary.py

  2. 导入必要的模块并创建一个继承自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的实现将在这里
    
  3. 使用BlockSchema定义输入和输出模式。这些模式指定了Block期望接收(输入)和生成(输出)的数据结构。

  4. 输入模式定义了Block将处理的数据结构。模式中的每个字段代表一个必需的输入数据。

  5. 输出模式定义了Block处理后将返回的数据结构。模式中的每个字段代表一个输出数据。

    示例:

    class Input(BlockSchema):
        topic: str  # 要获取Wikipedia摘要的主题
    
    class Output(BlockSchema):
        summary: str  # 来自Wikipedia的主题摘要
        error: str  # 如果请求失败,任何错误消息,错误字段需要命名为`error`。
    
  6. 实现__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_schemaoutput_schema:定义输入和输出数据的结构。

    让我们分解测试组件:

    • test_input:这是用于测试Block的示例输入。它应根据你的输入模式是有效的输入。

    • test_output:这是使用test_input运行Block时的预期输出。它应匹配你的输出模式。对于非确定性输出或当你只想断言类型时,可以使用Python类型而不是特定值。在这个例子中,("summary", str)断言输出键是"summary",其值是字符串。

    • test_mock:这对于进行网络调用的Block至关重要。它提供了一个模拟函数,在测试期间替换实际的网络调用。

    在这种情况下,我们正在模拟get_request方法,使其始终返回一个带有'extract'键的字典,模拟成功的API响应。这使我们能够在不进行实际网络请求的情况下测试Block的逻辑,实际网络请求可能会很慢、不可靠或受速率限制。

  7. 实现带有错误处理的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:
        ...

凭证将由后端的执行器自动注入。

APIKeyCredentialsOAuth2Credentials模型定义在这里。 要在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]接口的以下部分:

autogpt_platform/backend/backend/integrations/oauth/base.py
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本身,将处理程序添加到系统中还需要另外两件事:

autogpt_platform/backend/backend/integrations/oauth/__init__.py
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
autogpt_platform/backend/backend/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组件。

frontend/src/components/integrations/credentials-input.tsx
const providerIcons: Record<string, React.FC<{ className?: string }>> = {
  github: FaGithub,
  google: FaGoogle,
  notion: NotionLogoIcon,
};

您还需要将提供者添加到frontend/src/components/integrations/credentials-provider.tsx中的CredentialsProvider组件。

frontend/src/components/integrations/credentials-provider.tsx
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枚举。

frontend/src/lib/autogpt-server-api/types.ts
export type BlockIOCredentialsSubSchema = BlockIOSubSchemaMeta & {
  credentials_provider: "github" | "google" | "notion";
  credentials_scopes?: string[];
  credentials_types: Array<CredentialsType>;
};

示例:GitHub集成

blocks/github/issues.py
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
blocks/github/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集成

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变量,这用于设置无论用户请求什么都会请求的范围。

blocks/google/_auth.py
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 方法中实现错误处理。
  • 输出结果:使用 yieldrun 方法中输出结果。
  • 测试:在 init 方法中提供测试输入和输出以进行自动测试。

理解测试过程

块的测试由 test_block.py 处理,其执行以下操作:

  1. 它使用提供的 test_input 调用块。
    如果块有 credentials 字段,也会传入 test_credentials
  2. 如果提供了 test_mock,它会临时将指定方法替换为模拟函数。
  3. 然后它会断言输出与 test_output 匹配。

对于 WikipediaSummaryBlock:

  • 测试将使用主题 "Artificial Intelligence" 调用块。
  • 它将使用模拟函数而不是进行实际的 API 调用,模拟函数返回 {"extract": "summary content"}
  • 然后它会检查输出键是否为 "summary" 并且其值是否为字符串。

这种方法使我们能够在不依赖外部服务的情况下全面测试块的逻辑,同时也能适应非确定性输出。

有效块测试的技巧

  1. 提供现实的 test_input:确保你的测试输入涵盖典型用例。

  2. 定义适当的 test_output

  3. 对于确定性输出,使用特定的预期值。

  4. 对于非确定性输出或仅类型重要时,使用 Python 类型(例如,strintdict)。
  5. 你可以混合特定值和类型,例如 ("key1", str), ("key2", 42)

  6. 对网络调用使用 test_mock:这可以防止测试因网络问题或 API 变化而失败。

  7. 考虑省略无外部依赖块的 test_mock:如果你的块不进行网络调用或使用外部资源,你可能不需要模拟。

  8. 考虑边缘情况:包括测试 run 方法中潜在的错误条件。

  9. 更改块行为时更新测试:如果你修改了块,确保相应地更新测试。

通过遵循这些步骤,你可以创建扩展 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视频搜索和下载

市场营销

  • 作品集网站设计和增强