Cloudflare Agents 中 Think 子代理与程序化轮次全解析chat() RPC、saveMessages 与可恢复轮次实战指南【免费下载链接】agentsBuild and deploy AI Agents on Cloudflare项目地址: https://gitcode.com/GitHub_Trending/agents1/agentscloudflare/think是 Cloudflare Agents 生态中基于 Durable Object SQLite 的聊天 Agent 基类。它既可以作为顶层 Agent通过 WebSocket 与浏览器客户端对话也可以作为子代理Sub-agent通过 RPC 从父代理调用并支持完全脱离 WebSocket 的程序化轮次Programmatic Turns。本文以 docs/think/sub-agents.md 为主线深入剖析 Think 的chat()RPC 流式接口、saveMessages()程序化轮次、continueLastTurn()续写、轮次中止与 Chat Recovery 可恢复机制并结合packages/think/src/think.ts的源码实现与 e2e 测试进行印证。读完本文你将掌握如何在父代理中流式调用子代理并转发事件、如何从定时任务或 Webhook 触发无连接模型轮次、如何跨 RPC 边界取消轮次以及如何让一轮推理在 Durable Object 被驱逐后安全恢复。实验性说明Think 的 API 表面已趋于稳定但在其正式毕业Graduate之前仍可能演进。本文示例基于当前仓库源码。一、Think 的双重身份与轮次入口总览Think 在架构上同时承担两种角色这也是理解本文全部 API 的前提角色接入方式典型场景顶层 AgentWebSocketuseAgentChat连接浏览器面向用户的聊天界面子代理RPC父代理通过subAgent()拿到子实例再调用chat()父代理将任务委托给子代理流式接收结果在这两种角色之上Think 还支持程序化轮次不经过 WebSocket直接注入消息并触发模型推理。saveMessages()、submitMessages()、continueLastTurn()都属于这一类。Think 的多个轮次入口最终都汇聚到一个统一入口runTurn(options)见 docs/think/index.md按options.mode分为三种模式模式适用场景返回值对应快捷方法wait默认调用方可以阻塞等待模型响应完成PromiseTurnResultsaveMessages()submit调用方需要快速、持久化的受理确认稍后再查状态PromiseSubmitMessagesResultsubmitMessages()stream调用方希望响应通过回调流式输出RPC 场景Promisevoidchat()本文聚焦其中与子代理和程序化轮次最相关的三个入口chat()RPC 流式、saveMessages()阻塞式程序化轮次与continueLastTurn()续写以及贯穿这三者的恢复与取消机制。至于通用框架原语subAgent、onBeforeSubAgent、useAgent({ sub })、parentAgent、hasSubAgent、listSubAgents、路由形态参见仓库内的 docs/agents/sub-agents.md。二、通过 chat() 实现子代理 RPC 流式调用2.1 方法签名当 Think 作为子代理时chat()会运行完整的一轮持久化用户消息 → 运行 agentic loop模型推理与工具循环→ 持久化助手响应并通过回调实时推送事件流。其在 think.ts 中的实现正是文档所述的流程——onStart先行触发携带requestId随后进入基于队列的_admitTurn执行体推理结果通过_streamResultToRpcCallback逐块流回调用方async chat( userMessage: string | UIMessage, callback: StreamCallback, options?: ChatOptions ): Promisevoid2.2 StreamCallback四个必选事件与一个可选事件interface StreamCallback { onStart(event: { requestId: string }): void | Promisevoid; onEvent(json: string): void | Promisevoid; onDone(): void | Promisevoid; onError(error: string): void | Promisevoid; }各回调的触发时机方法触发时机onStart(event)开始工作前触发暴露requestId供后续取消使用onEvent(json)每个流式分块触发一次JSON 序列化的UIMessageChunkonDone()一轮完成后、助手消息已持久化后触发onError(error)轮次中出现错误时触发需要特别指出的是当前源码中的 StreamCallback 在文档基础上新增了一个可选成员onInterrupted?()。它只在「流内中断」时触发例如chatStreamStallTimeoutMs看门狗超时触发了有界恢复此时该次尝试的最终结果不会通过此回调送达——要么稍后由调度中的续写在另一个 isolate 调用中、没有此回调的情况下产生答案要么恢复预算已耗尽、轮次已带外终结。它既不是onDone本次尝试未完成也不是onError原始 stall 不在此处作为终态错误暴露。省略该回调完全向后兼容默认为空操作但对需要精确区分「已终结」与「仍在恢复中」的消费者很有价值。2.3 ChatOptions信号、客户端工具与渠道元数据interface ChatOptions { signal?: AbortSignal; }文档中的核心字段字段说明signal用于在流式中途取消轮次的AbortSignal而实际源码中 ChatOptions 还携带了更多字段均属于可选的进阶能力字段说明signalAbortSignal跨 DO 边界时不可用见下文「跨 DO 边界」一节clientTools客户端或父代理在运行时定义的客户端工具 schema镜像 WebSocket 协议中携带的clientTools父代理委托给子代理时子代理仍可访问客户端工具onClientToolCall客户端工具调用的执行器用于在同一轮内完成「模型调用客户端工具 → 拿到结果 → 继续推理」的往返。若不提供客户端工具调用将没有结果轮次以悬空工具调用结束channel该轮所属的渠道 idmetadata服务器侧元数据与channel一起持久化到该轮用户消息上恢复/续写时从持久化历史中重新解析从源码可见chat()在内部会为每个requestId获取或链接一个AbortSignalthis._aborts.getSignal(requestId)与linkExternal(requestId, options?.signal)并把用户消息规范化后追加进会话树、广播给已连接客户端再进入推理循环。2.4 工具归属工具属于子代理调用chat()时无法通过options.tools传入工具。工具的持久化能力必须定义在子代理自身子代理的getTools()子代理的 extensions扩展子代理挂载的 MCP 工具客户端工具 schemaclientTools经ChatOptions.clientTools按轮转发。源码在 think.ts 中保留了向后兼容检查遗留调用者若向chat()传入options.tools会收到一条console.warn警告且该值会被忽略。若需要父代理对子代理的编排式委托应使用runAgentTool()/agentTool()见 docs/agents/agent-tools.md。2.5 完整示例父代理调用子代理import { Think, Session } from cloudflare/think; import type { StreamCallback } from cloudflare/think; export class ParentAgent extends ThinkEnv { getModel() { /* ... */ } async delegateToChild(task: string) { const child await this.subAgent(ChildAgent, child-1); const chunks: string[] []; await child.chat(task, { onStart: (event) { console.log(Child started:, event.requestId); }, onEvent: (json) { chunks.push(json); // Optionally forward to a connected client }, onDone: () { console.log(Child completed); }, onError: (error) { console.error(Child failed:, error); } }); return chunks; } } export class ChildAgent extends ThinkEnv { getModel() { /* ... */ } getSystemPrompt() { return You are a research assistant. Analyze data and report findings.; } }父代理通过this.subAgent(ChildAgent, child-1)获取子代理实例子代理通常作为父代理的持久化 facet 挂载拥有独立的 SQLite 数据库与状态再调用其chat()。onEvent收到的每个 JSON 分块都是UIMessageChunk的序列化形式可原样转发给 WebSocket 客户端。2.6 传入字符串还是 UIMessagechat()的第一参数既可以是普通字符串也可以是UIMessage。字符串会被自动包装为一条文本 user 消息// 下面两种写法等价 await child.chat(Analyze this data, callback); await child.chat( { id: crypto.randomUUID(), role: user, parts: [{ type: text, text: Analyze this data }] }, callback );实际上chat()的第一参数类型是TurnInputMessages还支持传函数形态(current) UIMessage[]该函数会在轮到该消息执行时基于当前队列中的会话历史求值这与saveMessages()的函数形态语义一致。2.7 取消一轮子代理轮次AbortSignal在单进程内可用但 RPC 边界上有序列化限制见后文。RPC 安全的取消方式是组合onStart与cancelChat()onStart拿到requestId之后从另一个 RPC 调用如父代理的错误处理分支用该 id 取消let requestId: string | undefined; const callback: StreamCallback { onStart(event) { requestId event.requestId; }, onEvent(json) { // Forward stream chunks. }, onDone() {}, onError(error) { console.error(error); } }; const turn child.chat(Long analysis task, callback); // Later, from another RPC call or failure handler: if (requestId) { await child.cancelChat(requestId, client disconnected); } await turn;如果调用方与被调用方没有被 Workers RPC 分隔例如同一 DO 内部的直接方法调用也可以直接传AbortSignal取消const controller new AbortController(); setTimeout(() controller.abort(), 30_000); await child.chat(Long analysis task, callback, { signal: controller.signal });取消的语义被中止后已经流出的部分助手消息仍然会被持久化partial assistant message 保留符合可恢复流resumable stream的约定。三、saveMessages无 WebSocket 的程序化轮次3.1 签名与返回值saveMessages在没有 WebSocket 连接的情况下注入消息并触发模型轮次。它适用于定时触发的响应、Webhook 触发的轮次、主动型 Agentproactive agents、从onChatResponse链式发起后续轮次等场景。async saveMessages( messages: UIMessage[] | ((current: UIMessage[]) UIMessage[] | PromiseUIMessage[]), options?: SaveMessagesOptions ): PromiseSaveMessagesResult返回{ requestId, status, error? }其中status取值为completed、error、skipped、aborted之一status出现时机completed轮次完整运行结束error轮次已开始但流报告了错误error字段在可用时携带流错误消息skipped轮次在运行中被作废例如chat-clear清空会话用户消息已持久化但没有运行模型aborted轮次在完成前通过options.signal或chat-request-cancel被取消部分助手分块仍然持久化3.2 静态消息形态await this.saveMessages([ { id: crypto.randomUUID(), role: user, parts: [{ type: text, text: Time for your daily summary. }] } ]);注意saveMessages的id由调用方生成。源码实现think.ts中saveMessages内部会为新轮次crypto.randomUUID()一个requestId然后委托给_runProgrammaticMessagesTurn通过_admitTurn进入排队机制——这也意味着多个saveMessages调用会按队列串行执行而不会并发抢占。3.3 函数形态取最新会话状态当多个saveMessages调用排队时函数形态会在轮次真正启动时基于最新消息求值await this.saveMessages((current) [ ...current, { id: crypto.randomUUID(), role: user, parts: [{ type: text, text: Continue your analysis. }] } ]);这对于「先确认前一任务已完成、再追加后续指令」的链式场景非常有用因为current反映的是入队那一刻之后的最新会话快照。3.4 定时响应getScheduledTasks()通过getScheduledTasks()声明周期性触发的提示词轮次Think 会在启动时对声明进行对账reconcile持久化下一次触发的 one-shot 计划并在每次运行后重新武装下一次触发export class MyAgent extends ThinkEnv { getModel() { /* ... */ } getScheduledTasks() { return { dailyReport: { schedule: every day at 09:00, timezone: UTC, prompt: Generate the daily report. } }; } }调度 DSL 支持every n minutes、every n hours、every day at HH:mm、every weekday at HH:mm、every week on monday,wednesday at HH:mm等形态。墙钟时间调度要求显式时区内联timezone、任务级timezone或getDefaultTimezone()三选一。如果闹钟迟到了Think 只运行本该运行的那一次并安排下一次不会回填错过的运行。每个任务必须且只能定义prompt或handler之一prompt任务会通过submitMessages()创建持久化提交handler任务则用于应用自有的工作如创建 Workflow 运行或写入运行台账交付语义为 at-least-once请使用idempotencyKey/occurrenceKey自行实现幂等。3.5 从 onChatResponse 链式发起后续轮次async onChatResponse(result: ChatResponseResult) { if (result.status completed this.needsFollowUp(result.message)) { await this.saveMessages([{ id: crypto.randomUUID(), role: user, parts: [{ type: text, text: Now summarize what you found. }] }]); } }onChatResponse在当前轮完成后触发此时再调用saveMessages会排队发起一轮新的推理实现「先分析、再总结」的多阶段 Agent 行为。3.6 外部取消options.signalsaveMessages接受AbortSignal调用方无需知道内部生成的requestId即可从外部取消。该信号与 Think 每轮的AbortController建立链接中止时推理循环的信号被中止与chat-request-cancel走同一条路径已经流出的部分分块持久化到可恢复流resumable streamsaveMessages以{ status: aborted }解析onChatResponse以status: aborted触发。如果传入的信号在调用saveMessages时已处于已中止状态则不会运行任何推理工作。class MyAgent extends ThinkEnv { async runWithTimeout(text: string) { const controller new AbortController(); setTimeout(() controller.abort(), 30_000); const { status } await this.saveMessages( [ { id: crypto.randomUUID(), role: user, parts: [{ type: text, text }] } ], { signal: controller.signal } ); if (status aborted) { console.log(Turn cancelled by external signal); } } }3.7 跨 Durable Object 边界的注意事项AbortSignal不能作为 RPC 参数跨 DO 边界传递——workerd 的 JSRPC 层会在序列化阶段拒绝它。因此必须在调用saveMessages的那个 DO 内部构造AbortController把父进程的取消意图通过可序列化的机制桥接进来。对于 Agent 编排场景优先使用 Agent ToolsrunAgentTool()与agentTool()已经处理了父代中止信号的桥接、子代理本地的saveMessages({ signal })、事件转发、回放与清理对 Think 与AIChatAgent子代理均适用。对于更低层的自定义 RPC 模式可以从子代理返回一个ReadableStream由父代理取消其 readerworkerd 会把该取消传播回源流的cancel回调子代理在此回调中中止本地控制器。3.8 Hibernation休眠与恢复Think 的聊天恢复在子代理中同样生效底层 fiber 存储在子代理自己的 SQLite 数据库中顶层父代理维护一个「活跃子 fiber」的小型索引当父代理的 alarm 触发时会把恢复检查路由进所属的子代理即使子代理原本处于空闲状态恢复也以子代理作为this执行恢复后的续写continuation可以在子代理内部调用schedule()——物理 alarm 仍归顶层父代理所有父代理会把续写路由回子代理。外部信号的局限外部信号只存在于内存中。如果 DO 在轮次中途休眠恢复后的轮次通过continueLastTurn()运行不再携带原始的options.signal——监听器已在驱逐时丢失恢复路径无法回溯到原始调用方。实际影响在 DO 重启之后才中止的信号对恢复后的轮次没有效果需要恢复轮次响应全新信号的子类应覆写onChatRecovery在原始调用方已消失时拒绝续写return { continue: false }恢复最适合拥有自己客户端重连路径的长期存活聊天子代理若父代理在 agent-tool 运行中途重启Agent Tools 会重新挂接到子代理的持久化运行、回放已存储的分块并在配置的 reattach 预算内等待真实的终态结果。四、continueLastTurn续写最后一轮助手轮次continueLastTurn用于不注入新用户消息的情况下恢复最后一条助手轮次典型场景是拿到工具结果后继续、或从中断恢复后继续protected async continueLastTurn( body?: Recordstring, unknown, options?: SaveMessagesOptions ): PromiseSaveMessagesResult如果最后一条消息不是助手消息则返回{ requestId, status: skipped }。可选的body参数覆盖该次续写存储的 body省略时沿用上一轮的 body。可选的options.signal接受外部AbortSignal契约与saveMessages一致。绝大多数应用不应直接调用此方法。它是面向高级子类与恢复路径的原语面向用户、由服务端触发的轮次通常使用saveMessages()或submitMessages()submitMessages的持久化受理语义参见 docs/think/programmatic-submissions.md。需要注意的是continueLastTurn会铸造一个新的requestId源码注释也明确指出这一点它区别于原始轮次的 id。五、中止进行中的轮次对于无法拿到requestId、但需要「粗粒度取消当前运行」句柄的调用方例如通过 RPC 驱动、一次只跑一轮的子代理助手Think 暴露了两个 protected 方法protected abortRequest(requestId: string, reason?: unknown): void protected abortAllRequests(): voidabortRequest(id, reason?)按 id 中止指定进行中的轮次。若该 id 不存在对应控制器则为空操作no-op。等价于客户端发送chat-request-cancel。abortAllRequests()中止注册表中所有进行中的控制器。适用于不追踪 id 的单用途子代理。两者产生的终态与chat-request-cancel完全一致推理循环终止、部分分块持久化、该轮的ChatResponseResult报告status: aborted。当以程序化方式驱动轮次时优先使用saveMessages/continueLastTurn的options.signal——它从轮次一开始就把取消意图注入进去调用方无需知道 id。补充在公开方法层Think 还提供了对应的cancelChat(requestId, reason?)与cancelAllChats()见 think.ts前者是本文 2.7 节 RPC 安全取消所用的公开方法。另外abortAllRequests()与cancelAllChats()都不会重置排队中的轮次、续写定时器或提交并发状态——完整的清理chat-clear语义由resetTurnState()负责。六、Chat Recovery让轮次跨越驱逐生存6.1 原理一切入口都包在 runFiber 里Think 把聊天轮次包装在Durable Object fiber中实现持久化执行durable execution。当 DO 在轮次中途被驱逐eviction如部署、内存回收重启后可以恢复该轮次。这对顶层 Agent 与子代理都成立对于子代理由顶层父代理的 alarm 驱动恢复检查回到子 facet。每一条轮次入口路径都被runFiber包裹WebSocket 聊天、子代理chat()RPC、自动续写auto-continuation、saveMessages()、submitMessages()执行以及continueLastTurn()。持久化恢复始终开启chatRecovery只用于调整恢复预算与终态行为。一个值得注意的行为阻塞类模式不能嵌套。在活跃轮次内部例如工具execute中调用wait/stream/continuation会抛错因为会死锁轮次队列此时应改用runTurn({ mode: submit })持久化、在当前轮释放队列后运行或addMessages()仅写会话树、不触发推理、因此不会死锁参见 docs/think/index.md。6.2 onChatRecovery 钩子当 DO 重启后检测到被中断的聊天 fiber 时Think 调用onChatRecovery钩子onChatRecovery(ctx: ChatRecoveryContext): ChatRecoveryOptions | void其默认实现think.ts返回{}——即按默认策略持久化部分输出并在可行时继续/重试。6.3 ChatRecoveryContext中断现场快照字段类型说明incidentIdstring本次恢复事件的稳定 IDattemptnumber该事件的当前尝试次数从 1 开始maxAttemptsnumber配置的尝试次数上限到达即终态耗尽recoveryKindretry \| continue恢复是重试未获回答的用户轮次还是续写部分完成的助手轮次streamIdstring被中断轮次的流 IDrequestIdstring被中断轮次的请求 IDpartialTextstring中断前已生成的文本partialPartsMessagePart[]中断前已累积的 partsrecoveryDataunknown \| null轮次中this.stash()存入的数据messagesUIMessage[]当前会话历史lastBodyRecordstring, unknown?被中断轮次的 bodylastClientToolsClientToolSchema[]?被中断轮次的客户端工具createdAtnumber底层 fiber 启动时的 epoch 毫秒数6.4 ChatRecoveryOptions恢复策略字段类型说明persistboolean?是否持久化部分助手消息continueboolean?是否通过continueLastTurn()自动开始新一轮6.5 典型覆写示例export class MyAgent extends ThinkEnv { getModel() { /* ... */ } onChatRecovery(ctx: ChatRecoveryContext) { console.log( Recovering turn ${ctx.requestId}, partial: ${ctx.partialText.length} chars ); return { persist: true, continue: true }; } }persist: true保存部分消息continue: trueAgent 达到稳定状态后Think 调用continueLastTurn()自动续写。流开始前的中断当ctx.streamId 且ctx.partialText 、但最新持久化的消息仍是未获回答的用户消息时Think 默认自动重试该用户轮次除非continue为falseonChatRecovery(ctx: ChatRecoveryContext): ChatRecoveryOptions { if (!ctx.streamId !ctx.partialText) { console.log(Recovering a pre-stream interruption); } return {}; }抑制过度老化的续写对孤立过久、重放不再安全的轮次基于ctx.createdAt设置时间门槛onChatRecovery(ctx: ChatRecoveryContext): ChatRecoveryOptions { if (Date.now() - ctx.createdAt 2 * 60 * 1000) { return { continue: false }; } return {}; }6.6 恢复预算与限制为chatRecovery赋一个对象来调整恢复允许运行多久、何时放弃。只要轮次持续取得前向进展就可以无限期地穿越中断存活——时长本身不是约束——只要它保持在maxRecoveryWork兜底限制之下。恢复只会被下列限制之一「封存」seal。注意chatRecovery false已不再支持。若自动续写不安全请从onChatRecovery()返回{ continue: false }若取消意图必须在休眠后仍然存活请把取消意图持久化存储并在该钩子中检查。关于副作用与成本的指导参见 docs/agents/chat-agents.md。export class MyAgent extends ThinkEnv { chatRecovery { maxAttempts: 10, noProgressTimeoutMs: 5 * 60 * 1000, maxRecoveryWork: 1000, maxOomRetries: 3, terminalMessage: The assistant was interrupted and could not recover., // 从第二次恢复尝试起被咨询。返回 false 以停止。 // 以 config.shouldKeepRecovering(ctx) 方式调用因此它不绑定到 // agent 实例——请在自己以 ctx.recoveryRootRequestId 为键的存储中 // 记录真实的 token/成本开销。 async shouldKeepRecovering(ctx) { return (await getSpendForTurn(ctx.recoveryRootRequestId)) MAX_SPEND; }, async onExhausted(ctx) { console.warn(Recovery exhausted, ctx.incidentId, ctx.reason); } }; }各配置项语义Think 与cloudflare/ai-chat共用同一套恢复配置字段默认值说明maxAttempts10尝试次数上限。有前向进展时重置因此它拦住的是「无进展的 alarm 紧循环」而非健康的长轮次stableTimeoutMs10_000每次尝试等待 isolate 达到稳定状态的时间超时则重新调度noProgressTimeoutMs300_0005 分钟主要的「卡住轮次」约束无前向进展的最长时长到达即封存。每次有进展的尝试都会重置它maxRecoveryWork1000失控循环防护自事件开启以来累计产生的 content/工具单位超过该值即封存仍在进展的轮次work_budget_exceeded。它是一个慷慨的有限兜底使「不断输出内容但永不收敛」例如每次恢复都在流式中途 OOM 的 isolate的轮次无法永远循环。非常长的 agentic 轮次可调大或设为InfinitymaxOomRetries3DO 内存限制重置isolate 超过 128 MB 限制的紧凑重试预算。OOM 通常在重跑时再次 OOM但也可能是瞬时峰值所以恢复重试几次后以out_of_memory封存。只统计以 OOM 结束的尝试。设为0表示第一次 OOM 即封存shouldKeepRecovering—调用方策略从第二次尝试起被咨询。返回false停止恢复。这是 token/成本预算的挂接点ctx.work是粗略的段计数不是 tokenterminalMessage通用消息放弃恢复时展示给用户的消息onExhausted—放弃恢复时调用一次。检查ctx.reasonctx.reason在onExhausted钩子中取值为以下之一no_progress_timeout卡住、max_attempts_exceeded无进展 alarm 循环、work_budget_exceeded失控、out_of_memory反复的内存限制重置、recovery_aborted你的shouldKeepRecovering返回了false、stable_timeout极端抖动。完整共享参考见 docs/agents/chat-agents.md。最后的 OOM 兜底一种严重到绕过上述预算的内存限制重置例如 DO 在醒来加载状态时就 OOM早于恢复运行会在 alarm 边界被捕获当连续maxAlarmMemoryLimitStrikes基础Agent静态选项默认3次 alarm 以内存限制重置结束时被中断的轮次以out_of_memory封存平台 alarm 重试循环被停止发出alarm:memory_limit_reset事件。这限制了循环和账单但不会缩小工作集——真正不再适配 128 MB 的轮次需要更小的上下文。七、稳定性检测hasPendingInteraction 与 waitUntilStableThink 提供了检测 Agent 是否处于稳定状态的方法——无待处理的工具结果、无待批准的审批、无进行中的轮次protected hasPendingInteraction(): boolean只要存在任何带待处理工具调用无结果或待审批的助手消息就返回true。源码实现think.ts基于_pendingInteractionPromise是否非空判断并与_clientResolvableToolNames()联动排除不需要人类介入的孤儿工具调用避免waitUntilStable被永久卡住。protected async waitUntilStable(options?: { timeout?: number }): Promiseboolean返回一个 PromiseAgent 达到稳定状态时解析为true超时则解析为false。const stable await this.waitUntilStable({ timeout: 30_000 }); if (stable) { await this.saveMessages([ { id: crypto.randomUUID(), role: user, parts: [{ type: text, text: Now that you are done, summarize. }] } ]); }这在「先等当前轮次收敛含等待人类审批/客户端工具回放、再安全追加新指令」的场景中非常关键。waitUntilStable也是恢复路径_chatRecoveryContinueDetached在调用continueLastTurn()之前使用的前置闸门——只有确认 isolate 已稳定续写才不会被进行中的状态干扰。八、源码与测试验证想进一步印证本文内容可以在当前仓库中查看以下实现与测试chat()的完整实现packages/think/src/think.ts#L7988-L8136 —— 展示了requestId生成、外部信号链接、options.tools忽略警告、客户端工具按轮转发、_runChatRecoveryFiber包裹推理等全部环节saveMessages的排队与状态判定packages/think/src/think.ts#L11384-L11392 及_runProgrammaticMessagesTurn——可以看到skipped队列代数变化与aborted外部信号的判定位置取消 API 族packages/think/src/think.ts#L12494-L12521 ——cancelChat/cancelAllChats/abortRequest/abortAllRequests的实现全部委托给内部 abort 注册表恢复钩子默认实现packages/think/src/think.ts#L15968-L15972e2e 混沌测试packages/think/src/e2e-tests/chat-recovery.test.ts —— 该测试实际启动wrangler dev通过 WebSocket 发消息启动慢流在流式中途SIGKILL杀死进程模拟真实 DO 驱逐再以相同持久化目录重启最后验证onChatRecovery被触发、部分文本被持久化、fiber 行被清理。这是观察恢复语义的最直接入口相关单元与 e2e 测试还覆盖了agent-tool重挂接/回放如 agent-tool-reattach-recovery.test.ts、run-turn三种模式run-turn.test.ts与 stall 恢复e2e-tests/stall-recovery.test.ts。九、如何选择正确的轮次 API最后把本文介绍的能力放入决策矩阵完整版见 docs/think/index.md使用场景推荐 API浏览器用户发送聊天消息useAgentChatWebSocket 聊天协议服务端代码可以等待模型响应saveMessages()服务端代码需要快速持久化受理 稍后查状态submitMessages()代码应创建周期性提示词轮次或处理器getScheduledTasks()父代码需要向特定子代理直接流式 RPCsubAgent(...).chat()父代理委托任务给保留的子代理含回放/中止桥接agentTool()或runAgentTool()向会话写入消息但不触发模型轮次addMessages()高级子类或恢复代码续写助手轮次continueLastTurn()核心取舍chat()用于低层父到子的流式 RPC前提是你自己掌控转发、取消与回放策略saveMessages()用于调用方拥有触发权且能等待轮次结束submitMessages()用于超时歧义会让重试变得不安全的场景而 Agent Tools 则是在父模型或 Workflow 委托子代理时、希望获得「保留子运行 事件回放 中止桥接 UI 下钻」的推荐方案。无论选择哪条入口Think 的恢复纤维都会让这轮推理在 DO 驱逐后继续存在。【免费下载链接】agentsBuild and deploy AI Agents on Cloudflare项目地址: https://gitcode.com/GitHub_Trending/agents1/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考