AutoGPT — 图与数据流(扁平化动态 pin)

本章讲什么: 块是点,连线是边。这一章讲 agent 的静态结构(Node/Link),以及整个项目最聪明的一招:用字符串编码的「动态 pin」,把上游输出里任意深的嵌套值,精确地喂到下游块的某个输入字段。

1. 静态结构:Node 与 Link

一张图(GraphModel,data/graph.py:385)就是一堆节点 + 一堆连线。

Node(data/graph.py:104) = 「画布上的一个块实例」。它记 block_id(指向哪个块)、input_default(用户在面板里填死的常量输入)、还有进出的连线。它的 .block 属性(data/graph.py:123)按 id 取出真正的块;块被删了就返回一个占位的 _UnknownBlockBase,而不是崩。
Link(data/graph.py:82) = 一条有向连线,四元组定义:

字段	含义
`source_id`	上游节点
`source_name`	上游的输出 pin 名(可能是个编码过的路径,见下)
`sink_id`	下游节点
`sink_name`	下游的输入字段名
`is_static`	这条线的数据可否被下游多次消费

1.1 起始节点:从哪开始跑

引擎需要知道「图从哪几个节点开始」。starting_nodes(data/graph.py:402)的规则很直接:

# data/graph.py:402 GraphModel.starting_nodes
outbound_nodes = {link.sink_id for link in self.links}  # 所有「被连入」的节点
input_nodes = {n.id for n in self.nodes if n.block.block_type == BlockType.INPUT}
return [n for n in self.nodes
        if n.id not in outbound_nodes or n.id in input_nodes]

白话:没有任何输入连线的节点(没人喂它)就是起点;外加显式的 Input 块(图的入参)。

2. 核心机制:扁平化动态 pin

2.1 要解决的小问题

上游块 yield "result", {"items": [{"name": "Alice"}, ...]}。下游块只想要 Alice 这个字符串,塞到它的 username 输入。怎么在「连线」这个静态结构里,表达「取 result 的 items[0].name」?

AutoGPT 的答案:不引入复杂的连线对象,而是把这条取值路径压进 source_name 字符串本身。这就是「扁平化动态 pin」。

2.2 思路/直觉:用分隔符编码路径

它给 dict / list / object 三种取值各定义一个分隔符,把路径写成一个字符串(data/dynamic_fields.py 里 DICT_SPLIT/LIST_SPLIT/OBJC_SPLIT)。于是:

上游 yield 的 pin 名:  "result"
Link 的 source_name:  "result" + DICT_SPLIT+"items" + LIST_SPLIT+"0" + OBJC_SPLIT+"name"
                       └─ 基名 ─┘ └──────────── 取值路径(编码进字符串)────────────┘

路由时,引擎拿 source_name 去剥上游实际产出的那个对象,一层层走下去,取出最终值;取不到就返回 None(那条线这次不传数据)。

2.3 真实实现:解码(出方向)

parse_execution_output(data/dynamic_fields.py:152)就是「按 link_output_selector 从 output 里挖出嵌套值」。核心循环(data/dynamic_fields.py:218):

# data/dynamic_fields.py:218 —— 沿编码路径逐级下钻
cur = data
for delim, ident in tokens:
    if delim == LIST_SPLIT:        # 列表索引
        cur = cur[int(ident)]      # 越界/类型不符 → return None
    elif delim == DICT_SPLIT:      # 字典键
        cur = cur[ident]
    elif delim == OBJC_SPLIT:      # 对象属性
        cur = getattr(cur, ident)
return cur

任何一步失配(键不存在、索引越界、类型不对)都安全地返回 None(data/dynamic_fields.py:161 docstring 明确这一点)。这让「上游输出形状和连线期望不完全一致」不会炸引擎,只是那条线不流数据。

2.4 真实实现:编码回构(入方向)

反过来,下游一个块可能有好几条线分别喂它的 config.host、config.port。引擎收到的是一堆扁平 key,要重新拼成嵌套对象再交给块。这是 merge_execution_input(data/dynamic_fields.py:299):

