自定义节点¶

内置节点不够用时，编写自定义节点即可扩展 plaita。自定义节点与内置节点走同一套 NodeRegistry 机制，使用上无差别。

实现一个自定义节点¶

继承 Node，设 node_type / node_name，实现 execute(self, execution=None)：

from plaita import Node

class GreetNode(Node):
    node_type = "greet"        # 流程 JSON 里 "type" 字段的值
    node_name = "打招呼"

    def execute(self, execution=None):
        # execution.evaluate 解析其中可能含的表达式
        name = execution.evaluate(self.output)
        return f"hello, {name}"

实现 execute，不要覆写 run

Node.run() 会调用 execute() 再做 _validate_output。请实现 execute(self, execution=None)，覆写 run 会绕过输出校验。这是旧文档示例的常见错误。

注册节点¶

使用自定义节点前要先注册到 NodeRegistry：

from plaita import Flow, NodeRegistry, get_default_registry

# 推荐：用独立的 NodeRegistry
registry = NodeRegistry()
registry.register(GreetNode)

# 或注册到默认注册表
get_default_registry().register(GreetNode)

register 返回节点类本身，可作装饰器用：

@get_default_registry().register
class GreetNode(Node):
    node_type = "greet"
    ...

注册时机

必须在执行流程之前注册。Flow 在解析 nodes 时会通过 get_default_registry().parse_node 查找节点类，未注册会抛 unRecognized node type。

在流程中使用¶

{
    "flow_id": "greet_flow",
    "inputType": { "dataType": "object" },
    "nodes": [
        { "type": "start", "id": "start", "next": "greet" },
        { "type": "greet", "id": "greet", "output": "$INPUT.name", "next": "end" },
        { "type": "end", "id": "end", "output": "$NODE.greet", "resultType": "success" }
    ]
}

get_default_registry().register(GreetNode)
flow = Flow.from_string(open("greet_flow.json").read())
print(flow.run(name="kongjie"))  # => "hello, kongjie"

访问上下文¶

execution（FlowExecution facade）代理到 ExecutionContext，提供：

能力	用法
求值表达式	`execution.evaluate(value)`
读状态	`execution.get_state("$INPUT")` 或 `execution.context`
写状态	`execution.set_state(key, value)`
全局变量	`execution.get_global_variable(key)`
子流程	`execution.get_child_execution()`
event bus	`execution.get_or_create_event_bus()`
协作取消	`execution.cancel_event.is_set()`

class SumNode(Node):
    node_type = "sum"

    def execute(self, execution=None):
        nums = execution.evaluate(self.output)  # 期望是数组
        return sum(nums)

异步节点¶

需要异步 I/O 的节点设 async_node = True 并实现 arun：

import asyncio
from plaita import Node

class AsyncFetchNode(Node):
    node_type = "async_fetch"
    async_node = True

    async def arun(self, execution=None):
        url = execution.evaluate(self.output)
        # 用 aiohttp 之类异步拉取
        await asyncio.sleep(0.1)
        return {"url": url, "data": "..."}

NodeRunner 会用 asyncio.wait_for 驱动 arun，超时同样生效。

事件节点（可挂起）¶

要支持断点续执挂起，继承 EventNode 并实现 on_event / on_timeout / on_cancel。详见断点续执 - 扩展节点。

输入/输出校验¶

可在 validate()（构造期调用）与 _validate_input / _validate_output（运行期）加校验：

class SafeNode(Node):
    node_type = "safe"

    def validate(self):
        assert self.output is not None, "output is required"

    def _validate_output(self, result):
        assert isinstance(result, str), "result must be str"

执行外部命令（子进程安全）¶

节点里 spawn 子进程（跑门禁命令、捕获 CLI 输出、起 agent CLI）时，不要把宿主 os.environ 直接交给子进程：worker 进程的环境里有平台凭据（PLAITA_CREDENTIALS_KEY、 DB/Redis 连接串、provider token），全量透传等于把凭据交出去。用公共层 plaita.subprocess_env 按白名单重建：

