宜博 / vokoworld

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() {},
  }
}