streaming.mdx 7.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182
  1. ---
  2. title: "流式响应:逐字呈现"
  3. description: "为什么 Claude Code 的回答是'打字机效果'而不是一整块弹出"
  4. ---
  5. ## 为什么需要流式
  6. 想象 AI 需要 30 秒才能生成完整回答——如果等 30 秒后才一次性显示,用户体验是灾难性的。
  7. 流式响应让用户**实时看到 AI 的思考过程**:
  8. - 文字逐字出现,用户能提前判断方向是否正确
  9. - 工具调用的参数在生成过程中就能预览
  10. - 长时间任务不会让用户觉得"卡死了"
  11. ## `BetaRawMessageStreamEvent` 核心事件类型
  12. 流式 API 返回的是一系列 `BetaRawMessageStreamEvent`,每种事件类型对应流式响应的不同阶段(`src/services/api/claude.ts`):
  13. ```
  14. message_start ← 消息开始,包含 model、usage 初始值
  15. ├── content_block_start ← 内容块开始(text / tool_use / thinking)
  16. │ ├── content_block_delta ← 增量数据(text_delta / input_json_delta / thinking_delta)
  17. │ ├── content_block_delta ← ... 持续到达
  18. │ └── content_block_stop ← 内容块结束,yield AssistantMessage
  19. ├── content_block_start ← 下一个内容块...
  20. │ └── ...
  21. └── message_delta ← stop_reason + 最终 usage
  22. message_stop ← 消息结束
  23. ```
  24. ### 事件处理状态机
  25. `src/services/api/claude.ts:1980-2298` 实现了一个基于 `switch(part.type)` 的状态机:
  26. | 事件类型 | 处理逻辑 | 状态变更 |
  27. |----------|----------|----------|
  28. | `message_start` | 初始化 `partialMessage`,记录 TTFT(首字节延迟) | `usage` 初始化 |
  29. | `content_block_start` | 按 `part.index` 创建对应类型的内容块 | `contentBlocks[index]` 初始化 |
  30. | `content_block_delta` | 按子类型增量追加数据 | text / thinking / input 累加 |
  31. | `content_block_stop` | 构建完整 `AssistantMessage` 并 yield | 消息推入 `newMessages` |
  32. | `message_delta` | 更新 stop_reason 和最终 usage | 写回最后一条消息 |
  33. | `message_stop` | 无操作(流结束标记) | — |
  34. ### 内容块类型及其增量数据
  35. `content_block_start` 中的 `content_block.type` 决定了如何处理后续 delta:
  36. | 内容块类型 | Delta 类型 | 累加逻辑 |
  37. |-----------|-----------|----------|
  38. | `text` | `text_delta` | `text += delta.text` |
  39. | `thinking` | `thinking_delta` + `signature_delta` | `thinking += delta.thinking`,`signature = delta.signature` |
  40. | `tool_use` | `input_json_delta` | `input += delta.partial_json`(JSON 字符串增量拼接) |
  41. | `server_tool_use` | `input_json_delta` | 同 tool_use |
  42. | `connector_text` | `connector_text_delta` | 特殊连接器文本(feature flag 控制) |
  43. 关键设计:`content_block_start` 时所有文本字段初始化为空字符串,只通过 `content_block_delta` 累加。这是因为 SDK 有时在 start 和 delta 中重复发送相同文本。
  44. ## 文本 chunk 和 tool_use block 的交织
  45. 一次 AI 响应可能包含多个内容块,交替出现:
  46. ```
  47. content_block_start (text, index=0) "我来帮你修复这个 bug。"
  48. content_block_delta (text_delta) "首先..."
  49. content_block_stop (index=0)
  50. content_block_start (tool_use, index=1) { name: "Read", input: "..." }
  51. content_block_delta (input_json_delta) '{"file_p' → 'ath":' → '"src/foo.ts"}'
  52. content_block_stop (index=1)
  53. content_block_start (text, index=2) "我已经看到了问题所在..."
  54. content_block_stop (index=2)
  55. ```
  56. 每个 `content_block_stop` 触发一次 `yield`,将完整的 AssistantMessage 推送给消费者。这意味着一个 AI 响应会产生**多条** `AssistantMessage`——文本消息和工具调用消息交替产出。
  57. `stop_reason` 要等到 `message_delta` 才确定(可能是 `end_turn`、`tool_use`、`max_tokens` 等),所以最后一条消息的 `stop_reason` 是**回写**的:
  58. ```typescript
  59. // claude.ts:2246 — 直接属性修改,不用对象替换
  60. // 因为 transcript 写队列持有 message.message 的引用
  61. const lastMsg = newMessages.at(-1)
  62. if (lastMsg) {
  63. lastMsg.message.usage = usage
  64. lastMsg.message.stop_reason = stopReason
  65. }
  66. ```
  67. ## 流式中的错误处理
  68. ### 网络断开
  69. 流式连接依赖 SSE(Server-Sent Events)。当连接中断时:
  70. 1. **Stream idle watchdog**:定时检测事件间隔,超过阈值(stall)触发告警和重试
  71. 2. **Stream abort**:如果 watchdog 检测到长时间无事件,抛出错误进入重试流程
  72. 3. **非流式降级**:作为最后手段,回退到非流式请求(一次性获取完整响应)
  73. ```typescript
  74. // claude.ts:2338-2355 — 检测空流
  75. // 1. 完全没有事件 → 代理返回了非 SSE 响应
  76. // 2. 有 message_start 但没有 content_block_stop → 流被截断
  77. ```
  78. ### API 限流
  79. 当 API 返回限流错误时,系统使用 `withRetry` 包装器进行指数退避重试。重试逻辑考虑了:
  80. - 错误类型(429 限流 vs 500 服务器错误)
  81. - 重试次数上限
  82. - 退避间隔
  83. ### Token 超限
  84. 两种 token 超限场景有不同的处理:
  85. | 场景 | stop_reason | 处理方式 |
  86. |------|------------|----------|
  87. | **输出超限** | `max_tokens` | 生成错误消息,建议设置 `CLAUDE_CODE_MAX_OUTPUT_TOKENS` |
  88. | **上下文窗口超限** | `model_context_window_exceeded` | 触发 compaction 压缩对话历史后重试 |
  89. ```typescript
  90. // claude.ts:2267-2293
  91. if (stopReason === 'max_tokens') {
  92. yield createAssistantAPIErrorMessage({ error: 'max_output_tokens', ... })
  93. }
  94. if (stopReason === 'model_context_window_exceeded') {
  95. // 复用 max_output_tokens 的恢复路径
  96. yield createAssistantAPIErrorMessage({ error: 'max_output_tokens', ... })
  97. }
  98. ```
  99. ### 流式停滞检测
  100. 系统持续监控事件到达间隔,检测"停滞"(stall):
  101. ```typescript
  102. // claude.ts:1940-1966
  103. const STALL_THRESHOLD_MS = 10_000 // 10 秒无事件视为停滞
  104. if (timeSinceLastEvent > STALL_THRESHOLD_MS) {
  105. stallCount++
  106. totalStallTime += timeSinceLastEvent
  107. logEvent('tengu_streaming_stall', { stall_duration_ms, stall_count, ... })
  108. }
  109. ```
  110. 多个 stall 累积后,watchdog 可能决定中断流并触发重试。
  111. ## 工具执行的流式反馈
  112. BashTool 的命令执行也是流式的——通过 `onProgress` 回调逐行推送输出:
  113. ```
  114. BashTool.call() → runShellCommand() → AsyncGenerator
  115. ├── 每秒轮询输出文件 → onProgress(lastLines, allLines, ...)
  116. ├── yield { type: 'progress', output, fullOutput, elapsedTimeSeconds }
  117. └── return { code, stdout, interrupted, ... }
  118. ```
  119. UI 层通过 `useToolCallProgress` hook 实时展示命令输出,而不是等命令完全结束。长时间运行的命令还支持自动后台化(`shouldAutoBackground`)。
  120. ## 多 Provider 适配
  121. | Provider | 流式协议 | 特殊处理 |
  122. |----------|----------|----------|
  123. | **Anthropic Direct** | 原生 SSE | 延迟最低,TTFT 最快 |
  124. | **AWS Bedrock** | AWS SDK 流式接口 | 需要额外的 beta header 和认证 |
  125. | **Google Vertex** | gRPC → 事件流 | 通过 `getMergedBetas()` 适配 |
  126. | **Azure** | Anthropic 兼容 API | 自定义 base URL |
  127. 所有 Provider 通过统一的 `Stream<BetaRawMessageStreamEvent>` 抽象层屏蔽差异。上层代码(QueryEngine、REPL)不需要关心底层用的是哪个 Provider。
  128. ### Provider 选择
  129. `src/utils/model/providers.ts` 中的 `getAPIProvider()` 根据配置决定使用哪个 Provider:
  130. ```typescript
  131. // 根据 api_provider 配置选择:
  132. // "anthropic" → 直连
  133. // "bedrock" → AWS SDK
  134. // "vertex" → Google SDK
  135. // 第三方 base URL → 自动检测
  136. ```
  137. 每个 Provider 需要适配的细节包括:认证方式、beta header、请求参数格式、错误码映射——但这些差异在 `claude.ts` 的 `queryStream()` 函数中被统一处理。