LLM 适配器与流式：从请求到 settlement

专栏：DeepSeek Harness · 第 11 / 14 篇
DeepSeek HarnessLLM流式

第 7 篇的 step() 里，模型调用浓缩成了一句 preparedCall?.stream(request) ?? this.loopCtx.llm.stream(request)。这篇把这一行背后的整个世界展开。

适配器层的结构

先看这层的架构：消费方、运行时、waterfall、注册表与适配器的关系。

图表（llm-adapter-streaming.md）

ctx.llm：一个 waterfall 通向所有适配器

ctx.llm（LlmRuntime）对外只有一条流式入口，而它本身是一个 waterfall——适配器查找发生在 waterfall 的终端 continuation：

// packages/llm/llm/src/index.ts:53-74（节选）
declare module '@deepseek-ai/cordis' {
  interface Context {
    llm: LlmRuntime
  }
  interface Events {
    /**
     * Waterfall around every streaming model call (retry, replay, routing).
     * … A LOOP-built request carries the process-local
     * {@link markAgentLoopRequest} identity and arrives deep-frozen
     * (mutation throws): its content is a pure function of the session log …,
     * so listeners read it, never rewrite it.
     * @mode waterfall
     */
    'llm/stream'(this: LlmRuntime, options: GenerateOptions, next: () => AsyncIterable<StreamChunk>): AsyncIterable<StreamChunk>
  }
}
// packages/llm/llm/src/index.ts:1093-1104
stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
  return this.streamWithRegistration(options)
}

private streamWithRegistration(options: GenerateOptions, prepared?: PreparedDispatch) {
  return this.ctx.waterfall(
    this,
    'llm/stream',
    options,
    () => this.adapterStream(options, prepared),
  )
}

这意味着：任何插件都可以拦截 llm/stream 短路（自产 chunk，如缓存/回放）、或调 next() 走到真适配器。loop 构建的请求深度冻结——监听器只能读，不能改写消息。

适配器注册是全或无的：ctx.llm.registerAdapter(['deepseek-official'], adapter)，重复路由抛错，随 fiber 释放自动注销，还带 replace() 原子换路由。

适配器契约：唯一必选的方法是 stream

// packages/llm/llm/src/index.ts:257-278（节选）
export abstract class LlmAdapter ... {
  /**
   * Bind exact model metadata and the eventual request dispatch to one adapter
   * generation. Dynamic adapters override this so settings changes between
   * preparation and dispatch cannot combine one generation's capabilities with
   * another's endpoint.
   */
  async prepareCall(provider: string, model: string, signal?: AbortSignal): Promise<PreparedAdapterCall> {
    return {
      model: await this.resolveModel(provider, model, signal),
      stream: options => this.stream(options),
    }
  }

  /**
   * Stream one model call as raw chunks. The only required method.
   * @returns the chunk stream, obeying the adapter contract documented on StreamChunk.
   */
  abstract stream(options: GenerateOptions): AsyncIterable<StreamChunk>
}

prepareCall 存在的意义是代际绑定：loop 先准备（解析模型元数据、锁定本次调用的适配器注册与重试策略），再发流——设置热变更时，不会把 A 代际的能力结果配到 B 代际的端点上。

StreamChunk：七种分片的原始协议

// packages/llm/llm/src/types.ts:370-390
export type StreamChunk =
  | { type: 'block-start'; index: number; blockType: ContentBlockType }
  | { type: 'text-delta'; index: number; text: string }
  | { type: 'reasoning-delta'; index: number; text: string }
  | { type: 'tool-call-delta'; index: number; id: ToolCallId; name?: string; argumentsDelta: string }
  | { type: 'block-end'; index: number; block: ContentBlock }
  | { type: 'usage'; usage: TokenUsage }
  | {
    type: 'finish'
    reason: FinishReason        // stop / tool-calls / max-tokens / aborted / error
    replayState?: ReplayEnvelope
  }

适配器契约的硬规则：usage 必须在 finish 之前、之后不得再有分片；工具参数全程是原始 JSON 字符串（argumentsDelta 续片重发空串表示「不变」）；错误只有两条合法路径——抛异常（传输故障）或带内 finish{kind:'error'|'aborted'}（协议内故障），不允许第三种。block-end 直接携带组装好的完整块，消费方可以不自己拼 delta。

双轨制：实时帧 vs 持久 settlement

模型流的消费分两条轨道，这是本篇最重要的区分：

轨道一（实时，进程本地）：agent/assistant-stream 事件的三种帧：

// packages/core/agent/src/runtime-types.ts:74-107（节选）
export type AssistantStreamFrame =
  | { readonly type: 'start'; readonly attemptId: LlmAttemptId; readonly revision: number;
      readonly turn: number; readonly step: number }
  | { readonly type: 'chunk'; readonly attemptId: LlmAttemptId; ...; readonly chunk: StreamChunk }
  | { readonly type: 'end'; readonly attemptId: LlmAttemptId; ...;
      readonly outcome:
        | { readonly kind: 'committed'; readonly eventType: 'assistant/message' | 'assistant/attempt';
            readonly seq: SessionSeq }
        | { readonly kind: 'abandoned' } }

