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 entityobserve观察已存在的实体订阅其状态变化send向目标实体投递消息。这三者共同构成了实体协作的创建—订阅—通信闭环。底层实现位于 packages/agents-runtime/src/context-factory.tshandler 收到的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?)各参数含义如下表参数类型说明typestring实体类型名必须已注册idstring子实体的唯一 IDargsRecordstring, unknown传递给子 handler作为ctx.args使用opts.initialMessageunknown投递给子实体的第一条消息opts.initialMessageTypestring可选为initialMessage指定 inbox 消息类型opts.wakeWake何时重新唤醒父实体见下文opts.tagsRecordstring, string应用到子实体的键值标签opts.observeboolean是否同时观察该子实体默认trueopts.sandboxSpawnSandboxOption子实体的沙箱配置或继承方式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.tsWake联合类型还支持{ 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?)—— 从指定实体 forkctx.forkSelf(id, opts?)—— 从当前实体 fork。从源码看context-factory.tsforkSelf只是把config.entityUrl作为源实体 URL 转发给doFork。新 fork 默认是发起 fork 实体的子实体并自动注册一个runFinishedincludeResponse: 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.tsinterface EntityHandle { entityUrl: string type?: string db: EntityStreamDB // 被观察实体流的 TanStack DB events: ChangeEvent[] send(msg: unknown): PromiseSendResult // 发送后续消息 status(): ChildStatus | undefined }entityUrl—— 目标实体的 URL是send、observe、持久化关联的关键字段db—— 被观察实体流对应的 TanStack DB可直接查询该实体的集合数据events—— 本次观察期间累积的变更事件send(msg)—— 向该实体发送后续消息返回SendResultstatus()—— 返回子实体当前状态。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 内同步等待子实体的输出。正确做法是用 wake 条件 spawn 或 observe 子实体持久化足够多的元数据例如child.entityUrl用于后续关联子实体立即 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共享状态 DBsourceType: db带 schema—— 返回SharedStateHandle ObservationHandle可对共享集合做类型化读写其他观察源如 cron、webhook 等—— 返回ObservationHandle。实体状态集合的定义方式见 defining-entities.mdwake 条件的完整语义见 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 的形态首次唤醒firstWakespawn 子实体并配置 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 managers 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。小结与下一步实体协作的完整工具箱可归纳为一张速查表API作用关键语义ctx.spawn(type, id, args?, opts?)创建子实体只创建重复调用抛错observe默认truectx.fork / forkSelf从历史分支实体默认子实体 runFinishedwakeobserve: false即发即忘ctx.observe(entity(url), opts?)订阅已存在实体每次唤醒获取句柄的标准方式handle.send(msg)给实体发消息进入目标inbox集合ctx.send(url, payload, opts?)即发即忘投递支持type与afterMsctx.sleep()回到空闲可再次被唤醒不销毁实体handle.status()查子实体状态返回ChildStatus或undefined建议按以下顺序继续深入先阅读 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考