Day 02 / 共 20 天 · 第 1 周 核心概念

Block：积木的概念

第一个核心概念。Block 是有"输入口/输出口"和 run() 逻辑的可复用功能积木。今天建立心智模型、读一个最简 Block 的真实代码。核心代码在 blocks/_base.py。

📍 你在整门课的位置 · 第 1 周 核心概念
D1 全景→ D2 Block 概念→ D3 Graph 图→ D4 运行平台
💡 今天的类比世界观：Block = 一台"带插头的家用电器" Block 就像一台家用电器：有进料口（输入口）、出料口（输出口），内部一个马达（run）干一件事。Input/Output Schema = 说明书上写清"能插什么、会出什么"；SchemaField = 面板上带标注的旋钮（数值范围多少、默认几档、给前端画什么控件）；error 引脚 = 电器的漏电保护/保险丝，出故障从专门的口报警，而不是整机炸掉。今天都用"电器"来想。

👶 小白：run() 为什么要写成 async def 还带 yield（异步生成器），不直接 return 一个结果？

👨‍🏫 老师：因为一台设备可能陆续吐出好几件产品，而不是一次给一个。return 像自动贩卖机"咣当"掉一件就结束；yield 像流水线设备加工好一件就送出一件、边算边出——下游能立刻接着处理，不用干等全部算完。而 async 让设备"联网等外部响应时不霸占整条线"，好让别的块并行干活。

L01

Block 心智模型

🤔 痛点：为什么不把"读网页+调 LLM+发邮件"写成一个大函数就好？ 写成一坨代码，换个场景（比如只想读网页不发邮件）就得复制粘贴改一遍；非程序员更是完全没法用。能不能把每个小能力做成标准零件，让人像拼乐高一样自由组合、还能复用？
💡 本质：Block = 一个"有标准接口口子"的可复用功能单元 AutoGPT 把每个能力封装成 Block：规定好输入口（要什么）、输出口（给什么）、中间一个 run() 干活。有了统一的"口子规范"，任意两个 Block 就能对接、复用、可视化拖拽——这就是"积木"心智，也是整个平台的原子。

一个 Block 就像电路里的一个元件——有输入引脚、输出引脚，中间一段处理逻辑：

输入 → WordCount Block数单词和字符 → 输出

输入口：text　｜　输出口：word_count、character_count

先澄清一个坑：文件名有迷惑性——data/block.py 里没有 Block 基类！它只放类型别名和"同步到数据库"的逻辑。真正的 Block 基类在 blocks/_base.py:550。
Block vs Node 的区别（重要） Block 是"能力"（一个类）——描述"这种积木能干什么"。Node 是"这个 Block 在某张图里的一次摆放"（Day 03）——带具体的连线和配置。就像"电阻"这个元件型号（Block）vs "电路图里第 3 个位置那个电阻"（Node）。同一个 LLM Block 可以在一张图里放好几份 Node，各自配置不同。Block = 逻辑/代码，Node = 数据/配置——逻辑与配置分离。
⚠️ 常见误解：小白常以为"Block 基类在 data/block.py"——名字太像了。其实那里只放类型别名和"同步到数据库"逻辑，真正的 Block 抽象基类在 blocks/_base.py:550。记住：data/ 放数据模型，blocks/_base.py 放能力基类。
🧱 一句话 + 类比 Block 就像生活中的乐高积木：每块有固定的凸点（输出口）和凹槽（输入口），你不用管它内部塑料怎么注塑成型（run() 的实现），只要凸凹接口对得上就能拼。一句话：Block = 一块规定好接口的乐高，"选块 + 拼接"就能搭出任意造型。
L02

run 是异步生成器（关键设计）

看类型别名（data/block.py:15）——它揭示了 Block 的核心：

