AutoGPT Platform 数据类 Block 完全指南：字典、列表、JSON、文件、持久化、SQL 与截图实战解析

本文是一份围绕 AutoGPT Platform 可视化流程中 Data（数据处理）类 Block 的完整技术指南。Data 类 Block 用于创建、读取与操作各类数据结构，覆盖字典、列表、CSV/Excel 表格、JSON 序列化、跨运行持久化存储，以及 SQL 查询与网页截图等能力。读完本文，你将掌握每个 Block 的输入/输出契约、参数默认值与边界约束、源码级实现原理，并能直接在工作流搭建中组合出符合实际业务需求的数据处理管线。

---

一、Data 类 Block 概览：它们负责什么

在 AutoGPT Platform（autogpt_platform/）中，Block 是构建 Agent 工作流的最小可复用单元。Data 类 Block 在代码中通过 BlockCategory.DATA 归类，负责"数据结构的创建、读取与加工"，官方定义覆盖：列表（List）、字典（Dictionary）、表格（Spreadsheet）与持久化存储（Persistent storage）几大类。它们通常不调用外部 AI 模型，而是充当工作流中的数据搬运、转换与落地节点。

从源码位置看，Data 类 Block 分散在 autogpt_platform/backend/backend/blocks 目录下的多个模块中：

Block	所属源码文件	归类
Create Dictionary / Create List	data_manipulation.py	DATA
File Read	text.py	TEXT + DATA
JSON Decoder / JSON Encoder	json_blocks.py	DATA
Persist Information / Retrieve Information	persistence.py	DATA
Read Spreadsheet	spreadsheet.py	TEXT + DATA
SQL Query	sql_query_block.py	DATA
Screenshot Web Page	screenshotone.py	DATA

下面按功能族逐一拆解。

---

二、字典与列表：结构化数据的"积木"

Create Dictionary（创建字典）

定位：一次性创建包含指定键值对的字典。适合"启动时已知全部取值"的场景，而非增量构建。其源码实现非常直白——输入值经过 Pydantic 模式校验后原样透传：

async def run(self, input_data: Input, **kwargs) -> BlockOutput:
    try:
        yield "dictionary", input_data.values
    except Exception as e:
        yield "error", f"Failed to create dictionary: {str(e)}"

输入输出契约：

方向	字段	类型	必填	说明
输入	values	Dict[str, Any]	是	要组成字典的键值对（编辑器占位示例：{'name': 'Alice', 'age': 25}）
输出	dictionary	Dict[str, Any]	—	创建出的字典
输出	error	str	—	创建失败时的错误信息

该 Block 的注册 ID 为 b924ddf4-de4f-4b56-9a85-358930dcbc91，在 __init__ 中同时内置了 test_input/test_output 用于框架自检（如 {"name": "Alice", "age": 25, "city": "New York"} 原样输出）。

典型场景：

• API 请求负载构造：先把全部必填字段拼成一个完整请求体对象，再交给 HTTP 请求类 Block；
• 配置对象：用预定义值初始化服务或工作流所需的设置字典；
• 数据映射：把输入数据转换成下游 Block 期望的特定 key 结构。

Create List（创建列表，支持按规模/Token 分批产出）

定位：把一组值组合成列表；同时具备"分批产出"能力——当设置了 max_size 或 max_tokens 时，会按批次（chunk）逐次 yield，而不是一次性吐出一个大列表。这对处理受 API 限流或内存约束约束的大数据集尤其有用。

输入输出契约：

方向	字段	类型	必填	说明
输入	values	List[Any]	是	要合并成新列表的值（占位示例：['Alice', 25, True]）
输入	max_size	int	否	每批最大元素数；设置后列表将按该大小分批产出
输入	max_tokens	int	否	每批最大 Token 数；设置后每批元素总量不超该 Token 上限
输出	list	List[Any]	—	创建出的列表（应用分批时为其中一批）

源码级分批原理（见 data_manipulation.py 中 CreateListBlock.run）：它遍历 values，对每个元素先用 estimate_token_count_str(value)（来自 backend.util.prompt）估算 Token 数；若"累加后超出任一限制"则先把已收集的 chunk yield 出去再开新批；最后在列表为空或收尾时仍会产出一个 chunk（保证空输入也有产出路径）。max_size/max_tokens 字段在源码中都标为 advanced=True，即编辑器默认折叠、需展开高级选项配置。

典型场景：

• 批量处理：为有调用速率限制的 API 把大数据集切成合理大小；
• LLM Token 管理：按 Token 上限切分文本块交给语言模型处理；
• 并行处理：生成一批工作项，交由多个 Block 并发消费。

