Electric Agents 实体协作指南：spawn / fork / observe 与子实体协调实战

Electric Agents 实体协作指南：spawn / fork / observe 与子实体协调实战

【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric

本文以 Electric Agents（构建于 Electric sync 之上的 Agent 平台）为背景，系统讲解实体（Entity）之间的协作机制：通过spawn派生子实体、用fork分支会话、借助EntityHandle观察与发消息、以及用send、sleep、wake完成跨实体协调。读完本文，你将掌握如何在实体 handler 中编排"管理者—工人"（manager-worker）多智能体流程，并理解其底层基于 durable stream 的状态语义。

实体协作的核心模型

Entities 是 Electric Agents 中的最小执行单元，每个实体拥有独立的 handler、durable stream 与 state。实体之间的协作主要依靠三条路径完成：

• spawn：创建子实体（child entity）；
• observe：观察已存在的实体，订阅其状态变化；
• send：向目标实体投递消息。

这三者共同构成了实体协作的"创建—订阅—通信"闭环。底层实现位于 packages/agents-runtime/src/context-factory.ts，handler 收到的ctx对象将spawn、fork、forkSelf、observe、send、sleep等能力分别委托给config.doSpawn、config.doFork、config.doObserve、config.executeSend等运行时实现。

spawn：创建子实体

spawn是创建子实体的唯一入口：

const child = await ctx.spawn(type, id, args?, opts?)

各参数含义如下表：

spawn 是"只创建"操作

spawn是纯创建操作：如果(type, id)组合在实体的 manifest 中已经存在，调用会直接抛错。从源码中的类型签名可以看出，ctx.spawn委托给config.doSpawn(type, id, args, opts)（见 context-factory.ts），其 opts 仅包含initialMessage、wake、tags、observe等创建期选项，没有"更新已有实体"的语义。测试用例 process-wake.test.ts 专门验证了重复 spawn 同一(type, id)会失败。

如果你需要获取一个已存在子实体的句柄，应当改用observe(entity(url))，而非重复调用spawn。

wake 选项：父实体何时被重新唤醒

wake控制子实体完成后父实体 handler 的重新调用时机，支持三种形式：

• 'runFinished'—— 子实体的一次 agent run 完成时唤醒父实体。默认情况下，子实体的文本响应会包含在 wake 事件中。
• { on: 'runFinished', includeResponse?: boolean }—— 同上，但设置includeResponse: false可省略子实体的文本响应。
• { on: 'change', collections?: string[], debounceMs?: number, timeoutMs?: number }—— 指定集合（collections）发生变化时唤醒。

从类型定义看（types.ts），Wake联合类型还支持{ on: 'change', ops?: TagOperation[] }形式的标签操作监听，以及debounceMs（去抖）与timeoutMs（超时）控制。

spawn返回一个EntityHandle，用于后续与子实体交互。

子实体的沙箱隔离

当子实体需要独立的文件系统、进程或网络访问权限，或希望继承父实体的沙箱时，应使用沙箱机制。opts.sandbox的类型SpawnSandboxOption（见 types.ts）支持两种取值：

• 'inherit'—— 直接继承父 wake 已解析的沙箱（profile、key、persistent）；若父实体本身没有沙箱，则优雅地回退为无沙箱；
• 对象形式 —— 指定profile，可选的scope/persistent，或通过inherit: true显式继承、key加入某个共享沙箱。

完整的安全边界与配置方法见 sandboxing.md。

fork：从历史分支出新实体

fork从另一个实体的最新已完成 run 的历史创建新实体。典型场景是：分支（branch）一段会话，尝试不同的后续走向：

const fork = await ctx.forkSelf("variant-a", { initialMessage: { text: "Explore the risky option instead." }, tags: { branch: "variant-a" }, })

• ctx.fork(sourceEntityUrl, id, opts?)—— 从指定实体 fork；
• ctx.forkSelf(id, opts?)—— 从当前实体 fork。

从源码看（context-factory.ts），forkSelf只是把config.entityUrl作为源实体 URL 转发给doFork。新 fork 默认是发起 fork 实体的子实体，并自动注册一个runFinished（includeResponse: true）的 wake，因此父实体会在 fork 的下一轮 run 结束时被唤醒。

