内置节点

下表列出 plaita 默认注册表中的内置节点，共 23 种：18 种随 plaita.node 内置（_BUILTIN_NODES），另 5 种扩展节点经 plaita.nodes entry_points 随包自动注册（无需额外安装）。node_type 即流程 JSON 中 type 字段的值。

type	类	展示名	分支	异步	需 extra	用途
start	Start	开始				流程起点，无逻辑
end	End	结束				流程终点，返回结果或抛业务错误
assignment	Assignment	赋值				求值并产出值，支持上游输出选择
switch	Switch	判断	✓			按 branches 条件多路跳转
if	Bool	IF	✓			二分支：条件真走 next，假走 else_next
case	SwitchLegacy	CASE	✓			按 target 等值匹配 cases
loop	Loop	重复				对集合循环执行子流程，可带中止条件
while	While	条件循环				按条件反复执行子流程，带 max_iterations 上限保护
map	Map	映射				对集合映射，可并发
filter	Filter	过滤				按子流程返回 bool 过滤集合
find	Find	查找				返回首个子流程返回真的元素
reduce	Reduce	归纳				对集合逐项归约
child	InlineFlow	内联子逻辑				内联子流程，可共享父上下文
reference	ReferenceFlow	引用逻辑				引用独立子流程（不共享父上下文）
parallel	Parallel	并行				多分支并行执行（thread/process）
http	HTTP	http			http	发起 HTTP 请求
event	EventNode	事件节点		✓		等待事件，用于断点续执挂起
mock	Mock	数据固定				占位节点，原样返回 value 字段（试跑固定下游输入用）
delay	DelayNode	延迟节点				延迟指定时间后触发事件继续流程
approval	ApprovalNode	审批节点				发起人工审批，等待审批决策后继续
redis_queue	RedisQueueNode	Redis队列节点				与 Redis 队列交互（入队/出队触发）
kafka_queue	KafkaQueueNode	Kafka队列节点				与 Kafka 队列交互
http_callback	HttpCallbackNode	HTTP回调节点			http	发起 HTTP 回调并等待响应

扩展节点的运行前提

delay / approval / redis_queue / kafka_queue / http_callback 这 5 种扩展节点经 entry_points 自动注册、可直接解析；但它们是事件驱动的——实际执行需相应基础设施（Redis / Kafka / 审批与回调服务），通常配合 server extra 的外延服务使用。详见扩展节点。

未默认注册的节点

• code（CodeNode）：0.5.0 起移出默认注册表。使用前须显式：

from plaita.node import register_code_node
register_code_node()  # 默认 docker 沙箱；无 Docker 时传 default_backend="subprocess"|"unsafe"|"restricted"

后端白名单（防流程 JSON 逐节点降级到 unsafe）见 code 沙箱后端白名单。

• calculate / redis（需 redis extra）：存在于源码但不在 _BUILTIN_NODES，自行 register。

start / end

start 无逻辑，仅作为起点标识。

end 的 resultType 决定行为（不写时默认 success；写了无法识别的值会打 logger.warning 并按 success 处理，不会静默返回 None）：

resultType	行为
success（默认）	求值 output 作为流程结果返回
nop	返回 None
error	抛 FlowResultError(error.code, error.message)，透传业务错误码

{ "type": "end", "id": "end", "output": "$INPUT.name", "resultType": "success" }
{ "type": "end", "id": "bad", "resultType": "error", "error": { "code": 4001, "message": "参数非法" } }

assignment

求值 output（或从 upstreamOutput 选择上游输出）作为节点结果。

{
    "type": "assignment",
    "id": "greet",
    "output": "$F.concat('hi ', $INPUT.name)",
    "outputType": { "dataType": "string" },
    "next": "end"
}

upstreamOutput 用于多上游汇聚时按 upstream（上一个节点 id）挑选对应输出， 类型是列表，每项形如 {"upstream": "<上游节点id>", "output": "<表达式>"}：

