AI 项目额度与权限控制

要点

AI 项目的额度控制涉及：请求次数、Token 用量、模型权限、并发数
额度检查应该在请求进入 AI API 之前完成
Token 用量需要从 AI API 的响应中提取，不能预估
计费系统需要处理：预付费（余额）、后付费（账单）、订阅（计划）
额度耗尽时应该有明确的错误提示和升级引导
AI 项目的权限控制需要区分：模型访问、功能开关、数据范围

1. 额度类型

1.1 请求次数

限制每月/每天的请求次数：

export const quotas = pgTable('quotas', {
  id: uuid('id').primaryKey().defaultRandom(),
  userId: uuid('user_id').notNull(),
  period: text('period').notNull(),  // daily, monthly
  maxRequests: integer('max_requests').notNull(),
  usedRequests: integer('used_requests').default(0).notNull(),
  resetAt: timestamp('reset_at').notNull(),
})

1.2 Token 用量

限制每月使用的 Token 数量：

export const tokenQuotas = pgTable('token_quotas', {
  id: uuid('id').primaryKey().defaultRandom(),
  userId: uuid('user_id').notNull(),
  period: text('period').notNull(),
  maxTokens: integer('max_tokens').notNull(),
  usedTokens: integer('used_tokens').default(0).notNull(),
  resetAt: timestamp('reset_at').notNull(),
})

1.3 模型权限

不同计划可以使用不同的模型：

const MODEL_PERMISSIONS = {
  free: ['gpt-3.5-turbo', 'claude-3-haiku'],
  pro: ['gpt-3.5-turbo', 'gpt-4', 'claude-3-haiku', 'claude-3-sonnet'],
  enterprise: ['gpt-3.5-turbo', 'gpt-4', 'gpt-4-turbo', 'claude-3-haiku', 'claude-3-sonnet', 'claude-3-opus'],
}

1.4 并发限制

限制同时进行的请求数：

// Redis 记录并发数
async function checkConcurrency(userId: string, maxConcurrent: number) {
  const current = await redis.get(`concurrent:${userId}`)
 
  if (current && parseInt(current) >= maxConcurrent) {
    throw new Error('Too many concurrent requests')
  }
 
  await redis.incr(`concurrent:${userId}`)
  await redis.expire(`concurrent:${userId}`, 60)  // 1 分钟过期
}
 
async function releaseConcurrency(userId: string) {
  await redis.decr(`concurrent:${userId}`)
}

2. 额度检查

2.1 中间件

const quotaMiddleware = createMiddleware(async (c, next) => {
  const user = c.get('user')
 
  // 检查请求次数
  const requestQuota = await db.query.quotas.findFirst({
    where: and(
      eq(quotas.userId, user.id),
      eq(quotas.period, 'monthly')
    ),
  })
 
  if (requestQuota && requestQuota.usedRequests >= requestQuota.maxRequests) {
    return c.json({
      error: 'Monthly request quota exceeded',
      upgradeUrl: '/billing/upgrade',
    }, 402)
  }
 
  // 检查 Token 用量
  const tokenQuota = await db.query.tokenQuotas.findFirst({
    where: and(
      eq(tokenQuotas.userId, user.id),
      eq(tokenQuotas.period, 'monthly')
    ),
  })
 
  if (tokenQuota && tokenQuota.usedTokens >= tokenQuota.maxTokens) {
    return c.json({
      error: 'Monthly token quota exceeded',
      upgradeUrl: '/billing/upgrade',
    }, 402)
  }
 
  // 检查模型权限
  const model = c.get('model')
  const plan = user.plan || 'free'
  const allowedModels = MODEL_PERMISSIONS[plan as keyof typeof MODEL_PERMISSIONS]
 
  if (!allowedModels.includes(model)) {
    return c.json({
      error: `Model ${model} not available for ${plan} plan`,
      upgradeUrl: '/billing/upgrade',
    }, 403)
  }
 
  // 检查并发
  await checkConcurrency(user.id, plan === 'free' ? 1 : 5)
 
  await next()
})

2.2 额度更新

请求完成后更新用量：

app.post('/v1/chat/completions',
  quotaMiddleware,
  async (c) => {
    const user = c.get('user')
    const body = await c.req.json()
 
    try {
      // 调用 AI API
      const response = await callAI(body)
 
      // 提取 Token 用量
      const usage = response.usage
 
      // 更新用量
      await updateUsage(user.id, {
        requests: 1,
        promptTokens: usage.prompt_tokens,
        completionTokens: usage.completion_tokens,
        totalTokens: usage.total_tokens,
      })
 
      return c.json(response)
    } finally {
      // 释放并发
      await releaseConcurrency(user.id)
    }
  }
)

