工作流数据模型:DAG 是怎么被描述的

30 秒导读: PySpur 里,"一个 agent / 一条工作流"不是一段代码,而是一份数据——一张节点表加一张连线表,构成有向无环图(DAG,Directed Acyclic Graph)。这份数据由几个 Pydantic 模型定义,并被一组校验规则守着底线(必须恰好一个入口、节点 id 不能重复……)。读懂这份数据结构,就读懂了整个系统的地基:后面的执行引擎(02)、节点体系(03)全都围着它转。

本章只讲数据模型与它的校验,不讲怎么执行(留给 02-executor.md),也不讲单个节点内部干了什么(留给 03-node-system.md)。全部内容锚定在一个文件:backend/pyspur/schemas/workflow_schemas.py。

1. 这是什么(零基础也能懂)

一句话定义: 一条 PySpur 工作流 = 一个 DAG,用 JSON / Pydantic 对象描述:有哪些节点(每步做什么)、节点之间怎么连线(数据从谁流到谁)。

打个比方: 把工作流想成一张地铁线路图。

节点(node) = 一个个站点(输入站、某个处理站、输出站)。
连线(link) = 站与站之间的轨道,规定了"车"(数据)从哪站开往哪站。
校验(validator) = 通车前的安检:必须有且只有一个始发站、站名不能重名、终点站最多一个……不合规就不让这张图跑起来。

关键心智模型:这是纯声明式的数据,不是命令式的代码。 你不写"先调 A 再调 B",你只描述这张图长什么样;真正"按图施工、决定先跑谁"的是执行引擎。所以这一层没有任何执行逻辑,只有"结构 + 约束"。

用起来什么样: 一份最小工作流,拆开看就是 nodes(节点数组)+ links(连线数组)两块。下面这段是从源码 __main__ 里搬来的真实例子(backend/pyspur/execution/workflow_executor.py:__main__),做的是"输入一个问题 → BestOfN 节点生成回答 → 输出":

{
  "nodes": [
    { "id": "input_node",  "node_type": "InputNode",   "config": { "output_schema": { "question": "string" } } },
    { "id": "bon_node",    "node_type": "BestOfNNode",  "config": { "samples": 1, "llm_info": { "model": "gpt-4o", ... }, ... } },
    { "id": "output_node", "node_type": "OutputNode",   "config": { "output_map": { "response": "bon_node.response" }, ... } }
  ],
  "links": [
    { "source_id": "input_node", "target_id": "bon_node" },
    { "source_id": "bon_node",   "target_id": "output_node" }
  ],
  "test_inputs": [ { "id": 1733466671014, "question": "<p>Is altruism inherently selfish?</p>" } ]
}

这张图对应的结构一眼就能读出来:

 input_node ──▶ bon_node ──▶ output_node
 (InputNode)   (BestOfNNode) (OutputNode)

三个节点、两条线,一条直链。这就是 PySpur 眼中"一个工作流"最朴素的样子。

2. 顶层全景(几个模型怎么拼)

整份数据模型由 4 个核心 Pydantic 类拼成,外加一个枚举 SpurType。它们的包含关系是:

WorkflowDefinitionSchema            "整张图"     schemas/workflow_schemas.py:98
  ├── spur_type: SpurType           工作流品类(workflow / chatbot / agent)
  ├── nodes:  List[WorkflowNodeSchema]     一张节点表    :40
  │            └── coordinates: WorkflowNodeCoordinatesSchema  :26  (画布 x/y)
  │            └── dimensions:  WorkflowNodeDimensionsSchema   :33  (画布宽高)
  │            └── subworkflow: WorkflowDefinitionSchema (可选,递归嵌套子图)
  ├── links:  List[WorkflowLinkSchema]     一张连线表    :86
  └── test_inputs: List[Dict]              画布上"跑一下"用的样例输入

各部件一句话职责:

