Convex Agents Usage Tracking

用途

用量跟踪记录每个 Agent 使用了多少 token，从而实现准确计费、成本监控和性能优化。这对于理解大模型成本和用户影响至关重要。

何时使用本技能

• 根据 token 消耗量向用户计费
• 监控 API 成本
• 优化 Agent 效率
• 按用户、Agent 或团队跟踪用量
• 生成发票和成本报告
• 对高用量进行告警
• 分析成本趋势

如何使用

1. 配置用量处理器

创建一个处理器来记录用量：

// convex/agents/myAgent.ts
import { Agent } from "@convex-dev/agent";
import { components } from "../_generated/api";
import { openai } from "@ai-sdk/openai";
import { internal } from "../_generated/api";

const myAgent = new Agent(components.agent, {
  name: "My Agent",
  languageModel: openai.chat("gpt-4o-mini"),
  usageHandler: async (ctx, args) => {
    const {
      userId,
      threadId,
      agentName,
      model,
      provider,
      usage, // { inputTokens, outputTokens, totalTokens }
      providerMetadata,
    } = args;

    // Save usage to database
    await ctx.runMutation(internal.usage.recordUsage, {
      userId,
      threadId,
      agentName,
      model,
      provider,
      inputTokens: usage.inputTokens,
      outputTokens: usage.outputTokens,
      totalTokens: usage.totalTokens,
      timestamp: Date.now(),
    });
  },
});

2. 存储用量记录

保存用量数据以供后续分析：

// convex/usage.ts
import { internalMutation } from "./_generated/server";
import { v } from "convex/values";
import { defineTable, defineSchema } from "convex/server";

export const schema = defineSchema({
  usage: defineTable({
    userId: v.string(),
    threadId: v.optional(v.string()),
    agentName: v.optional(v.string()),
    model: v.string(),
    provider: v.string(),
    inputTokens: v.number(),
    outputTokens: v.number(),
    totalTokens: v.number(),
    cost: v.number(), // Cost in dollars
    date: v.string(), // ISO date for daily rollups
    billingPeriod: v.string(), // YYYY-MM for monthly
  })
    .index("billingPeriod_userId", ["billingPeriod", "userId"])
    .index("date_userId", ["date", "userId"])
    .index("userId", ["userId"]),

  invoices: defineTable({
    userId: v.string(),
    billingPeriod: v.string(),
    amount: v.number(),
    status: v.union(
      v.literal("pending"),
      v.literal("paid"),
      v.literal("failed")
    ),
    generatedAt: v.number(),
  })
    .index("billingPeriod_userId", ["billingPeriod", "userId"])
    .index("userId", ["userId"]),
});

export const recordUsage = internalMutation({
  args: {
    userId: v.string(),
    threadId: v.optional(v.string()),
    agentName: v.optional(v.string()),
    model: v.string(),
    provider: v.string(),
    inputTokens: v.number(),
    outputTokens: v.number(),
    totalTokens: v.number(),
  },
  handler: async (
    ctx,
    {
      userId,
      threadId,
      agentName,
      model,
      provider,
      inputTokens,
      outputTokens,
      totalTokens,
    }
  ) => {
    const today = new Date().toISOString().split("T")[0];
    const billingPeriod = today.substring(0, 7); // YYYY-MM

    // Calculate cost (example: $0.015 per 1M input tokens, $0.060 per 1M output)
    const cost =
      (inputTokens / 1_000_000) * 0.015 + (outputTokens / 1_000_000) * 0.06;

    await ctx.db.insert("usage", {
      userId,
      threadId,
      agentName,
      model,
      provider,
      inputTokens,
      outputTokens,
      totalTokens,
      cost,
      date: today,
      billingPeriod,
    });
  },
});

3. 按用户查询用量

获取特定用户的总用量：

// convex/usage.ts
import { query } from "./_generated/server";
import { v } from "convex/values";

export const getUserUsage = query({
  args: { userId: v.string() },
  handler: async (ctx, { userId }) => {
    const records = await ctx.db
      .query("usage")
      .withIndex("userId", (q) => q.eq("userId", userId))
      .collect();

    const totals = records.reduce(
      (acc, record) => ({
        inputTokens: acc.inputTokens + record.inputTokens,
        outputTokens: acc.outputTokens + record.outputTokens,
        totalTokens: acc.totalTokens + record.totalTokens,
        cost: acc.cost + record.cost,
      }),
      { inputTokens: 0, outputTokens: 0, totalTokens: 0, cost: 0 }
    );

    return totals;
  },
});