import subprocess
from plaita.subprocess_env import build_subprocess_env, clip_output

proc = subprocess.Popen(cmd, env=build_subprocess_env(extra={"RECURSIVE_BIN": "recursive"}))
stdout = proc.stdout.read().decode()
return {"stdout": clip_output(stdout, 4000)}   # 超长输出头尾保留，别把全量塞进状态

build_subprocess_env(extra=...) 只放行 SUBPROCESS_ENV_ALLOWLIST（PATH/HOME/ TMPDIR/locale 等）里的宿主变量，再叠加模块级 SUBPROCESS_ENV_EXTRA 与调用点 extra。
子进程确实需要的凭证/旋钮，走 extra 显式声明——白名单即安全边界。
跨进程执行单元还必须显式传播租户与凭据路由（PLAITA_TENANT_ID + 基础凭据文件/密钥路径，子进程入口恢复租户上下文）；ContextVar 不跨进程，忘了传不等于 default。注意 plaita#256 首版的 env 传播在 A42-06 后已收窄：subprocess 沙箱与 worker 同用户、文件系统不受限，不能作为租户凭据隔离边界，非 default 租户会在 spawn 前 fail-closed；租户级 code 节点用 docker（不挂载宿主凭据）。
clip_output(text, cap) 超阈值时保留头 1/4 + 尾 3/4 并标注省略量（诊断信息在尾部）。节点输出会进 checkpoint / 事件流，无截断的大输出对低权限 viewer 也是泄露面。

最佳实践¶

用 execution.evaluate 处理表达式，不要假设字段值是字面量。
节点保持无状态，所有状态写进 execution 上下文，便于分布式持久化。
用 plaita.logger 记日志，与系统日志集成；避免 print。
明确定义 node_type / node_name，前者是流程 JSON 的键。
错误处理交给框架：节点抛异常会被 NodeRunner 按 errorHandler 策略处理，无需自己 try/except 吞错。

分发节点库¶

把领域节点打包成独立 Python 库，通过 plaita.nodes entry_points 实现自动发现，无需用户手动 register。详见节点注册表与插件。

自定义 Node vs 数据源工具¶

两件事容易混在一起，选型时先分清：

	自定义 Node（本页）	plaita-ai 工具层
包	`plaita`	`plaita-ai`
何时用	新控制流语义、领域专用节点、要进 JSON/`type` 字段	把 HTTP/SQL/向量/Python 函数暴露给 Agent / `@flow`
写法	继承 `Node`，设 `node_type`，实现 `execute`	`@tool` / `HttpToolSource` / YAML 清单 → 动态节点
`@flow` 调用	`MY_NODE(...)`（`node_type` 大写）	`GET_USER(...)` 或兼容 `TOOL(action=...)`
文档	本页	工具节点与数据源

examples/agent/ 里的 ToolNode 是教学用自定义节点，用来演示「怎么写 Node」；生产集成请走 plaita-ai 工具层，不要把两套 ToolNode 当成同一个类。

完整可运行示例：Agent 编排¶

仓库里的 examples/agent/ 给出三个开箱即跑的 LLM 相关节点（LLMNode 调 LLM、ToolNode 注册 Python 函数为工具、RetrieverNode 内存检索），并串成 RAG / Tool-use / Router 三个端到端 Agent 案例。内置 FakeLLM 无需 API key 即可运行，也可换真实 LLM。

# examples/ 不随 wheel 分发，需 clone 仓库后在仓库根目录运行
git clone https://github.com/jeffkit/plaita.git
cd plaita
python -m examples.agent.demo

详见 examples/agent/README.md 与应用场景 - Agent 编排。

下一步¶

节点注册表与插件
工具节点与数据源 —— 自定义 Node 不够用、或要接 HTTP/SQL/向量时
应用场景 - Agent 编排
API: plaita.node