提示：同文件还提供 AddToDictionaryBlock、FindInDictionaryBlock、AddToListBlock、FindInListBlock、ConcatenateListsBlock、FlattenListBlock 等更细粒度的操作类 Block（归类 BASIC），构建复杂数据结构时可结合使用。

---

三、File Read：按行/按字节流式读取任意文件

定位：从 URL、data URI 或本地路径读取文件，以字符串返回内容；并支持按分隔符、字节大小做分块（chunk）产出，配合跳行/跳字节参数，做到"大文件不整体载入内存即可流式处理"。

源码执行链路（见 text.py 的 FileReadBlock.run）：

1. 通过 store_media_file(...) 统一落地媒体文件（统一处理 URL、data URI 等来源），再经 get_exec_file_path() 解析出真实本地路径，若文件不存在直接抛 ValueError；
2. 以 UTF-8 打开读取；若抛 UnicodeDecodeError 则回退 latin-1 编码再次尝试；
3. 应用 skip_size（按字符跳过开头）→ 按 delimiter 切分为行/条目 → 应用 skip_rows（跳过前 N 条）→ 应用 row_limit（截取前 N 条）；
4. 将剩余内容按 size_limit（字节）切成多个 chunk 逐个 yield；当 size_limit <= 0 时整段作为一个 chunk 产出。

输入输出契约：

方向	字段	类型	必填	默认值	说明
输入	file_input	str (file)	是	—	要读取的文件（URL、data URI 或本地路径）
输入	delimiter	str	否	""	用于把内容切成行/chunk 的分隔符（如 '\n' 表示按行）
输入	size_limit	int	否	0	每个 chunk 的最大字节数（0 表示不限）
输入	row_limit	int	否	0	最多处理的行数（0 表示不限；需配合 delimiter）
输入	skip_size	int	否	0	从文件开头跳过的字符数
输入	skip_rows	int	否	0	从开头跳过的行数（需配合 delimiter）
输出	content	str	—	—	文件内容；应用分隔/大小限制时按 chunk 逐个产出
输出	error	str	—	—	操作失败时的错误信息

该 Block 的测试用例演示了 data URI 输入："data:text/plain;base64,SGVsbG8gV29ybGQ=" 会被解码为内容 "Hello World"。

典型场景：

• 日志处理：逐行读取并过滤、转换日志条目；
• 大文档分析：分块读取长文本做摘要/分析，规避内存问题；
• 数据导入：逐行读取文本数据文件以入库。

---

四、JSON 编解码：与结构化数据的桥接

两个 Block 位于 json_blocks.py，底层都使用项目封装并优化的 orjson（backend.util.json 中的 loads/dumps），而非标准库 json，以获得更高序列化性能。

JSON Encoder（编码器）

将任意值/结构编码为 JSON 字符串，支持 dict、list、str、int、float、bool、None 的嵌套组合（Python None → JSON null，布尔值自动对应 true/false）。若传入无法序列化的类型（自定义对象、datetime、无自定义序列化的 set），run 会捕获 orjson.JSONEncodeError / RecursionError / TypeError / ValueError 并统一提升为带 "JSON Encoding Error" 前缀的 ValueError，中止该 Block 执行并汇入框架的异常处理。官方建议：对大数值精度问题或不可序列化类型，先转成字符串或字典再喂给编码器。

输入输出契约：

方向	字段	类型	必填	说明
输入	data	Data	是	要编码的对象/列表/字符串等任意结构
输出	json_str	str	—	输入数据的 JSON 字符串表示

测试用例：{"name": "AutoGPT", "active": True} → '{"name":"AutoGPT","active":true}'。

典型场景：POST/PUT 请求体构造、把结构化工流数据导出成 JSON 保存、复杂结构序列化用于结构化日志。

JSON Decoder（解码器）

将 JSON 字符串解码为原生 Python 结构（对象、列表、字符串、数字等）。合法输入必须严格符合 JSON 语法，例如 '{"active": true, "val": null}' 会解码成 Python 字典，其中 true→True、null→None。若输入格式非法（缺引号、尾随逗号等），内部解析器抛异常，Block 捕获后提升为 "JSON Decoding Error" 前缀的 ValueError 中止执行。空字符串、深层嵌套等边界场景由解析器安全处理（极深嵌套可能受解析递归深度限制）。

输入输出契约：

方向	字段	类型	必填	说明
输入	json_str	str	是	待解码的 JSON 字符串
输出	data	Data	—	从 JSON 字符串解码出的值

典型场景：解析外部 API 的 JSON 响应、解码 JSON 配置字符串、从 JSON Webhook 负载提取嵌套字段做动态决策。

---

五、Persist Information / Retrieve Information：跨运行键值持久化

这对 Block 为 Agent 提供"跨多次运行仍然生效的记忆"，适合保存上次状态、计数器、累计数据等。注意：同一 key 被并行读写时存在 read→write 竞态条件，官方文档与类 docstring 均明确提示需谨慎。