export const getMonthlyUsageByUser = query({
  args: { billingPeriod: v.string() },
  handler: async (ctx, { billingPeriod }) => {
    const records = await ctx.db
      .query("usage")
      .withIndex("billingPeriod_userId", (q) =>
        q.eq("billingPeriod", billingPeriod)
      )
      .collect();

    // Group by userId
    const byUser: Record<string, any> = {};
    for (const record of records) {
      if (!byUser[record.userId]) {
        byUser[record.userId] = {
          userId: record.userId,
          totalTokens: 0,
          cost: 0,
          records: 0,
        };
      }
      byUser[record.userId].totalTokens += record.totalTokens;
      byUser[record.userId].cost += record.cost;
      byUser[record.userId].records += 1;
    }

    return Object.values(byUser);
  },
});

4. 生成月度发票

根据用量数据创建发票：

// convex/usage.ts
import { action } from "./_generated/server";
import { v } from "convex/values";

export const generateInvoices = action({
  args: { billingPeriod: v.string() },
  handler: async (ctx, { billingPeriod }) => {
    // Get all usage for the period
    const records = await ctx.db
      .query("usage")
      .withIndex("billingPeriod_userId", (q) =>
        q.eq("billingPeriod", billingPeriod)
      )
      .collect();

    // Group by user
    const byUser: Record<string, number> = {};
    for (const record of records) {
      byUser[record.userId] = (byUser[record.userId] || 0) + record.cost;
    }

    // Create invoices
    for (const [userId, amount] of Object.entries(byUser)) {
      const existingInvoice = await ctx.db
        .query("invoices")
        .filter(
          (inv) =>
            inv.billingPeriod === billingPeriod && inv.userId === userId
        )
        .first();

      if (!existingInvoice) {
        await ctx.db.insert("invoices", {
          userId,
          billingPeriod,
          amount,
          status: "pending",
          generatedAt: Date.now(),
        });
      }
    }

    return { invoicesCreated: Object.keys(byUser).length };
  },
});

5. 按 Agent 跟踪用量

比较不同 Agent 之间的效率：

// convex/usage.ts
import { query } from "./_generated/server";
import { v } from "convex/values";

export const getUsageByAgent = query({
  args: { userId: v.string(), billingPeriod: v.string() },
  handler: async (ctx, { userId, billingPeriod }) => {
    const records = await ctx.db
      .query("usage")
      .withIndex("billingPeriod_userId", (q) =>
        q.eq("billingPeriod", billingPeriod)
      )
      .filter((r) => r.userId === userId)
      .collect();

    // Group by agent
    const byAgent: Record<string, any> = {};
    for (const record of records) {
      const agent = record.agentName || "unknown";
      if (!byAgent[agent]) {
        byAgent[agent] = {
          agent,
          totalTokens: 0,
          inputTokens: 0,
          outputTokens: 0,
          cost: 0,
          calls: 0,
        };
      }
      byAgent[agent].totalTokens += record.totalTokens;
      byAgent[agent].inputTokens += record.inputTokens;
      byAgent[agent].outputTokens += record.outputTokens;
      byAgent[agent].cost += record.cost;
      byAgent[agent].calls += 1;
    }

    return Object.values(byAgent).sort((a, b) => b.cost - a.cost);
  },
});

6. 设置定时发票生成

每月自动生成发票：

// convex/crons.ts
import { cronJobs } from "convex/server";
import { internal } from "./_generated/api";

const crons = cronJobs();

// Generate invoices on the 2nd day of each month at midnight UTC
crons.monthly(
  "generateMonthlyInvoices",
  { day: 2, hourUTC: 0, minuteUTC: 0 },
  internal.usage.generateInvoices,
  { billingPeriod: calculatePreviousMonth() }
);

export default crons;

function calculatePreviousMonth(): string {
  const now = new Date();
  const month = now.getMonth() === 0 ? 11 : now.getMonth() - 1;
  const year = now.getMonth() === 0 ? now.getFullYear() - 1 : now.getFullYear();
  return `${year}-${String(month + 1).padStart(2, "0")}`;
}

7. 监控用量告警

对高用量进行告警：

// convex/usage.ts
import { action } from "./_generated/server";
import { v } from "convex/values";