BlockInput = dict[str, Any]                          # 输入：口名 → 数据
BlockOutputEntry = tuple[str, Any]                   # 输出：(口名, 值)
BlockOutput = AsyncGenerator[BlockOutputEntry, None] # 输出：一个异步生成器！
读法：run() 的返回类型是异步生成器——它用 yield ("口名", 值) 可以产出多次。所以一个 Block 天然支持"一个输入 → 多个输出"，甚至流式地陆续产出。
普通 return（只能给 1 次） return 值 1 个结果，函数结束 async generator（yield 多次） run() 异步生成器 yield "word_count", 4 yield "character_count", 19 yield ... （可继续，可流式）
上：普通函数 return 一次就结束；下：run() 是异步生成器，能 yield 多个 (口名, 值)——所以一个 Block 可有多个输出口、可流式产出。
为什么用生成器而不是普通 return？ 普通函数 return 一次就结束、只能返回一个值。生成器能 yield 很多次——这让 Block 能：① 有多个输出口（yield "word_count", 4 然后 yield "character_count", 19）；② 流式产出（比如一个"逐行读文件"块，每读一行 yield 一次，下游能边收边处理）；③ 一个输出口产出多个值（列表展开）。异步（async）则让 Block 里能 await 网络请求等 I/O 而不阻塞。这个设计让 Block 既灵活又高效。
L03

最简 Block 精读

看一个最纯粹的 Block（blocks/count_words_and_char_block.py:11）——理解结构的最佳入门：

class WordCharacterCountBlock(Block):
    class Input(BlockSchemaInput):          # ← 内嵌输入 schema
        text: str = SchemaField(description="要统计的文本", advanced=False)
    class Output(BlockSchemaOutput):        # ← 内嵌输出 schema
        word_count: int = SchemaField(description="单词数")
        character_count: int = SchemaField(description="字符数")
        error: str = SchemaField(...)       # error 口（L06）
    def __init__(self):
        super().__init__(id="ab2a782d-...", categories={BlockCategory.TEXT},
                         input_schema=..., output_schema=...,
                         test_input={"text": "Hello, how are you?"},
                         test_output=[("word_count", 4), ("character_count", 19)])
    async def run(self, input_data: Input, **kwargs) -> BlockOutput:
        yield "word_count", len(input_data.text.split())
        yield "character_count", len(input_data.text)
读法：一个 Block 的标准结构三段式：① 内嵌 Input/Output 两个类（定义有哪些输入/输出口）；② __init__ 里声明元数据（唯一 id、分类、测试样例）；③ run() 里 yield 输出。这个块无副作用、无凭据，最纯粹。
📝 举个例子：喂给这个 Block 一句话，它 yield 什么 输入 input_data.text = "Hello, how are you?"：
① run() 执行 "Hello, how are you?".split() = ["Hello,","how","are","you?"] → 长度 4 → yield "word_count", 4；
② 执行 len("Hello, how are you?") = 19（含空格标点）→ yield "character_count", 19。
这正好等于 Block 自带的 test_output=[("word_count", 4), ("character_count", 19)]——测试用例就是"标准答案"。
注意 test_input/test_output 每个 Block 都自带测试样例——给定 test_input 应该产出 test_output。这是很好的设计：Block 自带"我该怎么用、正确输出长啥样"的示例，既能自动测试、又是文档。把测试数据做成 Block 定义的一部分——保证每个积木都可验证。
L04

Input/Output Schema

BlockSchema（blocks/_base.py:194）继承 Pydantic BaseModel，是输入/输出的"类型契约"。关键能力：

  • jsonschema()（:197）：转成前端能用的 JSON Schema——前端画布就靠它画出"输入表单/输出引脚"。
  • validate_data()（:227）：执行前校验用户填的输入。
  • 派生出 BlockSchemaInput（输入基类）、BlockSchemaOutput（输出基类，自带 error 字段）。
一个聪明的自动校验（_base.py:337 的 __pydantic_init_subclass__）：每当有人定义新 Schema 子类，Pydantic 自动跑校验，强制命名规范——名为 credentials 或 *_credentials 的字段必须是凭证类型，反之亦然。这保证整个平台凭证字段命名统一，前端能据此识别"哪些字段要弹连接账号的 UI"（Day 08）。
L05

SchemaField：带 UI 元数据的字段

