/** * LLMBase —— 适配器三段式的「公共一处」。 * * 公共能力全在这里、且只在这里(生命周期编排): * - attempt 生命周期(acquire→fetch→消费流→release)、每 attempt 独立超时与组合 signal * - SSE 帧切分外移至 sse.js、mime 魔数外移至 media.js,本文件一行 import 接入、不含其逻辑 * - 增量回调编排(text→on_content、reasoning→on_thinking、tool_call→on_tool_call) * - usage 汇总与 reasoning/text 累积 * - 重试退避(瞬态错误重试,经 @drama/service-call 纯函数判定) * - trace(经注入的 tracePort,缺省 no-op;本包不写盘、不读全局上下文) * - 多模态消息中间态构建(images/videos → 统一中间态,交由子类落各厂商形态) * * 子类(providers/llm..js)只覆写两件事: * - async chat(req, opts):拼本厂商 body/headers,调 this.request(...) * - messageToValue(rawFrame):把本厂商 SSE 帧解析成统一 { type, ... } * * 边界纪律:构造里不建连接、不 mkdir、不 appendFile;apiKey/baseUrl 不静态持有, * 每次 request 从注入的 keyProvider.acquire() 取、finally release()。 */ import { LLMError } from './errors.js' import { mapHttpStatus, mapThrownError, isAbortError, mapChatFinishReason } from './map-error.js' import { iterateSse } from './sse.js' import { sniffMime, parseDataUrl } from './media.js' import { normalizeResponseFormat } from './response-format.js' import { DEFAULT_RETRY_POLICY, shouldRetry, backoffDelayMs, isRetryableReason, } from '@drama/service-call' /** 外部 abort reason 常量(与 map-error.js 的 'aborted' 对齐,消除魔法字符串)。 */ const ABORTED_REASON = 'aborted' export const EMPTY_USAGE = { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0, cached_tokens: 0, cache_write_tokens: 0, reasoning_tokens: 0, } /** * @typedef {import('./map-error.js').LLMErrorReason} LLMErrorReason * @typedef {import('@drama/service-call').RetryPolicy} RetryPolicy */ /** * 多模态归一后的单一形态:恰含 (mime+data) 或 url 其一。 * @typedef {object} MediaPart * @property {'image'|'video'} kind * @property {string} [mime] 本地字节的 mime(sniff/dataUrl 得),与 data 同时存在。 * @property {string} [data] base64 负载。 * @property {string} [url] 远端直传 URL(无本地 mime)。 */ /** * 统一中间帧(messageToValue 的产物)。messageToValue 可返回单个 ParsedValue、其数组、或 null; * 同帧多产物(如 Gemini 同帧 text+usage、OpenAI 同帧多路 tool_call 分片)以数组返回,base 展开消费。 * @typedef {object} ParsedValue * @property {'text'|'reasoning'|'tool_call'|'tool_call_delta'|'usage'|'metadata'|'finish'} type * @property {string} [text] type==='text'|'reasoning' 时的增量。 * @property {Array} [tool_calls] type==='tool_call' 时的完整(中性)调用列表。 * @property {number} [index] type==='tool_call_delta' 时的分片归属序号(同一 index 聚合为一个调用)。 * @property {string} [id] type==='tool_call_delta' 时的调用 id(首帧给出)。 * @property {string} [name] type==='tool_call_delta' 时的工具名(首帧给出)。 * @property {string} [args_fragment] type==='tool_call_delta' 时的 arguments JSON 分片(跨帧拼接)。 * @property {Partial} [usage] * @property {string} [provider_request_id] * @property {string} [service_tier] * @property {string|null} [finish_reason] */ /** * request() 返回的中性聚合结果(不含任何厂商专有字段)。 * @typedef {object} StreamResult * @property {string} content * @property {string} [reasoning] * @property {Array} [tool_calls] * @property {import('./index.js').ChatUsage} usage * @property {string|null} provider_request_id * @property {string|null} service_tier * @property {string|null} [finish_reason] * @property {{ ttft_ms:number|null, duration_ms:number }} [timings] * 计时(仅成功路径有):ttft_ms 为首个 content/reasoning 帧相对 startedAt 的偏移(无帧则 null), * duration_ms 为流消费完成相对 startedAt 的偏移。 */ export class LLMBase { /** * @param {object} cfg * @param {import('./index.js').KeyProvider} cfg.keyProvider * @param {string} cfg.provider api_type,决定协议子类与 LLMError.provider。 * @param {string} [cfg.vendor] keyProvider 隔离维度(厂商身份);缺省回落 provider。 * @param {string} [cfg.model] 默认模型。 * @param {import('./index.js').TracePort} cfg.tracePort * @param {RetryPolicy} [cfg.retryPolicy] * @param {() => number} [cfg.nowMs] 注入时钟(计时用);缺省 Date.now。 */ constructor({ keyProvider, provider, vendor, model, tracePort, retryPolicy, nowMs }) { this.keyProvider = keyProvider this.provider = provider this.vendor = vendor ?? provider this.model = model this.tracePort = tracePort this.retryPolicy = retryPolicy ?? DEFAULT_RETRY_POLICY // 全局 fetch,无注入缝:测试打真实 HTTP 上游(base_url 指向本地真服务),不再注入假 fetch。 this.httpFetch = fetch this.nowMs = nowMs ?? Date.now } /** * 子类必须覆写:拼本厂商请求并返回中性 StreamResult。 * @param {import('./index.js').ChatRequest} _req * @param {import('./index.js').ChatOpts} [_opts] * @returns {Promise} */ async chat(_req, _opts) { throw new Error('LLMBase.chat must be overridden by a provider subclass') } /** * 子类必须覆写:把一帧厂商 SSE data 的已解析 JSON 转成统一 ParsedValue(或 null 跳过)。 * @param {any} _rawFrame * @returns {ParsedValue | null} */ messageToValue(_rawFrame) { throw new Error('LLMBase.messageToValue must be overridden by a provider subclass') } /** * 能力声明钩子:本 provider 是否支持视频输入。缺省 false(openai/anthropic/moonshot/azure_foundry), * 视频厂商(gemini/volcengine)覆写返回 true。与 supportsThinkingParam() 同为能力钩子, * 非厂商 if 分支——video 护栏在 request() 单点消费它,基类零厂商分支。 * @protected * @returns {boolean} */ supportsVideo() { return false } /** * 多模态归一入口(唯一 mime 归一点):把 images/videos(Buffer|string)按 MECE 四态归一为 * 单一 MediaPart 形态(恰含 mime+data 或 url 其一),交由子类落各厂商消息形态。 * 落地处零 mime 回退:未识别字节在此一次性 badRequest,子类信任 part.mime 非空。 * @param {Array|undefined} items * @param {'image'|'video'} kind * @returns {Array} */ buildMediaParts(items, kind) { if (!items || items.length === 0) return [] return items.map((it) => { if (Buffer.isBuffer(it)) { const mime = sniffMime(it) if (mime == null) throw this._mediaError(kind) return { kind, mime, data: it.toString('base64') } } if (it.startsWith('data:')) { const parsed = parseDataUrl(it) if (parsed == null) throw this._mediaError(kind) return { kind, mime: parsed.mime, data: parsed.data } } if (it.startsWith('http')) return { kind, url: it } const mime = sniffMime(Buffer.from(it, 'base64')) if (mime == null) throw this._mediaError(kind) return { kind, mime, data: it } }) } /** * @private * @param {'image'|'video'} kind * @returns {LLMError} */ _mediaError(kind) { return new LLMError({ provider: this.provider, reason: 'badRequest', retryable: false, message: `[${this.provider}] unrecognized ${kind} bytes (no mime magic match)`, }) } /** * responseFormat 分派表钩子:provider 覆写返回「归一 type → 落地函数」表。 * 缺省空表——任何 responseFormat 都由 applyResponseFormat 统一 badRequest(不支持结构化输出)。 * 支持厂商只声明自己能表达的 type(如 moonshot 仅 json_object),不能表达的 type 天然落 badRequest, * 无需各写 if-else。 * @protected * @returns {Record, headers:Record, rf:{schema?:object, name?:string}) => void>} */ responseFormatHandlers() { return {} } /** * responseFormat(structured output)唯一落地入口:入口一次归一 → 按归一 type 查 provider 分派表 → * 命中即落地,未命中显式 badRequest。子类零 if-else、零静默忽略;每个 provider 的 chat 调它一次(null 时 no-op)。 * @protected * @param {Record} body * @param {Record} headers * @param {object|undefined} responseFormat * @returns {void} */ applyResponseFormat(body, headers, responseFormat) { if (responseFormat == null) return const normalized = normalizeResponseFormat(responseFormat, this.provider) const handler = this.responseFormatHandlers()[normalized.type] if (!handler) { throw new LLMError({ provider: this.provider, reason: 'badRequest', retryable: false, message: `[${this.provider}] does not support responseFormat type '${normalized.type}'`, }) } handler(body, headers, normalized) } /** * 公共请求执行:acquire key → 带重试地发请求并解析流 → reportUsage/reportRateLimited → release。 * 子类调它,传入已拼好的「发请求闭包」`doFetch(lease)`,本方法负责其余全部编排。 * * @param {object} args * @param {(lease: import('./index.js').KeyLease) => Promise} args.doFetch * 子类用 lease.apiKey/baseUrl 拼 body/headers 并发 fetch,返回原始 Response。 * @param {import('./index.js').ChatRequest} args.req * @param {import('./index.js').ChatOpts} [args.opts] * @param {string} [args.affinity_key] * @returns {Promise} */ async request({ doFetch, req, opts = {}, affinity_key }) { // 视频护栏(唯一点):所有子类 chat 都经此。缺省 supportsVideo()===false 覆盖 // openai/anthropic/moonshot/azure_foundry;视频厂商各自覆写返回 true。基类零厂商分支。 if (req.videos?.length && !this.supportsVideo()) { throw new LLMError({ provider: this.provider, reason: 'badRequest', retryable: false, message: `[${this.provider}] video input is not supported by this provider`, }) } const span = this.tracePort.startSpan(`llm.chat:${this.provider}`) span.setAttr('provider', this.provider) span.setAttr('model', req.model ?? this.model ?? '') // sessionId 一次入口归一:有值才落 span 属性(对齐 gen_ai 语义约定),无值不写占位。 const sessionId = opts.trace_ctx?.session_id if (sessionId) span.setAttr('gen_ai.conversation.id', sessionId) // 计时起点:整个 request(含重试)相对它测 ttft / duration,交由 _consumeStream 采样。 const startedAt = this.nowMs() // 单一重试循环:attempt 从 0(首次)起算;每次失败按 @drama/service-call 纯函数判定是否再试。 let attempt = 0 try { for (;;) { if (opts.signal?.aborted) { throw new LLMError({ provider: this.provider, reason: ABORTED_REASON, retryable: false }) } try { return await this._attempt({ doFetch, req, opts, affinity_key, span, startedAt }) } catch (err) { const mapped = this._toLLMError(err) // 外部 abort 不算可重试错误、不重试、不告警;内部 timeout 不落此短路(reason 不同),正常进重试。 if (isAbortError(err) || mapped.reason === ABORTED_REASON) { span.addEvent('aborted', {}) throw mapped } attempt += 1 span.addEvent('error', { reason: mapped.reason, attempt }) if (!shouldRetry(attempt, mapped.retryable && isRetryableReason(mapped.reason), this.retryPolicy)) { throw mapped } await sleep(backoffDelayMs(attempt, this.retryPolicy), opts.signal) } } } finally { span.end() } } /** * 单次尝试:acquire key → doFetch → 非 2xx 归一并上报 → 解析流 → reportUsage → 必 release。 * 任何失败都抛(由 request 的循环决定是否重试)。 * @private * @param {object} a * @param {(lease: import('./index.js').KeyLease) => Promise} a.doFetch * @param {import('./index.js').ChatRequest} a.req * @param {import('./index.js').ChatOpts} a.opts * @param {string} [a.affinity_key] * @param {import('./index.js').Span} a.span * @param {number} a.startedAt * @returns {Promise} */ async _attempt({ doFetch, req, opts, affinity_key, span, startedAt }) { const lease = await this.keyProvider.acquire({ vendor: `text:${this.vendor}`, model: req.model ?? this.model, affinity_key, }) // 这次 attempt 是哪把 key 服务的——审计要归因(认证失败时尤其),故 lease.keyId 必须离开这里。 // 本包不认识 keyrouter:keyId 只是 lease 上的一个值,经回调原样交给上层,不做解释。 opts.on_key_lease?.(lease.key_id) // 每 attempt 独立计时:内部 AbortController 用 TimeoutError 中止,AbortSignal.any 组合外部 opts.signal。 const ac = new AbortController() const signals = opts.signal ? [ac.signal, opts.signal] : [ac.signal] const attemptSignal = AbortSignal.any(signals) const timer = opts.timeout_ms ? setTimeout(() => ac.abort(new DOMException('llm attempt timeout', 'TimeoutError')), opts.timeout_ms) : null try { const res = await doFetch(lease, { signal: attemptSignal }) if (!res.ok) { const { reason, retryable } = mapHttpStatus(res.status) const bodyText = await safeReadText(res) const body = httpErrorBody(res, bodyText) const errorInfo = this.httpErrorInfo(res, body, reason) if (reason === 'rateLimit') this.keyProvider.report_rate_limited(lease.key_id, retryAfterMs(res)) else if (reason === 'auth') this.keyProvider.report_error(lease.key_id, 'auth') else if (reason === 'transient') this.keyProvider.report_error(lease.key_id, 'transient') throw new LLMError({ provider: this.provider, reason, retryable, cause: body, ...errorInfo, detail: { type: 'http', status: res.status, headers: Object.fromEntries(res.headers.entries()), body, }, }) } const result = await this._consumeStream(res, attemptSignal, opts, span, startedAt) this.keyProvider.report_usage(lease.key_id, result.usage) span.addEvent('usage', result.usage) return result } finally { if (timer) clearTimeout(timer) // release 必被 await(兼容同步/异步);失败不覆盖返回值,仅 span 落痕。 try { await lease.release() } catch (e) { span.addEvent('release_error', { message: String(e?.message ?? e) }) } } } /** * 消费 SSE 流:分帧 → messageToValue → 累积 text/reasoning/tool_calls/usage + 触发回调。 * @private * @param {Response} res * @param {AbortSignal} attemptSignal 每 attempt 的组合信号(内部超时 + 外部 abort);中途可中断读循环。 * @param {import('./index.js').ChatOpts} opts * @param {import('./index.js').Span} span * @param {number} startedAt 计时起点(this.nowMs() 域);ttft/duration 相对它测。 * @returns {Promise} */ async _consumeStream(res, attemptSignal, opts, span, startedAt) { let content = '' let reasoning = '' /** @type {number | undefined} 首个 content/reasoning 帧的到达时刻(TTFT 采样点,只采一次)。 */ let firstFrameAt /** @type {Array} */ const tool_calls = [] /** * 分片 tool_call 聚合缓冲(协议无关,base 独有):index → { id?, name?, argsText }。 * 真实流的 function.arguments 跨多帧分片,provider 逐帧发 tool_call_delta,base 按 index 聚合, * 流结束时对每个完整调用一次 JSON.parse 得中性 { id, name, args(object) }。每次 _consumeStream 独立、不跨 attempt。 * @type {Map} */ const toolCallBuffers = new Map() /** @type {StreamResult['usage']} */ let usage = { ...EMPTY_USAGE } let provider_request_id = null let service_tier = null let finish_reason = null // 只有输出已经通过回调交付给调用方,才存在不可撤销的流式副作用;无回调时结果尚未离开 // Promise,失败后仍可安全重试。已交付后禁止 retry/fallback,避免混入两个 attempt 的内容。 let emittedOutput = false try { for await (const frame of iterateSse(res, attemptSignal)) { opts.on_raw_frame?.(frame) // messageToValue 单值/数组统一归一为数组,在入口一处展开(同帧多产物:text+usage、多路 tool_call 分片)。 for (const value of toValues(this.messageToValue(frame))) { switch (value.type) { case 'text': if (value.text) { content += value.text firstFrameAt ??= this.nowMs() if (opts.on_content) { emittedOutput = true opts.on_content(value.text) } } break case 'reasoning': if (value.text) { reasoning += value.text firstFrameAt ??= this.nowMs() if (opts.on_thinking) { emittedOutput = true opts.on_thinking(value.text) } } break case 'tool_call': // 完整(非分片)中性调用:provider 已给 { id, name, args(object) }(整帧下发的厂商)。 if (value.tool_calls?.length) { tool_calls.push(...value.tool_calls) if (opts.on_tool_call) { emittedOutput = true opts.on_tool_call(value.tool_calls) } } break case 'tool_call_delta': bufferToolCallDelta(toolCallBuffers, value) break case 'usage': if (value.usage) usage = { ...usage, ...value.usage } break case 'metadata': provider_request_id = value.provider_request_id service_tier = value.service_tier break case 'finish': finish_reason = normalizeFinishReason(value.finish_reason) break default: throw new Error(`Unhandled parsed value type: ${value.type}`) } } } } catch (err) { if (!emittedOutput) throw err const mapped = this._toLLMError(err) throw new LLMError({ provider: this.provider, reason: mapped.reason, retryable: false, cause: err, message: `[${this.provider}] stream failed after output started: ${mapped.reason}`, detail: { type: 'stream', error: mapped.detail }, }) } // 流结束:把聚合完成的分片 tool_call 落成中性 { id, name, args(object) },一次性回调。 const aggregated = finalizeToolCalls(toolCallBuffers) if (aggregated.length) { tool_calls.push(...aggregated) opts.on_tool_call?.(aggregated) } span.addEvent('content', { length: content.length }) // Anthropic 分帧只给输入和输出分量,流结束时补齐总量。 if (!usage.total_tokens) { usage = { ...usage, total_tokens: usage.prompt_tokens + usage.completion_tokens, } } const finishError = chatFinishReasonError({ provider: this.provider, finish_reason, content_length: content.length, usage, provider_request_id, }) if (finishError) { if (emittedOutput) { throw new LLMError({ provider: this.provider, reason: finishError.reason, retryable: false, cause: finishError, message: `[${this.provider}] stream ended with finish_reason=${finish_reason} after output started`, detail: { type: 'stream_finish_reason', finish_reason, content_length: content.length, usage, }, provider_request_id, }) } throw finishError } /** @type {StreamResult} */ const out = { content, usage, provider_request_id, service_tier, finish_reason } if (reasoning) out.reasoning = reasoning if (tool_calls.length) out.tool_calls = tool_calls // 计量取整只在出口一次:ttft 仅在有帧时有值,duration 恒为流消费总时长。 out.timings = { ttft_ms: firstFrameAt === undefined ? null : firstFrameAt - startedAt, duration_ms: this.nowMs() - startedAt, } return out } /** * 把任意抛出物归一为 LLMError(已是 LLMError 则原样)。 * @private * @param {unknown} err * @returns {LLMError} */ _toLLMError(err) { if (err instanceof LLMError) return err const { reason, retryable } = mapThrownError(err) if (!(err instanceof Error)) throw new Error(`[${this.provider}] provider 抛出了非 Error 值`) return new LLMError({ provider: this.provider, reason, retryable, cause: err, message: err.name === 'KeyRouterError' ? `[${this.provider}] ${err.message}` : undefined, detail: { type: 'runtime', name: err.name, message: err.message }, }) } httpErrorInfo(res, _body, reason) { return { message: `[${this.provider}] HTTP ${res.status}: ${reason}`, provider_request_id: res.headers.get('x-request-id'), } } } function normalizeFinishReason(reason) { if (reason === null || typeof reason === 'string') return reason throw new Error(`finish_reason must be string or null, got ${typeof reason}`) } function chatFinishReasonError({ provider, finish_reason, content_length, usage, provider_request_id }) { const mapped = mapChatFinishReason(finish_reason) if (mapped === null) return null return new LLMError({ provider, reason: mapped.reason, retryable: mapped.retryable, message: `[${provider}] chat completion ended with finish_reason=${finish_reason}`, detail: { type: 'finish_reason', finish_reason, content_length, usage, }, provider_request_id, }) } /** * messageToValue 返回归一为 ParsedValue 数组(单值/数组/null 三态一处收口),消费端零形态判断。 * @param {import('./base.js').ParsedValue | Array | null | undefined} v * @returns {Array} */ function toValues(v) { if (v == null) return [] return Array.isArray(v) ? v : [v] } /** * 把一帧 tool_call_delta 并入按 index 聚合的缓冲:id/name 首帧给出(合并),args_fragment 跨帧拼接。 * @param {Map} buffers * @param {{ index?:number, id?:string, name?:string, args_fragment?:string }} delta * @returns {void} */ function bufferToolCallDelta(buffers, delta) { const index = delta.index ?? 0 let buf = buffers.get(index) if (!buf) { buf = { argsText: '' } buffers.set(index, buf) } if (delta.id != null) buf.id = delta.id if (delta.name != null) buf.name = delta.name if (delta.args_fragment) buf.argsText += delta.args_fragment } /** * 聚合缓冲 → 中性 tool_call 列表:按 index 升序,每个 argsText 做一次 JSON.parse(args 恒为对象)。 * args 为空时归一为 {};非法 JSON 由 JSON.parse 自然抛出(真实上游不会给坏 JSON,坏则是真错误)。 * @param {Map} buffers * @returns {Array<{ id?:string, name?:string, args:object }>} */ function finalizeToolCalls(buffers) { return [...buffers.entries()] .sort((a, b) => a[0] - b[0]) .map(([, buf]) => ({ id: buf.id, name: buf.name, args: JSON.parse(buf.argsText || '{}') })) } /** * 安全读取响应文本(错误体),失败返回空串。 * @param {Response} res * @returns {Promise} */ async function safeReadText(res) { try { return await res.text() } catch { return '' } } function httpErrorBody(res, bodyText) { const contentType = res.headers.get('content-type') return contentType !== null && contentType.startsWith('application/json') ? JSON.parse(bodyText) : bodyText } /** * 从 429 响应取 Retry-After(秒)→ 毫秒;无则 undefined。 * @param {Response} res * @returns {number | undefined} */ function retryAfterMs(res) { const h = res.headers?.get?.('retry-after') if (!h) return undefined const sec = Number(h) return Number.isFinite(sec) ? sec * 1000 : undefined } /** * 可被 abort 的 sleep。 * @param {number} ms * @param {AbortSignal} [signal] * @returns {Promise} */ function sleep(ms, signal) { return new Promise((resolve, reject) => { if (signal?.aborted) { reject(abortReason(signal)) return } const t = setTimeout(() => { signal?.removeEventListener?.('abort', onAbort) resolve() }, ms) function onAbort() { clearTimeout(t) reject(abortReason(signal)) } signal?.addEventListener?.('abort', onAbort, { once: true }) }) } /** * 中止原因:优先透传 signal.reason(内部超时的 TimeoutError / 外部的 AbortError),缺省造 AbortError。 * @param {AbortSignal} [signal] * @returns {unknown} */ function abortReason(signal) { return signal?.reason ?? Object.assign(new Error('aborted'), { name: 'AbortError' }) }