DeepSeek Harness: StreamChunk 协议与错误处理

Q: 如何计算流式输出的 TTFB？

记录第一个 chunk 到达的时间： typescript const start = Date.now() for await (const chunk of stream) { if (chunk.type === 'text') { const ttfb = Date.now() - start ctx.logger.info(`TTFB: ${ttfb}ms`) break } }

Q: 流式输出和完整输出可以同时支持吗？

可以。适配器同时实现 complete() 和 stream() 方法。调用者根据需要选择。

Q: 如何模拟流式输出进行测试？

typescript async function* mockStream(): AsyncIterable<StreamChunk> { yield { type: 'text', content: 'Hello' } yield { type: 'text', content: ', world!' } yield { type: 'done', reason: 'stop' } } ---

最后更新：2026-08-31

流式输出是 Agent 交互的核心体验——用户不想等 30 秒才看到完整回复，而是一字一字地看 Agent 思考。StreamChunk 协议是 DSH 流式输出的统一抽象，而错误处理确保流在异常时优雅降级。

💡 提示：流式输出的难点不是"发送"，而是"出错时怎么办"。StreamChunk 协议把错误也当作一种 chunk，让消费者可以统一处理正常和异常情况。

📋 前置知识：已完成 24-llm-adapter.md，理解 LLM 适配器

1. 你将学到

流式输出协议：StreamChunk
chunk 类型：text/tool_call/tool_result
错误恢复机制
中断与取消
重试策略
日志与可观测性

2. StreamChunk 协议详解

StreamChunk 序列图

(1) ▶ 示例 1

TYPESCRIPT

type StreamChunk =
  | TextChunk
  | ToolCallChunk
  | ToolResultChunk
  | ErrorChunk
  | DoneChunk

interface TextChunk {
  type: 'text'
  content: string
}

interface ToolCallChunk {
  type: 'tool_call'
  id: string
  name: string
  arguments: string
  index: number
}

interface ToolResultChunk {
  type: 'tool_result'
  id: string
  toolCallId: string
  result: any
  isError: boolean
}

interface ErrorChunk {
  type: 'error'
  error: Error
  recoverable: boolean
  retryAfter?: number
}

interface DoneChunk {
  type: 'done'
  reason: 'stop' | 'tool_use' | 'length' | 'cancel' | 'error'
  usage?: TokenUsage
}

interface TokenUsage {
  promptTokens: number
  completionTokens: number
  totalTokens: number
}

(2) ▶ 示例 2

graph LR
    START[流开始] --> TEXT[TextChunk × N]
    TEXT --> TC[ToolCallChunk]
    TC --> TR[ToolResultChunk]
    TR --> TEXT2[TextChunk × N]
    TEXT2 --> DONE[DoneChunk]

一个典型的 Agent 对话流：

LLM 输出文本 → TextChunk
LLM 决定调用工具 → ToolCallChunk
工具执行完成 → ToolResultChunk
LLM 继续输出 → TextChunk
流结束 → DoneChunk

▶ 示例 3

TYPESCRIPT

const stream = ctx.llm.stream({
  messages: [{ role: 'user', content: '列出项目文件' }],
  tools: availableTools
})

for await (const chunk of stream) {
  switch (chunk.type) {
    case 'text':
      process.stdout.write(chunk.content)
      break

    case 'tool_call':
      console.log(`\n🔧 调用工具: ${chunk.name}`)
      console.log(`   参数: ${chunk.arguments}`)
      break

    case 'tool_result':
      if (chunk.isError) {
        console.log(`   ❌ 工具错误: ${chunk.result}`)
      } else {
        console.log(`   ✅ 工具结果: ${JSON.stringify(chunk.result)}`)
      }
      break

    case 'error':
      console.error(`\n⚠️ 错误: ${chunk.error.message}`)
      if (chunk.recoverable) {
        console.log(`   将在 ${chunk.retryAfter}ms 后重试`)
      }
      break

    case 'done':
      console.log(`\n✅ 完成 (${chunk.reason})`)
      if (chunk.usage) {
        console.log(`   Token 使用: ${chunk.usage.totalTokens}`)
      }
      break
  }
}

3. chunk 类型详解

(1) TextChunk

文本内容片段，增量输出：

TYPESCRIPT

// LLM 输出 "Hello, world!"
// 可能产生多个 TextChunk:
// chunk 1: { type: 'text', content: 'Hello' }
// chunk 2: { type: 'text', content: ', ' }
// chunk 3: { type: 'text', content: 'world' }
// chunk 4: { type: 'text', content: '!' }

消费者应拼接所有 TextChunk 而非逐个展示。

(2) ToolCallChunk

LLM 请求调用工具：

TYPESCRIPT

{
  type: 'tool_call',
  id: 'call_abc123',
  name: 'file_edit',
  arguments: '{"action":"read","path":"src/index.ts"}',
  index: 0
}

注意 arguments 是 JSON 字符串，需要解析。

