diff --git a/docs/features/direct-model-agent/README.md b/docs/features/direct-model-agent/README.md index 6a0b82d5..a6899076 100644 --- a/docs/features/direct-model-agent/README.md +++ b/docs/features/direct-model-agent/README.md @@ -33,6 +33,8 @@ Task、Job、Subjob、Run 和 Subagent 的对象关系以 ## 功能边界 - `process_execute` 和 `subagent_delegate` 是 GoodBuddy 直连模型内置工具，不是外部 MCP。 +- 直连文本模型的普通问答、工具轮次和上下文摘要会在尚无可见输出时，对瞬时网络错误或 + 单次请求超时最多自动重试 3 次。 - OpenCode、Continue 和 DeepSeek Harness 继续使用各自原生执行与委派能力。 - Ask 不允许进程执行；Subagent 若在 Ask 中使用，只能继承 Ask 的只读能力。 - Execute 表示用户授权当前执行空间账号的完整能力，不增加第二套工具审批或权限档位。 diff --git a/docs/features/direct-model-agent/logic-design.md b/docs/features/direct-model-agent/logic-design.md index c59a5750..4e0f8caf 100644 --- a/docs/features/direct-model-agent/logic-design.md +++ b/docs/features/direct-model-agent/logic-design.md @@ -5,8 +5,8 @@ | 项目 | 内容 | | --- | --- | | 状态 | 已实施 | -| 版本 | 0.2 | -| 日期 | 2026-09-01 | +| 版本 | 0.3 | +| 日期 | 2026-09-03 | | 关联 PRD | [PRD](./prd.md) | | 关联故事 | [User Stories](./user-stories.md) | @@ -22,6 +22,8 @@ 8. 子任务是父请求内部 Job/Subjob，不创建顶层 Task 或独立 Conversation。 9. 父请求取消、Runtime 释放或应用退出会传播到全部活动子级和进程。 10. OpenCode、Continue 和 DeepSeek Harness 的原生能力不被本功能替换或包装。 +11. 直连模型只在尚未显示文本或推理内容时自动重试瞬时网络错误或请求超时，最多重试 + 3 次；用户取消和已显示的部分输出禁止重放。 ## 2. 状态维度 @@ -33,6 +35,7 @@ | 本机 Shell | `available`、`unavailable` | | Subagent 深度 | `0`、`1` | | 父请求 | `active`、`cancelling`、`terminal` | +| 模型请求重试 | `initial`、`retry 1..3`、`terminal` | ## 3. 工具可见性决策表 @@ -158,7 +161,26 @@ type SubagentDelegateResult = { 已经提交的正常终态不被稍后到达的取消覆盖。父请求进入终态后，迟到的子级文本和工具事件 只用于资源清理，不再写入会话。 -## 11. 逻辑完整性评估 +## 11. 模型请求重试 + +```text +attempt + ├─ success → completed + ├─ cancel → cancelled + ├─ non-retryable failure → failed + ├─ transient failure after visible output → failed_with_partial_output + └─ transient failure before visible output + ├─ retry count < 3 → backoff → attempt + └─ retry count = 3 → failed +``` + +- 瞬时失败包括常见连接重置、连接/网络不可达、DNS 临时失败、连接管道中断和请求超时。 +- 三次重试分别等待 500 ms、1 s 和 2 s；每次尝试重新建立请求并使用独立请求超时。 +- 普通问答、工具模型轮次和上下文摘要使用同一规则。 +- 重试不重复已经执行的工具；工具只会在模型完整返回一轮工具调用后开始执行。 +- 退避等待绑定父请求取消信号，取消优先于启动下一次尝试。 + +## 12. 逻辑完整性评估 | 领域 | 结论 | | --- | --- | @@ -167,6 +189,7 @@ type SubagentDelegateResult = { | 本机三平台 | 完整，统一契约和平台 Shell 决策已定义 | | Subagent 递归 | 完整，深度固定为 1 | | 取消与失败 | 完整，父子和进程传播及优先级已定义 | +| 模型网络恢复 | 完整，有限重试、输出重放和取消边界已定义 | | 产品对象 | 完整，遵循 Task/Job 权威模型 | | SSH 远端执行 | 明确不在首版；未来必须增加 Agent 后端后才能开放 | diff --git a/docs/features/direct-model-agent/prd.md b/docs/features/direct-model-agent/prd.md index 0b77bd70..4aeafbfc 100644 --- a/docs/features/direct-model-agent/prd.md +++ b/docs/features/direct-model-agent/prd.md @@ -5,8 +5,8 @@ | 项目 | 内容 | | --- | --- | | 状态 | 已实施，跨平台 CI 待验证 | -| 版本 | 0.2 | -| 日期 | 2026-09-01 | +| 版本 | 0.3 | +| 日期 | 2026-09-03 | | 适用产品 | GoodBuddy 桌面端 | | 目标平台 | Windows、macOS、Linux，x64 与 arm64 | | 关联领域 | [Task 与 Job 统一领域模型](../task-and-job/task-and-job-model.md) | @@ -169,6 +169,14 @@ Ask 禁止命令和写入。Execute 是用户对当前执行空间账号的完 - 不改变现有 Agent Runtime 的工具清单或权限语义。 - 不修改系统 PATH、Shell Profile 或用户终端会话。 +### FR-11 模型请求重试 + +- 直连文本模型的普通问答、工具轮次和上下文摘要遇到瞬时网络错误或单次请求超时时， + 在尚未显示文本或推理内容的前提下最多自动重试 3 次。 +- 重试使用递增等待时间并显示当前次数；用户取消必须立即终止等待和后续请求。 +- 一旦本次模型调用已经显示文本或推理内容，不自动重放该调用，避免重复回答。 +- 非网络类 Provider 错误、协议错误和用户取消不进入自动重试。 + ## 6. 产品验收 - [ ] Windows 直连模型 Execute 使用 PowerShell 创建文件、运行测试并读取退出码。 @@ -182,4 +190,6 @@ Ask 禁止命令和写入。Execute 是用户对当前执行空间账号的完 - [ ] 取消父请求和退出应用后没有残留 GoodBuddy 命令进程。 - [ ] 本机工具环境选择对新命令生效，Main 持有的模型凭据不进入子进程。 - [ ] 子任务不进入 Task Center，不创建独立可导航 Conversation。 +- [ ] TLS 建连重置和请求超时在未输出内容时最多自动重试 3 次，成功后继续原请求。 +- [ ] 重试等待可取消，已显示部分内容的失败不自动重放。 - [ ] 全量测试、类型检查、Lint 和生产构建通过。 diff --git a/docs/features/direct-model-agent/progress.md b/docs/features/direct-model-agent/progress.md index a7dd9bcb..fb5bc694 100644 --- a/docs/features/direct-model-agent/progress.md +++ b/docs/features/direct-model-agent/progress.md @@ -6,8 +6,8 @@ | --- | --- | | 文档类型 | 功能进度 | | 状态 | 已实施，本机验收完成 | -| 版本 | 0.2 | -| 日期 | 2026-09-02 | +| 版本 | 0.3 | +| 日期 | 2026-09-03 | | 关联入口 | [直连模型 Agent 能力](./README.md) | ## 当前结论 @@ -55,6 +55,8 @@ - [x] Ask/Execute、Runtime target 和本机执行空间双重检查。 - [x] DeepSeek Harness、OpenCode 和 Continue 隔离。 - [x] 通用工具活动和“设置 > MCP > 直连模型”编程能力目录。 +- [x] 普通问答、工具轮次和上下文摘要的瞬时网络错误及请求超时最多自动重试 3 次。 +- [x] 递增退避、重试状态、取消传播和已显示输出不重放。 ### 阶段 3：编程 Subagent @@ -104,7 +106,19 @@ 0、委派编程 Subagent 再次运行最终程序并综合结果；最终源码成功验收运行包含 8 次真实 模型调用。 - BigToken 连接的首次验收在 TLS 建立前被 `ECONNRESET` 中断，未进入模型工具路径；切换 - 到仓库已有的 DeepSeek 连接后完成验证。 + 到仓库已有的 DeepSeek 连接后完成验证。2026-09-03 已针对该故障补充统一模型请求重试。 + +## 2026-09-03 模型请求重试验证证据 + +- `ModelAgentRuntime` 对普通问答、工具轮次和上下文摘要统一处理瞬时 Node/Undici 网络错误 + 与请求超时；首次失败后最多重试 3 次，退避为 500 ms、1 s、2 s。 +- 聚焦测试覆盖嵌套 `fetch failed` → `ECONNRESET` 恢复、工具轮次恢复、四次尝试后终止、 + 退避取消，以及已显示部分流式输出时不重放。 +- `npm test -- --run src/main/agent/model-runtime.test.ts`： + `72 passed, 1 skipped`。 +- 全量 `npm test`：`3484 passed, 51 skipped`。 +- `npm run typecheck`、`npm run lint` 和 `npm run build`：通过。 +- 最小真实直连文本模型验证：`1 passed, 13 skipped`，实际模型调用 1 次。 - 全量 `npm test`：`3410 passed, 50 skipped`。 - `npm run typecheck`：通过。 diff --git a/docs/features/direct-model-agent/technical-design.md b/docs/features/direct-model-agent/technical-design.md index a6be823d..9e97f691 100644 --- a/docs/features/direct-model-agent/technical-design.md +++ b/docs/features/direct-model-agent/technical-design.md @@ -5,8 +5,8 @@ | 项目 | 内容 | | --- | --- | | 状态 | 已实施 | -| 版本 | 0.2 | -| 日期 | 2026-09-01 | +| 版本 | 0.3 | +| 日期 | 2026-09-03 | | 关联 PRD | [PRD](./prd.md) | | 功能逻辑 | [功能逻辑设计](./logic-design.md) | | 相关架构 | [助手工作栏与执行空间](../assistant-workbar/prd.md)、[本机工具环境](../local-tool-environment/technical-design.md) | @@ -319,12 +319,19 @@ type SubagentActor = | Shell 无法启动 | 工具失败 | | cwd 无效或越界 | 工具调用前失败 | | 超时/取消 | 终止进程树，返回已有输出和原因 | +| 直连模型瞬时网络错误/请求超时，尚无可见输出 | 500 ms、1 s、2 s 退避，最多自动重试 3 次 | +| 直连模型失败，已有文本或推理输出 | 保留部分输出并失败，不自动重放 | | 子级模型失败 | 返回 `failed`、部分输出和错误 | | 子级工具失败 | 由子模型处理；最终无法完成时返回子级失败 | | 父请求已终止 | 拒绝新调用，丢弃迟到内容并清理 | 不要用宽泛 catch 抹掉取消、超时、Provider HTTP 状态或进程退出上下文。 +模型请求重试由 `ModelAgentRuntime` 的统一流式请求边界处理，覆盖普通问答、工具轮次和复用 +同一 Runtime 的上下文摘要。重试分类检查错误及有界 `cause` 链中的 Node/Undici 网络错误 +码；每次尝试创建独立超时信号。退避等待监听父请求取消。只有完整模型轮次返回后才执行 +工具，因此未产生可见文本或推理内容的重试不会重复执行工具。 + ## 11. 性能 - Shell 探测按执行服务实例缓存，不在每轮模型调用重复探测。 @@ -347,6 +354,8 @@ type SubagentActor = - 深度 1 过滤、并发 3、队列 20、取消传播和部分输出。 - 编程 Subagent 不创建 Task/Conversation。 - DeepSeek Harness/OpenCode/Continue 不获得本功能工具。 +- TLS 建连 `ECONNRESET`、Undici 连接超时、单次请求超时、三次重试上限和退避取消。 +- 已产生文本或推理增量后发生网络错误时不自动重放。 ### 12.2 本机集成测试 diff --git a/docs/features/direct-model-agent/ui-design.md b/docs/features/direct-model-agent/ui-design.md index c9bb6edc..d6fe3467 100644 --- a/docs/features/direct-model-agent/ui-design.md +++ b/docs/features/direct-model-agent/ui-design.md @@ -5,8 +5,8 @@ | 项目 | 内容 | | --- | --- | | 状态 | 已实施 | -| 版本 | 0.2 | -| 日期 | 2026-09-01 | +| 版本 | 0.3 | +| 日期 | 2026-09-03 | | 设计系统 | [GoodBuddy 统一界面设计系统](../../../UI-DESIGN.md) | | 关联 PRD | [PRD](./prd.md) | | 关联逻辑 | [功能逻辑设计](./logic-design.md) | @@ -117,6 +117,8 @@ GoodBuddy 服务崩溃。 - Runtime 初始化、Shell 探测或全局服务失败可通过应用通知提示，并在设置状态卡提供重试。 - 同一错误不同时显示活动块副本和通知副本。 - 失败后保留用户 Prompt、命令摘要、工作目录和已有输出。 +- 直连模型自动重试通过现有聊天状态显示“模型网络请求失败，正在重试（n/3）”，不增加 + 独立按钮、弹窗或通知副本；用户仍可使用当前停止操作取消请求。 ## 7. 键盘与无障碍 diff --git a/docs/features/direct-model-agent/user-stories.md b/docs/features/direct-model-agent/user-stories.md index 7d2ba751..b6fa8276 100644 --- a/docs/features/direct-model-agent/user-stories.md +++ b/docs/features/direct-model-agent/user-stories.md @@ -5,8 +5,8 @@ | 项目 | 内容 | | --- | --- | | 状态 | 已实施，跨平台 CI 待验证 | -| 版本 | 0.2 | -| 日期 | 2026-09-01 | +| 版本 | 0.3 | +| 日期 | 2026-09-03 | | 关联 PRD | [直连模型 Agent 能力 PRD](./prd.md) | | 关联逻辑 | [功能逻辑设计](./logic-design.md) | @@ -70,6 +70,14 @@ Given 命令仍在运行，When 用户取消父请求，Then 当前 Shell 及其 Given 命令输出超过上限，When 工具返回结果，Then 保留有用的开头和结尾并明确 `truncated: true`，And 模型与用户都不会把结果理解为完整日志。 +### US-A7 从瞬时模型网络故障恢复 + +关联：FR-7、FR-9、FR-11。 + +Given 直连模型在 TLS 建连、网络传输或单次请求超时时失败且尚未显示输出，When 请求仍未 +取消，Then GoodBuddy 显示当前重试次数并使用递增等待时间最多重试 3 次，And 成功后继续 +原问答或工具轮次。Given 已经显示文本或推理内容，Then 不自动重放该调用。 + ## 3. Epic B：编程闭环 ### US-B1 修改后运行验证 diff --git a/src/main/agent/model-runtime.test.ts b/src/main/agent/model-runtime.test.ts index 96883d31..c52b5c57 100644 --- a/src/main/agent/model-runtime.test.ts +++ b/src/main/agent/model-runtime.test.ts @@ -1367,6 +1367,179 @@ describe('ModelAgentRuntime', () => { await expect(stream.next()).rejects.toThrow('upstream unavailable') }) + it('retries a nested ECONNRESET before streaming output', async () => { + vi.useFakeTimers() + try { + const connectionReset = Object.assign( + new Error( + 'Client network socket disconnected before secure TLS connection was established' + ), + { code: 'ECONNRESET', port: 443 } + ) + const fetcher = vi + .fn() + .mockRejectedValueOnce( + new TypeError('fetch failed', { cause: connectionReset }) + ) + .mockResolvedValueOnce( + new Response(createEventStream('重试后成功'), { + status: 200, + headers: { 'content-type': 'text/event-stream' } + }) + ) + const runtime = new ModelAgentRuntime({ + apiKey: 'test-key', + baseUrl: 'https://bigtoken.ai', + model: 'sonnet-5', + protocol: 'anthropic-messages', + authentication: 'api-key', + fetcher + }) + const events: RuntimeEvent[] = [] + const consume = async (): Promise => { + for await (const event of runtime.run( + { + requestId: crypto.randomUUID(), + conversationId: crypto.randomUUID(), + prompt: 'test' + }, + new AbortController().signal + )) { + events.push(event) + } + } + const completed = consume() + + await vi.runAllTimersAsync() + await completed + + expect(fetcher).toHaveBeenCalledTimes(2) + expect(events).toContainEqual( + expect.objectContaining({ + type: 'status', + message: '模型网络请求失败，正在重试（1/3）' + }) + ) + expect(events).toContainEqual( + expect.objectContaining({ + type: 'text', + delta: '重试后成功' + }) + ) + expect(events.at(-1)).toMatchObject({ type: 'done' }) + } finally { + vi.useRealTimers() + } + }) + + it('does not retry a transient stream failure after publishing output', async () => { + const connectionReset = Object.assign( + new Error('socket disconnected'), + { code: 'ECONNRESET' } + ) + let sentOutput = false + const body = new ReadableStream({ + pull(controller) { + if (sentOutput) { + controller.error(connectionReset) + return + } + sentOutput = true + controller.enqueue( + new TextEncoder().encode( + [ + 'event: content_block_delta', + 'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"已经显示"}}', + '', + '' + ].join('\n') + ) + ) + } + }) + const fetcher = vi.fn(async () => + new Response(body, { + status: 200, + headers: { 'content-type': 'text/event-stream' } + }) + ) + const runtime = new ModelAgentRuntime({ + apiKey: 'test-key', + baseUrl: 'https://bigtoken.ai', + model: 'sonnet-5', + protocol: 'anthropic-messages', + authentication: 'api-key', + fetcher + }) + const events: RuntimeEvent[] = [] + const consume = async (): Promise => { + for await (const event of runtime.run( + { + requestId: crypto.randomUUID(), + conversationId: crypto.randomUUID(), + prompt: 'test' + }, + new AbortController().signal + )) { + events.push(event) + } + } + + await expect(consume()).rejects.toThrow('socket disconnected') + expect(fetcher).toHaveBeenCalledOnce() + expect(events).toContainEqual( + expect.objectContaining({ + type: 'text', + delta: '已经显示' + }) + ) + expect(events).not.toContainEqual( + expect.objectContaining({ + message: expect.stringContaining('正在重试') + }) + ) + }) + + it('cancels immediately during model retry backoff', async () => { + const controller = new AbortController() + const fetcher = vi.fn(async () => { + throw Object.assign(new Error('socket disconnected'), { + code: 'ECONNRESET' + }) + }) + const runtime = new ModelAgentRuntime({ + apiKey: 'test-key', + baseUrl: 'https://bigtoken.ai', + model: 'sonnet-5', + protocol: 'anthropic-messages', + authentication: 'api-key', + fetcher + }) + const stream = runtime.run( + { + requestId: crypto.randomUUID(), + conversationId: crypto.randomUUID(), + prompt: 'test' + }, + controller.signal + ) + + await expect(stream.next()).resolves.toMatchObject({ + value: { type: 'status', message: 'sonnet-5 正在思考' } + }) + await expect(stream.next()).resolves.toMatchObject({ + value: { + type: 'status', + message: '模型网络请求失败，正在重试（1/3）' + } + }) + const pendingRetry = stream.next() + controller.abort(new Error('用户取消')) + + await expect(pendingRetry).rejects.toThrow('用户取消') + expect(fetcher).toHaveBeenCalledOnce() + }) + it('parses CRLF event separators split across response chunks', async () => { const payload = createEventStream('split CRLF').replaceAll('\n', '\r\n') const splitAt = payload.indexOf('\r\n\r\n') + 3 @@ -1418,6 +1591,16 @@ describe('ModelAgentRuntime', () => { it('aborts a model request that exceeds the runtime timeout', async () => { vi.useFakeTimers() try { + const fetcher = vi.fn( + async (_input, init) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener( + 'abort', + () => reject(init.signal?.reason), + { once: true } + ) + }) + ) const runtime = new ModelAgentRuntime({ apiKey: 'test-key', baseUrl: 'https://bigtoken.ai', @@ -1425,35 +1608,27 @@ describe('ModelAgentRuntime', () => { protocol: 'anthropic-messages', authentication: 'api-key', requestTimeoutMs: 50, - fetcher: vi.fn( - async (_input, init) => - new Promise((_resolve, reject) => { - init?.signal?.addEventListener( - 'abort', - () => reject(init.signal?.reason), - { once: true } - ) - }) - ) + fetcher }) - const stream = runtime.run( - { - requestId: crypto.randomUUID(), - conversationId: crypto.randomUUID(), - prompt: 'test' - }, - new AbortController().signal - ) - await expect(stream.next()).resolves.toMatchObject({ - value: { type: 'status' } - }) - const result = stream.next() - const assertion = expect(result).rejects.toThrow( + const consume = async (): Promise => { + for await (const _event of runtime.run( + { + requestId: crypto.randomUUID(), + conversationId: crypto.randomUUID(), + prompt: 'test' + }, + new AbortController().signal + )) { + void _event + } + } + const assertion = expect(consume()).rejects.toThrow( '模型接口请求超时' ) - await vi.advanceTimersByTimeAsync(50) + await vi.runAllTimersAsync() await assertion + expect(fetcher).toHaveBeenCalledTimes(4) } finally { vi.useRealTimers() } @@ -1489,23 +1664,23 @@ describe('ModelAgentRuntime', () => { ) }) }) - const stream = runtime.run( - { - requestId: crypto.randomUUID(), - conversationId: crypto.randomUUID(), - prompt: 'test' - }, - new AbortController().signal - ) - await expect(stream.next()).resolves.toMatchObject({ - value: { type: 'status' } - }) - const result = stream.next() - const assertion = expect(result).rejects.toThrow( + const consume = async (): Promise => { + for await (const _event of runtime.run( + { + requestId: crypto.randomUUID(), + conversationId: crypto.randomUUID(), + prompt: 'test' + }, + new AbortController().signal + )) { + void _event + } + } + const assertion = expect(consume()).rejects.toThrow( '模型接口请求超时' ) - await vi.advanceTimersByTimeAsync(50) + await vi.runAllTimersAsync() await assertion expect(responseSignal?.aborted).toBe(true) } finally { @@ -1693,6 +1868,75 @@ describe('ModelAgentRuntime', () => { expect(toolProvider.listTools).not.toHaveBeenCalled() }) + it('retries transient network failures for direct-model tool rounds', async () => { + vi.useFakeTimers() + try { + const fetcher = vi + .fn() + .mockRejectedValueOnce( + Object.assign(new Error('connect timeout'), { + code: 'UND_ERR_CONNECT_TIMEOUT' + }) + ) + .mockResolvedValueOnce( + Response.json({ + choices: [ + { + message: { + role: 'assistant', + content: '工具轮次重试成功' + } + } + ] + }) + ) + const runtime = new ModelAgentRuntime({ + baseUrl: 'http://127.0.0.1:11434/v1', + model: 'qwen3', + protocol: 'openai-chat-completions', + authentication: 'none', + fetcher, + toolProvider: createToolProvider() + }) + const events: RuntimeEvent[] = [] + const consume = async (): Promise => { + for await (const event of runtime.run( + { + requestId: crypto.randomUUID(), + conversationId: crypto.randomUUID(), + prompt: '使用工具能力回答', + workMode: 'execute' + }, + new AbortController().signal, + async () => 'once' + )) { + events.push(event) + } + } + const completed = consume() + + await vi.runAllTimersAsync() + await completed + + expect(fetcher).toHaveBeenCalledTimes(2) + expect(events).toContainEqual( + expect.objectContaining({ + type: 'status', + message: '模型网络请求失败，正在重试（1/3）' + }) + ) + expect(events).toContainEqual( + expect.objectContaining({ + type: 'text', + delta: '工具轮次重试成功' + }) + ) + expect(events.at(-1)).toMatchObject({ type: 'done' }) + } finally { + vi.useRealTimers() + } + }) + it('streams OpenAI-compatible reasoning deltas before the answer', async () => { const stream = [ `data: ${JSON.stringify({ diff --git a/src/main/agent/model-runtime.ts b/src/main/agent/model-runtime.ts index 34a4d8f5..913a785c 100644 --- a/src/main/agent/model-runtime.ts +++ b/src/main/agent/model-runtime.ts @@ -188,6 +188,24 @@ const maxStreamBlockBytes = 1024 * 1024 const maxToolArgumentBytes = 128 * 1024 const maxToolContextBytes = 1024 * 1024 const defaultModelRequestTimeoutMs = 10 * 60_000 +const modelRequestRetryDelaysMs = [500, 1_000, 2_000] as const +const retryableModelNetworkErrorCodes = new Set([ + 'EAI_AGAIN', + 'ECONNABORTED', + 'ECONNREFUSED', + 'ECONNRESET', + 'EHOSTDOWN', + 'EHOSTUNREACH', + 'ENETDOWN', + 'ENETRESET', + 'ENETUNREACH', + 'EPIPE', + 'ETIMEDOUT', + 'UND_ERR_BODY_TIMEOUT', + 'UND_ERR_CONNECT_TIMEOUT', + 'UND_ERR_HEADERS_TIMEOUT', + 'UND_ERR_SOCKET' +]) const noModelTools: ModelToolProviderLike = { listTools: async () => [], @@ -573,6 +591,72 @@ function normalizeRequestError( throw error } +function isRetryableModelRequestError(error: unknown): boolean { + const seen = new Set() + let current = error + for (let depth = 0; depth < 8; depth += 1) { + if ( + current === null || + (typeof current !== 'object' && + typeof current !== 'function') || + seen.has(current) + ) { + break + } + seen.add(current) + const record = current as Record + if ( + typeof record.code === 'string' && + retryableModelNetworkErrorCodes.has( + record.code.toLocaleUpperCase() + ) + ) { + return true + } + if (typeof record.message === 'string') { + if ( + record.message === '模型接口请求超时' || + /(?:fetch failed|failed to fetch|network(?: request)? (?:error|failed)|socket disconnected|connection (?:closed|reset)|\bECONNRESET\b|\bETIMEDOUT\b)/iu.test( + record.message + ) + ) { + return true + } + } + current = record.cause + } + return false +} + +function waitForModelRetry( + delayMs: number, + signal: AbortSignal +): Promise { + signal.throwIfAborted() + return new Promise((resolve, reject) => { + let settled = false + const finish = (callback: () => void): void => { + if (settled) { + return + } + settled = true + clearTimeout(timeout) + signal.removeEventListener('abort', abort) + callback() + } + const abort = (): void => { + finish(() => reject(signal.reason)) + } + const timeout = setTimeout(() => { + finish(resolve) + }, delayMs) + signal.addEventListener('abort', abort, { once: true }) + if (signal.aborted) { + abort() + } + }) +} + function parseGeneratedImage(value: unknown): { data: string mimeType: 'image/png' | 'image/jpeg' | 'image/webp' @@ -1734,6 +1818,71 @@ export class ModelAgentRuntime implements AgentRuntime { } } + private async *retryModelStream( + createAttempt: () => AsyncGenerator, + signal: AbortSignal, + requestId: string + ): AsyncGenerator { + for ( + let retryCount = 0; + retryCount <= modelRequestRetryDelaysMs.length; + retryCount += 1 + ) { + signal.throwIfAborted() + const attempt = createAttempt() + let step: IteratorResult | undefined + let outputPublished = false + let failed = false + let failure: unknown + try { + step = await attempt.next() + while (!step.done) { + if ( + step.value.type === 'text' || + step.value.type === 'reasoning' + ) { + outputPublished = true + } + yield step.value + step = await attempt.next() + } + } catch (error) { + failed = true + failure = error + } finally { + if (!step?.done) { + await attempt.return(undefined as T).catch(() => undefined) + } + } + if (!failed) { + if (!step?.done) { + throw new Error('模型请求流未返回最终结果') + } + return step.value + } + if ( + outputPublished || + signal.aborted || + retryCount >= modelRequestRetryDelaysMs.length || + !isRetryableModelRequestError(failure) + ) { + throw failure + } + yield { + requestId, + type: 'status', + message: + `模型网络请求失败，正在重试（${retryCount + 1}/` + + `${modelRequestRetryDelaysMs.length}）` + } + await waitForModelRetry( + modelRequestRetryDelaysMs[retryCount]!, + signal + ) + } + throw new Error('模型请求重试状态无效') + } + async getStatus(): Promise { const imageGeneration = this.capability === 'image-generation' return { @@ -2577,6 +2726,29 @@ export class ModelAgentRuntime implements AgentRuntime { anthropic: boolean, signal: AbortSignal, requestId: string + ): AsyncGenerator { + return yield* this.retryModelStream( + () => + this.requestToolModelOnce( + messages, + tools, + system, + anthropic, + signal, + requestId + ), + signal, + requestId + ) + } + + private async *requestToolModelOnce( + messages: Array>, + tools: ModelToolDefinition[], + system: string, + anthropic: boolean, + signal: AbortSignal, + requestId: string ): AsyncGenerator { const responses = this.options.protocol === 'openai-responses' const streamOpenAIChat = !responses && !anthropic @@ -2867,6 +3039,141 @@ export class ModelAgentRuntime implements AgentRuntime { } } + private async *requestTextModelOnce( + request: AgentExecutionRequest, + signal: AbortSignal, + system: string, + messages: + | AnthropicApiMessage[] + | Array> + ): AsyncGenerator< + RuntimeEvent, + { answer: string; usage: ModelUsageAccumulator }, + void + > { + const anthropic = this.options.protocol === 'anthropic-messages' + const responses = this.options.protocol === 'openai-responses' + const modelRequest = await this.fetchWithTimeout( + this.getEndpoint(), + { + method: 'POST', + headers: this.getHeaders(), + body: JSON.stringify( + responses + ? { + model: this.options.model, + stream: true, + instructions: system, + input: messages + } + : anthropic + ? { + model: this.options.model, + max_tokens: this.anthropicProtocolMaximumTokens, + stream: true, + system, + messages + } + : { + model: this.options.model, + stream: true, + stream_options: { + include_usage: true + }, + messages + } + ) + }, + signal + ) + const response = modelRequest.response + let answer = '' + const usage = { + reported: false + } satisfies ModelUsageAccumulator + try { + if (!response.ok) { + const responseText = await readBoundedResponseText(response, { + maxBytes: 128 * 1024, + missingBodyMessage: '模型接口未返回响应内容', + tooLargeMessage: '模型接口响应超过安全限制' + }) + let detail: string | undefined + try { + detail = getErrorMessage( + responseText.trim() + ? JSON.parse(responseText) + : undefined + ) + } catch { + detail = undefined + } + throw new Error( + detail ?? `模型接口请求失败（HTTP ${response.status}）` + ) + } + + let receivedStop = false + for await (const block of readBoundedSseBlocks(response)) { + const parsed = parseStreamBlock(block, this.options.protocol) + if (parsed.usage) { + applyUsageUpdate(usage, parsed.usage) + } + if (parsed.reasoningDelta) { + yield { + requestId: request.requestId, + type: 'reasoning', + delta: parsed.reasoningDelta + } + } + const { delta } = parsed + if (delta) { + answer += delta + yield { + requestId: request.requestId, + type: 'text', + delta + } + } + if (parsed.stopped) { + receivedStop = true + break + } + } + if (!receivedStop) { + throw new Error('模型接口流式响应意外中断') + } + if (!answer) { + throw new Error('模型接口返回了空内容') + } + return { answer, usage } + } catch (error) { + return normalizeRequestError(error, modelRequest.timedOut()) + } finally { + modelRequest.clear() + } + } + + private async *requestTextModel( + request: AgentExecutionRequest, + signal: AbortSignal, + system: string, + messages: + | AnthropicApiMessage[] + | Array> + ): AsyncGenerator< + RuntimeEvent, + { answer: string; usage: ModelUsageAccumulator }, + void + > { + return yield* this.retryModelStream( + () => + this.requestTextModelOnce(request, signal, system, messages), + signal, + request.requestId + ) + } + private async *compactAgentRunContext( request: AgentExecutionRequest, baseMessages: readonly Record[], @@ -3700,152 +4007,57 @@ export class ModelAgentRuntime implements AgentRuntime { [], anthropic || responses ? system : '' ) - const modelRequest = await this.fetchWithTimeout( - this.getEndpoint(), + const modelResponse = yield* this.requestTextModel( + request, + signal, + system, + messages + ) + const { answer, usage } = modelResponse + const completedHistory = [ + ...(identifiedRequest.history ?? + this.conversations.get(request.conversationId) ?? + []), { - method: 'POST', - headers: this.getHeaders(), - body: JSON.stringify( - responses - ? { - model: this.options.model, - stream: true, - instructions: system, - input: messages - } - : anthropic - ? { - model: this.options.model, - max_tokens: this.anthropicProtocolMaximumTokens, - stream: true, - system, - messages - } - : { - model: this.options.model, - stream: true, - stream_options: { - include_usage: true - }, - messages - } - ) + id: request.currentUserMessageId, + role: 'user', + content: request.prompt }, + { + id: request.currentAssistantMessageId, + role: 'assistant', + content: answer + } + ] satisfies ConversationMessage[] + this.saveConversation(request.conversationId, completedHistory) + + const usageEvent = createUsageEvent( + request.requestId, + anthropic ? 'anthropic' : 'openai', + this.options.model, + usage + ) + const contextMetricsEvent = this.createContextMetricsEvent( + request.requestId, + usage, + estimatedRequestTokens + estimateTextTokens(answer) + ) + if (contextMetricsEvent) { + yield contextMetricsEvent + } + if (usageEvent) { + yield usageEvent + } + yield* this.finalizeConversationContext( + identifiedRequest, + completedHistory, + contextMetricsEvent?.contextTokens ?? + estimatedRequestTokens + estimateTextTokens(answer), signal ) - const response = modelRequest.response - let answer = '' - const usage = { - reported: false - } satisfies ModelUsageAccumulator - try { - if (!response.ok) { - const responseText = await readBoundedResponseText(response, { - maxBytes: 128 * 1024, - missingBodyMessage: '模型接口未返回响应内容', - tooLargeMessage: '模型接口响应超过安全限制' - }) - let detail: string | undefined - try { - detail = getErrorMessage( - responseText.trim() - ? JSON.parse(responseText) - : undefined - ) - } catch { - detail = undefined - } - throw new Error( - detail ?? `模型接口请求失败（HTTP ${response.status}）` - ) - } - - let receivedStop = false - - for await (const block of readBoundedSseBlocks(response)) { - const parsed = parseStreamBlock(block, this.options.protocol) - if (parsed.usage) { - applyUsageUpdate(usage, parsed.usage) - } - if (parsed.reasoningDelta) { - yield { - requestId: request.requestId, - type: 'reasoning', - delta: parsed.reasoningDelta - } - } - const { delta } = parsed - if (delta) { - answer += delta - yield { - requestId: request.requestId, - type: 'text', - delta - } - } - - if (parsed.stopped) { - receivedStop = true - break - } - } - - if (!receivedStop) { - throw new Error('模型接口流式响应意外中断') - } - if (!answer) { - throw new Error('模型接口返回了空内容') - } - - const completedHistory = [ - ...(identifiedRequest.history ?? - this.conversations.get(request.conversationId) ?? - []), - { - id: request.currentUserMessageId, - role: 'user', - content: request.prompt - }, - { - id: request.currentAssistantMessageId, - role: 'assistant', - content: answer - } - ] satisfies ConversationMessage[] - this.saveConversation(request.conversationId, completedHistory) - - const usageEvent = createUsageEvent( - request.requestId, - anthropic ? 'anthropic' : 'openai', - this.options.model, - usage - ) - const contextMetricsEvent = this.createContextMetricsEvent( - request.requestId, - usage, - estimatedRequestTokens + estimateTextTokens(answer) - ) - if (contextMetricsEvent) { - yield contextMetricsEvent - } - if (usageEvent) { - yield usageEvent - } - yield* this.finalizeConversationContext( - identifiedRequest, - completedHistory, - contextMetricsEvent?.contextTokens ?? - estimatedRequestTokens + estimateTextTokens(answer), - signal - ) - yield { - requestId: request.requestId, - type: 'done' - } - } catch (error) { - return normalizeRequestError(error, modelRequest.timedOut()) - } finally { - modelRequest.clear() + yield { + requestId: request.requestId, + type: 'done' } }