数据截至 2025-07-20(上游 commit 97b9856e49f2)

节点体系:BaseNode 抽象、动态 Pydantic 模型与工厂注册

30 秒导读: PySpur 的工作流是一张节点组成的图。这一章讲一个节点是怎么被定义、实例化、执行的,以及一个新节点怎么插进系统——这是 PySpur "可扩展性"的核心。你会看到一个反复出现的主题:Pydantic 模型不是写死的,而是运行时动态生成的。

本章只讲节点框架本身。具体的 LLM / Agent 节点逻辑在 05-llm-and-agent.md;节点被谁按什么顺序调用(拓扑调度)在 02-executor.md;节点在图里怎么连线在 01-workflow-model.md。

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

一句话定义: 一个"节点"就是工作流里的一个盒子——喂给它一些输入,它跑一段逻辑,吐出一些输出。PySpur 里所有盒子都长在同一个抽象基类 BaseNode 上。

它要解决什么: 工作流平台需要很多种节点(调 LLM、跑 Python、发 Slack、读 Google Sheet……),而且要让人能轻松加新的。如果每种节点各写各的接口,执行引擎就没法统一调度。所以 PySpur 定了一套统一契约:不管你是什么节点,都必须交出三样东西。

一个节点 = 三件套 + 一个方法:

交出什么	类型	干什么用
`config_model`	`BaseNodeConfig` 子类	节点的配置(比如 LLM 节点的 prompt、模型名)
`input_model`	`BaseNodeInput` 子类	节点吃进来的输入结构(来自上游节点)
`output_model`	`BaseNodeOutput` 子类	节点吐出去的输出结构
`run()`	`async` 方法	真正的逻辑:吃 input,返回 output

一句话直觉: 把节点想成一个带类型签名的函数——config 是它被创建时定死的参数,input 是每次调用喂进来的参数,output 是返回值,run() 是函数体。特别之处在于:这些"类型签名"很多是运行时才拼出来的,因为一个节点的上游是谁、要输出什么字段,得等工作流搭好才知道。

2. 顶层全景(一个节点怎么被定义、造出来、跑起来)

下面这张图是本章的骨架。从上到下读:定义时 → 实例化时 → 执行时,三个阶段各自动态生成不同的 Pydantic 模型。

                    ┌─────────────────────────────────────────┐
   写节点的人         │  定义一个节点类(继承 BaseNode)          │
   ─────────────▶   │  声明: config_model / output_model /     │
                    │        run() 逻辑 / (可选) @register     │
                    └───────────────────┬─────────────────────┘
                                        │
             ┌──────────────────────────┼──────────────────────────┐
             │ 实例化(工厂造节点)      │                          │
             ▼                          ▼                          │
   ┌───────────────────┐     ┌────────────────────────┐           │
   │  NodeFactory       │     │  BaseNode.__init__      │           │
   │  .create_node()    │────▶│    └─ setup()           │           │
   │  按名查表          │     │       动态生成           │           │
   │  importlib 动态导入 │     │       output_model      │           │
   └───────────────────┘     └────────────────────────┘           │
                                        │                          │
             ┌──────────────────────────┘                          │
             │ 执行(执行引擎调用)                                 │
             ▼                                                     │
   ┌───────────────────────────────────────────────┐              │
   │  await node(input)   →  __call__               │              │
   │   1. 按 input 动态拼 input_model(create_model) │              │
   │   2. 校验输入                                   │              │
   │   3. await run(input) ← 你的逻辑               │              │
   │   4. 用 output_model 校验输出                   │              │
   └───────────────────────────────────────────────┘              │
                                                                   │
   ┌───────────────────────────────────────────────────────────┐  │
   │  新节点怎么进系统?两条路 ◀───────────────────────────────────┘
   │   A. 静态清单 node_types.py 的 SUPPORTED_NODE_TYPES         │
   │   B. @NodeRegistry.register 装饰器 + discover 递归自注册     │
   │      (含 @tool_function:一个 Python 文件就是一个新节点)    │
   └───────────────────────────────────────────────────────────┘

