DeepSeek-Harness 源码深读(11):直连 DeepSeek 的适配器,一行重试都没写

摘要：把 DeepSeek 接进 harness 的包用原生 fetch 直连 /chat/completions，SSE 字节流翻译成统一协议。[DONE] 之前流不可信，块的关闭压到哨兵，缓存命中从 prompt_tokens 里减掉，读挂起由空闲看门狗掐断。重试呢？一行没写，计数记在会话日志里由隔壁插件执行。传输层七个文件逐段拆。

在 harness 里把 DeepSeek 模型接进来的包叫 @deepseek-ai/dsh-llm-deepseek。打开主体文件 adapter.ts，文件头注释给自己划了界：transport-only（adapter.ts:4）。只管传输。你大概会本能地去找重试、找超时策略、找配置校验，全都没有。发请求用原生 fetch，读响应走 SSE，出事抛一个带稳定错误码的 LlmError，到此为止。

我们把一次请求的一生从头跟到尾。案例定下来：会话里问 deepseek-v4-pro「这段代码为什么会死锁」，thinking 开着，effort 为 high。这次请求要依次经过请求序列化、fetch、SSE 分帧、增量翻译、[DONE] 收尾、usage 记账。中途它会出事三次：流被掐断、读挂起、服务器 500，我们逐个看它怎么死、死因如何被记录、最后由谁来救。

一、传输主体只有一行

request() 是传输的主体（adapter.ts:304），fetch、请求头、错误处理都在里面，但真正消费响应的只有末尾一行（adapter.ts:384）：

    yield* translate(parseSse(response.body, onComment))

这段是整条消费管道。响应体先进 parseSse 变成一个个 data 载荷，再进 translate 变成 harness 统一的 StreamChunk。注意看它有多薄：两个函数名之外没有任何中间状态，帧的重组、增量的翻译、结束的收尾全压在这两个函数里。

请求头也讲究（adapter.ts:323-335）。除了 authorization: Bearer 和 accept: text/event-stream，还带归因头、匿名 user-id，purpose 为 compaction 时再附一个 x-deepseek-harness-compact: '1'（:332-334）。另外两个刻意的位置：payload 的 JSON.stringify 放在 try 块之外（:320-321 注释），TRANSPORT 标签精确覆盖传输边界，序列化炸了不会被记成传输故障；fetch 也没走 Cordis HTTP 服务，:337-338 留着 TODO，换来的是零额外依赖。

parseSse（sse.ts:28）把分帧整个外包给 eventsource-parser：

  const events = stream    .pipeThrough(new TextDecoderStream())    .pipeThrough(new EventSourceParserStream({ onComment }))  for await (const { data } of events) {    yield data    if (data === DONE) return  }  throw new LlmError('SSE stream ended without [DONE]', 'STREAM_CLOSED')

这段是 SSE 层的全部实现。跨 chunk 重组、UTF-8 断裂、CRLF、注释与非 data 字段的跳过，全在 EventSourceParserStream（sse.ts:34）里按规范处理，本模块留下的只有 DeepSeek 协议语义。注意看最后两行：[DONE] 这个结束标记在这里不被消化，原样 yield 出去，收尾权留给 translate；循环走完了还没等到 [DONE]，直接抛 STREAM_CLOSED。为什么这么硬？SSE 规范要求事件在空行终止符处才派发，EOF 处没有终止符的尾巴属于截断，截断的模型输出不可信，宁可报错也绝不把半个 payload 当正常完成 flush 出去。

主线案例第一次出事：网关在 reasoning 输出到一半时掐断连接，字节流干净地走到 EOF，[DONE] 没来。调用方拿到的是带 STREAM_CLOSED 码的明确报错，下游没有任何半个块被放出去。

一次请求的一生，连同三种死法和救它的人，全景就是下面这张：

五步管道，三种死法各映射稳定错误码，重试回路带着会话日志计数回到 fetch

二、增量翻译：块什么时候开，什么时候关

payload 进了 translate（translate.ts:86）。它维护的状态极少（:87-93）：text 和 reasoning 各最多一个开块，工具调用按 wire index 记在一张表里，外加两个延迟变量 pendingFinish 和 pendingUsage。

第一个规矩藏在 thinking 模式的第一个增量里：

      const reasoning = delta?.reasoning_content      if (typeof reasoning === 'string' && reasoning.length > 0) {        if (!reasoningBlock) {          reasoningBlock = open('reasoning')          yield { type: 'block-start', index: reasoningBlock.index, blockType: 'reasoning' }        }        reasoningBlock.text += reasoning        yield { type: 'reasoning-delta', index: reasoningBlock.index, text: reasoning }      }

这段是 reasoning 增量的处理。思考内容在 wire 上先于正文到达，而且第一个 delta 是空字符串，这是对线上流的实测结论，types.ts 的字段注释里专门记了一笔（types.ts:127-131）。注意看 length > 0 这个条件，空串被挡在门外，不会开出一场空块开场。我们的死锁提问发出后，第一批真正带字的 reasoning 到达，block-start 这时才发出去，之后每个增量跟一条 reasoning-delta。text 走同样的路（:142-150），tool_calls 按 call.index 复用或新建块，id 和函数名只在首 delta 出现，arguments 片段逐段累积（:152-170）。

第二个规矩是收尾的时机。finish_reason 和 usage 到手先扣下（:172-179），压到 [DONE] 才统一发射：

    if (payload === DONE) {      for (const block of order) {        yield { type: 'block-end', index: block.index, block: closeBlock(block) }      }      if (pendingUsage) yield { type: 'usage', usage: pendingUsage }      const reason = pendingFinish ?? { kind: 'stop' as const }      yield {        type: 'finish',        reason: reason.kind === 'stop' && order.length === 0          ? {            kind: 'error',            failure: { message: 'model returned a completed response with no content', code: EMPTY_RESPONSE_CODE },          }          : reason,      }      return    }

这段是 [DONE] 到达时的收尾。注意看顺序：所有 block-end 按开块顺序补齐，然后 usage，最后 finish，finish 之后绝无 chunk 这条协议不变量由这个顺序结构性保证。usage 取最新一份，同时兼容附在 finish chunk 和末尾 usage-only chunk 两种 wire 形态（:177-179）。中段还藏着一个降级：finish_reason 是 stop 但一个块都没开过，翻译成 EMPTY_RESPONSE 错误（:110-115），一次「成功的空消息」就此变成显式失败，重试执行器认的恰恰是这个码。

三、usage 记账：有一笔要减掉

[DONE] 前那笔 usage 账里有一个坑。DeepSeek 文档写明 prompt_tokens 含缓存命中，等于命中加未命中；harness 的 TokenUsage 约定是不相交计数（translate.ts:46-49 注释）。mapUsage 做的就是这笔减法（translate.ts:53）：

export function mapUsage(usage: WireUsage): TokenUsage {  const cacheRead = usage.prompt_tokens_details?.cached_tokens ?? usage.prompt_cache_hit_tokens  const reasoning = usage.completion_tokens_details?.reasoning_tokens  return {    inputTokens: usage.prompt_tokens - (cacheRead ?? 0),    outputTokens: usage.completion_tokens,    ...cacheRead !== undefined ? { cacheReadTokens: cacheRead } : {},    ...reasoning !== undefined ? { reasoningTokens: reasoning } : {},  }}

这段是 wire usage 到 harness usage 的换算。注意看 inputTokens 那行，命中部分被减掉之后才作为输入记账，cacheReadTokens 单独成列。不减的话，同一个 token 会在输入和缓存读里被数两遍，账面输入量虚高。下一篇 token-meter 的计量账本，吃的正是这份不相交口径。

四、读挂起的时候，谁动手掐

主线案例第二次出事：模型思考了很久，流安静下来，一次读挂在那里，没有 EOF 也没有错误，纯挂起。stream() 开头几行就是为这个场景准备的（adapter.ts:255）：

    const consumer = new AbortController()    const upstream = options.signal === undefined      ? consumer.signal      : AbortSignal.any([options.signal, consumer.signal])    using watchdog = idleWatchdog(upstream, connection.streamIdleTimeoutMs, STREAM_IDLE_TIMEOUT_CODE)

