AI调用记录管理模块

CREATE TABLE ai_call_logs (
  id              BIGINT PRIMARY KEY AUTO_INCREMENT,
  call_id         VARCHAR(64) NOT NULL UNIQUE,       -- 全局唯一调用 ID
  tenant_id       VARCHAR(64) NOT NULL,              -- 租户 ID
  user_id         VARCHAR(64),                       -- 用户 ID
  feature_module  VARCHAR(128) NOT NULL,             -- 功能模块标识
  provider        VARCHAR(32) NOT NULL,              -- 模型提供商：openai / anthropic / ...
  model           VARCHAR(64) NOT NULL,              -- 模型名称：gpt-4o / claude-sonnet-4-20250514 / ...
  request_type    VARCHAR(16) NOT NULL DEFAULT 'chat', -- chat / embedding / image / audio
  input_tokens    INT NOT NULL DEFAULT 0,
  output_tokens   INT NOT NULL DEFAULT 0,
  total_tokens    INT GENERATED ALWAYS AS (input_tokens + output_tokens) STORED,
  cost_usd        DECIMAL(12, 6) NOT NULL DEFAULT 0, -- 单次调用成本（美元）
  latency_ms      INT,                               -- 端到端延迟
  ttft_ms         INT,                               -- 首 Token 延迟
  status          VARCHAR(16) NOT NULL,              -- success / error / timeout / rate_limited
  error_code      VARCHAR(64),                       -- 错误码
  error_message   TEXT,                              -- 错误信息
  prompt_hash     VARCHAR(64),                       -- prompt 哈希（用于去重和聚合）
  metadata        JSON,                              -- 扩展字段：temperature、top_p 等
  created_at      TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
 
  INDEX idx_tenant_time (tenant_id, created_at DESC),
  INDEX idx_model_time (model, created_at DESC),
  INDEX idx_feature_time (feature_module, created_at DESC),
  INDEX idx_status_time (status, created_at DESC),
  INDEX idx_user_time (user_id, created_at DESC)
);

CREATE TABLE ai_model_pricing (
  id              INT PRIMARY KEY AUTO_INCREMENT,
  provider        VARCHAR(32) NOT NULL,
  model           VARCHAR(64) NOT NULL,
  request_type    VARCHAR(16) NOT NULL DEFAULT 'chat',
  input_price     DECIMAL(12, 6) NOT NULL,  -- 每百万 input token 价格（美元）
  output_price    DECIMAL(12, 6) NOT NULL,  -- 每百万 output token 价格（美元）
  effective_from  DATE NOT NULL,            -- 价格生效日期
  effective_to    DATE,                     -- 价格失效日期（NULL 表示当前有效）
  currency        VARCHAR(8) NOT NULL DEFAULT 'USD',
 
  UNIQUE INDEX uk_model_effective (provider, model, request_type, effective_from)
);

CREATE TABLE ai_call_daily_stats (
  id              BIGINT PRIMARY KEY AUTO_INCREMENT,
  stat_date       DATE NOT NULL,
  tenant_id       VARCHAR(64) NOT NULL,
  feature_module  VARCHAR(128),
  provider        VARCHAR(32) NOT NULL,
  model           VARCHAR(64) NOT NULL,
  status          VARCHAR(16) NOT NULL,
  call_count      INT NOT NULL DEFAULT 0,
  total_input_tokens  BIGINT NOT NULL DEFAULT 0,
  total_output_tokens BIGINT NOT NULL DEFAULT 0,
  total_cost_usd  DECIMAL(14, 6) NOT NULL DEFAULT 0,
  avg_latency_ms  INT,
  p95_latency_ms  INT,
  p99_latency_ms  INT,
  error_count     INT NOT NULL DEFAULT 0,
 
  UNIQUE INDEX uk_daily_dim (stat_date, tenant_id, feature_module, provider, model, status)
);

SELECT
  feature_module,
  DATE_FORMAT(created_at, '%Y-%m') AS month,
  COUNT(*) AS call_count,
  SUM(input_tokens) AS total_input_tokens,
  SUM(output_tokens) AS total_output_tokens,
  SUM(cost_usd) AS total_cost,
  AVG(latency_ms) AS avg_latency
FROM ai_call_logs
WHERE tenant_id = ?
  AND created_at >= DATE_SUB(CURDATE(), INTERVAL 3 MONTH)
GROUP BY feature_module, DATE_FORMAT(created_at, '%Y-%m')
ORDER BY total_cost DESC;

SELECT
  DATE(created_at) AS stat_date,
  COUNT(*) AS total_calls,
  SUM(CASE WHEN status != 'success' THEN 1 ELSE 0 END) AS error_count,
  ROUND(SUM(CASE WHEN status != 'success' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) AS error_rate
FROM ai_call_logs
WHERE tenant_id = ?
  AND created_at >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)
