Electric Agents HandlerContext 完全指南状态、协调与执行控制【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric导读HandlerContext是 Electric Agents 平台中每个实体Entity处理器Handler的第一个参数它为 Agent 处理器提供了状态访问、实体间协调原语与 Agent 运行配置的统一入口。本文基于仓库中的官方 API 参考文档website/docs/agents/reference/handler-context.md结合 packages/agents-runtime 源码逐项讲解HandlerContext的完整接口——从state/db数据访问、wake唤醒机制到spawn/fork/observe实体编排、上下文组合与目标管理再到信号处理与沙箱隔离。读完本文你将能够熟练编写自定义实体处理器并利用 fork、观察、延迟消息等原语构建多 Agent 协作应用。HandlerContext 是什么在 Electric Agents 中每个实体如聊天会话、工作流任务都通过handler(ctx, wake)处理唤醒事件。HandlerContext就是传入的第一个参数ctx它聚合了该实体在本次唤醒中的全部能力状态与数据state实体自定义状态集合、dbTanStack DB 实例、events触发本次唤醒的变更事件协调原语spawn生成子实体、fork分叉会话、observe观察外部源、send向其他实体发消息Agent 配置useAgent(config)配置 LLM Agentagent.run()执行 Agent 循环执行控制signal取消信号、onSignal生命周期信号、sleep结束处理而不运行 Agent。接口定义位于 packages/agents-runtime/src/types.ts#L1057-L1234源码中的实际接口比文档中的简化示意更精确——它带有四个泛型参数TState、TArgs、TActions、TDb分别约束状态集合、spawn 参数、自定义动作与数据库实例的类型从而为每个实体提供类型安全的上下文。export interface HandlerContext TState extends StateProxy StateProxy, TArgs extends ReadonlyRecordstring, unknown ReadonlyRecordstring, unknown, TActions extends Recordstring, (...args: Arrayany) unknown Record string, (...args: Arrayany) unknown , TDb extends EntityStreamDBWithActions EntityStreamDBWithActions, { /* ... */ }实体定义通过EntityDefinition.handler声明处理器types.ts#L1257-L1265ctx的类型由实体定义中的state、creationSchema、actions自动推导export interface EntityDefinition... { handler: ( ctx: HandlerContext StateProxyFromTState, EntityArgsTCreationSchema, HandlerActionsTActions, EntityStreamDBWithActionsTState, TActions , wake: WakeEvent ) void | Promisevoid }HandlerContext实例由 context-factory.ts 在每次唤醒会话wake-session时装配测试用例可在 test/context-factory.test.ts 中看到其完整的构造方式。属性总览属性类型说明firstWakeboolean实体尚无持久化 manifest 条目时的初始设置阶段为true可用状态检查做一次性初始化wakeHandlerWake当前唤醒的规范化视图等价于处理器第二个参数slashCommandsSlashCommandHelpers读写暴露给结构化 composer 输入的斜杠命令定义tagsReadonlyEntityTags与实体关联的键值元数据principalRuntimePrincipal \| undefined触发当前唤醒的主体当服务端提供时entityUrlstring实体 URL 路径例如/chat/my-convoentityTypestring注册的实体类型名例如chatargsReadonlyRecordstring, unknown实体创建时传入的 spawn 参数dbEntityStreamDBWithActions实体的 TanStack DB 实例含已注册动作stateTState按集合名索引的代理对象每个属性是一个StateCollectionProxyeventsArrayChangeEvent触发本次唤醒的变更事件actionsRecordstring, (...args) unknown实体定义actions工厂提供的自定义非 CRUD 动作自动生成的 CRUD 动作位于ctx.db.actions与ctx.stateelectricToolsAgentTool[]宿主提供的运行时级工具需要时可展开进 Agent 配置可能为空signalAbortSignal当前唤醒应提前停止如关闭或SIGINT时触发 abort可传给可取消的耗时工作sandboxSandbox本次唤醒会话的沙箱运行时工具用它进行文件系统、进程与网络访问attachmentsAttachmentsApi读取与创建该实体由 manifest 支撑的附件其中值得深入的是state。StateProxy Recordstring, StateCollectionProxytypes.ts#L493每个集合代理提供insert/update/delete/get/toArray五个方法types.ts#L481-L491export interface StateCollectionProxyT extends object Recordstring, unknown, ... { insert: (row: TInsert) EntityTransaction update: (key: TKey, updater: (draft: T) void) EntityTransaction delete: (key: TKey) EntityTransaction get: (key: TKey) T | undefined toArray: ArrayT }firstWake 的一次性初始化firstWake仅在实体尚未持久化任何 manifest 条目时为true。官方建议用状态检查来做一次性普通状态初始化例如首次唤醒时写入默认配置handler: async (ctx) { if (ctx.firstWake ctx.state.settings.toArray().length 0) { ctx.state.settings.insert({ key: locale, value: zh-CN }) } }HandlerWake唤醒的规范化视图ctx.wake是对处理器第二个参数WakeEvent的规范化便捷视图types.ts#L911-L941源码中的完整类型如下export type HandlerWake InboxHandlerWake | OtherHandlerWake export type InboxHandlerWake { type: inbox source: string raw: WakeEvent message: { type: string; payload: unknown; from?: string } } export type OtherHandlerWake { type: other wakeType: string source: string payload?: unknown raw: WakeEvent }inbox 唤醒实体收到来自其他实体或用户的消息message字段包含消息类型、载荷与发送者other 唤醒其他类型的唤醒如定时任务 cron、观察结果、信号wakeType区分具体类型。底层WakeEvent还包含fromOffset/toOffset/eventCount/summary/fullRef等流位置信息供运行时定位事件批次。处理 inbox 消息的典型写法handler: async (ctx, wake) { if (ctx.wake.type inbox) { const { type, payload } ctx.wake.message if (type user_message) { await ctx.agent.run(payload.text) } } }方法总览方法返回类型说明useAgent(config)AgentHandle配置 LLM Agent必须在agent.run()之前调用useContext(config)void声明带 token 预算与缓存层的上下文来源timelineMessages(opts?)ArrayTimestampedMessage将实体时间线投影为有序 LLM 消息数组insertContext(id, entry)void插入持久化上下文条目跨唤醒保留重复 id 覆盖旧条目removeContext(id)void按 id 移除上下文条目getContext(id)ContextEntry \| undefined按 id 读取上下文条目listContext()ArrayContextEntry列出全部上下文条目setGoal(input)GoalEntry设置或替换实体的活动目标clearGoal()boolean清除活动目标返回是否移除了目标getGoal()GoalEntry \| undefined读取活动目标markGoalComplete(summary?)GoalEntry \| undefined标记活动目标完成可选记录总结updateGoalUsage(tokens, opts?)GoalEntry \| undefined向活动目标累加 token 用量并可更新状态agent.run(input?)PromiseAgentRunResult运行已配置的 Agent 循环可选input会作为用户消息追加spawn(type, id, args?, opts?)PromiseEntityHandle生成子实体fork(sourceUrl, id, opts?)PromiseEntityHandle在另一实体最新完成运行处分叉它forkSelf(id, opts?)PromiseEntityHandlectx.fork(ctx.entityUrl, id, opts)的便捷封装observe(source, opts?)EntityHandle \| SharedStateHandle \| ObservationHandle观察一个来源返回类型取决于来源类型unobserve(sourceRef)Promisevoid停止按 sourceRef 观察某个 pg-sync 来源mkdb(id, schema)SharedStateHandleT创建新的共享状态流send(entityUrl, payload, opts?)PromiseSendResult向另一实体发送消息onSignal(handler)void注册本次唤醒期间的生命周期信号处理recordRun()RunHandle在内置runs集合记录非 LLM 运行replyText(text)void不调用 LLM 直接写入合成助手文本回复setTag(key, value)Promisevoid设置实体标签deleteTag(key)Promisevoid删除实体标签sleep()void结束处理而不运行 Agent实体保持空闲直到下次唤醒上下文组合Context 相关方法上下文组合是HandlerContext中最核心的 Agent 能力。四个方法协同工作useContext声明来源timelineMessages投影时间线insertContext/removeContext/getContext/listContext管理持久化上下文条目。useContext声明带 token 预算与缓存层级的上下文来源。缓存层级常量CACHE_TIERS定义于 types.ts由 context-factory 装配timelineMessages将实体时间线投影为有序 LLM 消息数组典型用途是作为易失来源的content函数。其底层投影选项TimelineProjectionOpts支持since与自定义projection回调types.ts#L438-L441持久化上下文条目ContextEntry结构为{ id, name, attrs?, content, insertedAt }types.ts#L334-L343通过insertContext写入后跨唤醒保留重复 id 覆盖旧条目。handler: async (ctx) { // 声明一个带预算的上下文来源 ctx.useContext({ /* token budget 与缓存层级配置 */ }) // 插入跨唤醒持久化的上下文条目 ctx.insertContext(policy, { name: policy, content: Always reply in Chinese., }) // 读取 / 列举 / 移除 const entry ctx.getContext(policy) const all ctx.listContext() ctx.removeContext(policy) }目标管理Goal API目标管理让 Agent 拥有可追踪的任务目标。GoalEntry的源码结构types.ts#L352-L364export interface GoalEntry { id: string objective: string status: GoalStatus tokenBudget: number | null // null 表示无上限省略则使用运行时默认值 tokensUsed: number summary?: string // markGoalComplete 记录完成说明 createdAt: string // ISO 字符串与其他 manifest 类型一致 updatedAt: string }GoalInput接受objective、可选的status与tokenBudgetnull为无上限省略用运行时默认。典型流程handler: async (ctx) { if (ctx.firstWake) { ctx.setGoal({ objective: 调研三种方案并输出对比报告, tokenBudget: 20000, }) } // ... Agent 执行 ... ctx.updateGoalUsage(tokensUsed) ctx.markGoalComplete(已完成三方案对比报告) }Agent 配置与运行useAgent(config)必须在agent.run()之前调用二者返回句柄类型如下types.ts#L943-L1029export interface AgentHandle { run: (input?: string, abortSignal?: AbortSignal) PromiseAgentRunResult } export type AgentRunResult { result?: unknown writes: ArrayChangeEvent toolCalls: Array{ name: string; args: unknown; result: unknown } usage: { tokens: number; duration: number } }AgentConfig支持systemPrompt、model字符串或模型对象、provider、tools、getApiKey、reasoning/thinkingBudgets、onPayload、onStepEnd每步结束时的 token 统计回调预算核算应使用uncachedInput output、modelTimeoutMs、modelMaxRetries、testResponses等types.ts#L953-L983。handler: async (ctx) { const agent ctx.useAgent({ systemPrompt: You are a helpful assistant., model: gpt-4o, tools: [...ctx.electricTools, myCustomTool], }) const result await agent.run(Summarize the conversation.) }从源码看agent.run()的底层执行会通过 outbound bridgeoutbound-bridge.ts在runs集合写入运行记录并派发text_delta事件这正是聊天 UI 渲染消息流的来源。实体编排spawn 与 sendspawn用于生成子实体opts接受tags、observe、initialMessage、initialMessageType、wake、sandbox。源码注释明确types.ts#L1115-L1132observe: false时父实体不订阅子实体流返回的EntityHandle是 fire-and-forget 的——访问.status会抛出异常适用于高扇出high-fanout且父实体不关心子状态的场景wake可自定义子实体完成后的唤醒条件。handler: async (ctx) { const child await ctx.spawn(worker, analysis-1, { task: summarize }, { initialMessage: Review the current workspace., tags: { role: worker }, wake: { on: runFinished, includeResponse: true }, }) }send向其他实体发送消息支持type与延迟投递afterMs毫秒await ctx.send(/chat/other-agent, { text: 请检查结果 }, { type: review_request, afterMs: 1000, // 延迟 1 秒投递 })SendResult可能为{ sent: true; targetUrl }或{ queued: true; targetUrl }types.ts#L466-L468延迟投递走内部待发送队列。Forking会话分叉ctx.fork(sourceEntityUrl, id, opts?)在源实体最新完成运行其main流上处创建子分叉forkSelf(id, opts?)等价于ctx.fork(ctx.entityUrl, id, opts)types.ts#L1133-L1173。const fork await ctx.forkSelf(variant-a, { initialMessage: { text: Try a different approach. }, tags: { branch: variant-a }, })分叉的关键语义源码注释确认新分叉默认创建为当前实体的子实体与spawn相同的父所有权模型并在分叉时注册runFinished includeResponse唤醒——分叉下次运行结束时当前实体会带着响应被唤醒id必须显式提供因为调用方需要预先知道新 URL/horton/id中的id且请求在重试时具有幂等性相同 id 由服务端去重initialMessage将分叉 发送合并为一次调用但不是原子操作它会在分叉创建与派发链接之后才发送因此部分失败可能留下空闲的已派发分叉tags会叠加在从源实体复制的标签之上wake可覆盖默认的runFinished includeResponse订阅例如为高扇出分叉设置 debounceobserve: false完全退出父关系无唤醒、无 manifest 条目、无回复路径即 fire-and-forget。// fire-and-forget 分叉 await ctx.forkSelf(variant-b, { observe: false })观察外部来源observe 与 unobserveobserve(source, opts?)是重载方法返回类型取决于来源类型types.ts#L1174-L1185实体来源sourceType: entity→EntityHandledb 来源sourceType: db带 schema→SharedStateHandle ObservationHandle其他来源 →ObservationHandle。官方推荐使用electric-ax/agents-runtime导出的辅助函数构造ObservationSourceobservation-sources.tsimport { entity, cron, entities, db, pgSync } from electric-ax/agents-runtime handler: async (ctx) { // 观察另一个实体的运行结果 await ctx.observe(entity(/chat/parent), { wake: { on: runFinished, includeResponse: true }, }) // 观察定时任务 await ctx.observe(cron(daily-report, { schedule: 0 9 * * * })) // 观察共享状态流 const shared await ctx.observe(db(findings-stream, { findings: { schema: z.object({ key: z.string() }), type: finding, primaryKey: key }, })) // 观察 Postgres 同步源 await ctx.observe(pgSync({ url: process.env.DATABASE_URL, tables: [issues] })) }unobserve(sourceRef)用于停止观察某个 pg-sync 来源它只移除当前实体在该源上的唤醒共享的 pg-sync bridge 会继续为其他观察者运行types.ts#L1186-L1191。mkdb创建共享状态流mkdb(id, schema)创建新的共享状态流返回SharedStateHandleT——按 schema 集合名键入的集合代理并暴露流 idtypes.ts#L533-L544。共享状态 schema 使用SharedStateCollectionSchemaschemaZod 或任意 Standard Schema 校验器、type持久流事件类型如finding、primaryKey主键字段名必须是字符串字段。const shared ctx.mkdb(findings-stream, { findings: { schema: z.object({ key: z.string(), domain: z.string(), finding: z.string() }), type: finding, primaryKey: key, }, }) shared.findings.insert({ key: f1, domain: web, finding: XSS risk })沙箱Sandboxctx.sandbox在唤醒会话开始时从实体的沙箱 profile 中选择。源码注释明确其生命周期types.ts#L1087-L1098由运行时按entity.sandbox.profile指定的 profile 配置未选择时回退为当前工作目录不受限模式在每个唤醒会话开始时创建在processWake的外层finally中释放——处理器绝不能直接调用sandbox.dispose()同一唤醒会话处理同一实体的多个排队唤醒时复用同一个沙箱跨会话则重建会话间的状态保持是 provider 的职责。编写需要文件系统、子进程或网络访问的自定义工具时应通过ctx.sandbox使行为遵循当前沙箱 profile。生成的子实体可以继承或选择沙箱await ctx.spawn(worker, analysis, args, { sandbox: inherit, initialMessage: Review the current workspace., })附件Attachmentsctx.attachments暴露与实体关联的、由 manifest 支撑的附件。完整 APItypes.ts#L378-L386export interface AttachmentsApi { list(filter?: { subject?: AttachmentCreateInput[subject]; role?: input | output }): ArrayManifestAttachmentEntry get(id: string): ManifestAttachmentEntry | undefined read(id: string): PromiseUint8Array create(input: AttachmentCreateInput): PromiseManifestAttachmentEntry }AttachmentCreateInput接受bytesUint8Array | ArrayBuffer | Blob、mimeType、filename以及subject类型可为inbox/run/text/tool_call/context并关联对应 key、roleinput/output、meta。运行时用它水合图片与文件上下文context-factory 中MAX_HYDRATED_IMAGE_ATTACHMENTS 4、单附件上限 10MB自定义处理器或工具也可用它检查上传的文件handler: async (ctx) { const files ctx.attachments.list({ role: input }) for (const file of files) { const bytes await ctx.attachments.read(file.id) // 处理文件字节 } }斜杠命令Slash Commandsctx.slashCommands暴露注册在实体上的结构化 composer 命令。静态命令来自实体类型定义EntityDefinition.slashCommands处理器可以添加或替换动态命令供发送composer_input消息的 UI composer 使用ctx.slashCommands.register({ name: summarize, description: Summarize the current session, })处理 composer 载荷时使用ctx.wake或处理器的wake参数检查传入的composer_input消息动态命令定义在 composer-input.ts 中校验validateSlashCommandDefinitions。生命周期信号与取消ctx.signal用于可取消的耗时工作ctx.onSignal()用于处理器交付的生命周期信号。两者分工明确源码注释确认types.ts#L1202-L1215SIGINT中止当前活跃的处理器调用通过ctx.signalAbortSignal传播——把 signal 传给 fetch、子进程等可取消操作即可优雅停止SIGSTOP/SIGCONT运行时控制暂停/恢复SIGKILL终态不可捕获运行时当前将SIGHUP、SIGTERM、SIGUSR交付给onSignal处理器。handler: async (ctx) { // 可取消的耗时工作 const resp await fetch(url, { signal: ctx.signal }) // 生命周期信号处理 ctx.onSignal(async ({ signal, reason }) { if (signal SIGTERM) { await cleanup(reason) } }) }注意SIGINT会中止当前处理器调用并传递到ctx.signal因此它不会出现在onSignal中——后者只接收运行时交付的SIGHUP、SIGTERM、SIGUSR。RunHandle记录非 LLM 运行recordRun()服务于在ctx.agent.run()之外执行工作、但仍想暴露运行生命周期事件的处理器。例如包装 CLI 子进程、HTTP 调用等外部操作使观察者能通过runFinished唤醒被通知。完整接口types.ts#L1038-L1055export interface RunHandle { /** 生成的运行键如 run-3与实体 runs 集合中的值一致 */ readonly key: string /** 结束运行并写入对应 runs 集合更新满足 runFinished 唤醒匹配器 */ end(opts: { status: completed | failed; finishReason?: string }): void /** 以 text_delta 事件附加响应文本通过 run_id 关联多次调用按序追加 */ attachResponse(text: string): void }handler: async (ctx) { const run ctx.recordRun() try { const output await runExternalCli(analyze, args) run.attachResponse(output) run.end({ status: completed }) } catch (err) { run.end({ status: failed, finishReason: String(err) }) } }attachResponse()追加与运行关联的文本增量可包含在runFinished唤醒载荷中观察者需设置wake: { on: runFinished, includeResponse: true }。LLM 驱动的实体无需手动调用——useAgent流程已通过 outbound bridge 内部记录运行。相关行为由 test/record-run.test.ts 覆盖。其他方法速查replyText(text)写入完整的runs texts text_delta序列使聊天 UI 将其渲染为普通助手消息。适用于运行时驱动的回复斜杠命令、错误消息完全不涉及 LLM源码注释见 types.ts#L1224-L1230setTag/deleteTag管理实体的键值元数据标签sleep()结束处理器而不运行 Agent实体保持空闲直到下次唤醒principalRuntimePrincipal结构为{ url, key?, kind?, id? }types.ts#L546-L551表示引发当前唤醒的主体。完整示例一个使用多原语的处理器综合上述能力一个研究型 Agent 实体处理器可以这样组织import { entity, db, cron } from electric-ax/agents-runtime import { z } from zod export const researcher defineEntity({ type: researcher, state: { findings: { schema: z.object({ key: z.string(), topic: z.string() }) } }, handler: async (ctx) { // 1. 首次唤醒初始化目标并观察定时任务 if (ctx.firstWake) { ctx.setGoal({ objective: 每日生成一份技术简报, tokenBudget: 30000 }) await ctx.observe(cron(daily, { schedule: 0 8 * * * })) } // 2. 处理 inbox 消息或 cron 唤醒 if (ctx.wake.type inbox) { const { payload } ctx.wake.message await ctx.insertContext(topic, { name: topic, content: payload.topic }) } // 3. 配置并运行 Agent const agent ctx.useAgent({ systemPrompt: You are a research assistant., model: gpt-4o, tools: [...ctx.electricTools], }) const result await agent.run() // 4. 更新目标用量并写入共享状态 ctx.updateGoalUsage(result.usage.tokens) ctx.state.findings.insert({ key: Date.now().toString(), topic: result.result?.toString() ?? }) // 5. 分叉一个变体做备选方案 const fork await ctx.forkSelf(variant-${Date.now()}, { initialMessage: { text: Propose an alternative approach. }, observe: false, // fire-and-forget }) }, })总结HandlerContext是 Electric Agents 实体处理器的统一入口其设计围绕三条主线状态与数据state/db提供类型安全的实体数据访问events携带唤醒事件attachments管理文件与图片协调与编排spawn/fork/observe/send构成多 Agent 协作原语配合mkdb共享状态流与tags元数据可构建父-子、观察者-被观察者、消息驱动的复杂拓扑Agent 执行控制useAgentagent.run驱动 LLM 循环useContext/timelineMessages/context entries 组合上下文窗口Goal API 追踪任务目标signal/onSignal/sandbox提供生命周期与隔离保障recordRun/replyText让非 LLM 工作也融入统一运行模型。实际使用中请记住几个关键约束useAgent必须先于agent.run()调用处理器不得调用sandbox.dispose()由processWake拥有SIGINT通过ctx.signal传播而非onSignalfork 的id必须显式提供以保证幂等。深入理解这些语义后你可以参考 test/context-factory.test.ts 与 test/process-wake.test.ts 中的测试用例验证并学习各 API 的真实调用方式。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考