高级算子库架构与生命周期钩子 (Advanced Libraries)

在绝大多数常规开发场景下，一个简单的 griptape_nodes_library.json 清单文件足以完整定义一个算子库：引擎读取清单、导入其中枚举的节点模块，并无缝注册。

然而，高级算子库 (Advanced Library) 允许开发者在独立的 Python 类中，深度介入算子库自身的五个关键生命周期节点：节点加载前、节点加载后、库卸载前、引擎收集请求处理器时、以及引擎收集分发后钩子时。

---

什么时候需要编写高级算子库？

❌ 无需编写的情况：若算子库仅由一组预先固化在文件中的标准节点类构成，直接在清单的 nodes 中声明即可，不需要编写任何高级库代码。

✅ 必须编写的场景： - 进程级稀缺资源的统一申请与销毁：如统一申请与释放 GPU Context 显存上下文、初始化 C++ 原生动态链接库 SDK 句柄、维系长连接池或后台独立监控线程； - 动态自适应注册未在清单中枚举的隐式算子：基于本地数据文件、云端元数据或动态代码生成机制按需长出节点（详见 免清单动态注册算子）； - 承载并向系统开放库专属的全局请求总线类型：让本库、其他算子库或外部 MCP 客户端能够通过统一总线调用本库的核心底层能力（详见 get_request_handlers）； - 作为旁路观察者监听引擎原生系统事件：例如在每一次用户存盘工作流后，自动触发本库的审计上报或自动导出（详见 get_post_dispatch_hooks）； - 向系统注册工作流发布器等竞争性扩展提供商。

---

声明与加载配置约定 (Wiring up)

在清单 griptape_nodes_library.json 中添加 advanced_library_path，指向高级库 Python 文件（相对路径）：

{
  "name": "My Library",
  "advanced_library_path": "advanced_library.py",
  "nodes": []
}

在该 Python 文件中继承 AdvancedNodeLibrary 基类，并按需重写对应的钩子（所有钩子默认均为 no-op 空实现）：

from griptape_nodes.node_library.advanced_node_library import AdvancedNodeLibrary


class MyLibrary(AdvancedNodeLibrary):
    def after_library_nodes_loaded(self, library_data, library) -> None:
        print(f"算子库加载完毕，共注册 {len(library.get_registered_nodes())} 个节点")

必须严格遵守的三大硬性加载法则：

1. 类必须在该文件中直接显式定义：引擎通过静态比对类的 __module__ 与导入模块名来识别。严禁通过 from ... import ... 导入外部定义的子类；
2. 首项匹配胜出 (First match wins)：模块扫描时只认第一个符合条件的 AdvancedNodeLibrary 子类；请确保模块内有且仅有一个此类；
3. 构造函数 __init__ 不得包含任何必填实参：引擎以无参形式实例化该类；所有动态状态请在内部初始化或在后续生命周期钩子中按序挂载。

若违反上述任一法则，引擎会将该库标记为 UNUSABLE（不可用），并在界面展示 AdvancedLibraryLoadFailureProblem。

---

算子库装载与卸载时序链路 (The Load Sequence)

深刻理解各个钩子的调度次序是正确掌控生命周期的核心前提。算子库挂载时的完整执行拓扑：

步序	引擎内部执行逻辑	涉及的重要开发者上下文
1	解析并静态强校验 griptape_nodes_library.json 清单	—
2	将本算子库根目录及其虚拟环境的 site-packages 注入到 sys.path	此时本库内的所有子模块均可原生直接 import
3	动态导入高级库模块，并无参实例化你的类	⚠️ 此时该库尚未正式入驻注册表，__init__ 中不可查自身
4	将 Library 实例正式写入全局 LibraryRegistry 注册表	—
5	持久化清单中声明的全部库配置项	—
6	⚡ 回调 before_library_nodes_loaded	在此处可动态修改 library_data.nodes 列表
7	引擎迭代遍历 library_data.nodes，真正装载并注册每一个算子类型	读取的是第 6 步被可能动态修改后的 live 列表
8	注册前端 JavaScript 小部件 Widgets	—
9	⚡ 回调 after_library_nodes_loaded	此时所有节点已装配完毕，可调用 get_registered_nodes()
10	⚡ 回调 get_request_handlers	收集并注册该库对外开放的自定义请求处理器
11	⚡ 回调 get_post_dispatch_hooks	收集并挂载该库对全局事件的分发后监听钩子
12	评估算子库完整健康度 (Fitness) 并标记为 LOADED 就绪	—

卸载流程相反，在注销任何实际资源前率先触发 before_library_unregistered，确保清理代码执行期间底层网络与句柄依然鲜活有效。

---

五大生命周期钩子详解

1. before_library_nodes_loaded

def before_library_nodes_loaded(self, library_data: LibrarySchema, library: Library) -> None: ...

在库已注册但在任何具体节点类被解析前触发。主要用于准备节点加载所依赖的前置环境，或动态向 library_data.nodes 列表追加合成的节点元数据。传入的 library_data 是引擎内部直接持有的实时引用。

2. after_library_nodes_loaded

def after_library_nodes_loaded(self, library_data: LibrarySchema, library: Library) -> None: ...

在清单及动态追加的所有节点类已完全完成类型加载与注册后触发。此时可安全调用 library.get_registered_nodes() 巡检全量节点。

3. before_library_unregistered

