账单高了 30%，却说不清贵在哪
「回声」上线 AI 摘要功能的第二个月，财务发现模型账单比预估高了 30%。团队打开自建看板，只看到一个大数：本月总成本。再往下问“哪个租户、哪个模型、哪一天”，看板答不上来。
排查持续了三天。第一层原因是网关重试：同一次业务请求在超时后重发，产生了两个不同的 requestId，计量系统当成两次调用。第二层原因是失败请求没有记账：模型已经处理完输入、在返回途中超时，用量真实发生，但代码在异常分支直接返回。第三层原因是日切：账单按 UTC 零点切，内部报表按北京时间切，跨日的那几个小时被两边各算了一次。
这三个问题都不是“算错了”，而是“记的方式撑不住对账”。本节从用量事件出发，把计量、幂等、日切和对账串成一条可测试的链路。
文中出现的金额、比例与耗时，除非标注为本机测试输出，都是案例假设值。
计量事件的四个要素
一条用量事件要能独立回答四个问题：谁用的、什么时候用的、用了多少、该付多少钱。
“谁”包含租户与模型两个维度。只记租户，无法和提供方按模型的账单对齐；只记模型，无法做租户配额与内部结算。
“什么时候”不只是时间戳，还包括账期归属。时间戳是事实，账期是策略，两者必须分开存。
“用了多少”至少要记输入 token 与输出 token。很多提供方对两者定价不同，合并成一个总数会让对账失去意义。
“多少钱”用整数微元记录。浮点数在累加几千条事件后会出现可见误差，而账单差异往往正是几厘钱量级。整数微元把舍入问题收敛到唯一一次向上取整。
事件还必须有 eventId 与 requestId。前者标识一次投递，后者标识一次业务调用。两者缺一，幂等就无从谈起。
两层幂等键
网关到计量系统之间通常是至少一次投递：消息队列会重发，消费者会重启，上游会重试。计量系统必须自己保证“同一次调用只入账一次”。
第一层是传输级幂等，键是 eventId。同一个事件重复到达，直接返回 duplicate。
第二层是业务级幂等，键是 requestId + modelId。上游可能生成了新的 eventId，但业务上仍是同一次调用。这一层能挡住“换了消息 id 的重放”。
两层都会遇到“键相同但内容不同”的情况。这时不能静默覆盖，也不能取最新值，而应返回 conflict 并保留第一条。内容不同说明上游状态错乱，覆盖只会把问题藏起来；保留第一条至少让对账时能发现差异。
冲突不是异常，它是需要人看的信号。计量系统可以把冲突计数暴露成指标，但不要自动修复。
日切是业务策略
“这一天”到底是哪一天，取决于合同，而不是服务器时区。
提供方账单通常按自己的账期切，内部报表按业务所在时区切，两边可能相差几个小时。实现上不要把 new Date().getDate() 直接当账期，而要显式传入 UTC 偏移，把日切规则写成一个参数。
本文的实现默认按 UTC，需要按北京时间切时传 480。测试里专门构造了一个 UTC 前一天、东八区当天的时间点，断言两种偏移得到不同的日期键。这个测试看起来琐碎，却能防止“上线后才发现在月底少算一天”。
对账：先求和，再逐行比较
对账不是比较两个总数。总数相等可能只是两个方向的错误互相抵消。
正确顺序是：先按“租户 + 模型”分组求和，再逐行比较。提供方账单可能把同一租户同一模型拆成多行，内部事件则是一条条明细，所以两边都要先聚合到同一粒度。
比较时先看 token，再看金额。token 不一致说明计量口径有问题，例如分词器版本不同、缓存命中没有扣除、流式响应的输出被重复计数。金额不一致但 token 一致，才更像价格表或舍入问题。
金额可以设一个极小的容差，用来吸收计价舍入；token 不设容差。把 token 差异也放进容差，等于允许“少记了一些调用”被当成正常。
对账结果分五种状态：完全一致、只有内部账、只有提供方账、token 不一致、金额不一致。每种状态对应不同的处理人，不能合并成一个“对账失败”。
差异不是噪声
只有内部账，常见于提供方账单延迟或免费额度；只有提供方账，常见于网关在记账前崩溃，或者事件被消费者丢弃。前者等账单补齐，后者要查投递链路。
token 不一致，先查分词器与缓存。很多提供方对缓存命中的输入按折扣价计费，但 token 数仍按全量展示，内部如果直接乘单价就会多算。
金额不一致但 token 一致，先查价格表版本。模型降价、阶梯价、汇率和税都可能造成差异，逐项排除比“重新算一遍”有效。
实现
下面的实现包含事件校验、两层幂等、按日聚合、配额窗口与日结对账。它不依赖数据库，但保留了真实系统必须有的约束。
// file: src/observe/metering.ts
// tier: L1
// verify: node --test test/observe/metering.test.ts
import { AiError } from "../core/errors.ts";
import { systemClock } from "../core/types.ts";
import type { Clock, TenantId } from "../core/types.ts";
/** 一条用量事件是一次模型调用的结算凭据，必须可重放、可去重。 */
export interface UsageEvent {
eventId: string;
requestId: string;
tenantId: TenantId;
modelId: string;
/** 事件发生时间；账期归属由日切偏移决定，不由时区常识决定。 */
occurredAt: number;
inputTokens: number;
outputTokens: number;
costMicros: number;
}
export interface UsageAggregate {
tenantId: string;
modelId: string;
requests: number;
inputTokens: number;
outputTokens: number;
costMicros: number;
}
function fail(message: string): never {
throw new AiError("invalid-request", `metering: ${message}`);
}
function requireText(value: string, field: string): void {
if (typeof value !== "string" || value.trim() === "") fail(`${field} is required`);
}
function requireCount(value: number, field: string): void {
if (!Number.isInteger(value) || value < 0) fail(`${field} must be a non-negative integer`);
}
function validateEvent(event: UsageEvent): void {
for (const field of ["eventId", "requestId", "tenantId", "modelId"] as const) requireText(event[field], field);
for (const field of ["occurredAt", "inputTokens", "outputTokens", "costMicros"] as const) requireCount(event[field], field);
}
const PAYLOAD_FIELDS: Array<keyof UsageEvent> = ["requestId", "tenantId", "modelId", "occurredAt", "inputTokens", "outputTokens", "costMicros"];
function samePayload(left: UsageEvent, right: UsageEvent): boolean {
return PAYLOAD_FIELDS.every((field) => left[field] === right[field]);
}
/** 默认按 UTC 计算；需要按 +08:00 日切时显式传入 480。 */
export function dayKeyOf(timestampMs: number, utcOffsetMinutes = 0): string {
return new Date(timestampMs + utcOffsetMinutes * 60_000).toISOString().slice(0, 10);
}
export type RecordStatus = "accepted" | "duplicate" | "conflict";
/**
* 内存账本。eventId 是传输级幂等键，requestId + modelId 是业务级幂等键；
* 两层都命中才算重复，任一键携带不同数字都算冲突。
*/
export class UsageLedger {
private readonly events: Map<string, UsageEvent>;
private readonly byRequest: Map<string, string>;
private readonly clock: Clock;
constructor(clock: Clock = systemClock) {
this.events = new Map();
this.byRequest = new Map();
this.clock = clock;
}
record(event: UsageEvent): { status: RecordStatus; reason?: "event-id" | "request-key"; duplicateOf?: string } {
validateEvent(event);
const seenByEvent = this.events.get(event.eventId);
if (seenByEvent !== undefined) {
return samePayload(seenByEvent, event)
? { status: "duplicate", reason: "event-id", duplicateOf: seenByEvent.eventId }
: { status: "conflict", reason: "event-id", duplicateOf: seenByEvent.eventId };
}
const requestKey = `${event.requestId}\u0000${event.modelId}`;
const seenEventId = this.byRequest.get(requestKey);
if (seenEventId !== undefined) {
const seen = this.events.get(seenEventId);
if (seen !== undefined) {
return samePayload(seen, event)
? { status: "duplicate", reason: "request-key", duplicateOf: seen.eventId }
: { status: "conflict", reason: "request-key", duplicateOf: seen.eventId };
}
}
this.events.set(event.eventId, event);
this.byRequest.set(requestKey, event.eventId);
return { status: "accepted" };
}
get size(): number {
return this.events.size;
}
/** 按租户 + 模型聚合并稳定排序，便于逐字节比对。 */
aggregate(options: { day?: string; utcOffsetMinutes?: number } = {}): UsageAggregate[] {
const offset = options.utcOffsetMinutes ?? 0;
const buckets = new Map<string, UsageAggregate>();
for (const event of this.events.values()) {
if (options.day !== undefined && dayKeyOf(event.occurredAt, offset) !== options.day) continue;
const key = `${event.tenantId}\u0000${event.modelId}`;
const bucket = buckets.get(key);
if (bucket === undefined) {
buckets.set(key, {
tenantId: event.tenantId, modelId: event.modelId, requests: 1,
inputTokens: event.inputTokens, outputTokens: event.outputTokens, costMicros: event.costMicros,
});
} else {
bucket.requests += 1;
bucket.inputTokens += event.inputTokens;
bucket.outputTokens += event.outputTokens;
bucket.costMicros += event.costMicros;
}
}
return [...buckets.values()].sort((left, right) => {
if (left.tenantId !== right.tenantId) return left.tenantId < right.tenantId ? -1 : 1;
return left.modelId < right.modelId ? -1 : left.modelId > right.modelId ? 1 : 0;
});
}
currentDay(utcOffsetMinutes = 0): string {
return dayKeyOf(this.clock.now(), utcOffsetMinutes);
}
quota(tenantId: TenantId, limitMicros: number, options: { day?: string; utcOffsetMinutes?: number } = {}) {
requireCount(limitMicros, "limitMicros");
const usedMicros = this.aggregate(options).filter((row) => row.tenantId === tenantId)
.reduce((total, row) => total + row.costMicros, 0);
return {
tenantId,
limitMicros,
usedMicros,
remainingMicros: Math.max(0, limitMicros - usedMicros),
exceeded: usedMicros > limitMicros,
utilization: limitMicros === 0 ? (usedMicros === 0 ? 0 : 1) : usedMicros / limitMicros,
};
}
}
/** 提供方账单行；同一账期内同租户 + 模型可能拆成多行，对账时先求和。 */
export interface InvoiceLine {
tenantId: string;
modelId: string;
inputTokens: number;
outputTokens: number;
costMicros: number;
}
export interface ReconcileRow {
tenantId: string;
modelId: string;
status: "matched" | "internal-only" | "invoice-only" | "token-mismatch" | "cost-mismatch";
internal: UsageAggregate | null;
invoice: InvoiceLine | null;
/** 内部账减提供方账；正数表示我们记多了，负数表示少记了。 */
delta: { inputTokens: number; outputTokens: number; costMicros: number };
}
export interface ReconciliationTotals {
internalCostMicros: number;
invoiceCostMicros: number;
deltaCostMicros: number;
mismatchedRows: number;
internalOnlyRows: number;
invoiceOnlyRows: number;
}
export interface Reconciliation {
day: string;
rows: ReconcileRow[];
totals: ReconciliationTotals;
balanced: boolean;
}
function deltaOf(internal: UsageAggregate | null, invoice: InvoiceLine | null) {
const pick = (field: "inputTokens" | "outputTokens" | "costMicros") => (internal?.[field] ?? 0) - (invoice?.[field] ?? 0);
return { inputTokens: pick("inputTokens"), outputTokens: pick("outputTokens"), costMicros: pick("costMicros") };
}
function statusOf(
internal: UsageAggregate | null,
invoice: InvoiceLine | null,
delta: { inputTokens: number; outputTokens: number; costMicros: number },
tolerance: number,
): ReconcileRow["status"] {
if (internal !== null && invoice === null) return "internal-only";
if (internal === null && invoice !== null) return "invoice-only";
if (delta.inputTokens !== 0 || delta.outputTokens !== 0) return "token-mismatch";
return Math.abs(delta.costMicros) > tolerance ? "cost-mismatch" : "matched";
}
/**
* 日结对账：先按租户 + 模型求和，再逐行比较 token 与金额。
* token 不一致永远单独标记，金额容差只用于吸收计价舍入。
*/
export function reconcile(
ledger: UsageLedger,
day: string,
invoiceLines: InvoiceLine[],
options: { utcOffsetMinutes?: number; toleranceMicros?: number } = {},
): Reconciliation {
const tolerance = options.toleranceMicros ?? 0;
requireCount(tolerance, "toleranceMicros");
const invoices = new Map<string, InvoiceLine>();
for (const line of invoiceLines) {
requireText(line.tenantId, "invoice.tenantId");
requireText(line.modelId, "invoice.modelId");
for (const field of ["inputTokens", "outputTokens", "costMicros"] as const) {
requireCount(line[field], `invoice.${field}`);
}
const key = `${line.tenantId}\u0000${line.modelId}`;
const existing = invoices.get(key);
if (existing === undefined) {
invoices.set(key, { ...line });
} else {
existing.inputTokens += line.inputTokens;
existing.outputTokens += line.outputTokens;
existing.costMicros += line.costMicros;
}
}
const internal = new Map<string, UsageAggregate>();
for (const row of ledger.aggregate({ day, utcOffsetMinutes: options.utcOffsetMinutes })) {
internal.set(`${row.tenantId}\u0000${row.modelId}`, row);
}
const rows: ReconcileRow[] = [];
for (const key of [...new Set([...internal.keys(), ...invoices.keys()])].sort()) {
const internalRow = internal.get(key) ?? null;
const invoiceRow = invoices.get(key) ?? null;
const delta = deltaOf(internalRow, invoiceRow);
rows.push({
tenantId: internalRow?.tenantId ?? invoiceRow?.tenantId ?? "",
modelId: internalRow?.modelId ?? invoiceRow?.modelId ?? "",
status: statusOf(internalRow, invoiceRow, delta, tolerance),
internal: internalRow,
invoice: invoiceRow,
delta,
});
}
const totals = {
internalCostMicros: rows.reduce((sum, row) => sum + (row.internal?.costMicros ?? 0), 0),
invoiceCostMicros: rows.reduce((sum, row) => sum + (row.invoice?.costMicros ?? 0), 0),
deltaCostMicros: rows.reduce((sum, row) => sum + row.delta.costMicros, 0),
mismatchedRows: rows.filter((row) => row.status === "token-mismatch" || row.status === "cost-mismatch").length,
internalOnlyRows: rows.filter((row) => row.status === "internal-only").length,
invoiceOnlyRows: rows.filter((row) => row.status === "invoice-only").length,
};
const balanced = totals.mismatchedRows === 0 && totals.internalOnlyRows === 0 && totals.invoiceOnlyRows === 0;
return { day, rows, totals, balanced };
}
record 返回三种状态而不是布尔值，是为了让调用方能区分“正常去重”和“需要人工看的冲突”。如果只返回成功或失败，冲突会被当成成功吞掉。
测试与真实输出
测试覆盖重复投递、同键异值冲突、业务键去重、非法用量、分组聚合、日切偏移、配额超限、逐行对账、单边缺失、token 优先、金额容差与多行账单求和。执行：
node --test test/observe/metering.test.ts
本机一次运行的真实输出摘要如下，耗时随机器变化：
ℹ tests 12
ℹ pass 12
ℹ fail 0
ℹ duration_ms 149.511542
日切与容差的测试都用了精确断言，而不是“差不多”。计量逻辑一旦用“差不多”测试，线上就会用“差很多”回报。
配额不是账单
配额和账单共用一本账，但目的不同。账单回答“该付多少”，配额回答“现在还能不能用”。
配额必须实时，账单可以延迟。用提供方账单驱动配额，会让用户在账单出来之前一直可用；用内部事件驱动账单，又可能因为事件丢失而少收。两者应该同源但不同步：配额读实时聚合，账单读经过对账的日结结果。
小结
计量系统的核心不是“把数字算出来”，而是让每个数字都能被追溯到一条事件、一个租户、一个模型和一个账期。事件要有两层幂等键，日切要显式配置，对账要先聚合再逐行比较，token 差异永远单独标记。
下一节离开“花了多少”，转向“效果如何”：采纳率、重试率、转人工率与投诉如何从线上采集，又怎样和离线评测对应。
阅读导航：上一节：12.1 一次 AI 请求的最小观测字段 · 下一节：12.3 质量信号的线上采集 。
常见错误与边界
- 只按
eventId去重。上游换了消息 id 的重放会重复入账。 - 同键异值时覆盖旧记录。冲突被藏起来，月底对账才发现金额对不上。
- 用服务器本地时区做日切。服务器迁移或容器时区变化会直接改变账期。
- 失败请求不记用量。已经消耗的 token 不会因为异常而退回。
- 对账只比总数。两个方向的错误可能刚好抵消。
- 给 token 也设容差。少记调用会被当成正常舍入。
- 用提供方账单驱动实时配额。账单延迟会让配额长期失效。
- 差异没有处理记录。同一个差异会每个月重复出现。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。