Dataset 数据模块:评测的"米面粮油"
从今天起我们进入 Coze Loop 的业务模块。第一站是 backend/modules/data——不管是评测集、还是给评估器打分喂数据,最终都要落到"数据集里有哪些条目(Item)"这件事上。今天把 Dataset 从"新建 → 写数据 → 打版本 → 导入导出"这条线走一遍,顺便认识给数据"打标签"的 Tag 体系。
为什么从 Dataset 开始
| 概念 | 代码位置 | 大白话 |
|---|---|---|
Dataset | domain/dataset/entity/dataset.go | 数据集本身:名字、分类、规格上限 |
Schema | domain/dataset/entity/schema.go | 这个数据集有哪些字段(列) |
Item | domain/dataset/entity/item.go | 一行具体数据 |
Version | domain/dataset/entity/version.go | 某个时刻的只读快照 |
Job | domain/dataset/entity/job.go | 导入/导出等异步任务 |
Tag | domain/tag/ | 给条目打标签,用于筛选、人工标注 |
Dataset 长什么样:从实体到表结构
先看核心实体 backend/modules/data/domain/dataset/entity/dataset.go(节选真实字段):
type Dataset struct {
ID int64
AppID int32
SpaceID int64
SchemaID int64
Name string
Description *string
Category DatasetCategory // 业务场景分类:general/training/validation/evaluation
Status DatasetStatus // 可用/已删除/已过期/导入中…
SecurityLevel SecurityLevel // 安全等级
Visibility DatasetVisibility // 可见性
Spec *DatasetSpec // 规格:最大条数/字段数/单条大小
Features *DatasetFeatures // 功能开关:能否改 schema、多轮、多模态
LatestVersion string // 最新版本号
NextVersionNum int64 // 下一个版本的数字版本号
LastOperation DatasetOpType // 最近一次操作(用于判断"是否有未提交改动")
...
}
IsChangeUncommitted()
dataset.go:40-47 有个方法判断"这个数据集是不是有没打版本的改动"——只要 LastOperation 不是"创建"或"创建版本",就说明有未提交的变更。前端会用这个字段提示用户"你改了数据但还没生成新版本,实验跑的可能不是你以为的那批数据"。这是个很容易被忽略但很重要的细节。对应的 MySQL 表在 release/deployment/docker-compose/bootstrap/mysql-init/init-sql/dataset.sql,字段和实体几乎一一对应:
CREATE TABLE IF NOT EXISTS `dataset` (
`id` bigint unsigned NOT NULL AUTO_INCREMENT,
`space_id` bigint unsigned NOT NULL DEFAULT '0' COMMENT '空间 ID',
`schema_id` bigint unsigned NOT NULL DEFAULT '0' COMMENT 'Schema ID',
`name` varchar(255) NOT NULL DEFAULT '' COMMENT '数据集名称',
`category` varchar(64) NOT NULL DEFAULT '' COMMENT '业务场景分类',
`status` varchar(128) NOT NULL DEFAULT '' COMMENT '状态',
`spec` json DEFAULT NULL COMMENT '规格配置',
`features` json DEFAULT NULL COMMENT '功能开关',
`latest_version` varchar(64) NOT NULL DEFAULT '' COMMENT '最新版本号',
`next_version_num` bigint unsigned NOT NULL DEFAULT '1',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_space_id_category_name` (`space_id`, `category`, `name`, `deleted_at`)
) ENGINE = InnoDB;
uk_space_id_category_name 保证"同一个空间下、同一个业务分类里,数据集名字不能重复"——但换个分类(比如从 general 换成 evaluation)名字就能重复,这也解释了为什么 Category 要放进唯一键。application 层四件套:谁负责什么
backend/modules/data/application/ 下按"业务动作"拆成了四个文件,全部挂在同一个 DatasetApplicationImpl 结构体上(Go 里同一个类型可以在多个文件里继续加方法):
dataset_app.go(338 行)
数据集本体的增删改查:CreateDataset / UpdateDataset / GetDataset / ListDatasets。
item_app.go(539 行,最大)
条目的批量写入/更新/删除/查询:BatchCreateDatasetItems / ListDatasetItems 等。
version_app.go(176 行)
版本管理:CreateDatasetVersion / ListDatasetVersions。
job_app.go(132 行)
导入导出等异步任务:ImportDataset 创建 Job,真正搬数据的活交给消费者。
以 CreateDataset(dataset_app.go:73)为例,看看一个写请求的标准套路:
func (h *DatasetApplicationImpl) CreateDataset(ctx context.Context, req *dataset.CreateDatasetRequest) (resp *dataset.CreateDatasetResponse, err error) {
userID := session.UserIDInCtxOrEmpty(ctx)
// 1. 鉴权:这个用户在这个空间有没有"创建评测集"的权限
err = h.auth.Authorization(ctx, &rpc.AuthorizationParam{
ObjectID: strconv.FormatInt(req.WorkspaceID, 10),
SpaceID: req.WorkspaceID,
ActionObjects: []*rpc.ActionObject{{Action: gptr.Of(rpc.CozeActionCreateLoopEvaluationSet), ...}},
})
// 2. DTO → DO:把接口请求结构体转成领域实体
set := &entity.Dataset{ Name: req.GetName(), Category: convertor.ConvertCategoryDTO2DO(req.GetCategory()), ... }
// 3. 内容审核:把名字/描述/字段拼起来送审
record, err := h.auditClient.Audit(ctx, audit.AuditParam{ AuditType: audit.AuditType_CozeLoopDatasetModify, ... })
// 4. 调用 domain 层 service 真正落库
...
}
domain/dataset/service 和 infra/repo 的事),只负责"这一次调用该走哪些检查、组装什么参数"。这也是为什么 Day 15 会说 application 层应该"薄"——它是编排者,不是执行者。domain/dataset/service 或 entity 的方法上。application 只负责"调用顺序",不负责"规则本身",否则规则会散落在到处都是,未来改一条规则要满仓库找文件。Item:一条数据怎么存下来
BatchCreateDatasetItems(item_app.go:31)要把这些复杂结构安全地校验、批量落库。authByDatasetID鉴权:这个数据集我能不能写h.prepare(ctx, req)按 Schema 校验每条数据的字段类型/长度,分成 goodItems / badItemsh.svc.BatchCreateItems调用 domain service,真正批量写库(支持"部分成功")h.buildResp把成功、失败的条目分别整理成响应,前端能看到哪几条为什么失败PartialAdd 选项让"997 条先进去,3 条报错告诉你原因",这是数据导入类功能几乎必须有的体验。数据表对应 .../mysql-init/init-sql/dataset_item.sql 和 dataset_item_snapshot.sql(打版本时的快照表)——草稿数据和"打版本后的快照"分表存放,快照表只读不再变化,这样实验引用某个版本时数据永远稳定。
dataset_item 表);点"创建版本"后,当前草稿会被复制一份写进 dataset_item_snapshot,从此这个版本号对应的数据永远不变——即使你之后继续改草稿。Day 13 的实验只认版本号,就是靠这个不变性保证"可复现"。Job:导入导出为什么要"异步"
job_app.go 里的 ImportDataset 并不会立刻去读你上传的 CSV/JSONL 文件解析写库——它只是创建一条 Job 记录立刻返回:
func (h *DatasetApplicationImpl) ImportDataset(ctx context.Context, req *dataset.ImportDatasetRequest) (r *dataset.ImportDatasetResponse, err error) {
ds, err := h.checkImportDatasetReq(ctx, req) // 校验字段映射是否匹配 Schema
job := h.buildJob(ctx, req, ds)
do := convertor.IOJobDTO2DO(job)
if err := h.svc.CreateIOJob(ctx, do); err != nil { return nil, err }
return &dataset.ImportDatasetResponse{JobID: gptr.Of(do.ID)}, nil // 立刻返回 JobID,不等文件处理完
}
JobID 轮询 GetDatasetIOJob 查看进度。IJobRunMsgHandler 接口里的 RunIOJob(dataset_app.go:31-34)就是消费者回调的入口。Tag 标签体系:给数据"打标"
data 域但独立成一套体系。
核心实体 TagKey(domain/tag/entity/tag_key.go:25-46):
type TagKey struct {
ID int64
TagKeyID int64
TagKeyName string
TagType TagType // 标签的形态:单选/多选/连续值…
TagTargetType []TagTargetType // 能打在什么对象上:数据集条目/实验结果…
TagValues []*TagValue // 可选的取值枚举
ChangeLogs []*ChangeLog // 变更历史
...
}
TagKey 定义"有一个叫『满意度』的标签,取值是 1-5 分"(相当于 Schema),真正打在某条数据上的"这条数据是 4 分"叫 AnnotateRecord(Day 13 会再遇到,用于实验结果的人工标注)。先定义规则、再产生数据是这个仓库里反复出现的建模思路。路由集中在 backend/api/handler/coze/loop/apis/tag_service.go,都挂在 /api/data/v1/tags* 下:
# 真实路由(tag_service.go 注释里的 @router)
POST /api/data/v1/tags # 创建标签
PATCH /api/data/v1/tags/:tag_key_id # 更新标签
POST /api/data/v1/tags/batch_update_status # 批量启停用
POST /api/data/v1/tags/search # 搜索标签
GET /api/data/v1/tags/:tag_key_id/detail # 标签详情
GET /api/data/v1/tag_spec # 标签规格(有哪些类型/限制)
POST /api/data/v1/tags/batch_get # 批量获取
POST /api/data/v1/tags/:tag_key_id/archive_option_tag # 归档某个可选项
/api/data/v2/datasets*(比如 POST /api/data/v2/datasets),而 Tag 和"校验条目"接口用的是 /api/data/v1/...——同一个模块下混用 v1/v2 很正常:v2 是 Dataset 接口做过一次不兼容重构后升的版本号,Tag 相关接口还没经历过这种重构,所以停在 v1。读路由时不要假设"同模块版本号一定统一",以每个接口自己的 @router 注释为准。今日小结 + 动手 + 预告
🧠 今天你应该能回答
- Dataset 由哪四个核心概念组成?(Schema / Item / Version / Job)
- 草稿数据和"打版本后的快照"分别存在哪张表?为什么要分开?
- 为什么导入导出接口要设计成"异步 Job"而不是同步等待?
- TagKey 和 AnnotateRecord 的关系类比 Dataset 里的哪两个概念?
- application 层的"标准四步"是什么?为什么业务规则不该写在这里?
✋ 动手 5 分钟(可选)
# 1. 看四件套 application 文件各自的方法列表
grep -n "^func (h \*DatasetApplicationImpl)" backend/modules/data/application/*.go
# 2. 数一数 Dataset 相关表有多少张
ls release/deployment/docker-compose/bootstrap/mysql-init/init-sql/ | grep dataset
# 3. 看 Dataset 实体的"未提交改动"判断逻辑
sed -n '40,47p' backend/modules/data/domain/dataset/entity/dataset.go
# 4. 找出 Tag 模块所有真实路由
grep -n "@router" backend/api/handler/coze/loop/apis/tag_service.go
evaluation_set_app.go 和 evaluator_app.go,认识评测的"裁判"——评估器(Evaluator),看它凭什么给一次回答判个分数。