export const checkUsageAlerts = action({
  args: {},
  handler: async (ctx, {}) => {
    const today = new Date().toISOString().split("T")[0];

    // Get today's usage by user
    const records = await ctx.db
      .query("usage")
      .filter((r) => r.date === today)
      .collect();

    const byUser: Record<string, number> = {};
    for (const record of records) {
      byUser[record.userId] = (byUser[record.userId] || 0) + record.cost;
    }

    // Alert on users exceeding $100/day
    const alerts = [];
    for (const [userId, cost] of Object.entries(byUser)) {
      if (cost > 100) {
        alerts.push({ userId, cost, reason: "Daily spend exceeded $100" });
      }
    }

    // Send alerts (email, Slack, etc.)
    for (const alert of alerts) {
      await ctx.runMutation(internal.notifications.sendAlert, alert);
    }

    return alerts;
  },
});

8. 用量分析仪表盘

查询用量以供展示：

// convex/usage.ts
import { query } from "./_generated/server";

export const getDashboardStats = query({
  args: { userId: v.string() },
  handler: async (ctx, { userId }) => {
    // This month's usage
    const today = new Date();
    const billingPeriod = `${today.getFullYear()}-${String(
      today.getMonth() + 1
    ).padStart(2, "0")}`;

    const monthlyRecords = await ctx.db
      .query("usage")
      .withIndex("billingPeriod_userId", (q) =>
        q.eq("billingPeriod", billingPeriod)
      )
      .filter((r) => r.userId === userId)
      .collect();

    const monthlyStats = monthlyRecords.reduce(
      (acc, r) => ({
        totalTokens: acc.totalTokens + r.totalTokens,
        cost: acc.cost + r.cost,
      }),
      { totalTokens: 0, cost: 0 }
    );

    // All time
    const allRecords = await ctx.db
      .query("usage")
      .withIndex("userId", (q) => q.eq("userId", userId))
      .collect();

    const allTimeStats = allRecords.reduce(
      (acc, r) => ({
        totalTokens: acc.totalTokens + r.totalTokens,
        cost: acc.cost + r.cost,
      }),
      { totalTokens: 0, cost: 0 }
    );

    return {
      monthly: monthlyStats,
      allTime: allTimeStats,
      averageDailyCost: monthlyStats.cost / Math.min(today.getDate(), 30),
    };
  },
});

关键原则

• 在生成时记录：在 usageHandler 中捕获用量
• 准确计算成本：使用服务商特定的定价
• 按月周期：让发票与自然月对齐
• 用户归属：始终跟踪是哪个用户产生了用量
• Agent 跟踪：了解哪些 Agent 最昂贵
• 归档：归档旧用量数据以满足合规要求

示例：完整的计费系统

// convex/billing/complete.ts
import { Agent } from "@convex-dev/agent";
import { components } from "../_generated/api";
import { openai } from "@ai-sdk/openai";
import { internal } from "../_generated/api";

// Cost per million tokens
const PRICING = {
  "gpt-4o-mini": {
    input: 0.015,
    output: 0.06,
  },
};

export const billingAgent = new Agent(components.agent, {
  name: "Billing Agent",
  languageModel: openai.chat("gpt-4o-mini"),
  usageHandler: async (ctx, { usage, userId, model }) => {
    if (!userId) return;

    const pricing = PRICING[model as keyof typeof PRICING] || {
      input: 0.001,
      output: 0.002,
    };

    const cost =
      (usage.inputTokens / 1_000_000) * pricing.input +
      (usage.outputTokens / 1_000_000) * pricing.output;

    await ctx.runMutation(internal.billing.recordUsage, {
      userId,
      model,
      inputTokens: usage.inputTokens,
      outputTokens: usage.outputTokens,
      totalTokens: usage.totalTokens,
      cost,
    });
  },
});

常见模式

• 按用户计费：每个用户收到一张发票
• 团队计费：汇总团队成员的使用量
• 按量付费：每条消息立即计费
• 月度开票：在月末收取费用
• 用量分级：为高用量用户提供折扣
• 预付额度：用户购买额度套餐

后续步骤

• 实现速率限制：参见 Convex Agents Rate Limiting 以控制成本
• 添加告警：为高用量发送通知
• 构建仪表盘：向用户展示用量分析
• 优化成本：分析哪些 Agent 最昂贵

故障排除

• 缺少用量记录：检查是否已配置 usageHandler
• 定价不匹配：验证成本计算是否与服务商一致
• 发票金额过大：检查是否存在失控的 token 生成
• 月度时间：验证计费周期计算是否与你的财年一致