部件一句话职责:

部件	干什么	文件
`BaseNode`	所有节点的抽象基类,定义三件套契约与 `__call__`/`run`/`setup`	`backend/pyspur/nodes/base.py`
`NodeFactory`	按节点类型名找到类,`importlib` 动态导入并实例化	`backend/pyspur/nodes/factory.py`
`node_types.py`	内置节点的静态清单(支持的/废弃的),按类别分组	`backend/pyspur/nodes/node_types.py`
`NodeRegistry`	装饰器自注册 + 递归扫描发现节点 + 加载用户自定义工具	`backend/pyspur/nodes/registry.py`
`@tool_function`	把一个普通 Python 函数包成一个节点类	`backend/pyspur/nodes/decorator.py`

3. 核心原理(逐个机制,由浅入深)

3.1 三件套契约:config / input / output

先看契约本身。BaseNode 声明了三个类型字段,任何节点都必须填上(base.py:79-81,BaseNode 类):

config_model: Type[BaseNodeConfig]   # 配置模型
output_model: Type[BaseNodeOutput]   # 输出模型
input_model: Type[BaseNodeInput]     # 输入模型

这三个基类各有分工,注释里说得很清楚:

BaseNodeConfig(base.py:22-44)——是唯一默认就带字段的。它自带 output_schema、output_json_schema、has_fixed_output 三个字段,还开了 extra: "allow" 允许附加任意字段。这几个字段正是"动态生成输出模型"的开关(下一节讲)。
BaseNodeOutput(base.py:47-53)——空基类。每种节点定义自己的输出模型继承它。
BaseNodeInput(base.py:56-63)——空基类,但注释点破了关键设计:"每个节点的输入模型会基于它的前驱节点动态创建,字段名就是节点 ID,字段类型是对应的输出模型。" 这就是为什么 input 模型不能写死。

为什么 input 要动态生成? 因为一个节点的输入取决于它在图里连了谁。同一个 MergeNode,连两个上游和连三个上游,输入结构就不一样。写死不可能,只能运行时拼。

3.2 输入模型:`call` 里按输入形态动态 `create_model`

它要解决的小问题: 执行引擎把上游结果塞给节点时,可能是两种形态——要么是一批上游节点的输出对象(BaseNodeOutput 实例),要么是一批原始值(字符串/数字/字典)。节点得把这堆东西拼成一个带类型的 Pydantic 模型再校验。

思路: 在 __call__(base.py:175-243)入口,先判断输入是不是 dict,再看 dict 里装的是什么,分两条路动态建模型:

                    input 进来
                        │
              是 dict 吗?──否──▶ 已是 BaseNodeInput,直接用
                        │是
        ┌───────────────┴────────────────┐
   全是 BaseNodeOutput/          否则(原始值:
   BaseNodeInput 实例?           str/int/dict...)
        │                              │
        ▼                              ▼
  create_composite_          就地 create_model:
  model_instance():          字段名=key,
  字段名=key,               字段类型=type(value)
  字段类型=各自的类           (base.py:211-220)
  (base.py:201-208)

第一条路(上游是节点输出)——调 create_composite_model_instance(base.py:150-173),用 create_model 把每个上游的输出类拼成一个复合输入模型,字段名保留原来的 key(通常是上游节点名):

# base.py:164-173,create_composite_model_instance 核心
return create_model(
    model_name,
    **{key: (instance.__class__, ...) for key, instance in instances.items()},
    __base__=BaseNodeInput,
    ...
)

这段把"上游名 → 上游输出模型类"逐个变成字段,组装出这个节点这一次调用的 input_model。

第二条路(上游是原始值)——直接在 __call__ 里 create_model,字段类型用 type(value) 现推(base.py:211-220)。

拼好模型后校验、存进 self._input,再 await self.run(input)(base.py:223-224)。跑完还要用 output_model 把结果校验一遍(base.py:226-243),校验失败会打印结果并抛 ValueError,方便 debug。

