LLM 通信三阶段

🔹 序列化 (Serialize) — 将 Harness 内部的消息格式转换为 OpenAI 兼容的 JSON 请求体

🔹 传输与 SSE 解析 (Transport + SSE) — 发起 HTTP 请求，逐帧解析 Server-Sent Events 流

🔹 翻译 (Translate) — 将 SSE 原始载荷翻译为 Harness 的 StreamChunk 协议

export class LlmRuntime extends Service {  private adapters = new Map<string, AdapterRegistration>()   // provider → 适配器注册表（路由核心）  private directory = new Map<string, LlmConfigurableProvider>() // 可配置 provider 目录  private discoveries = new Map<                               // 模型发现回调（provider → 发现函数）    string,    (request: LlmModelDiscoveryRequest) => Promise<readonly LlmDiscoveredModel[]>  >()  constructor(ctx: Context) {    super(ctx, 'llm')                                          // 在 Cordis 中注册为 "llm" 服务  }  /** 为给定 provider 路由注册适配器（全有或全无）。 */  registerAdapter(providers: string[], adapter: LlmAdapter): AdapterRegistrationHandle {    const owned = new Set<string>()                            // 本次注册当前持有的路由集合    let released = false                                       // disposer 是否已执行    const dispose = this.ctx.effect(function* (this: LlmRuntime) { // 注册为 fiber effect，随会话销毁      if (providers.length === 0) throw new LlmError('an adapter must register at least one provider', 'INVALID_ADAPTER')      this.commitRoutes(owned, this.prepareRoutes(providers, adapter, owned)) // 提交：provider 重复则整体拒绝      yield () => {                                            // disposer：销毁时释放路由        released = true        for (const provider of owned) this.adapters.delete(provider)        owned.clear()        this.emitAdaptersUpdated()                             // 通知拓扑观察者（非否决式）      }    }.bind(this), 'llm.registerAdapter()')    const handle = (() => void dispose()) as AdapterRegistrationHandle    handle.replace = (next: string[]): void => { /* 改写本次注册持有的路由 */ }    return handle  }  /** "准备-派发"两阶段：prepare 冻结配置，stream 只做一次性派发。 */  async prepareCall(config: LlmCallConfig, signal?: AbortSignal): Promise<PreparedLlmCall> {    const registration = this.registration(config.provider)    // 按 provider 查适配器；缺失抛 NO_ADAPTER    const resolved = await this.resolveCallFor(registration, config, signal) // 解析模型 + 合并默认值    const resolvedConfig = deepFreeze(structuredClone(resolved.config))     // 深冻结：防异步等待期间被外部修改    let dispatched = false                                     // 一次性派发标志    return Object.freeze({      config: resolvedConfig,                                  // 最终配置的冻结快照（重试也用这份）      retryPolicy: registration.retryPolicy,                   // provider 注册时绑定的重试策略      stream: (options: GenerateOptions): AsyncIterable<StreamChunk> => {        if (dispatched) throw new LlmError('a prepared LLM call can only be dispatched once', 'INVALID_PREPARED_CALL')        if (!callConfigEquals(options, resolvedConfig))        // 派发前配置被改动 → 拒绝          throw new LlmError('prepared LLM call config changed before adapter dispatch', 'INVALID_PREPARED_CALL')        dispatched = true                                      // 标记已派发，禁止二次调用        return this.streamWithRegistration(options, { registration, config: resolvedConfig })      },    })  }}

private async * adapterStream(  options: GenerateOptions,  prepared?: { registration: AdapterRegistration; config: LlmCallConfig },): AsyncGenerator<StreamChunk> {  let iterator: AsyncIterator<StreamChunk>  try {    const registration = prepared?.registration ?? this.registration(options.provider) // 复用已解析注册或现查    const resolvedConfig = prepared === undefined      ? (await this.resolveCallFor(registration, options, options.signal)).config      : prepared.config    if (prepared !== undefined && !callConfigEquals(options, resolvedConfig)) {      throw new LlmError('prepared LLM call config changed before adapter dispatch', 'INVALID_PREPARED_CALL')    }    const resolvedOptions = callConfigEquals(options, resolvedConfig)      ? options      : Object.isFrozen(options)        ? deepFreeze({ ...options, ...resolvedConfig })        // 冻结对象走深冻结合并，否则浅合并        : { ...options, ...resolvedConfig }    const adapter = registration.adapter                        // 从注册表取出目标适配器    const stream = adapter.stream(this.forAdapter(resolvedOptions, adapter)) // 派发：replay 状态按归属过滤    iterator = stream[Symbol.asyncIterator]()                   // 拿到异步迭代器  } catch (error: unknown) {    yield adapterFailureChunk(error, options.signal)            // 选型/派发失败 → 转为终端 finish chunk，不抛异常    return  }  let completed = false  try {    while (true) {      let item: { done: true } | { done: false; value: StreamChunk }      try {        const next = await iterator.next()                      // 拉取下一个 chunk        item = next.done ? { done: true } : { done: false, value: next.value }      } catch (error: unknown) {        completed = true                                        // 迭代中失败（网络/SSE 解析）→ 同样转为终端 chunk        yield adapterFailureChunk(error, options.signal)        return      }      if (item.done) { completed = true; return }               // 正常结束      // 在 yield 之前结束适配器侧 try：消费者/中间件失败必须保持"抛出"语义      yield item.value                                          // 向下游透传 chunk    }  } finally {    if (!completed) {                                           // 非正常退出（被 break/throw）时关闭迭代器      const close = iterator.return?.bind(iterator)      if (close) await close()    }  }}

async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {  // 每次 stream 调用只解析一次配置：连接事实与凭据在此冻结，  // 整个请求期间不会观察到配置变更；endpoint 和密钥来自同一代配置。  const connection = this.config.options()                      // 快照当前连接配置（baseURL、超时等）  const apiKey = await this.config.resolveApiKey(connection)    // 从该快照解析 API key  const userId = this.config.resolveUserId()                    // 匿名身份 user id  const consumer = new AbortController()                        // 消费者自终止控制器  const upstream = options.signal === undefined    ? consumer.signal    : AbortSignal.any([options.signal, consumer.signal])        // 调用方取消 + 消费者终止，任一触发即 abort  using watchdog = idleWatchdog(upstream, connection.streamIdleTimeoutMs, STREAM_IDLE_TIMEOUT_CODE) // SSE 空闲看门狗  const iterator = this.request(                                // 进入 HTTP 请求生成器    options, watchdog.signal, connection, apiKey, userId,    () => { watchdog.pulse() },                                 // SSE comment 到达时心跳，证明连接仍活跃  )[Symbol.asyncIterator]()  let exhausted = false  try {    while (true) {      const result = await watchdog.next(iterator)              // 经看门狗拉取：空闲超时在此触发      if (result.done) { exhausted = true; return }             // 流正常耗尽      yield result.value                                        // 透传 chunk    }  } catch (error: unknown) {    if (timeoutOf(watchdog.signal, STREAM_IDLE_TIMEOUT_CODE) !== undefined) {      throw new LlmError(`DeepSeek stream idle timeout after ${connection.streamIdleTimeoutMs}ms`, 'TIMEOUT', { cause: error }) // 空闲超时 → TIMEOUT（可重试）    }    if (options.signal?.aborted) {      throw new LlmError('DeepSeek request aborted by caller', 'ABORTED', { cause: error }) // 调用方取消 → ABORTED（不重试）    }    if (error instanceof LlmError) throw error                  // 已是分类错误，原样上抛    throw new LlmError(`DeepSeek API stream from ${connection.baseURL} failed`, 'TRANSPORT', { cause: error }) // 兜底 → TRANSPORT（可重试）  } finally {    consumer.abort('DeepSeek stream consumer stopped')          // 消费者侧终止，释放连接    if (!exhausted && iterator.return !== undefined) {      try { await iterator.return() } catch (_abortedTransportTeardown) { /* 终止权已归消费者 */ }    }  }}

三层超时/取消机制

🔹 AbortSignal.any — 调用方取消或消费者自终止，任意一个触发即 abort

🔹 idleWatchdog — SSE 流连续无数据超过阈值自动 abort，防止长连接挂死

🔹 watchdog.pulse() — SSE comment 到达时心跳，证明连接仍活跃

private async * request(  options: GenerateOptions, signal: AbortSignal, connection: DeepSeekConnectionOptions,  apiKey: *** userId: AnonymousUserId, onComment: () => void,): AsyncIterable<StreamChunk> {  const body = serializeRequest(options, connection.defaults)   // Harness 消息 → OpenAI 兼容 JSON  const payload = JSON.stringify(body)                           // 序列化失败不落入 TRANSPORT 标签  const headers = {    'authorization': `Bearer ${apiKey}`,                        // Bearer 鉴权    'content-type': 'application/json',    'accept': 'text/event-stream',                              // 声明要 SSE 流式响应    ...attributionHeaders(),                                    // Harness 归因头    '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' } : {}, // 压缩调用打标  }  let response: Response  try {    response = await fetch(`${connection.baseURL}/chat/completions`, { method: 'POST', headers, body: payload, signal })  } catch (error: unknown) {    if (signal.aborted) throw error                              // 外层 stream 负责区分取消与超时    // fetch 把所有传输失败（DNS/拒连/TLS/代理）包成裸 TypeError，细节在 cause 上    throw new LlmError(`DeepSeek API request to ${connection.baseURL} failed`, 'TRANSPORT', { cause: error })  }  if (!response.ok) {                                            // HTTP 错误分支：解析 provider 错误体    let message = `DeepSeek API error (HTTP ${response.status})`    let providerError: WireError['error']    try {      const parsed = await response.json() as WireError      providerError = parsed.error      if (providerError?.message) message = providerError.message // 优先用 provider 的报错文案    } catch { /* 只吞错误体解析失败：HTTP 状态码仍能定位问题 */ }    const delay = providerRetryAfterMs(response.headers.get('retry-after')) // 提取限流重试等待    const id = requestId(response.headers)                       // 提取请求追踪 id    throw new LlmError(message, httpErrorCode(response.status, providerError), {      status: response.status,      ...delay === undefined ? {} : { providerRetryAfterMs: delay },      ...id === undefined ? {} : { requestId: id },    })                                                           // 错误码交给重试策略判定  }  if (!response.body) throw new LlmError('DeepSeek API returned no response body', 'EMPTY_RESPONSE')  yield* translate(parseSse(response.body, onComment))           // SSE 帧解析 → StreamChunk 翻译，两级管道}

export function httpErrorCode(status: number, error?: WireError['error']): string {  if (status === 401 || status === 403) return 'AUTH'            // 鉴权失败：重试无意义，不可重试  const detail = [error?.code, error?.type, error?.message].filter(Boolean).join(' ')  if (isQuotaExceededError(detail)) return QUOTA_EXCEEDED_CODE    // 额度耗尽：不可重试  if (status === 429) return 'RATE_LIMIT'                         // 限流：可重试（配合 retry-after）  if (status === 400) {    if (isContextWindowExceededError(detail)) return CONTEXT_WINDOW_EXCEEDED_CODE // 上下文超长：不可重试    return 'INVALID_REQUEST'                                      // 请求非法：不可重试  }  if (status >= 500) return 'SERVER'                              // 服务端错误：可重试  return `HTTP_${status}`                                         // 兜底：按状态码原样上报}

export const DONE = '[DONE]'                                     // DeepSeek/OpenAI 在最后一个 chunk 后发送的终止载荷/** * 把 SSE 字节流解析为 data 载荷。帧重组、UTF-8/CRLF/BOM、comment 跳过、 * 多 data: 拼接都交给 eventsource-parser；本模块只保留 DeepSeek 协议语义： * [DONE] 原样 yield（调用方负责最终 flush），EOF 前没收到则抛错。 */export async function* parseSse(  stream: ReadableStream<BufferSource>,  onComment?: (comment: string) => void,                         // comment 仅作传输活性回调，不进入载荷流): AsyncGenerator<string> {  const events = stream    .pipeThrough(new TextDecoderStream())                        // 字节 → UTF-8 文本（处理跨 chunk 的多字节字符）    .pipeThrough(new EventSourceParserStream({ onComment }))     // SSE 帧解析：按空行终止事件分发  for await (const { data } of events) {    yield data                                                    // 逐条产出 data 载荷，保持到达顺序    if (data === DONE) return                                     // 收到 [DONE] 即正常结束  }  throw new LlmError('SSE stream ended without [DONE]', 'STREAM_CLOSED') // 流被截断：本次模型调用不可信}

export async function* translate(payloads: AsyncIterable<string>): AsyncGenerator<StreamChunk> {  let nextIndex = 0                                               // 块索引分配器  let textBlock: OpenBlock | undefined                            // 当前打开的文本块（至多一个）  let reasoningBlock: OpenBlock | undefined                       // 当前打开的推理块（至多一个）  const toolBlocks = new Map<number, OpenBlock>()                 // 工具调用块：按 wire index 多个并存  const order: OpenBlock[] = []                                   // 打开顺序，决定 block-end 输出顺序  let pendingFinish: FinishReason | undefined                     // finish reason 延迟到 [DONE] 才输出  let pendingUsage: TokenUsage | undefined                        // usage 同样延迟（可能附在 finish chunk 或 trailing chunk）  function open(kind: OpenBlock['kind']): OpenBlock {             // 打开一个新块并登记顺序    const block: OpenBlock = { index: nextIndex++, kind, text: '' }    order.push(block)    return block  }  for await (const payload of payloads) {    if (payload === DONE) {                                       // [DONE] 哨兵：统一收尾      for (const block of order) {                                // 按打开顺序逐个关闭块        yield { type: 'block-end', index: block.index, block: closeBlock(block) }      }      if (pendingUsage) yield { type: 'usage', usage: pendingUsage } // 再发 usage（若收到过）      const reason = pendingFinish ?? { kind: 'stop' as const }   // 没带 finish_reason 时按 stop 处理      yield {        type: 'finish',        reason: reason.kind === 'stop' && order.length === 0      // stop 但一个块都没打开 → 退化响应          ? { kind: 'error', failure: { message: 'model returned a completed response with no content', code: EMPTY_RESPONSE_CODE } }          : reason,                                               // 否则按原 finish reason 收尾      }      return    }    let chunk: WireChunk    try {      chunk = JSON.parse(payload) as WireChunk                    // 解析 wire chunk    } catch {      throw new LlmError(`malformed SSE payload: ${payload.slice(0, 120)}`, 'MALFORMED_RESPONSE') // 坏载荷直接中止流    }    for (const choice of chunk.choices ?? []) {                   // —— 逐 delta 组装见下一段代码 ——      const delta = choice.delta

      // 推理优先：thinking mode 下 reasoning 先于 text 交错出现；      // 首个空字符串 chunk 不得开块。      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 }      }      const content = delta?.content                               // 正文增量      if (typeof content === 'string' && content.length > 0) {        if (!textBlock) {                                          // 首次出现 → 打开文本块          textBlock = open('text')          yield { type: 'block-start', index: textBlock.index, blockType: 'text' }        }        textBlock.text += content        yield { type: 'text-delta', index: textBlock.index, text: content }      }      for (const call of delta?.tool_calls ?? []) {               // 工具调用：按 wire index 分块累积        let block = toolBlocks.get(call.index)        if (!block) {                                              // 新调用 → 开块          block = open('tool-call')          toolBlocks.set(call.index, block)          yield { type: 'block-start', index: block.index, blockType: 'tool-call' }        }        if (call.id !== undefined) block.callId = call.id          // id 可能只在首个 delta 出现        if (call.function?.name !== undefined) block.name = call.function.name        const fragment = call.function?.arguments ?? ''            // arguments 是流式 JSON 片段，逐段拼接        block.text += fragment        yield { type: 'tool-call-delta', index: block.index, id: CallId(block.callId ?? ''), ...block.name !== undefined ? { name: block.name } : {}, argumentsDelta: fragment }      }      if (typeof choice.finish_reason === 'string') {        pendingFinish = mapFinishReason(choice.finish_reason)      // 只记录，延迟到 [DONE] 输出      }    }    // usage 可能附在 finish chunk 上，也可能是单独的 trailing chunk —— 始终保留最新值    if (chunk.usage) pendingUsage = mapUsage(chunk.usage)  }  throw new LlmError('SSE payload stream ended without [DONE]', 'STREAM_CLOSED') // parseSse 保证不会走到这里}

/** 拼接消息中的文本块（user / tool-result 内容用）。 */function flattenText(blocks: ContentBlock[]): string {  return blocks.filter(block => block.type === 'text').map(block => block.text).join('')}/** 序列化一条 assistant 消息（文本 + 推理 + 工具调用）。 */function serializeAssistant(message: Message): WireMessage {  const text = flattenText(message.content)                       // 正文拼接为字符串  const reasoning = message.content.filter(block => block.type === 'reasoning').map(block => block.text).join('')  const toolCalls = message.content                                // 工具调用块 → wire tool_calls    .filter(block => block.type === 'tool-call')    .map(block => ({ id: block.id, type: 'function' as const, function: { name: block.name, arguments: block.arguments } }))  return {    role: 'assistant',    // 无文本回合发 "" —— 绝不发 null。纯工具调用回合：官方样例原样回放 content（即 ""），    // 部分网关直接拒绝 null；纯推理回合：线上 API 对 null-content/无 tool_calls 的 assistant    // 消息返回 400（"content or tool_calls must be set"），且该消息会持久留在会话日志里，    // 一旦为 null 会砖掉该会话后续所有回合。    content: text,    // 官方回放规则：reasoning_content 只在工具调用回合回传；纯文本回合被忽略，丢弃以省 token    ...toolCalls.length > 0 && reasoning.length > 0 ? { reasoning_content: reasoning } : {},    ...toolCalls.length > 0 ? { tool_calls: toolCalls } : {},  }}

/** * 序列化整段对话。harness 把工具结果放在 user 角色消息里， * DeepSeek 要求独立的 {role:'tool'} 消息 —— 这里完成语义转换。 */export function serializeMessages(messages: Message[]): WireMessage[] {  const wire: WireMessage[] = []  for (const message of messages) {    assertTextOnly(message.content)                                // chat-completions 路由纯文本，图片显式拒绝    if (message.role === 'system') {      wire.push({ role: 'system', content: flattenText(message.content) })      continue    }    if (message.role === 'assistant') {      wire.push(serializeAssistant(message))                       // assistant：文本 + 推理回放 + tool_calls      continue    }    // user 角色：工具结果在 harness 词汇里挂在 user 消息上，DeepSeek 要求拆成 role:'tool'    const toolResults = message.content.filter(block => block.type === 'tool-result')    const text = flattenText(message.content)                      // user 文本部分保留    if (text.length > 0 || toolResults.length === 0) {      wire.push({ role: 'user', content: text })                   // 有文本先发 user；纯工具结果消息则跳过空 user    }    for (const result of toolResults) {                            // 每个工具结果 → 独立 tool 消息      wire.push({        role: 'tool',        tool_call_id: result.toolCallId,                           // 关联到发起调用的 id        content: flattenText(result.content) || '(no output)',     // 空输出也要有内容占位      })    }  }  return wire                                                      // 顺序保持，工具结果展开为独立条目}

/** 指数退避 + 抖动：initialDelayMs * 2^(retry-1)，封顶 maxDelayMs。 */function localDelay(config: ResolvedRetryPolicy, retry: number, random: () => number): number {  const exponent = Math.min(retry - 1, 1024)                       // 指数上限，防溢出  const exponential = Math.min(config.initialDelayMs * 2 ** exponent, config.maxDelayMs)  const jitter = 1 - config.jitterRatio + 2 * config.jitterRatio * random() // ±jitterRatio 随机抖动，打散重试风暴  return Math.min(exponential * jitter, config.maxDelayMs)         // 最终仍不超过上限}/** agent/request-error 事件处理器：按 provider 绑定的策略决定是否重试。 */async function recover(  { agent, turn, step, provider, failure, retryPolicy: policy, signal }: Parameters<Events['agent/request-error']>[0],  next: () => Promise<RequestErrorAction>,                        // waterfall 下游（其他恢复插件）): Promise<RequestErrorAction> {  if (policy === undefined) return next()                          // 该 provider 没绑策略 → 交给下游  if (policy.mode === 'always') {                                  // always：重试所有失败直到成功/取消/销毁    if (signal.aborted || lifetime.signal.aborted) return          // 已取消则直接放弃    const downstream = await settleDownstream(next)                // 先问下游（错误被隔离，不炸插件）    if (downstream.type === 'decision' && downstream.decision?.kind === 'retry') {      return downstream.decision                                   // 下游已决定重试 → 尊重它    }  } else if (!policy.retryableCodes.includes(failure.code)) {      // normal：错误码不在可重试清单 → 交给下游    return next()  }  const policyKey = retryPolicyKey(policy)                         // 策略指纹：同 turn/step/provider/policy 才累计次数  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() // normal：达到上限 → 放弃  const retry = previousRetry + 1  const retryId = priorPolicyRetry?.data.retryId ?? RetryId(randomUUID())           // 同一轮重试链复用 id，新链生成 UUID  let delayMs: number  if (failure.providerRetryAfterMs !== undefined                   // provider 给了 retry-after（限流场景）    && Number.isFinite(failure.providerRetryAfterMs) && failure.providerRetryAfterMs > 0) {    if (failure.providerRetryAfterMs > policy.maxDelayMs) {      if (policy.mode === 'normal') return next()                 // normal：超出上限 → 放弃      delayMs = localDelay(policy, retry, random)                 // always：按本地退避走    } else {      delayMs = failure.providerRetryAfterMs                      // 在范围内 → 尊重 provider 的等待时间    }  } else {    delayMs = localDelay(policy, retry, random)                   // 无 retry-after → 指数退避 + 抖动  }  return backoff(agent, turn, step, failure, provider, policy, policyKey, retry, retryId, delayMs, signal)  // backoff：先持久化 llm/retry 事件（可取消等待前落盘），再执行可中断的延迟，最后返回 {kind:'retry'}}

/** 把适配器抛出的任意值归一化为 LlmFailure 结构体（终端 finish chunk 用）。 */export function normalizeLlmFailure(value: unknown): LlmFailure {  const error = value instanceof Error                             // 非 Error 抛出（字符串/对象）→ 包成 HarnessError    ? value    : new HarnessError(thrownMessage(value), 'UNKNOWN', { cause: value })  // 跨包拷贝保留自有数据但不保留类身份：只有两个自有属性都通过校验才信任携带的事实  const carried = ownFailureSnapshot(error)                        // 读 error.failure（防御式，见下）  if (carried !== undefined && carried.code === ownErrorCode(error)) return carried  return Object.freeze({                                           // 兜底：只保留 message + code    message: errorMessage(error),    code: harnessErrorCode(error),                                  // 非 HarnessError → 'UNKNOWN'（第三方 SDK 码不属我们的分类）  })}/** 用 getOwnPropertyDescriptor 读自有数据属性，绝不触发 SDK 定义的 getter。 */function ownFailureSnapshot(error: Error): LlmFailure | undefined {  try {    const descriptor = Object.getOwnPropertyDescriptor(error, 'failure') // 直接 error.failure 可能执行恶意/意外逻辑    return descriptor !== undefined && 'value' in descriptor       // 只认自有数据属性（非 accessor）      ? failureSnapshot(descriptor.value)                           // 校验 message/code/status/retryAfter/requestId 后冻结返回      : undefined  } catch (_sdkPropertyTrap) {    return undefined                                                // 任何异常都视为"没有携带事实"，走兜底路径  }}

LLM 通信数据流

🔹 Agent Loop → 构造 GenerateOptions（消息、工具、采样参数）

🔹 LlmRuntime.stream() → 通过 Cordis waterfall 分发 llm/stream 事件

🔹 LlmRuntime.adapterStream() → 适配器选型 + 异常隔离

🔹 DeepSeekAdapter.stream() → 配置热加载 + 三层超时控制

🔹 serializeRequest() → Harness 消息 → OpenAI 兼容 JSON

🔹 fetch() → HTTP POST 到 /chat/completions

🔹 parseSse() → 字节流 → SSE 数据载荷

🔹 translate() → JSON 载荷 → StreamChunk 协议

🔹 llm-retry 插件 → 监听 agent/request-error，执行重试策略