GROUP BY DATE(created_at)
ORDER BY stat_date;

// AI 调用中间件 —— 自动采集调用指标
interface CallMetrics {
  callId: string
  tenantId: string
  userId?: string
  featureModule: string
  provider: string
  model: string
  startTime: number
  firstTokenTime?: number  // 流式响应专用
  endTime: number
  inputTokens: number
  outputTokens: number
  status: 'success' | 'error' | 'timeout' | 'rate_limited'
  errorCode?: string
  errorMessage?: string
  metadata?: Record<string, unknown>
}
 
class AIMetricsCollector {
  private buffer: CallMetrics[] = []
  private flushInterval = 5000 // 5 秒批量写入
 
  async record(metrics: CallMetrics): Promise<void> {
    // 计算衍生指标
    const latency = metrics.endTime - metrics.startTime
    const ttft = metrics.firstTokenTime
      ? metrics.firstTokenTime - metrics.startTime
      : null
 
    // 计算成本
    const cost = await this.calculateCost(
      metrics.provider,
      metrics.model,
      metrics.inputTokens,
      metrics.outputTokens
    )
 
    // 写入缓冲区，批量落库
    this.buffer.push({
      ...metrics,
      latencyMs: latency,
      ttftMs: ttft ?? undefined,
      costUsd: cost,
    })
  }
 
  private async flush(): Promise<void> {
    if (this.buffer.length === 0) return
 
    const batch = this.buffer.splice(0, this.buffer.length)
 
    // 批量写入主表
    await db.insert(aiCallLogs).values(batch)
 
    // 异步更新聚合表
    await this.updateAggregates(batch)
  }
}

async function calculateCost(
  provider: string,
  model: string,
  inputTokens: number,
  outputTokens: number
): Promise<number> {
  // 查询当前有效的模型价格
  const pricing = await db.query.aiModelPricing.findFirst({
    where: and(
      eq(aiModelPricing.provider, provider),
      eq(aiModelPricing.model, model),
      lte(aiModelPricing.effectiveFrom, new Date()),
      or(
        isNull(aiModelPricing.effectiveTo),
        gte(aiModelPricing.effectiveTo, new Date())
      )
    ),
  })
 
  if (!pricing) return 0
 
  // 价格单位：每百万 token
  const inputCost = (inputTokens / 1_000_000) * pricing.inputPrice
  const outputCost = (outputTokens / 1_000_000) * pricing.outputPrice
 
  return Number((inputCost + outputCost).toFixed(6))
}

// Hono 路由 —— 调用列表查询
import { Hono } from 'hono'
import { eq, and, gte, lte, desc, sql } from 'drizzle-orm'
 
const callLogRouter = new Hono()
 
callLogRouter.get('/api/admin/ai-call-logs', async (c) => {
  const query = c.req.query()
  const page = Number(query.page) || 1
  const pageSize = Math.min(Number(query.pageSize) || 50, 200)
  const offset = (page - 1) * pageSize
 
  // 构建动态筛选条件
  const conditions = []
 
  if (query.tenantId) {
    conditions.push(eq(aiCallLogs.tenantId, query.tenantId))
  }
  if (query.model) {
    conditions.push(eq(aiCallLogs.model, query.model))
  }
  if (query.status) {
    conditions.push(eq(aiCallLogs.status, query.status))
  }
  if (query.featureModule) {
    conditions.push(eq(aiCallLogs.featureModule, query.featureModule))
  }
  if (query.startTime) {
    conditions.push(gte(aiCallLogs.createdAt, new Date(query.startTime)))
  }
  if (query.endTime) {
    conditions.push(lte(aiCallLogs.createdAt, new Date(query.endTime)))
  }
 
  const where = conditions.length > 0 ? and(...conditions) : undefined
 
  // 并行查询数据和总数
  const [data, totalResult] = await Promise.all([
    db.query.aiCallLogs.findMany({
      where,
      orderBy: [desc(aiCallLogs.createdAt)],
      limit: pageSize,
      offset,
      // 不返回完整的 error_message 和 metadata，减少传输量
      columns: {
        errorMessage: false,
        metadata: false,
      },
    }),
    db
      .select({ count: sql<number>`count(*)` })
      .from(aiCallLogs)
      .where(where),
  ])
 
  return c.json({
    data,
    pagination: {
      page,
      pageSize,
      total: totalResult[0].count,
      totalPages: Math.ceil(totalResult[0].count / pageSize),
    },
  })
})

