Mongoose aggregate() 聚合函数实战：从 $match 到 $group 的完整配置与验证

1. 订单统计场景下 Mongoose aggregate() 到底解决什么问题如果你用 Mongoose 做过后台统计大概率遇到过这种需求前端要一个「按订单状态分组每组多少钱、多少单」的汇总而不是把几万条订单全查出来在 Node 里reduce。这时候find()就不够用了得请出Model.aggregate()。aggregate()是 Mongoose 对 MongoDB 聚合管道的封装它把一批「阶段stage」串成流水线上一个阶段的输出文档直接喂给下一个阶段。你可以把它理解成一条工厂流水线——$match是质检口只放符合条件的货进来$group是分拣台按某个字段把货归类并计数$lookup是外协车间去另一张表取关联信息$project是打包台决定最后哪些字段装箱发走。它适合谁适合已经会用find()、populate()但一遇到「分组求和、多表关联统计、去重计数」就卡住的 Node 后端同学。本文以电商订单统计为场景把$match → $group → $lookup → $project这条完整链路拆开每一段都给可复制的配置和验证动作你在本地就能跑通并核对结果。先明确一个容易踩的认知aggregate()返回的不是 Mongoose Document而是普通 JS 对象数组plain object。所以别指望在结果上调用.save()或.populate()它走的是原生驱动那套。这一点决定了后面很多写法和find()不一样。我先把本文要用的数据模型定下来后面所有片段都基于它// models/Order.js const mongoose require(mongoose); const OrderSchema new mongoose.Schema({ orderNo: { type: String, required: true, unique: true }, userId: { type: mongoose.Schema.Types.ObjectId, ref: User }, status: { type: String, enum: [pending, paid, shipped, done, refund], default: pending }, amount: { type: Number, required: true }, // 订单金额分 items: [{ sku: String, qty: Number, price: Number }], createdAt: { type: Date, default: Date.now } }); module.exports mongoose.model(Order, OrderSchema);需求是统计每个status下的订单数量和总金额并且带上该状态最近一笔订单的时间最后按总金额倒序。这个需求刚好能把四个阶段都用上。下面从环境准备开始一步步来。2. 用 TaoToken 准备可调用的模型与 Key 前置配置写聚合逻辑时我习惯让一个模型帮我 review 管道写法和报错定位尤其是$group里_id写错、$lookup的localField对不上这类问题肉眼扫很难发现。这里用 TaoToken 来提供这个能力它兼容 OpenAI 风格的接口接入成本低。TaoToken 官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 这个不加 UTM。你需要先拿到一个 API Key入口在控制台的 API Keys 页面https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。拿到 Key 之后最省事的验证方式是直接在模型对话页发一条消息确认 Key 可用https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。如果对话能正常返回说明 Key 和网络都没问题再往代码里接。如果你打算长期在编码场景里用它辅助写聚合、查报错可以看下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有 Base URL、鉴权头、模型 ID 的完整说明。这里要强调一个关键点无论你用哪种客户端接入配置里必须同时写全三件套——Base URL、API Key、Model ID。少任何一个都会报鉴权或模型不存在的错。下面给一个通用的环境变量写法Node 项目里直接读# .env TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的key TAOTOKEN_MODEL_ID你的模型ID然后在脚本里这样调用把聚合管道贴给它做检查// scripts/review-pipeline.js require(dotenv).config(); async function reviewPipeline(pipeline) { const res await fetch(${process.env.TAOTOKEN_BASE_URL}/chat/completions, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${process.env.TAOTOKEN_API_KEY} }, body: JSON.stringify({ model: process.env.TAOTOKEN_MODEL_ID, messages: [ { role: system, content: 你是 MongoDB 聚合管道专家只指出管道里的语法和字段错误。 }, { role: user, content: 检查这段聚合管道\n${JSON.stringify(pipeline, null, 2)} } ] }) }); const data await res.json(); console.log(data.choices[0].message.content); } module.exports { reviewPipeline };注意Authorization头是Bearer加 Key中间有一个空格这个空格漏了会直接 401。Base URL 结尾不要多加/v1具体以接入文档为准写错路径同样会 404 或 401。把这段跑通后面写聚合时就能随时让它帮你核对字段名。3. 可复制的聚合管道配置从 $match 到 $project 完整片段这一节是核心我把订单统计的完整管道写出来然后逐段解释。先给完整版你可以直接复制到项目里改字段名// services/orderStats.js const Order require(../models/Order); async function getOrderStats({ startDate, endDate }) { const pipeline [ // 阶段一过滤时间范围只保留已支付及之后的订单 { $match: { createdAt: { $gte: new Date(startDate), $lte: new Date(endDate) }, status: { $in: [paid, shipped, done] } } }, // 阶段二按状态分组统计单量、总金额、最近下单时间 { $group: { _id: $status, orderCount: { $sum: 1 }, totalAmount: { $sum: $amount }, lastOrderAt: { $max: $createdAt }, orderIds: { $push: $_id } } }, // 阶段三关联用户表取每个状态下的样本用户这里用 first 演示 { $lookup: { from: users, localField: orderIds, foreignField: _id, as: relatedUsers } }, // 阶段四整理输出结构去掉不需要的中间字段 { $project: { _id: 0, status: $_id, orderCount: 1, totalAmountYuan: { $divide: [$totalAmount, 100] }, lastOrderAt: 1, userSampleCount: { $size: $relatedUsers } } }, // 阶段五按总金额倒序 { $sort: { totalAmountYuan: -1 } } ]; return Order.aggregate(pipeline); } module.exports { getOrderStats };现在逐段拆。$match放在管道最前面是有讲究的它能利用索引先把数据量压下来后面的$group、$lookup才不至于处理全表。$match里用的查询语法和find()完全一致$gte、$lte、$in都能用。这里我特意把status过滤成已支付之后的三种因为待支付和退款的订单不该计入营收统计。$group是整条管道的灵魂。它的_id是「分组依据」这里写$status意思是按status字段的值分组。注意$前缀不能丢丢了就变成按字符串字面量status分组结果只会有一组。orderCount: { $sum: 1 }表示每遇到一条文档就加 1等价于 SQL 的COUNT(*)。totalAmount: { $sum: $amount }是对amount字段求和。$max: $createdAt取组内最大时间也就是最近一笔。$push: $_id把组内所有订单 ID 收集成数组为下一步$lookup做准备。这里有个高频报错点$group里除了_id其他字段必须写成累加器对象$sum、$avg、$max、$min、$push、$addToSet、$first、$last等。如果你写成status: $status就会报The field status must be an accumulator object。记住分组键只能叫_id其他都是累加器。$lookup相当于 SQL 的LEFT JOIN。from是目标集合名注意是数据库里的实际集合名通常是模型名小写复数比如User模型对应userslocalField是本管道里的字段这里是orderIds数组foreignField是目标集合的字段_idas是结果挂载的字段名。因为localField是数组MongoDB 会自动做「数组对单值」的匹配把命中的用户都塞进relatedUsers。$project用来重塑输出。_id: 0表示不输出默认的_id这里它是分组键我们改名叫status了。status: $_id把分组键重命名。totalAmountYuan: { $divide: [$totalAmount, 100] }把「分」转成「元」$divide接收一个两元素数组。userSampleCount: { $size: $relatedUsers }取关联用户数组的长度。最后$sort按金额倒序-1是降序1是升序。如果你更习惯用配置文件管理管道也可以把它抽成 JSON方便和 TaoToken 的模型对话页对照检查{ collection: orders, pipeline: [ { $match: { status: { $in: [paid, shipped, done] } } }, { $group: { _id: $status, orderCount: { $sum: 1 }, totalAmount: { $sum: $amount } } }, { $project: { _id: 0, status: $_id, orderCount: 1, totalAmount: 1 } }, { $sort: { totalAmount: -1 } } ] }把这段 JSON 贴到模型对话里让它逐阶段解释比自己啃文档快很多。4. 验证请求与成功结果本地跑通并核对数据管道写完不能直接上线得先验证。我一般分三步造数据、跑聚合、核对结果。第一步造一批可控的测试数据。写个 seed 脚本插入几条不同状态、不同金额的订单// scripts/seed.js const mongoose require(mongoose); const Order require(../models/Order); async function seed() { await mongoose.connect(mongodb://127.0.0.1:27017/shop_test); await Order.deleteMany({}); await Order.insertMany([ { orderNo: A001, status: paid, amount: 10000, createdAt: new Date(2024-06-01) }, { orderNo: A002, status: paid, amount: 20000, createdAt: new Date(2024-06-02) }, { orderNo: A003, status: shipped, amount: 30000, createdAt: new Date(2024-06-03) }, { orderNo: A004, status: done, amount: 40000, createdAt: new Date(2024-06-04) }, { orderNo: A005, status: pending, amount: 50000, createdAt: new Date(2024-06-05) } ]); console.log(seed done); await mongoose.disconnect(); } seed();第二步跑聚合并打印结果// scripts/run-stats.js const mongoose require(mongoose); const { getOrderStats } require(../services/orderStats); async function main() { await mongoose.connect(mongodb://127.0.0.1:27017/shop_test); const result await getOrderStats({ startDate: 2024-06-01, endDate: 2024-06-30 }); console.log(JSON.stringify(result, null, 2)); await mongoose.disconnect(); } main();预期输出应该是这样pending被$match过滤掉了所以只有三组[ { status: done, orderCount: 1, totalAmountYuan: 400, lastOrderAt: 2024-06-04T00:00:00.000Z, userSampleCount: 0 }, { status: shipped, orderCount: 1, totalAmountYuan: 300, lastOrderAt: 2024-06-03T00:00:00.000Z, userSampleCount: 0 }, { status: paid, orderCount: 2, totalAmountYuan: 300, lastOrderAt: 2024-06-02T00:00:00.000Z, userSampleCount: 0 } ]核对要点paid组有两条金额 100002000030000 分转元是 300对得上done组 40000 分转 400 元对得上lastOrderAt取的是组内最大时间paid组应该是 06-02对得上。userSampleCount是 0因为测试数据里userId没填$lookup没匹配到这也符合预期。第三步用explain()看执行计划确认$match走了索引const explain await Order.aggregate(pipeline).explain(executionStats); console.log(explain.stages[0].$cursor.executionStats);如果totalDocsExamined远大于nReturned说明$match没走索引需要给createdAt和status建复合索引OrderSchema.index({ status: 1, createdAt: -1 });验证这一步别偷懒。我见过太多人管道写完直接上线结果$group的_id少写个$统计出来永远只有一组排查半天。本地跑一遍结果对得上再往下走。5. 本篇常见报错排查401、local proxy failed、reading choices 等聚合本身不报错但接入和调用环节的报错很集中。我把几个高频的列出来对照着查。401 Unauthorized最常见。原因通常是 API Key 写错、Authorization头少了Bearer前缀、或者 Key 前后带了空格。检查.env里TAOTOKEN_API_KEY的值确认没有多余引号。如果用的是 TaoToken去 API Keys 页面重新复制一次 Keyhttps://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。local proxy failed / connection refused这类报错说明请求根本没发出去通常是 Base URL 写错或本地网络配置问题。确认TAOTOKEN_BASE_URL是https://taotoken.net/api结尾不要多加斜杠或/v1。如果你在代码里用了自定义的 HTTP 客户端检查它有没有被环境变量里的代理设置干扰。Cannot read properties of undefined (reading choices)这个报错说明data.choices是 undefined也就是返回体结构和你预期的不一样。多半是请求失败但你没检查状态码直接取了data.choices[0]。改成先判断const res await fetch(url, options); if (!res.ok) { const text await res.text(); throw new Error(HTTP ${res.status}: ${text}); } const data await res.json(); console.log(data.choices?.[0]?.message?.content ?? no content);OAuth / token expired如果你用的是带 OAuth 的客户端比如某些 IDE 插件报这个说明 token 过期了重新走一遍授权流程即可。纯 API Key 方式不会遇到这个。聚合相关报错The field xxx must be an accumulator object说明$group里写了非累加器字段检查是不是把分组键写成了普通字段名分组键只能叫_id。$lookup返回空数组检查from的集合名对不对、localField和foreignField类型是否一致ObjectId 对 ObjectIdString 对 String类型不匹配匹配不上。如果你用的是 Claude Code 这类工具辅助写聚合配置里同样要写全三件套。Base URL 填https://taotoken.net/apiKey 填你的 API KeyModel ID 填你选的模型。三者缺一要么 401要么模型不存在。接入细节看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。排查时有个通用思路先确认请求发出去了没看状态码再看返回体结构打印原始 text最后才看业务逻辑。很多人一上来就怀疑聚合写错其实卡在鉴权那一步。6. 把聚合能力接进你的编码工作流聚合管道写多了会发现真正费时间的不是语法而是字段名对不上、阶段顺序放错、结果和预期差一点。这时候让模型帮你逐阶段核对比反复跑脚本快。你可以把getOrderStats的管道贴到模型对话页让它指出$lookup的from是否和实际集合名一致、$project有没有漏字段https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。如果你每天都要写这类查询把它固化进编码流程更划算。Coding Plan 适合长期在编辑器里辅助写聚合、查报错的场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。API Key 在控制台随时可以新建和轮换https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。最后留一个我常用的调试习惯把管道拆成单阶段跑。先只跑$match看过滤后的数据对不对再加$group看分组结果再加$lookup看关联有没有命中。一次加一个阶段哪一步结果不对就停在哪比一次性跑完整管道再猜哪里错高效得多。聚合管道本质是流水线逐段验证问题自然浮出来。