{
  "type": "assignment", "id": "merge",
  "upstreamOutput": [
    {"upstream": "fetch_user",  "output": "$NODE.fetch_user"},
    {"upstream": "fetch_order", "output": "$NODE.fetch_order"}
  ],
  "next": "e"
}

多分支汇总裁决的更简惯用法：End 节点的 output 直接写对象字面量聚合各分支 结果，例如 "output": {"user": "$NODE.fetch_user", "order": "$NODE.fetch_order"} ——未实际执行的分支其节点结果解析为 None，不需要 upstreamOutput 逐项挑选。

switch / if / case

详见 流程定义 - 节点与控制流 的分支示例。

• switch：branches[].condition + priority + isDefault，按优先级匹配
• if：condition + next（真）+ else_next（假）
• case：target + cases[]（等值匹配）+ default

条件 operator 见 流程定义 - 条件与运算符。

loop / while / map / filter / find / reduce

loop / map / filter / find / reduce 这类节点对集合操作，每个元素跑一个子流程（childFlow）。子流程输入为 item 与 index（reduce 为 first/second）。

节点	子流程输入	子流程输出	节点输出
loop	item, index	任意	最后一次结果（满足 condition 时中止）
while	item, index	任意	最后一轮结果（条件不满足即退出；未执行任何轮次时为 None）
map	item, index	任意	结果列表（concurrent: true 可并发）
filter	item, index	bool	命中元素组成的子集
find	item, index	bool	首个命中元素
reduce	first, second	同元素类型	归约结果（可带 initial）

{
    "type": "map",
    "id": "double_all",
    "collection": "$INPUT.numbers",
    "concurrent": true,
    "childFlow": {
        "inputType": { "dataType": "object" },
        "nodes": [
            { "type": "start", "id": "s", "next": "e" },
            { "type": "end", "id": "e", "output": "$F.mul($INPUT.item, 2)", "resultType": "success" }
        ]
    },
    "next": "end"
}

并发与副作用

map 的 concurrent 与 parallel 节点会并发执行。并发中不要使用带副作用的表达式函数（pop/set/clear 等），也不要在共享上下文上写竞争数据。

while

条件循环节点：以 condition 为继续条件反复执行子流程——条件满足则进入下一轮，不满足即退出。max_iterations（默认 1000，JSON 亦可写驼峰 maxIterations）是迭代上限保护，达到上限强制停止并打 logger.warning。子流程输入为 item（上一轮结果，首轮为 None）与 index（轮次，从 0 起）；condition 上下文可引用 $LOOP-ITEM / $LOOP-INDEX。节点输出为最后一轮子流程结果。

{
    "type": "while",
    "id": "retry_until_three",
    "condition": { "field": "$LOOP-INDEX", "operator": "lt", "value": 3 },
    "maxIterations": 100,
    "childFlow": {
        "inputType": { "dataType": "object" },
        "nodes": [
            { "type": "start", "id": "s", "next": "e" },
            { "type": "end", "id": "e", "output": "$INPUT.index" }
        ]
    },
    "next": "end"
}

condition 的字段结构同分支条件（field / operator / value，operator 取值见条件与运算符）。condition 缺省时不循环（仅执行一轮）。

mock

数据固定（pin）占位节点：原样返回 value 字段作为节点输出。主要用于编排台调试——把某次试运行的节点输出「固定」下来，后续试跑直接以固定值作为该节点结果，从而跳过真实执行（如 HTTP 调用）反复调试下游；也可手工充当测试桩。

{
    "type": "mock",
    "id": "stub_user",
    "value": { "name": "alice", "age": 30 },
    "next": "end"
}

child / reference

两者都执行子流程，区别在于上下文共享：

节点	父上下文	类比
child (InlineFlow)	子流程可引用父级及更高级数据	匿名函数闭包
reference (ReferenceFlow)	不使用父上下文，子流程由上层调度器注入	调用独立函数

{
    "type": "child",
    "id": "sub",
    "input": "$INPUT.payload",
    "childFlow": { "inputType": { "dataType": "object" }, "nodes": [ ... ] },
    "next": "end"
}

