数据截至 (上游 commit 7cdd13a1beec)
从图到产品:嵌入运行时、调试器、序列化与测试
30 秒导读: 前四章讲的是 Rivet 的"引擎内核"——节点与数据模型、GraphProcessor 拉取式引擎、control-flow-excluded 撑起的分支循环、LLM 节点与插件生态。但引擎自己不碰真实世界:它不读文件、不连 MCP、不跑代码、不存盘、不画界面。本章讲的就是包在引擎外面的那一圈"外壳与工具链"——把一张图变成能跑、能调、能存、能测、能交付的产品。
1. 这是什么(零基础也能懂)
一句话定义: 这一章讲的是 Rivet 从"一个能算图的库"变成"一个能用的产品"所需要的全部外围子系统。
先建立一个直觉。第 2 章里的 GraphProcessor 是一台纯粹的计算引擎:你喂给它一张图和一份ProcessContext(运行上下文),它就能把图算出来。但它故意不知道怎么读磁盘、怎么发 HTTP、怎么起 MCP 子进程——这些"脏活"全被抽象成接口,留给外面的宿主去实现。
这么设计的好处是同一个引擎能跑在两个完全不同的地方:
| 落地路径 | 谁在用 | 宿主包 |
|---|---|---|
| 库(嵌入到你自己的 Node 程序里) | 后端服务、脚本、Serverless | @ironclad/rivet-node |
| App(桌面图形界面里可视化编排) | 用鼠标连线搭图的人 | @ironclad/rivet-app(Tauri 桌面壳) |
它解决什么问题: 想象你在 Rivet 桌面 App 里用鼠标搭好了一张 LLM 工作流,现在你想:
- 把它部署成一个 HTTP API,让生产服务调用 → 靠
rivet serve(本章 §4)。 - 在本地脚本里嵌入它,读文件、调 MCP → 靠
runGraphInFile(本章 §2)。 - 图跑在远端服务器上,但你想用桌面 App 的界面实时看它每个节点的执行 → 靠 WebSocket 远程调试器(本章 §3)。
- 把图存成文件、明年用新版 Rivet 还能打开 → 靠版本化序列化(本章 §5)。
- 改了图之后想自动回归测试它有没有跑坏 → 靠 Trivet 测试框架(本章 §6)。
一句话直觉: 如果 GraphProcessor 是一台发动机,那么本章讲的就是底盘、油箱、仪表盘、钥匙和年检——发动机再好,没有这些也开不上路。
2. 顶层全景(外壳怎么把引擎包起来)
2.1 五个子系统一张图
本章的五个外围子系统,围绕中央的 GraphProcessor 各管一段。先看它们怎么协作:
┌──────────────────────────────────────┐
│ GraphProcessor(引擎内核) │
│ —— 见第 2 章,本章不重复 —— │
└──────────────────────────────────────┘
▲ ▲ │
注入运行上下文 │ │ attach 挂事件 │ 发事件
(ProcessContext)│ │ ▼
┌────────────────────────┴──┐ ┌────┴──────────────────────────┐
│ ① Node 宿主包 │ │ ② 远程调试器 (WebSocket) │
│ createProcessor / runGraph│ │ startDebuggerServer │
│ 注入 4 个 Native 实现: │ │ 把 nodeStart/nodeFinish/... │
│ NativeApi / Dataset / │ │ 广播给桌面 App 客户端 │
│ CodeRunner / MCP / RefLdr│ └───────────────────────────────┘
└────────────┬───────────────┘
│ 被复用
┌────────────┴───────────────┐ ┌───────────────────────────────┐
│ ③ CLI:rivet serve / run │ │ ④ 序列化(存/读 .rivet-*) │
│ 把图暴露成 HTTP / 命令行 │ │ serializeProject 固定写 v4 │
└────────────────────────────┘ │ deserialize 回退 v4→v3→v2→v1 │
└───────────────────────────────┘
┌────────────────────────────┐ ┌───────────────────────────────┐
│ ⑤ 桌面 App(Tauri 壳) │ │ ⑥ Trivet 测试框架 │
│ Vite/React + jotai + 自研 │ │ 用"验证图"给"被测图"断言 │
│ 画布 NodeCanvas/Wire │ │ runTrivet │
└────────────────────────────┘ └───────────────────────────────┘
怎么读这张图: 中间是引擎;上下左右六个框都是"外围"。①是所有落地方式的地基(它决定引擎能碰到哪些真实世界能力);③⑤都建立在①之上;②是把引擎的内部事件"直播"出去;④⑥则相对独立地负责持久化和质量。
2.2 各子系统一句话职责
| 子系统 | 干什么 | 关键文件 |
|---|---|---|
| Node 宿主包 | 组装 ProcessContext 并驱动引擎跑图 | packages/node/src/api.ts |
| 5 个 Native 实现 | 给引擎"手脚":读文件、存数据集、跑代码、连 MCP、加载引用项目 | packages/node/src/native/* |
| 远程调试器 | WebSocket 服务端,把引擎事件广播给远端 App | packages/node/src/debugger.ts |
| CLI | serve(HTTP API)/run(一次性命令) | packages/cli/src/ |
| 序列化 | 图/项目/数据集 ↔ YAML 文本,带版本回退 | packages/core/src/utils/serialization/* |
| 桌面 App | Tauri Rust 壳 + Vite/React 前端 + 自研画布 | packages/app/src/*、src-tauri/* |
| Trivet | 用一张"验证图"对"被测图"做断言的测试框架 | packages/trivet/src/* |
3. 核心机制一:Node 宿主包如何组装 ProcessContext
这一节讲整个外壳最关键的一件事:引擎不知道怎么碰真实世界,宿主替它把"手脚"都插好。
3.1 它要解决的小问题
GraphProcessor.processGraph(context, inputs, contextValues) 的第一个参数 context(ProcessContext)是一堆接口:NativeApi(文件系统)、DatasetProvider(数据集)、CodeRunner(执行代码节点)、MCPProvider(连 MCP 服务器)、ProjectReferenceLoader(加载被引用的子项目)……引擎只调接口方法,从不关心背后是 Node 的 fs 还是浏览器的 IndexedDB。
宿主包的工作,就是把这些接口用 Node 平台的真实能力实现出来,再一次性注入。
3.2 主入口:三个便利函数
@ironclad/rivet-node 对外暴露一条极短的调用链,从文件到输出只要一行:
// 示意,非源码:嵌入 Rivet 到你自己的 Node 程序
import { runGraphInFile } from '@ironclad/rivet-node';
const outputs = await runGraphInFile('./my.rivet-project', {
graph: 'main', // 图的 id 或名字
inputs: { question: 'hi' }, // 图输入
openAiKey: process.env.OPENAI_API_KEY,
});
// 重点看:你没碰任何 Native 实现,宿主默认全给你插好了
真实实现是三层套娃,一层比一层高层:
| 函数 | 干什么 | 源码 |
|---|---|---|
loadProjectFromFile(path) | 读文件文本 → loadProjectFromString 反序列化成 Project | packages/node/src/api.ts:25 |
createProcessor(project, options) | 造出 processor 并包一个 run(),run() 里注入 ProcessContext | packages/node/src/api.ts:44 |
runGraph / runGraphInFile | 上面两步的组合,一把梭 | packages/node/src/api.ts:35,101 |
3.3 关键一步:run() 里注入了什么
createProcessor 先调核心的 coreCreateProcessor(第 2 章的引擎工厂)拿到裸 processor,再返回一个替换过 run() 的对象。真正的注入发生在这个 run() 里(packages/node/src/api.ts:68-98):
// 真实源码骨架(packages/node/src/api.ts:68),已省略 settings 细节
async run() {
const outputs = await processor.processor.processGraph(
{
nativeApi: options.nativeApi ?? new NodeNativeApi(),
datasetProvider: options.datasetProvider,
mcpProvider: options.mcpProvider ?? new NodeMCPProvider(),
codeRunner: options.codeRunner ?? new NodeCodeRunner(),
projectReferenceLoader:
options.projectReferenceLoader ?? new NodeProjectReferenceLoader(),
settings: { /* openAiKey / pluginEnv / chatNodeTimeout ... */ },
// ...
},
processor.inputs,
processor.contextValues,
);
return outputs;
}
读法: 每个接口都是 options.X ?? new NodeX()——调用方给了就用给的,没给就 new 一个 Node 默认实现。这就是"手脚可插拔"的落点。processGraph 是引擎真正开跑的那个方法(见第 2 章)。
createProcessor 还顺手做了两件外壳杂务(packages/node/src/api.ts:50-58):把 processor.executor 标成 'nodejs';若传了 remoteDebugger,就 remoteDebugger.attach(processor)——把引擎接上本章 §4 的调试器。
3.4 五个 Native 实现,各补一只手脚
这五个类就是"引擎的手脚"。逐个看它们补的是什么能力:
| Native 实现 | 补的能力 | 关键源码 | 值得注意的坑 |
|---|---|---|---|
NodeNativeApi | 文件系统:readdir(支持递归/glob/ignore)、读写文本/二进制 | packages/node/src/native/NodeNativeApi.ts:15 | exec() 故意抛 Not Implemented(:73)——不让图任意执行 shell |
NodeDatasetProvider | 数据集读写,可落盘到 .rivet-data 文件 | packages/node/src/native/NodeDatasetProvider.ts:28 | fromProjectFile 把 .rivet-project 后缀换成 .rivet-data(:74) |
NodeCodeRunner | 执行 Code 节点里的用户 JS | packages/node/src/native/NodeCodeRunner.ts:5 | 用 new AsyncFunction(...) 动态构造函数(:56),按需注入 console/require/process/fetch/Rivet |
NodeMCPProvider | 连 MCP 服务器(HTTP/SSE/STDIO 三种传输) | packages/node/src/native/NodeMCPProvider.ts:8 | HTTP 先试 StreamableHTTP,失败回退 SSE(:21-30);STDIO 走子进程(:107) |
NodeProjectReferenceLoader | 加载被主项目引用的子项目 | packages/node/src/native/NodeProjectReferenceLoader.ts:7 | 按 reference.hintPaths 逐个相对路径试读,全失败才报错(:13-26) |
一个值得记住的安全边界: NodeNativeApi.exec 直接抛异常(packages/node/src/native/NodeNativeApi.ts:73)——即便宿主给了文件读写能力,也不开放任意命令执行。要跑代码,只能走 NodeCodeRunner 那条受控的 AsyncFunction 路径。
3.5 插件环境变量的自动拉取
宿主还替你把插件要的环境变量从 process.env 里抠出来。getPluginEnvFromProcessEnv(packages/node/src/api.ts:106)遍历注册表里每个插件的 configSpec,凡是标了 pullEnvironmentVariable 的字符串配置,就去 process.env 取对应变量塞进 pluginEnv。这样你在库里跑图时,插件密钥不用手动一个个传。
4. 核心机制二:WebSocket 远程调试器
这一节讲一个很"产品化"的能力:图在远端服务器上真跑,但你在本地桌面 App 里像看本地执行一样,实时看到每个节点的启停、输出、报错。
4.1 它要解决的小问题
桌面 App 的调试界面是最好用的——能看到每个节点高亮、看到中间输出、能暂停/继续。但生产环境里图往往跑在没有界面的服务器上。怎么让本地 App 的界面"遥控"并"透视"远端的执行?
答案:在服务器端起一个 WebSocket 服务,把引擎的每一个内部事件都序列化后广播给连上来的 App 客户端;App 反过来也能通过这条连接发 run/pause/abort 等指令。
4.2 数据流:双向
桌面 App(前端) 远端服务器(Node 宿主)
───────────── ──────────────────────
useRemoteDebugger startDebuggerServer
new WebSocket("ws://host:21888") WebSocketServer(port 21888)
│ │
│ ── run / pause / abort / ──────────▶│ handleMessage: ts-pattern
│ user-input / preload │ match(message.type)
│ │ │
│ │ processor.pause() 等
│ │ │
│ ◀── nodeStart / nodeFinish / ───────│ attach(): processor.on(evt)
│ partialOutput / done / error │ → broadcast(JSON)
▼ ▼
界面实时高亮/显示输出 引擎照常 processGraph
读法: 左边 App 发控制指令,右边服务器把指令翻译成对 processor 的方法调用;右边引擎发执行事件,服务器广播回左边界面。默认端口 21888。
4.3 服务端:startDebuggerServer
入口 startDebuggerServer(packages/node/src/debugger.ts:52)开一个 WebSocketServer,默认 port = 21888(:65)。每个连接的消息处理分两类:
① 客户端 → 引擎的控制指令,用 ts-pattern 的 match 分派(packages/node/src/debugger.ts:101):
| 消息 type | 动作 | 源码行 |
|---|---|---|
run | 调 dynamicGraphRun(动态起一次图运行) | debugger.ts:102 |
abort / pause / resume | processor.abort()/pause()/resume() | debugger.ts:141-149 |
user-input | processor.userInput(nodeId, answers)(回答 User Input 节点) | debugger.ts:150 |
preload | processor.preloadNodeData(...)(预置节点输出) | debugger.ts:154 |
set-dynamic-data | 上传整个项目(需 allowGraphUpload) | debugger.ts:122 |
② 引擎 → 客户端的事件广播,在 attach(processor) 里把 processor 的每个事件都转成 broadcast(packages/node/src/debugger.ts:206)。订阅的事件几乎覆盖引擎的全部生命周期:nodeStart、nodeFinish、nodeError、start、done、partialOutput、graphStart、pause、userInput……(debugger.ts:214-283)。
4.4 一个体现工程味的细节:partialOutput 限流
LLM 节点会逐 token 吐 partialOutput。如果每个 token 都通过 WebSocket JSON 序列化发一遍,会把连接打爆。所以 attach 里给 partialOutput 单独做了按节点限流(packages/node/src/debugger.ts:247-256):
// 真实源码(packages/node/src/debugger.ts:247)
processor.on('partialOutput', (data) => {
if (
lastPartialOutputsTimePerNode[data.node.id] == null ||
(lastPartialOutputsTimePerNode[data.node.id] ?? 0) + throttlePartialOutputs < Date.now()
) {
this.broadcast(processor, 'partialOutput', data);
lastPartialOutputsTimePerNode[data.node.id] = Date.now();
}
});
重点看: 每个节点记一个"上次广播时间",两次广播至少间隔 throttlePartialOutputs(默认 100ms)。注释里直白写着这么做的原因——不然序列化端会"ridiculous"。
4.5 客户端:App 怎么连
App 侧用 useRemoteDebugger 这个 hook 连接(packages/app/src/hooks/useRemoteDebugger.ts:27)。默认 URL 就是 ws://localhost:21888(:29),断线会用 setTimeout 自动重连(:68)。有一个特殊 URL ws://localhost:21889/internal 被判定为内部执行器(:39)——即 App 自带的 executor 子进程,不算"真正的远程调试"。
5. 核心机制三:CLI —— serve 与 run
这一节讲怎么不写一行代码就把图跑起来或部署出去。CLI 是 Node 宿主包最薄的一层封装。
5.1 两个命令
CLI 入口 packages/cli/src/cli.ts 用 yargs 注册了两个命令(:9-20):
| 命令 | 用途 | 实现 |
|---|---|---|
rivet run <projectFile> [graphName] | 跑一次图,把输出打到 stdout | packages/cli/src/commands/run.ts:40 |
rivet serve [projectFile] | 把图暴露成 HTTP REST API(Hono 服务) | packages/cli/src/commands/serve.ts:83 |
5.2 run:一次性执行
# 示意:跑一次图,从 --input 传参
rivet run ./my.rivet-project main --input question=hi --include-cost
run(packages/cli/src/commands/run.ts:40)做的事很直白:loadProjectFromFile 读项目 → 从 --input k=v / --context k=v / stdin 收集输入 → createProcessor(...).run() → JSON.stringify 打印输出。默认删掉 cost 字段,除非 --include-cost(:115)。
5.3 serve:部署成 HTTP API
serve(packages/cli/src/commands/serve.ts:83)用 Hono 起一个 HTTP 服务:POST / 收 JSON 输入、跑图、返回 JSON 输出(:115)。几个产品化选项:
| 选项 | 作用 | 源码 |
|---|---|---|
--dev | 每次请求都重新读项目文件(热更新) | serve.ts:116 |
--stream | 用 SSE(Server-Sent Events)把执行事件流式推给客户端 | serve.ts:128,200 |
--allow-specifying-graph-id | 允许在 URL 路径里指定要跑哪张图 | serve.ts:145 |
--expose-cost | 响应里带上 token 成本 | serve.ts:276 |
流式模式 streamGraph(serve.ts:200)会调 createProcessor 拿到 getSSEStream,把 nodeStart/nodeFinish/partialOutputs 事件包成 SSE 流返回——本质上是把 §4 那套事件,换个 HTTP 出口给非 App 的客户端用。
CLI 还内置了很贴心的图名纠错:若找不到指定的 graph,用 didyoumean2 给出"你是不是想跑 X?"的建议(serve.ts:318、run.ts 同理)。
6. 核心机制四:项目序列化与向后兼容
这一节讲 Rivet 怎么保证"今天存的图,明年用新版还能打开"。
6.1 它要解决的小问题
图是用户的资产,格式却会随版本演进。序列化层要同时满足两个要求:写的时候只写最新格式(简单、确定性),读的时候能 吃下所有历史格式(向后兼容)。
6.2 写:永远写 v4
serializeProject(packages/core/src/utils/serialization/serialization.ts:18)直接委托给 projectV4Serializer——只写 v4,不给选择。v4 序列化器(serialization_v4.ts:100)做两件关键事:
- 用
safe-stable-stringify做确定性排序(:110)——保证同一张图每次序列化出逐字节相同的文本,对 git diff 友好。 - 输出 YAML,顶层带
version: 4标记(serialization_v4.ts:114)。
6.3 读:v4 → v3 → v2 → v1 逐级回退
deserializeProject(packages/core/src/utils/serialization/serialization.ts:24)是一串 try/catch 瀑布:先试 v4 反序列化器,抛错就试 v3,再不行 v2,最后 v1,全挂才 throw 'Could not deserialize project'。
deserializeProject(text)
│
┌──────────▼──────────┐ 成功 → 返回 [Project, AttachedData]
│ projectV4Deserializer│──────────────────┐
└──────────┬──────────┘ │
抛错 │ │
┌──────────▼──────────┐ │
│ projectV3Deserializer│──成功 → 返回 ─────▶│
└──────────┬──────────┘ │
抛错 │ │
┌──────────▼──────────┐ │
│ projectV2Deserializer│──成功 → 返回 ─────▶│
└──────────┬ ──────────┘ │
抛错 │ │
┌──────────▼──────────┐ ▼
│ projectV1Deserializer│──成功 → 返回 调用方拿到 Project
└──────────┬──────────┘
抛错 │
throw 'Could not deserialize project'
怎么读: 从上往下是"从新到旧"逐个尝试,命中即返回;每级失败会
console.warn一条(如serialization.ts:33),方便排查是哪一版格式的问题。图(deserializeGraph)和数据集用同样的回退策略(serialization.ts:69)。
一个小设计: v4 反序列化器还能顺带解析出 AttachedData(附加数据),而 v3/v2/v1 回退时一律返回空的 {}(serialization.ts:36,45,54)——老格式没有这个概念,诚实地给空。