跳到主要内容

数据截至 (上游 commit 97b9856e49f2)

工作流数据模型: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 / o1openai/<原名>(例:gpt-4o → openai/gpt-4o)
claudeanthropic/<原名>(例: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 契约一览:

节点必需字段类型校验器
顶层 InputNodeuser_messagestrvalidate_chatbot_input_node :156
顶层 InputNodesession_idstr同上
顶层 OutputNodeassistant_messagestrvalidate_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 里发生了什么(全是本章的校验,没有任何执行):

  1. 三个 node 各自构造 → 每个都跑 default_title_to_id:三个 title 都是空串 "",于是分别被回填成 input_node / bon_node / output_node。
  2. bon_node 是 BestOfNNode → 跑 prefix_model_name_with_provider:llm_info.model 从 "gpt-4o" 被就地改写成 "openai/gpt-4o"。
  3. 全图字段校验:3 个 id 唯一 ✓;顶层 InputNode 恰好 1 个 ✓;顶层 OutputNode 1 个(≤1)✓。
  4. 全图 model 校验:没有 RouterNode,router 规范化跳过;spur_type 默认是 WORKFLOW(非 CHATBOT),两条 chatbot IO 校验跳过。
  5. 全部通过 → 得到一个结构合法的 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.pySpurType
整张 DAG 定义backend/pyspur/schemas/workflow_schemas.pyWorkflowDefinitionSchema
单个节点定义backend/pyspur/schemas/workflow_schemas.pyWorkflowNodeSchema
连线定义backend/pyspur/schemas/workflow_schemas.pyWorkflowLinkSchema
画布坐标 / 尺寸backend/pyspur/schemas/workflow_schemas.pyWorkflowNodeCoordinatesSchema / WorkflowNodeDimensionsSchema
空标题回落到 idbackend/pyspur/schemas/workflow_schemas.pydefault_title_to_id
模型名补 provider 前缀(gpt→openai/、claude→anthropic/)backend/pyspur/schemas/workflow_schemas.pyprefix_model_name_with_provider
id 唯一backend/pyspur/schemas/workflow_schemas.pynodes_must_have_unique_ids
恰好一个入口节点backend/pyspur/schemas/workflow_schemas.pymust_have_one_and_only_one_input_node
至多一个出口节点backend/pyspur/schemas/workflow_schemas.pymust_have_at_most_one_output_node
RouterNode target_handle 规范化backend/pyspur/schemas/workflow_schemas.pyvalidate_router_node_links
chatbot 输入 IO 校验(user_message/session_id)backend/pyspur/schemas/workflow_schemas.pyvalidate_chatbot_input_node
chatbot 输出 IO 校验(assistant_message)backend/pyspur/schemas/workflow_schemas.pyvalidate_chatbot_output_node
最小 workflow JSON 示例(InputNode→BestOfNNode→OutputNode)backend/pyspur/execution/workflow_executor.py__main__

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