mirror of
https://github.com/claude-code-best/claude-code.git
synced 2026-06-15 12:55:51 +00:00
docs: 重写流式响应,移除 API 事件类型枚举和代码片段
从核心设计选择讲起(流式不只是打字机效果), 将错误处理重组为三层防护体系, 增加 Provider 适配策略对比表和设计考量分析。 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,192 +1,111 @@
|
||||
---
|
||||
title: "流式响应机制 - Claude Code 打字机效果原理"
|
||||
description: "解析 Claude Code 流式响应实现:如何通过 SSE 逐 token 接收 AI 输出,实现实时打字机效果,提升用户等待体验。"
|
||||
keywords: ["流式响应", "SSE", "streaming", "实时输出", "API streaming"]
|
||||
sourceRef: "3ec5675 (2026-04-08)"
|
||||
title: "流式响应"
|
||||
description: "为什么流式是 Claude Code 的核心设计选择?理解流式传输的设计考量、错误处理策略和多 Provider 适配。"
|
||||
keywords: ["流式响应", "SSE", "streaming", "API streaming"]
|
||||
---
|
||||
|
||||
## 为什么需要流式
|
||||
## 为什么流式是核心设计
|
||||
|
||||
想象 AI 需要 30 秒才能生成完整回答——如果等 30 秒后才一次性显示,用户体验是灾难性的。
|
||||
|
||||
流式响应让用户**实时看到 AI 的思考过程**:
|
||||
- 文字逐字出现,用户能提前判断方向是否正确
|
||||
- 工具调用的参数在生成过程中就能预览
|
||||
- 长时间任务不会让用户觉得"卡死了"
|
||||
但流式不仅仅是为了"打字机效果"。在 Claude Code 中,流式是整个系统架构的基础假设:
|
||||
|
||||
## `BetaRawMessageStreamEvent` 核心事件类型
|
||||
- **工具并行执行**:AI 在流式输出过程中就可能发出工具调用,系统可以立即开始执行,不必等整个响应结束。这使得"AI 边想边做"成为可能
|
||||
- **实时反馈**:用户看到 AI 的思考方向后可以提前判断是否正确,必要时提前中断
|
||||
- **可取消性**:流式架构天然支持用户中断——随时可以终止正在进行的流
|
||||
- **工具执行反馈**:不仅是 AI 输出是流式的,工具执行(如 shell 命令)的输出也是流式的——用户实时看到命令输出
|
||||
|
||||
流式 API 返回的是一系列 `BetaRawMessageStreamEvent`,每种事件类型对应流式响应的不同阶段(`src/services/api/claude.ts`):
|
||||
**代价**:流式架构比一次性响应复杂得多——需要处理连接中断、部分数据、乱序事件等边界情况。
|
||||
|
||||
## 流式响应的概念模型
|
||||
|
||||
一次流式响应不是"一个完整的回答",而是一系列按顺序到达的事件流:
|
||||
|
||||
```
|
||||
message_start ← 消息开始,包含 model、usage 初始值
|
||||
├── content_block_start ← 内容块开始(text / tool_use / thinking)
|
||||
│ ├── content_block_delta ← 增量数据(text_delta / input_json_delta / thinking_delta)
|
||||
│ ├── content_block_delta ← ... 持续到达
|
||||
│ └── content_block_stop ← 内容块结束,yield AssistantMessage
|
||||
├── content_block_start ← 下一个内容块...
|
||||
│ └── ...
|
||||
└── message_delta ← stop_reason + 最终 usage
|
||||
message_stop ← 消息结束
|
||||
消息开始
|
||||
├── 内容块 1:文本 "我来帮你修复这个 bug。"
|
||||
├── 内容块 2:工具调用 { name: "Read", input: "..." }
|
||||
├── 内容块 3:文本 "我看到了问题..."
|
||||
└── ...
|
||||
消息结束(包含停止原因和 token 用量)
|
||||
```
|
||||
|
||||
### 事件处理状态机
|
||||
关键设计点:
|
||||
|
||||
`src/services/api/claude.ts` 中 `queryModelWithStreaming()` 函数的事件处理循环实现了一个基于 `switch(part.type)` 的状态机:
|
||||
### 内容块的增量累加
|
||||
|
||||
| 事件类型 | 处理逻辑 | 状态变更 |
|
||||
|----------|----------|----------|
|
||||
| `message_start` | 初始化 `partialMessage`,记录 TTFT(首字节延迟) | `usage` 初始化 |
|
||||
| `content_block_start` | 按 `part.index` 创建对应类型的内容块 | `contentBlocks[index]` 初始化 |
|
||||
| `content_block_delta` | 按子类型增量追加数据 | text / thinking / input 累加 |
|
||||
| `content_block_stop` | 构建完整 `AssistantMessage` 并 yield | 消息推入 `newMessages` |
|
||||
| `message_delta` | 更新 stop_reason 和最终 usage | 写回最后一条消息 |
|
||||
| `message_stop` | 无操作(流结束标记) | — |
|
||||
每个内容块(文本、思考、工具调用)都是通过增量数据逐步构建的。文本逐字到达,工具调用的 JSON 参数逐段到达。系统需要在内存中持续累加这些片段,直到一个内容块完整后才传递给消费者。
|
||||
|
||||
### 内容块类型及其增量数据
|
||||
### 多消息产出
|
||||
|
||||
`content_block_start` 中的 `content_block.type` 决定了如何处理后续 delta:
|
||||
因为一次 AI 响应可能包含多个内容块(文本和工具调用交替出现),每个完整的内容块都会触发一次消息传递。这意味着一个 API 响应会产生**多条消息**——文本消息和工具调用消息交替产出。
|
||||
|
||||
| 内容块类型 | Delta 类型 | 累加逻辑 |
|
||||
|-----------|-----------|----------|
|
||||
| `text` | `text_delta` | `text += delta.text` |
|
||||
| `thinking` | `thinking_delta` + `signature_delta` | `thinking += delta.thinking`,`signature = delta.signature` |
|
||||
| `tool_use` | `input_json_delta` | `input += delta.partial_json`(JSON 字符串增量拼接) |
|
||||
| `server_tool_use` | `input_json_delta` | 同 tool_use |
|
||||
| `connector_text` | `connector_text_delta` | 特殊连接器文本(feature flag 控制) |
|
||||
### 停止原因的回写
|
||||
|
||||
关键设计:`content_block_start` 时所有文本字段初始化为空字符串,只通过 `content_block_delta` 累加。这是因为 SDK 有时在 start 和 delta 中重复发送相同文本。
|
||||
AI 最终是"回答完毕"还是"需要调用工具",这个信息要到最后才知道。所以停止原因是在消息结束时**回写**到最后一条消息上的——消费者在收到中间消息时还不知道整轮对话是否结束。
|
||||
|
||||
## 文本 chunk 和 tool_use block 的交织
|
||||
## 错误处理:三层防护
|
||||
|
||||
一次 AI 响应可能包含多个内容块,交替出现:
|
||||
流式连接比一次性请求脆弱得多——网络波动、服务器过载、连接超时都可能导致中断。系统设计了三层防护:
|
||||
|
||||
```
|
||||
content_block_start (text, index=0) "我来帮你修复这个 bug。"
|
||||
content_block_delta (text_delta) "首先..."
|
||||
content_block_stop (index=0)
|
||||
content_block_start (tool_use, index=1) { name: "Read", input: "..." }
|
||||
content_block_delta (input_json_delta) '{"file_p' → 'ath":' → '"src/foo.ts"}'
|
||||
content_block_stop (index=1)
|
||||
content_block_start (text, index=2) "我已经看到了问题所在..."
|
||||
content_block_stop (index=2)
|
||||
```
|
||||
### 第一层:被动停滞检测
|
||||
|
||||
每个 `content_block_stop` 触发一次 `yield`,将完整的 AssistantMessage 推送给消费者。这意味着一个 AI 响应会产生**多条** `AssistantMessage`——文本消息和工具调用消息交替产出。
|
||||
系统记录每个事件到达的时间间隔。当间隔超过 30 秒时,记录为一次"停滞"并写入遥测。这是被动检测——只在下一个事件到达时才能发现之前的停滞,不会主动中断流。
|
||||
|
||||
`stop_reason` 要等到 `message_delta` 才确定(可能是 `end_turn`、`tool_use`、`max_tokens` 等),所以最后一条消息的 `stop_reason` 是**回写**的:
|
||||
**设计考量**:为什么不立即中断?因为 API 可能在做长时间的计算(如复杂推理),短暂的无响应不一定意味着故障。被动检测提供了观测能力,而不影响正常流程。
|
||||
|
||||
```typescript
|
||||
// claude.ts — stop_reason 回写逻辑(直接属性修改,不用对象替换)
|
||||
// 因为 transcript 写队列持有 message.message 的引用
|
||||
const lastMsg = newMessages.at(-1)
|
||||
if (lastMsg) {
|
||||
lastMsg.message.usage = usage
|
||||
lastMsg.message.stop_reason = stopReason
|
||||
}
|
||||
```
|
||||
### 第二层:主动空闲超时
|
||||
|
||||
## 流式中的错误处理
|
||||
如果 90 秒内没有收到任何事件(可通过环境变量配置),系统主动终止流并进入重试流程。
|
||||
|
||||
### 网络断开
|
||||
**设计考量**:这是兜底机制。真正的故障不能靠被动检测发现——因为被动检测依赖于"下一个事件到达",而如果连接已经死了,下一个事件永远不会到达。
|
||||
|
||||
流式连接依赖 SSE(Server-Sent Events)。当连接中断时,系统有两层检测机制:
|
||||
### 第三层:非流式降级
|
||||
|
||||
1. **被动停滞检测**(`src/services/api/claude.ts` 中 stall 检测逻辑):当下一个事件到达时,计算与上一个事件的时间间隔。超过阈值(30 秒,`STALL_THRESHOLD_MS = 30_000`)记录为一次 stall,累积计数并写入遥测日志。这是被动检测——仅在下一个 chunk 到达时才触发,不会主动中断流。
|
||||
2. **主动空闲超时看门狗**(`src/services/api/claude.ts` 中 `STREAM_IDLE_TIMEOUT_MS` 看门狗逻辑):使用 `setTimeout` 设置 90 秒(可通过 `CLAUDE_STREAM_IDLE_TIMEOUT_MS` 环境变量覆盖)的硬性超时。如果在此期间没有收到任何事件,主动终止流并抛出错误进入重试流程。
|
||||
3. **非流式降级**:作为最后手段,设置 `didFallBackToNonStreaming` 标志,通过 `executeNonStreamingRequest()` 回退到非流式请求(一次性获取完整响应)。
|
||||
作为最后手段,系统可以回退到非流式请求——一次性获取完整响应。失去了实时性,但保证了功能可用性。
|
||||
|
||||
```typescript
|
||||
// claude.ts — 被动停滞检测
|
||||
const STALL_THRESHOLD_MS = 30_000 // 30 秒无事件视为停滞
|
||||
let totalStallTime = 0
|
||||
let stallCount = 0
|
||||
### 降级策略对比
|
||||
|
||||
// claude.ts — 主动空闲超时
|
||||
const STREAM_IDLE_TIMEOUT_MS =
|
||||
parseInt(process.env.CLAUDE_STREAM_IDLE_TIMEOUT_MS || '', 10) || 90_000
|
||||
```
|
||||
| 策略 | 检测方式 | 响应时间 | 用户体验 |
|
||||
|------|----------|----------|----------|
|
||||
| 正常流式 | — | 最低延迟 | 逐字显示 |
|
||||
| 被动停滞检测 | 下一个事件到达时 | 不变 | 无感知 |
|
||||
| 主动超时中断 | 定时器触发 | 中断后重试延迟 | 短暂停顿后恢复 |
|
||||
| 非流式降级 | 重试失败后 | 等待完整响应 | 等待后一次性显示 |
|
||||
|
||||
### API 限流
|
||||
## Token 超限的两种场景
|
||||
|
||||
当 API 返回限流错误时,系统使用 `withRetry` 包装器进行指数退避重试。重试逻辑考虑了:
|
||||
- 错误类型(429 限流 vs 500 服务器错误)
|
||||
- 重试次数上限
|
||||
- 退避间隔
|
||||
两种不同的 token 超限需要不同的处理策略:
|
||||
|
||||
### Token 超限
|
||||
| 场景 | 含义 | 处理方式 |
|
||||
|------|------|----------|
|
||||
| **输出超限** | AI 话说了一半被切断 | 提升输出上限重试,或提示 AI "接着说" |
|
||||
| **上下文窗口超限** | 整个对话历史太长,塞不进 API | 触发自动压缩,用 AI 摘要替代原始对话 |
|
||||
|
||||
两种 token 超限场景有不同的处理:
|
||||
|
||||
| 场景 | stop_reason | 处理方式 |
|
||||
|------|------------|----------|
|
||||
| **输出超限** | `max_tokens` | 生成错误消息,建议设置 `CLAUDE_CODE_MAX_OUTPUT_TOKENS` |
|
||||
| **上下文窗口超限** | `model_context_window_exceeded` | 触发 compaction 压缩对话历史后重试 |
|
||||
|
||||
```typescript
|
||||
// claude.ts — stop_reason 处理
|
||||
if (stopReason === 'max_tokens') {
|
||||
yield createAssistantAPIErrorMessage({ error: 'max_output_tokens', ... })
|
||||
}
|
||||
if (stopReason === 'model_context_window_exceeded') {
|
||||
// 复用 max_output_tokens 的恢复路径
|
||||
yield createAssistantAPIErrorMessage({ error: 'max_output_tokens', ... })
|
||||
}
|
||||
```
|
||||
|
||||
### 流式停滞检测
|
||||
|
||||
系统持续监控事件到达间隔,检测"停滞"(stall):
|
||||
|
||||
```typescript
|
||||
// claude.ts — stall 检测逻辑
|
||||
const STALL_THRESHOLD_MS = 30_000 // 30 秒无事件视为停滞
|
||||
if (timeSinceLastEvent > STALL_THRESHOLD_MS) {
|
||||
stallCount++
|
||||
totalStallTime += timeSinceLastEvent
|
||||
logEvent('tengu_streaming_stall', { stall_duration_ms, stall_count, ... })
|
||||
}
|
||||
```
|
||||
|
||||
这是**被动检测**——仅在下一个 chunk 到达时才触发比较。与之互补的是 90 秒主动空闲超时看门狗(`STREAM_IDLE_TIMEOUT_MS`),会直接中断长时间无响应的流。
|
||||
关键区别:输出超限是"AI 话太多",可以通过调整上限解决;上下文超限是"给 AI 看的东西太多",必须通过压缩或删除来减少。
|
||||
|
||||
## 工具执行的流式反馈
|
||||
|
||||
BashTool 的命令执行也是流式的——通过 `onProgress` 回调逐行推送输出:
|
||||
不仅是 API 响应是流式的,工具执行本身也是流式的。例如 BashTool 执行 shell 命令时,命令的标准输出会实时推送给 UI——用户不需要等命令完全结束才能看到结果。
|
||||
|
||||
```
|
||||
BashTool.call() → runShellCommand() → AsyncGenerator
|
||||
├── 每秒轮询输出文件 → onProgress(lastLines, allLines, ...)
|
||||
├── yield { type: 'progress', output, fullOutput, elapsedTimeSeconds }
|
||||
└── return { code, stdout, interrupted, ... }
|
||||
```
|
||||
|
||||
UI 层通过 `useToolCallProgress` hook 实时展示命令输出,而不是等命令完全结束。长时间运行的命令还支持自动后台化(`shouldAutoBackground`)。
|
||||
长时间运行的命令还支持**自动后台化**:如果命令执行超过一定时间,系统自动将其移到后台,AI 可以继续处理其他任务,命令完成后再回调结果。
|
||||
|
||||
## 多 Provider 适配
|
||||
|
||||
| Provider | 流式协议 | 特殊处理 |
|
||||
|----------|----------|----------|
|
||||
| **firstParty** (Anthropic Direct) | 原生 SSE | 延迟最低,TTFT 最快 |
|
||||
| **AWS Bedrock** | AWS SDK 流式接口 | 需要额外的 beta header 和认证 |
|
||||
| **Google Vertex** | gRPC → 事件流 | 通过 `getMergedBetas()` 适配 |
|
||||
| **foundry** | Anthropic 兼容 API | 内部部署 |
|
||||
| **openai** | OpenAI 流式适配器 | 转换为 Anthropic 内部格式 |
|
||||
| **gemini** | Gemini 流式适配器 | 转换为 Anthropic 内部格式 |
|
||||
| **grok** (xAI) | Grok 流式适配器 | 转换为 Anthropic 内部格式 |
|
||||
系统支持 7 种 API Provider,每种有不同的流式协议和认证方式:
|
||||
|
||||
所有 Provider 通过统一的 `Stream<BetaRawMessageStreamEvent>` 抽象层屏蔽差异。上层代码(QueryEngine、REPL)不需要关心底层用的是哪个 Provider。
|
||||
| Provider 类别 | 流式方式 | 设计挑战 |
|
||||
|---------------|----------|----------|
|
||||
| Anthropic 直连 | 原生 SSE | 基准实现,其他 Provider 对齐它 |
|
||||
| 云平台(Bedrock/Vertex) | SDK 封装的流式接口 | 需要适配认证、beta header、参数格式 |
|
||||
| 第三方兼容(OpenAI/Gemini/Grok) | 各自的流式协议 | 需要转换为 Anthropic 内部格式 |
|
||||
|
||||
### Provider 选择
|
||||
**设计策略**:所有 Provider 通过统一的流式抽象层屏蔽差异。上层代码(编排层、交互层)不需要关心底层用的是哪个 Provider——它们只看到统一的事件流。
|
||||
|
||||
`src/utils/model/providers.ts` 中的 `getAPIProvider()` 根据配置决定使用哪个 Provider:
|
||||
这意味着切换 Provider 不需要修改任何业务逻辑,只需要在通信层适配新的协议。这也是为什么"通信层"是独立的一层。
|
||||
|
||||
```typescript
|
||||
// 根据 api_provider 配置选择:
|
||||
// "anthropic" → 直连
|
||||
// "bedrock" → AWS SDK
|
||||
// "vertex" → Google SDK
|
||||
// 第三方 base URL → 自动检测
|
||||
```
|
||||
## 接下来
|
||||
|
||||
每个 Provider 需要适配的细节包括:认证方式、beta header、请求参数格式、错误码映射——但这些差异在 `claude.ts` 的 `queryStream()` 函数中被统一处理。
|
||||
- **多轮对话** — 理解跨迭代的上下文管理
|
||||
- **上下文压缩** — 深入了解 token 超限时的自动压缩机制
|
||||
- **工具系统** — 了解工具执行的并行策略
|
||||
|
||||
Reference in New Issue
Block a user