cs-note/claude-code-best/docs/conversation/streaming.md

tags, create time

tags	create time
流式响应 SSE streaming API 多Provider	2026-06-09 22:15

流式响应机制 - Claude Code 打字机效果原理

概述

Claude Code 通过 SSE（Server-Sent Events）实现流式响应，逐 token 接收 AI 输出，实时呈现"打字机"效果。本文解析事件状态机、内容块交织、错误处理和多 Provider 适配的完整实现。

正文

为什么需要流式

想象 AI 需要 30 秒才能生成完整回答——如果等 30 秒后才一次性显示，用户体验是灾难性的。

流式响应让用户实时看到 AI 的思考过程:

• 文字逐字出现，用户能提前判断方向是否正确
• 工具调用的参数在生成过程中就能预览
• 长时间任务不会让用户觉得"卡死了"

BetaRawMessageStreamEvent 核心事件类型

流式 API 返回的是一系列 BetaRawMessageStreamEvent，每种事件类型对应流式响应的不同阶段（src/services/api/claude.ts）:

flowchart TD
    A["message_start 消息开始"] --> B["content_block_start 内容块开始"]
    B --> C["content_block_delta 增量数据"]
    C --> C
    C --> D["content_block_stop 内容块结束"]
    D --> B
    D --> E["message_delta stop_reason + 最终usage"]
    E --> F["message_stop 消息结束"]

事件处理状态机

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	无操作（流结束标记）	--

内容块类型及其增量数据

content_block_start 中的 content_block.type 决定了如何处理后续 delta:

内容块类型	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 控制）

[!tip] 设计细节 content_block_start 时所有文本字段初始化为空字符串，只通过 content_block_delta 累加。这是因为 SDK 有时在 start 和 delta 中重复发送相同文本。

文本 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——文本消息和工具调用消息交替产出。

stop_reason 要等到 message_delta 才确定（可能是 end_turn、tool_use、max_tokens 等），所以最后一条消息的 stop_reason 是回写的:

// claude.ts -- stop_reason 回写逻辑（直接属性修改，不用对象替换）
// 因为 transcript 写队列持有 message.message 的引用
const lastMsg = newMessages.at(-1)
if (lastMsg) {
  lastMsg.message.usage = usage
  lastMsg.message.stop_reason = stopReason
}

流式中的错误处理

网络断开

流式连接依赖 SSE。当连接中断时，系统有三层检测机制:

1. 被动停滞检测: 当下一个事件到达时，计算与上一个事件的时间间隔。超过阈值（30 秒，STALL_THRESHOLD_MS = 30_000）记录为一次 stall，累积计数并写入遥测日志。
2. 主动空闲超时看门狗: 使用 setTimeout 设置 90 秒（可通过 CLAUDE_STREAM_IDLE_TIMEOUT_MS 环境变量覆盖）的硬性超时。如果在此期间没有收到任何事件，主动终止流并抛出错误进入重试流程。
3. 非流式降级: 作为最后手段，设置 didFallBackToNonStreaming 标志，通过 executeNonStreamingRequest() 回退到非流式请求。

// claude.ts -- 被动停滞检测
const STALL_THRESHOLD_MS = 30_000  // 30 秒无事件视为停滞

// claude.ts -- 主动空闲超时
const STREAM_IDLE_TIMEOUT_MS =
  parseInt(process.env.CLAUDE_STREAM_IDLE_TIMEOUT_MS || '', 10) || 90_000

API 限流

当 API 返回限流错误时，系统使用 withRetry 包装器进行指数退避重试。重试逻辑考虑了:

• 错误类型（429 限流 vs 500 服务器错误）
• 重试次数上限
• 退避间隔

Token 超限

两种 token 超限场景有不同的处理:

场景	stop_reason	处理方式
输出超限	max_tokens	生成错误消息，建议设置 CLAUDE_CODE_MAX_OUTPUT_TOKENS
上下文窗口超限	model_context_window_exceeded	触发 compaction 压缩对话历史后重试

// 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', ... })
}

工具执行的流式反馈

BashTool 的命令执行也是流式的——通过 onProgress 回调逐行推送输出:

flowchart TD
    A["BashTool.call"] --> B["runShellCommand AsyncGenerator"]
    B --> C["每秒轮询输出文件"]
    C --> D["onProgress lastLines, allLines"]
    D --> E["yield progress, output, fullOutput"]
    B --> F["return code, stdout, interrupted"]

UI 层通过 useToolCallProgress hook 实时展示命令输出，而不是等命令完全结束。长时间运行的命令还支持自动后台化（shouldAutoBackground）。

多 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 内部格式

所有 Provider 通过统一的 Stream<BetaRawMessageStreamEvent> 抽象层屏蔽差异。上层代码（QueryEngine、REPL）不需要关心底层用的是哪个 Provider。

[!info] Provider 选择 src/utils/model/providers.ts 中的 getAPIProvider() 根据配置决定使用哪个 Provider。每个 Provider 需要适配认证方式、beta header、请求参数格式、错误码映射——但这些差异在 claude.ts 的 queryStream() 函数中被统一处理。

关联笔记

• the-loop - Agentic Loop 核心机制
• multi-turn - 多轮对话管理与 QueryEngine
• ../context/token-budget - Token 预算与输出限制