1. 从一次多模型调用统计需求说起mongoose aggregate 分组计数排序到底解决什么问题如果你正在用 Node.js 写后端手里有一张记录「每次模型调用」的集合某天产品经理跑过来问每个模型分别被调用了多少次按次数从高到低排个序给我。你第一反应可能是find出来在 JS 里 reduce数据量小还行几十万条就开始卡。这时候mongoose aggregate的$group$sort就是正解。aggregate是 MongoDB 的聚合管道你可以把它理解成一条流水线数据从集合里流出来经过$match过滤、$group分组、$sort排序最后吐出一个结果数组。$group负责「按某个字段把文档归类」$sum: 1负责「每遇到一条就加一」$sort负责「按计数字段升序或降序排列」。这三步组合起来就是「分组统计数量并排序」的标准链路。它适合谁适合所有用 Mongoose 操作 MongoDB、需要做统计报表、调用次数排行、用户行为聚合的开发者。尤其是现在很多团队用统一 Key 走多模型调用日志表里混着不同模型的记录按模型分组计数几乎是刚需。这篇就围绕这个场景把管道配置、字段选择、排序方向、结果校验一步步拆开讲代码可以直接复制到你的项目里跑。我试过在几十万条调用记录上直接find().lean()再手动统计内存直接飙到几百兆换成aggregate之后数据库端算完只返回几十条聚合结果响应从秒级降到毫秒级。这个差距就是聚合管道存在的意义。2. TaoToken 统一 Key 前置准备多模型调用日志从哪来要让统计有意义得先有数据。这里说的数据是每次调用大模型时落库的一条记录。很多团队的做法是用 TaoToken 的统一 Key 和 API 通道把不同模型的调用都收敛到同一个入口这样日志表结构统一统计起来也方便。TaoToken 的定位是统一的大模型 API 接入层官网在 https://taotoken.net API 入口是 https://taotoken.net/api 。你拿到一个 Key 之后可以在模型对话页面先验证通道是否通再在控制台里管理 Key 和额度。对于要做 coding 或者 Agent 长期任务的场景Coding Plan 会更合适因为它按订阅方式提供稳定的调用额度不用每次担心余额。具体操作上我一般这样落地写一个薄薄的封装函数每次调用模型前先记录一条日志字段包括model模型 ID、unit_name业务单元或团队名、created_at时间、status成功/失败。调用完成后再更新这条记录的状态和耗时。这样日志表里天然就有可分组的字段。为什么强调统一 Key因为如果你每个模型用不同的 Key、不同的 SDK日志字段很容易写乱有的记model_name有的记modelId聚合时_id就对不齐。统一走 TaoToken 之后模型 ID 的命名规范由你自己控制落库时保持一致后面$group的_id直接引用同一个字段就行。这里给一个日志 Schema 的参考字段名和后面聚合管道严格对应// models/callLog.js const mongoose require(mongoose); const callLogSchema new mongoose.Schema({ model: { type: String, required: true }, // 模型 ID如 gpt-4o、claude-3-5-sonnet unit_name: { type: String, required: true }, // 业务单元/团队标识 status: { type: String, default: success }, // success / failed latency_ms: { type: Number, default: 0 }, // 调用耗时 createdAt: { type: Date, default: Date.now } }); module.exports mongoose.model(CallLog, callLogSchema);拿到 Key 的入口在 API Keys 页面接入细节可以对照接入文档模型能力验证走模型对话。这几个入口在后面的 CTA 里会再给一次方便你直接跳。数据准备好了接下来就是核心把aggregate管道写对。3. 可复制的 aggregate 管道配置$group 分组计数 $sort 排序完整写法这一节是全文重点直接给可运行的管道配置。假设我们要统计每个unit_name下的调用次数并按次数降序排列。核心代码来自实际项目字段名和上一节的 Schema 一致。// services/statsService.js const CallLog require(../models/callLog); async function countByUnit() { const result await CallLog.aggregate([ // 第一步过滤只统计成功的调用减少参与分组的数据量 { $match: { status: success } }, // 第二步分组按 unit_name 归类每遇到一条就 count 1 { $group: { _id: { unit_name: $unit_name }, // {} 内是分组条件引用文档字段要加 $ count: { $sum: 1 } // 类似 count但这是管道函数必须配 $sum } }, // 第三步排序count 为 -1 表示降序1 表示升序 { $sort: { count: -1 } } ]); return result; } module.exports { countByUnit };逐段解释。$match放在管道最前面是性能关键它先把不需要的文档挡掉后面的$group处理的数据量就小了。$group的_id是分组键这里用了一个对象{ unit_name: $unit_name }好处是结果里_id会保留字段名可读性好如果你只写_id: $unit_name结果里_id就是一个裸字符串字段语义丢了。$sum: 1是计数累加的标准写法每进一条文档加一。$sort的count: -1是降序排行榜场景基本都是降序。如果你要按多个字段分组比如「按模型 按业务单元」统计_id里加字段即可{ $group: { _id: { model: $model, unit_name: $unit_name }, count: { $sum: 1 }, avgLatency: { $avg: $latency_ms } // 顺便算个平均耗时 } }排序也可以多级比如先按 count 降序count 相同再按_id.unit_name升序{ $sort: { count: -1, _id.unit_name: 1 } }这里有个容易踩的坑$sort的字段必须是管道里已经存在的字段。你在$group之后排序只能用_id和聚合出来的count、avgLatency这些想按原始文档的createdAt排序得在$group之前用$sort或者用$first/$max把时间带进分组结果。再给一个带时间范围过滤的版本实际项目里几乎都会加async function countByUnitInRange(start, end) { return CallLog.aggregate([ { $match: { status: success, createdAt: { $gte: start, $lte: end } } }, { $group: { _id: { unit_name: $unit_name }, count: { $sum: 1 } } }, { $sort: { count: -1 } }, { $limit: 20 // 只要前 20 名排行榜常用 } ]); }$limit放在$sort之后能进一步减少返回数据量。整个管道是顺序执行的顺序写错结果就不对比如$limit写在$sort前面就变成「先取 20 条再排序」完全不是你要的。4. 本地验证请求与成功结果跑通管道并校验数据配置写完得验证。我一般分三步先造几条测试数据再跑聚合最后人工核对结果。第一步插入测试数据。写个脚本模拟不同 unit_name 的调用记录// scripts/seed.js const mongoose require(mongoose); const CallLog require(../models/callLog); async function seed() { await mongoose.connect(mongodb://127.0.0.1:27017/taotoken_demo); await CallLog.deleteMany({}); await CallLog.insertMany([ { model: gpt-4o, unit_name: team-a, status: success }, { model: gpt-4o, unit_name: team-a, status: success }, { model: gpt-4o, unit_name: team-a, status: failed }, { model: claude-3-5-sonnet, unit_name: team-b, status: success }, { model: claude-3-5-sonnet, unit_name: team-b, status: success }, { model: gpt-4o-mini, unit_name: team-c, status: success } ]); console.log(seed done); await mongoose.disconnect(); } seed();跑node scripts/seed.js控制台输出seed done就说明数据进去了。第二步调用聚合函数并打印// scripts/verify.js const mongoose require(mongoose); const { countByUnit } require(../services/statsService); async function verify() { await mongoose.connect(mongodb://127.0.0.1:27017/taotoken_demo); const result await countByUnit(); console.log(JSON.stringify(result, null, 2)); await mongoose.disconnect(); } verify();预期输出注意 team-a 有一条 failed 被$match过滤掉了所以是 2 不是 3[ { _id: { unit_name: team-a }, count: 2 }, { _id: { unit_name: team-b }, count: 2 }, { _id: { unit_name: team-c }, count: 1 } ]第三步校验。校验的核心是「聚合结果能不能和原始数据对上」。你可以用countDocuments单独查一遍const total await CallLog.countDocuments({ status: success }); const sum result.reduce((acc, cur) acc cur.count, 0); console.log(原始成功总数:, total, 聚合计数总和:, sum); // 两者必须相等否则管道逻辑有问题如果total和sum不相等八成是$match条件和countDocuments的条件不一致或者$group的_id里有字段为 null 导致部分文档被归到同一个 null 组。这个校验动作我强烈建议每次都做尤其是管道改过之后。另外如果你在真实项目里用 TaoToken 统一 Key 落库日志量可能很大验证时记得加时间范围别把全表都聚合一遍。用$match卡住最近一天的数据既快又安全。5. 本篇常见报错排查401、local proxy failed、reading choices 与 OAuth 对照统计管道本身报错不多但结合多模型调用场景问题往往出在「数据来源」和「调用链路」上。下面按真实报错逐个对照。报错一401 Unauthorized。这个通常不是 aggregate 的问题而是你落库的日志里混入了调用失败的记录或者你在验证阶段想顺便调一次模型结果 Key 不对。检查两处一是 TaoToken 的 Key 是否复制完整有没有多余空格二是请求头里Authorization: Bearer key格式对不对。如果 Key 没问题还是 401去控制台确认这个 Key 是否被禁用或额度耗尽。报错二local proxy failed。这个报错一般出现在你本地起了代理工具或者环境变量里配了HTTP_PROXY/HTTPS_PROXY导致请求发不出去。排查方法先echo $HTTP_PROXY看有没有值有就临时unset掉再试。Node 项目里如果用了 axios它默认会读环境变量里的代理配置可以在请求配置里显式关掉。注意这里说的是排查本地环境变量不是让你去配什么网络工具把干扰项清掉就行。报错三Cannot read properties of undefined (reading choices)。这个报错说明你拿到的响应体里没有choices字段通常是接口返回了错误对象但你的代码直接去读response.choices[0]。正确做法是先判断const data response.data; if (!data || !data.choices) { console.error(响应异常:, JSON.stringify(data)); throw new Error(模型返回结构不符合预期); } const content data.choices[0].message.content;这样出错时你能看到真实的错误信息而不是一个模糊的 undefined 报错。报错四OAuth 相关报错。如果你用的是 Claude Code 这类工具它可能走 OAuth 授权流程。报错通常表现为 token 过期或授权失效。处理方式是重新走一遍授权或者在配置文件里更新 token。如果你是通过 TaoToken 的 Coding Plan 接入按文档里的配置方式填 Base URL、Key、Model ID 三件套即可不要混用旧的 OAuth 凭证。这里把三件套写全方便对照{ baseUrl: https://taotoken.net/api, apiKey: 你的 TaoToken Key, model: claude-3-5-sonnet }如果你用 CC Switch 或 Cline 的 MCP 配置也是同样的三件套逻辑Base URL 指向https://taotoken.net/apiKey 填控制台生成的Model ID 填你要用的模型。Codex 的auth.json里同样对应这三个字段别只填 Key 漏了 Base URL否则请求会打到默认地址上。排查顺序建议先确认 Key 和 Base URL再确认模型 ID最后看网络环境变量。三步走完大部分报错都能定位。6. 把统计接进你的工作流从聚合结果到调用优化管道跑通只是第一步真正有价值的是把统计结果用起来。我通常会把countByUnit的结果接到两个地方一个是管理后台的排行榜一个是每日定时任务把统计结果写进一张汇总表避免每次都实时聚合。定时任务可以这样写// jobs/dailyStats.js const cron require(node-cron); const CallLog require(../models/callLog); const DailyStat require(../models/dailyStat); cron.schedule(0 2 * * *, async () { const yesterday new Date(Date.now() - 24 * 60 * 60 * 1000); const result await CallLog.aggregate([ { $match: { status: success, createdAt: { $gte: yesterday } } }, { $group: { _id: { unit_name: $unit_name }, count: { $sum: 1 } } }, { $sort: { count: -1 } } ]); await DailyStat.deleteMany({ date: yesterday.toDateString() }); await DailyStat.insertMany( result.map(r ({ date: yesterday.toDateString(), unit_name: r._id.unit_name, count: r.count })) ); console.log(daily stats saved:, result.length); });这样后台查报表直接读DailyStat不用每次扫全量日志。聚合管道的性能优势在于数据库端计算但再快也架不住高频调用预计算是常规优化手段。另一个用法是异常监控。如果某个unit_name的失败率突然升高可以加一个按status分组的管道对比成功和失败的数量{ $group: { _id: { unit_name: $unit_name, status: $status }, count: { $sum: 1 } } }结果里每个 unit 会有 success 和 failed 两条一眼就能看出谁在报错。配合 TaoToken 控制台的调用记录能快速定位是 Key 问题还是模型问题。最后提醒一个索引问题。$match和$sort用到的字段建议加索引。比如status、createdAt、unit_name这三个字段建一个复合索引callLogSchema.index({ status: 1, createdAt: -1, unit_name: 1 });索引能让$match阶段少扫很多文档聚合速度提升明显。但索引不是越多越好写日志频繁的表索引太多会拖慢写入按实际查询模式取舍。到这里从数据落库、管道配置、本地验证到报错排查整条链路就闭环了。你可以先把countByUnit复制到项目里跑一遍确认结果和原始数据对得上再考虑接定时任务和索引优化。统计这件事先跑通再优化比一上来就设计复杂管道靠谱得多。