# 示意,非源码 —— 扁平 key 还原成嵌套
{"config~host": "db1", "config~port": 5432}
        │  merge_execution_input
        ▼
{"config": {"host": "db1", "port": 5432}}

它对每个 key 切出基名 + 路径,然后用递归的 _assign(data/dynamic_fields.py:245)按路径在结果里挖坑填值——list 不够长就补 None,dict/object 不存在就新建。

巧妙之处: 整个「任意深度的数据路由」用纯字符串编码 + 容错下钻实现,不需要在连线上挂表达式引擎或类型系统。出方向 parse 失败即 None、入方向 merge 自动建结构——两端都对脏数据宽容。这是把一个看起来要很复杂的特性,压成两个纯函数。

2.5 特殊情况:工具 pin

还有一类反过来的 pin:工具 pin。给「LLM 选工具调用」用——普通 pin 的路由信息在 link 的 source_name(selector)里;工具 pin 反过来,路由信息编码在输出 pin 名(即块产出的 emit key)里,标明要调哪个下游节点的哪个字段,而 link 的 selector 只是 "tools"。is_tool_pin(data/dynamic_fields.py:95)正是靠 tools_^_ 前缀(或恰好等于 "tools")来识别这类 pin。emit key 的具体格式见 parse_execution_output 里的工具 pin 分支注释 tools_^_{node_id}_~_{field}(data/dynamic_fields.py:195,该处用 _~_ 切出目标节点 id 与目标输入字段)。方向反了,但还是同一套字符串编码思路。

3. 静态 pin:一份输出喂多次

普通连线的数据是「一次性」的:上游产一份,下游消费一份。但有些值(比如配置、常量)要被下游反复用。这就是 Link.is_static(data/graph.py:87)。

引擎在路由时对静态线有专门处理(见 03 章 _register_next_executions):静态线的输入会被记下来,用来补全后续每一次下游执行缺失的那个输入(executor/manager.py:457 起)。所以一个标了 static_output=True 的块(如 StoreValueBlock),它的输出可以驱动下游跑很多次而不用每次都重新连。

4. 边界与局限

路径编码失配只会静默丢数据(返回 None),不会报错。好处是健壮,坏处是连线写错时排查靠日志(引擎会 log Skipped queueing ...,见 03 章)。
图本身可以有环(循环 agent 靠 ready 队列天然支持),但这也意味着没有编译期的 DAG 拓扑保证——是否终止取决于块逻辑。

5. 横向对比

和 Node-RED / ComfyUI 这类数据流编辑器同源:都是「节点 + 端口 + 连线」。AutoGPT 的特色是端口名即取值路径,省去了在连线上挂转换节点。
和 LangGraph 的 state graph 比:LangGraph 用一个共享 state 字典 + reducer;AutoGPT 是点对点逐 pin 传值,没有全局共享 state——更接近经典数据流而非状态机。

6. 代码地图

主题	文件	符号
图模型	`autogpt_platform/backend/backend/data/graph.py`	`GraphModel`
节点 / 连线	`data/graph.py`	`Node`、`NodeModel`、`Link`
起始节点判定	`data/graph.py`	`GraphModel.starting_nodes`
动态 pin 解码(出)	`data/dynamic_fields.py`	`parse_execution_output`
动态 pin 回构(入)	`data/dynamic_fields.py`	`merge_execution_input`、`_assign`
工具 pin 判定	`data/dynamic_fields.py`	`is_tool_pin`

1. 静态结构:Node 与 Link​

1.1 起始节点:从哪开始跑​

2. 核心机制:扁平化动态 pin​

2.1 要解决的小问题​

2.2 思路/直觉:用分隔符编码路径​

2.3 真实实现:解码(出方向)​

2.4 真实实现:编码回构(入方向)​

2.5 特殊情况:工具 pin​

3. 静态 pin:一份输出喂多次​

4. 边界与局限​

5. 横向对比​

6. 代码地图​