// 成本统计 —— 优先查聚合表，回退到主表
callLogRouter.get('/api/admin/ai-call-logs/cost-stats', async (c) => {
  const query = c.req.query()
  const tenantId = query.tenantId
  const granularity = query.granularity || 'day' // day / week / month
 
  if (!tenantId) {
    return c.json({ error: 'tenantId is required' }, 400)
  }
 
  // 使用聚合表查询，性能可控
  const stats = await db.query.aiCallDailyStats.findMany({
    where: and(
      eq(aiCallDailyStats.tenantId, tenantId),
      gte(aiCallDailyStats.statDate, new Date(query.startDate)),
      lte(aiCallDailyStats.statDate, new Date(query.endDate))
    ),
    orderBy: [aiCallDailyStats.statDate],
  })
 
  // 按时间粒度二次聚合
  const grouped = new Map<string, {
    callCount: number
    totalCost: number
    inputTokens: number
    outputTokens: number
    errorCount: number
  }>()
 
  for (const row of stats) {
    const key = formatDate(row.statDate, granularity)
    const existing = grouped.get(key) || {
      callCount: 0,
      totalCost: 0,
      inputTokens: 0,
      outputTokens: 0,
      errorCount: 0,
    }
 
    existing.callCount += row.callCount
    existing.totalCost += Number(row.totalCostUsd)
    existing.inputTokens += Number(row.totalInputTokens)
    existing.outputTokens += Number(row.totalOutputTokens)
    existing.errorCount += row.errorCount
 
    grouped.set(key, existing)
  }
 
  return c.json({
    data: Array.from(grouped.entries()).map(([date, stats]) => ({
      date,
      ...stats,
      avgCostPerCall: stats.callCount > 0
        ? stats.totalCost / stats.callCount
        : 0,
    })),
  })
})

// 每小时增量聚合任务
async function runHourlyAggregation(): Promise<void> {
  const oneHourAgo = new Date(Date.now() - 60 * 60 * 1000)
 
  // 获取需要聚合的时间范围内的原始数据
  const rawData = await db
    .select({
      statDate: sql<Date>`DATE(${aiCallLogs.createdAt})`,
      tenantId: aiCallLogs.tenantId,
      featureModule: aiCallLogs.featureModule,
      provider: aiCallLogs.provider,
      model: aiCallLogs.model,
      status: aiCallLogs.status,
      callCount: sql<number>`count(*)`,
      totalInputTokens: sql<number>`sum(${aiCallLogs.inputTokens})`,
      totalOutputTokens: sql<number>`sum(${aiCallLogs.outputTokens})`,
      totalCost: sql<string>`sum(${aiCallLogs.costUsd})`,
      avgLatency: sql<number>`avg(${aiCallLogs.latencyMs})`,
      p95Latency: sql<number>`percentile_cont(0.95) within group (order by ${aiCallLogs.latencyMs})`,
      p99Latency: sql<number>`percentile_cont(0.99) within group (order by ${aiCallLogs.latencyMs})`,
      errorCount: sql<number>`sum(case when ${aiCallLogs.status} != 'success' then 1 else 0 end)`,
    })
    .from(aiCallLogs)
    .where(gte(aiCallLogs.createdAt, oneHourAgo))
    .groupBy(
      sql`DATE(${aiCallLogs.createdAt})`,
      aiCallLogs.tenantId,
      aiCallLogs.featureModule,
      aiCallLogs.provider,
      aiCallLogs.model,
      aiCallLogs.status
    )
 
  // Upsert 到聚合表
  for (const row of rawData) {
    await db
      .insert(aiCallDailyStats)
      .values({
        statDate: row.statDate,
        tenantId: row.tenantId,
        featureModule: row.featureModule,
        provider: row.provider,
        model: row.model,
        status: row.status,
        callCount: row.callCount,
        totalInputTokens: row.totalInputTokens,
        totalOutputTokens: row.totalOutputTokens,
        totalCostUsd: row.totalCost,
        avgLatencyMs: row.avgLatency,
        p95LatencyMs: row.p95Latency,
        p99LatencyMs: row.p99Latency,
        errorCount: row.errorCount,
      })
      .onConflictDoUpdate({
        target: [
          aiCallDailyStats.statDate,
          aiCallDailyStats.tenantId,
          aiCallDailyStats.featureModule,
          aiCallDailyStats.provider,
          aiCallDailyStats.model,
          aiCallDailyStats.status,
        ],
        set: {
          callCount: sql`${aiCallDailyStats.callCount} + ${row.callCount}`,
          totalInputTokens: sql`${aiCallDailyStats.totalInputTokens} + ${row.totalInputTokens}`,
          totalOutputTokens: sql`${aiCallDailyStats.totalOutputTokens} + ${row.totalOutputTokens}`,
          totalCostUsd: sql`${aiCallDailyStats.totalCostUsd} + ${row.totalCost}`,
        },
      })
  }
}