ForkOptions（见 types.ts）与spawn的 opts 语义一一对应：initialMessage、wake（省略时默认{ on: 'runFinished', includeResponse: true }）、tags（在从源实体复制的标签之上追加）、observe（默认true）。

如果希望"即发即忘"（fire-and-forget）式分支——不建立父子关系、不订阅 wake——传入observe: false即可。

EntityHandle：实体的统一句柄

spawn与observe都会返回EntityHandle，其接口定义见 types.ts：

interface EntityHandle { entityUrl: string type?: string db: EntityStreamDB // 被观察实体流的 TanStack DB events: ChangeEvent[] send(msg: unknown): Promise<SendResult> // 发送后续消息 status(): ChildStatus | undefined }

• entityUrl—— 目标实体的 URL，是send、observe、持久化关联的关键字段；
• db—— 被观察实体流对应的 TanStack DB，可直接查询该实体的集合数据；
• events—— 本次观察期间累积的变更事件；
• send(msg)—— 向该实体发送后续消息，返回SendResult；
• status()—— 返回子实体当前状态。

status()返回ChildStatus对象（尚无已知状态时返回undefined），包含.status、.entity_url、.entity_type和.key字段（ChildStatus = ChildStatusEntry，见 types.ts）。

SendResult（见 types.ts）是一个判别联合：

type SendResult = | { sent: true; targetUrl: string } | { queued: true; targetUrl: string }

即消息可能已送达（sent），也可能因目标实体当前不可用而被入队延迟投递（queued）。

子实体完成后的续接：不要在同一个 wake 里等待

协调子实体的关键原则：不要在同一个 wake 内同步等待子实体的输出。正确做法是：

1. 用 wake 条件 spawn 或 observe 子实体；
2. 持久化足够多的元数据（例如child.entityUrl）用于后续关联子实体；
3. 立即 return，等待子实体完成触发下一次 wake。

async handler(ctx, wake) { if (ctx.firstWake) { const child = await ctx.spawn( "worker", "analyst", { systemPrompt: "Analyze this input", tools: ["read"] }, { initialMessage: "Initial task.", wake: { on: "runFinished", includeResponse: true }, } ) ctx.state.children.insert({ key: child.entityUrl, url: child.entityUrl, status: "running", }) return } const finished = wake.payload?.finished_child if (finished) { ctx.state.children.update(finished.url, (draft) => { draft.status = finished.run_status draft.response = finished.response ?? "" }) } }

要点拆解：

• ctx.firstWake：实体首次被唤醒时执行 spawn，此后不再重复创建。测试辅助代码firstWake: false等字段（见 context-test-helpers.ts）也印证了 handler 每次唤醒都会携带该标志；
• 子实体完成事件：第二次 wake 时从wake.payload?.finished_child读取完成信息（finished.url、finished.run_status、finished.response），更新自己在ctx.state中持久化的子实体记录；
• includeResponse: true适用简单文本交接；若子实体的输出是结构化数据或体量较大，更推荐让子实体直接写入共享状态（shared state），仅以runFinishedwake 作为续接信号，避免把大块文本塞进 wake 事件。共享状态的创建与订阅见 shared-state.md。

这一模式在运行时测试中得到了完整覆盖：例如 process-wake.test.ts 中 spawn 子 worker 时配置{ wake: { on: 'runFinished', includeResponse: true } }，验证了父实体按预期被唤醒。

observe：观察已存在的实体

observe用于订阅一个已存在的实体，而不重新创建它：

const handle = await ctx.observe(entity(entityUrl), { wake: { on: "change", collections: ["runs", "childStatus"] }, })

返回EntityHandle。wake用于在观察目标发生变化时重新调用父 handler，例如上例监听runs与childStatus两个集合的变更。

observe的能力比spawn更通用：从 types.ts 的重载签名可以看到，ctx.observe支持三种观察源：

• 实体流（sourceType: 'entity'）—— 观察实体，返回EntityHandle；
• 共享状态 DB（sourceType: 'db'，带 schema）—— 返回SharedStateHandle & ObservationHandle，可对共享集合做类型化读写；
• 其他观察源（如 cron、webhook 等）—— 返回ObservationHandle。

实体状态集合的定义方式见 defining-entities.md，wake 条件的完整语义见 waking-entities.md。

send：向其他实体发送消息

send是"即发即忘"（fire-and-forget）的消息投递：

ctx.send("/assistant/target-id", { text: "Hello" }) ctx.send("/assistant/target-id", payload, { type: "custom_type" })

消息会出现在目标实体的inbox集合中。第二个示例演示了如何通过{ type: "custom_type" }指定 inbox 消息类型——这也对应spawn的initialMessageType与ForkOptions.initialMessage的底层投递机制。

底层executeSend（见 context-factory.ts）还支持afterMs参数做延迟投递，send的返回SendResult可区分"已送达"与"已入队"两种结果。

sleep：将实体置回空闲

sleep让实体回到空闲状态，结束当前 handler 调用：

ctx.sleep()

实体仍然存活，可以再次被 inbox 消息或观察到的变化唤醒。实现上（context-factory.ts）只是设置sleepRequested = true，由getSleepRequested()驱动进程将实体置回 idle，而非销毁实体。

与既有子实体交互：spawn 一次，observe 多次

spawn只在firstWake创建子实体；后续 wake 需要用observe获取句柄来交互：

async handler(ctx, wake) { if (ctx.firstWake) { await ctx.spawn( "worker", "analyst", { systemPrompt: "...", tools: ["read"] }, { initialMessage: "Initial task.", wake: { on: "runFinished", includeResponse: true }, } ) } const analyst = await ctx.observe(entity("/worker/analyst")) if (wake.type === "inbox") { analyst.send(wake.payload) } }

这段代码展示了两种 wake 的形态：

• 首次唤醒（firstWake）：spawn 子实体并配置 wake；
• 后续唤醒：通过observe拿到子实体句柄，检查wake.type—— 若为inbox，则说明有外部消息进来，将其透传给子实体analyst.send(wake.payload)。

一句话总结：spawn负责"创建一次"，observe负责"每次唤醒都获取句柄"——它是在创建之后与子实体交互的标准方式。这与spawn的只创建语义（重复调用会抛错）形成了严格互补。

Worker 与带认证的 API：最少权限原则

内置的worker实体类型是最少权限（least-privilege）沙箱。它接收：systemPrompt、选定的tools子集、可选的sharedDb配置，以及 spawn 时投递的initialMessage。内置 worker 的实现见 packages/agents/src/agents/worker.ts。

不要把密钥插进 worker 的 prompt 或消息

绝对不要把process.env.API_KEY、认证 token 等密钥插值进 worker 的 prompt 或消息——这些内容会持久化在实体的 durable stream 中，等于把凭据写进了不可变的审计日志。

推荐模式：Manager 侧预取（manager-side prefetch）

正确姿势是"管理者取数，工人干活"：由 manager 完成带认证的请求，再把原始数据（而非凭据）传给 worker：

// In the manager's tool: const response = await fetch(apiUrl, { headers: { Authorization: `Bearer ${process.env.API_KEY}` }, }) const data = await response.json() // Pass data, not credentials, to the worker await ctx.spawn( "worker", id, { systemPrompt: "Summarise this data.", tools: ["read"] }, { initialMessage: JSON.stringify(data), wake: { on: "runFinished", includeResponse: true }, } )

数据经initialMessage传入，worker 全程不接触凭据。

需要后续认证调用时：注册自定义 worker 类型

当 worker 需要做后续的带认证调用（例如分页拉取、按条件取数）时，不要使用内置的worker类型，而应在应用里注册一个自定义 worker 实体类型，在注册时通过闭包捕获凭据（credential at registration time）。这样认证能力封装在实体类型内部，既满足最少权限，又不把密钥写入 durable stream。

小结与下一步

实体协作的完整工具箱可归纳为一张速查表：

建议按以下顺序继续深入：先阅读 writing-handlers.md 理解 handler 与ctx的完整能力，再结合 managing-state.md 掌握ctx.state的持久化用法，随后通过 sandboxing.md 配置子实体的隔离边界。若想从整体上理解这些能力如何被运行时调度与测试，可进一步查看 context-factory.ts 与 process-wake.test.ts 中的实现与测试用例。

【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric