diff --git a/data/README.md b/data/README.md index 76ce4c9..5111788 100644 --- a/data/README.md +++ b/data/README.md @@ -1,18 +1,41 @@ # data/ — 数据目录 +## 层（语义） + +```text +raw + ↓ preprocessing +processed + ↓ canonical resolution（script/canonical，冻结契约） +canonical + ↓ SFT/RL export（script/verl/sft/export、script/verl/rl/export） +parquet（data/sft、data/rl） +``` + +- `raw/`：source of truth 的原始导入物（CSV/XLSX / 原始包）。**训练脚本不得直接修改**。 + 真原始表格由数据维护方在 `data/raw/.{csv,xlsx}` 提供（preprocessing + `prepare` 默认查找该路径）。`data/raw/` 为 gitignored，需要时由脚本/维护方创建。 +- `processed//`：预处理结果（`all.json` + `train.json` / `val.json` / + `test.json` + `split_report.json`）。split 按 record id 隔离，是后续 canonical 的 + 输入层，也是 `data/` 归一化源层的正式位置。 +- `canonical//`：canonical 契约层（`all.json` + `resolution_report.json`）。 + **后续所有算法（SFT/RL/评估）统一消费该层**；`data_level` 仅作 provenance。 +- `sft/`、`rl/`：可再生训练派生产物（`*.parquet` + export_report.json），gitignored； + 源码变动后可删除并重新 export。 +- `knowledge/standards_map/`：标准知识（12 份词典，入库），不属于训练数据生成产物。 + ## 下载 数据来自 ModelScope ```bash export MODELSCOPE_TOKEN=xxx # token 走环境变量，禁止写入脚本/文档/仓库 -git clone . +git clone . # 导入到 raw/（原始 CSV/XLSX） ``` - ## 各领域概况（2025-08-10 快照） -| 领域 | 业务域 | 记录数 | train/val/test | +| 领域 | 业务域 | 记录数(processed all.json) | train/val/test | |---|---|---|---| | finance | 信托核心系统 | 568 | 439/69/58 | | infra | 钢铁基建（=shougang 子集） | 64 | 40/15/9 | @@ -22,4 +45,104 @@ git clone . ## 入库内容 - `knowledge/standards_map/`：12 份多领域分类分级标准词典（公开标准知识，入库） -- 本文档：下载说明 +- 本文档：下载说明 + 分层语义 + +## 如何使用 `data/processed`（processed → canonical → parquet → 训练/评估） + +> 这是把 `data/processed` 交付给他人后、从零生成训练数据的标准步骤。 +> 只需要：仓库代码（含 `cfg/task/registry` + `cfg/task/corpus`，clone 自带）+ `data/processed` +> + Python 3.10+ 且装有 pyarrow。**不需要** `raw` / `knowledge` / 已有 `canonical`/`sft`/`rl`。 + +每步失败会清晰报错且不产生产物（fail-fast）；已产出时重跑需 `--overwrite`。 + +**1) canonical（processed → `data/canonical//all.json` + resolution_report.json）** + +```bash +python -m script.canonical.targets --overwrite --datasets finance infra pers_info shougang +# 或单数据集：--dataset pers_info +``` + +**2) SFT parquet（canonical → `data/sft//{train,val,test}.parquet`）** + +```bash +for ds in finance infra pers_info shougang; do + python -m script.verl.sft.export \ + --canonical data/canonical/$ds/all.json \ + --split-dir data/processed/$ds \ + --output-dir data/sft/$ds \ + --registry cfg/task/registry/$ds.registry.json \ + --corpus cfg/task/corpus/$ds.corpus.json \ + --metadata-fields field_name field_description +done +``` + +**3) RL parquet（canonical → `data/rl//{train,val,test}.parquet`，五字段）** + +```bash +for ds in finance infra pers_info shougang; do + python -m script.verl.rl.export \ + --canonical data/canonical/$ds/all.json \ + --split-dir data/processed/$ds \ + --output-dir data/rl/$ds \ + --dataset $ds \ + --registry cfg/task/registry/$ds.registry.json \ + --corpus cfg/task/corpus/$ds.corpus.json \ + --metadata-fields field_name field_description +done +``` + +**4) 校验**（契约 + token 预算；token 预算需要模型 tokenizer） + +```bash +python -m script.verl.sft.validate --dataset-dir data/sft/pers_info \ + --registry cfg/task/registry/pers_info.registry.json --corpus cfg/task/corpus/pers_info.corpus.json \ + --metadata-fields field_name field_description +python -m script.verl.sft.check_token_budget --dataset-dir data/sft/pers_info \ + --model --max-length 512 +``` + +**5) 训练 / 评估**（需要 verl 环境 + GPU + 模型） + +```bash +# SFT baseline（7B LoRA） +DATASET=pers_info DATA_DIR=data/sft/pers_info MODEL_PATH= \ + bash script/verl/sft/run_baseline.sh +# RL smoke（GRPO） +DATASET=pers_info TRAIN_FILE=data/rl/pers_info/train.parquet VAL_FILE=data/rl/pers_info/val.parquet \ + MODEL_PATH= bash script/verl/rl/grpo_smoke.sh +# 评估（choice protocol） +python -m script.verl.sft.evaluate_baseline --model-path \ + --data data/sft/pers_info/test.parquet --registry cfg/task/registry/pers_info.registry.json \ + --report tmp/eval.json +python -m script.verl.sft.evaluate_true_e2e --model-path \ + --data data/sft/pers_info/test.parquet --registry cfg/task/registry/pers_info.registry.json \ + --corpus cfg/task/corpus/pers_info.corpus.json --report tmp/eval_e2e.json +``` + +**预期产物 / 校验**（与迁移前仓库逐字节一致的基准行数） + +| dataset | canonical resolved | SFT/RL train | val | test | +|---|---|---|---|---| +| finance | 531 | 806 | 138 | 114 | +| infra | 64 | 80 | 30 | 18 | +| pers_info | 176 | 280 | 36 | 36 | +| shougang | 18393 | 29042 | 3852 | 3892 | + +（shougang canonical 含 1022 条 `placeholder` 不入训；finance 有 34 missing_leaf / 3 path_mismatch 不入训。） +生成 SFT/RL parquet 为**纯 pyarrow 计算**（无 torch/GPU）；训练/评估阶段才需要 GPU 与模型。 +新数据集上线另见 `docs/新数据集运行说明.md`（raw → processed 的完整流程）。 + +## 迁移说明（2026-08，data layout refactor） + +- 原 `data//all.json + splits`（预处理归一化源层）→ `data/processed//`。 +- 原 `data//canonical/*` → `data/canonical//`。 +- 新增 `data/raw/`（原始物）、`data/legacy/`（历史遗留物）。 +- **注意：data 数据文件均为 gitignored，本 PR 不会通过 git 自动迁移现有本地数据。** + 已有旧布局（`data//`）的环境需要手动将 `all.json`+splits 移至 + `data/processed//`、`canonical/*` 移至 `data/canonical//`， + 或按 `data/README.md` §「如何使用 data/processed」从 raw 重新生成。 +- 删除与 `all.json` 逐字节重复的 `data/shougang/all_shougang.json`。 +- `data//corpus.json`（legacy finance corpus，仅 analysis 对照用）→ + `data/legacy/finance.corpus.json`；正式 corpus 唯一来源为 `cfg/task/corpus/`。 +- 本次迁移只改目录结构与路径引用，样本 / label / split / canonical resolution / + prompt / reward 均未改动。 diff --git a/docs/SFT_BASELINE.md b/docs/SFT_BASELINE.md index 39f30d2..b6b71a1 100644 --- a/docs/SFT_BASELINE.md +++ b/docs/SFT_BASELINE.md @@ -28,7 +28,7 @@ VERL_ENV_DIR=/envs/verl \ ## Export Training labels come **exclusively** from the canonical dataset contract: -records are read from `data//canonical/all.json`, only +records are read from `data/canonical//all.json`, only `resolution_status == "resolved"` records with a target whose `category_id` belongs to the LeafRegistry enter training, and the ground truth is always `target.category_id`. `classification.level_1..level_4` stay provenance only; @@ -44,8 +44,8 @@ resolves the five candidates by category_id against the canonical corpus. ```bash python -m script.verl.sft.export \ - --canonical data/pers_info/canonical/all.json \ - --split-dir data/pers_info \ + --canonical data/canonical/pers_info/all.json \ + --split-dir data/processed/pers_info \ --output-dir data/sft/pers_info \ --registry cfg/task/registry/pers_info.registry.json \ --corpus cfg/task/corpus/pers_info.corpus.json \ diff --git a/docs/design/data_contract.md b/docs/design/data_contract.md index a7ac65f..d401e74 100644 --- a/docs/design/data_contract.md +++ b/docs/design/data_contract.md @@ -15,7 +15,7 @@ truth for registry, corpus, ground truth, SFT labels, evaluation and reward. `docs/design/prompt_interface.md`); they never leak back into canonical records or reward semantics. -## 2. Canonical sample schema (`data//canonical/all.json`) +## 2. Canonical sample schema (`data/canonical//all.json`) Input sample keeps the original fields (`classification`, `metadata`, `data_level`, `label_status`, …) untouched as provenance. The resolver @@ -72,7 +72,7 @@ appends: Only records with `resolution_status == "resolved"` **and** `target.category_id ∈ LeafRegistry` **and** `target.leaf_name == registry.get(category_id).name` enter training; violations fail fast (no silent skip). Split membership follows -the original `data//{train,val,test}.json` by record id. `canonical` +the original `data/processed//{train,val,test}.json` by record id. `canonical` resolved / trainable counts (trainable = resolved within split boundaries): | dataset | canonical resolved | trainable | outside splits | diff --git a/docs/phase_reports/stage3c_report.md b/docs/phase_reports/stage3c_report.md index 72be8a8..7fb95b6 100644 --- a/docs/phase_reports/stage3c_report.md +++ b/docs/phase_reports/stage3c_report.md @@ -1,5 +1,11 @@ # 阶段 3C：canonical data contract 接入 SFT 数据流水线 +> **2026-08 data layout migration note**：本文档正文中的路径为迁移前的旧布局 +> （`data//all.json` + `data//canonical/`）。新布局： +> `data/processed//all.json`（含 splits）→ `data/canonical//all.json` +> （canonical 契约）→ `data/sft//`、`data/rl//`（parquet）。 +> 内容/语义未变，仅目录结构。 + ## 1. 修改/新增文件 修改： diff --git "a/docs/\346\226\260\346\225\260\346\215\256\351\233\206\350\277\220\350\241\214\350\257\264\346\230\216.md" "b/docs/\346\226\260\346\225\260\346\215\256\351\233\206\350\277\220\350\241\214\350\257\264\346\230\216.md" index 355970c..d035bd6 100644 --- "a/docs/\346\226\260\346\225\260\346\215\256\351\233\206\350\277\220\350\241\214\350\257\264\346\230\216.md" +++ "b/docs/\346\226\260\346\225\260\346\215\256\351\233\206\350\277\220\350\241\214\350\257\264\346\230\216.md" @@ -1,5 +1,9 @@ # TransClass 新数据集运行说明 +> 本文档覆盖 `raw → processed`（数据预处理/切分/语料）。 +> 拿到 `data/processed` 之后如何生成 `canonical → SFT/RL parquet → 训练/评估` 的标准步骤 +> 见 `data/README.md` §「如何使用 data/processed」。 + ## 一、放置原始数据 将 CSV 或 XLSX 文件放到： diff --git a/script/analysis/analyze_dataset_corpus_alignment.py b/script/analysis/analyze_dataset_corpus_alignment.py index 16d591d..aef66ee 100644 --- a/script/analysis/analyze_dataset_corpus_alignment.py +++ b/script/analysis/analyze_dataset_corpus_alignment.py @@ -2,8 +2,8 @@ string matching (no semantic models, no fuzzy auto-fixing of labels). Reads: - - datasets: data//all.json (normalized TransClass JSON) - - leaf corpora: data//corpus.json (level_4 + description documents) + - datasets: data/processed//all.json (normalized TransClass JSON) + - leaf corpora: data/legacy/*.corpus.json (legacy/archived dataset corpus) - standards: data/knowledge/standards_map/*.json Writes: @@ -25,6 +25,10 @@ PROJECT_ROOT = Path(__file__).resolve().parents[2] DEFAULT_DATA_DIR = PROJECT_ROOT / "data" +# post-migration layout: normalized records live under processed/ +DEFAULT_PROCESSED_DIR = DEFAULT_DATA_DIR / "processed" +# legacy / archived dataset-level artifacts (never a formal corpus source) +DEFAULT_LEGACY_DIR = DEFAULT_DATA_DIR / "legacy" DEFAULT_OUT_DIR = PROJECT_ROOT / "artifacts" / "generated" / "alignment" CLASSIFICATION_FIELDS = ("level_1", "level_2", "level_3", "level_4") @@ -158,7 +162,7 @@ def _json_safe(value: Any) -> Any: def load_dataset(name: str) -> list[dict[str, Any]]: - path = DEFAULT_DATA_DIR / name / "all.json" + path = DEFAULT_PROCESSED_DIR / name / "all.json" with path.open("r", encoding="utf-8") as file: data = json.load(file) if not isinstance(data, list): @@ -608,7 +612,7 @@ def match_dataset_to_corpus( def discover_datasets() -> dict[str, list[dict[str, Any]]]: datasets: dict[str, list[dict[str, Any]]] = {} - for child in sorted(DEFAULT_DATA_DIR.iterdir()): + for child in sorted(DEFAULT_PROCESSED_DIR.iterdir()): all_path = child / "all.json" if child.is_dir() and all_path.is_file(): datasets[child.name] = load_dataset(child.name) @@ -617,10 +621,16 @@ def discover_datasets() -> dict[str, list[dict[str, Any]]]: def discover_corpora() -> dict[str, tuple[Path, list[dict[str, Any]]]]: corpora: dict[str, tuple[Path, list[dict[str, Any]]]] = {} - for child in sorted(DEFAULT_DATA_DIR.iterdir()): - corpus_path = child / "corpus.json" - if child.is_dir() and corpus_path.is_file(): - corpora[f"corpus:{child.name}"] = (corpus_path, _iter_entries(corpus_path)) + # legacy dataset-level corpus (e.g. the archive-only finance corpus, now + # under data/legacy/); the formal corpus is cfg/task/corpus and is covered + # by the dataset/corpus alignment via canonical records, not here. + if DEFAULT_LEGACY_DIR.is_dir(): + for path in sorted(DEFAULT_LEGACY_DIR.glob("*.corpus.json")): + # 文件形如 finance.corpus.json -> dataset "finance"（避免 Path.stem + # 产生 finance.corpus 双后缀，与下游 build_schema_issues 的硬编码 + # "corpus:" 保持一致） + dataset = path.name.removesuffix(".corpus.json") + corpora[f"corpus:{dataset}"] = (path, _iter_entries(path)) standards_dir = DEFAULT_DATA_DIR / "knowledge" / "standards_map" for path in sorted(standards_dir.glob("*.json")): if path.name == "generate_standards_map.py": @@ -853,7 +863,7 @@ def build_schema_issues( "whitespace removal but different raw text" ), "detail": grouped_text + " — fix in the upstream input " - "(data/finance/all.json), never auto-repair.", + "(data/processed/finance/all.json), never auto-repair.", } ) l3 = finance["level_stats"]["level_3"] @@ -888,7 +898,7 @@ def build_schema_issues( issues.append( { "severity": "high", - "where": "corpus:finance (data/finance/corpus.json)", + "where": "corpus:finance (data/legacy/finance.corpus.json)", "issue": "leaf-only corpus: category identity is the bare level_4 name; " "no path/code is kept, so any future leaf-name collision is unresolvable", "detail": "all 220 unique level_4 labels; no path or code fields", @@ -1089,7 +1099,7 @@ def render_markdown(report: dict[str, Any]) -> str: add("# 数据对齐分析报告 (dataset ↔ corpus alignment)") add("") add("> 方法约束：仅使用精确字符串与空白归一化匹配，不使用语义模型或模糊匹配修标签。") - add("> 数据来源：`data//all.json`、`data//corpus.json`、`data/knowledge/standards_map/*.json`。") + add("> 数据来源：`data/processed//all.json`、`data/legacy/*.corpus.json`、`data/knowledge/standards_map/*.json`。") add("") # 1. datasets @@ -1234,7 +1244,7 @@ def render_markdown(report: dict[str, Any]) -> str: conclusions = report["conclusions"] add("### 4.1 实际有哪些 dataset") add("") - add(", ".join(f"`{name}`" for name in conclusions["datasets_found"]) + "（以存在 `all.json` 为准）") + add(", ".join(f"`{name}`" for name in conclusions["datasets_found"]) + "（以存在 processed/all.json 为准）") add("") add("### 4.2 每个 dataset 对应哪个 corpus") add("") diff --git a/script/canonical/cli.py b/script/canonical/cli.py index 1ab215d..af1eac8 100644 --- a/script/canonical/cli.py +++ b/script/canonical/cli.py @@ -12,7 +12,7 @@ Sources: - shougang / infra: data/knowledge/standards_map/guanji_dict.json - finance: data/knowledge/standards_map/financial_standards_dict.json -- pers_info: data//all.json (dataset universe, no standard exists) +- pers_info: data/processed//all.json (dataset universe, no standard exists) All builds and coverage diagnostics are computed before anything is written; every output file is written exactly once. Artifact sources are @@ -60,9 +60,11 @@ def _load_json(path: Path) -> Any: def _load_records(data_dir: Path, dataset: str) -> list[dict[str, Any]]: - records = _load_json(data_dir / dataset / "all.json") + records = _load_json(data_dir / "processed" / dataset / "all.json") if not isinstance(records, list): - raise ValueError(f"{data_dir / dataset / 'all.json'} must be a JSON list") + raise ValueError( + f"{data_dir / 'processed' / dataset / 'all.json'} must be a JSON list" + ) return records @@ -91,7 +93,7 @@ def _annotate_coverage( data_dir: Path, ) -> None: """Read-only registry vs resolver-ID coverage; mutates the report only.""" - records_path = data_dir / dataset / "all.json" + records_path = data_dir / "processed" / dataset / "all.json" if not records_path.is_file(): report.dataset_id_coverage = {"available": False} return diff --git a/script/canonical/targets.py b/script/canonical/targets.py index e3148b9..5be40c4 100644 --- a/script/canonical/targets.py +++ b/script/canonical/targets.py @@ -3,11 +3,12 @@ Usage: python -m script.canonical.targets [--dataset finance] [--datasets ...] [--overwrite] -Writes per dataset: - data//canonical/all.json +Input: data/processed//all.json (normalized records) +Writes into data/canonical//: + all.json every input record unchanged (classification untouched) plus "resolution_status" and, for resolved records, "target". - data//canonical/resolution_report.json + resolution_report.json status counts, unresolved details, registry facts, input sha256. The LeafRegistry is the final constraint: a resolver target only counts as @@ -36,7 +37,19 @@ def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--data-dir", type=Path, default=PROJECT_ROOT / "data") + parser.add_argument( + "--processed-dir", + type=Path, + default=PROJECT_ROOT / "data" / "processed", + help="Where processed records live: processed-dir//all.json", + ) + parser.add_argument( + "--canonical-dir", + type=Path, + default=PROJECT_ROOT / "data" / "canonical", + help="Where the canonical contract is written: canonical-dir//", + ) + parser.add_argument("--registry-dir", type=Path, default=PROJECT_ROOT / "cfg" / "task" / "registry") parser.add_argument("--corpus-dir", type=Path, default=PROJECT_ROOT / "cfg" / "task" / "corpus") parser.add_argument("--dataset", type=str, choices=list(DEFAULT_DATASETS)) @@ -59,10 +72,10 @@ def main(argv: list[str] | None = None) -> int: # 1. fail fast before building/writing anything if not args.overwrite: existing = [ - Path(args.data_dir) / dataset / "canonical" / name + Path(args.canonical_dir) / dataset / name for dataset in datasets for name in ("all.json", "resolution_report.json") - if (Path(args.data_dir) / dataset / "canonical" / name).exists() + if (Path(args.canonical_dir) / dataset / name).exists() ] if existing: raise FileExistsError( @@ -78,7 +91,8 @@ def main(argv: list[str] | None = None) -> int: prepared.append( prepare_canonical_dataset( dataset, - data_dir=args.data_dir, + processed_dir=args.processed_dir, + canonical_dir=args.canonical_dir, registry_dir=args.registry_dir, corpus_dir=args.corpus_dir, ) diff --git a/script/verl/rl/export.py b/script/verl/rl/export.py index 937c951..13d4fd8 100644 --- a/script/verl/rl/export.py +++ b/script/verl/rl/export.py @@ -22,7 +22,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser.add_argument( "--canonical", required=True, - help="data//canonical/all.json (canonical dataset records)", + help="data/canonical//all.json (canonical dataset records)", ) parser.add_argument( "--split-dir", diff --git a/script/verl/sft/export.py b/script/verl/sft/export.py index c7635d4..0d3f29e 100644 --- a/script/verl/sft/export.py +++ b/script/verl/sft/export.py @@ -17,7 +17,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser.add_argument( "--canonical", required=True, - help="data//canonical/all.json (canonical dataset records)", + help="data/canonical//all.json (canonical dataset records)", ) parser.add_argument( "--split-dir", diff --git a/src/agent/task/canonical_dataset.py b/src/agent/task/canonical_dataset.py index 318877f..6ab703d 100644 --- a/src/agent/task/canonical_dataset.py +++ b/src/agent/task/canonical_dataset.py @@ -9,13 +9,17 @@ auto-repaired. Output design (per dataset): - data//canonical/all.json + data/canonical//all.json every input record, unchanged (classification untouched), plus: - "resolution_status": one of the ResolutionStatus values - "target": canonical target (only for resolved records) - data//canonical/resolution_report.json + data/canonical//resolution_report.json status counts, unresolved details, registry facts, deterministic metadata (input sha256; no timestamps, no machine-local paths). + +Input runs from data/processed//all.json (preprocessing output); +canonical is a pure path-renumbering of the processed layer, never a rewrite +of samples / labels / split membership. """ from __future__ import annotations @@ -163,18 +167,23 @@ def resolve_record( def prepare_canonical_dataset( dataset: str, *, - data_dir: str | Path, + processed_dir: str | Path, + canonical_dir: str | Path, registry_dir: str | Path, corpus_dir: str | Path, ) -> tuple[CanonicalDatasetResult, list[dict[str, Any]]]: """Load, validate, resolve and build the report for one dataset. - Pure computation: writes nothing. The caller decides when to write via - write_canonical_dataset(), which enables cross-dataset fail-fast (no - partial outputs when any selected dataset fails). + Reads the processed records (``processed_dir//all.json``) and + resolves them against the LeafRegistry; the canonical contract is targeted + at ``canonical_dir//all.json``. Pure computation: writes nothing. + The caller decides when to write via write_canonical_dataset(), which + enables cross-dataset fail-fast (no partial outputs when any selected + dataset fails). """ - data_dir = Path(data_dir) - input_path = data_dir / dataset / "all.json" + processed_dir = Path(processed_dir) + canonical_dir = Path(canonical_dir) + input_path = processed_dir / dataset / "all.json" if not input_path.is_file(): raise FileNotFoundError(f"input dataset not found: {input_path}") @@ -250,7 +259,7 @@ def prepare_canonical_dataset( code_unresolved_by_leaf[result.leaf] += 1 output_records.append(canonical) - out_dir = data_dir / dataset / "canonical" + out_dir = canonical_dir / dataset out_all = out_dir / "all.json" unresolved_details: dict[str, Any] = { @@ -308,15 +317,22 @@ def write_canonical_dataset( def build_canonical_dataset( dataset: str, *, - data_dir: str | Path, + processed_dir: str | Path, + canonical_dir: str | Path, registry_dir: str | Path, corpus_dir: str | Path, overwrite: bool = False, ) -> CanonicalDatasetResult: - """Convenience wrapper: prepare + overwrite/existence check + write.""" - data_dir = Path(data_dir) - out_all = data_dir / dataset / "canonical" / "all.json" - out_report = data_dir / dataset / "canonical" / "resolution_report.json" + """Convenience wrapper: prepare + overwrite/existence check + write. + + Reads processed records from ``processed_dir//all.json`` and + writes the canonical contract to ``canonical_dir//all.json`` + + ``resolution_report.json``. + """ + processed_dir = Path(processed_dir) + canonical_dir = Path(canonical_dir) + out_all = canonical_dir / dataset / "all.json" + out_report = canonical_dir / dataset / "resolution_report.json" if (out_all.exists() or out_report.exists()) and not overwrite: raise FileExistsError( f"canonical output exists for {dataset}: {out_all.parent} " @@ -324,7 +340,8 @@ def build_canonical_dataset( ) result, output_records = prepare_canonical_dataset( dataset, - data_dir=data_dir, + processed_dir=processed_dir, + canonical_dir=canonical_dir, registry_dir=registry_dir, corpus_dir=corpus_dir, ) diff --git a/src/agent/training/sft/dataset.py b/src/agent/training/sft/dataset.py index 1f19c6e..4fe3fbf 100644 --- a/src/agent/training/sft/dataset.py +++ b/src/agent/training/sft/dataset.py @@ -1,7 +1,7 @@ """Exporter and validator for the VeRL SFT messages parquet baseline. Production labels come EXCLUSIVELY from the canonical dataset contract: -records are read from data//canonical/all.json, only +records are read from data/canonical//all.json, only resolution_status == "resolved" records with a target whose category_id belongs to the LeafRegistry enter training, and the ground truth is always record["target"]["category_id"]. classification.level_1..level_4 stay diff --git a/tests/analysis/test_alignment_legacy_corpus.py b/tests/analysis/test_alignment_legacy_corpus.py new file mode 100644 index 0000000..36b277a --- /dev/null +++ b/tests/analysis/test_alignment_legacy_corpus.py @@ -0,0 +1,50 @@ +"""Alignment CLI tests: legacy corpus discovery keys after the data-layout +migration (data/legacy/*.corpus.json). + +Regression guard for the ``Path.stem`` double-suffix bug: a file named +``finance.corpus.json`` must register as ``corpus:finance`` (the key that +``build_schema_issues`` hardcodes), never ``corpus:finance.corpus``. +""" + +from __future__ import annotations + +import json + +import pytest + +from script.analysis.analyze_dataset_corpus_alignment import ( + DEFAULT_LEGACY_DIR, + discover_corpora, +) + + +@pytest.fixture +def legacy_dir(tmp_path, monkeypatch): + (tmp_path / "finance.corpus.json").write_text( + json.dumps({"category1": "definition one"}), encoding="utf-8" + ) + (tmp_path / "shougang.corpus.json").write_text( + json.dumps({"category2": "definition two"}), encoding="utf-8" + ) + monkeypatch.setattr( + "script.analysis.analyze_dataset_corpus_alignment.DEFAULT_LEGACY_DIR", + tmp_path, + ) + return tmp_path + + +def test_legacy_corpus_keys_use_dataset_name_not_double_suffix(legacy_dir): + corpora = discover_corpora() + assert "corpus:finance" in corpora + assert "corpus:shougang" in corpora + # the Path.stem double-suffix ("finance.corpus") must never be registered + assert "corpus:finance.corpus" not in corpora + assert "corpus:shougang.corpus" not in corpora + for key in corpora: + assert not key.endswith(".corpus"), f"bad legacy key {key!r}" + + +def test_legacy_corpus_stem_matches_hardcoded_consumer_key(legacy_dir): + corpora = discover_corpora() + # build_schema_issues() looks up standard_stats["corpus:finance"] + assert "corpus:finance" in corpora diff --git a/tests/rl/test_rl_canonical_e2e.py b/tests/rl/test_rl_canonical_e2e.py index 4d9906c..446b9d6 100644 --- a/tests/rl/test_rl_canonical_e2e.py +++ b/tests/rl/test_rl_canonical_e2e.py @@ -1,7 +1,7 @@ """End-to-end stage-4A validation on real data (skipped in CI without data/). Exercises the full RL data path on the four production datasets: -data//canonical/all.json -> resolved-only export -> VeRL v0.8.0 RL +data/canonical//all.json -> resolved-only export -> VeRL v0.8.0 RL parquet, asserting registry consistency, candidate-fidelity, determinism, VeRL-compatible column shape and validate_rl_dataset acceptance. """ @@ -24,6 +24,8 @@ ROOT = Path(__file__).resolve().parents[2] DATA = ROOT / "data" +CANONICAL = DATA / "canonical" +PROCESSED = DATA / "processed" REGISTRY_DIR = ROOT / "cfg" / "task" / "registry" CORPUS_DIR = ROOT / "cfg" / "task" / "corpus" DATASETS = ("finance", "infra", "pers_info", "shougang") @@ -34,12 +36,12 @@ def _trainable_resolved(dataset: str) -> int: - with (DATA / dataset / "canonical" / "all.json").open(encoding="utf-8") as handle: + with (CANONICAL / dataset / "all.json").open(encoding="utf-8") as handle: canonical = json.load(handle) by_id = {str(record.get("id")): record for record in canonical} split_ids: set[str] = set() for split in ("train", "val", "test"): - with (DATA / dataset / f"{split}.json").open(encoding="utf-8") as handle: + with (PROCESSED / dataset / f"{split}.json").open(encoding="utf-8") as handle: for record in json.load(handle): split_ids.add(str(record.get("id"))) return sum( @@ -51,7 +53,7 @@ def _trainable_resolved(dataset: str) -> int: @pytest.fixture(scope="module") def real_data_available() -> bool: - return all((DATA / dataset / "canonical" / "all.json").is_file() for dataset in DATASETS) + return all((CANONICAL / dataset / "all.json").is_file() for dataset in DATASETS) @pytest.fixture(scope="module") @@ -70,8 +72,8 @@ def exports(tmp_path_factory, real_data_available: bool): config = TaskConfig.from_mapping({"metadata_fields": list(METADATA_FIELDS)}) out = out_root / dataset report = export_rl_dataset( - DATA / dataset / "canonical" / "all.json", - DATA / dataset, + CANONICAL / dataset / "all.json", + PROCESSED / dataset, out, dataset, registry_path, @@ -102,7 +104,7 @@ def test_resolved_counts_match_canonical_pipeline(exports: dict) -> None: def test_canonical_resolved_counts_are_stable(exports: dict) -> None: for dataset, expected in EXPECTED_CANONICAL_RESOLVED.items(): - with (DATA / dataset / "canonical" / "all.json").open(encoding="utf-8") as handle: + with (CANONICAL / dataset / "all.json").open(encoding="utf-8") as handle: canonical = json.load(handle) resolved = sum( 1 for record in canonical if record.get("resolution_status") == "resolved" @@ -142,9 +144,9 @@ def test_no_assistant_gold_and_no_algorithm_fields(exports: dict) -> None: def test_no_unresolved_samples_in_exports(exports: dict) -> None: for dataset, bundle in exports.items(): for split, details in bundle["report"]["splits"].items(): - with (DATA / dataset / f"{split}.json").open(encoding="utf-8") as handle: + with (PROCESSED / dataset / f"{split}.json").open(encoding="utf-8") as handle: split_records = json.load(handle) - with (DATA / dataset / "canonical" / "all.json").open(encoding="utf-8") as handle: + with (CANONICAL / dataset / "all.json").open(encoding="utf-8") as handle: canonical = json.load(handle) canonical_by_id = {str(record.get("id")): record for record in canonical} unresolved_in_split = sum( @@ -203,8 +205,8 @@ def test_export_is_deterministic(exports: dict, tmp_path: Path) -> None: for dataset, bundle in exports.items(): out2 = tmp_path / f"{dataset}-again" report2 = export_rl_dataset( - DATA / dataset / "canonical" / "all.json", - DATA / dataset, + CANONICAL / dataset / "all.json", + PROCESSED / dataset, out2, dataset, bundle["registry_path"], diff --git a/tests/sft/test_sft_canonical_e2e.py b/tests/sft/test_sft_canonical_e2e.py index 069bc75..d78f33e 100644 --- a/tests/sft/test_sft_canonical_e2e.py +++ b/tests/sft/test_sft_canonical_e2e.py @@ -1,6 +1,6 @@ """End-to-end stage-3C validation on real data (skipped in CI without data/). -Exercises the full canonical pipeline: data//canonical/all.json -> +Exercises the full canonical pipeline: data/canonical//all.json -> resolved-only export -> stage1/stage2 parquet, asserting registry consistency, universe completeness, determinism and VeRL-readable output. """ @@ -19,6 +19,8 @@ ROOT = Path(__file__).resolve().parents[2] DATA = ROOT / "data" +CANONICAL = DATA / "canonical" +PROCESSED = DATA / "processed" REGISTRY_DIR = ROOT / "cfg" / "task" / "registry" CORPUS_DIR = ROOT / "cfg" / "task" / "corpus" DATASETS = ("finance", "infra", "pers_info", "shougang") @@ -31,12 +33,12 @@ def _trainable_resolved(dataset: str) -> int: """Resolved canonical records whose id lies inside the original split boundaries (computed from the real data, never hard-coded).""" - with (DATA / dataset / "canonical" / "all.json").open(encoding="utf-8") as handle: + with (CANONICAL / dataset / "all.json").open(encoding="utf-8") as handle: canonical = json.load(handle) by_id = {str(record.get("id")): record for record in canonical} split_ids: set[str] = set() for split in ("train", "val", "test"): - with (DATA / dataset / f"{split}.json").open(encoding="utf-8") as handle: + with (PROCESSED / dataset / f"{split}.json").open(encoding="utf-8") as handle: for record in json.load(handle): split_ids.add(str(record.get("id"))) return sum( @@ -48,7 +50,7 @@ def _trainable_resolved(dataset: str) -> int: @pytest.fixture(scope="module") def real_data_available() -> bool: - return all((DATA / dataset / "canonical" / "all.json").is_file() for dataset in DATASETS) + return all((CANONICAL / dataset / "all.json").is_file() for dataset in DATASETS) @pytest.fixture(scope="module") @@ -68,8 +70,8 @@ def exports(tmp_path_factory, real_data_available: bool): config = TaskConfig.from_mapping({"metadata_fields": list(METADATA_FIELDS)}) out = out_root / dataset report = export_sft_dataset( - DATA / dataset / "canonical" / "all.json", - DATA / dataset, + CANONICAL / dataset / "all.json", + PROCESSED / dataset, out, registry_path, config, @@ -101,7 +103,7 @@ def test_canonical_resolved_counts_are_stable(exports: dict) -> None: """Stage-3B canonical resolved counts hold; records outside the split boundaries are reported (data-pipeline fact), not silently added.""" for dataset, expected in EXPECTED_CANONICAL_RESOLVED.items(): - with (DATA / dataset / "canonical" / "all.json").open(encoding="utf-8") as handle: + with (CANONICAL / dataset / "all.json").open(encoding="utf-8") as handle: canonical = json.load(handle) resolved = sum( 1 for record in canonical if record.get("resolution_status") == "resolved" @@ -133,10 +135,10 @@ def test_no_unresolved_samples_in_exports(exports: dict) -> None: report = bundle["report"] # split records that are not resolved are skipped, never exported for split, details in report["splits"].items(): - with (DATA / dataset / f"{split}.json").open(encoding="utf-8") as handle: + with (PROCESSED / dataset / f"{split}.json").open(encoding="utf-8") as handle: split_records = json.load(handle) canonical_by_id = {} - with (DATA / dataset / "canonical" / "all.json").open(encoding="utf-8") as handle: + with (CANONICAL / dataset / "all.json").open(encoding="utf-8") as handle: for record in json.load(handle): canonical_by_id[str(record.get("id"))] = record unresolved_in_split = sum( @@ -244,8 +246,8 @@ def test_export_is_deterministic(exports: dict, tmp_path: Path) -> None: for dataset, bundle in exports.items(): out2 = tmp_path / f"{dataset}-again" report2 = export_sft_dataset( - DATA / dataset / "canonical" / "all.json", - DATA / dataset, + CANONICAL / dataset / "all.json", + PROCESSED / dataset, out2, bundle["registry_path"], bundle["config"], diff --git a/tests/task/test_canonical_dataset.py b/tests/task/test_canonical_dataset.py index 24759a4..c71edf2 100644 --- a/tests/task/test_canonical_dataset.py +++ b/tests/task/test_canonical_dataset.py @@ -1,7 +1,7 @@ """Stage-3B tests: canonical dataset resolution against the LeafRegistry. Fixture tests run on checked-in standards_map files / inline data (CI-safe). -Real-data integration assertions run only when data//all.json is +Real-data integration assertions run only when data/processed//all.json is available locally and exercise the full build_canonical_dataset pipeline. """ @@ -233,8 +233,8 @@ def test_pers_info_without_description_still_resolves(pers_registry: LeafRegistr def _write_tmp_dataset(tmp_path: Path, dataset: str, records: list[dict]) -> None: - data_dir = tmp_path / "data" - out = data_dir / dataset / "all.json" + data_root = tmp_path / "data" + out = data_root / "processed" / dataset / "all.json" out.parent.mkdir(parents=True, exist_ok=True) with out.open("w", encoding="utf-8") as handle: json.dump(records, handle, ensure_ascii=False) @@ -260,7 +260,8 @@ def test_pipeline_preserves_classification_and_keeps_unresolved( _write_tmp_dataset(tmp_path, "finance", records) result = build_canonical_dataset( "finance", - data_dir=tmp_path / "data", + processed_dir=tmp_path / "data" / "processed", + canonical_dir=tmp_path / "data" / "canonical", registry_dir=REGISTRY_DIR, corpus_dir=CORPUS_DIR, overwrite=False, @@ -274,7 +275,7 @@ def test_pipeline_preserves_classification_and_keeps_unresolved( } assert result.resolved_targets_in_registry is True - with (tmp_path / "data" / "finance" / "canonical" / "all.json").open(encoding="utf-8") as handle: + with (tmp_path / "data" / "canonical" / "finance" / "all.json").open(encoding="utf-8") as handle: canonical = json.load(handle) by_id = {record["id"]: record for record in canonical} # classification is provenance, never rewritten (whitespace preserved) @@ -299,7 +300,8 @@ def _run_cli(tmp_path: Path, *extra: str) -> int: return targets_cli.main( [ - "--data-dir", str(tmp_path / "data"), + "--processed-dir", str(tmp_path / "data" / "processed"), + "--canonical-dir", str(tmp_path / "data" / "canonical"), "--registry-dir", str(REGISTRY_DIR), "--corpus-dir", str(CORPUS_DIR), *extra, @@ -316,8 +318,8 @@ def test_cli_first_run_succeeds_and_second_fails(tmp_path: Path) -> None: _write_tmp_dataset(tmp_path, "finance", records) # first run without --overwrite on a fresh out-dir succeeds assert _run_cli(tmp_path, "--dataset", "finance") == 0 - assert (tmp_path / "data" / "finance" / "canonical" / "all.json").is_file() - assert (tmp_path / "data" / "finance" / "canonical" / "resolution_report.json").is_file() + assert (tmp_path / "data" / "canonical" / "finance" / "all.json").is_file() + assert (tmp_path / "data" / "canonical" / "finance" / "resolution_report.json").is_file() # second run without --overwrite fails fast before writing anything with pytest.raises(FileExistsError, match="refusing to overwrite"): _run_cli(tmp_path, "--dataset", "finance") @@ -330,11 +332,11 @@ def test_cli_is_deterministic_across_two_runs(tmp_path: Path) -> None: ] _write_tmp_dataset(tmp_path, "finance", records) assert _run_cli(tmp_path, "--dataset", "finance") == 0 - first_all = (tmp_path / "data" / "finance" / "canonical" / "all.json").read_bytes() - first_report = (tmp_path / "data" / "finance" / "canonical" / "resolution_report.json").read_bytes() + first_all = (tmp_path / "data" / "canonical" / "finance" / "all.json").read_bytes() + first_report = (tmp_path / "data" / "canonical" / "finance" / "resolution_report.json").read_bytes() assert _run_cli(tmp_path, "--dataset", "finance", "--overwrite") == 0 - second_all = (tmp_path / "data" / "finance" / "canonical" / "all.json").read_bytes() - second_report = (tmp_path / "data" / "finance" / "canonical" / "resolution_report.json").read_bytes() + second_all = (tmp_path / "data" / "canonical" / "finance" / "all.json").read_bytes() + second_report = (tmp_path / "data" / "canonical" / "finance" / "resolution_report.json").read_bytes() assert first_all == second_all assert first_report == second_report @@ -376,7 +378,8 @@ def test_code_unresolved_pipeline_no_crash(tmp_path: Path) -> None: ) result = build_canonical_dataset( "shougang", - data_dir=tmp_path / "data", + processed_dir=tmp_path / "data" / "processed", + canonical_dir=tmp_path / "data" / "canonical", registry_dir=REGISTRY_DIR, # real shared guanji registry (233) corpus_dir=tmp_path / "corpus", ) @@ -385,7 +388,7 @@ def test_code_unresolved_pipeline_no_crash(tmp_path: Path) -> None: "count": 1, "by_leaf": {"科研进程管控": 1}, } - with (tmp_path / "data" / "shougang" / "canonical" / "all.json").open(encoding="utf-8") as handle: + with (tmp_path / "data" / "canonical" / "shougang" / "all.json").open(encoding="utf-8") as handle: canonical = json.load(handle) by_id = {record["id"]: record for record in canonical} assert by_id["has-code"]["resolution_status"] == "resolved" @@ -409,21 +412,22 @@ def test_cross_dataset_fail_fast_no_partial_writes(tmp_path: Path) -> None: with pytest.raises(FileNotFoundError, match="input dataset not found"): targets_cli.main( [ - "--data-dir", str(tmp_path / "data"), + "--processed-dir", str(tmp_path / "data" / "processed"), + "--canonical-dir", str(tmp_path / "data" / "canonical"), "--registry-dir", str(REGISTRY_DIR), "--corpus-dir", str(CORPUS_DIR), "--datasets", "finance", "shougang", ] ) - assert not (tmp_path / "data" / "finance" / "canonical").exists() - assert not (tmp_path / "data" / "shougang" / "canonical").exists() + assert not (tmp_path / "data" / "canonical" / "finance").exists() + assert not (tmp_path / "data" / "canonical" / "shougang").exists() def test_invalid_record_non_mapping_is_audited(tmp_path: Path) -> None: """Non-object records (bare strings in the JSON array) resolve to invalid_record instead of raising AttributeError, and stay auditable.""" - data_dir = tmp_path / "data" - out = data_dir / "finance" / "all.json" + data_root = tmp_path / "data" + out = data_root / "processed" / "finance" / "all.json" out.parent.mkdir(parents=True, exist_ok=True) with out.open("w", encoding="utf-8") as handle: json.dump( @@ -438,12 +442,12 @@ def test_invalid_record_non_mapping_is_audited(tmp_path: Path) -> None: ) result = build_canonical_dataset( "finance", - data_dir=data_dir, + processed_dir=data_root / "processed", canonical_dir=data_root / "canonical", registry_dir=REGISTRY_DIR, corpus_dir=CORPUS_DIR, ) assert result.status_counts == {"invalid_record": 3, "resolved": 1} - with (data_dir / "finance" / "canonical" / "all.json").open(encoding="utf-8") as handle: + with (data_root / "canonical" / "finance" / "all.json").open(encoding="utf-8") as handle: canonical = json.load(handle) invalid = [r for r in canonical if r["resolution_status"] == "invalid_record"] assert len(invalid) == 3 @@ -461,14 +465,14 @@ def test_invalid_record_non_mapping_is_audited(tmp_path: Path) -> None: @pytest.fixture(scope="module") def real_data_dir() -> Path | None: path = ROOT / "data" - if not (path / "finance" / "all.json").is_file(): + if not (path / "processed" / "finance" / "all.json").is_file(): return None return path def _copy_real_input(tmp_path: Path, dataset: str, real_data: Path) -> Path: - src = real_data / dataset / "all.json" - dst = tmp_path / "data" / dataset / "all.json" + src = real_data / "processed" / dataset / "all.json" + dst = tmp_path / "data" / "processed" / dataset / "all.json" dst.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(src, dst) return dst @@ -479,7 +483,8 @@ def test_real_finance_counts(tmp_path: Path, real_data_dir: Path | None) -> None pytest.skip("real data/ not available") _copy_real_input(tmp_path, "finance", real_data_dir) result = build_canonical_dataset( - "finance", data_dir=tmp_path / "data", + "finance", processed_dir=tmp_path / "data" / "processed", + canonical_dir=tmp_path / "data" / "canonical", registry_dir=REGISTRY_DIR, corpus_dir=CORPUS_DIR, ) assert result.input_records == 568 @@ -507,7 +512,8 @@ def test_real_infra_counts(tmp_path: Path, real_data_dir: Path | None) -> None: pytest.skip("real data/ not available") _copy_real_input(tmp_path, "infra", real_data_dir) result = build_canonical_dataset( - "infra", data_dir=tmp_path / "data", + "infra", processed_dir=tmp_path / "data" / "processed", + canonical_dir=tmp_path / "data" / "canonical", registry_dir=REGISTRY_DIR, corpus_dir=CORPUS_DIR, ) assert result.status_counts == {"resolved": 64} @@ -519,7 +525,8 @@ def test_real_pers_info_counts(tmp_path: Path, real_data_dir: Path | None) -> No pytest.skip("real data/ not available") _copy_real_input(tmp_path, "pers_info", real_data_dir) result = build_canonical_dataset( - "pers_info", data_dir=tmp_path / "data", + "pers_info", processed_dir=tmp_path / "data" / "processed", + canonical_dir=tmp_path / "data" / "canonical", registry_dir=REGISTRY_DIR, corpus_dir=CORPUS_DIR, ) assert result.status_counts == {"resolved": 176} @@ -531,7 +538,8 @@ def test_real_shougang_counts(tmp_path: Path, real_data_dir: Path | None) -> Non pytest.skip("real data/ not available") _copy_real_input(tmp_path, "shougang", real_data_dir) result = build_canonical_dataset( - "shougang", data_dir=tmp_path / "data", + "shougang", processed_dir=tmp_path / "data" / "processed", + canonical_dir=tmp_path / "data" / "canonical", registry_dir=REGISTRY_DIR, corpus_dir=CORPUS_DIR, ) assert result.status_counts == { diff --git a/tests/task/test_label_whitespace_variants.py b/tests/task/test_label_whitespace_variants.py index 2e73a73..376d879 100644 --- a/tests/task/test_label_whitespace_variants.py +++ b/tests/task/test_label_whitespace_variants.py @@ -4,13 +4,13 @@ removing all whitespace produces the same key but the raw strings differ (e.g. ``"经营 管理"`` vs ``"经营管理"``) and REPORTS them. It never auto-repairs labels — label fixes are manual upstream edits in -``data//all.json`` (the formal input), and this test exists so a +``data/processed//all.json`` (the formal input), and this test exists so a re-introduced variant is caught before downstream artifacts are regenerated. Two tiers, matching the repo's CI convention: - logic tier: runs always on inline data (CI-safe), -- data tier: skips when ``data//all.json`` is unavailable locally. +- data tier: skips when ``data/processed//all.json`` is unavailable locally. """ from __future__ import annotations @@ -97,7 +97,7 @@ def test_find_whitespace_variants_clean_labels_report_nothing() -> None: @pytest.mark.parametrize("dataset", DATASETS) def test_real_dataset_has_no_whitespace_variant_labels(dataset: str) -> None: - record_path = DATA_DIR / dataset / "all.json" + record_path = DATA_DIR / "processed" / dataset / "all.json" if not record_path.is_file(): pytest.skip(f"data/{dataset}/all.json not available")