部件	干什么	位置(`schemas/workflow_schemas.py`)
`SpurType`	枚举:这张图是普通工作流、聊天机器人、还是自治 agent	`:12`
`WorkflowDefinitionSchema`	整张 DAG——持有 nodes / links / test_inputs / spur_type,并挂着全图级校验	`:98`
`WorkflowNodeSchema`	一个节点——id / 标题 / 类型 / config / 画布坐标,可选子工作流	`:40`
`WorkflowLinkSchema`	一条连线——source→target,加可选的 handle(具体连哪个输出/输入口)	`:86`
`WorkflowNodeCoordinatesSchema` / `...Dimensions...`	节点在可视化画布上的位置与大小(纯 UI 用)	`:26` / `:33`

术语:PySpur 把"工作流"这个东西统称 spur。SpurType(:12)区分三种 spur:WORKFLOW(标准工作流)、CHATBOT(带聊天式 IO + 会话管理的工作流)、AGENT(会自己调工具的自治 agent 节点,也有聊天式 IO)。品类不同,校验规则会不同(见 §5)。

主线走一遍(高层): 前端画布 / API 递上来一份 JSON → WorkflowDefinitionSchema.model_validate(...) 把它解析成对象,同时跑完所有校验 → 得到一个结构合法的 DAG 对象 → 交给执行引擎去跑。本章负责到"结构合法的 DAG 对象"为止。

3. 三个核心模型逐个看

3.1 WorkflowNodeSchema —— 一个节点

它描述什么: 图里的一步。字段见 schemas/workflow_schemas.py:40-61:

字段	含义	备注
`id`	节点在图里的唯一标识	全图必须唯一(§4.1)
`title`	显示名	空则回落到 id(见下)
`parent_id`	父节点 id	用于嵌套/分组;顶层节点为 `None`
`node_type`	节点类型名	字符串,如 `"InputNode"` / `"BestOfNNode"`;执行时靠它查工厂(03)
`config`	该节点的配置字典	动态输出 schema、`llm_info` 等都塞这里
`coordinates` / `dimensions`	画布位置 / 尺寸	纯 UI
`subworkflow`	可选的子工作流	类型又是 `WorkflowDefinitionSchema`,递归嵌套

注意 subworkflow 的类型是 Optional["WorkflowDefinitionSchema"](:61)——一个节点内部可以再挂一整张图。所以数据模型天然是可递归的:图里能套图。

这个节点模型挂了两个 model_validator(mode="after")(对象构造完之后跑的整体校验):

① default_title_to_id(:63-67)——标题兜底。 如果 title 去空格后是空串,就把 title 设成 id。一句话:没起名的节点,用它的 id 当显示名,保证画布上不会出现空标题。

② prefix_model_name_with_provider(:69-83)——给模型名补 provider 前缀。 这是一段历史兼容逻辑,只对 node_type 为 "SingleLLMCallNode" 或 "BestOfNNode" 的节点生效。它从 config["llm_info"]["model"] 取模型名,按前缀改写:

原始模型名前缀	改写为
`gpt` / `chatgpt` / `o1`	`openai/<原名>`(例:`gpt-4o` → `openai/gpt-4o`)
`claude`	`anthropic/<原名>`(例:`claude-3-5-sonnet` → `anthropic/claude-3-5-sonnet`)

真实片段(schemas/workflow_schemas.py:72-82):

if self.node_type in ("SingleLLMCallNode", "BestOfNNode"):
    llm_info = self.config.get("llm_info")
    assert llm_info is not None
    if (llm_info["model"].startswith("gpt")
            or llm_info["model"].startswith("chatgpt")
            or llm_info["model"].startswith("o1")):
        llm_info["model"] = f"openai/{llm_info['model']}"
    if llm_info["model"].startswith("claude"):
        llm_info["model"] = f"anthropic/{llm_info['model']}"

为什么要这么做: PySpur 下游用统一的 provider/model 命名来路由到不同厂商。早期存下来的 spur 只写了裸模型名(gpt-4o),所以每次加载时都在这里"就地升级"成带前缀的格式,老数据也能跑。

