跳到主要内容

AutoGPT — 积木的契约(Block 抽象)

本章讲什么: AutoGPT 里一切都是「块」。搞懂一个块的契约(输入怎么定义、输出怎么产生、execute() 帮你做了什么),后面的图和引擎就只是「把很多块按依赖跑起来」。

1. 一个块要解决的小问题​

平台有 ~180 个块(指用户在画布上可见、已注册的具体块;按 id 数,需运行 load_all_blocks 才能精确定数,代码里 class XxxBlock( 定义约 500+ 处含抽象/基类,不可直接 grep 当数)。它们五花八门:调 LLM、抓网页、发 Slack、做加法。要让一个通用引擎能调度它们、能在前端画布上展示它们的输入框、能校验用户填的数据、能计费——就必须让每个块都遵守同一份契约。

这份契约就是 Block 抽象基类(blocks/_base.py:550 Block)。

2. 思路/直觉:块 = 带类型的异步生成器​

核心直觉三句话:

  • 输入是一个 Pydantic 模型(Input 子类)——既是类型定义,又能自动生成给前端的 JSON Schema 表单。
  • 输出不是 return 一个值,而是 yield 多个 (pin名, 数据)——因为一个块可以有多个输出管脚(pin),还可能流式产出多条。
  • run() 是 async 生成器——天然支持流式 LLM、长任务、产出多条结果。
# 示意,非源码 —— 一个最小块的形状
class AddBlock(Block):
    class Input(BlockSchemaInput):
        a: int = SchemaField(description="加数 a")
        b: int = SchemaField(description="加数 b")
    class Output(BlockSchemaOutput):
        sum: int = SchemaField(description="和")

    async def run(self, input_data: Input, **kwargs):
        yield "sum", input_data.a + input_data.b   # (pin 名, 值)

3. 真实实现:一个块的三件套​

看真源码里最简单的有用块 StoreValueBlock(把输入原样存住并转发,可被多次消费):

# blocks/basic.py:69 StoreValueBlock
class Input(BlockSchemaInput):
    input: Any = SchemaField(description="Trigger ...")
    data: Any = SchemaField(description="...", default=None)
class Output(BlockSchemaOutput):
    output: Any = SchemaField(description="The stored data ...")

async def run(self, input_data: Input, **kwargs) -> BlockOutput:
    yield "output", input_data.data or input_data.input

三件套都在 __init__ 里登记(blocks/basic.py:90):

  • id:一个固定的 UUID,持久化在数据库里、永不变(blocks/_base.py:576 注释:「persisted in the DB ... unique and constant」)。块改名了 id 也不能变,否则老图里的节点就指不到块了。
  • input_schema / output_schema:上面那两个 Pydantic 子类。
  • test_input / test_output:平台有一套自动跑「每个块都能正常工作」的测试(AGENTS.md 里 test_block.py),靠的就是这对样例。
  • static_output=True:声明这个块的输出可被下游多次消费(见 02 章静态 pin)。

3.1 输出/输入的类型别名​

块的输入输出在类型层面就定死了形状(data/block.py:15):

# data/block.py:15
BlockInput = dict[str, Any]                 # 1 个输入 pin <- 1 份数据
BlockOutputEntry = tuple[str, Any]          # 输出 = (名字, 值)
BlockOutput = AsyncGenerator[BlockOutputEntry, None]  # 1 个输出 pin -> N 份数据

「输入是 dict、输出是 (名, 值) 的异步流」——这是引擎和块之间唯一的数据接口。

4. 深入:execute() 这条校验管线​

引擎从不直接调 run(),而是调 block.execute()(blocks/_base.py:727)→ 它再调 _execute()(blocks/_base.py:819)。_execute 是一条前后都包了校验和拦截的管线。按顺序它做了这些事:

输入 dict
  │
  ├─① 人审拦截     is_block_exec_need_review() —— 敏感动作 + 安全模式开 → 暂停等审批
  │
  ├─② 输入校验     input_schema.validate_data() —— 不符 schema 直接 BlockInputError
  │                (dry-run 时放过 credentials 字段)
  │
  ├─③ 自动凭据兜底  缺 auto-credentials → 明确报错而不是让 run() 崩在 SDK 深处
  │
  ├─④ 真正执行     async for ... in self.run(input_schema(**非空字段)):
  │       │
  │       ├─ 若 yield 的 pin 名 == "error" → 抛 BlockExecutionError(块自报错误的约定)
  │       ├─ 标准块还会校验每个输出值符合 output schema
  │       └─ yield 给上层
  └────────────────────────────────────────────────────────────

几个值得记住的设计:

  • error 是一个约定俗成的输出 pin。 任何块只要 yield "error", "出错了",_execute 就会把它转成异常停掉节点(blocks/_base.py:927)。所以 BlockSchemaOutput 基类自带一个 error: str 字段(blocks/_base.py:471)——这是平台级的统一错误通道。

  • 异常分级包装。 execute()(blocks/_base.py:739)把 run() 里冒出来的异常分类:已经是 BlockError 的原样抛;ValueError(通常是用户输入问题)包成 BlockExecutionError;其它未知的包成 BlockUnknownError。这让上层引擎能区分「用户的错」和「平台的错」(后者才上报 Sentry)。

  • 凭据字段的命名是被强制的。 schema 在子类化时就校验:名为 credentials 或 *_credentials 的字段必须是 CredentialsMetaInput 类型,反之亦然(blocks/_base.py:338 __pydantic_init_subclass__)。这保证引擎能可靠地找出「哪些字段是凭据」从而去凭据库取真值。

5. 块是怎么被「发现」的(注册表)​

没有手写的注册列表。启动时 load_all_blocks()(blocks/__init__.py:17)扫描 blocks/ 目录下所有 .py,import 它们,然后用 Block.__subclasses__() 递归找出所有子类(blocks/__init__.py:115 _all_subclasses)。

这一步顺带做了一堆启动期硬校验(blocks/__init__.py:51 起),不合规直接启动失败:

规则代码位置
类名必须以 Block 结尾(抽象类以 Base 结尾)blocks/__init__.py:54
id 必须是合法 36 字符 UUID,且全局唯一blocks/__init__.py:66
每个字段必须是 SchemaField(带 json_schema_extra)blocks/__init__.py:91
布尔字段必须有默认值blocks/__init__.py:96
error 输出字段必须是 strblocks/__init__.py:80

结果用 @cached(ttl_seconds=3600) 缓存,所以 get_blocks() 很便宜。运行时一个块实例由 get_block(block_id)(blocks/__init__.py:130)按 id 取到。

巧妙之处: 「约定即注册」。加一个新块只需在 blocks/ 下新建文件、写个 XxxBlock(Block),启动时自动被发现并校验。无需改任何中央清单——这对一个有几十个贡献者、~180 个块的项目是关键的可扩展性设计。

6. 边界与局限​

  • 块不能跨实例共享内存状态:run() 是无状态函数,要存东西得用 StoreValueBlock 或外部存储。这是 dataflow 模型的必然约束。
  • 同一份凭据同一时刻只能被一个块用——执行引擎对凭据加了 Redis 锁(见 03 章),并发跑两个用同一 key 的块会串行化。
  • 块的输出 schema 校验只对 BlockType.STANDARD 生效(blocks/_base.py:931);特殊块(Agent、Webhook 等)绕过。

7. 横向对比​

  • 和 LangGraph / 经典 agent 框架比:AutoGPT 的「块」更接近 ETL/iPaaS 的算子而非「工具(tool)」。工具是给 LLM 选的;这里块是用户手工连线的图节点,LLM 只是其中一种块。
  • 和老的 classic AutoGPT(同仓库 classic/,已停更)比:老版是「LLM 自己决定下一步」的自治循环;新平台把控制流从 LLM 手里拿回来,交给用户画的图——更可控、可计费、可托管,代价是少了「完全自主」。

8. 代码地图​

主题文件符号
Block 基类autogpt_platform/backend/backend/blocks/_base.pyBlock
执行管线(校验/人审/包装)blocks/_base.pyBlock._execute、Block.execute
输入/输出 schema 基类blocks/_base.pyBlockSchemaInput、BlockSchemaOutput
凭据字段命名校验blocks/_base.pyBlockSchema.__pydantic_init_subclass__
类型别名data/block.pyBlockInput、BlockOutput、BlockOutputEntry
块自动发现 + 启动校验blocks/__init__.pyload_all_blocks、_all_subclasses
取块blocks/__init__.pyget_block、get_blocks
最简示例块blocks/basic.pyStoreValueBlock