parallel

多分支并行执行，mode 选择执行方式：thread（默认）/ process（coroutine 模式已于 0.4.0 下线）。thread 模式受 GIL 限制，只对 IO 型分支有加速——纯 CPU 分支实测 32 workers 与 2 workers 同速，CPU 密集场景请用 process。joinBranches 列出的分支会汇聚结果（{branchName: result}），其余作为后台分支 fire-and-forget。isConditional 为真时按 branches[].condition 过滤要执行的分支。

{
    "type": "parallel",
    "id": "fanout",
    "mode": "thread",
    "joinBranches": ["a", "b"],
    "branches": [
        { "name": "a", "flow": { "nodes": [ ... ] }, "input": "$INPUT.x" },
        { "name": "b", "flow": { "nodes": [ ... ] }, "input": "$INPUT.y" }
    ],
    "next": "end"
}

process 模式的可序列化要求

mode: "process" 使用进程池，分支 flow 与节点必须可 pickle。threading.Event 等不可序列化对象无法跨进程。

code

执行用户代码（需 code extra 才能跑 JS；Python 走沙箱后端）。0.5.0 起不在默认注册表，使用前须：

from plaita.node import register_code_node
register_code_node()  # 默认 docker；无 Docker 时显式传 default_backend

流程 JSON 里的 sandbox_backend 是逐节点可覆盖的：不设白名单时，流程作者写 "sandbox_backend": "unsafe" 就能在宿主进程内 raw exec 任意代码。生产部署入口 （worker / console）因此默认施加白名单：PLAITA_SANDBOX_ALLOWED_BACKENDS（未配置 → docker）∪ 生效默认后端，白名单外的后端在解析期即被拒绝。库调用方自己接线 时用 register_code_node(allowed_backends=(...))，或经 plaita.node.resolve_sandbox_allowed_backends() 取同一口径。详见 运维 Runbook · code 沙箱后端白名单。

language 为 js 或 python，代码需定义一个 run 函数，input 作为参数传入。

language: "js" 默认被拒（plaita#29）：js 的历史实现（PyExecJS）绕开整个档位体系 （无隔离 / 无超时 / 无取消），因此语言白名单默认只放行 python。放行须运营者显式配置 PLAITA_SANDBOX_ALLOWED_LANGUAGES=python,js（库调用方用 register_code_node(allowed_languages=(...))，或经 plaita.node.resolve_sandbox_allowed_languages()）；放行后 js 同样按 sandbox_backend 走档位：docker 跑 PLAITA_SANDBOX_DOCKER_NODE_IMAGE（默认 node:20-alpine）容器， subprocess 跑 node -e（PLAITA_SANDBOX_NODE_BIN，默认 node；10s 墙钟 + 超时/取消 整组击杀），unsafe 仍是裸 PyExecJS。restricted 只有 Python 实现，js 声明它会在解析期 报错。详见 运维 Runbook · code 语言白名单。

default_backend="subprocess" 时子进程受 PLAITA_SANDBOX_TIMEOUT（墙钟秒，默认 10； 不是 CPU 时间——睡眠与计算消耗同一份预算；0/负值在启动期具名拒绝）约束。 地址空间上限 PLAITA_SANDBOX_MEMORY_MB（MB）默认 0 = 不限制：非 0 值在 Linux 上经 RLIMIT_AS 真实生效，会卡住 node/npm/pnpm 系命令的启动（V8 启动期预留 GB 级虚拟地址， 表现为截断的崩溃栈）——这类命令请保持 0。

多租户凭据（plaita#256 / A42-06）：subprocess 后端与 worker 同用户、文件系统 不受限，且所有租户共用同一把解密密钥——子进程不是凭据隔离边界：租户 B 的用户代码 可以 set_current_tenant("default") 改写路由，或直接读取 default 凭据文件并解密 （A42-06 探针）。因此非 default 租户在该后端 spawn 之前即被拒绝（具名 TenantContextPropagationError，消息含 PLAITA_SUBPROCESS_TENANT_ISOLATION_UNAVAILABLE）， 用户代码零执行机会；default 租户语义不变。租户级 code 节点请用 docker 后端（容器内 不挂载宿主凭据文件）；unsafe / restricted 在宿主进程内沿用执行上下文，同样不是 凭据隔离边界。