3. Token 用量提取

3.1 从响应中提取

async function callAI(body: any) {
  const response = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${process.env.OPENAI_API_KEY}`,
    },
    body: JSON.stringify(body),
  })
 
  const data = await response.json()
 
  // 提取用量
  return {
    ...data,
    usage: {
      prompt_tokens: data.usage.prompt_tokens,
      completion_tokens: data.usage.completion_tokens,
      total_tokens: data.usage.total_tokens,
    },
  }
}

3.2 流式响应的用量

流式响应需要从最后一个 chunk 提取用量：

app.post('/v1/chat/completions/stream',
  quotaMiddleware,
  async (c) => {
    const user = c.get('user')
    const body = await c.req.json()
 
    const response = await fetch('https://api.openai.com/v1/chat/completions', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        Authorization: `Bearer ${process.env.OPENAI_API_KEY}`,
      },
      body: JSON.stringify({
        ...body,
        stream: true,
      }),
    })
 
    const reader = response.body!.getReader()
    const encoder = new TextEncoder()
 
    let totalTokens = 0
 
    const stream = new ReadableStream({
      async start(controller) {
        let buffer = ''
 
        while (true) {
          const { done, value } = await reader.read()
 
          if (done) {
            // 更新用量
            await updateUsage(user.id, {
              requests: 1,
              totalTokens,
            })
 
            controller.close()
            break
          }
 
          buffer += new TextDecoder().decode(value)
 
          const lines = buffer.split('\n')
          buffer = lines.pop() || ''
 
          for (const line of lines) {
            if (line.startsWith('data: ')) {
              const data = line.slice(6)
 
              if (data === '[DONE]') {
                controller.enqueue(encoder.encode('data: [DONE]\n\n'))
                continue
              }
 
              try {
                const parsed = JSON.parse(data)
 
                // 提取用量（如果包含）
                if (parsed.usage) {
                  totalTokens = parsed.usage.total_tokens
                }
 
                controller.enqueue(encoder.encode(`data: ${data}\n\n`))
              } catch {
                // 忽略解析错误
              }
            }
          }
        }
      },
    })
 
    return new Response(stream, {
      headers: {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        Connection: 'keep-alive',
      },
    })
  }
)

4. 计费系统

4.1 预付费（余额）

用户先充值，后消费：

export const userBalance = pgTable('user_balance', {
  userId: uuid('user_id').primaryKey().references(() => users.id),
  balance: decimal('balance', { precision: 10, scale: 2 }).default('0').notNull(),
  updatedAt: timestamp('updated_at').defaultNow().notNull(),
})
 
// 充值
app.post('/billing/topup', async (c) => {
  const user = c.get('user')
  const { amount } = await c.req.json()
 
  // 调用支付接口
  await processPayment(user.id, amount)
 
  // 更新余额
  await db
    .update(userBalance)
    .set({
      balance: sql`${userBalance.balance} + ${amount}`,
    })
    .where(eq(userBalance.userId, user.id))
 
  return c.json({ balance: await getBalance(user.id) })
})
 
// 扣费
async function deductBalance(userId: string, amount: number) {
  const balance = await getBalance(userId)
 
  if (balance < amount) {
    throw new Error('Insufficient balance')
  }
 
  await db
    .update(userBalance)
    .set({
      balance: sql`${userBalance.balance} - ${amount}`,
    })
    .where(eq(userBalance.userId, userId))
}

4.2 订阅（计划）

用户订阅计划，按月/年付费：

export const subscriptions = pgTable('subscriptions', {
  id: uuid('id').primaryKey().defaultRandom(),
  userId: uuid('user_id').notNull(),
  plan: text('plan').notNull(),  // free, pro, enterprise
  status: text('status').notNull(),  // active, canceled, past_due
  currentPeriodStart: timestamp('current_period_start').notNull(),
  currentPeriodEnd: timestamp('current_period_end').notNull(),
  canceledAt: timestamp('canceled_at'),
})
 
// 订阅检查
app.use('/api/*', async (c, next) => {
  const user = c.get('user')
 
  const subscription = await db.query.subscriptions.findFirst({
    where: eq(subscriptions.userId, user.id),
  })
 
  if (!subscription || subscription.status !== 'active') {
    // 免费计划
    c.set('plan', 'free')
  } else {
    c.set('plan', subscription.plan)
  }
 
  await next()
})

4.3 后付费（账单）

先消费，后结算：

export const usageRecords = pgTable('usage_records', {
  id: uuid('id').primaryKey().defaultRandom(),
  userId: uuid('user_id').notNull(),
  model: text('model').notNull(),
  promptTokens: integer('prompt_tokens').notNull(),
  completionTokens: integer('completion_tokens').notNull(),
  cost: decimal('cost', { precision: 10, scale: 6 }).notNull(),
  createdAt: timestamp('created_at').defaultNow().notNull(),
})
 
// 记录用量
async function recordUsage(userId: string, model: string, usage: {
  promptTokens: number
  completionTokens: number
}) {
  const pricing = await getModelPricing(model)
 
  const cost =
    (usage.promptTokens / 1000) * pricing.promptPricePer1K +
    (usage.completionTokens / 1000) * pricing.completionPricePer1K
 
  await db.insert(usageRecords).values({
    userId,
    model,
    promptTokens: usage.promptTokens,
    completionTokens: usage.completionTokens,
    cost: cost.toString(),
  })
}
 
// 生成账单
async function generateInvoice(userId: string, period: { start: Date; end: Date }) {
  const records = await db
    .select({
      totalCost: sql<string>`sum(${usageRecords.cost})`,
      totalTokens: sql<number>`sum(${usageRecords.promptTokens} + ${usageRecords.completionTokens})`,
    })
    .from(usageRecords)
    .where(and(
      eq(usageRecords.userId, userId),
      gte(usageRecords.createdAt, period.start),
      lte(usageRecords.createdAt, period.end)
    ))
 
  return {
    amount: records[0].totalCost,
    tokens: records[0].totalTokens,
    period,
  }
}

5. 错误处理

5.1 额度耗尽

// 返回 402 Payment Required
return c.json({
  error: {
    message: 'Monthly quota exceeded',
    type: 'quota_exceeded',
    code: 'quota_exceeded',
    upgradeUrl: 'https://app.example.com/billing/upgrade',
    currentUsage: {
      requests: 1000,
      maxRequests: 1000,
    },
  },
}, 402)

5.2 模型不可用

// 返回 403 Forbidden
return c.json({
  error: {
    message: `Model ${model} is not available for your plan`,
    type: 'model_not_available',
    code: 'model_not_available',
    availableModels: ['gpt-3.5-turbo', 'claude-3-haiku'],
    upgradeUrl: 'https://app.example.com/billing/upgrade',
  },
}, 403)

5.3 余额不足

// 返回 402 Payment Required
return c.json({
  error: {
    message: 'Insufficient balance',
    type: 'insufficient_balance',
    code: 'insufficient_balance',
    currentBalance: 0.50,
    estimatedCost: 1.00,
    topupUrl: 'https://app.example.com/billing/topup',
  },
}, 402)

6. 监控与告警

6.1 用量监控

// 监控异常用量
async function monitorUsage() {
  const users = await db.query.users.findMany()
 
  for (const user of users) {
    const usage = await getCurrentUsage(user.id)
    const quota = await getQuota(user.id)
 
    // 超过 80% 告警
    if (usage.usedTokens > quota.maxTokens * 0.8) {
      await sendAlert({
        type: 'quota_warning',
        userId: user.id,
        usage: usage.usedTokens,
        quota: quota.maxTokens,
      })
    }
 
    // 超过 100% 通知
    if (usage.usedTokens > quota.maxTokens) {
      await sendNotification({
        type: 'quota_exceeded',
        userId: user.id,
      })
    }
  }
}
 
// 定时任务
cron.schedule('0 * * * *', monitorUsage)  // 每小时检查

6.2 用量仪表板

app.get('/api/usage', async (c) => {
  const user = c.get('user')
 
  const usage = await getCurrentUsage(user.id)
  const quota = await getQuota(user.id)
 
  return c.json({
    current: {
      requests: usage.usedRequests,
      tokens: usage.usedTokens,
    },
    quota: {
      maxRequests: quota.maxRequests,
      maxTokens: quota.maxTokens,
    },
    percentage: {
      requests: (usage.usedRequests / quota.maxRequests) * 100,
      tokens: (usage.usedTokens / quota.maxTokens) * 100,
    },
    resetAt: quota.resetAt,
  })
})

总结

AI 项目的额度控制涉及请求次数、Token 用量、模型权限、并发数。计费系统需要支持预付费、订阅、后付费。

这一节涉及到的几个实践：

额度类型：请求次数、Token 用量、模型权限、并发限制
额度检查：中间件统一检查，请求前验证
Token 提取：从 AI API 响应中提取，流式响应特殊处理
计费系统：预付费、订阅、后付费
错误处理：402 额度耗尽、403 模型不可用
监控告警：用量监控、异常告警、用量仪表板

额度控制是 AI SaaS 的核心。检查要快、计费要准、错误要清晰。用户应该随时能看到自己的用量和剩余额度。

下一篇看认证安全最佳实践——密码策略、Token 安全、攻击防护。