(3) ToolResultChunk

工具执行后的结果：

TYPESCRIPT

{
  type: 'tool_result',
  id: 'result_xyz789',
  toolCallId: 'call_abc123',
  result: { content: 'export const name = ...' },
  isError: false
}

isError: true 表示工具执行失败，result 中包含错误信息。

(4) 多工具调用

一次对话可能调用多个工具：

TEXT 📖 仅展示

TextChunk: "我来检查两个文件"
ToolCallChunk: { name: "file_edit", id: "call_1", index: 0 }
ToolCallChunk: { name: "file_edit", id: "call_2", index: 1 }
ToolResultChunk: { toolCallId: "call_1", result: ... }
ToolResultChunk: { toolCallId: "call_2", result: ... }
TextChunk: "两个文件的内容如上..."
DoneChunk: { reason: "stop" }

4. 错误恢复机制

(1) 错误分类

TYPESCRIPT

// 不可恢复错误（如 API Key 无效）
{
  type: 'error',
  error: new Error('Invalid API key'),
  recoverable: false
}

// 可恢复错误（如速率限制）
{
  type: 'error',
  error: new Error('Rate limit exceeded'),
  recoverable: true,
  retryAfter: 5000
}

// 可恢复错误（如网络抖动）
{
  type: 'error',
  error: new Error('Connection timeout'),
  recoverable: true,
  retryAfter: 2000
}

(2) 自动恢复流程

graph TD
    ERR[ErrorChunk] --> CHECK{recoverable?}
    CHECK -->|否| FAIL[终止流 + DoneChunk<br/>reason: error]
    CHECK -->|是| WAIT[等待 retryAfter]
    WAIT --> RETRY[重试请求]
    RETRY --> SUCCESS{成功?}
    SUCCESS -->|是| CONTINUE[继续流]
    SUCCESS -->|否| ERR2[再次 ErrorChunk]
    ERR2 --> CHECK2{重试次数用尽?}
    CHECK2 -->|否| WAIT
    CHECK2 -->|是| FAIL

(3) 手动恢复

TYPESCRIPT

for await (const chunk of stream) {
  if (chunk.type === 'error') {
    if (chunk.recoverable) {
      ctx.logger.warn(`流错误，可恢复: ${chunk.error.message}`)
      // 框架自动重试
    } else {
      ctx.logger.error(`流错误，不可恢复: ${chunk.error.message}`)
      break
    }
  }
}

(4) 工具错误的处理

工具执行失败时，框架将错误作为 ToolResultChunk 返回给 LLM：

TYPESCRIPT

{
  type: 'tool_result',
  id: 'result_1',
  toolCallId: 'call_1',
  result: { error: true, message: 'Permission denied: /etc/passwd' },
  isError: true
}

LLM 收到错误后可以选择：

换一个路径重试
告知用户权限不足
用其他方式完成任务

5. 中断与取消

(1) 用户取消

用户在 Web UI 点击"停止"按钮时，流被取消：

TYPESCRIPT

const controller = new AbortController()

const stream = ctx.llm.stream(request, { signal: controller.signal })

// 用户取消
controller.abort()

// 流产生 DoneChunk
// { type: 'done', reason: 'cancel' }

(2) 超时取消

TYPESCRIPT

const controller = new AbortController()
const timer = setTimeout(() => controller.abort(), 60000)

try {
  for await (const chunk of ctx.llm.stream(request, { signal: controller.signal })) {
    // 处理 chunk
  }
} finally {
  clearTimeout(timer)
}

(3) 条件取消

TYPESCRIPT

let tokenCount = 0

for await (const chunk of stream) {
  if (chunk.type === 'text') {
    tokenCount += chunk.content.length
    if (tokenCount > 10000) {
      controller.abort()
      break
    }
  }
}

(4) 取消的清理

取消后框架确保：

停止接收新 chunk
已接收的 chunk 正常处理
产生 DoneChunk { reason: 'cancel' }
释放网络连接

6. 重试策略

(1) 内置重试策略

错误类型	重试次数	退避策略
速率限制 (429)	3	指数退避
网络超时	2	固定间隔
服务器错误 (5xx)	2	指数退避
认证错误 (401)	0	不重试
请求错误 (400)	0	不重试

(2) 配置重试

TYPESCRIPT

export const Config = Schema.object({
  maxRetries: Schema.number().default(3).description('Maximum retry attempts'),
  retryDelay: Schema.number().default(1000).description('Initial retry delay (ms)'),
  retryMultiplier: Schema.number().default(2).description('Delay multiplier for exponential backoff')
})

(3) 自定义重试逻辑

TYPESCRIPT