{
    "type": "code",
    "id": "transform",
    "language": "python",
    "code": "def run(x):\n    return x.upper()",
    "input": "$INPUT.text",
    "next": "end"
}

http

发起 HTTP 请求，需 http extra（requests + aiohttp）。支持表达式寻址、headers、body、代理等。响应可通过 $NODE.http_id.data / .status / .headers 引用。。可选装 fast extra（orjson）加速大 payload 的 JSON 编解码（缺失自动回退标准库）

{
    "type": "http",
    "id": "call_api",
    "method": "POST",
    "url": "https://api.example.com/users",
    "headers": { "Content-Type": "application/json" },
    "body": { "name": "$INPUT.name" },
    "timeout": "PT5S",
    "errorHandler": { "strategy": "continue_with", "defaultValue": { "data": null } },
    "next": "end"
}

访问策略（SSRF 防护）：requestTimeout（默认 30s，上限 600s，必须是正的有限秒数； 它是整个请求含所有重定向跳的总预算，每跳只拿剩余额度，不再逐跳重新起算——#267）、 allowedHosts / deniedHosts（精确域名 / *.suffix / CIDR）、blockPrivateNetworks （拒绝回环/内网/链路本地等私网段）、maxRedirects（默认 5，上限 20，必须是 非负整数：负值 / 非有限 / 小数在构造期具名拒绝，0 合法 = 不跟随重定向；同步与 异步同口径，超限报 Too many redirects）。 上述边界均在构造期校验：保存/发布/加载时越界直接失败，避免单个节点长期占用共享 worker。负值必须在构造期拒绝——sync/async 的重定向循环对负值解释不同（sync 会先发 一跳，async 的 range(max_redirects+1) 为空则一跳都不发）。执行级取消 （cancel_requested）在每跳发送前复查，取消后不再发起下一跳。 重定向不使用客户端自动跟随（sync/async 同一形状）：节点逐跳手动跟随——策略激活 时每跳复检、跨源剥离凭据头、中间跳 body 丢弃不读；方法改写与跨源剥离规则对齐 requests（301 仅把 POST 转 GET，302/303 非 HEAD 转 GET，307/308 保留方法与 body； 同 host 换 scheme/端口也算跨源）。 策略激活的请求建连解析经同一策略校验并只连校验过的 IP（关闭 DNS rebinding TOCTOU）， 并且忽略 HTTP_PROXY 等代理环境变量（经代理的连接由代理侧解析，pinning 无从 生效）。运营者级强制见 运维 Runbook （PLAITA_HTTP_BLOCK_PRIVATE=1 后节点声明只能更严不能更松）。

响应体默认上限 10MiB（PLAITA_HTTP_MAX_RESPONSE_BYTES 可调），超限报错—— 流式读，不会把超大 body 读进内存/状态；重定向中间跳的 body 也在上限之内（不读）。

event

事件节点，用于断点续执。async_node=True，在 Distributed 模式下执行后挂起，等待外部事件到达恢复。详见 断点续执。

{
    "type": "event",
    "id": "wait_approval",
    "event_type": "$INPUT.approval_topic",
    "event_filter": { "request_id": "req-42" }
}

字段说明：

• 字段名为 event_type / event_filter（无驼峰归一化，JSON/YAML 里必须写小写下划线形式）。
• 字段值不支持 {{ }} 模板插值；event_type 支持 $ 前缀表达式——以 $ 开头时执行期对上下文整体求值（上例在运行时解析为 $INPUT.approval_topic 的值，如 "approval.leave"）。
• event_filter 为字面量字典等值匹配。

on_event / on_timeout / on_cancel / on_error 分别处理恢复/超时/取消/错误情况。

下一步

• 自定义节点
• API: plaita.node