作用域（scope）机制与存储键生成

持久化的隔离程度由 scope 决定（代码见 persistence.py 的 get_storage_key）：

scope	语义	实际存储键
within_agent（默认）	该 Agent 的所有运行共享	agent#{graph_id}#{key}
across_agents	该用户的所有 Agent 共享	global#{key}

底层读写经由 get_database_manager_async_client() 的 set_execution_kv_data / get_execution_kv_data 方法，落在执行期键值数据存储中；数据在被显式覆盖前持续有效，从而支撑状态管理与配置跨运行保持。

Persist Information 输入输出

方向	字段	类型	必填	说明
输入	key	str	是	存储信息所用的键
输入	value	Value	是	要存储的值
输入	scope	"within_agent" | "across_agents"	否	持久化作用域（默认 within_agent）
输出	value	Value	—	被存储的值（写入成功即回显）

Retrieve Information 输入输出

方向	字段	类型	必填	说明
输入	key	str	是	要取回信息的键
输入	scope	"within_agent" | "across_agents"	否	与写入时一致的作用域
输入	default_value	Default Value	否	key 不存在时返回的默认值
输出	value	Value	—	取回的值，或默认值

组合使用案例（摘自文档）：

• 用户偏好：保存语言、通知偏好等，供后续运行使用；
• 进度追踪：记录"最后处理到哪个 item ID"，实现断点续批；
• API Token 缓存：跨多次执行复用刷新后的 Token；
• 状态恢复：工作流启动时用 Retrieve Information 读回上次保存的运行状态以保持连续性。

---

六、Read Spreadsheet：读取 CSV 与 Excel 表格

定位：读取 CSV 与 Excel 文件，输出为"字典列表 + 逐行产出"两种形态；Excel（.xlsx/.xls）会被自动转换为 CSV 后再解析。解析规则为：每一行转成一个字典，列头作为 key。

源码要点（见 spreadsheet.py）：

• 数据来源二选一：直接给 contents（内联 CSV 文本），或给 file_input（URL/data URI/本地路径的文件）；代码优先处理 file_input，通过 store_media_file + get_exec_file_path 落盘，扩展名属 .xlsx/.xls 时先做 Excel→CSV 转换；
• CSV 解析由 Python 标准库 csv 完成，所有解析选项均透传可配。

输入输出契约（含源码确认的默认值）：

方向	字段	类型	必填	默认值	说明
输入	contents	str	否	—	内联的 CSV/表格数据文本
输入	file_input	str (file)	否	—	要读取的 CSV/Excel 文件；Excel 自动转 CSV
输入	delimiter	str	否	,	CSV 分隔符
输入	quotechar	str	否	"	字段引号字符
输入	escapechar	str	否	\	转义分隔符的字符
输入	has_header	bool	否	true	是否含表头行
输入	skip_rows	int	否	0	从文件开头跳过的行数
输入	strip	bool	否	true	是否去除取值两侧空白
输入	skip_columns	List[str]	否	[]	从输出中排除的列
输入	produce_singular_result	bool	否	false	为 true 时仅逐行产出 row（可能较慢）；为 false 时产出整批 rows
输出	row	Dict[str, str]	—	—	单行数据（逐行产出模式）
输出	rows	List[Dict[str, str]]	—	—	全部数据（整批产出模式）
输出	error	str	—	—	失败时的错误信息

测试用例展示了两种产出形态：对 "a, b, c\n1,2,3\n4,5,6"，整批模式输出 [{"a": "1", "b": "2", "c": "3"}, {"a": "4", "b": "5", "c": "6"}]，逐行模式则把每个 row 分别 yield。

典型场景：导入产品目录/联系人/库存等导出的表格；解析其他系统的 CSV 报表；按行批量处理邮箱地址、用户记录或配置数据。

---

七、SQL Query：安全地直连三类数据库执行查询

定位：通过 SQLAlchemy 连接 PostgreSQL、MySQL 或 MSSQL 执行 SQL；默认只读以保证安全，需写入时显式关闭 read_only。

输入输出契约（含源码确认的默认值与范围约束，见 sql_query_block.py）：