轨道二（持久）：settlement 时落日志——成功是 assistant/message（内嵌完整的紧凑流与 usage），失败/重试/取消是 assistant/attempt（保留到达终局的流，但不进模型历史）。时序上先持久提交、后发终结帧：end 帧的 committed 携带刚落日志的 seq，UI 拿到它就能从实时模式无缝切到从日志回放。

loop 侧的分支逻辑（错误→attempt→征询重试）：

// packages/core/agent-loop/src/agent.ts:425-448（节选）
const finish = live.finish
if (finish.kind === 'error' || finish.kind === 'aborted') {
  live.settle('assistant/attempt',
    () => this.session.append('assistant/attempt', { turn, step, stream: live.stream }).seq)
  const action = await this.dispatch.waterfall(
    'agent/request-error', { turn, step, provider: request.provider, failure: finish.failure, ... },
    () => Promise.resolve<RequestErrorAction>(undefined),
  )
  if (action?.kind !== 'retry') {
    throw new LlmError(finish.failure.message, finish.failure.code, finish.failure)
  }
  continue      // 重开一步：llm-retry 插件就挂在这里
}

注意一条纪律：一次适配器调用 = 一次提供方尝试。适配器内部禁用库级重试；重试发生在 agent 层的持久步骤边界，每次尝试都在日志里留下独立带编号的 attempt。

deepseek 适配器：一个具体实现

llm-deepseek 包展示了接入一个真实提供方的全部要素。注册：

// packages/llm/llm-deepseek/src/index.ts:455-476（节选）
const adapter = new DeepSeekAdapter({
  options,
  resolveApiKey,            // 经 ctx.credentials seam，每次请求重新解析 → 密钥轮换免重启
  resolveUserId,
  resolveAttachments: () => ctx.get('attachments'),
  resolveImageAccess: (attachments, ref) => resolveImageAttachmentAccess(
    attachments,
    hostPath => ctx.get('fs')?.processPathFromHostPath(hostPath),  // ← 消费 fs seam！
    ref,
  ),
  prepareExtensions: (request) => { ... ctx.get('deepseekLlmApiExtensions')?.prepare(request) ... },
})
ctx.llm.registerConfigurableProviders([{ provider: PROVIDER, displayName: 'DeepSeek', ... }])
const registration = ctx.llm.registerAdapter([PROVIDER], adapter)

请求头（端点 ${baseURL}/chat/completions，默认 https://api.deepseek.com，兼容 OpenAI 网关）：

// packages/llm/llm-deepseek/src/adapter.ts:531-543
const headers = {
  'authorization': `Bearer ${apiKey}`,
  'content-type': 'application/json',
  'accept': 'text/event-stream',
  ...attributionHeaders(),
  'x-deepseek-harness-user-id': String(userId),
  ...options.sessionId !== undefined ? { 'x-deepseek-harness-session-id': String(options.sessionId) } : {},
  ...options.purpose === 'compaction' ? { 'x-deepseek-harness-compact': '1' } : {},
}

流式路径由 300 秒空闲看门狗包裹（到期映射 TIMEOUT，不吞调用方中止）；SSE 用 eventsource-parser 严格分帧，[DONE] 前流提前结束抛 STREAM_CLOSED；translate 层把 wire chunk 翻成 harness StreamChunk（连「把缓存命中 token 从 prompt_tokens 里拆出来保持计数不相交」这种细节都处理了）。

wire 扩展是另一个有意思的设计：dsh_plugin_packages、dsh_session_log 等插件向官方 API 请求注册 dsh_ 前缀的顶层字段（如会话日志的至少一次交付），字段在 messages/工具 schema 之外——不增加模型输入 token、不改变模型可见前缀。HTTP 2xx 后、读 SSE 前才确认接受。

全链路时序

图表（llm-adapter-streaming.md）

与其他模块的联系

  • session（第 6 篇）：assistant/message / assistant/attempt 内嵌紧凑流，是回放与遥测的权威；
  • agent-loop（第 7 篇）：agent/request-error waterfall 上的 llm-retry 插件在持久步骤边界重试；
  • 凭据（第 12 篇）：适配器经 ctx.credentials 每请求解析密钥——轮换免重启；
  • fs（第 10 篇）：附件路径经 processPathFromHostPath 映射进执行世界——seam 思想无处不在。

全篇的钥匙还是第 6 篇那句「模型可见即已记录」：流的实时帧是瞬态的，持久世界只认 assistant/message / assistant/attempt 里内嵌的流。UI 断线重连、进程崩溃恢复、回放审计，全都从日志重建——实时帧只是锦上添花。

下一篇转产品视角：这一切如何被包装成人能用的审批流、权限档与三种应用形态。

← 返回文章列表