关键细节: input_model 是每次调用都可能被重写的实例属性(self.input_model = ...)。这跟"类上声明的类型注解"不同——注解是占位,真正生效的是运行时拼出来的这个。

3.3 输出模型:`setup()` 按配置动态生成

它要解决的小问题: 有些节点的输出字段是用户在配置里定的(比如"结构化输出"节点让你自己写 JSON schema)。这种输出模型没法在代码里写死,得从配置里的 schema 现生成。

开关在配置里: BaseNodeConfig 的 has_fixed_output(默认 False)和 output_json_schema(默认一个 {"output": string} 的 schema)。当 has_fixed_output=True,setup() 就从 output_json_schema 生成输出模型(base.py:104-114):

# base.py:109-114,BaseNode.setup()
if self._config.has_fixed_output:
    schema = json.loads(self._config.output_json_schema)
    model = pydantic_utils.json_schema_to_model(
        schema, model_class_name=self.name, base_class=BaseNodeOutput
    )
    self.output_model = model

这里 json_schema_to_model(utils/pydantic_utils.py:27-56)把一段 JSON schema 递归翻译成 Pydantic 模型类:读 properties 生成字段、读 $defs 生成嵌套子模型,最后 create_model 出一个继承 BaseNodeOutput 的类。

setup() 在 __init__ 末尾被调用(base.py:102),所以节点一造出来,输出模型就位。

对照: create_output_model_class(base.py:116-148)是另一条更简单的路——它吃 output_schema(Dict[str,str],像 {"answer": "string"}),用一张类型映射表把 "string"→str、"integer"→int 等转好再 create_model。两种方式都指向同一目标:输出结构由配置决定,运行时建模型。

3.4 节点即工具:`function_schema` 与 `call_as_tool`

它要解决的小问题: Agent 要能把别的节点当"工具"调用(function calling)。那就得把节点的配置翻译成 LLM 认识的函数 schema,并能用一批参数直接执行。

两个成员配合:

function_schema(base.py:263-316,@property)——把 config_model 的 JSON schema 转成 OpenAI 风格的 {"type":"function","function":{name, description, parameters}}。描述取自节点类的 docstring。如果 has_fixed_output=True,它会把 has_fixed_output 和 output_json_schema 这两个内部字段从参数里剔掉(base.py:283-294)——因为那是框架细节,不该暴露给 LLM。
call_as_tool(base.py:318-333)——吃一个 arguments 字典,用它 model_validate 出配置、新建一个节点实例、再用同一批参数校验成输入模型,直接 await node_instance.run(input_model):

# base.py:326-333,call_as_tool 核心
config_model = self.config_model.model_validate(arguments)
node_instance = self.__class__(self.name, config_model, self.context)
input_model = self.input_model.model_validate(arguments)
node_instance._input = input_model
return await node_instance.run(input_model)

这条链让"任何节点"天然就是"一个可被 Agent 调用的工具"。Agent 节点怎么用它循环调工具,见 05-llm-and-agent.md。

附带一提 VisualTag(base.py:13-19)——节点在 UI 上的小标签(缩写 + 十六进制颜色)。get_default_visual_tag(base.py:349-379)用节点名首字母拼缩写,并用类名的 md5 从一组固定色里挑颜色(确定性:同名节点每次同色)。纯展示,不影响执行。

4. 深入实现:新节点怎么进系统

到这里,"一个节点如何被定义/执行"讲完了。剩下的问题是:系统怎么知道有哪些节点、怎么按名字把类找出来。有两条并行的路。

4.1 路 A:静态清单 `SUPPORTED_NODE_TYPES`

内置节点写在 node_types.py 的 SUPPORTED_NODE_TYPES(node_types.py:9-192)——一个按类别分组的字典,每项给出 node_type_name / module / class_name 三元组:

# node_types.py:22-32(节选 AI 类别)
"AI": [
    {"node_type_name": "SingleLLMCallNode",
     "module": ".nodes.llm.single_llm_call",
     "class_name": "SingleLLMCallNode"},
    {"node_type_name": "AgentNode",
     "module": ".nodes.llm.agent",
     "class_name": "AgentNode"},
    ...
]