async function streamWithRetry(
  ctx: Context,
  request: LLMRequest,
  maxRetries = 3
): Promise<void> {
  let attempt = 0
  
  while (attempt <= maxRetries) {
    try {
      for await (const chunk of ctx.llm.stream(request)) {
        if (chunk.type === 'error' && chunk.recoverable) {
          attempt++
          const delay = Math.min(1000 * Math.pow(2, attempt), 30000)
          ctx.logger.warn(`retry ${attempt}/${maxRetries} in ${delay}ms`)
          await sleep(delay)
          break  // 重新进入 while 循环
        }
        // 处理正常 chunk
      }
      return  // 成功完成
    } catch (error) {
      attempt++
      if (attempt > maxRetries) throw error
    }
  }
}

7. 日志与可观测性

(1) 流式日志

TYPESCRIPT

ctx.on('llm/stream/start', (request) => {
  ctx.logger.info(`stream started: model=${request.model}`)
})

ctx.on('llm/stream/chunk', (chunk) => {
  ctx.logger.debug(`chunk: type=${chunk.type}`)
})

ctx.on('llm/stream/end', (done) => {
  ctx.logger.info(`stream ended: reason=${done.reason}, tokens=${done.usage?.totalTokens}`)
})

(2) 性能指标

TYPESCRIPT

interface StreamMetrics {
  ttfb: number              // Time to First Byte
  totalDuration: number     // 总耗时
  chunkCount: number        // chunk 数量
  toolCallCount: number     // 工具调用次数
  retryCount: number        // 重试次数
  tokenUsage: TokenUsage    // Token 使用
}

(3) 结构化日志

TYPESCRIPT

ctx.logger.info('llm_request', {
  model: request.model,
  messageCount: request.messages.length,
  toolCount: request.tools?.length || 0,
  stream: true
})

ctx.logger.info('llm_response', {
  model: response.model,
  duration: Date.now() - startTime,
  tokens: response.usage?.totalTokens
})

❓ 常见问题

Q 流被取消后，已调用的工具会回滚吗？

A 不会。工具执行是即时生效的（如文件已创建）。取消只停止后续的 LLM 输出。

Q ErrorChunk 后流一定会结束吗？

A 不一定会结束。如果是 recoverable 错误，框架重试后流可能继续。只有不可恢复错误或重试用尽后，流才产生 DoneChunk 结束。

Q 如何计算流式输出的 TTFB？

A 记录第一个 chunk 到达的时间：

typescript const start = Date.now() for await (const chunk of stream) { if (chunk.type === 'text') { const ttfb = Date.now() - start ctx.logger.info(`TTFB: ${ttfb}ms`) break } }

Q 多个 ToolCallChunk 的 index 有什么用？

A index 标识并行工具调用的顺序。同一个 LLM 响应可能包含多个 tool_call，index 从 0 开始递增。

Q 流式输出和完整输出可以同时支持吗？

A 可以。适配器同时实现 complete() 和 stream() 方法。调用者根据需要选择。

Q 如何模拟流式输出进行测试？

typescript async function* mockStream(): AsyncIterable<StreamChunk> { yield { type: 'text', content: 'Hello' } yield { type: 'text', content: ', world!' } yield { type: 'done', reason: 'stop' } }

---

📖 小节

StreamChunk 五种类型：text、tool_call、tool_result、error、done
流的生命周期：文本输出 → 工具调用 → 工具结果 → 继续输出 → 完成
错误分可恢复和不可恢复，可恢复错误自动重试
取消通过 AbortController 实现，产生 DoneChunk { reason: 'cancel' }
重试策略按错误类型不同，指数退避是默认策略
日志和可观测性通过生命周期事件和结构化日志实现

📝 作业

1. ⭐ 基础题：编写一个消费者，遍历 StreamChunk 流，分别统计 text/tool_call/tool_result/error/done chunk 的数量。使用模拟流测试。

2. ⭐⭐ 进阶题：实现一个带超时的流消费逻辑——如果 30 秒内没有收到任何 chunk，自动取消流。使用 AbortController 实现取消。

3. ⭐⭐⭐ 挑战题：实现一个自定义重试包装器，包装 ctx.llm.stream()，在遇到可恢复错误时自动重试（指数退避，最多 3 次）。记录每次重试的延迟和结果。模拟一个交替成功的 LLM 适配器进行测试。

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

关于我们

DeepSeek Harness: StreamChunk 协议与错误处理

1. 你将学到

2. StreamChunk 协议详解

(1) ▶ 示例 1

(2) ▶ 示例 2

▶ 示例 3

3. chunk 类型详解

(1) TextChunk

(2) ToolCallChunk

(3) ToolResultChunk

(4) 多工具调用

4. 错误恢复机制

(1) 错误分类

(2) 自动恢复流程

(3) 手动恢复

(4) 工具错误的处理

5. 中断与取消

(1) 用户取消

(2) 超时取消

(3) 条件取消

(4) 取消的清理

6. 重试策略

(1) 内置重试策略

(2) 配置重试

(3) 自定义重试逻辑

7. 日志与可观测性

(1) 流式日志

(2) 性能指标

(3) 结构化日志

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好