坑: 这里对 llm_info 用了 assert ... is not None(:74)。也就是说,一个 SingleLLMCallNode / BestOfNNode 的 config 必须带 llm_info,否则加载阶段就会 AssertionError。这是这两类节点的隐性硬约束。

3.2 WorkflowLinkSchema —— 一条连线

它描述什么: 数据从哪个节点流到哪个节点。字段见 schemas/workflow_schemas.py:86-95:

字段	含义
`source_id`	源节点 id(数据从这里出)
`target_id`	目标节点 id(数据流到这里)
`source_handle`	源节点的输出口名(可选);一个节点可能有多个命名输出
`target_handle`	目标节点的输入口名(可选)

直觉: 只有 source_id + target_id 时,就是"整包数据从 A 流到 B"。当一个节点有多个输出/输入口时,才需要 handle 来指名道姓连哪个口。上面 §1 的最小例子里两条 link 都没写 handle,就是最简单的"整包直连"。target_handle 在 RouterNode 场景会被特殊规范化(见 §5)。

3.3 WorkflowDefinitionSchema —— 整张图

它描述什么: 见 schemas/workflow_schemas.py:98-104,就 4 个字段:

字段	含义
`nodes`	节点表 `List[WorkflowNodeSchema]`
`links`	连线表 `List[WorkflowLinkSchema]`
`test_inputs`	画布上点"运行"时喂的样例输入,`List[Dict]`,默认空
`spur_type`	品类,默认 `SpurType.WORKFLOW`

它自己不含执行逻辑;它的价值全在挂着的一堆校验器——这些校验才是"什么样的图算合法"的定义。下面两节专门讲校验。

4. 结构约束:三条 `field_validator`

这三条挂在 nodes 字段上(@field_validator("nodes")),负责守住 DAG 的基本结构底线。它们是"这张图能不能成立"的门槛。

4.1 节点 id 必须唯一 —— `nodes_must_have_unique_ids`(`:106-112`)

把所有 node.id 收集起来,比较列表长度和去重后集合的长度,不等就报错 "Node IDs must be unique."。为什么重要: links 靠 id 指认节点、执行时也靠 id 索引输出;id 撞车整个图就语义崩了。

4.2 必须恰好一个入口节点 —— `must_have_one_and_only_one_input_node`(`:114-122`)

筛出所有 node_type == "InputNode" 且 parent_id is None(即顶层)的节点,数量不等于 1 就报错 "Workflow must have exactly one input node."。

一句话:每张图有且只有一个始发站。 注意 parent_id is None 的条件——子工作流里的 InputNode 不算,只数顶层的。

4.3 至多一个输出节点 —— `must_have_at_most_one_output_node`(`:124-132`)

同理筛顶层 OutputNode,数量 > 1 就报错。注意措辞是 at most:输出节点可以是 0 个或 1 个,但不能多个。入口是"恰好一个",出口是"至多一个"——这个不对称值得记住。

三条约束一览:

约束	规则	报错文案
id 唯一	所有 `node.id` 不重复	Node IDs must be unique.
唯一入口	顶层 `InputNode` 恰好 1 个	Workflow must have exactly one input node.
至多一出口	顶层 `OutputNode` ≤ 1 个	Workflow must have at most one output node.

小注(inferred): 这三条挂了 @classmethod + @field_validator 两个装饰器,顺序上 @classmethod 在外层(:106-107 等)。Pydantic v2 的 field_validator 本会自身处理 classmethod 绑定,这里的显式 @classmethod 属于冗余写法,但不影响校验语义。

5. 语义约束:三条 `model_validator`(全图级)

field_validator 只看单个字段;要看"节点+连线+品类"的组合是否合理,就得用 model_validator(mode="after")——它在整张图构造完之后跑,能同时访问 nodes / links / spur_type。这里有三条。

5.1 RouterNode 的 target_handle 规范化 —— `validate_router_node_links`(`:134-154`)

这条不只是校验,还会就地改数据。它遍历所有 link,只要 link 的源节点是 RouterNode,就把该 link 的 target_handle 规整成统一格式 <source_id>.<handle>。