类别一览(节选):

类别	代表节点
Input/Output	`InputNode`、`OutputNode`
AI	`SingleLLMCallNode`、`AgentNode`、`RetrieverNode`、`BestOfNNode`
Code Execution	`PythonFuncNode`
Logic	`RouterNode`、`CoalesceNode`、`MergeNode`、`StaticValueNode`
Experimental	`ForLoopNode`
Integrations	一堆 GitHub / Reddit / Slack / Google Sheets 节点
Search / Tools	`ExaSearchNode`、`SendEmailNode`

另有 DEPRECATED_NODE_TYPES(node_types.py:194-240)列废弃节点(如 MCTSNode、TreeOfThoughtsNode)——它们仍被 is_valid_node_type 认作合法(node_types.py:262-264),这样老工作流不会因为节点被弃用就打不开,只是不在新建面板里推荐。

4.2 工厂:`NodeFactory.create_node` 按名查表 + `importlib` 动态导入

create_node(factory.py:66-105)是"从类型名到实例"的总入口。流程:

create_node(node_name, node_type_name, config)
        │
        ▼
  is_valid_node_type? ──否──▶ raise ValueError
        │是
        ▼
  在 SUPPORTED_NODE_TYPES 里找 module+class  (factory.py:79-86)
        │找不到
        ▼
  在 NodeRegistry 里找 module+class          (factory.py:89-98)
        │仍找不到 ──▶ raise ValueError
        ▼
  importlib.import_module(module, package="pyspur")   (factory.py:103)
  getattr(module, class_name)                          (factory.py:104)
  node_class(name, config=config_model(**config))      (factory.py:105)

关键就一行 importlib 动态导入——内置节点不在启动时全部加载,而是用到哪个才 import 哪个(factory.py:103-105)。注意它优先查静态清单,查不到才查注册表(factory.py:78-98),两套注册机制在这里汇合。

工厂顶部注释还定了命名约定(factory.py:18-24):类叫 <X>Node、配置叫 <X>NodeConfig、输入/输出同理,一个模块只放一个节点类。

4.3 路 B:`@NodeRegistry.register` 装饰器 + 递归自注册

它要解决的小问题: 每加一个节点都去改 node_types.py 很烦。装饰器让节点自己声明自己。

NodeRegistry.register(registry.py:26-116)是个带参装饰器,给节点类打上类别/显示名/logo 等元数据,并把它的 module + class_name 记进 _nodes 字典。它还支持 position 参数控制在类别里的排序:整数=绝对位置、"after:X" / "before:X"=相对某节点(registry.py:84-106)。

光有装饰器还不够——得有人去触发这些类被 import(装饰器只在类定义被执行时才生效)。这就是 discover_nodes(registry.py:149-176):

# registry.py:159-171(节选)
package = importlib.import_module(package_path)          # 默认 "pyspur.nodes"
base_path = Path(package.__file__).resolve().parent
cls._discover_in_directory(base_path, package_path)     # 递归 import 每个 .py
cls.discover_tool_functions()                            # 顺带加载用户工具

_discover_in_directory(registry.py:126-147)递归遍历目录,对每个非下划线开头的 .py 文件 importlib.import_module。import 的副作用就是执行文件里的 @register,节点于是"自注册"。子目录递归下去。

一句话精华: discover 不主动"注册"任何东西,它只负责把文件 import 进来,让文件里的装饰器自己跑。注释写得很直白:"Import module but don't register nodes - they'll self-register if decorated"(registry.py:139)。

4.4 用户自定义工具:`discover_tool_functions` 扫 `PROJECT_ROOT/tools`

discover_tool_functions(registry.py:181-276)专门加载用户放在项目里的工具——这正是 README 所说 "🐍 Python-Based: Add new nodes by creating a single Python file"(README.md:68)的落地。

流程:

