"""算子开发权威指南配套示例算子 (Example Control Node)。 本文件经过精心设计并附带详尽注释： - 演示核心开发概念（算子类 + 参数 Parameter + 交互特征 Traits + process() 执行体）； - 展示如何定制现代化 UI 交互形态（显示名称、多行文本域、数值滑块等）； - 演示轻量级预检校验钩子，让配置错误在工作流运行*之前*即刻拦截并呈现。 注意：本示例直接采用底层原生的 `Parameter` API，以便完整展示内部各个活动部件。 在工业级算子库开发中，对于常规类型推荐使用预封装的高阶辅助构造器（如 `ParameterString`、`ParameterInt` 等）。 """ # ruff: noqa: INP001 from __future__ import annotations import random from datetime import datetime from typing import TYPE_CHECKING, Any from griptape_nodes.exe_types.core_types import ( Parameter, ParameterButtonGroup, ParameterMode, ParameterTypeBuiltin, ) from griptape_nodes.exe_types.node_types import DataNode from griptape_nodes.exe_types.param_types.parameter_button import ParameterButton from griptape_nodes.traits.options import Options from griptape_nodes.traits.slider import Slider if TYPE_CHECKING: from griptape_nodes.traits.button import Button, ButtonDetailsMessagePayload class ExampleNode(DataNode): def __init__(self, name: str, metadata: dict[Any, Any] | None = None) -> None: # 必须始终优先调用基类构造函数，以便引擎完成内部状态初始化。 # `name` 为用户在界面上看到的算子名称（并在日志和错误提示中显示）。 super().__init__(name, metadata) # -------- # 输入 (Inputs) / 属性 (Properties) / 输出 (Outputs) # # 一个 Parameter 端口可以自由组合以下三种核心模式： # - INPUT: 接受来自上游算子的连线输入 # - PROPERTY: 可在前端算子属性面板中直接手动输入与编辑 # - OUTPUT: 可向拓扑下游算子输出连线 # # 在真实算子中通常采用以下最佳实践： # - 对既支持连线传入又支持手动填写的配置参数，赋予 INPUT + PROPERTY # - 对由 `process()` 计算产出的衍生数据，赋予仅 OUTPUT 模式 # -------- # 自由文本输入参数（支持连线输入 + 属性面板编辑 + 输出直通） # # `type` 是该参数在引擎底层的原生主类型。内建基础类型可通过 # `ParameterTypeBuiltin..value` 获取（例如 "str", "int", "float"）。 self.add_parameter( Parameter( name="free_text", tooltip="输入任意文本内容", type=ParameterTypeBuiltin.STR.value, allowed_modes={ParameterMode.INPUT, ParameterMode.PROPERTY, ParameterMode.OUTPUT}, ui_options={ # `display_name` 决定在前端图形界面中展示的友好标签 "display_name": "自由文本 (Free Text)", # `multiline` 将输入框展开为支持换行的多行富文本域 "multiline": True, }, # 提供一个有意义的默认初始值，使得节点拖入画布后立即可用 default_value="hello from the example node", ) ) # 下拉单选参数 (Dropdown) # # 特征 Traits 用于为参数挂载动态交互行为。`Options(...)` 会在前端渲染为 # 下拉选择菜单，并将选项持久化记录至 ui_options 确保状态稳定性。 self.add_parameter( Parameter( name="dropdown", tooltip="从预设候选项中选择一项", type=ParameterTypeBuiltin.STR.value, allowed_modes={ParameterMode.INPUT, ParameterMode.PROPERTY, ParameterMode.OUTPUT}, traits={Options(choices=["yes", "no", "maybe"])}, ui_options={ "display_name": "下拉选择 (Dropdown)", }, default_value="yes", ) ) # 整数滑块参数 (Integer Slider) # # `Slider(min_val, max_val)` 在前端添加可视化拖动滑块，并提供范围约束校验。 # 对于数值型参数，还可以在 ui_options 中指定微调步长 `step`。 self.add_parameter( Parameter( name="integer_slider", tooltip="拖动滑块选择数值大小", type=ParameterTypeBuiltin.INT.value, allowed_modes={ParameterMode.INPUT, ParameterMode.PROPERTY, ParameterMode.OUTPUT}, traits={Slider(min_val=1, max_val=10)}, ui_options={ "display_name": "整数滑块 (Integer Slider)", "step": 1, }, default_value=5, ) ) # 输出文本参数 (仅作为输出端口) # # 仅限 OUTPUT 的参数通常在 `process()` 执行体内被计算赋值，不允许用户在界面手动修改。 self.add_parameter( Parameter( name="reversed_text", tooltip="对自由文本反转单词顺序后的输出结果", type=ParameterTypeBuiltin.STR.value, allowed_modes={ParameterMode.OUTPUT}, ui_options={ "display_name": "反转文本 (Reversed Text)", "multiline": True, }, ) ) # 随机浮点数输出参数 (仅 OUTPUT) self.add_parameter( Parameter( name="random_float", tooltip="介于 0 到整数滑块设定值之间的随机浮点数", type=ParameterTypeBuiltin.FLOAT.value, allowed_modes={ParameterMode.OUTPUT}, ui_options={ "display_name": "随机浮点数 (Random Float)", }, ) ) # 触发更新当前时间的交互按钮 (Button) # # 按钮组件通过 ParameterButtonGroup 与 ParameterButton 联合构建。 # `on_click` 异步事件回调直接挂载至 ParameterButton，当用户在前端点击时被触发执行。 with ParameterButtonGroup(name="datetime_button_group") as datetime_buttons: ParameterButton( name="update_datetime", label="更新日期与时间", icon="calendar", on_click=self._update_datetime, ) self.add_node_element(datetime_buttons) # 只读日期时间显示参数 # # 此参数以标准化格式呈现当前日期时间。 # 其属性仅声明为 PROPERTY（不支持 INPUT/OUTPUT 连线），并在 ui_options 中开启 readonly 锁定编辑。 self.add_parameter( Parameter( name="datetime_display", tooltip="当前记录的日期与时间", type=ParameterTypeBuiltin.STR.value, allowed_modes={ParameterMode.PROPERTY}, ui_options={ "display_name": "当前日期/时间 (Date/Time)", "readonly": True, }, default_value="点击上方按钮获取最新时间", ) ) def _update_datetime( self, _button: Button, _button_payload: ButtonDetailsMessagePayload, ) -> None: """按钮点击回调：使用当前时区时间刷新 datetime_display 属性值。 采用紧凑且通用的 ISO 友好格式（如 2024-02-04 14:30:45）。 """ # 获取当前带时区的时间戳并格式化 current_datetime = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S") self.set_parameter_value("datetime_display", current_datetime) def process(self) -> None: """运行算子计算核心体。 当该节点在工作流中被调度执行时，引擎会自动调用 `process()`。 通过 `get_parameter_value(...)` 读取输入端口参数值， 并将计算产出的成果写入 `self.parameter_output_values` 字典中。 """ # 读取当前参数值（优先取自上游连线输入，无连线时取前端属性配置面板设置的值） free_text = self.get_parameter_value("free_text") or "" dropdown = self.get_parameter_value("dropdown") integer_slider = self.get_parameter_value("integer_slider") or 0 # 反转自由文本中的各个单词顺序 reversed_words = " ".join(reversed(str(free_text).split())) self.parameter_output_values["reversed_text"] = reversed_words # 将下拉选择的值直通传递至下游端口 self.parameter_output_values["dropdown"] = dropdown # 计算随机浮点数 # 注意：`random.uniform(a, b)` 要求数值型输入。本代码为演示用途； # 生产级算子应当稳健处理 None 或非预期非法值。 random_float = round(random.uniform(0, float(integer_slider)), 3) # noqa: S311 self.parameter_output_values["random_float"] = random_float def validate_before_workflow_run(self) -> list[Exception] | None: """在整个工作流启动运行前执行静态预检校验。 工作流编排引擎会在执行任何计算前优先调用此钩子函数， 从而将可预防的配置缺陷（如必填项为空）在运行前及时暴露给用户。 """ errors = [] free_text_value = self.get_parameter_value("free_text") # 检查 'free_text' 是否为空白 if not free_text_value: errors.append(ValueError(f"算子 '{self.name}' 的 'free_text' 端口不能为空，请输入文本。")) return errors or None