为什么要: RouterNode(路由/分支节点)会把数据分发到不同分支,每个分支对应一个命名 handle。为了让下游能无歧义地识别"这条线来自哪个 router 的哪个分支",target_handle 统一带上 源节点id. 前缀。

规整逻辑(:145-152)分三步:

target_handle = link.target_handle or link.source_id   # 缺省用 source_id 兜底
if "." in target_handle:  target_handle = 点号后半段     # 去掉可能已有的旧前缀
if not startswith(f"{source_id}."):                      # 没带正确前缀就补上
    link.target_handle = f"{source_id}.{target_handle}"

一句话:保证每条从 RouterNode 出来的连线,target_handle 一定形如 router1.branchA。

5.2 & 5.3 chatbot 品类的 IO 校验

当 spur_type == SpurType.CHATBOT 时,额外强制入口/出口节点带上聊天必需的字段。普通 WORKFLOW 不受这两条约束。

输入节点校验 —— validate_chatbot_input_node(:156-199): 找到顶层 InputNode,把它 config 里的 output_json_schema 解析成模型,检查字段:

必须有 user_message,且类型是 str(:180-183)
必须有 session_id,且类型是 str(:185-188)

缺哪个就把它加进 missing_fields,最后统一报错列出缺失项。JSON 解析失败则报 "Invalid JSON schema in input node output_json_schema"(:195-198)。

输出节点校验 —— validate_chatbot_output_node(:201-236): 对称地检查顶层 OutputNode 的 schema:

必须有 assistant_message,且类型是 str(:221-224)

为什么: chatbot 这种 spur 要接会话系统——进来得知道"用户说了什么(user_message)、这是哪个会话(session_id)",出去得给出"助手回了什么(assistant_message)"。这三个字段是聊天式 IO 的契约,靠校验硬性钉死。

chatbot IO 契约一览:

节点	必需字段	类型	校验器
顶层 InputNode	`user_message`	`str`	`validate_chatbot_input_node` `:156`
顶层 InputNode	`session_id`	`str`	同上
顶层 OutputNode	`assistant_message`	`str`	`validate_chatbot_output_node` `:201`

6. 完整例子:从 JSON 到合法 DAG

把前面所有点串起来看那份真实的 __main__ 示例(backend/pyspur/execution/workflow_executor.py,文件末尾)。它调用 WorkflowDefinitionSchema.model_validate({...}),喂进去的字典就是本章讲的全部结构:

        input_node                bon_node                    output_node
       ┌───────────┐            ┌───────────────┐           ┌────────────┐
   ──▶ │ InputNode │ ──link──▶  │  BestOfNNode  │ ──link──▶ │ OutputNode │ ──▶
       │ output_   │            │ llm_info:     │           │ output_map:│
       │  schema:  │            │  gpt-4o →     │           │  response ←│
       │  question │            │  openai/gpt-4o│           │  bon_node. │
       └───────────┘            └───────────────┘           │  response  │
                                                            └────────────┘

这一次 model_validate 里发生了什么(全是本章的校验,没有任何执行):

三个 node 各自构造 → 每个都跑 default_title_to_id:三个 title 都是空串 "",于是分别被回填成 input_node / bon_node / output_node。
bon_node 是 BestOfNNode → 跑 prefix_model_name_with_provider:llm_info.model 从 "gpt-4o" 被就地改写成 "openai/gpt-4o"。
全图字段校验:3 个 id 唯一 ✓;顶层 InputNode 恰好 1 个 ✓;顶层 OutputNode 1 个(≤1)✓。
全图 model 校验:没有 RouterNode,router 规范化跳过;spur_type 默认是 WORKFLOW(非 CHATBOT),两条 chatbot IO 校验跳过。
全部通过 → 得到一个结构合法的 WorkflowDefinitionSchema 对象。

到此为止,这份数据"是一张合法的 DAG"这件事已经被完全确立。接下来"谁先跑、怎么把 input_node 的输出喂给 bon_node、失败了怎么办",是 02-executor.md 的事。

7. 边界与本章不覆盖的

本层不做 DAG 无环检查(inferred)。 名字叫 DAG,但 workflow_schemas.py 里的校验只管 id 唯一、入口/出口数量、handle 格式,没有遍历 links 检测环。"无环"这个性质由谁保证(前端画布、或执行阶段的拓扑排序)不在本文件——见 02-executor.md。
不校验 link 两端 id 真的存在于 nodes 里(inferred)。 三条结构校验没有做"source_id/target_id 必须指向已存在节点"的检查。悬空连线是否被拦,得看执行阶段。
node_type 只是个字符串。 这一层不认识 "BestOfNNode" 到底是什么、config 里该有哪些键(除了那两类节点的 llm_info 断言)。字符串→真实节点类的映射、节点内部 schema,属于节点体系(03-node-system.md)。
不讲执行、并发、暂停恢复。 那些是 02 / 04 / 05。

8. 代码地图(导航索引)

全部在 backend/pyspur/schemas/workflow_schemas.py,除最后一行外。

主题	文件路径	符号名
spur 品类枚举(workflow/chatbot/agent)	`backend/pyspur/schemas/workflow_schemas.py`	`SpurType`
整张 DAG 定义	`backend/pyspur/schemas/workflow_schemas.py`	`WorkflowDefinitionSchema`
单个节点定义	`backend/pyspur/schemas/workflow_schemas.py`	`WorkflowNodeSchema`
连线定义	`backend/pyspur/schemas/workflow_schemas.py`	`WorkflowLinkSchema`
画布坐标 / 尺寸	`backend/pyspur/schemas/workflow_schemas.py`	`WorkflowNodeCoordinatesSchema` / `WorkflowNodeDimensionsSchema`
空标题回落到 id	`backend/pyspur/schemas/workflow_schemas.py`	`default_title_to_id`
模型名补 provider 前缀(gpt→openai/、claude→anthropic/)	`backend/pyspur/schemas/workflow_schemas.py`	`prefix_model_name_with_provider`
id 唯一	`backend/pyspur/schemas/workflow_schemas.py`	`nodes_must_have_unique_ids`
恰好一个入口节点	`backend/pyspur/schemas/workflow_schemas.py`	`must_have_one_and_only_one_input_node`
至多一个出口节点	`backend/pyspur/schemas/workflow_schemas.py`	`must_have_at_most_one_output_node`
RouterNode target_handle 规范化	`backend/pyspur/schemas/workflow_schemas.py`	`validate_router_node_links`
chatbot 输入 IO 校验(user_message/session_id)	`backend/pyspur/schemas/workflow_schemas.py`	`validate_chatbot_input_node`
chatbot 输出 IO 校验(assistant_message)	`backend/pyspur/schemas/workflow_schemas.py`	`validate_chatbot_output_node`
最小 workflow JSON 示例(InputNode→BestOfNNode→OutputNode)	`backend/pyspur/execution/workflow_executor.py`	`__main__`

下一步: 02-executor.md — 这张合法 DAG 是怎么被拓扑调度、异步并发地跑起来的。

1. 这是什么(零基础也能懂)​

2. 顶层全景(几个模型怎么拼)​

3. 三个核心模型逐个看​

3.1 WorkflowNodeSchema —— 一个节点​

3.2 WorkflowLinkSchema —— 一条连线​

3.3 WorkflowDefinitionSchema —— 整张图​

4. 结构约束:三条 field_validator​

4.1 节点 id 必须唯一 —— nodes_must_have_unique_ids(:106-112)​

4.2 必须恰好一个入口节点 —— must_have_one_and_only_one_input_node(:114-122)​

4.3 至多一个输出节点 —— must_have_at_most_one_output_node(:124-132)​

5. 语义约束:三条 model_validator(全图级)​

5.1 RouterNode 的 target_handle 规范化 —— validate_router_node_links(:134-154)​

5.2 & 5.3 chatbot 品类的 IO 校验​

6. 完整例子:从 JSON 到合法 DAG​

7. 边界与本章不覆盖的​

8. 代码地图(导航索引)​