跳转至

自定义算子开发指南 (Developing Nodes)

本节为希望为 Griptape Nodes 生态构建自定义扩展算子的开发者提供了系统而详尽的工程参考文档。

面向 AI 编程助手与 Coding Agents

本开发章节的所有技术文档均提供了经过后处理的轻量 Markdown 格式，专门用于 AI 编程助手（如 Claude Code、Cursor、GitHub Copilot 等）快速检索与理解；完整的机器可读文档索引请参见 面向智能体索引。

推荐使用方式： 将上述 URL 提供给你的 AI 助手并提示： "请阅读该算子开发指南: [URL]，并指导我构建一个自定义 Griptape 算子"

算子架构与设计哲学 (Introduction)

Griptape 算子节点是高度模块化的可视化工作流组件，赋予用户通过拖拽与连线构建复杂 AI 流水线的直观体验。本节系统剖析了从极简基础概念到工业级健壮节点架构的全套设计范式。

如果你是首次接触算子开发，强烈建议从 新手入门指南 (Getting Started) 开始起步。它以平缓的学习曲线带你搭建开发环境并手把手完成第一个节点的编码。

所有自定义节点均直接或间接继承自 BaseNode 的四个核心子类之一：

  • DataNode (数据处理节点)：专注于纯粹的数据流转、变换与转换任务；
  • ControlNode (控制流节点)：具备显式 exec_in / exec_out 控制流连线，用于管控执行次序与异步长任务；
  • StartNode (流程起点节点)：工作流的唯一起始入口标记；
  • EndNode (流程终点节点)：工作流的最终收束收口标记。

核心设计概念 (Core Concepts)

1. 核心基类选择矩阵 (Base Classes)

  • DataNode：处理纯数据计算，无需外部控制流介入。适用于同步、即时的数据变换（如数学运算、字符串修剪、格式格式化等）——只要所有必填输入端口数据就绪，便会立即触发计算；
  • ControlNode：通过 exec_in 与 exec_out 强显式连线管理执行调度拓扑。面向网络 API 请求、重型 AI 推理或长周期轮询任务的强制基类——可通过重写 async def aprocess() 编写现代异步逻辑，或通过 AsyncResult 将阻塞操作委派给后台多线程池。如果你的节点需要调用云端接口并轮询计算结果，必须继承此类；
  • StartNode：工作流的起点触发器；
  • EndNode：工作流的终点出口标记。

2. 强类型属性参数体系 (Parameters)

通过 Parameter 核心类统一声明算子的输入端口 (Inputs)、输出端口 (Outputs) 以及内部属性面板 (Properties)。参数体系原生支持：

  • 严密的底层强类型静态与运行时校验；
  • 丰富的前端 UI 自定义控件绑定；
  • 端口连线类型与数量约束；
  • 出厂默认值兜底；
  • 特征属性控件 Traits（下拉枚举 Options、数值滑动条 Slider、动作触发按钮 Button、拾色器 ColorPicker 等）。

完整定义请参见 参数规范指南。

3. 计算核心逻辑方法 (Process Method)

同步节点的计算逻辑封装在 process() 方法内。将计算产出的结果赋值给字典 self.parameter_output_values 即可对外广播输出。对于长耗时的异步网络操作，请重写 async def aprocess() 异步协程方法——详见 生命周期与执行流。

4. 节点解析状态机 (Node States)

  • UNRESOLVED (未就绪/未解析)：初始冷态，或输入数据变动后的待计算状态；
  • RESOLVING (计算中/解析中)：当前正处于 process() 逻辑执行流水线中；
  • RESOLVED (已完成/已解析)：计算平稳结束，输出端口数据已广播生效。

5. 拓扑连线生命周期响应 (Connections)

引擎提供了精细的生命周期回调函数（如 validate_connection、on_connection_created、on_connection_deleted），供节点在连线建立或拔除时自主验证类型兼容性并动态增删衍生端口。

6. 全局事件总线捕获 (Events)

利用 on_griptape_event 回调钩子，算子能够近乎实时地捕获并响应 Griptape 底层框架抛出的各类系统级与运行时事件。

算子开发标准骨架 (Basic Node Structure)

以下是一个最小化可用数据算子的标准化 Python 代码范例：

from typing import Any
from griptape_nodes.exe_types.core_types import Parameter, ParameterMode
from griptape_nodes.exe_types.node_types import DataNode


class MyNode(DataNode):
    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self.category = "Text Processing"
        self.description = "将输入的英文字符串无损转换为全大写形态"

        # 声明输入端口
        self.add_parameter(
            Parameter(
                name="input_text",
                input_types=["str"],
                type="str",
                tooltip="待转换的原始文本",
            )
        )

        # 声明输出端口
        self.add_parameter(
            Parameter(
                name="output_text",
                output_type="str",
                tooltip="转换后生成的大写文本",
            )
        )

    def process(self) -> None:
        # 提取输入端传入的数据
        val = self.get_parameter_value("input_text")
        transformed = val.upper() if val else ""

        # 将计算成果广播至输出端口
        self.parameter_output_values["output_text"] = transformed

算子开发全景文档导航 (Documentation Structure)

基于官方脚手架模板极速起步 (Start from Template)

搭建生产级算子库的最快方式是直接派生官方的标准模板仓库：

👉 Griptape Nodes Library Template 官方脚手架 (查看 Readme 说明)

该模板已预置了工业级标准样板代码、单元测试框架与规范文档拓扑。推荐流水线步骤：

  1. 从模板生成仓库：在 GitHub 上点击 “Use this template” 创建属于你的专属代码仓库；
  2. 拉取至本地开发环境：克隆仓库至本地 Griptape Nodes 的开发工作区目录；
  3. 定制项目信息：修改目录结构并重命名 pyproject.toml 中的包元数据与作者信息；
  4. 编码算子逻辑：继承 ControlNode 或 DataNode，声明标准输入输出参数并实现 process() 计算体；
  5. 配置算子库清单：在 library.json 中登记算子类名与菜单所属分组；
  6. 引擎挂载调试：在图形界面的 Library Management 中以本地模式添加并热重载测试；
  7. 验证与打包交付：在画布中拖入你的算子，搭建端到端工作流验证稳定性与性能。