def before_library_unregistered(self, library_data: LibrarySchema, library: Library) -> None: ...

在算子库被注销拔除前触发。所有由你创建的 GPU 资源、后台轮询线程、Socket 句柄必须在此完成干净释放。此处的异常会被系统捕获并吞掉，防止清理缺陷卡死主引擎。

---

4. get_request_handlers：向系统开放专属请求服务

def get_request_handlers(self) -> list[tuple[type[RequestPayload], Callable]]: ...

声明一组 (请求类型, 对应处理函数) 元组。任何其他算子、甚至外部接入的 MCP 智能体客户端，均可通过事件总线分发该请求，由你的高级库捕获并给出计算响应。

标准实现“三部曲”：

第 1 步：在公共模块中声明载荷数据结构 (Payloads)

from dataclasses import dataclass
from griptape_nodes.retained_mode.events.payload_registry import PayloadRegistry
from griptape_nodes.retained_mode.events.base_events import (
    RequestPayload,
    ResultPayloadSuccess,
    ResultPayloadFailure,
    WorkflowNotAlteredMixin,
)

@dataclass
@PayloadRegistry.register
class ConvertColorspaceRequest(RequestPayload):
    color: tuple[float, float, float]
    source: str
    target: str

@dataclass
@PayloadRegistry.register
class ConvertColorspaceResultSuccess(WorkflowNotAlteredMixin, ResultPayloadSuccess):
    color: tuple[float, float, float]

@dataclass
@PayloadRegistry.register
class ConvertColorspaceResultFailure(WorkflowNotAlteredMixin, ResultPayloadFailure):
    pass

第 2 步：在高级库钩子中返回映射元组

def get_request_handlers(self) -> list[tuple[type[RequestPayload], Callable]]:
    return [(ConvertColorspaceRequest, self._handle_convert_colorspace)]

def _handle_convert_colorspace(self, request: ConvertColorspaceRequest) -> ResultPayload:
    # 核心计算逻辑，返回 ConvertColorspaceResultSuccess 或 ConvertColorspaceResultFailure
    ...

第 3 步：在业务算子节点中消费该请求

result = GriptapeNodes.handle_request(ConvertColorspaceRequest(color=(1.0, 0.0, 0.0), source="rgb", target="hsv"))
if result.failed():
    raise RuntimeError(f"颜色空间转换失败: {result.result_details}")
success = cast("ConvertColorspaceResultSuccess", result)

⚠️ 关键约束： - 请求分发必须在 process() 中进行，严禁在节点的 __init__ 中分发请求（会触发死锁与 Strict Mode 违规）； - 全局唯一排他性：同一个请求类型全引擎只允许存在一个主处理器； - 仅限主编排器进程：在独立 Worker 子进程中运行的库无法向主进程暴露此能力。

---

5. get_post_dispatch_hooks：旁路感知系统事件

def get_post_dispatch_hooks(self) -> list[tuple[type[RequestPayload], Callable]]: ...

返回一组监听映射。当引擎原生处理器完成了对特定请求的处理并产出结果后，旁路调用你的回调函数 (request, result) -> None：

def get_post_dispatch_hooks(self):
    return [(SaveWorkflowRequest, self._on_workflow_saved)]

async def _on_workflow_saved(self, request: RequestPayload, result: ResultPayload) -> None:
    if not isinstance(result, SaveWorkflowResultSuccess):
        return
    await self._append_audit_line(result.file_path)

核心特征与约束： - 纯旁路通知：返回值被直接丢弃，回调无法更改操作结果，也无法打断流程； - 只读原则：严禁在回调中修改请求或结果载荷； - 严禁在回调中递归分发引擎请求：这会破坏引擎全局流水线的内部操作栈深度。

---

免清单动态自适应算子生成 (Dynamic Node Registration)

若算子体系是完全基于外部配置文件（如 JSON/YAML 规范）驱动生成的，可以在清单中保留 "nodes": []，完全在 before_library_nodes_loaded 中动态追加 NodeDefinition：

class MyLibrary(AdvancedNodeLibrary):
    def before_library_nodes_loaded(self, library_data, library) -> None:
        library_data.nodes.extend(
            NodeDefinition(
                class_name=spec["class_name"],
                file_path="generated_nodes.py",
                metadata=NodeMetadata(
                    category="dynamic",
                    description=spec["description"],
                    display_name=spec["display_name"],
                ),
            )
            for spec in load_specs_from_disk()
        )

动态类生成的底层要求：

引擎在实例化节点时会导入 generated_nodes.py 并提取对应的类名。你可以通过模块级 __getattr__ (PEP 562) 动态动态合成类并注入缓存：

def __getattr__(name: str) -> type[DataNode]:
    spec = find_spec(name)
    if spec is None:
        raise AttributeError(f"模块未找到对应算子规格: {name}")
    node_class = build_node_class(spec)
    globals()[name] = node_class  # 缓存至全局，确保多次提取为同一单例对象
    return node_class

⚠️ 致命陷阱：动态构建类时必须显式标注 __module__ 与 __qualname__！

# 必须显式传递 __module__ 与 __qualname__，否则 Python ABC 会将其归入 "abc" 模块，
# 导致包含该算子的工作流存盘反序列化时全面崩溃！
return type(
    spec["class_name"],
    (DataNode,),
    {
        "__init__": __init__,
        "process": process,
        "__module__": __name__,
        "__qualname__": spec["class_name"],
    },
)