方向	字段	类型	必填	默认值	说明
输入	database_type	"postgres" | "mysql" | "mssql"	否	postgres	目标数据库引擎
输入	host	str (password)	是	—	主机名/IP；按密钥字段处理避免泄露基础设施信息；私网/内网 IP 会被拦截
输入	port	int	否	—	端口（留空用默认值：PostgreSQL 5432、MySQL 3306、MSSQL 1433），合法范围 1–65535
输入	database	str	是	—	库名
输入	query	str	是	—	要执行的 SQL
输入	credentials	数据库用户名/密码凭据	是	—	由凭据字段选择器提供
输入	read_only	bool	否	true	开启（默认）只允许 SELECT 且会话置为只读；关闭后可执行写操作
输入	timeout	int	否	30	查询超时秒数（上限 120）
输入	max_rows	int	否	1000	返回行数上限（上限 10000）
输出	results	List[Dict[str, Any]]	—	—	查询结果（行字典列表）
输出	columns	List[str]	—	—	结果列名
输出	row_count	int	—	—	返回行数
输出	truncated	bool	—	—	结果被 max_rows 截断时为 true，提示库内仍有更多数据
输出	affected_rows	int	—	—	写查询（INSERT/UPDATE/DELETE）影响的行数
输出	error	str	—	—	查询失败时的错误信息

安全机制（源码与文档共同确认）：

• 单语句校验：借助 sqlparse 校验查询只含单条语句，防范多语句拼接型 SQL 注入（_validate_single_statement）；
• SSRF 防护：host 经 resolve_and_check_blocked 解析与阻断校验，私网/内网地址会被拒绝；
• 只读兜底：默认下会话置为只读且事务始终回滚，read_only 关闭后才允许 INSERT/UPDATE/DELETE/CREATE/DROP 等写操作；
• 相关辅助逻辑集中在 sql_query_helpers.py，并有完整单测覆盖 sql_query_block_test.py。

典型场景：拉取活跃用户数/营收/漏斗数据做分析看板；用写查询管理数据；建表改索引等 schema 管理；在同一工作流内连接多种数据库做跨源汇总。

---

八、Screenshot Web Page：网页截图

定位：调用 ScreenshotOne API 对指定 URL 截图，输出图片文件数据。

源码确认的默认参数（见 screenshotone.py）：

方向	字段	类型	必填	默认值	说明
输入	credentials	ScreenshotOne API Key 凭据	是	—	平台凭据选择器提供
输入	url	str	是	—	待截图网址
输入	viewport_width	int	否	1920	视口宽（像素）
输入	viewport_height	int	否	1080	视口高（像素）
输入	full_page	bool	否	false	是否截取整页长图
输入	format	"png" | "jpeg" | "webp"	否	png	输出图片格式
输入	block_ads	bool	否	true	是否拦截广告
输入	block_cookie_banners	bool	否	true	是否拦截 Cookie 横幅
输入	block_chats	bool	否	true	是否拦截聊天小组件
输入	cache	bool	否	false	是否启用缓存（对重复抓同页更高效）
输出	image	str (file)	—	—	截图图片数据
输出	error	str	—	—	失败时的错误信息

典型场景：为文档/报告/归档捕获页面截图做可视化资料；定期抓取竞品站点跟踪设计与内容变化；抓取页面渲染结果做视觉回归测试。

---

九、组合实战建议

把上面的 Block 串成典型管线时，可参考下列几种成熟组合（均基于前述 Block 的输入输出契约）：

1. 批量处理 + 断点续批：Create List 设 max_size 分批发散 → 各批处理完用 Persist Information 记录最后进度 → 下次运行用 Retrieve Information 取回并跳过已完成部分（注意 read→write 竞态）；
2. 外部数据入库：File Read / Read Spreadsheet 读出原始内容 → JSON Decoder 或逐行字典 → SQL Query（默认只读）校验后关闭 read_only 执行批量 INSERT；
3. API 对接闭环：Create Dictionary 构造请求体 → JSON Encoder 序列化 → 请求完成后 JSON Decoder 解析响应 → FindInDictionaryBlock 提取关键字段。

---

十、结语与扩展阅读

Data 类 Block 是 AutoGPT Platform 工作流中把"AI 决策"与"真实数据"衔接起来的底座：字典/列表负责组织内存数据，File Read 与 Read Spreadsheet 负责外部数据输入，JSON 编解码负责与 API 生态互通，持久化对负责跨运行记忆，SQL Query 则把数据库能力安全地带入可视化编排。每个 Block 都在源码中以 id + Input/Output Schema + test_input/test_output 的自描述结构注册（backend/blocks/_base.py），既可在编辑器中直接拖拽，也方便二次开发。

想进一步了解 Block 开发范式与整站编排能力，可继续阅读：

• 数据操作类更多玩法：data_manipulation.py（字典增删改查、列表拼接/展平/去重等）；
• JSON 工具与编码细节：backend/util/json 与 json_blocks.py；
• 持久化存储服务端实现：见 get_execution_kv_data/set_execution_kv_data 所在的数据访问层，以及相关数据库迁移（如 20250702224504_add_node_exec_kv_data）；
• 面向贡献者的 Block 开发指南：docs/platform/new_blocks.md 与 docs/platform/agent-blocks.md。