import { randomUUID } from 'node:crypto' import * as v from 'valibot' import { LLMError } from './errors.js' import { EMPTY_USAGE } from './base.js' import { loadRegistry, registeredProviders } from './registry.js' import { noopNotifyPort, noopTracePort } from './ports/no-op.js' const NOTIFY_FAIL_THRESHOLD = 3 const KeyProviderShape = v.object({ acquire: v.function(), report_usage: v.function(), report_rate_limited: v.function(), report_error: v.function(), }) const ChatRequestSchema = v.object({ messages: v.pipe( v.array(v.object({ role: v.picklist(['system', 'user', 'assistant', 'tool']), content: v.any(), prompt_cache_breakpoint: v.optional(v.boolean()), })), v.minLength(1), ), model: v.optional(v.string()), temperature: v.optional(v.number()), max_tokens: v.optional(v.number()), response_format: v.optional(v.any()), tools: v.optional(v.array(v.any())), thinking: v.optional(v.any()), images: v.optional(v.array(v.any())), videos: v.optional(v.array(v.any())), prompt_cache_options: v.optional(v.strictObject({ mode: v.picklist(['implicit', 'explicit']), ttl: v.optional(v.literal('30m')), })), }) /** * provider 调用、供应商成本、调用日志、积分结算的统一编排入口。 * costPort/billingPort/auditPort 要么三者全部注入,要么全部省略用于运营探针。 */ export function createLLM({ keyProvider, provider = 'openai', vendor = provider, model, costPort, billingPort, notifyPort, tracePort, auditPort, nowMs, newId, retryPolicy, }) { assertKeyProvider(keyProvider, provider) const accounting = accountingPorts({ costPort, billingPort, auditPort }) const notify = notifyPort ?? noopNotifyPort const trace = tracePort ?? noopTracePort const now = nowMs ?? Date.now const genId = newId ?? randomUUID const table = loadRegistry() if (!table[provider]) { throw new LLMError({ provider, reason: 'badRequest', retryable: false, message: `createLLM: unknown provider '${provider}'. Registered providers: [${registeredProviders().join(', ')}]`, }) } let instance let consecutiveFailures = 0 async function ensureInstance() { if (instance) return instance const module = await table[provider]() instance = new module.default({ keyProvider, provider, vendor, model, tracePort: trace, retryPolicy, nowMs: now }) return instance } async function chat(req, opts = {}) { const parsed = parseChatRequest(req, provider) const impl = await ensureInstance() const usedModel = parsed.model ?? model ?? '' const requestId = genId() const billingOperationId = opts.billing_ctx?.idempotency_key ?? requestId const startedAt = now() const sessionId = opts.trace_ctx?.session_id ?? null let failureUsage = EMPTY_USAGE let failureTtftMs = null let keyId = null const implOpts = { ...opts, on_key_lease: (id) => { keyId = id } } let result try { result = await impl.chat(parsed, implOpts) failureUsage = result.usage failureTtftMs = result.timings?.ttft_ms ?? null consecutiveFailures = 0 } catch (error) { const llmError = normalizeLlmError(error, provider) const { provider_request_id } = llmError if (accounting) { accounting.auditPort.record({ provider: vendor, model: usedModel, request_id: requestId, billing_operation_id: billingOperationId, session_id: sessionId, user_id: opts.billing_ctx?.user_id ?? null, key_id: keyId, created_at: new Date(now()), status: llmError.reason === 'aborted' ? 'aborted' : 'error', ttft_ms: failureTtftMs, duration_ms: now() - startedAt, usage: failureUsage, provider_request_id, service_tier: null, pricing_snapshot: null, supplier_cost_micros: null, currency: null, billing_mode: null, billing_policy_snapshot: null, charged_points: null, ledger_id: null, error_reason: llmError.reason, error_message: llmError.message, error_detail: llmError.detail, messages: parsed.messages, response_text: null, reasoning_text: null, mode: opts.mode ?? null, correlation_id: opts.correlation_id ?? null, attempt_index: opts.attempt_index ?? null, }) } if (llmError.reason !== 'aborted') { consecutiveFailures += 1 if (consecutiveFailures >= NOTIFY_FAIL_THRESHOLD) { consecutiveFailures = 0 await notify.notify( { kind: 'llm.consecutiveFailures', provider, reason: llmError.reason }, opts.billing_ctx, ) } } throw llmError } const { content, reasoning, tool_calls, usage, provider_request_id, service_tier, timings = { ttft_ms: null, duration_ms: now() - startedAt }, } = result const metered_usage = { kind: 'token', ...usage } const out = { content, usage, metered_usage, provider: vendor, model: usedModel, timings, request_id: requestId, billing_operation_id: billingOperationId, provider_request_id, service_tier, } if (reasoning) out.reasoning = reasoning if (tool_calls) out.tool_calls = tool_calls if (!accounting) return out const calculated = await accounting.costPort.calculate({ provider: vendor, model: usedModel, usage: metered_usage }) await accounting.auditPort.recordProviderSuccess({ provider: vendor, model: usedModel, request_id: requestId, billing_operation_id: billingOperationId, session_id: sessionId, user_id: opts.billing_ctx?.user_id ?? null, key_id: keyId, created_at: new Date(now()), status: 'ok', ...timings, usage, provider_request_id, service_tier, pricing_snapshot: calculated.pricing_snapshot, supplier_cost_micros: calculated.supplier_cost_micros, currency: calculated.currency, billing_mode: null, billing_policy_snapshot: null, charged_points: null, ledger_id: null, error_reason: null, error_message: null, error_detail: null, messages: parsed.messages, response_text: content, reasoning_text: reasoning ?? null, mode: opts.mode ?? null, correlation_id: opts.correlation_id ?? null, attempt_index: opts.attempt_index ?? null, }) const settlement = await accounting.billingPort.settle({ call_id: requestId, user_id: opts.billing_ctx?.user_id, call_ref: { type: 'llm_operation', id: billingOperationId }, supplier_cost: { supplier_cost_micros: calculated.supplier_cost_micros, currency: calculated.currency, }, policy: opts.billing_ctx?.policy, idempotency_key: billingOperationId, }) out.accounting = { call_ref: { type: 'llm_operation', id: billingOperationId }, usage: metered_usage, pricing_snapshot: calculated.pricing_snapshot, supplier_cost_micros: calculated.supplier_cost_micros, currency: calculated.currency, billing_mode: settlement.billing_mode, billing_policy_snapshot: settlement.billing_policy_snapshot, charged_points: settlement.charged_points, ledger_id: settlement.ledger_id, } return out } return { chat } } function parseChatRequest(req, provider) { try { return v.parse(ChatRequestSchema, req) } catch (cause) { throw new LLMError({ provider, reason: 'badRequest', retryable: false, cause, message: 'createLLM.chat: invalid ChatRequest', }) } } function assertKeyProvider(keyProvider, provider) { if (!keyProvider) { throw new LLMError({ provider, reason: 'badRequest', retryable: false, message: 'createLLM: keyProvider is required' }) } try { v.parse(KeyProviderShape, keyProvider) } catch (cause) { throw new LLMError({ provider, reason: 'badRequest', retryable: false, cause, message: 'createLLM: keyProvider does not satisfy the KeyProvider port shape', }) } } function accountingPorts({ costPort, billingPort, auditPort }) { const ports = [costPort, billingPort, auditPort] const present = ports.filter(Boolean).length if (present === 0) return null if (present !== ports.length) { throw new Error('createLLM: costPort、billingPort、auditPort 必须同时注入') } if (typeof costPort.calculate !== 'function') throw new Error('createLLM: costPort.calculate 必须是函数') if (typeof billingPort.settle !== 'function') throw new Error('createLLM: billingPort.settle 必须是函数') if ( typeof auditPort.record !== 'function' || typeof auditPort.recordProviderSuccess !== 'function' ) { throw new Error('createLLM: auditPort 缺少 record/recordProviderSuccess') } return { costPort, billingPort, auditPort } } function normalizeLlmError(error, provider) { if (error instanceof LLMError) return error if (!(error instanceof Error)) throw new Error(`[${provider}] provider 抛出了非 Error 值`) return new LLMError({ provider, reason: 'unknown', retryable: false, cause: error, detail: { type: 'runtime', name: error.name, message: error.message }, }) } export function createStaticKeyProvider({ api_key, base_url }) { if (!api_key) { throw new LLMError({ provider: 'static', reason: 'badRequest', retryable: false, message: 'createStaticKeyProvider: api_key is required', }) } return { async acquire() { return { api_key, base_url, account_id: null, key_id: null, release() {} } }, report_usage() {}, report_rate_limited() {}, report_error() {}, } }