这段是读挂起的守卫。调用方的取消信号和消费者自己的控制器汇成 upstream，idleWatchdog 拿着它与 streamIdleTimeoutMs（默认 300 秒，adapter.ts:97）上岗。注意看它量的是空闲：单次流读挂够 300 秒才触发，模型连着思考几分钟、字节还在流动就完全不碰。若上的是总超时，长思考会被拦腰截断；这只看门狗把「模型在想」和「管道死了」分成了两回事。

SSE 注释也参与喂狗。注释经 onComment 回调上报（sse.ts 文档注释称其为 transport-activity），这个回调在 stream() 里就是 watchdog.pulse()（adapter.ts:267），一条注释也算一次传输活动。

消费循环逐块经过 watchdog.next（:272），出了异常，catch 里的归因有先后（:279-291）：先查狗是否叫了，是则 TIMEOUT；再查调用方是否取消，是则 ABORTED；已是 LlmError 的原样透传；剩下的统一包成带端点的 TRANSPORT。fetch 的裸 TypeError（DNS、TLS、代理故障全被它裹成一模一样的 fetch failed）也是在这里被补上诊断（:347-358）。finally 里 consumer.abort() 一刀切断底层下载（:293），消费者提前退出就不会白下载；流没耗尽则补一个 iterator.return() 并吞掉拆卸时的 abort 异常（:294-299）。

五、死要死得有名字：错误码和一本会话账

第三次出事：服务器 500。非 2xx 的处理在 request() 里（adapter.ts:361-379），错误体 message、retry-after 头、x-request-id 都被摘出来挂在 LlmError 上。状态码到稳定码的对照是 httpErrorCode（adapter.ts:150）：

  if (status === 401 || status === 403) return 'AUTH'  if (status === 413) return 'INVALID_REQUEST'  const detail = [error?.code, error?.type, error?.message].filter(Boolean).join(' ')  if (isQuotaExceededError(detail)) return QUOTA_EXCEEDED_CODE  if (status === 429) return 'RATE_LIMIT'  if (status === 400) {    if (isContextWindowExceededError(detail)) return CONTEXT_WINDOW_EXCEEDED_CODE    return 'INVALID_REQUEST'  }  if (status >= 500) return 'SERVER'  return `HTTP_${status}`

这段是 HTTP 状态到 harness 错误码的映射。注意看配额检测排在 429 前面，同一个 429，配额耗尽是更精确的诊断，所以配额码优先。未识别的 finish_reason 走同一条保守路线，统一映射成 error（translate.ts:31-43）。我们的 500 在这里得到名字：SERVER。

名字有了，重试在哪？在隔壁。执行器是另一个包 dsh-llm-retry，挂在 agent loop 的 agent/request-error 扩展点上（llm-retry/src/index.ts:210），默认认五个码：EMPTY_RESPONSE、RATE_LIMIT、SERVER、TIMEOUT、TRANSPORT（llm/src/retry-policy.ts:18-24），与上面的映射产出精确对齐。错误码是 adapter 与重试执行器之间的全部耦合面，执行器对 DeepSeek 一无所知。它的计数方式值得单独看（llm-retry/src/index.ts:182）：

    const priorPolicyRetry = agent.session.events.findLast((event): event is SessionEvent<'llm/retry'> =>      event.type === 'llm/retry'      && event.data.turn === turn      && event.data.step === step      && event.data.provider === provider      && event.data.policyKey === policyKey,    )    const previousRetry = priorPolicyRetry?.data.retry ?? 0    if (policy.mode === 'normal' && previousRetry >= policy.maxRetries) return next()    const retry = previousRetry + 1