从环境变量读 PROJECT_ROOT,拼出 PROJECT_ROOT/tools 目录(registry.py:190-199)。没设或不存在就记 error 返回。
递归扫描,但只进 Python 包目录(必须有 __init__.py,见 _is_package_dir,registry.py:204-206)。
对每个 .py import,遍历模块属性,挑出合法的 tool function(_is_valid_tool_function,registry.py:231-239:是 ToolFunction、其 node_class 继承 FunctionToolNode、且带齐 display_name/config_model/input_model/output_model)。
把它们注册到 "Custom Tools" 类别(registry.py:208-229)。

于是:丢一个带 @tool_function 的 .py 到 tools/ 里,它就成了 UI 里可用的新节点——不用碰框架代码。

4.5 `@tool_function`:把普通函数包成一个节点类

最后看这块魔法怎么施展。tool_function(decorator.py:110-315)是个装饰器工厂,读你函数的签名和类型注解,自动生成 config/input/output 三件套,并造出一个 FunctionToolNode 子类。

它从函数里读出什么、生成什么:

从函数读	生成什么	代码位置
函数名	节点名 / 显示名(下划线转标题)	`decorator.py:139-140`
docstring	节点描述(docstring)	`decorator.py:141`,`266`
返回类型注解	`output_model`	`decorator.py:161-191`
每个参数 + 默认值	`config_model` 的字段	`decorator.py:203-246`
——	`input_model`(空壳,输入由图上位置决定)	`decorator.py:151-159`

输出模型分三种情况生成(decorator.py:163-191):返回类型若已是 BaseNodeOutput 就直接用;若是别的 Pydantic 模型,用 json_schema_to_model 转;若是原始类型(str/dict 等),就造一个只有单字段 output 的模型。

配置模型把每个函数参数变成一个字段,并塞进 output_json_schema 和 has_fixed_output(decorator.py:235-246)——这样它接回 3.3 节 setup() 的动态输出逻辑。

执行时——FunctionToolNode.run(decorator.py:63-88)从 config 里取出各参数值,关键一步:用 jinja2 渲染字符串参数,把 input 注入模板:

# decorator.py:73-79,run() 核心
for param_name, param_value in kwargs.items():
    if isinstance(param_value, str):
        template = Template(param_value)          # jinja2
        kwargs[param_name] = template.render(input=input)
result = self._func(**kwargs)                     # 调你的原函数

这让你能在配置里写 "Hello {{ input.input_data }}",运行时被上游输入填充。异步函数也支持(decorator.py:82-83 检测 __await__)。

文件末尾的 if __name__ == "__main__" 块(decorator.py:318-427)是可直接跑的活示例,演示了普通调用、节点执行、自定义输出模型、jinja2 模板三种用法——想快速上手可以照着跑。

教学示例:一个最小的自定义工具节点

# 示意,基于 decorator.py:324-327 的真实用法
from pyspur.nodes.decorator import tool_function

@tool_function(name="add", description="两数相加", category="Math")
def add(a: int, b: int = 0) -> int:   # 参数→config 字段;返回类型→output 模型
    return a + b                       # 函数体就是 run() 要跑的逻辑
# 放进 PROJECT_ROOT/tools/ 下某个带 __init__.py 的包里,
# discover_tool_functions 会自动把它注册成 "Custom Tools" 里的 add 节点。

5. 巧妙之处(可借鉴的技术)

一切皆 create_model。 input/output/config 三件套里,凡是"结构取决于运行时信息"的,一律用 Pydantic 的 create_model 现场造类,而不是写死(base.py:132、164、211;decorator.py:151、182、235)。既拿到运行时灵活性,又保住了静态校验。
discover 只 import,不注册。 发现机制不维护一张手写清单,而是借 import 的副作用触发装饰器自注册(registry.py:139-140)。加节点=加文件,系统自己发现。
懒加载。 NodeFactory 用到哪个节点才 importlib 哪个模块(factory.py:103),启动不必导入全部节点(整个 Integrations 类别一大堆第三方依赖)。
废弃 ≠ 失效。 DEPRECATED_NODE_TYPES 仍被 is_valid_node_type 认可(node_types.py:262-264),保证老工作流向后兼容。
节点天生是工具。 function_schema + call_as_tool(base.py:263-333)让任意节点无需改造就能被 Agent 当 function 调用,统一了"节点"和"工具"两个概念。

6. 边界与局限

input_model 会被运行时重写。 类上声明的 input_model 只是占位;真正生效的是 __call__ 里现拼的那个(base.py:201、211)。读代码时别被类注解误导。
自定义工具依赖 PROJECT_ROOT + __init__.py。 环境变量没设、或 tools/ 目录不是 Python 包,discover_tool_functions 会静默记 error 返回、什么都不加载(registry.py:190-206)。
命名约定是硬约束。 工厂靠 <X>Node / <X>NodeConfig 这套约定工作(factory.py:18-24),不遵守会找不到类。
静态清单优先于注册表。 同名节点若两处都有,create_node 用静态清单里的(factory.py:78-98,先查 SUPPORTED_NODE_TYPES)。

7. 横向对比

节点体系是 PySpur 可扩展性的地基,与本组其它章的关系:

节点在图里如何被描述、如何连边 → 01-workflow-model.md
谁按什么顺序 await node(input)(拓扑调度、并发、失败传播) → 02-executor.md
暂停/恢复如何穿过节点执行 → 04-human-in-loop.md
LLM/Agent 节点如何用 function_schema/call_as_tool 做结构化输出与工具循环 → 05-llm-and-agent.md

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

主题	文件	符号
三件套契约 + `__call__`/`run`/`setup`	`backend/pyspur/nodes/base.py`	`BaseNode`
配置基类(output_schema / has_fixed_output)	`backend/pyspur/nodes/base.py`	`BaseNodeConfig`
动态拼输入模型(上游是节点输出)	`backend/pyspur/nodes/base.py`	`create_composite_model_instance`
动态生成输出模型(按配置 schema)	`backend/pyspur/nodes/base.py`	`setup`
从 output_schema 建输出模型	`backend/pyspur/nodes/base.py`	`create_output_model_class`
节点即工具	`backend/pyspur/nodes/base.py`	`function_schema` / `call_as_tool`
UI 视觉标签	`backend/pyspur/nodes/base.py`	`VisualTag` / `get_default_visual_tag`
按名查表 + importlib 动态导入实例化	`backend/pyspur/nodes/factory.py`	`NodeFactory.create_node`
内置节点静态清单 / 废弃清单	`backend/pyspur/nodes/node_types.py`	`SUPPORTED_NODE_TYPES` / `DEPRECATED_NODE_TYPES`
合法性校验	`backend/pyspur/nodes/node_types.py`	`is_valid_node_type`
装饰器自注册	`backend/pyspur/nodes/registry.py`	`NodeRegistry.register`
递归扫描发现节点	`backend/pyspur/nodes/registry.py`	`discover_nodes` / `_discover_in_directory`
加载用户自定义工具	`backend/pyspur/nodes/registry.py`	`discover_tool_functions`
把函数包成节点	`backend/pyspur/nodes/decorator.py`	`tool_function` / `FunctionToolNode`
JSON schema → Pydantic 模型	`backend/pyspur/utils/pydantic_utils.py`	`json_schema_to_model`

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

2. 顶层全景(一个节点怎么被定义、造出来、跑起来)​

3. 核心原理(逐个机制,由浅入深)​

3.1 三件套契约:config / input / output​

3.2 输入模型:__call__ 里按输入形态动态 create_model​

3.3 输出模型:setup() 按配置动态生成​

3.4 节点即工具:function_schema 与 call_as_tool​

4. 深入实现:新节点怎么进系统​

4.1 路 A:静态清单 SUPPORTED_NODE_TYPES​

4.2 工厂:NodeFactory.create_node 按名查表 + importlib 动态导入​

4.3 路 B:@NodeRegistry.register 装饰器 + 递归自注册​

4.4 用户自定义工具:discover_tool_functions 扫 PROJECT_ROOT/tools​

4.5 @tool_function:把普通函数包成一个节点类​

5. 巧妙之处(可借鉴的技术)​

6. 边界与局限​

7. 横向对比​

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