Block 不用 Pydantic 原生 Field，而用封装的 SchemaField（data/model.py:263）——它把 UI/行为元数据塞进字段：placeholder、secret（密码型，前端打码）、advanced（收进高级折叠区）、hidden、depends_on（字段依赖）等。

一个贴心的默认值逻辑 model.py:283：字段没有默认值 → 强制 advanced=False（因为是必填，得让用户显眼地看到）；有默认值且没显式指定 → 默认 advanced=True（收进"高级"折叠区，不打扰新手）。框架用这个小逻辑自动决定"哪些输入口显眼、哪些藏起来"——让画布界面对新手友好（只显示必填的），对高级用户也够用（展开高级）。这种对 UI 体验的用心，是"面向非程序员的可视化平台"该有的细致。
L06

error 引脚约定

BlockSchemaOutput 默认带一个 error: str 字段（_base.py:471）。这是全局的错误处理约定——看执行包装 _execute()（_base.py:922）：

async for output_name, output_data in self.run(...):
    if output_name == "error":
        raise BlockExecutionError(...)   # ★ yield "error" = 失败信号，转成异常
    # 每条输出还按 schema 校验
    yield output_name, output_data
读法：Block 只要 yield "error", "出错了..."，框架就把它转成异常、中断执行。这就是为什么每个输出 schema 默认带 error 字段——它是"Block 报告失败"的统一渠道。
为什么用"约定的 error 口"而不是抛异常？ 两种都支持，但 error 口更符合"数据流"心智——Block 的一切产出（包括失败）都是"从某个口 yield 出的数据"。而且 error 是一个输出口，可以被连线——你能把某个块的 error 口连到"发告警"块，实现"这步失败就通知我"的可视化错误处理。把错误也变成可连线的输出口——这是数据流范式的优雅之处。另外注意 _execute() 还做了输入校验、敏感操作人工审核、输出 schema 校验——它是 run() 外面的一层"执行生命周期管理"。
L07

自动发现注册

怎么让 300+ 个 Block 被系统认识？load_all_blocks()（blocks/__init__.py:17）——放进 blocks/ 目录就自动加载，无需手动注册：

# 递归扫描 blocks/ 下所有 .py，逐个 import，
# 再用 Block.__subclasses__() 找出所有 Block 子类
# 结果 @cached(ttl=3600) 缓存一小时
读法：约定优于配置——你在 blocks/ 目录放一个符合规范的 Block 类文件，它自动被发现注册。加载时有一堆严格校验（fail-fast）：类名必须以 Block 结尾、id 必须是唯一 UUID、字段必须是 SchemaField……不合规立刻报错。
initialize_blocks()（data/block.py:21）再把内存里的 Block 同步到数据库（AgentBlock 表）——比对 id/schema 有变化就 upsert。为什么要同步到 DB？因为前端要列出所有可用 Block（画布左侧的积木清单）、Graph 里的 Node 用 block_id 引用 Block——这些都需要 Block 在数据库里有记录。"代码里定义 Block → 自动发现 → 同步到 DB → 前端可拖拽"这条链让加新积木极其简单。
L08

今日小结 + 动手

🧠 今天你应该能回答

  • Block 是什么？和 Node 的区别？（能力/类 vs 摆放/数据）
  • run 为什么是异步生成器？带来什么能力？
  • 一个 Block 的标准三段式结构？test_input 有什么用？
  • error 引脚约定是什么？为什么把错误做成输出口？
  • Block 怎么被自动发现和注册？

✋ 动手

P=autogpt_platform/backend/backend
sed -n '15,20p' $P/data/block.py                          # 类型别名
sed -n '11,40p' $P/blocks/count_words_and_char_block.py   # 最简 Block
sed -n '456,490p' $P/blocks/_base.py                      # BlockSchemaInput/Output
sed -n '17,60p' $P/blocks/__init__.py                     # 自动发现
明天预告 · Day 03：第二个核心概念——Graph（图）。读 data/graph.py：Node（节点）+ Link（连线）怎么把 Block 连成一个智能体、图怎么校验、起始节点怎么确定、agent-as-block 子图。
← Day 01 全景 Day 03 · Graph 图 →