这段是重试号的来历。findLast 直接翻会话事件日志，找同 turn、同 step、同 provider、同策略指纹的上一条 llm/retry 事件，取它的号加一。注意看这里没有内存计数器，重试号持久化在会话日志里，进程崩溃重启后接着数，同一步不会从 1 再试一遍。等待的顺序也是先记账后行动：append('llm/retry') 落盘，再进入可取消的延迟，睡醒后补一条 llm/retry-started（:150-153）。退避公式是指数加对称抖动（:58-63），500ms 起、10 秒封顶、抖动比 0.1；provider 给了 retry-after 且不超过封顶就直接采纳（:194-205）。

六、改了配置，下一发请求才换血

还有一条线没对：请求进行到一半，你在设置页把 baseURL 换成内网网关，这条流毫无感知。连接事实在 stream() 开头就被冻结成一份快照（adapter.ts:234），密钥也从这份快照解析（:253），端点和密钥永远同代，新密钥打旧端点这种组合在结构上就凑不出来。下一次调用重新解析，新配置自然到位。

重解析的缓存在插件侧（llm-deepseek/src/index.ts:236）：

    const raw = current()    if (raw === lastRaw && lastGood !== undefined) return lastGood    try {      const next = resolveAdapterOptions(raw, launchEnvironmentOf(ctx))      lastRaw = raw      lastGood = next      return next    } catch (error) {      // …      if (lastGood === undefined) throw error      lastRaw = raw      ctx.logger.error('llm-deepseek: keeping the last good configuration after an invalid settings section')      ctx.logger.error(error)      return lastGood    }

这段是配置生命周期的核心。raw 引用没变直接命中缓存；变了就重新过 resolveAdapterOptions，所有默认值和边界在那里重新判定（:188-230），baseURL 走配置、受信环境层 $DEEPSEEK_BASE_URL、公网端点三级回退（:216-218）。注意看 catch 分支，坏的新设置不清空运行时，日志记一句，继续用上一份好的；从来没好过（组合期）才向上抛，快速失败。设置热更新走的就是这条链（installSettingsSection，index.ts:307）。

retryPolicy 是唯一的例外。注册表在注册时捕获策略（:294），逐请求解析刷不动它，设置变化时用 registration.replace([PROVIDER]) 在一个同步注册表段内原地重注册（:303）。为什么没有选 dispose 再重新注册？注释写得很直白：那样会短暂发布一个空路由集，观察者会看到这个 provider 消失又出现（:298-302）。

结语

回到主线那次提问，把一生对完账。请求体在 serializeRequest 成型，纯 tool-call turn 的 assistant 消息发空串不发 null，注释里记着线上 API 对 null 的 400 实测（serialize.ts:176-184）；fetch 带归因头出门（adapter.ts:327）；SSE 字节经 EventSourceParserStream 分帧（sse.ts:34）；reasoning 的第一个空 delta 被 length > 0 拦下（translate.ts:133）；三种内容按开块顺序在 [DONE] 处统一收尾（translate.ts:103-105）；usage 把缓存命中从输入里减掉（translate.ts:57）。三种死法各有名字：截断叫 STREAM_CLOSED，挂起叫 TIMEOUT，服务器 500 叫 SERVER；名字进了会话日志，重试号在日志里接着数（llm-retry/src/index.ts:189）。

拆完这条链上的七个文件，我印象最深的是注释的密度。空串不开块（types.ts:127-131）、空串不发 null（serialize.ts:176-184）、缓存要相减（translate.ts:46-49），每条怪规矩旁边都有一行注释写明出处，线上 400、官方样例或实测 wire 行为，后来的人不用重新踩一遍坑就知道规矩为什么存在。

深读线下一篇走 token-meter。这一篇里 usage 被减成不相交计数交给下游，谁在消费这份口径、会话的 token 账本怎么对，token-meter 会接着把这笔账算到底。

你的流式客户端读到截断的流，是报错重试，还是把半截输出当正常完成交差？评论区聊聊。

本系列基于 DeepSeek Harness 源码（MIT，0.1.1-rc.1）与官方 Agent Notes 整理，仓库：github.com/deepseek-ai/deepseek-harness。有收获就点个关注，下一篇见。