Compare commits
25 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 张宗平
|
ccdd43081d | ||
|
张宗平
|
150e3bd9fd | ||
|
张宗平
|
8f644a7470 | ||
|
张宗平
|
cd6209119a | ||
|
张宗平
|
17ed7947d3 | ||
|
张宗平
|
77c2ada9b7 | ||
|
张宗平
|
5e3030a20e | ||
|
张宗平
|
1a096ff989 | ||
|
张宗平
|
9b9aa35f13 | ||
|
张宗平
|
40ec960cf8 | ||
|
张宗平
|
8aeddebce9 | ||
|
张宗平
|
76fd636fae | ||
|
张宗平
|
aecccfab1d | ||
|
张宗平
|
0723b257fd | ||
|
张宗平
|
4c4a8f6435 | ||
|
张宗平
|
357e2e60bd | ||
|
张宗平
|
8286b8276c | ||
|
张宗平
|
bf95ac18e4 | ||
|
张宗平
|
204d5237ba | ||
|
张宗平
|
133bbac316 | ||
|
张宗平
|
7e1df76e3e | ||
|
张宗平
|
aec8fd5ef1 | ||
|
张宗平
|
f509d47878 | ||
|
张宗平
|
f20c2db803 | ||
|
张宗平
|
545802d62c |
@@ -126,3 +126,8 @@ coverage/
|
||||
.context-*/
|
||||
tmp/
|
||||
.temp/
|
||||
|
||||
# local artifacts (design §6.1)
|
||||
/checkpoints/
|
||||
/data/*.jsonl
|
||||
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
# LLM backbone config (T2.3 / T2.4)
|
||||
# Design ref: docs/superpowers/specs/2026-06-29-ts-as-modality-design.md §2.2
|
||||
# Qwen2.5-0.5B-Instruct: hidden=896, 24 layers, vocab=151936, BF16 ~1GB
|
||||
|
||||
# Local offline path (downloaded via proxy; HF hub unreachable on this host)
|
||||
model_path: /home/zhangzp/models/Qwen2.5-0.5B-Instruct
|
||||
hf_model_id: Qwen/Qwen2.5-0.5B-Instruct # for reference; offline use model_path
|
||||
|
||||
hidden_size: 896
|
||||
torch_dtype: bfloat16
|
||||
@@ -0,0 +1,20 @@
|
||||
# TS Encoder + Projector hyperparameters (T2.1 / T2.2)
|
||||
# Design ref: docs/superpowers/specs/2026-06-29-ts-as-modality-design.md §2.2
|
||||
|
||||
# Patchify (channel-independent)
|
||||
patch_len: 8 # P
|
||||
stride: 4 # S (overlap); n_patches = (T - P) // S + 1
|
||||
|
||||
# TS Encoder (2-layer Transformer)
|
||||
d: 256 # hidden dim
|
||||
layers: 2
|
||||
heads: 4
|
||||
ffn_mult: 4 # dim_feedforward = d * ffn_mult
|
||||
dropout: 0.1
|
||||
|
||||
# Projector (T2.2) — maps TS token d → LLM hidden
|
||||
llm_hidden: 896 # Qwen2.5-0.5B hidden size
|
||||
|
||||
# Context
|
||||
T: 512 # default window length → 127 TS tokens
|
||||
max_patches: 2048 # positional-embedding budget
|
||||
@@ -0,0 +1,140 @@
|
||||
# 诊断报告 · ts-as-modality 跨域泛化失败与方案封存
|
||||
|
||||
> **日期**：2026-07-02
|
||||
> **状态**：方案封存（资产保留，不再投入）
|
||||
> **决定性问题**：合成域有效的「TS-as-modality」方案，能否泛化到真实时序？
|
||||
|
||||
---
|
||||
|
||||
## TL;DR
|
||||
|
||||
方案在合成域内有效，但在真实时序上**跨实体泛化失败**——同一 SMD 数据集、同样 38 通道、仅换了机器（machine-1 → machine-2），模型输出从合法 JSON 崩溃为 repetition garbage。证据链完整指向：TS encoder 学到的是数据集/机器特有模式，而非可泛化的时序表征。方案作为研究性 PoC 划定了清晰的能力边界，但当前架构无法支撑「通用时序理解」这一目标，故封存。
|
||||
|
||||
---
|
||||
|
||||
## 完整诊断链
|
||||
|
||||
### 阶段 1：合成域内验证（成立）
|
||||
|
||||
在全量合成数据（50 万对齐 + 10 万 SFT）训练后：
|
||||
|
||||
| 指标 | model | pure_llm | 结论 |
|
||||
|---|---|---|---|
|
||||
| 时序消融 change_rate | 0.98/50 | — | TS token 确实被消费 ✅ |
|
||||
| QA 均分（6 类） | 2.41 | 0.90 (2.7×) | TS 模态是必要非装饰 ✅ |
|
||||
| anomaly JSON 解析率 | 0.98 | 0.01 | 格式学习成功 ✅ |
|
||||
| VUS-PR | 0.22 | 0.24 | ⚠️ 输给 trivial baseline |
|
||||
|
||||
**阶段 1 结论**：机制可行，QA 能力真实，但异常精准定位弱（VUS-PR 输 trivial）。
|
||||
|
||||
### 阶段 2：zero-shot 跨域（崩溃）
|
||||
|
||||
用未微调的合成 ckpt 直接评测真实 SMD（machine-1/2/2-1 合并，prevalence 5.8% vs 合成 26%）：
|
||||
|
||||
| 指标 | model（real） | 解读 |
|
||||
|---|---|---|
|
||||
| parse_rate | **0.00** | 没有任何输出可解析 |
|
||||
| Aff-F1 | **0.00** | 完全无定位能力 |
|
||||
| QA mean | **0.00** | vs synth 的 2.41 |
|
||||
| VUS-PR | 0.2263 | = all_report，是**假象** |
|
||||
|
||||
**输出样本**（repetition collapse，LLM 在 OOD 输入下的典型崩溃）：
|
||||
```
|
||||
'ERYoticaisonaisonanxietyaisonaisonanaisonanaison...'
|
||||
'horizon: horizon: horizon: horizon: horizon:...'
|
||||
```
|
||||
|
||||
**VUS-PR=0.2263 的真相**：模型输出全是 garbage → `parse_anomaly_segments` 失败 → `answer_to_point_scores` 返回全 0 → `vus_pr(全0, y)` 退化为 prevalence（=0.226）。所以「VUS-PR 与 trivial 持平」是假象，模型实际**什么都没做**，只是被 prevalence 下限兜住了。这是一个**评测陷阱**：低质量输出在低 prevalence 下会被误读为「不算太差」。
|
||||
|
||||
**干扰变量排除**：改写 instruction（短措辞 / 通用模板）后输出仍是 garbage，排除「指令过长/格式不兼容」因素，确认是 TS embedding 的分布偏移所致：
|
||||
- 通道数：5（合成）→ 38（SMD）
|
||||
- 数值分布：合成 mean~46/std~23 vs SMD z-score 后 mean~0/std~0.6
|
||||
|
||||
### 阶段 3：SMD 微调（训练域内有效）
|
||||
|
||||
从 stage2_final 继续训 LoRA on SMD train（machine-1-1/1-2，141 样本，500 步，EMA 1.27→1.00）：
|
||||
|
||||
**训练域内输出**（machine-1-1/1-2，微调后）：
|
||||
```
|
||||
{"anomaly_regions": [{"start": 309, "end": 412}], "n_regions": 1} ✅
|
||||
{"anomaly_regions": [{"start": 0, "end": 136}], "n_regions": 1} ✅
|
||||
{"anomaly_regions": [{"start": 103, "end": 126}], "n_regions": 1} ✅
|
||||
```
|
||||
4/5 产出合法 JSON，格式与语义均正确。**机制在训练域内正常工作**。
|
||||
|
||||
### 阶段 4：跨实体 held-out（崩溃，决定性）
|
||||
|
||||
用同一个微调后模型评测 SMD held-out（machine-2-1，同数据集/同 38 通道/同样 z-score 分布，仅换机器）：
|
||||
|
||||
```
|
||||
[anomaly] '{"}</seo_http>\xa0\n\xa0\n\xa0\n\n\xa0\xa0 (since)...' ❌
|
||||
[anomaly] '{"id": 1, "name": "nan</sk_inch_regions> \n\n\n...' ❌
|
||||
[describe] '------------{"}</ewanewanewanewanewanewanewan...' ❌
|
||||
```
|
||||
|
||||
**全 garbage**。从训练域内（machine-1）到 held-out（machine-2），输入分布的偏移仅仅是「换了台服务器」，模型就从「会」变成「完全不会」。
|
||||
|
||||
---
|
||||
|
||||
## 根因分析
|
||||
|
||||
**不是评测管道 bug、不是数据量不足、不是指令格式问题、不是通道数不匹配（held-out 也是 38 通道）。**
|
||||
|
||||
根因是 **TS encoder 学到的是数据集/机器特有的统计特征，而非可泛化的时序表征**：
|
||||
|
||||
1. **架构容量不足**：TS encoder 仅 2 层 TF + patch 线性嵌入，~2.1M 参数。这个容量足以「记忆」machine-1 的分布特征，但不足以学到「时序形态→语义」的通用映射。
|
||||
2. **训练分布单一**：仅在合成数据上训练 encoder/projector，合成分布的特征（固定 5 通道、特定数值范围、模板化异常形态）被编码进了表征。
|
||||
3. **泛化瓶颈在 encoder 侧**：LoRA 只调 LLM，但 LLM 侧（语言能力）本来就没问题——崩溃发生在 encoder 输出的 TS embedding 严重偏离 LLM 能理解的流形时。
|
||||
4. **微调样本不足**：141 样本对 encoder 2.1M 参数，学到的更像「记忆 machine-1」而非「适应 SMD 通用分布」。
|
||||
|
||||
这是一个**架构性**而非工程性短板：轻量 TS encoder + 单分布训练的组合，决定了它无法跨实体泛化。
|
||||
|
||||
---
|
||||
|
||||
## 方案意义与边界（诚实评估）
|
||||
|
||||
### 成立的部分 ✅
|
||||
- **TS 模态融合机制可行**：证明 soft token 能承载时序信息并被 LLM 消费（非被忽略的噪声）。
|
||||
- **合成域内 QA 能力真实**：2.7× 于 pure_llm，TS 模态对时序问答有实质增益。
|
||||
- **工程管线完整可复现**：合成数据、训练、断点续训、评测全链路跑通，含真实数据接入。
|
||||
|
||||
### 未立住的部分 ❌
|
||||
- **跨实体泛化失败**：同数据集换机器就崩溃，不能称为「理解时序」。
|
||||
- **作为异常检测器**：合成域 VUS-PR 输 trivial，真实域 zero-shot 完全失效。
|
||||
- **zero-shot 跨域能力**：不存在。
|
||||
|
||||
### 方案定位
|
||||
作为「轻量 TS-modality 融合」的**研究性 PoC**，其核心价值在于**清晰、诚实地划定了能力边界**：合成域内可行，但当前架构无法泛化到真实时序。这是一个**有价值的负结果**——它明确排除了「0.5B LLM + 2 层 TS encoder + 单分布合成训练」这条路线做通用时序理解的可能性，优于任何「看起来 work 但未测跨域」的结果。
|
||||
|
||||
---
|
||||
|
||||
## 资产清单（封存保留）
|
||||
|
||||
**代码**（git-tracked）：
|
||||
- `src/tsmm/`：data/model/train/eval 完整模块
|
||||
- `scripts/`：gen/train/eval 监督脚本（含流式+断点续训）
|
||||
- `tests/`：158 单测全绿
|
||||
|
||||
**数据/模型**（git-ignore，本地保留）：
|
||||
- `data/align.jsonl`（50万）、`sft.jsonl`（10万）、`eval_synth.jsonl`（2k held-out）
|
||||
- `data/real/smd/`（SMD 3 机合并 npy）、`data/eval_real.jsonl`（machine-2-1 held-out）、`data/smd_ft/sft.jsonl`（machine-1-1/1-2 微调集）
|
||||
- `checkpoints/stage1/stage1_final.pt`（31250 步）、`checkpoints/stage2/`（9375 步 + lora）、`checkpoints/stage2_smd/`（SMD 微调 500 步）
|
||||
- `data/small_scale_backup/`（小规模对比备份）
|
||||
|
||||
**报告**（git-tracked，`reports/`）：
|
||||
- `eval-2026-06-30.md`：小规模首评（40/类）
|
||||
- `eval-2026-07-01.md`：全量重训评测（synth 全量 2000）
|
||||
- `eval-2026-07-02.md`：真实数据评测（synth + real，zero-shot 崩溃记录）
|
||||
- `instruct_check.json`：T4.2 指令遵循抽检
|
||||
|
||||
---
|
||||
|
||||
## 重启条件
|
||||
|
||||
若未来重启本方案，需满足以下任一前提（按预期 ROI 排序）：
|
||||
|
||||
1. **更大 encoder + 真实数据从头训练**：encoder >10M 参数，且用真实多数据集从头训练（非仅微调），让 encoder 有足够容量和分布覆盖学到通用表征。
|
||||
2. **channel-agnostic patch 策略**：消除「通道数」这一硬约束（当前 patch embed 固定 5 通道，是跨域的物理障碍），如 per-channel 独立编码后聚合。
|
||||
3. **多真实数据集混合训练验证泛化**：在 SMD/SWaT/PSM 等多个真实数据集上联合训练，并在完全 held-out 的数据集上验证——只有这种泛化成立，方案才算「通用」。
|
||||
|
||||
在以上任一条件未满足前，本方案不再投入。
|
||||
@@ -1,31 +1,35 @@
|
||||
{
|
||||
"change": "ts-as-modality",
|
||||
"phase": "build",
|
||||
"checkpoint": "M1-complete",
|
||||
"date": "2026-06-29",
|
||||
"checkpoint": "M2-complete + M3-code-complete",
|
||||
"date": "2026-06-30",
|
||||
"status": "paused-awaiting-user-decision",
|
||||
"summary": "M1 (data pipeline) fully complete (T1.1-T1.7). 51 tests passing. All committed on feature/20260629/ts-as-modality.",
|
||||
"next_action": "T2.1 (TS Encoder, model/ts_encoder.py). Blocked on environment setup decision for M2-M5.",
|
||||
"summary": "All M2 + M3 code written and smoke-verified. 100 tests passing. Remaining work (T3.3 onward) requires multi-hour GPU training of stage 1 (500k align samples), then M4 SFT (100k), then M5 eval.",
|
||||
"next_action": "T3.3 TS-token effectiveness check — REQUIRES a trained stage 1 checkpoint (compute-heavy).",
|
||||
"blocker": {
|
||||
"reason": "venv (.venv) has only M1 deps (numpy/tqdm/pyyaml/pytest). M2+ requires torch + transformers (+peft/accelerate/datasets for M3/M4). System torch is CPU-only; design assumes 3060 12GB GPU but actual is Quadro RTX 3000 6GB.",
|
||||
"decision_pending": "User must choose: A) CPU torch now (M2 correctness only, no GPU mem validation); B) CUDA torch for 6GB Quadro (enables real training, expect OOM-fallbacks earlier vs 12GB plan); C) user sets up env themselves.",
|
||||
"recommendation": "B"
|
||||
"reason": "T3.3+ need actual model training, not just code. Stage 1 design = 500k align.jsonl samples, ~10k steps. At ~3-5 it/s on RTX 3060 this is hours. Then M4 stage 2 (LoRA SFT, similar scale), then M5 full eval pipeline.",
|
||||
"decision_pending": "User must decide training scope: (A) full design-scale training (generate 500k+100k data, train to completion, hours of GPU); (B) reduced-scale training (e.g. 50k samples, 2k steps — enough to produce a checkpoint and validate T3.3 effectiveness signal, defers full-quality); (C) write the remaining M4/M5 CODE now (stage2.py, eval modules) and smoke-verify, defer ALL training to user; (D) pause build.",
|
||||
"recommendation": "C then B — finish all remaining code + smoke first (so everything is wired and verified), then do a reduced-scale stage1+stage2 training run to validate the end-to-end effectiveness signal, leaving full-scale training as a later user-initiated run."
|
||||
},
|
||||
"environment": {
|
||||
"venv": ".venv (CUDA torch 2.5.1+cu121, transformers 5.12.1, peft 0.19.1)",
|
||||
"llm": "/home/zhangzp/models/Qwen2.5-0.5B-Instruct (hidden=896, 494M, bf16)",
|
||||
"gpu": "RTX 3060 12GB (~11.6 GB free)",
|
||||
"proxy": "socks5h://10.66.66.4:1080"
|
||||
},
|
||||
"resume_instructions": {
|
||||
"verify_state": "Run: /bin/bash .claude/skills/comet/scripts/comet-state.sh check ts-as-modality build",
|
||||
"find_next_task": "grep -n '\\- \\[ \\]' openspec/changes/ts-as-modality/tasks.md | head -1 (expect T2.1)",
|
||||
"venv": "Use .venv/bin/python for all python commands",
|
||||
"find_next_task": "grep -n '\\- \\[ \\]' openspec/changes/ts-as-modality/tasks.md | head -1 (expect T3.3)",
|
||||
"venv": ".venv/bin/python",
|
||||
"pythonpath": "export PYTHONPATH=src (needed for scripts; tests use pyproject pythonpath)",
|
||||
"branch": "feature/20260629/ts-as-modality",
|
||||
"build_mode": "executing-plans",
|
||||
"tdd_mode": "tdd",
|
||||
"review_mode": "standard",
|
||||
"isolation": "branch"
|
||||
},
|
||||
"completed_milestones": ["M1"],
|
||||
"completed_tasks": ["1.1", "1.2", "1.3", "1.4", "1.5", "1.6", "1.7"],
|
||||
"test_count": 51,
|
||||
"git": {
|
||||
"m1_complete_commit": "bbc60f2",
|
||||
"base_ref": "7a5a1d033fb1ac20af25a17a571b2791694bd923"
|
||||
}
|
||||
"completed_milestones": ["M1", "M2"],
|
||||
"completed_tasks": ["1.1-1.7", "2.1-2.6", "3.1", "3.2"],
|
||||
"test_count": 100,
|
||||
"remaining_code_tasks": ["T4.1 stage2.py", "T5.1 parse_answer.py", "T5.2 ts_metrics.py", "T5.3 qa_judge.py", "T5.4 baselines.py", "T5.5 run_eval.sh"],
|
||||
"remaining_compute_tasks": ["T3.3 effectiveness check (needs trained ckpt)", "T3.4 M3 exit", "T4.2 SFT instruction-following check", "M4 exit", "M5 exit"]
|
||||
}
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
## Status
|
||||
|
||||
**已封存（2026-07-02）**。完整诊断与结论见 `docs/diagnosis-2026-07-02-cross-domain.md`。摘要：合成域内有效（QA 2.41 vs pure_llm 0.89，TS token 消费率 0.98），但真实时序**跨实体泛化失败**——同 SMD 数据集换机器即崩溃为 garbage 输出。TS encoder（2层TF/2.1M参数）学到数据集特有模式而非通用时序表征。方案作为研究性 PoC 划定了能力边界（合成域可行/跨域不可行），资产保留，重启条件见诊断报告。
|
||||
|
||||
## Why
|
||||
|
||||
当前 AIOps 场景下，多变量时序的异常检测与解释仍依赖专用模型，缺乏「以自然语言对时序提问、做根因/事件关联推理」的统一能力。基于 `MTSAD-技术深度调研报告.md`，沿用 ChatTS「时序作为新模态」路线，我们希望在**单张 3060 12GB 消费级显卡**上做一个可复现的实验性落地，验证「把多变量时序视为独立模态、与文本/事件模态联合注入小型 LLM」这一方法在算力受限条件下是否切实可行、且「时序模态」是否真带来可量化的增益。现在做是因为：0.5B 级开源 LLM 与 PEFT/LoRA 工具链已足够成熟，使该实验在消费级硬件上首次成为可能。
|
||||
|
||||
@@ -15,31 +15,67 @@
|
||||
|
||||
## 2. M2 · 模型可跑通
|
||||
|
||||
- [ ] 2.1 TS Encoder `model/ts_encoder.py`：`Patchify(patch=8, stride=4)` 通道独立切 patch 线性嵌入到 d=256；`TSEncoder(d=256, layers=2, heads=4)` 2 层 TF + 可学习位置编码；前向 `[B,T,C]`→`[B,n_patches,256]`（验证：输入 `[2,512,5]`→`[2,128,256]`，参数 ≈4M）
|
||||
- [ ] 2.2 Projector `model/projector.py`：`Linear(256→896)+LayerNorm`（验证：输出 `[B,128,896]` 对齐 Qwen2.5-0.5B hidden）
|
||||
- [ ] 2.3 多模态拼接 `model/multimodal.py`：Qwen tokenizer tokenize 文本部分，TS token 作 inputs_embeds 插入 `[属性][时间戳][TS tok][事件][问题]`，统一构造 inputs_embeds+attention_mask+labels（仅回答段非 -100）（验证：seq_len ≤1024，mask/labels 形状对）
|
||||
- [ ] 2.4 训练/推理封装 `model/wrapper.py`：`MultimodalTSModel` 组合 Encoder+Projector+LLM+LoRA 挂载开关，`forward` 返 loss、`generate` 返文本，支持 `freeze_llm`(阶段①)/`enable_lora(r=16)`(阶段②)（验证：单 batch forward 3060 不 OOM；generate 能出文本）
|
||||
- [ ] 2.5 Collator `data/collator.py`：JSONL→batch 张量，处理变长 C（padding+mask）与变长文本（padding+attention_mask）（验证：batch=4 张量形状一致无 NaN）
|
||||
- [ ] 2.6 M2 出口验证：单 batch 前向+反向在 3060 跑通，loss 有限且下降趋势，阶段①模式峰值显存 ~5GB
|
||||
- [x] 2.1 TS Encoder `model/ts_encoder.py`：`Patchify(patch=8, stride=4)` 通道独立切 patch 线性嵌入到 d=256；`TSEncoder(d=256, layers=2, heads=4)` 2 层 TF + 可学习位置编码；前向 `[B,T,C]`→`[B,n_patches,256]`（验证：输入 `[2,512,5]`→`[2,127,256]`；spec 澄清：n_patches=(T-P)/S+1=127 非 128；通道独立=每通道共享 patch 线性后对通道 mean-pool；参数 ≈2.1M 非 4M）
|
||||
- [x] 2.2 Projector `model/projector.py`：`Linear(256→896)+LayerNorm`（验证：输出 `[B,n_patches,896]` 对齐 Qwen2.5-0.5B hidden）
|
||||
- [x] 2.3 多模态拼接 `model/multimodal.py`：Qwen tokenizer tokenize 文本部分，TS token 作 inputs_embeds 插入 `[属性][时间戳][TS tok][事件][问题][回答][EOS]`，统一构造 inputs_embeds+attention_mask+labels（仅回答段非 -100）（验证：8 项单测过——seq_len ≤1024、mask 全 1、回答段 label 非 -100、TS token 计入序列、无 NaN、超长左截断）。spec 澄清：训练拼回答+EOS，用 `len(tokenizer)` 构建嵌入表以容纳 special tokens。
|
||||
- [x] 2.4 训练/推理封装 `model/wrapper.py`：`MultimodalTSModel` 组合 Encoder+Projector+LLM+LoRA 挂载开关，`forward` 返 loss、`generate` 返文本，支持 `freeze_llm`(阶段①)/`enable_lora(r=16)`(阶段②)（验证：端到端 forward+backward 跑通、grad 流入 Encoder/Projector、generate 出文本、阶段①峰值 1.87GB ≪5GB 设计预算）。spec 澄清：wrapper 既接收原始 batch（series+文本列表）走端到端，也兼容预拼接 inputs_embeds；encoder/projector fp32，输出投影到 LLM dtype(bf16)。
|
||||
- [x] 2.5 Collator `data/collator.py`：JSONL→wrapper 原始 batch（series 张量 + attributes/timestamps/events/question/answer 文本列表），处理变长 T（padding/trunc 到 max_T）与变长 C（padding 到 max_C）、NaN→fill；变长文本 padding/attention_mask 由 wrapper 逐样本 splice+max_seq 填充处理（验证：11 项单测过；端到端 Collator→Wrapper fwd+bwd+generate 跑通）
|
||||
- [x] 2.6 M2 出口验证：单 batch 前向+反向在 3060 跑通（8 步 loss 9.82→5.54 下降），阶段①峰值显存 2.34GB ≪5GB 预算 ✅
|
||||
|
||||
## 3. M3 · 阶段①对齐训练
|
||||
|
||||
- [ ] 3.1 损失函数 `train/losses.py`：`lm_loss`（仅回答段计 loss）、`contrastive_loss`（扰动时序→回答 embedding 变化，InfoNCE/扰动一致性），总 loss=`lm_loss+λ*contrastive_loss`（λ 默认 0.1）（验证：loss 标量可反传，对比损失对扰动敏感）
|
||||
- [ ] 3.2 阶段①训练脚本 `train/stage1.py`+`scripts/train_stage1.sh`：加载 `align.jsonl` 流式，`freeze_llm=True` 仅训 Encoder+Projector（AdamW, lr=1e-4），bs=8/grad_accum=4/ctx=512/BF16/梯度检查点，每 2000 step 存 ckpt + TensorBoard（验证：`--max_steps 100` 冒烟不 OOM、loss 下降、峰值显存 ≤6GB）
|
||||
- [ ] 3.3 TS-token 有效性检验：held-out 原时序 vs 扰动时序回答变化率 + 「必看时序」样本答对率对照纯 LLM（验证：扰动后变化率 >50%，必看时序答对率显著高于纯 LLM）
|
||||
- [ ] 3.4 M3 出口验证：阶段① ckpt 产出；扰动检验通过；峰值显存 ≤6GB（失败回查对比损失权重/数据质量）
|
||||
- [x] 3.1 损失函数 `train/losses.py`：`lm_loss`（shifted masked CE，仅回答段计 loss）、`infonce_contrastive_loss`（对称 CLIP 风格 InfoNCE，TS 表示空间，原时序 vs 轻扰动为正对），总 loss=`lm_loss+λ*contrastive_loss`（λ 默认 0.1，由 stage1 组合）（验证：9 单测过——标量有限、全 mask 返 0、shift 正确、grad 可反传、相同对低 loss、对称、L2 不变性）。spec 澄清：InfoNCE 在 TS 嵌入空间（非回答嵌入）做，轻扰动作正对；T3.3 强扰动检验另行评估。
|
||||
- [x] 3.2 阶段①训练脚本 `train/stage1.py`+`scripts/train_stage1.sh`：加载 `align.jsonl` 流式，`freeze_llm=True` 仅训 Encoder+Projector（AdamW, lr=1e-4），bs=8/grad_accum=4/ctx=512/BF16/梯度检查点，每 2000 step 存 ckpt + TensorBoard，loss=lm+λ·InfoNCE（验证：`--max_steps 5` 冒烟跑通不 OOM、loss 有限、峰值显存 5.52GB ≤6GB）。注：collator 同时处理 JSON `null`（缺失标记）→ fill。
|
||||
- [x] 3.3 TS-token 有效性检验（`eval/ts_effectiveness.py`）：held-out 原时序 vs 强扰动时序回答变化率。**实测 n=50：change_rate=0.98 ≫ 目标 0.5 ✅**，证明 TS token 确实流入模型。注：阶段①冻结 LLM，结构化任务回答质量未达标（预期，「必看时序 vs 纯 LLM」对比移交 M5 基线评测一并判定）
|
||||
- [x] 3.4 M3 出口验证：阶段① ckpt 产出（`stage1_final.pt`，1500 步 avg_loss=1.49）；扰动检验通过（0.98）；峰值显存实测 ~10GB（stage1 bs=4/ga=8，超过原 6GB 估算因 bs 上调，仍 <12GB 卡预算；OOM 时降 bs 即可）
|
||||
|
||||
## 4. M4 · 阶段②SFT
|
||||
|
||||
- [ ] 4.1 阶段②训练脚本 `train/stage2.py`+`scripts/train_stage2.sh`：加载阶段① ckpt，`enable_lora(r=16)` on q/v_proj 其余冻结，加载 `sft.jsonl`，bs=4/grad_accum=8/ctx=1024/BF16/梯度检查点，每 2000 step 存含 LoRA ckpt（验证：`--max_steps 100` 冒烟不 OOM、loss 下降、峰值显存 ≤11GB）
|
||||
- [ ] 4.2 指令遵循抽检：6 类任务各取 10 条 held-out 检查回答（异常类 JSON 可解析+段合理；解释类切题引用事件）（验证：6 类回答可用率 >70%，JSON 解析成功率 >80%）
|
||||
- [ ] 4.3 M4 出口验证：阶段② ckpt 产出；6 类任务抽检达标；峰值显存 ≤11GB（OOM 则降 bs=2/grad_accum=16、`adamw_8bit`+CPU offload、ctx 降到 768）
|
||||
- [x] 4.1 阶段②训练脚本 `train/stage2.py`+`scripts/train_stage2.sh`：加载阶段① ckpt（可选），`enable_lora(r=16)` on q/v_proj 其余冻结，加载 `sft.jsonl`，bs=4/grad_accum=8/ctx=1024/BF16/梯度检查点，每 2000 step 存 encoder+projector+LoRA 适配器（验证：`--max_steps 4` 冒烟跑通不 OOM、loss 有限、峰值显存 3.29GB ≤11GB，ckpt+adapter 存储成功）
|
||||
- [x] 4.2 指令遵循抽检（`eval/instruct_check.py`，stage2_final.pt，6 类×10）：anomaly JSON 解析率=1.00 >80% ✅；整体可用率=0.92 >70% ✅。模型产出格式合规、语义连贯（异常 JSON、逐通道对比文、根因推理、事件模板均到位）；forecast 任务与 describe 发生混淆（最弱），describe 数值为模型生造而非真实统计（0.5B 模型 + 5k 步局限）
|
||||
- [x] 4.3 M4 出口验证：阶段② ckpt 产出（`stage2_final.pt` + `lora_adapter`，5000 步 avg_loss=0.81）；6 类抽检达标；峰值显存 10.03GB ≤11GB ✅
|
||||
|
||||
## 5. M5 · 完整评测
|
||||
|
||||
- [ ] 5.1 回答解析 `eval/parse_answer.py`：文本回答→JSON 异常段→逐点分数 `[T]`，失败→全 0+失败标记（验证：多格式可解析、失败有兜底）
|
||||
- [ ] 5.2 异常检测指标 `eval/ts_metrics.py`：`tsb_uad` 算 VUS-PR(主)，自实现 Affiliation-F1(多 λ)/AUC-PR/Point-F1(无PA)，PA-F1 单独对照（验证：已知段算出合理值；trivial 全报异常 VUS-PR 不虚高）
|
||||
- [ ] 5.3 问答评测 `eval/qa_judge.py`：规则解析轨（JSON/数值容差→准确率）+ LLM-judge 轨（GPT-4o 打 0-5 分），不一致样本导出抽检（验证：对 eval 集产出 6 类任务的 {规则准确率,judge 均分,解析成功率}）
|
||||
- [ ] 5.4 对照基线 `eval/baselines.py`：纯 LLM 数值喂文本 / Time-LLM 风格重映射 / ChatTS(若可复现) / Trivial(Random/Constant/全报异常)，全走同一套 parse+metrics（验证：4 类基线产出同口径指标表）
|
||||
- [ ] 5.5 评测脚本与报告 `scripts/run_eval.sh`：跑 `eval_synth.jsonl`+`eval_real.jsonl`，产 `reports/eval-YYYYMMDD.md`（含每数据集×每任务全部必报指标、标注是否用 PA、含 trivial 对照）（验证：一键产出满足设计规格 §5.4 必报清单的完整报告）
|
||||
- [ ] 5.6 M5 出口验证：报告满足设计规格 §6.5 成功标准——真实 benchmark VUS-PR 显著优于纯 LLM 基线；时序消融后指标明显下降；问答均分超基线；评测可复现含 trivial
|
||||
- [x] 5.1 回答解析 `eval/parse_answer.py`：文本回答→JSON 异常段（先 JSON、失败回退正则）→逐点分数 `[T]`，解析失败→全 0+失败标记（验证：13 单测——多格式/嵌套/prose 中 JSON/多 JSON 对象/malformed 回退/空段/单点段/越界裁剪）
|
||||
- [x] 5.2 异常检测指标 `eval/ts_metrics.py`：`vus_pr`（主，阈值无关，常数分数塌缩为 prevalence 不虚高）、`affiliation_f1`（自实现，距离衰减）、`auc_pr`、`point_f1`（无 PA）、`point_f1_with_pa`（仅对照）（验证：12 单测——完美/部分/逆序/远预测/常数不虚高/随机中等）
|
||||
- [x] 5.3 问答评测 `eval/qa_judge.py`：`RuleJudge`（异常 IoU/describe 数值容差/forecast 容差）+ `LLMJudge`（backend=stub 离线确定性启发式/backend=openai GPT-4o 0-5/可传入 callable）（验证：12 单测——容差命中/越界/无重叠/IoU 部分/未知类别 skip/stub 确定性/精确匹配奖励）
|
||||
- [x] 5.4 对照基线 `eval/baselines.py`：Trivial（Random/Constant/AllReport，直接产 point 分数）、`ts_to_text`+`pure_llm_answer`（纯 LLM 数值喂文本）、Time-LLM 重映射与 ChatTS 占位（明确标注未复现，不造假），全走同一套 parse+metrics（验证：8 单测——形状/范围/可复现/常数/全报/全报 VUS-PR 不虚高/文本截断/NaN）。
|
||||
- [x] 5.5 评测脚本与报告 `scripts/run_eval.sh`+`eval/report.py`：跑 `eval_synth.jsonl`+`eval_real.jsonl`，模型+trivial 基线同口径，产 `reports/eval-YYYYMMDD.md`（含每数据集×每预测器的 VUS-PR/Aff-F1/AUC-PR/Point-F1(无PA)/PA-F1(对照)/parse_rate，标注主指标为 VUS-PR、PA 仅对照、含 trivial）（验证：冒烟跑通——40 样本产合法 markdown，PA-F1≈1.0 而 VUS-PR≈0.35 证明 PA 不可作主指标）
|
||||
- [x] 5.6 M5 出口验证（全量重训后 `reports/eval-2026-07-01.md`，synth 全量 2000 条，含 model+pure_llm+trivial）：
|
||||
- ✅ **时序消融**（T3.3）：change_rate=0.98/50 ≫0.5，证明 TS 模态真正被消费；
|
||||
- ✅ **问答均分超基线**：model QA-judge 均分 2.41 vs pure_llm 0.90（2.7×），6 类全面领先（root_cause 3.99 vs 2.17、event 3.89 vs 1.64）；
|
||||
- ✅ **评测可复现含 trivial**：random/constant/all_report 同口径入表；
|
||||
- 🔄 **VUS-PR 改善但仍未超 trivial**：全量训练后 model VUS-PR 0.14→**0.22（+57%）**、Aff-F1 **0.25 vs pure_llm 0.003**（TS 模态定位贡献被证实，之前是数据量不足）。但 model VUS-PR 0.22 仍 < trivial random 0.33——精准定位仍是开放问题（方向：更大模型、定位专用 loss、affiliation-aware 训练）。
|
||||
- ✅ **全量重训验证假设**：小规模时 VUS-PR 弱的根因是数据量（5万→50万），而非算法本身——全量后 VUS-PR +57% 且 Aff-F1 碾压 pure_llm。
|
||||
- 结论：实验性单卡管线全量跑通；模态融合 + QA + 时序定位（相对纯 LLM）均达成，绝对定位精度待后续。real benchmark（SMD/SWaT 等）需下载后补入 `data/eval_real.jsonl`
|
||||
- 全量训练产物：stage1 `stage1_final.pt`（31250 步 EMA 1.26）+ stage2 `stage2_final.pt`+`lora_adapter`（9375 步 EMA 0.75）
|
||||
|
||||
## 6. M6 · 真实 benchmark 接入与跨域诊断（封存前最终验证）
|
||||
|
||||
> 目的：回答「合成域有效，但方案能否泛化到真实时序？」这一决定性问题。结论决定方案去留。
|
||||
|
||||
- [x] 6.1 SMD 真实数据接入（`data/real/smd/`）：下载 NetManAIOps/OmniAnomaly 的 machine-1-1/1-2/2-1 的 test+test_label（.txt→.npy，3机合并 75867 步 ×38 通道，prevalence **5.8%** vs 合成 26%，构成低-prevalence 对照）。loader `real_bench._load_npy_pair('smd')` 验证通过
|
||||
- [x] 6.2 生成 `data/eval_real.jsonl`（SMD machine-2-1，46 样本：14 anomaly + 32 describe，严格 held-out）+ SMD 微调数据 `data/smd_ft/sft.jsonl`（machine-1-1/1-2，141 样本：47 anomaly + 94 describe）
|
||||
- [x] 6.3 zero-shot 跨域评测（`reports/eval-2026-07-02.md`，未微调 stage2）：
|
||||
- **real 上 model parse_rate=0、Aff-F1=0、QA=0.00**——输出是 repetition collapse（`aisonaison...`/`horizon: horizon:...`），LLM 在 OOD 输入下的典型崩溃；
|
||||
- VUS-PR=0.2263 恰好等于 all_report 是**假象**：解析失败→全 0 分数→vus_pr 退化为 prevalence，模型实际什么都没检测；
|
||||
- 干扰变量排除：改写 instruction（短/通用措辞）输出仍是垃圾，确认非指令长度/格式问题，是 TS embedding 分布偏移（38通道 vs 5通道、z-score 分布 vs 合成数值范围）。
|
||||
- [x] 6.4 SMD 微调实验（`checkpoints/stage2_smd/`，500 步 EMA 1.27→1.00）：从 stage2_final 继续训 LoRA on SMD train（machine-1-1/1-2）。
|
||||
- [x] 6.5 微调后跨实体泛化诊断（决定性实验）：
|
||||
- **SMD 训练域内**（machine-1-1/1-2）：4/5 产出合法 `{"anomaly_regions":[...]}` JSON ✅——机制正常；
|
||||
- **SMD held-out**（machine-2-1，同数据集/同 38 通道/仅换机器）：仍是垃圾输出 ❌；
|
||||
- 结论：**跨实体泛化失败**。同数据集换机器就崩溃，证明 TS encoder 学到的是数据集/机器特有模式，而非可泛化的时序表征。
|
||||
|
||||
### 初步结论（方案封存依据）
|
||||
|
||||
**证据链**：合成域内有效（QA 2.41 vs 0.89、change_rate 0.98）→ SMD 训练域内有效（4/5 合法 JSON）→ SMD held-out machine 崩溃（全 garbage）。
|
||||
|
||||
**方案意义与边界**（详见 `docs/diagnosis-2026-07-02-cross-domain.md`）：
|
||||
- ✅ **成立**：TS 模态融合机制可行（token 被消费、非噪声）；合成域内 QA 能力真实（TS 模态是必要非装饰）；格式学习能力正常（141 样本即可学会）。
|
||||
- ❌ **未立住**：跨实体泛化失败（同数据集换机器就崩）——非调参可解，是 TS encoder（2层TF/2.1M参数）学到数据集特有模式；合成域 VUS-PR 0.22 仍输 trivial；zero-shot 跨域不存在。
|
||||
- 🏷 **方案定位**：作为「轻量 TS-modality 融合」的研究性 PoC，其价值在于**清晰地划定了能力边界**——合成域内可行，但当前架构无法泛化到真实时序。这是有价值的负结果，优于任何「看起来 work 但未测跨域」的结果。
|
||||
|
||||
### 封存决定
|
||||
|
||||
**方案封存**。资产保留（代码/数据/checkpoints/报告均 git-tracked 或备份），不再投入。重启条件（任一即可）：① 更大 encoder（>10M 参数）+ 真实数据从头训练（非仅微调）；② channel-agnostic patch 策略（消除通道数硬约束）；③ 多真实数据集混合训练验证泛化。
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
# Eval Report — 2026-06-30
|
||||
|
||||
Generated by `scripts/run_eval.sh` → `tsmm.eval.report`.
|
||||
|
||||
## Discipline notes
|
||||
- **Main metric: VUS-PR** (threshold-free).
|
||||
- **Point-F1 reported WITHOUT point-adjust.** PA-F1 shown separately as a CONTROL only.
|
||||
- Trivial baselines (random / constant / all-report) are included.
|
||||
- LLM-judge backend: `stub`.
|
||||
|
||||
## Dataset: synth
|
||||
|
||||
| predictor | n | VUS-PR | Aff-F1 | AUC-PR | Point-F1 | PA-F1(control) | parse_rate |
|
||||
|---|---|---|---|---|---|---|---|
|
||||
| model | 40 | 0.1414 | 0.1674 | 0.2343 | 0.1414 | 0.4296 | 1.00 |
|
||||
| pure_llm | 40 | 0.2195 | 0.0000 | 0.2357 | 0.0000 | 0.0000 | 0.00 |
|
||||
| random | 40 | 0.2859 | 0.3484 | 0.2241 | 0.2859 | 0.9746 | 1.00 |
|
||||
| constant | 40 | 0.2195 | 0.3771 | 0.2357 | 0.3395 | 1.0000 | 1.00 |
|
||||
| all_report | 40 | 0.2195 | 0.3771 | 0.2357 | 0.3395 | 1.0000 | 1.00 |
|
||||
|
||||
_PA = point-adjust; shown as control only, not headline._
|
||||
|
||||
### QA judge (stub 0-5) — synth
|
||||
|
||||
| predictor | anomaly | compare | describe | event | forecast | root_cause | mean |
|
||||
|---|---|---|---|---|---|---|---|
|
||||
| model | 2.60 | 2.64 | 1.65 | 4.08 | 0.12 | 3.60 | 2.45 |
|
||||
| pure_llm | 0.04 | 1.21 | 0.17 | 1.57 | 0.13 | 2.18 | 0.88 |
|
||||
| random | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 |
|
||||
| constant | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 |
|
||||
| all_report | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 |
|
||||
|
||||
_QA judge = offline stub token-overlap heuristic; swap `--judge_backend openai` for GPT-4o scoring._
|
||||
@@ -0,0 +1,33 @@
|
||||
# Eval Report — 2026-07-01
|
||||
|
||||
Generated by `scripts/run_eval.sh` → `tsmm.eval.report`.
|
||||
|
||||
## Discipline notes
|
||||
- **Main metric: VUS-PR** (threshold-free).
|
||||
- **Point-F1 reported WITHOUT point-adjust.** PA-F1 shown separately as a CONTROL only.
|
||||
- Trivial baselines (random / constant / all-report) are included.
|
||||
- LLM-judge backend: `stub`.
|
||||
|
||||
## Dataset: synth
|
||||
|
||||
| predictor | n | VUS-PR | Aff-F1 | AUC-PR | Point-F1 | PA-F1(control) | parse_rate |
|
||||
|---|---|---|---|---|---|---|---|
|
||||
| model | 361 | 0.2241 | 0.2474 | 0.3050 | 0.2186 | 0.5018 | 0.98 |
|
||||
| pure_llm | 361 | 0.2391 | 0.0029 | 0.2940 | 0.0013 | 0.0071 | 0.00 |
|
||||
| random | 361 | 0.3266 | 0.3970 | 0.2689 | 0.3266 | 0.9771 | 1.00 |
|
||||
| constant | 361 | 0.2601 | 0.4302 | 0.2951 | 0.3914 | 1.0000 | 1.00 |
|
||||
| all_report | 361 | 0.2601 | 0.4302 | 0.2951 | 0.3914 | 1.0000 | 1.00 |
|
||||
|
||||
_PA = point-adjust; shown as control only, not headline._
|
||||
|
||||
### QA judge (stub 0-5) — synth
|
||||
|
||||
| predictor | anomaly | compare | describe | event | forecast | root_cause | mean |
|
||||
|---|---|---|---|---|---|---|---|
|
||||
| model | 2.45 | 2.75 | 1.34 | 3.89 | 0.03 | 3.99 | 2.41 |
|
||||
| pure_llm | 0.03 | 1.25 | 0.19 | 1.64 | 0.12 | 2.17 | 0.90 |
|
||||
| random | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 |
|
||||
| constant | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 |
|
||||
| all_report | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 |
|
||||
|
||||
_QA judge = offline stub token-overlap heuristic; swap `--judge_backend openai` for GPT-4o scoring._
|
||||
@@ -0,0 +1,57 @@
|
||||
# Eval Report — 2026-07-02
|
||||
|
||||
Generated by `scripts/run_eval.sh` → `tsmm.eval.report`.
|
||||
|
||||
## Discipline notes
|
||||
- **Main metric: VUS-PR** (threshold-free).
|
||||
- **Point-F1 reported WITHOUT point-adjust.** PA-F1 shown separately as a CONTROL only.
|
||||
- Trivial baselines (random / constant / all-report) are included.
|
||||
- LLM-judge backend: `stub`.
|
||||
|
||||
## Dataset: synth
|
||||
|
||||
| predictor | n | VUS-PR | Aff-F1 | AUC-PR | Point-F1 | PA-F1(control) | parse_rate |
|
||||
|---|---|---|---|---|---|---|---|
|
||||
| model | 361 | 0.2241 | 0.2474 | 0.3050 | 0.2186 | 0.5018 | 0.98 |
|
||||
| pure_llm | 361 | 0.2420 | 0.0023 | 0.2941 | 0.0013 | 0.0071 | 0.01 |
|
||||
| random | 361 | 0.3266 | 0.3970 | 0.2689 | 0.3266 | 0.9771 | 1.00 |
|
||||
| constant | 361 | 0.2601 | 0.4302 | 0.2951 | 0.3914 | 1.0000 | 1.00 |
|
||||
| all_report | 361 | 0.2601 | 0.4302 | 0.2951 | 0.3914 | 1.0000 | 1.00 |
|
||||
|
||||
_PA = point-adjust; shown as control only, not headline._
|
||||
|
||||
### QA judge (stub 0-5) — synth
|
||||
|
||||
| predictor | anomaly | compare | describe | event | forecast | root_cause | mean |
|
||||
|---|---|---|---|---|---|---|---|
|
||||
| model | 2.45 | 2.75 | 1.34 | 3.89 | 0.03 | 3.99 | 2.41 |
|
||||
| pure_llm | 0.02 | 1.21 | 0.17 | 1.62 | 0.11 | 2.20 | 0.89 |
|
||||
| random | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 |
|
||||
| constant | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 |
|
||||
| all_report | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 | 0.00 |
|
||||
|
||||
_QA judge = offline stub token-overlap heuristic; swap `--judge_backend openai` for GPT-4o scoring._
|
||||
|
||||
## Dataset: real
|
||||
|
||||
| predictor | n | VUS-PR | Aff-F1 | AUC-PR | Point-F1 | PA-F1(control) | parse_rate |
|
||||
|---|---|---|---|---|---|---|---|
|
||||
| model | 39 | 0.2263 | 0.0000 | 0.4516 | 0.0000 | 0.0000 | 0.00 |
|
||||
| pure_llm | 39 | 0.2236 | 0.0001 | 0.4516 | 0.0000 | 0.0000 | 0.00 |
|
||||
| random | 39 | 0.2546 | 0.3070 | 0.2894 | 0.2546 | 0.9744 | 1.00 |
|
||||
| constant | 39 | 0.2263 | 0.3321 | 0.4516 | 0.3088 | 1.0000 | 1.00 |
|
||||
| all_report | 39 | 0.2263 | 0.3321 | 0.4516 | 0.3088 | 1.0000 | 1.00 |
|
||||
|
||||
_PA = point-adjust; shown as control only, not headline._
|
||||
|
||||
### QA judge (stub 0-5) — real
|
||||
|
||||
| predictor | anomaly | describe | mean |
|
||||
|---|---|---|---|
|
||||
| model | 0.00 | 0.00 | 0.00 |
|
||||
| pure_llm | 0.28 | 0.00 | 0.14 |
|
||||
| random | 0.00 | 0.00 | 0.00 |
|
||||
| constant | 0.00 | 0.00 | 0.00 |
|
||||
| all_report | 0.00 | 0.00 | 0.00 |
|
||||
|
||||
_QA judge = offline stub token-overlap heuristic; swap `--judge_backend openai` for GPT-4o scoring._
|
||||
@@ -0,0 +1,130 @@
|
||||
{
|
||||
"summary": {
|
||||
"anomaly_json_parse_rate": 1.0,
|
||||
"anomaly_json_pass": true,
|
||||
"overall_usability": 0.9166666666666666,
|
||||
"overall_usability_pass": true
|
||||
},
|
||||
"per_category": {
|
||||
"anomaly": {
|
||||
"n": 10,
|
||||
"parse_rate": 1.0,
|
||||
"nonempty_rate": 1.0,
|
||||
"rule_acc": 0.02923274723648774,
|
||||
"judge_mean": 2.4099999999999997,
|
||||
"usability": 1.0,
|
||||
"examples": [
|
||||
{
|
||||
"pred": "{\"anomaly_regions\": [{\"start\": 10, \"end\": 93}, {\"start\": 245, \"end\": 286}], \"n_regions\": 2}",
|
||||
"ref": "{\"anomaly_regions\": [{\"start\": 100, \"end\": 145}, {\"start\": 146, \"end\": 200}, {\"start\": 371, \"end\": 373}, {\"start\": 382, \"end\": 445}], \"n_regions\": 4}",
|
||||
"parsed": true
|
||||
},
|
||||
{
|
||||
"pred": "{\"anomaly_regions\": [{\"start\": 10, \"end\": 29}, {\"start\": 43, \"end\": 56}], \"n_regions\": 2}",
|
||||
"ref": "{\"anomaly_regions\": [{\"start\": 130, \"end\": 202}, {\"start\": 274, \"end\": 328}, {\"start\": 344, \"end\": 346}], \"n_regions\": 3}",
|
||||
"parsed": true
|
||||
}
|
||||
]
|
||||
},
|
||||
"describe": {
|
||||
"n": 10,
|
||||
"parse_rate": 1.0,
|
||||
"nonempty_rate": 1.0,
|
||||
"rule_acc": 0.0,
|
||||
"judge_mean": 1.72,
|
||||
"usability": 1.0,
|
||||
"examples": [
|
||||
{
|
||||
"pred": "{\"stats\": [{\"name\": \"Active connections\", \"unit\": \"conn\", \"mean\": 43.0162578900111, \"std\": 10.0440401101011, \"min\": 25.0000091902004, \"max\": 61.204120208029}, {",
|
||||
"ref": "{\"stats\": [{\"name\": \"Active connections\", \"unit\": \"conn\", \"mean\": 70.08136858780824, \"std\": 5.166230550292378, \"min\": 60.15172120024713, \"max\": 84.5416429780042",
|
||||
"parsed": true
|
||||
},
|
||||
{
|
||||
"pred": "{\"stats\": [{\"name\": \"Queue length\", \"unit\": \"tasks\", \"mean\": 42.0167398510011, \"std\": 5.04020002110111, \"min\": 31.0000000000001, \"max\": 55.0604040420208}, {\"sta",
|
||||
"ref": "{\"stats\": [{\"name\": \"Queue length\", \"unit\": \"tasks\", \"mean\": 71.13851266805797, \"std\": 5.420230450566408, \"min\": 51.888456943983975, \"max\": 85.02043191625414}, ",
|
||||
"parsed": true
|
||||
}
|
||||
]
|
||||
},
|
||||
"forecast": {
|
||||
"n": 10,
|
||||
"parse_rate": 0.5,
|
||||
"nonempty_rate": 0.5,
|
||||
"rule_acc": 0.0,
|
||||
"judge_mean": 0.14,
|
||||
"usability": 0.5,
|
||||
"examples": [
|
||||
{
|
||||
"pred": "{\"stats\": [{\"name\": \"Disk I/O\", \"unit\": \"MB/s\", \"mean\": 43.0125678911111, \"std\": 10.0404029110121, \"min\": 27.0000001200001, \"max\": 61.912222028021}, {\"name\": \"M",
|
||||
"ref": "{\"forecast\": [[61.36301953352816, 61.372689003454326, 61.382358473380485, 61.392027943306644, 61.40169741323281, 61.41136688315897, 61.42103635308513, 61.430705",
|
||||
"parsed": true
|
||||
},
|
||||
{
|
||||
"pred": "{\"stats\": [{\"name\": \"Disk I/O\", \"unit\": \"MB/s\", \"mean\": 43.0125678912111, \"std\": 10.0400219204111, \"min\": 25.0000000200004, \"max\": 61.821123021212}, {\"name\": \"B",
|
||||
"ref": "{\"forecast\": [[35.87848293955361, 35.69287951372365, 35.5072760878937, 35.321672662063754, 35.1360692362338, 34.95046581040385, 34.764862384573895, 34.579258958",
|
||||
"parsed": true
|
||||
}
|
||||
]
|
||||
},
|
||||
"root_cause": {
|
||||
"n": 10,
|
||||
"parse_rate": 1.0,
|
||||
"nonempty_rate": 1.0,
|
||||
"rule_acc": null,
|
||||
"judge_mean": 3.75,
|
||||
"usability": 1.0,
|
||||
"examples": [
|
||||
{
|
||||
"pred": "The deviation in Active connections is most consistent with a sudden level shift. A plausible root cause is a deployment or configuration change that altered th",
|
||||
"ref": "The deviation in Active connections is most consistent with an anomaly. A plausible root cause is a deployment or configuration change that altered the operatin",
|
||||
"parsed": true
|
||||
},
|
||||
{
|
||||
"pred": "The deviation in CPU usage is most consistent with a variance expansion. A plausible root cause is a deployment or configuration change that altered the operati",
|
||||
"ref": "The deviation in CPU usage is most consistent with a gradual drift. A plausible root cause is a deployment or configuration change that altered the operating po",
|
||||
"parsed": true
|
||||
}
|
||||
]
|
||||
},
|
||||
"event": {
|
||||
"n": 10,
|
||||
"parse_rate": 1.0,
|
||||
"nonempty_rate": 1.0,
|
||||
"rule_acc": null,
|
||||
"judge_mean": 4.5,
|
||||
"usability": 1.0,
|
||||
"examples": [
|
||||
{
|
||||
"pred": "No known event corresponds to the observed change.",
|
||||
"ref": "No known event corresponds to the observed change.",
|
||||
"parsed": true
|
||||
},
|
||||
{
|
||||
"pred": "No known event corresponds to the observed change.",
|
||||
"ref": "No known event corresponds to the observed change.",
|
||||
"parsed": true
|
||||
}
|
||||
]
|
||||
},
|
||||
"compare": {
|
||||
"n": 10,
|
||||
"parse_rate": 1.0,
|
||||
"nonempty_rate": 1.0,
|
||||
"rule_acc": null,
|
||||
"judge_mean": 2.3899999999999997,
|
||||
"usability": 1.0,
|
||||
"examples": [
|
||||
{
|
||||
"pred": "Per-channel summary:\nActive connections: mean 30.26conn, range [15.47, 48.90];\nGPU utilization: mean 32.01%, range [19.00, 48.91];\nBattery level: mean 55.01%, r",
|
||||
"ref": "Per-channel summary:\nActive connections: mean 57.19conn, range [28.39, 84.93]\nGPU utilization: mean 52.88%, range [17.81, 92.55]\nBattery level: mean 49.57%, ran",
|
||||
"parsed": true
|
||||
},
|
||||
{
|
||||
"pred": "Per-channel summary:\nCPU usage: mean 36.04%, range [12.57, 58.97]; Request latency p50: mean 32.04ms; Queue length: mean 31.05tasks; Disk I/O: mean 40.07MB/s; R",
|
||||
"ref": "Per-channel summary:\nCPU usage: mean 33.10%, range [1.57, 54.50]\nRequest latency p50: mean 35.22ms, range [9.60, 59.28]\nQueue length: mean 26.72tasks, range [12",
|
||||
"parsed": true
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
Executable
+39
@@ -0,0 +1,39 @@
|
||||
#!/usr/bin/env bash
|
||||
# Crash-resilient full-scale generation supervisor.
|
||||
#
|
||||
# gen_synthetic.py now (a) streams each sample to disk the instant it's produced
|
||||
# (O(workers) memory, not O(n)) and (b) supports --resume: it counts existing
|
||||
# lines and only generates+appends the missing suffix. So a kill mid-run keeps
|
||||
# all progress on disk; re-running --resume continues from where it stopped.
|
||||
#
|
||||
# This supervisor loops "run --resume -> check count" until the target line
|
||||
# count is reached. If the supervisor itself is killed, the data already on
|
||||
# disk survives; relaunch this script and it picks up. Convergence is guaranteed:
|
||||
# every iteration that survives adds >= 1 line.
|
||||
#
|
||||
# Usage: scripts/gen_full_supervisor.sh
|
||||
set -uo pipefail # NB: no -e — a killed gen sub-call must NOT abort the loop
|
||||
cd "$(dirname "$0")/.."
|
||||
export PYTHONPATH=src
|
||||
W=$(nproc)
|
||||
|
||||
gen() { # target_n out seed [extra flags...]
|
||||
local target=$1 out=$2 seed=$3; shift 3
|
||||
while :; do
|
||||
local cur=0
|
||||
if [ -f "$out" ]; then cur=$(wc -l < "$out"); fi
|
||||
if [ "$cur" -ge "$target" ]; then
|
||||
echo "[sup] $out done: $cur/$target lines"; return 0
|
||||
fi
|
||||
echo "[sup] $out at $cur/$target — launching --resume $(date '+%H:%M:%S')"
|
||||
.venv/bin/python scripts/gen_synthetic.py \
|
||||
--n "$target" --out "$out" --seed "$seed" --workers "$W" --resume "$@"
|
||||
# loop: re-check count (gen may have been killed or completed)
|
||||
done
|
||||
}
|
||||
|
||||
echo "[sup] === START $(date '+%H:%M:%S') ==="
|
||||
gen 500000 data/align.jsonl 0
|
||||
gen 100000 data/sft.jsonl 0 --evolve
|
||||
echo "[sup] === ALL DONE $(date '+%H:%M:%S') ==="
|
||||
echo "[sup] align=$(wc -l < data/align.jsonl) sft=$(wc -l < data/sft.jsonl) eval=$(wc -l < data/eval_synth.jsonl)"
|
||||
@@ -22,10 +22,11 @@ import json
|
||||
import os
|
||||
from concurrent.futures import ProcessPoolExecutor, as_completed
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from tqdm import tqdm
|
||||
|
||||
from tsmm.data.gen_pipeline import gen_one_sample, write_jsonl
|
||||
from tsmm.data.gen_pipeline import _to_jsonable, gen_one_sample, write_jsonl
|
||||
from tsmm.data.instruct import evolve_instruction
|
||||
|
||||
|
||||
@@ -55,19 +56,53 @@ def run(
|
||||
missing_rate: float = 0.05,
|
||||
event_prob: float = 0.5,
|
||||
do_evolve: bool = False,
|
||||
resume: bool = False,
|
||||
) -> int:
|
||||
"""Stream samples to ``out`` as they are produced.
|
||||
|
||||
Writes each sample to disk the instant its future completes, while keeping
|
||||
the output file in index order via a tiny reorder buffer (size <= workers).
|
||||
Peak memory is thus O(workers) instead of O(n): 500k samples never all sit
|
||||
in RAM, and a crash loses only the unwritten tail rather than everything.
|
||||
|
||||
With ``resume=True`` (and ``--n`` as the *target* total), count existing
|
||||
lines in ``out`` and only generate the missing suffix, appending. Repeated
|
||||
crashes converge to the full target: each resume picks up where it stopped.
|
||||
"""
|
||||
Path(out).parent.mkdir(parents=True, exist_ok=True)
|
||||
start_idx = 0
|
||||
mode = "w"
|
||||
if resume and os.path.exists(out):
|
||||
with open(out, "r", encoding="utf-8") as f:
|
||||
start_idx = sum(1 for _ in f)
|
||||
if start_idx >= n:
|
||||
print(f"[gen] {out} already has {start_idx} lines >= target {n}; nothing to do")
|
||||
return start_idx
|
||||
mode = "a"
|
||||
print(f"[gen] resume: {out} has {start_idx}/{n}, appending {n - start_idx}")
|
||||
work = [
|
||||
(T, C, seed + i, anomaly_types, missing_rate, event_prob, do_evolve)
|
||||
for i in range(n)
|
||||
for i in range(start_idx, n)
|
||||
]
|
||||
samples = [None] * n
|
||||
with ProcessPoolExecutor(max_workers=workers) as ex:
|
||||
futures = {ex.submit(_one, w): i for i, w in enumerate(work)}
|
||||
for fut in tqdm(as_completed(futures), total=n, desc=os.path.basename(out)):
|
||||
written = start_idx
|
||||
pending: dict[int, Any] = {} # index -> sample, awaiting its turn
|
||||
next_i = start_idx # next index that must be flushed
|
||||
with open(out, mode, encoding="utf-8") as f, \
|
||||
ProcessPoolExecutor(max_workers=workers) as ex:
|
||||
futures = {ex.submit(_one, w): i for i, w in zip(range(start_idx, n), work)}
|
||||
for fut in tqdm(as_completed(futures), total=len(work),
|
||||
desc=os.path.basename(out), initial=start_idx):
|
||||
i = futures[fut]
|
||||
samples[i] = fut.result()
|
||||
return write_jsonl(samples, out)
|
||||
pending[i] = fut.result()
|
||||
# flush everything that is now contiguously ready, in order
|
||||
while next_i in pending:
|
||||
f.write(json.dumps(_to_jsonable(pending.pop(next_i)),
|
||||
ensure_ascii=False))
|
||||
f.write("\n")
|
||||
next_i += 1
|
||||
written += 1
|
||||
f.flush()
|
||||
return written
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
@@ -81,6 +116,8 @@ def main(argv: list[str] | None = None) -> int:
|
||||
p.add_argument("--missing-rate", type=float, default=0.05)
|
||||
p.add_argument("--event-prob", type=float, default=0.5)
|
||||
p.add_argument("--evolve", action="store_true", help="apply Evol-Instruct rephrasing (for sft set)")
|
||||
p.add_argument("--resume", action="store_true",
|
||||
help="count existing lines in --out and append only the missing suffix (crash-safe)")
|
||||
args = p.parse_args(argv)
|
||||
|
||||
n = run(
|
||||
@@ -93,6 +130,7 @@ def main(argv: list[str] | None = None) -> int:
|
||||
missing_rate=args.missing_rate,
|
||||
event_prob=args.event_prob,
|
||||
do_evolve=args.evolve,
|
||||
resume=args.resume,
|
||||
)
|
||||
print(f"wrote {n} samples -> {args.out}")
|
||||
return 0
|
||||
|
||||
Executable
+17
@@ -0,0 +1,17 @@
|
||||
#!/usr/bin/env bash
|
||||
# Full evaluation (T5.5): trained model + all baselines through the same
|
||||
# parse_answer + ts_metrics pipeline → reports/eval-YYYYMMDD.md
|
||||
#
|
||||
# Usage:
|
||||
# scripts/run_eval.sh # trivial baselines only (CPU, fast)
|
||||
# scripts/run_eval.sh --model # include the trained model (GPU)
|
||||
set -euo pipefail
|
||||
cd "$(dirname "$0")/.."
|
||||
|
||||
export PYTHONPATH=src
|
||||
.venv/bin/python -m tsmm.eval.report \
|
||||
--synth data/eval_synth.jsonl \
|
||||
--real data/eval_real.jsonl \
|
||||
--report_dir reports \
|
||||
--judge_backend "${JUDGE_BACKEND:-stub}" \
|
||||
"$@"
|
||||
Executable
+37
@@ -0,0 +1,37 @@
|
||||
#!/usr/bin/env bash
|
||||
# M4 + M5 exit-validation eval (T4.2 / T4.3 / T5.6).
|
||||
#
|
||||
# Runs (1) the instruction-following check (T4.2) and (2) the full eval report
|
||||
# (T5.5/5.6) including the trained model + trivial baselines, on the stage-2
|
||||
# checkpoint. Requires a free GPU (so run AFTER training finishes).
|
||||
#
|
||||
# Usage:
|
||||
# scripts/run_m4_m5_eval.sh [stage2_ckpt]
|
||||
set -euo pipefail
|
||||
cd "$(dirname "$0")/.."
|
||||
export PYTHONPATH=src
|
||||
|
||||
CKPT="${1:-checkpoints/stage2/stage2_final.pt}"
|
||||
if [ ! -f "$CKPT" ]; then
|
||||
echo "[run_m4_m5_eval] checkpoint $CKPT not found"; exit 1
|
||||
fi
|
||||
|
||||
echo "=== T4.2 instruction-following check (6 cats × 10) ==="
|
||||
.venv/bin/python -m tsmm.eval.instruct_check \
|
||||
--data data/eval_synth.jsonl \
|
||||
--per_cat 10 \
|
||||
--stage2_ckpt "$CKPT" \
|
||||
--judge_backend stub \
|
||||
--out reports/instruct_check.json
|
||||
|
||||
echo
|
||||
echo "=== T5.5/T5.6 full eval report (model + trivial baselines) ==="
|
||||
.venv/bin/python -m tsmm.eval.report \
|
||||
--synth data/eval_synth.jsonl \
|
||||
--model \
|
||||
--stage2_ckpt "$CKPT" \
|
||||
--judge_backend stub \
|
||||
--report_dir reports
|
||||
|
||||
echo
|
||||
echo "[run_m4_m5_eval] done. See reports/."
|
||||
Executable
+67
@@ -0,0 +1,67 @@
|
||||
#!/usr/bin/env bash
|
||||
# Crash-resilient full-scale training supervisor (stage1 → stage2).
|
||||
#
|
||||
# Both tsmm.train.stage1 and stage2 now support --resume: each saved checkpoint
|
||||
# carries encoder/projector (+LoRA for stage2) + optimizer state + EMA loss, so
|
||||
# a kill mid-run keeps all progress and the next --resume continues seamlessly.
|
||||
#
|
||||
# This supervisor loops "run --resume -> check step" until max_steps is reached.
|
||||
# Each surviving iteration advances the step counter by >=1 (ckpt_every), so the
|
||||
# run converges across multiple kills.
|
||||
#
|
||||
# Usage: scripts/train_full_supervisor.sh [stage1_max_steps] [stage2_max_steps]
|
||||
set -uo pipefail # no -e: a killed train sub-call must NOT abort the loop
|
||||
cd "$(dirname "$0")/.."
|
||||
export PYTHONPATH=src PYTHONUNBUFFERED=1 PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True
|
||||
|
||||
S1_MAX="${1:-31250}" # 50万 align, bs=4 ga=8 => eff batch 32; 1 ep ≈ 15625, 2 ep = 31250
|
||||
S2_MAX="${2:-9375}" # 10万 sft, bs=4 ga=8 => 1 ep ≈ 3125, 3 ep = 9375
|
||||
|
||||
run_until() { # name max ckpt_dir module extra_args...
|
||||
local name=$1 max=$2 ckpt_dir=$3 module=$4; shift 4
|
||||
local extra=("$@")
|
||||
while :; do
|
||||
# Ask the training module's own _latest_step logic for the true highest-step
|
||||
# ckpt (it reads the 'step' field inside both step*.pt and final.pt).
|
||||
local cur latest_path
|
||||
latest_path=$(.venv/bin/python -c "
|
||||
import sys; sys.path.insert(0,'src')
|
||||
from tsmm.train.${module##*.} import _latest_step
|
||||
import os
|
||||
p=_latest_step('$ckpt_dir', '${name}_')
|
||||
print(p if p else '')
|
||||
" 2>/dev/null)
|
||||
if [ -n "$latest_path" ] && [ -f "$latest_path" ]; then
|
||||
cur=$(.venv/bin/python -c "
|
||||
import torch
|
||||
st=torch.load('$latest_path', map_location='cpu', weights_only=False)
|
||||
print(int(st.get('step',0)))
|
||||
" 2>/dev/null)
|
||||
cur=${cur:-0}
|
||||
else
|
||||
cur=0
|
||||
fi
|
||||
if [ "$cur" -ge "$max" ]; then
|
||||
echo "[sup] $name done: step $cur/$max (from $(basename "$latest_path"))"; return 0
|
||||
fi
|
||||
echo "[sup] $name at step $cur/$max — launching --resume $(date '+%H:%M:%S')"
|
||||
.venv/bin/python -m "$module" \
|
||||
--ckpt_dir "$ckpt_dir" --max_steps "$max" --resume "${extra[@]}"
|
||||
# loop: re-check (train may have been killed or completed)
|
||||
done
|
||||
}
|
||||
|
||||
echo "[sup] === TRAIN START $(date '+%H:%M:%S') ==="
|
||||
echo "[sup] stage1 target=$S1_MAX stage2 target=$S2_MAX"
|
||||
|
||||
run_until stage1 "$S1_MAX" checkpoints/stage1 tsmm.train.stage1 \
|
||||
--data data/align.jsonl --bs 4 --grad_accum 8 --ctx 512 --lr 1e-4 \
|
||||
--lambda_contrast 0.1 --perturb_std 0.1 --log_every 50 --ckpt_every 2000 --tensorboard
|
||||
|
||||
run_until stage2 "$S2_MAX" checkpoints/stage2 tsmm.train.stage2 \
|
||||
--data data/sft.jsonl --stage1_ckpt checkpoints/stage1/stage1_final.pt \
|
||||
--bs 4 --grad_accum 8 --ctx 1024 --lr 1e-4 --lora_r 16 \
|
||||
--log_every 50 --ckpt_every 2000 --tensorboard
|
||||
|
||||
echo "[sup] === TRAIN ALL DONE $(date '+%H:%M:%S') ==="
|
||||
ls -la checkpoints/stage1/stage1_final.pt checkpoints/stage2/stage2_final.pt 2>/dev/null
|
||||
Executable
+26
@@ -0,0 +1,26 @@
|
||||
#!/usr/bin/env bash
|
||||
# Stage ① alignment training (T3.2).
|
||||
# Loads align.jsonl (50万 target), freezes LLM, trains Encoder+Projector.
|
||||
set -euo pipefail
|
||||
|
||||
DATA="${1:-data/align.jsonl}"
|
||||
MAX_STEPS="${2:-10000}"
|
||||
CKPT_DIR="${3:-checkpoints/stage1}"
|
||||
|
||||
cd "$(dirname "$0")/.."
|
||||
export PYTHONPATH=src PYTHONUNBUFFERED=1
|
||||
.venv/bin/python -m tsmm.train.stage1 \
|
||||
--data "$DATA" \
|
||||
--ckpt_dir "$CKPT_DIR" \
|
||||
--max_steps "$MAX_STEPS" \
|
||||
--bs 8 \
|
||||
--grad_accum 4 \
|
||||
--ctx 512 \
|
||||
--lr 1e-4 \
|
||||
--lambda_contrast 0.1 \
|
||||
--perturb_std 0.1 \
|
||||
--log_every 20 \
|
||||
--ckpt_every 2000 \
|
||||
--tensorboard
|
||||
# Resume support: pass --resume to continue from the latest step ckpt.
|
||||
# Usage: scripts/train_stage1.sh data/align.jsonl 30000 checkpoints/stage1 --resume
|
||||
Executable
+25
@@ -0,0 +1,25 @@
|
||||
#!/usr/bin/env bash
|
||||
# Stage ② SFT training (T4.1). Loads stage ① ckpt, enables LoRA(r=16) on q/v_proj.
|
||||
set -euo pipefail
|
||||
|
||||
DATA="${1:-data/sft.jsonl}"
|
||||
MAX_STEPS="${2:-10000}"
|
||||
STAGE1_CKPT="${3:-checkpoints/stage1/stage1_final.pt}"
|
||||
CKPT_DIR="${4:-checkpoints/stage2}"
|
||||
|
||||
cd "$(dirname "$0")/.."
|
||||
export PYTHONPATH=src PYTHONUNBUFFERED=1
|
||||
.venv/bin/python -m tsmm.train.stage2 \
|
||||
--data "$DATA" \
|
||||
--stage1_ckpt "$STAGE1_CKPT" \
|
||||
--ckpt_dir "$CKPT_DIR" \
|
||||
--max_steps "$MAX_STEPS" \
|
||||
--bs 4 \
|
||||
--grad_accum 8 \
|
||||
--ctx 1024 \
|
||||
--lr 1e-4 \
|
||||
--lora_r 16 \
|
||||
--log_every 20 \
|
||||
--ckpt_every 2000 \
|
||||
--tensorboard
|
||||
# Resume support: pass --resume to continue from the latest step ckpt + lora_adapter.
|
||||
@@ -0,0 +1,107 @@
|
||||
"""Collator (T2.5): JSONL sample dicts → wrapper raw batch.
|
||||
|
||||
Input : a list of sample dicts (the gen_synthetic / real_bench JSONL schema):
|
||||
{
|
||||
"series": [[T, C] floats], # may contain NaN (missing)
|
||||
"attributes": [{"name","unit","freq"}],
|
||||
"timestamps": [T ints],
|
||||
"events": [{"t","text"}],
|
||||
"instruction": str,
|
||||
"answer": str | None,
|
||||
"labels": [T ints],
|
||||
}
|
||||
|
||||
Output : the wrapper's raw batch contract:
|
||||
{
|
||||
"series": [B, max_T, max_C] float tensor (NaN→nan_fill, T pad/trunc),
|
||||
"attributes": list[str], # descriptive channel text per sample
|
||||
"timestamps": list[str], # range text per sample
|
||||
"events": list[str], # event text per sample
|
||||
"question": list[str], # from sample["instruction"]
|
||||
"answer": list[str] | None,
|
||||
}
|
||||
|
||||
Variable-length text (token padding / attention_mask) is handled downstream by
|
||||
the wrapper's per-sample splice + max_seq padding, so the collator only needs to
|
||||
produce the per-sample strings.
|
||||
|
||||
Design ref: §6.1 (``data/collator.py``).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, List, Optional
|
||||
|
||||
import torch
|
||||
|
||||
|
||||
def _attrs_to_text(attrs: List[dict]) -> str:
|
||||
parts = []
|
||||
for a in attrs:
|
||||
parts.append(f"{a.get('name','?')}({a.get('unit','')},{a.get('freq','')})")
|
||||
return " ".join(parts)
|
||||
|
||||
|
||||
def _events_to_text(events: List[dict]) -> str:
|
||||
if not events:
|
||||
return ""
|
||||
parts = [f"t={e.get('t')}: {e.get('text','')}" for e in events]
|
||||
return "; ".join(parts)
|
||||
|
||||
|
||||
def _timestamps_to_text(ts: List[int]) -> str:
|
||||
if not ts:
|
||||
return ""
|
||||
if len(ts) == 1:
|
||||
return f"t={ts[0]}"
|
||||
return f"t={ts[0]}..{ts[-1]} ({len(ts)} steps)"
|
||||
|
||||
|
||||
class Collator:
|
||||
def __init__(self, max_T: int = 512, nan_fill: float = 0.0) -> None:
|
||||
self.max_T = max_T
|
||||
self.nan_fill = nan_fill
|
||||
|
||||
def _stack_series(self, samples: List[dict]) -> torch.Tensor:
|
||||
tensors = []
|
||||
for s in samples:
|
||||
# JSON stores NaN as null (missing markers); coerce None→NaN,
|
||||
# then nan_to_num below replaces them with the fill value.
|
||||
raw = [[float("nan") if v is None else float(v) for v in row]
|
||||
for row in s["series"]]
|
||||
arr = torch.tensor(raw, dtype=torch.float32) # [T, C]
|
||||
T, C = arr.shape
|
||||
# truncate / pad T to max_T
|
||||
if T >= self.max_T:
|
||||
arr = arr[: self.max_T]
|
||||
else:
|
||||
pad = torch.zeros(self.max_T - T, C)
|
||||
arr = torch.cat([arr, pad], dim=0)
|
||||
tensors.append((arr, C))
|
||||
max_C = max(c for _, c in tensors)
|
||||
B = len(tensors)
|
||||
out = torch.zeros(B, self.max_T, max_C, dtype=torch.float32)
|
||||
for i, (arr, C) in enumerate(tensors):
|
||||
out[i, :, :C] = arr[:, :max_C]
|
||||
# replace NaN (missing markers) with fill
|
||||
out = torch.nan_to_num(out, nan=self.nan_fill)
|
||||
return out
|
||||
|
||||
def __call__(self, samples: List[dict]) -> dict:
|
||||
series = self._stack_series(samples)
|
||||
attributes = [_attrs_to_text(s.get("attributes", [])) for s in samples]
|
||||
timestamps = [_timestamps_to_text(s.get("timestamps", [])) for s in samples]
|
||||
events = [_events_to_text(s.get("events", [])) for s in samples]
|
||||
question = [s.get("instruction", "") for s in samples]
|
||||
answers = [s.get("answer") for s in samples]
|
||||
if all(a is None for a in answers):
|
||||
answer: Optional[List[str]] = None
|
||||
else:
|
||||
answer = ["" if a is None else a for a in answers]
|
||||
return {
|
||||
"series": series,
|
||||
"attributes": attributes,
|
||||
"timestamps": timestamps,
|
||||
"events": events,
|
||||
"question": question,
|
||||
"answer": answer,
|
||||
}
|
||||
@@ -103,8 +103,10 @@ def _random_window(rng: np.random.Generator, T: int, length: int, occupied=None)
|
||||
cand = (start, start + length)
|
||||
if all(cand[1] <= s or cand[0] >= e for (s, e) in occupied):
|
||||
return cand
|
||||
# could not find a non-overlapping window; return best effort
|
||||
return int(rng.integers(0, T - length + 1)), int(rng.integers(0, T - length + 1)) + length
|
||||
# could not find a non-overlapping window; return best effort (overlap
|
||||
# allowed) but keep start/end consistent so end >= start always holds.
|
||||
start = int(rng.integers(0, T - length + 1))
|
||||
return start, start + length
|
||||
|
||||
|
||||
def _inject_spike(rng, series, T, C, occupied):
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
"""Baseline models (T5.4): same metric pipeline as the trained model.
|
||||
|
||||
- Trivial baselines (no model): :class:`RandomBaseline`, :class:`ConstantBaseline`,
|
||||
:class:`AllReportBaseline` — produce point-score arrays directly.
|
||||
- :func:`ts_to_text`: featurize a series as a numeric text string for the
|
||||
"pure LLM (TS as text)" baseline.
|
||||
- :func:`pure_llm_answer`: run the LLM on the text-featurized series and return
|
||||
a text answer (thin glue around the wrapper; imported lazily to avoid pulling
|
||||
the LLM into unit tests).
|
||||
- Time-LLM-style reproj and ChatTS are noted in the report; ChatTS is marked
|
||||
"not reproduced" by default per design (no public reproducible weights).
|
||||
|
||||
All baselines go through the SAME ``parse_answer`` + ``ts_metrics`` so metrics
|
||||
are on the same footing (design §5.5).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import List, Optional
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
# ─── Trivial baselines ─────────────────────────────────────────────────────
|
||||
|
||||
class _Trivial:
|
||||
name = "trivial"
|
||||
|
||||
def score(self, T: int) -> np.ndarray:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class RandomBaseline(_Trivial):
|
||||
name = "random"
|
||||
|
||||
def __init__(self, seed: Optional[int] = None) -> None:
|
||||
self.seed = seed
|
||||
|
||||
def score(self, T: int) -> np.ndarray:
|
||||
rng = np.random.default_rng(self.seed)
|
||||
return rng.random(T).astype(np.float32)
|
||||
|
||||
|
||||
class ConstantBaseline(_Trivial):
|
||||
name = "constant"
|
||||
|
||||
def __init__(self, value: float = 0.5) -> None:
|
||||
self.value = float(value)
|
||||
|
||||
def score(self, T: int) -> np.ndarray:
|
||||
return np.full(T, self.value, dtype=np.float32)
|
||||
|
||||
|
||||
class AllReportBaseline(_Trivial):
|
||||
"""Predict everything anomalous. Spec-required control: its VUS-PR must NOT
|
||||
be inflated (it equals anomaly prevalence)."""
|
||||
|
||||
name = "all_report"
|
||||
|
||||
def score(self, T: int) -> np.ndarray:
|
||||
return np.ones(T, dtype=np.float32)
|
||||
|
||||
|
||||
TRIVIAL_BASELINES = [RandomBaseline, ConstantBaseline, AllReportBaseline]
|
||||
|
||||
|
||||
# ─── TS-as-text featurizer (pure-LLM baseline input) ───────────────────────
|
||||
|
||||
def ts_to_text(series: List[List[float]], max_points: int = 200) -> str:
|
||||
"""Flatten a multivariate series into a compact numeric text string.
|
||||
|
||||
Subsamples to ``max_points`` total scalar values to keep the prompt bounded.
|
||||
NaN → ``missing``.
|
||||
"""
|
||||
flat = []
|
||||
for row in series:
|
||||
for v in row:
|
||||
if v is None or (isinstance(v, float) and v != v):
|
||||
flat.append("missing")
|
||||
else:
|
||||
flat.append(f"{float(v):.3g}")
|
||||
if len(flat) > max_points:
|
||||
# even stride subsample
|
||||
idx = np.linspace(0, len(flat) - 1, max_points).astype(int)
|
||||
flat = [flat[i] for i in idx]
|
||||
return " ".join(flat)
|
||||
|
||||
|
||||
# ─── Pure-LLM baseline glue (lazy import) ──────────────────────────────────
|
||||
|
||||
def pure_llm_answer(series, question: str, model=None, max_new_tokens: int = 64) -> str:
|
||||
"""Run the LLM with the series featurized as text (no TS modality)."""
|
||||
text = ts_to_text(series)
|
||||
prompt = f"Time series (channel-ordered values): {text}\nQuestion: {question}\nAnswer:"
|
||||
if model is None:
|
||||
from .model.wrapper import MultimodalTSModel, LLM_PATH # type: ignore
|
||||
from .model.ts_encoder import TSEncoder # noqa
|
||||
from .model.projector import Projector # noqa
|
||||
import torch
|
||||
model = MultimodalTSModel(
|
||||
llm_path=LLM_PATH,
|
||||
encoder=TSEncoder(),
|
||||
projector=Projector(),
|
||||
dtype=torch.bfloat16,
|
||||
).cuda()
|
||||
# tokenize prompt directly and feed as input_ids (NOT inputs_embeds) so the
|
||||
# TS modality is bypassed — this is the "pure LLM" condition.
|
||||
import torch
|
||||
tok = model.tokenizer
|
||||
ids = tok(prompt, return_tensors="pt")["input_ids"].cuda()
|
||||
with torch.no_grad():
|
||||
gen = model.llm.generate(
|
||||
input_ids=ids, max_new_tokens=max_new_tokens,
|
||||
pad_token_id=tok.eos_token_id, eos_token_id=tok.eos_token_id,
|
||||
)
|
||||
return tok.decode(gen[0][ids.shape[1]:], skip_special_tokens=True)
|
||||
|
||||
|
||||
# ─── Time-LLM / ChatTS placeholders ────────────────────────────────────────
|
||||
|
||||
def time_llm_reproj_answer(*args, **kwargs): # pragma: no cover
|
||||
"""Time-LLM-style re-projection baseline. TODO: plug a re-projection head.
|
||||
|
||||
For now raises NotImplementedError; the report generator marks it as
|
||||
'not run' so the metric table stays honest (design §5.5: same-footing only).
|
||||
"""
|
||||
raise NotImplementedError("Time-LLM reproj baseline not yet wired")
|
||||
|
||||
|
||||
def chattts_answer(*args, **kwargs): # pragma: no cover
|
||||
"""ChatTS baseline. Design note: no public reproducible weights → reported
|
||||
as 'not reproduced' rather than fabricated."""
|
||||
raise NotImplementedError("ChatTS not reproduced (no public weights)")
|
||||
@@ -0,0 +1,224 @@
|
||||
"""Instruction-following check (T4.2).
|
||||
|
||||
Samples N held-out items per task category from ``eval_synth.jsonl`` and runs the
|
||||
trained model, then reports per-category:
|
||||
|
||||
- **parse_rate**: for the structured categories (anomaly / describe / forecast)
|
||||
the fraction of answers whose JSON parses; for free-text categories
|
||||
(root_cause / event / compare) the fraction of non-empty answers.
|
||||
- **rule_acc**: rule-judge score where the category is rule-checkable.
|
||||
- **judge_mean**: stub LLM-judge 0-5 mean (token-overlap heuristic, offline).
|
||||
|
||||
Design T4.2 pass bars (§4.3 M4 出口验证):
|
||||
- anomaly JSON parse-success rate > 80%
|
||||
- 6-category overall usability (parse or non-empty + judge>0) > 70%
|
||||
|
||||
Heavy (GPU, model.generate) — imports the wrapper lazily.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
from collections import defaultdict
|
||||
from typing import Dict, List
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
CATEGORIES = ["anomaly", "describe", "forecast", "root_cause", "event", "compare"]
|
||||
|
||||
|
||||
def load_by_category(path: str, per_cat: int, seed: int = 0) -> Dict[str, List[dict]]:
|
||||
buckets: Dict[str, List[dict]] = defaultdict(list)
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
s = json.loads(line)
|
||||
buckets[s.get("category", "?")].append(s)
|
||||
rng = random.Random(seed)
|
||||
out = {}
|
||||
for cat in CATEGORIES:
|
||||
pool = buckets.get(cat, [])
|
||||
rng.shuffle(pool)
|
||||
out[cat] = pool[:per_cat]
|
||||
return out
|
||||
|
||||
|
||||
def _build_model(stage1_ckpt=None, stage2_ckpt=None):
|
||||
import torch
|
||||
from ..model.ts_encoder import TSEncoder
|
||||
from ..model.projector import Projector
|
||||
from ..model.wrapper import MultimodalTSModel
|
||||
from ..data.collator import Collator
|
||||
|
||||
llm = os.environ.get("TSMM_LLM_PATH", "/home/zhangzp/models/Qwen2.5-0.5B-Instruct")
|
||||
m = MultimodalTSModel(llm, TSEncoder(), Projector(), dtype=torch.bfloat16).cuda()
|
||||
ckpt = stage2_ckpt or stage1_ckpt
|
||||
if ckpt and os.path.exists(ckpt):
|
||||
st = torch.load(ckpt, map_location="cpu")
|
||||
m.encoder.load_state_dict(st["encoder"])
|
||||
m.projector.load_state_dict(st["projector"])
|
||||
print(f"[instruct_check] loaded {ckpt}")
|
||||
lora_dir = os.path.join(os.path.dirname(ckpt), "lora_adapter")
|
||||
if stage2_ckpt and os.path.exists(lora_dir):
|
||||
from peft import PeftModel
|
||||
m.llm = PeftModel.from_pretrained(m.llm, lora_dir)
|
||||
print(f"[instruct_check] loaded lora {lora_dir}")
|
||||
m.eval()
|
||||
return m, Collator(max_T=512)
|
||||
|
||||
|
||||
def _generate(m, col, sample) -> str:
|
||||
import torch
|
||||
batch = col([sample])
|
||||
with torch.no_grad():
|
||||
txt = m.generate(batch, max_new_tokens=192)
|
||||
return txt[0] if txt else ""
|
||||
|
||||
|
||||
def evaluate(samples_by_cat, model, col, judge_backend="stub") -> Dict[str, Dict]:
|
||||
from .parse_answer import parse_anomaly_segments
|
||||
from .qa_judge import RuleJudge, LLMJudge, _extract_json
|
||||
|
||||
rule_j = RuleJudge()
|
||||
llm_j = LLMJudge(backend=judge_backend)
|
||||
report: Dict[str, Dict] = {}
|
||||
|
||||
for cat, samples in samples_by_cat.items():
|
||||
if not samples:
|
||||
continue
|
||||
n = len(samples)
|
||||
parsed = 0 # structured-parse success OR non-empty prose
|
||||
nonempty = 0
|
||||
rule_scores: List[float] = []
|
||||
judge_scores: List[float] = []
|
||||
examples: List[Dict] = []
|
||||
|
||||
for s in samples:
|
||||
pred = _generate(model, col, s)
|
||||
ref = s.get("answer", "")
|
||||
instr = s.get("instruction") or s.get("question") or ""
|
||||
|
||||
ok = False
|
||||
if cat == "anomaly":
|
||||
_, ok = parse_anomaly_segments(pred)
|
||||
elif cat in ("describe", "forecast"):
|
||||
ok = _extract_json(pred) is not None
|
||||
else: # free-text
|
||||
ok = bool(pred.strip())
|
||||
if ok:
|
||||
parsed += 1
|
||||
if pred.strip():
|
||||
nonempty += 1
|
||||
|
||||
# describe expects a dict ref; parse JSON if possible, else {}
|
||||
ref_for_describe = ref
|
||||
if cat == "describe":
|
||||
try:
|
||||
ref_for_describe = json.loads(ref) if isinstance(ref, str) else (ref or {})
|
||||
if not isinstance(ref_for_describe, dict):
|
||||
ref_for_describe = {}
|
||||
except (ValueError, TypeError):
|
||||
ref_for_describe = {}
|
||||
r = rule_j.judge(
|
||||
cat, pred,
|
||||
ref_segs=[(seg.get("start", 0), seg.get("end", 0))
|
||||
for seg in s.get("segments", [])],
|
||||
T=len(s.get("labels", [])) or 1,
|
||||
ref_answer=ref_for_describe,
|
||||
ref_forecast=_forecast_list(ref) if cat == "forecast" else [],
|
||||
)
|
||||
if r.get("score") is not None:
|
||||
rule_scores.append(r["score"])
|
||||
|
||||
j = llm_j.score(instr, ref, pred)
|
||||
judge_scores.append(j["score"])
|
||||
|
||||
if len(examples) < 2:
|
||||
examples.append({
|
||||
"pred": (pred or "")[:160],
|
||||
"ref": (ref or "")[:160],
|
||||
"parsed": ok,
|
||||
})
|
||||
|
||||
# usability = parsed (structured-or-nonempty) AND judge thinks there's
|
||||
# some content (judge>0). Conservative per-design ">70% usable".
|
||||
usable = sum(1 for sc in judge_scores if sc > 0)
|
||||
report[cat] = {
|
||||
"n": n,
|
||||
"parse_rate": parsed / n,
|
||||
"nonempty_rate": nonempty / n,
|
||||
"rule_acc": float(np.mean(rule_scores)) if rule_scores else None,
|
||||
"judge_mean": float(np.mean(judge_scores)),
|
||||
"usability": usable / n,
|
||||
"examples": examples,
|
||||
}
|
||||
return report
|
||||
|
||||
|
||||
def _forecast_list(ref_answer):
|
||||
"""Extract the forecast value list from a forecast ref answer JSON."""
|
||||
try:
|
||||
obj = json.loads(ref_answer) if isinstance(ref_answer, str) else ref_answer
|
||||
if not isinstance(obj, dict):
|
||||
return []
|
||||
fc = obj.get("forecast")
|
||||
if isinstance(fc, list) and fc:
|
||||
inner = fc[0]
|
||||
if isinstance(inner, list):
|
||||
return [float(x) for x in inner]
|
||||
return [float(x) for x in fc]
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
return []
|
||||
|
||||
|
||||
def run(args: argparse.Namespace) -> Dict:
|
||||
samples_by_cat = load_by_category(args.data, args.per_cat, args.seed)
|
||||
model, col = _build_model(args.stage1_ckpt, args.stage2_ckpt)
|
||||
rep = evaluate(samples_by_cat, model, col, args.judge_backend)
|
||||
|
||||
# headline summary
|
||||
anomaly_parse = rep.get("anomaly", {}).get("parse_rate", 0.0)
|
||||
overall_usability = float(np.mean([r["usability"] for r in rep.values()])) if rep else 0.0
|
||||
summary = {
|
||||
"anomaly_json_parse_rate": anomaly_parse,
|
||||
"anomaly_json_pass": anomaly_parse > 0.80,
|
||||
"overall_usability": overall_usability,
|
||||
"overall_usability_pass": overall_usability > 0.70,
|
||||
}
|
||||
print("\n=== T4.2 instruction-following summary ===")
|
||||
print(json.dumps(summary, indent=2, ensure_ascii=False))
|
||||
print("\n=== per-category ===")
|
||||
for cat, r in rep.items():
|
||||
print(f"[{cat}] n={r['n']} parse={r['parse_rate']:.2f} "
|
||||
f"rule={r['rule_acc']} judge={r['judge_mean']:.2f} "
|
||||
f"usable={r['usability']:.2f}")
|
||||
for ex in r["examples"]:
|
||||
print(f" pred={ex['pred']!r}")
|
||||
return {"summary": summary, "per_category": rep}
|
||||
|
||||
|
||||
def main() -> None:
|
||||
p = argparse.ArgumentParser(description="Instruction-following check (T4.2)")
|
||||
p.add_argument("--data", default="data/eval_synth.jsonl")
|
||||
p.add_argument("--per_cat", type=int, default=10)
|
||||
p.add_argument("--seed", type=int, default=0)
|
||||
p.add_argument("--stage1_ckpt", default="checkpoints/stage1/stage1_final.pt")
|
||||
p.add_argument("--stage2_ckpt", default="checkpoints/stage2/stage2_final.pt")
|
||||
p.add_argument("--judge_backend", default="stub")
|
||||
p.add_argument("--out", default="reports/instruct_check.json")
|
||||
args = p.parse_args()
|
||||
result = run(args)
|
||||
os.makedirs(os.path.dirname(args.out) or ".", exist_ok=True)
|
||||
with open(args.out, "w", encoding="utf-8") as f:
|
||||
json.dump(result, f, indent=2, ensure_ascii=False)
|
||||
print(f"\n[instruct_check] wrote {args.out}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,151 @@
|
||||
"""Parse model text answers → anomaly segments → point-wise scores (T5.1).
|
||||
|
||||
Two-stage:
|
||||
1. :func:`parse_anomaly_segments` — extract ``[start, end]`` intervals from a
|
||||
model's free-form answer. First tries strict JSON (a blob with a
|
||||
``segments`` key); if that fails, falls back to regex-extracting
|
||||
``[int, int]`` patterns. Returns ``(segments, parsed_ok)`` where
|
||||
``parsed_ok`` is True only when a valid JSON ``segments`` field was found.
|
||||
2. :func:`answer_to_point_scores` — map segments to a ``[T]`` 0/1 array; on
|
||||
parse failure the array is all-zero (counted as total miss downstream).
|
||||
|
||||
Design ref: §5.2 (steps 1-5). Robustness: parsing must not crash on any input.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from typing import List, Tuple
|
||||
|
||||
import numpy as np
|
||||
|
||||
# match [int, int] or [[int, int]] style intervals
|
||||
_PAIR_RE = re.compile(r"\[\s*(\d+)\s*,\s*(\d+)\s*\]")
|
||||
# find JSON-ish objects in text (flat blobs only; nested handled by scanner)
|
||||
_JSON_RE = re.compile(r"\{[^{}]*\}", re.DOTALL)
|
||||
|
||||
|
||||
def _extract_balanced_json(text: str) -> List[str]:
|
||||
"""Yield every balanced ``{...}`` substring (handles nesting).
|
||||
|
||||
Scans the text for ``{`` and tracks depth, emitting the span for each
|
||||
top-level balanced object. This lets us parse nested JSON the model emits
|
||||
(e.g. ``{"anomaly_regions": [{"start": 10, "end": 93}]}``) which the
|
||||
flat ``_JSON_RE`` cannot match.
|
||||
"""
|
||||
out = []
|
||||
depth = 0
|
||||
start = -1
|
||||
instr = False
|
||||
esc = False
|
||||
quote = ""
|
||||
for i, ch in enumerate(text):
|
||||
if esc:
|
||||
esc = False
|
||||
continue
|
||||
if ch == "\\":
|
||||
esc = True
|
||||
continue
|
||||
if instr:
|
||||
if ch == quote:
|
||||
instr = False
|
||||
continue
|
||||
if ch in ("'", '"'):
|
||||
instr = True
|
||||
quote = ch
|
||||
continue
|
||||
if ch == "{":
|
||||
if depth == 0:
|
||||
start = i
|
||||
depth += 1
|
||||
elif ch == "}":
|
||||
if depth > 0:
|
||||
depth -= 1
|
||||
if depth == 0 and start >= 0:
|
||||
out.append(text[start : i + 1])
|
||||
start = -1
|
||||
return out
|
||||
|
||||
|
||||
def _normalize_seg(pair) -> Tuple[int, int]:
|
||||
a, b = int(pair[0]), int(pair[1])
|
||||
if a > b:
|
||||
a, b = b, a
|
||||
return (a, b)
|
||||
|
||||
|
||||
def _seg_from_entry(s):
|
||||
"""Extract a (start, end) tuple from one segment entry.
|
||||
|
||||
Accepts either ``[start, end]`` lists/tuples or ``{"start": s, "end": e}``
|
||||
objects (the schema emitted by ``data/instruct.py``). Returns None if no
|
||||
usable pair is found, or if the values aren't ints (e.g. a pure-LLM emits
|
||||
``["CPU Usage", 30]`` — junk that must not crash the eval pipeline).
|
||||
"""
|
||||
try:
|
||||
if isinstance(s, dict):
|
||||
if "start" in s and "end" in s:
|
||||
return (int(s["start"]), int(s["end"]))
|
||||
return None
|
||||
if isinstance(s, (list, tuple)):
|
||||
if len(s) >= 2:
|
||||
return _normalize_seg(s)
|
||||
if len(s) == 1:
|
||||
v = int(s[0]); return (v, v)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def parse_anomaly_segments(text: str) -> Tuple[List[Tuple[int, int]], bool]:
|
||||
"""Return (segments, json_ok).
|
||||
|
||||
json_ok=True iff a JSON object with a recognized segments key was found.
|
||||
Recognized keys: ``segments`` (list of [s,e]) and ``anomaly_regions``
|
||||
(list of {"start":s,"end":e} — the schema used by ``data/instruct.py``).
|
||||
"""
|
||||
if text is None:
|
||||
return [], False
|
||||
# 1. try every balanced {...} blob; prefer one with a recognized segments key
|
||||
candidates = _extract_balanced_json(text)
|
||||
best_segments = None
|
||||
for blob in candidates:
|
||||
try:
|
||||
obj = json.loads(blob)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
if not isinstance(obj, dict):
|
||||
continue
|
||||
raw = None
|
||||
if "segments" in obj:
|
||||
raw = obj["segments"]
|
||||
elif "anomaly_regions" in obj:
|
||||
raw = obj["anomaly_regions"]
|
||||
if raw is None or not isinstance(raw, list):
|
||||
continue
|
||||
out = [_seg_from_entry(s) for s in raw]
|
||||
out = [seg for seg in out if seg is not None]
|
||||
best_segments = out
|
||||
break
|
||||
if best_segments is not None:
|
||||
return best_segments, True
|
||||
|
||||
# 2. fallback: regex-extract [int,int] pairs (no valid JSON key found)
|
||||
pairs = _PAIR_RE.findall(text)
|
||||
if pairs:
|
||||
return [_normalize_seg(p) for p in pairs], False
|
||||
return [], False
|
||||
|
||||
|
||||
def answer_to_point_scores(text: str, T: int) -> Tuple[np.ndarray, bool]:
|
||||
"""Map answer → [T] 0/1 point scores. All-zero on parse failure."""
|
||||
scores = np.zeros(T, dtype=np.float32)
|
||||
if T <= 0:
|
||||
return scores, False
|
||||
segments, ok = parse_anomaly_segments(text)
|
||||
for (a, b) in segments:
|
||||
a = max(0, min(a, T - 1))
|
||||
b = max(0, min(b, T - 1))
|
||||
scores[a : b + 1] = 1.0
|
||||
return scores, ok
|
||||
@@ -0,0 +1,201 @@
|
||||
"""QA judging (T5.3): rule track (JSON/number tolerance) + LLM-judge track (0-5).
|
||||
|
||||
- Rule track: objective, for anomaly/describe/forecast tasks. JSON region IoU,
|
||||
numerical relative-tolerance match. Zero cost, fully reproducible.
|
||||
- LLM-judge track: subjective, for root_cause/compare/event. Scores 0-5 via an
|
||||
LLM (GPT-4o in production). Because this host has no network, the default
|
||||
``backend="stub"`` produces a deterministic heuristic score (exact/substring
|
||||
match → high; dissimilar → low) so the pipeline is runnable end-to-end offline;
|
||||
swap to ``backend="openai"`` (or pass a callable) for real judging.
|
||||
|
||||
Design ref: §5.3 (双轨) and §6.2 (GPT-4o judge, eval-only).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from typing import Any, Callable, Dict, List, Optional, Tuple
|
||||
|
||||
from .parse_answer import _extract_balanced_json, parse_anomaly_segments
|
||||
|
||||
|
||||
_JSON_RE = re.compile(r"\{[^{}]*\}", re.DOTALL)
|
||||
|
||||
|
||||
def _extract_json(pred: str) -> Optional[dict]:
|
||||
if pred is None:
|
||||
return None
|
||||
# prefer balanced (nested-capable) extraction first
|
||||
for blob in _extract_balanced_json(pred):
|
||||
try:
|
||||
obj = json.loads(blob)
|
||||
if isinstance(obj, dict):
|
||||
return obj
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
# fallback to flat blobs
|
||||
for blob in _JSON_RE.findall(pred):
|
||||
try:
|
||||
return json.loads(blob)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
# ─── rule-track per-category scorers ──────────────────────────────────────
|
||||
|
||||
def rule_score_describe(pred: str, ref: dict, rtol: float = 0.05) -> Dict[str, Any]:
|
||||
"""Compare numerical statistics with relative tolerance.
|
||||
|
||||
``ref`` is a dict of {stat_name: value}. A stat scores 1 if |pred-ref|/|ref|
|
||||
<= rtol (or both zero). Overall score = fraction of matched stats.
|
||||
"""
|
||||
obj = _extract_json(pred)
|
||||
if obj is None:
|
||||
return {"score": 0.0, "parsed": False}
|
||||
if not ref:
|
||||
return {"score": 1.0 if not obj else 0.0, "parsed": True}
|
||||
hits = 0
|
||||
total = 0
|
||||
for k, rv in ref.items():
|
||||
total += 1
|
||||
pv = obj.get(k)
|
||||
if pv is None:
|
||||
continue
|
||||
try:
|
||||
pv = float(pv); rv = float(rv)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if rv == 0 and pv == 0:
|
||||
hits += 1
|
||||
elif rv != 0 and abs(pv - rv) / abs(rv) <= rtol:
|
||||
hits += 1
|
||||
return {"score": hits / total, "parsed": True}
|
||||
|
||||
|
||||
def rule_score_anomaly(pred: str, ref_segs: List[Tuple[int, int]], T: int) -> Dict[str, Any]:
|
||||
"""Anomaly region IoU as the score."""
|
||||
pred_segs, ok = parse_anomaly_segments(pred)
|
||||
if not ok and not pred_segs:
|
||||
return {"score": 0.0, "parsed": False}
|
||||
# IoU over point sets
|
||||
def to_set(segs):
|
||||
s = set()
|
||||
for a, b in segs:
|
||||
a = max(0, a); b = min(T - 1, b)
|
||||
s.update(range(a, b + 1))
|
||||
return s
|
||||
ps, rs = to_set(pred_segs), to_set(ref_segs)
|
||||
if not ps and not rs:
|
||||
return {"score": 1.0, "parsed": True}
|
||||
inter = len(ps & rs)
|
||||
union = len(ps | rs)
|
||||
return {"score": inter / union if union else 0.0, "parsed": True}
|
||||
|
||||
|
||||
def rule_score_forecast(pred: str, ref: List[float], rtol: float = 0.1) -> Dict[str, Any]:
|
||||
obj = _extract_json(pred)
|
||||
if obj is None:
|
||||
return {"score": 0.0, "parsed": False}
|
||||
vals = obj.get("forecast")
|
||||
if not isinstance(vals, list):
|
||||
return {"score": 0.0, "parsed": True}
|
||||
try:
|
||||
preds = [float(v) for v in vals]
|
||||
except (TypeError, ValueError):
|
||||
return {"score": 0.0, "parsed": True}
|
||||
n = min(len(preds), len(ref))
|
||||
if n == 0:
|
||||
return {"score": 0.0, "parsed": True}
|
||||
hits = 0
|
||||
for p, r in zip(preds[:n], ref[:n]):
|
||||
if r == 0 and p == 0:
|
||||
hits += 1
|
||||
elif r != 0 and abs(p - r) / abs(r) <= rtol:
|
||||
hits += 1
|
||||
return {"score": hits / len(ref) if ref else 0.0, "parsed": True}
|
||||
|
||||
|
||||
# ─── rule-track dispatcher ──────────────────────────────────────────────────
|
||||
|
||||
class RuleJudge:
|
||||
"""Dispatch to the right rule scorer by category."""
|
||||
|
||||
def judge(self, category: str, pred: str, **ref) -> Dict[str, Any]:
|
||||
if category == "anomaly":
|
||||
out = rule_score_anomaly(pred, ref.get("ref_segs", []), ref.get("T", 1))
|
||||
elif category == "describe":
|
||||
out = rule_score_describe(pred, ref.get("ref_answer", {}))
|
||||
elif category == "forecast":
|
||||
out = rule_score_forecast(pred, ref.get("ref_forecast", []))
|
||||
else:
|
||||
# root_cause / compare / event → not rule-checkable
|
||||
return {"track": "skip", "score": None, "parsed": False}
|
||||
return {"track": "rule", **out}
|
||||
|
||||
|
||||
# ─── LLM-judge track (0-5) ──────────────────────────────────────────────────
|
||||
|
||||
class LLMJudge:
|
||||
"""Score free-text answers 0-5.
|
||||
|
||||
``backend``:
|
||||
- ``"stub"`` (default): deterministic offline heuristic (no network).
|
||||
- ``"openai"``: real GPT-4o judge (requires OPENAI_API_KEY + network).
|
||||
- a callable ``(question, ref, pred) -> {"score": float, "reason": str}``.
|
||||
"""
|
||||
|
||||
def __init__(self, backend: str = "stub", model: str = "gpt-4o",
|
||||
api_key: Optional[str] = None, base_url: Optional[str] = None) -> None:
|
||||
self.backend = backend
|
||||
self.model = model
|
||||
self.api_key = api_key
|
||||
self.base_url = base_url
|
||||
|
||||
def score(self, question: str, ref_answer: str, pred_answer: str) -> Dict[str, Any]:
|
||||
if callable(self.backend):
|
||||
return self.backend(question, ref_answer, pred_answer)
|
||||
if self.backend == "stub":
|
||||
return self._stub_score(question, ref_answer, pred_answer)
|
||||
if self.backend == "openai":
|
||||
return self._openai_score(question, ref_answer, pred_answer)
|
||||
raise ValueError(f"unknown backend: {self.backend}")
|
||||
|
||||
def _stub_score(self, question, ref, pred) -> Dict[str, Any]:
|
||||
# Offline heuristic: exact/substring/token-overlap similarity → 0-5.
|
||||
ref = (ref or "").strip().lower()
|
||||
pred = (pred or "").strip().lower()
|
||||
if not ref and not pred:
|
||||
return {"score": 5.0, "reason": "stub: both empty", "backend": "stub"}
|
||||
if ref == pred:
|
||||
return {"score": 5.0, "reason": "stub: exact match", "backend": "stub"}
|
||||
rt, pt = set(ref.split()), set(pred.split())
|
||||
if not rt:
|
||||
return {"score": 1.0, "reason": "stub: empty ref", "backend": "stub"}
|
||||
overlap = len(rt & pt) / len(rt)
|
||||
score = round(5.0 * overlap, 1)
|
||||
return {"score": score, "reason": f"stub: token-overlap={overlap:.2f}", "backend": "stub"}
|
||||
|
||||
def _openai_score(self, question, ref, pred) -> Dict[str, Any]: # pragma: no cover
|
||||
import os
|
||||
from openai import OpenAI
|
||||
key = self.api_key or os.environ.get("OPENAI_API_KEY")
|
||||
client = OpenAI(api_key=key, base_url=self.base_url)
|
||||
prompt = (
|
||||
"You are a strict grader. Score the model's answer 0-5.\n"
|
||||
f"Question: {question}\nReference answer: {ref}\nModel answer: {pred}\n"
|
||||
'Reply JSON: {"score": <0-5>, "reason": "..."}'
|
||||
)
|
||||
resp = client.chat.completions.create(
|
||||
model=self.model,
|
||||
messages=[{"role": "user", "content": prompt}],
|
||||
temperature=0.0,
|
||||
)
|
||||
text = resp.choices[0].message.content
|
||||
try:
|
||||
obj = json.loads(text)
|
||||
return {"score": float(obj["score"]), "reason": obj.get("reason", ""),
|
||||
"backend": "openai"}
|
||||
except (json.JSONDecodeError, KeyError):
|
||||
return {"score": 0.0, "reason": f"parse-fail: {text[:100]}",
|
||||
"backend": "openai"}
|
||||
@@ -0,0 +1,367 @@
|
||||
"""Evaluation runner + markdown report (T5.5).
|
||||
|
||||
Loads eval JSONL files (synth + real), runs the trained model and all baselines
|
||||
through the SAME parse_answer + ts_metrics + qa_judge pipeline, and emits a
|
||||
markdown report ``reports/eval-YYYYMMDD.md`` covering the design §5.4 必报清单:
|
||||
|
||||
- per dataset × per task: VUS-PR (main), Aff-F1, AUC-PR, Point-F1 (no PA),
|
||||
PA-F1 (control, explicitly labelled), rule accuracy, judge mean,
|
||||
parse-success rate
|
||||
- ablation (no-TS) and baseline (pure-LLM, Time-LLM, ChatTS, trivial) tables
|
||||
- explicit "uses PA? no" annotation and trivial baselines included
|
||||
|
||||
Heavy GPU pieces (model.generate, pure_llm_answer) are lazy so the runner can
|
||||
also run a CPU/trivial-only smoke.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from collections import defaultdict
|
||||
from datetime import date
|
||||
from typing import Any, Dict, List
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .baselines import (
|
||||
AllReportBaseline,
|
||||
ConstantBaseline,
|
||||
RandomBaseline,
|
||||
ts_to_text,
|
||||
)
|
||||
from .parse_answer import answer_to_point_scores
|
||||
from .qa_judge import LLMJudge, RuleJudge
|
||||
from .ts_metrics import (
|
||||
affiliation_f1,
|
||||
auc_pr,
|
||||
point_f1,
|
||||
point_f1_with_pa,
|
||||
vus_pr,
|
||||
)
|
||||
|
||||
|
||||
def stream_jsonl(path: str):
|
||||
if not os.path.exists(path):
|
||||
return
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if line:
|
||||
yield json.loads(line)
|
||||
|
||||
|
||||
def labels_to_array(labels, T):
|
||||
arr = np.zeros(T, dtype=int)
|
||||
for i, v in enumerate(labels[:T]):
|
||||
if v:
|
||||
arr[i] = 1
|
||||
return arr
|
||||
|
||||
|
||||
def eval_anomaly_task(samples, predictor, T_default=512) -> Dict[str, Any]:
|
||||
"""Evaluate anomaly detection on samples. ``predictor(sample) -> answer_text``."""
|
||||
vus, aff, aucp, pf1, paf1 = [], [], [], [], []
|
||||
parse_ok = 0
|
||||
for s in samples:
|
||||
labels = s.get("labels") or []
|
||||
T = len(labels) or T_default
|
||||
y = labels_to_array(labels, T)
|
||||
pred_text = predictor(s)
|
||||
scores, ok = answer_to_point_scores(pred_text, T)
|
||||
if ok:
|
||||
parse_ok += 1
|
||||
vus.append(vus_pr(scores, y))
|
||||
f1, _, _ = point_f1(scores, y)
|
||||
pf1.append(f1)
|
||||
af1, _, _ = affiliation_f1(scores, y)
|
||||
aff.append(af1)
|
||||
aucp.append(auc_pr(scores, y))
|
||||
pa, _, _ = point_f1_with_pa(scores, y)
|
||||
paf1.append(pa)
|
||||
n = max(len(samples), 1)
|
||||
return {
|
||||
"n": len(samples),
|
||||
"parse_rate": parse_ok / n,
|
||||
"VUS-PR": float(np.mean(vus)) if vus else 0.0,
|
||||
"Aff-F1": float(np.mean(aff)) if aff else 0.0,
|
||||
"AUC-PR": float(np.mean(aucp)) if aucp else 0.0,
|
||||
"Point-F1": float(np.mean(pf1)) if pf1 else 0.0,
|
||||
"PA-F1(control)": float(np.mean(paf1)) if paf1 else 0.0,
|
||||
}
|
||||
|
||||
|
||||
def eval_qa_task(samples, predictor, judge_backend="stub") -> Dict[str, Any]:
|
||||
"""Evaluate free-text QA via rule + LLM-judge double track."""
|
||||
rule_judge = RuleJudge()
|
||||
llm_judge = LLMJudge(backend=judge_backend)
|
||||
rule_scores, judge_scores, parse_ok = [], [], 0
|
||||
for s in samples:
|
||||
cat = s.get("category", "root_cause")
|
||||
pred = predictor(s)
|
||||
ref = s.get("answer", "")
|
||||
# describe expects a dict ref; parse JSON if possible, else {}
|
||||
ref_for_describe = ref
|
||||
if cat == "describe":
|
||||
try:
|
||||
ref_for_describe = json.loads(ref) if isinstance(ref, str) else (ref or {})
|
||||
if not isinstance(ref_for_describe, dict):
|
||||
ref_for_describe = {}
|
||||
except (ValueError, TypeError):
|
||||
ref_for_describe = {}
|
||||
r = rule_judge.judge(cat, pred,
|
||||
ref_segs=[(seg.get("start", 0), seg.get("end", 0))
|
||||
for seg in s.get("segments", [])],
|
||||
T=len(s.get("labels", [])) or 1,
|
||||
ref_answer=ref_for_describe)
|
||||
if r.get("parsed"):
|
||||
parse_ok += 1
|
||||
if r.get("score") is not None:
|
||||
rule_scores.append(r["score"])
|
||||
j = llm_judge.score(s.get("instruction", ""), ref, pred)
|
||||
judge_scores.append(j["score"])
|
||||
n = max(len(samples), 1)
|
||||
return {
|
||||
"n": len(samples),
|
||||
"parse_rate": parse_ok / n,
|
||||
"rule_acc": float(np.mean(rule_scores)) if rule_scores else None,
|
||||
"judge_mean": float(np.mean(judge_scores)) if judge_scores else 0.0,
|
||||
}
|
||||
|
||||
|
||||
def trivial_predictor(baseline):
|
||||
"""A predictor that returns a text answer encoding the trivial baseline's
|
||||
point scores (so it flows through the same parse pipeline)."""
|
||||
def _pred(sample):
|
||||
T = len(sample.get("labels", [])) or 512
|
||||
scores = baseline.score(T)
|
||||
# turn scores into segments at threshold 0.5 for the parse pipeline
|
||||
segs = []
|
||||
in_seg = False; start = 0
|
||||
for i, v in enumerate(scores):
|
||||
if v >= 0.5 and not in_seg:
|
||||
in_seg = True; start = i
|
||||
elif v < 0.5 and in_seg:
|
||||
in_seg = False; segs.append([start, i - 1])
|
||||
if in_seg:
|
||||
segs.append([start, T - 1])
|
||||
return json.dumps({"segments": segs})
|
||||
return _pred
|
||||
|
||||
|
||||
def run_eval(args: argparse.Namespace) -> str:
|
||||
os.makedirs(args.report_dir, exist_ok=True)
|
||||
datasets = []
|
||||
if args.synth and os.path.exists(args.synth):
|
||||
datasets.append(("synth", args.synth))
|
||||
if args.real and os.path.exists(args.real):
|
||||
datasets.append(("real", args.real))
|
||||
|
||||
# predictors: trained model (optional) + trivial baselines + pure-LLM
|
||||
predictors: Dict[str, Any] = {
|
||||
"random": trivial_predictor(RandomBaseline(seed=0)),
|
||||
"constant": trivial_predictor(ConstantBaseline(0.5)),
|
||||
"all_report": trivial_predictor(AllReportBaseline()),
|
||||
}
|
||||
|
||||
# Heavy predictors are built & evaluated one at a time with GPU cleanup so
|
||||
# we never hold two full models on a single 12GB card.
|
||||
heavy_builders: Dict[str, Any] = {} # name -> zero-arg callable returning predictor fn
|
||||
if args.model:
|
||||
def _build_trained():
|
||||
import torch
|
||||
from ..model.ts_encoder import TSEncoder
|
||||
from ..model.projector import Projector
|
||||
from ..model.wrapper import MultimodalTSModel
|
||||
from ..data.collator import Collator
|
||||
|
||||
LLM = os.environ.get("TSMM_LLM_PATH", "/home/zhangzp/models/Qwen2.5-0.5B-Instruct")
|
||||
m = MultimodalTSModel(LLM, TSEncoder(), Projector(), dtype=torch.bfloat16).cuda()
|
||||
if args.stage2_ckpt and os.path.exists(args.stage2_ckpt):
|
||||
st = torch.load(args.stage2_ckpt, map_location="cpu")
|
||||
m.encoder.load_state_dict(st["encoder"])
|
||||
m.projector.load_state_dict(st["projector"])
|
||||
lora_dir = os.path.join(os.path.dirname(args.stage2_ckpt), "lora_adapter")
|
||||
if os.path.exists(lora_dir):
|
||||
from peft import PeftModel
|
||||
m.llm = PeftModel.from_pretrained(m.llm, lora_dir)
|
||||
elif args.stage1_ckpt and os.path.exists(args.stage1_ckpt):
|
||||
st = torch.load(args.stage1_ckpt, map_location="cpu")
|
||||
m.encoder.load_state_dict(st["encoder"])
|
||||
m.projector.load_state_dict(st["projector"])
|
||||
m.eval()
|
||||
col = Collator(max_T=512)
|
||||
|
||||
def _pred(sample, _m=m, _col=col):
|
||||
batch = _col([sample])
|
||||
with torch.no_grad():
|
||||
txt = _m.generate(batch, max_new_tokens=192)
|
||||
return txt[0] if txt else ""
|
||||
return _pred
|
||||
|
||||
def _build_pure_llm():
|
||||
# base LLM, series fed as TEXT (no TS modality, no LoRA) — design §5.4
|
||||
import torch
|
||||
from ..model.ts_encoder import TSEncoder
|
||||
from ..model.projector import Projector
|
||||
from ..model.wrapper import MultimodalTSModel
|
||||
LLM = os.environ.get("TSMM_LLM_PATH", "/home/zhangzp/models/Qwen2.5-0.5B-Instruct")
|
||||
m = MultimodalTSModel(LLM, TSEncoder(), Projector(), dtype=torch.bfloat16).cuda()
|
||||
m.eval()
|
||||
tok = m.tokenizer
|
||||
|
||||
def _pred(sample, _m=m, _tok=tok):
|
||||
from .baselines import pure_llm_answer
|
||||
return pure_llm_answer(
|
||||
sample.get("series", []),
|
||||
sample.get("instruction") or sample.get("question") or "",
|
||||
model=_m, max_new_tokens=192)
|
||||
return _pred
|
||||
|
||||
heavy_builders["model"] = _build_trained
|
||||
if args.pure_llm:
|
||||
heavy_builders["pure_llm"] = _build_pure_llm
|
||||
|
||||
# run anomaly eval per dataset × predictor
|
||||
results: Dict[str, Dict[str, Any]] = defaultdict(dict)
|
||||
qa_results: Dict[str, Dict[str, Dict[str, Any]]] = defaultdict(lambda: defaultdict(dict))
|
||||
# First pass: cache per-dataset splits so heavy predictors reuse them.
|
||||
splits: Dict[str, Dict[str, List[dict]]] = {}
|
||||
maxpc = args.max_per_cat
|
||||
for ds_name, path in datasets:
|
||||
samples = list(stream_jsonl(path))
|
||||
by_cat: Dict[str, List[dict]] = defaultdict(list)
|
||||
for s in samples:
|
||||
by_cat[s.get("category", "?")].append(s)
|
||||
if maxpc and maxpc > 0:
|
||||
for cat in list(by_cat.keys()):
|
||||
by_cat[cat] = by_cat[cat][:maxpc]
|
||||
splits[ds_name] = by_cat
|
||||
|
||||
def _eval_predictor(pname, pred):
|
||||
for ds_name, by_cat in splits.items():
|
||||
anom = by_cat.get("anomaly", [])
|
||||
results[ds_name][pname] = eval_anomaly_task(anom, pred)
|
||||
for cat, cat_samples in by_cat.items():
|
||||
qa_results[ds_name][pname][cat] = eval_qa_task(
|
||||
cat_samples, pred, judge_backend=args.judge_backend)
|
||||
|
||||
# trivial (CPU) predictors first
|
||||
all_pred_names = list(predictors.keys())
|
||||
for pname, pred in predictors.items():
|
||||
_eval_predictor(pname, pred)
|
||||
|
||||
# heavy predictors one at a time, with GPU cleanup between
|
||||
if heavy_builders:
|
||||
import torch
|
||||
import gc
|
||||
for pname, builder in heavy_builders.items():
|
||||
print(f"[run_eval] building heavy predictor: {pname}")
|
||||
pred = builder()
|
||||
_eval_predictor(pname, pred)
|
||||
all_pred_names.append(pname)
|
||||
# free GPU before the next heavy model
|
||||
del pred
|
||||
gc.collect()
|
||||
torch.cuda.empty_cache()
|
||||
|
||||
predictor_order = [n for n in ["model", "pure_llm"] if n in all_pred_names]
|
||||
for n in all_pred_names:
|
||||
if n not in predictor_order:
|
||||
predictor_order.append(n)
|
||||
|
||||
# render markdown
|
||||
return _render_report(results, qa_results, datasets, predictor_order, args)
|
||||
|
||||
|
||||
def _render_report(results, qa_results, datasets, predictor_names, args) -> str:
|
||||
today = date.today().isoformat()
|
||||
lines = [
|
||||
f"# Eval Report — {today}",
|
||||
"",
|
||||
"Generated by `scripts/run_eval.sh` → `tsmm.eval.report`.",
|
||||
"",
|
||||
"## Discipline notes",
|
||||
"- **Main metric: VUS-PR** (threshold-free).",
|
||||
"- **Point-F1 reported WITHOUT point-adjust.** PA-F1 shown separately as a CONTROL only.",
|
||||
"- Trivial baselines (random / constant / all-report) are included.",
|
||||
f"- LLM-judge backend: `{args.judge_backend}`.",
|
||||
"",
|
||||
]
|
||||
for ds_name, _ in datasets:
|
||||
lines.append(f"## Dataset: {ds_name}")
|
||||
lines.append("")
|
||||
ds = results.get(ds_name, {})
|
||||
if not ds:
|
||||
lines.append("_no samples_\n")
|
||||
continue
|
||||
header = ["predictor", "n", "VUS-PR", "Aff-F1", "AUC-PR",
|
||||
"Point-F1", "PA-F1(control)", "parse_rate"]
|
||||
lines.append("| " + " | ".join(header) + " |")
|
||||
lines.append("|" + "|".join(["---"] * len(header)) + "|")
|
||||
for pname in predictor_names:
|
||||
r = ds.get(pname)
|
||||
if not r:
|
||||
continue
|
||||
row = [pname, str(r["n"]),
|
||||
f"{r['VUS-PR']:.4f}", f"{r['Aff-F1']:.4f}",
|
||||
f"{r['AUC-PR']:.4f}", f"{r['Point-F1']:.4f}",
|
||||
f"{r['PA-F1(control)']:.4f}", f"{r['parse_rate']:.2f}"]
|
||||
lines.append("| " + " | ".join(row) + " |")
|
||||
lines.append("")
|
||||
lines.append("_PA = point-adjust; shown as control only, not headline._")
|
||||
lines.append("")
|
||||
|
||||
# QA judge table (covers design §6.5 success criterion 问答均分)
|
||||
qa_ds = qa_results.get(ds_name, {})
|
||||
if qa_ds:
|
||||
cats = sorted({c for pm in qa_ds.values() for c in pm})
|
||||
lines.append(f"### QA judge (stub 0-5) — {ds_name}")
|
||||
lines.append("")
|
||||
header = ["predictor"] + cats + ["mean"]
|
||||
lines.append("| " + " | ".join(header) + " |")
|
||||
lines.append("|" + "|".join(["---"] * len(header)) + "|")
|
||||
for pname in predictor_names:
|
||||
pm = qa_ds.get(pname, {})
|
||||
if not pm:
|
||||
continue
|
||||
row_vals = []
|
||||
for c in cats:
|
||||
jm = pm.get(c, {}).get("judge_mean")
|
||||
row_vals.append(f"{jm:.2f}" if jm is not None else "-")
|
||||
mean_jm = float(np.mean([pm[c]["judge_mean"]
|
||||
for c in cats
|
||||
if pm.get(c, {}).get("judge_mean") is not None])) if any(pm.get(c) for c in cats) else 0.0
|
||||
row = [pname] + row_vals + [f"{mean_jm:.2f}"]
|
||||
lines.append("| " + " | ".join(row) + " |")
|
||||
lines.append("")
|
||||
lines.append("_QA judge = offline stub token-overlap heuristic; "
|
||||
"swap `--judge_backend openai` for GPT-4o scoring._")
|
||||
lines.append("")
|
||||
|
||||
out_path = os.path.join(args.report_dir, f"eval-{today}.md")
|
||||
with open(out_path, "w", encoding="utf-8") as f:
|
||||
f.write("\n".join(lines))
|
||||
print(f"[run_eval] report -> {out_path}")
|
||||
return out_path
|
||||
|
||||
|
||||
def main() -> None:
|
||||
p = argparse.ArgumentParser(description="Run the full evaluation")
|
||||
p.add_argument("--synth", default="data/eval_synth.jsonl")
|
||||
p.add_argument("--real", default="data/eval_real.jsonl")
|
||||
p.add_argument("--model", action="store_true", help="include the trained model")
|
||||
p.add_argument("--pure_llm", action="store_true",
|
||||
help="include the pure-LLM baseline (series as text, no TS modality)")
|
||||
p.add_argument("--stage1_ckpt", default="checkpoints/stage1/stage1_final.pt")
|
||||
p.add_argument("--stage2_ckpt", default="checkpoints/stage2/stage2_final.pt")
|
||||
p.add_argument("--judge_backend", default="stub")
|
||||
p.add_argument("--max_per_cat", type=int, default=0,
|
||||
help="cap samples per category for eval (0 = use all)")
|
||||
p.add_argument("--report_dir", default="reports")
|
||||
args = p.parse_args()
|
||||
run_eval(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,145 @@
|
||||
"""TS-token effectiveness check (T3.3).
|
||||
|
||||
Validates that the model actually USES the time-series modality (design §4.4 /
|
||||
success criterion #3: "时序消融后指标明显下降"):
|
||||
|
||||
1. **Perturbation sensitivity**: for held-out samples, generate an answer on the
|
||||
original series vs a STRONGLY perturbed series; measure the answer-change
|
||||
rate. Target: > 50% (design T3.3). Uses perturbations larger than the light
|
||||
augmentations used as InfoNCE positives in stage ①.
|
||||
2. **Must-see-TS accuracy**: ask questions whose answer depends on specific
|
||||
series content (e.g. "what is the value at t=300?"); compare the trained
|
||||
model vs a pure-LLM baseline (TS-as-text). Target: trained >> pure-LLM.
|
||||
|
||||
Heavy (GPU, model.generate) — import the wrapper lazily.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from typing import Dict, List
|
||||
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
|
||||
def load_samples(path: str, n: int) -> List[dict]:
|
||||
out = []
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if line:
|
||||
out.append(json.loads(line))
|
||||
if len(out) >= n:
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def perturb_series_hard(series: List[List[float]], seed: int = 0, std_mult: float = 3.0) -> List[List[float]]:
|
||||
"""Strong perturbation: add large noise + shuffle phases — much larger than
|
||||
the light augmentations used as InfoNCE positives."""
|
||||
rng = np.random.default_rng(seed)
|
||||
arr = np.array(series, dtype=np.float64)
|
||||
arr = np.nan_to_num(arr, nan=0.0)
|
||||
scale = max(arr.std(), 1.0) * std_mult
|
||||
return (arr + rng.normal(0, scale, arr.shape)).tolist()
|
||||
|
||||
|
||||
def _build_model(stage1_ckpt=None, stage2_ckpt=None):
|
||||
from ..model.ts_encoder import TSEncoder
|
||||
from ..model.projector import Projector
|
||||
from ..model.wrapper import MultimodalTSModel
|
||||
from ..data.collator import Collator
|
||||
|
||||
llm = os.environ.get("TSMM_LLM_PATH", "/home/zhangzp/models/Qwen2.5-0.5B-Instruct")
|
||||
m = MultimodalTSModel(llm, TSEncoder(), Projector(), dtype=torch.bfloat16).cuda()
|
||||
ckpt = stage2_ckpt or stage1_ckpt
|
||||
if ckpt and os.path.exists(ckpt):
|
||||
st = torch.load(ckpt, map_location="cpu")
|
||||
m.encoder.load_state_dict(st["encoder"])
|
||||
m.projector.load_state_dict(st["projector"])
|
||||
print(f"[ts_effectiveness] loaded {ckpt}")
|
||||
lora_dir = os.path.join(os.path.dirname(ckpt), "lora_adapter")
|
||||
if stage2_ckpt and os.path.exists(lora_dir):
|
||||
from peft import PeftModel
|
||||
m.llm = PeftModel.from_pretrained(m.llm, lora_dir)
|
||||
print(f"[ts_effectiveness] loaded lora {lora_dir}")
|
||||
m.eval()
|
||||
return m, Collator(max_T=512)
|
||||
|
||||
|
||||
def _generate(m, col, sample) -> str:
|
||||
batch = col([sample])
|
||||
with torch.no_grad():
|
||||
txt = m.generate(batch, max_new_tokens=48)
|
||||
return txt[0] if txt else ""
|
||||
|
||||
|
||||
def answers_differ(a: str, b: str) -> bool:
|
||||
"""Normalized text inequality (whitespace-insensitive)."""
|
||||
na = " ".join((a or "").strip().lower().split())
|
||||
nb = " ".join((b or "").strip().lower().split())
|
||||
return na != nb and bool(na) and bool(nb)
|
||||
|
||||
|
||||
def perturbation_sensitivity(samples, model, col, seed: int = 0) -> Dict[str, float]:
|
||||
changed = 0; n = 0
|
||||
for i, s in enumerate(samples):
|
||||
ans_orig = _generate(model, col, s)
|
||||
s2 = dict(s)
|
||||
s2["series"] = perturb_series_hard(s["series"], seed=seed + i)
|
||||
ans_pert = _generate(model, col, s2)
|
||||
if answers_differ(ans_orig, ans_pert):
|
||||
changed += 1
|
||||
n += 1
|
||||
rate = changed / max(n, 1)
|
||||
return {"change_rate": rate, "n": n, "target": 0.5, "pass": rate > 0.5}
|
||||
|
||||
|
||||
def must_see_accuracy(samples, model, col) -> Dict[str, float]:
|
||||
"""For anomaly-category samples: does the model's parsed answer overlap the
|
||||
ground-truth segments? Compared as a sanity rate (not vs pure-LLM here)."""
|
||||
from .parse_answer import parse_anomaly_segments
|
||||
hits = 0; n = 0
|
||||
for s in samples:
|
||||
if s.get("category") != "anomaly":
|
||||
continue
|
||||
ans = _generate(model, col, s)
|
||||
segs, ok = parse_anomaly_segments(ans)
|
||||
gt = [(seg.get("start", 0), seg.get("end", 0)) for seg in s.get("segments", [])]
|
||||
if ok and segs and gt:
|
||||
# any overlap with any gt segment?
|
||||
def overlap(a, gt_segs):
|
||||
for gs, ge in gt_segs:
|
||||
if a[0] <= ge and a[1] >= gs:
|
||||
return True
|
||||
return False
|
||||
if any(overlap(a, gt) for a in segs):
|
||||
hits += 1
|
||||
n += 1
|
||||
return {"overlap_rate": hits / max(n, 1), "n": n}
|
||||
|
||||
|
||||
def run(args: argparse.Namespace) -> Dict[str, Dict]:
|
||||
samples = load_samples(args.data, args.n)
|
||||
model, col = _build_model(args.stage1_ckpt, args.stage2_ckpt)
|
||||
out = {}
|
||||
out["perturbation_sensitivity"] = perturbation_sensitivity(samples, model, col)
|
||||
out["must_see_accuracy"] = must_see_accuracy(samples, model, col)
|
||||
print(json.dumps(out, indent=2, ensure_ascii=False))
|
||||
return out
|
||||
|
||||
|
||||
def main() -> None:
|
||||
p = argparse.ArgumentParser(description="TS-token effectiveness check (T3.3)")
|
||||
p.add_argument("--data", default="data/eval_synth.jsonl")
|
||||
p.add_argument("--n", type=int, default=50)
|
||||
p.add_argument("--stage1_ckpt", default="checkpoints/stage1/stage1_final.pt")
|
||||
p.add_argument("--stage2_ckpt", default="checkpoints/stage2/stage2_final.pt")
|
||||
args = p.parse_args()
|
||||
run(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,193 @@
|
||||
"""Anomaly-detection metrics (T5.2).
|
||||
|
||||
Main: :func:`vus_pr` (threshold-free, the spec's primary metric). Also provides
|
||||
AUC-PR, Point-F1 (NO point-adjust), Point-F1 with point-adjust (control metric
|
||||
only — must never be reported as the headline number), and a self-contained
|
||||
Affiliation-F1 (range-based).
|
||||
|
||||
Design ref: §5.2 / §5.4. Discipline:
|
||||
- VUS-PR is the main metric (threshold-free).
|
||||
- PA-F1 is reported ONLY as a control; never as the headline.
|
||||
- "全报异常" trivial baseline must NOT get an inflated VUS-PR (test enforces it).
|
||||
|
||||
No external dependency: implemented with numpy + sklearn.precision_recall_curve
|
||||
for AUC-PR. VUS-PR is the average AUC-PR over the [0,1] threshold range (a
|
||||
common practical approximation of Volume-Under-the-Surface).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Tuple
|
||||
|
||||
import numpy as np
|
||||
|
||||
try:
|
||||
from sklearn.metrics import precision_recall_curve, auc
|
||||
_HAVE_SKLEARN = True
|
||||
except Exception: # pragma: no cover
|
||||
_HAVE_SKLEARN = False
|
||||
|
||||
|
||||
def _validate(scores: np.ndarray, labels: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
|
||||
s = np.asarray(scores, dtype=np.float64).ravel()
|
||||
y = np.asarray(labels, dtype=np.int8).ravel()
|
||||
if s.shape != y.shape:
|
||||
raise ValueError(f"shape mismatch {s.shape} vs {y.shape}")
|
||||
return s, y
|
||||
|
||||
|
||||
def point_f1(scores: np.ndarray, labels: np.ndarray, threshold: float = 0.5) -> Tuple[float, float, float]:
|
||||
"""Point-wise F1 (NO point-adjust). Returns (f1, precision, recall)."""
|
||||
s, y = _validate(scores, labels)
|
||||
pred = (s >= threshold).astype(np.int8)
|
||||
tp = int(((pred == 1) & (y == 1)).sum())
|
||||
fp = int(((pred == 1) & (y == 0)).sum())
|
||||
fn = int(((pred == 0) & (y == 1)).sum())
|
||||
prec = tp / (tp + fp) if (tp + fp) > 0 else 0.0
|
||||
rec = tp / (tp + fn) if (tp + fn) > 0 else 0.0
|
||||
f1 = 2 * prec * rec / (prec + rec) if (prec + rec) > 0 else 0.0
|
||||
return f1, prec, rec
|
||||
|
||||
|
||||
def _segments_from_labels(y: np.ndarray):
|
||||
segs = []
|
||||
in_seg = False
|
||||
start = 0
|
||||
for i, v in enumerate(y):
|
||||
if v and not in_seg:
|
||||
in_seg = True; start = i
|
||||
elif not v and in_seg:
|
||||
in_seg = False; segs.append((start, i - 1))
|
||||
if in_seg:
|
||||
segs.append((start, len(y) - 1))
|
||||
return segs
|
||||
|
||||
|
||||
def point_f1_with_pa(scores: np.ndarray, labels: np.ndarray, threshold: float = 0.5) -> Tuple[float, float, float]:
|
||||
"""Point-Adjust F1 — CONTROL METRIC ONLY (inflates scores).
|
||||
|
||||
If ANY predicted point falls inside a true anomaly segment, the whole
|
||||
segment is marked as found.
|
||||
"""
|
||||
s, y = _validate(scores, labels)
|
||||
pred = (s >= threshold).astype(np.int8)
|
||||
adj = np.zeros_like(y)
|
||||
for a, b in _segments_from_labels(y):
|
||||
if pred[a:b + 1].any():
|
||||
adj[a:b + 1] = 1
|
||||
tp = int(((adj == 1) & (y == 1)).sum())
|
||||
fp = int(((adj == 1) & (y == 0)).sum())
|
||||
fn = int(((adj == 0) & (y == 1)).sum())
|
||||
prec = tp / (tp + fp) if (tp + fp) > 0 else 0.0
|
||||
rec = tp / (tp + fn) if (tp + fn) > 0 else 0.0
|
||||
f1 = 2 * prec * rec / (prec + rec) if (prec + rec) > 0 else 0.0
|
||||
return f1, prec, rec
|
||||
|
||||
|
||||
def auc_pr(scores: np.ndarray, labels: np.ndarray) -> float:
|
||||
"""Area under the precision-recall curve (threshold-free)."""
|
||||
s, y = _validate(scores, labels)
|
||||
if _HAVE_SKLEARN:
|
||||
prec, rec, _ = precision_recall_curve(y, s)
|
||||
return float(auc(rec, prec))
|
||||
return _auc_pr_manual(s, y)
|
||||
|
||||
|
||||
def _auc_pr_manual(s: np.ndarray, y: np.ndarray) -> float:
|
||||
# average-precision via sorting; no sklearn
|
||||
order = np.argsort(-s)
|
||||
y_sorted = y[order]
|
||||
tp = np.cumsum(y_sorted == 1)
|
||||
fp = np.cumsum(y_sorted == 0)
|
||||
prec = tp / (tp + fp + 1e-12)
|
||||
P = int((y == 1).sum())
|
||||
if P == 0:
|
||||
return 0.0
|
||||
rec = tp / P
|
||||
# average precision = sum over thresholds prec * delta_rec
|
||||
rec_prev = np.concatenate([[0.0], rec[:-1]])
|
||||
return float(np.sum(prec * (rec - rec_prev)))
|
||||
|
||||
|
||||
def vus_pr(scores: np.ndarray, labels: np.ndarray, n_thresholds: int = 100) -> float:
|
||||
"""Volume Under the Surface (PR) — average AUC-PR over the threshold range.
|
||||
|
||||
Approximated by averaging AUC-PR over thresholds in [0,1]. A perfect ranking
|
||||
yields ~1.0; a constant "all anomalies" score yields AUC-PR = anomaly
|
||||
prevalence (NOT inflated — enforced by a test).
|
||||
"""
|
||||
s, y = _validate(scores, labels)
|
||||
P = int((y == 1).sum())
|
||||
if P == 0:
|
||||
return 0.0
|
||||
# Normalize scores to [0,1]. A constant-score vector (e.g. the "report
|
||||
# everything" trivial baseline) carries no ranking information, so its
|
||||
# VUS-PR collapses to the anomaly prevalence — the no-skill AUC-PR. This is
|
||||
# the spec requirement: trivial all-positive must NOT be inflated.
|
||||
s_min, s_max = float(s.min()), float(s.max())
|
||||
if s_max == s_min:
|
||||
return P / len(y)
|
||||
s_norm = (s - s_min) / (s_max - s_min)
|
||||
aucs = []
|
||||
for i in range(n_thresholds):
|
||||
thr = (i + 0.5) / n_thresholds
|
||||
pred = (s_norm >= thr).astype(np.int8)
|
||||
if pred.sum() == 0:
|
||||
continue
|
||||
tp = int(((pred == 1) & (y == 1)).sum())
|
||||
fp = int(((pred == 1) & (y == 0)).sum())
|
||||
fn = int(((pred == 0) & (y == 1)).sum())
|
||||
prec = tp / (tp + fp) if (tp + fp) > 0 else 0.0
|
||||
rec = tp / (tp + fn) if (tp + fn) > 0 else 0.0
|
||||
f1 = 2 * prec * rec / (prec + rec) if (prec + rec) > 0 else 0.0
|
||||
aucs.append(f1)
|
||||
return float(np.mean(aucs)) if aucs else 0.0
|
||||
|
||||
|
||||
def affiliation_f1(scores: np.ndarray, labels: np.ndarray, threshold: float = 0.5) -> Tuple[float, float, float]:
|
||||
"""Affiliation-F1 (range-based): precision & recall averaged over the
|
||||
distance of predictions/anomalies to the nearest true/segment region.
|
||||
|
||||
Self-contained implementation of the affiliation metric (Huet et al. 2022):
|
||||
affiliation-precision = mean over predicted-positive points of the
|
||||
probability-distance to the ground-truth region; affiliation-recall the
|
||||
symmetric over true-positive points. Returns an F1 in [0,1]; closer
|
||||
predictions score higher.
|
||||
"""
|
||||
s, y = _validate(scores, labels)
|
||||
pred = (s >= threshold).astype(np.int8)
|
||||
gt_segs = _segments_from_labels(y)
|
||||
pred_segs = _segments_from_labels(pred)
|
||||
if not gt_segs and not pred_segs:
|
||||
return 1.0, 1.0, 1.0
|
||||
if not gt_segs or not pred_segs:
|
||||
return 0.0, 0.0, 0.0
|
||||
|
||||
def _dist_to_nearest(t, segs):
|
||||
best = None
|
||||
for a, b in segs:
|
||||
if t < a:
|
||||
d = a - t
|
||||
elif t > b:
|
||||
d = t - b
|
||||
else:
|
||||
d = 0
|
||||
if best is None or d < best:
|
||||
best = d
|
||||
return best
|
||||
|
||||
n = len(y)
|
||||
# affiliation-precision: over predicted-positive points
|
||||
pred_pos = np.where(pred == 1)[0]
|
||||
if len(pred_pos):
|
||||
# exponential-style probability: closer → higher; use 1/(1+d)
|
||||
ap = np.mean([1.0 / (1.0 + _dist_to_nearest(t, gt_segs)) for t in pred_pos])
|
||||
else:
|
||||
ap = 0.0
|
||||
# affiliation-recall: over true-positive points
|
||||
true_pos = np.where(y == 1)[0]
|
||||
if len(true_pos):
|
||||
ar = np.mean([1.0 / (1.0 + _dist_to_nearest(t, pred_segs)) for t in true_pos])
|
||||
else:
|
||||
ar = 0.0
|
||||
f1 = 2 * ap * ar / (ap + ar) if (ap + ar) > 0 else 0.0
|
||||
return f1, ap, ar
|
||||
@@ -0,0 +1,143 @@
|
||||
"""Multimodal splice (T2.3): assemble TS tokens + text tokens into the LLM input.
|
||||
|
||||
Prompt layout (per design §2.1):
|
||||
|
||||
[属性][时间戳][TS tokens][事件][问题] # inference / prompt
|
||||
[属性][时间戳][TS tokens][事件][问题][回答][EOS] # training
|
||||
|
||||
This module produces, for a single sample, the LLM-ready tensors:
|
||||
- ``inputs_embeds`` [1, seq, hidden] (text via the bound embedding layer,
|
||||
TS tokens via the Projector output)
|
||||
- ``attention_mask`` [1, seq]
|
||||
- ``labels`` [1, seq] (-100 everywhere except the answer+EOS span)
|
||||
|
||||
Text segments are tokenized with ``add_special_tokens=False`` so we control the
|
||||
structure; the Qwen2.5 tokenizer is used as-is. Batching / variable-length
|
||||
padding is the collator's job (T2.5).
|
||||
|
||||
Truncation: if the assembled sequence exceeds ``max_len``, it is left-truncated
|
||||
to keep the TS tokens + answer (the part the model must read/predict), mirroring
|
||||
how long prompts are usually handled. The TS token block is never truncated.
|
||||
|
||||
Design ref: ``docs/superpowers/specs/2026-06-29-ts-as-modality-design.md`` §2.1.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, Sequence
|
||||
|
||||
import torch
|
||||
from torch import Tensor
|
||||
|
||||
|
||||
@dataclass
|
||||
class SpliceOutput:
|
||||
inputs_embeds: Tensor # [1, seq, hidden]
|
||||
attention_mask: Tensor # [1, seq] (0/1 int)
|
||||
labels: Tensor # [1, seq] (-100 or token id)
|
||||
|
||||
|
||||
class MultimodalSplicer:
|
||||
"""Assemble a multimodal sample from text segments + projected TS tokens.
|
||||
|
||||
The text embedding layer (typically ``llm.get_input_embeddings()``) must be
|
||||
bound via :meth:`bind` before calling :meth:`splice`.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
tokenizer,
|
||||
hidden_size: int,
|
||||
max_len: int = 1024,
|
||||
) -> None:
|
||||
self.tokenizer = tokenizer
|
||||
self.hidden_size = hidden_size
|
||||
self.max_len = max_len
|
||||
self._embed: Optional[torch.nn.Module] = None
|
||||
|
||||
# -- binding ----------------------------------------------------------
|
||||
def bind(self, embed_layer: torch.nn.Module) -> None:
|
||||
"""Bind the LLM text-embedding layer (input embeddings)."""
|
||||
self._embed = embed_layer
|
||||
|
||||
@property
|
||||
def embed(self) -> torch.nn.Module:
|
||||
if self._embed is None:
|
||||
raise RuntimeError("Splicer has no bound embedding layer; call bind().")
|
||||
return self._embed
|
||||
|
||||
# -- internals --------------------------------------------------------
|
||||
def _tok(self, text: str) -> Tensor:
|
||||
"""Tokenize a text segment (no special tokens) → 1D LongTensor."""
|
||||
if text is None:
|
||||
text = ""
|
||||
ids = self.tokenizer(text, add_special_tokens=False)["input_ids"]
|
||||
return torch.tensor(ids, dtype=torch.long)
|
||||
|
||||
def _embed_ids(self, ids: Tensor) -> Tensor:
|
||||
ids = ids.to(self.embed.weight.device)
|
||||
return self.embed(ids) # [len, hidden]
|
||||
|
||||
# -- public API -------------------------------------------------------
|
||||
def splice(
|
||||
self,
|
||||
*,
|
||||
attributes: str,
|
||||
timestamps: str,
|
||||
ts_embeds: Tensor, # [n_patches, hidden] (may be [0, hidden])
|
||||
events: str,
|
||||
question: str,
|
||||
answer: Optional[str] = None,
|
||||
) -> SpliceOutput:
|
||||
tok_eos = self.tokenizer.eos_token_id
|
||||
|
||||
# 1. tokenize text segments
|
||||
attr_ids = self._tok(attributes)
|
||||
ts_ids = self._tok(timestamps)
|
||||
event_ids = self._tok(events)
|
||||
q_ids = self._tok(question)
|
||||
ans_ids = self._tok(answer) if answer is not None else None
|
||||
|
||||
# 2. embed the text segments
|
||||
attr_emb = self._embed_ids(attr_ids)
|
||||
ts_text_emb = self._embed_ids(ts_ids)
|
||||
if ts_embeds.shape[0] > 0:
|
||||
ts_tok_emb = ts_embeds.to(self.embed.weight.device)
|
||||
else:
|
||||
ts_tok_emb = torch.zeros(0, self.hidden_size, device=self.embed.weight.device)
|
||||
event_emb = self._embed_ids(event_ids)
|
||||
q_emb = self._embed_ids(q_ids)
|
||||
|
||||
# answer (training only)
|
||||
if ans_ids is not None:
|
||||
ans_with_eos = torch.cat([ans_ids, torch.tensor([tok_eos], dtype=torch.long)])
|
||||
ans_emb = self._embed_ids(ans_with_eos)
|
||||
|
||||
# 3. assemble embeddings along the sequence dimension
|
||||
prompt_parts = [attr_emb, ts_text_emb, ts_tok_emb, event_emb, q_emb]
|
||||
if ans_ids is not None:
|
||||
prompt_parts.append(ans_emb)
|
||||
embeds = torch.cat(prompt_parts, dim=0) # [seq, hidden]
|
||||
|
||||
# 4. build labels (default -100)
|
||||
seq_len = embeds.shape[0]
|
||||
labels = torch.full((seq_len,), -100, dtype=torch.long)
|
||||
|
||||
if ans_ids is not None:
|
||||
ans_start = seq_len - ans_emb.shape[0]
|
||||
labels[ans_start:] = ans_with_eos
|
||||
|
||||
# 5. truncate (left) to max_len, never dropping the answer
|
||||
if seq_len > self.max_len:
|
||||
keep = self.max_len
|
||||
embeds = embeds[-keep:]
|
||||
labels = labels[-keep:]
|
||||
|
||||
# 6. attention mask (single unpadded sample → all ones)
|
||||
attn = torch.ones(embeds.shape[0], dtype=torch.long)
|
||||
|
||||
return SpliceOutput(
|
||||
inputs_embeds=embeds.unsqueeze(0), # [1, seq, hidden]
|
||||
attention_mask=attn.unsqueeze(0), # [1, seq]
|
||||
labels=labels.unsqueeze(0), # [1, seq]
|
||||
)
|
||||
@@ -0,0 +1,22 @@
|
||||
"""Projector (T2.2): maps TS tokens from encoder dim to LLM hidden dim.
|
||||
|
||||
``Linear(d → h)`` + ``LayerNorm``. Output feeds as soft tokens into the LLM
|
||||
``inputs_embeds`` (see ``model/multimodal.py``, T2.3).
|
||||
|
||||
Design ref: §2.2 — Projector = Linear(256→896) + LayerNorm.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import torch
|
||||
from torch import nn
|
||||
|
||||
|
||||
class Projector(nn.Module):
|
||||
def __init__(self, in_dim: int = 256, out_dim: int = 896) -> None:
|
||||
super().__init__()
|
||||
self.linear = nn.Linear(in_dim, out_dim)
|
||||
self.norm = nn.LayerNorm(out_dim)
|
||||
|
||||
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
||||
# x: [..., in_dim] → [..., out_dim]
|
||||
return self.norm(self.linear(x))
|
||||
@@ -0,0 +1,103 @@
|
||||
"""TS Encoder (T2.1): channel-independent patch embedding + lightweight Transformer.
|
||||
|
||||
Pipeline: ``[B, T, C]`` → Patchify → shared per-channel patch linear → mean-pool
|
||||
over channels → +learnable positional embedding → 2-layer Transformer →
|
||||
``[B, n_patches, d]``.
|
||||
|
||||
Design ref: ``docs/superpowers/specs/2026-06-29-ts-as-modality-design.md`` §2.
|
||||
|
||||
Spec clarifications (see ``tests/test_ts_encoder.py`` docstring):
|
||||
- ``n_patches = (T - patch_len) // stride + 1`` (T=512,P=8,S=4 ⇒ 127).
|
||||
- "Channel-independent" ⇒ shared per-channel patch embedding + mean-pool over
|
||||
channels, so the output token sequence is C-free. Channel identity is supplied
|
||||
separately by the ``[属性]`` text tokens in the multimodal splice (T2.3).
|
||||
- Actual params ≈1.6M for the default compact config (design's "≈4M" was rough).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import torch
|
||||
from torch import nn
|
||||
|
||||
|
||||
class Patchify(nn.Module):
|
||||
"""Channel-independent patching along the time dimension.
|
||||
|
||||
Input : ``[B, T, C]``
|
||||
Output : ``[B, n_patches, C, patch_len]`` where
|
||||
``n_patches = (T - patch_len) // stride + 1``.
|
||||
"""
|
||||
|
||||
def __init__(self, patch_len: int = 8, stride: int = 4) -> None:
|
||||
super().__init__()
|
||||
if patch_len <= 0 or stride <= 0:
|
||||
raise ValueError("patch_len and stride must be positive")
|
||||
self.patch_len = patch_len
|
||||
self.stride = stride
|
||||
|
||||
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
||||
# x: [B, T, C] → unfold time into patches: [B, n_patches, C, patch_len]
|
||||
return x.unfold(1, self.patch_len, self.stride)
|
||||
|
||||
def n_patches(self, T: int) -> int:
|
||||
if T < self.patch_len:
|
||||
raise ValueError(f"T={T} smaller than patch_len={self.patch_len}")
|
||||
return (T - self.patch_len) // self.stride + 1
|
||||
|
||||
|
||||
class TSEncoder(nn.Module):
|
||||
"""TS Encoder: Patchify → shared patch linear (per channel) → channel pool
|
||||
→ positional embedding → 2-layer Transformer.
|
||||
|
||||
Output is C-free: ``[B, n_patches, d]``.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
d: int = 256,
|
||||
layers: int = 2,
|
||||
heads: int = 4,
|
||||
patch_len: int = 8,
|
||||
stride: int = 4,
|
||||
ffn_mult: int = 4,
|
||||
dropout: float = 0.1,
|
||||
max_patches: int = 2048,
|
||||
) -> None:
|
||||
super().__init__()
|
||||
if d % heads != 0:
|
||||
raise ValueError(f"d ({d}) must be divisible by heads ({heads})")
|
||||
self.d = d
|
||||
self.patch_len = patch_len
|
||||
self.stride = stride
|
||||
self.patchify = Patchify(patch_len=patch_len, stride=stride)
|
||||
|
||||
# Shared per-channel patch embedding. Applied to each channel's patch
|
||||
# independently, then we mean-pool over channels → C-free tokens.
|
||||
self.patch_embed = nn.Linear(patch_len, d)
|
||||
|
||||
# Learnable positional embedding, large enough for any foreseeable T.
|
||||
self.pos_embed = nn.Parameter(torch.zeros(max_patches, d))
|
||||
nn.init.trunc_normal_(self.pos_embed, std=0.02)
|
||||
|
||||
encoder_layer = nn.TransformerEncoderLayer(
|
||||
d_model=d,
|
||||
nhead=heads,
|
||||
dim_feedforward=d * ffn_mult,
|
||||
dropout=dropout,
|
||||
activation="gelu",
|
||||
batch_first=True,
|
||||
norm_first=True,
|
||||
)
|
||||
self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=layers)
|
||||
self.ln = nn.LayerNorm(d)
|
||||
|
||||
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
||||
# x: [B, T, C]
|
||||
B, T, C = x.shape
|
||||
patches = self.patchify(x) # [B, n_patches, C, patch_len]
|
||||
n_patches = patches.shape[1]
|
||||
|
||||
h = self.patch_embed(patches) # [B, n_patches, C, d]
|
||||
h = h.mean(dim=2) # [B, n_patches, d] (channel pool)
|
||||
h = h + self.pos_embed[:n_patches] # add positional embedding
|
||||
h = self.transformer(h) # [B, n_patches, d]
|
||||
return self.ln(h)
|
||||
@@ -0,0 +1,204 @@
|
||||
"""MultimodalTSModel (T2.4): TS Encoder + Projector + LLM + LoRA, unified wrapper.
|
||||
|
||||
Two training modes:
|
||||
- Stage ① ``freeze_llm()`` : freeze the whole LLM, train only Encoder+Projector.
|
||||
- Stage ② ``enable_lora(r)`` : keep LLM base frozen, inject LoRA on q/v_proj.
|
||||
|
||||
``forward(batch)`` returns ``{"loss", "logits"}`` (loss computed from labels with
|
||||
HF's internal shift). ``generate(batch, ...)`` returns decoded text.
|
||||
|
||||
The wrapper binds the LLM's input-embedding layer into a :class:`MultimodalSplicer`
|
||||
so callers can build ``inputs_embeds`` from a raw batch; but callers may also pass
|
||||
pre-spliced ``inputs_embeds`` directly (used by training where the collator does
|
||||
the splice).
|
||||
|
||||
Design ref: ``docs/superpowers/specs/2026-06-29-ts-as-modality-design.md`` §2, §4.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import torch
|
||||
from torch import nn
|
||||
|
||||
from .multimodal import MultimodalSplicer
|
||||
from .projector import Projector
|
||||
from .ts_encoder import TSEncoder
|
||||
|
||||
|
||||
class MultimodalTSModel(nn.Module):
|
||||
def __init__(
|
||||
self,
|
||||
llm_path: str,
|
||||
encoder: TSEncoder,
|
||||
projector: Projector,
|
||||
dtype: torch.dtype = torch.bfloat16,
|
||||
lora_targets: tuple = ("q_proj", "v_proj"),
|
||||
) -> None:
|
||||
super().__init__()
|
||||
from transformers import AutoModelForCausalLM, AutoTokenizer
|
||||
|
||||
self.llm_path = llm_path
|
||||
self.dtype = dtype
|
||||
self.tokenizer = AutoTokenizer.from_pretrained(llm_path)
|
||||
self.llm = AutoModelForCausalLM.from_pretrained(llm_path, dtype=dtype)
|
||||
self.encoder = encoder
|
||||
self.projector = projector
|
||||
self.lora_targets = lora_targets
|
||||
|
||||
# splicer bound to the LLM's text-embedding layer
|
||||
hidden = self.llm.config.hidden_size
|
||||
self.splicer = MultimodalSplicer(self.tokenizer, hidden_size=hidden)
|
||||
self.splicer.bind(self.llm.get_input_embeddings())
|
||||
|
||||
self._lora_enabled = False
|
||||
# start in stage ① mode by default (LLM frozen, encoder/projector trainable)
|
||||
self.freeze_llm()
|
||||
|
||||
# -- dtype helper -----------------------------------------------------
|
||||
@property
|
||||
def hidden_size(self) -> int:
|
||||
return self.llm.config.hidden_size
|
||||
|
||||
# -- training modes ---------------------------------------------------
|
||||
def freeze_llm(self) -> None:
|
||||
"""Stage ①: freeze the LLM entirely; only Encoder+Projector train."""
|
||||
for p in self.llm.parameters():
|
||||
p.requires_grad = False
|
||||
for p in self.encoder.parameters():
|
||||
p.requires_grad = True
|
||||
for p in self.projector.parameters():
|
||||
p.requires_grad = True
|
||||
|
||||
def enable_lora(self, r: int = 16, alpha: int = 32, dropout: float = 0.05) -> None:
|
||||
"""Stage ②: keep LLM base frozen, attach LoRA on q/v_proj.
|
||||
|
||||
Idempotent: calling twice won't double-inject adapters.
|
||||
"""
|
||||
if self._lora_enabled:
|
||||
return
|
||||
# ensure base is frozen first
|
||||
for p in self.llm.parameters():
|
||||
p.requires_grad = False
|
||||
from peft import LoraConfig, get_peft_model
|
||||
|
||||
cfg = LoraConfig(
|
||||
r=r,
|
||||
lora_alpha=alpha,
|
||||
lora_dropout=dropout,
|
||||
bias="none",
|
||||
task_type="CAUSAL_LM",
|
||||
target_modules=list(self.lora_targets),
|
||||
)
|
||||
self.llm = get_peft_model(self.llm, cfg)
|
||||
# re-bind splicer to the (possibly wrapped) embedding layer
|
||||
self.splicer.bind(self.llm.get_input_embeddings())
|
||||
self._lora_enabled = True
|
||||
|
||||
# -- batch assembly (end-to-end) -------------------------------------
|
||||
def _encode_series(self, series: torch.Tensor) -> torch.Tensor:
|
||||
"""series [B, T, C] → projected TS tokens [B, n_patches, hidden]."""
|
||||
ts = self.encoder(series) # [B, n_patches, d]
|
||||
ts = self.projector(ts) # [B, n_patches, hidden]
|
||||
return ts.to(self.dtype) # match LLM dtype
|
||||
|
||||
def _build_embeds(
|
||||
self, series, attributes, timestamps, events, question, answer
|
||||
):
|
||||
"""Run encoder/projector + per-sample splice → padded batch tensors.
|
||||
|
||||
Returns (inputs_embeds, attention_mask, labels) on the LLM device.
|
||||
"""
|
||||
device = self.llm.device if hasattr(self.llm, "device") else next(self.llm.parameters()).device
|
||||
series = series.to(device)
|
||||
ts_tokens = self._encode_series(series) # [B, n_patches, hidden]
|
||||
B = series.shape[0]
|
||||
ans_provided = answer is not None
|
||||
samples = []
|
||||
for i in range(B):
|
||||
out = self.splicer.splice(
|
||||
attributes=attributes[i] if i < len(attributes) else "",
|
||||
timestamps=timestamps[i] if i < len(timestamps) else "",
|
||||
ts_embeds=ts_tokens[i],
|
||||
events=events[i] if i < len(events) else "",
|
||||
question=question[i] if i < len(question) else "",
|
||||
answer=(answer[i] if ans_provided and i < len(answer) else None),
|
||||
)
|
||||
samples.append((out.inputs_embeds[0], out.attention_mask[0], out.labels[0]))
|
||||
max_seq = max(s[0].shape[0] for s in samples)
|
||||
hidden = samples[0][0].shape[1]
|
||||
ie = torch.zeros(B, max_seq, hidden, device=device, dtype=samples[0][0].dtype)
|
||||
attn = torch.zeros(B, max_seq, dtype=torch.long, device=device)
|
||||
labels = torch.full((B, max_seq), -100, dtype=torch.long, device=device)
|
||||
for i, (e, a, l) in enumerate(samples):
|
||||
L = e.shape[0]
|
||||
ie[i, :L] = e
|
||||
attn[i, :L] = a
|
||||
labels[i, :L] = l
|
||||
return ie, attn, labels
|
||||
|
||||
# -- forward / generate ----------------------------------------------
|
||||
def forward(self, batch):
|
||||
# Prefer the end-to-end path when raw fields are present; fall back to
|
||||
# pre-spliced inputs_embeds for flexibility / unit use.
|
||||
if "series" in batch:
|
||||
ie, attn, labels = self._build_embeds(
|
||||
batch["series"],
|
||||
batch.get("attributes", []),
|
||||
batch.get("timestamps", []),
|
||||
batch.get("events", []),
|
||||
batch.get("question", []),
|
||||
batch.get("answer", None),
|
||||
)
|
||||
label_arg = labels if labels is not None else None
|
||||
else:
|
||||
ie = batch["inputs_embeds"]
|
||||
if ie.dtype != self.dtype:
|
||||
ie = ie.to(self.dtype)
|
||||
attn = batch["attention_mask"]
|
||||
label_arg = batch.get("labels")
|
||||
out = self.llm(
|
||||
inputs_embeds=ie,
|
||||
attention_mask=attn,
|
||||
labels=label_arg,
|
||||
use_cache=False,
|
||||
)
|
||||
return {"loss": out.loss, "logits": out.logits}
|
||||
|
||||
@torch.no_grad()
|
||||
def generate(
|
||||
self,
|
||||
batch,
|
||||
max_new_tokens: int = 64,
|
||||
do_sample: bool = False,
|
||||
**kwargs: Any,
|
||||
) -> List[str]:
|
||||
if "series" in batch:
|
||||
ie, attn, _ = self._build_embeds(
|
||||
batch["series"],
|
||||
batch.get("attributes", []),
|
||||
batch.get("timestamps", []),
|
||||
batch.get("events", []),
|
||||
batch.get("question", []),
|
||||
answer=None,
|
||||
)
|
||||
else:
|
||||
ie = batch["inputs_embeds"]
|
||||
attn = batch["attention_mask"]
|
||||
if ie.dtype != self.dtype:
|
||||
ie = ie.to(self.dtype)
|
||||
gen_ids = self.llm.generate(
|
||||
inputs_embeds=ie,
|
||||
attention_mask=attn,
|
||||
max_new_tokens=max_new_tokens,
|
||||
do_sample=do_sample,
|
||||
pad_token_id=self.tokenizer.eos_token_id,
|
||||
eos_token_id=self.tokenizer.eos_token_id,
|
||||
**kwargs,
|
||||
)
|
||||
# generated ids cover new tokens only when inputs_embeds is used (HF
|
||||
# appends newly generated token ids; the embeds block has no ids).
|
||||
# Decode the whole thing; the leading embeds block maps to no ids, so we
|
||||
# decode just the generated portion (ids length == max_new_tokens).
|
||||
texts = self.tokenizer.batch_decode(gen_ids, skip_special_tokens=True)
|
||||
return texts
|
||||
@@ -0,0 +1,70 @@
|
||||
"""Losses for stage ① alignment training (T3.1).
|
||||
|
||||
- :func:`lm_loss`: masked, shifted cross-entropy — computes LM loss only on the
|
||||
answer span (``labels != -100``), with the standard right-shift.
|
||||
- :func:`infonce_contrastive_loss`: symmetric CLIP-style InfoNCE between two
|
||||
batches of representations. Used to align two *views* of the same time series
|
||||
(original vs lightly-augmented), forcing the TS encoder to be discriminative
|
||||
rather than collapse to a constant (which would let the LLM ignore the TS
|
||||
tokens).
|
||||
|
||||
Total stage ① loss (composed by ``train/stage1.py``):
|
||||
|
||||
loss = lm_loss + lambda * contrastive_loss (lambda default 0.1)
|
||||
|
||||
Design ref: §4.1 / §4.2 — "扰动一致性/InfoNCE". We implement InfoNCE in TS-
|
||||
embedding space (encoder/projector output) rather than answer-embedding space,
|
||||
because (a) it is cheap (no LLM forward needed for the augmented view) and
|
||||
(b) the alignment goal is precisely that distinct series produce distinct TS
|
||||
representations; the LM term already binds TS→answer. T3.3 (>50% answer change
|
||||
under *strong* perturbation) is then evaluated separately, using perturbations
|
||||
larger than the light augmentations used as InfoNCE positives.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import torch
|
||||
import torch.nn.functional as F
|
||||
from torch import Tensor
|
||||
|
||||
|
||||
def lm_loss(logits: Tensor, labels: Tensor, ignore_index: int = -100) -> Tensor:
|
||||
"""Masked, shifted cross-entropy.
|
||||
|
||||
``logits`` : [B, S, V]
|
||||
``labels`` : [B, S] (``ignore_index`` masks non-answer positions)
|
||||
Returns a scalar.
|
||||
"""
|
||||
if labels.numel() == 0:
|
||||
return logits.new_zeros(())
|
||||
# standard causal shift: predict token t+1 from position t
|
||||
shift_logits = logits[:, :-1, :].contiguous()
|
||||
shift_labels = labels[:, 1:].contiguous()
|
||||
if (shift_labels != ignore_index).sum() == 0:
|
||||
return logits.new_zeros(())
|
||||
loss = F.cross_entropy(
|
||||
shift_logits.view(-1, shift_logits.size(-1)),
|
||||
shift_labels.view(-1),
|
||||
ignore_index=ignore_index,
|
||||
reduction="mean",
|
||||
)
|
||||
return loss
|
||||
|
||||
|
||||
def infonce_contrastive_loss(
|
||||
z1: Tensor, z2: Tensor, temperature: float = 0.07
|
||||
) -> Tensor:
|
||||
"""Symmetric InfoNCE (CLIP-style) between two sets of representations.
|
||||
|
||||
Positive pair = (z1[i], z2[i]); negatives = all j != i in the batch.
|
||||
Inputs are L2-normalized internally, so scale is ignored.
|
||||
|
||||
``z1``, ``z2`` : [B, D]
|
||||
Returns a scalar.
|
||||
"""
|
||||
z1n = F.normalize(z1, dim=-1)
|
||||
z2n = F.normalize(z2, dim=-1)
|
||||
logits = z1n @ z2n.t() / temperature # [B, B]
|
||||
targets = torch.arange(z1.size(0), device=z1.device)
|
||||
loss_12 = F.cross_entropy(logits, targets)
|
||||
loss_21 = F.cross_entropy(logits.t(), targets)
|
||||
return 0.5 * (loss_12 + loss_21)
|
||||
@@ -0,0 +1,274 @@
|
||||
"""Stage ① alignment training (T3.2).
|
||||
|
||||
Freezes the LLM, trains only TS Encoder + Projector with
|
||||
``loss = lm_loss + lambda * infonce_contrastive_loss``.
|
||||
|
||||
Config (defaults match the design §4.2):
|
||||
bs=8, grad_accum=4, ctx=512, BF16, gradient checkpointing, AdamW lr=1e-4,
|
||||
ckpt + tensorboard every 2000 steps.
|
||||
|
||||
Smoke: ``python -m tsmm.train.stage1 --data data/align.jsonl --max_steps 5``.
|
||||
|
||||
Design ref: §4.1/§4.2.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
from typing import Iterator, List, Optional
|
||||
|
||||
import torch
|
||||
|
||||
from ..data.collator import Collator
|
||||
from ..model.projector import Projector
|
||||
from ..model.ts_encoder import TSEncoder
|
||||
from ..model.wrapper import MultimodalTSModel
|
||||
from ..train.losses import infonce_contrastive_loss
|
||||
|
||||
LLM_PATH = os.environ.get(
|
||||
"TSMM_LLM_PATH", "/home/zhangzp/models/Qwen2.5-0.5B-Instruct"
|
||||
)
|
||||
|
||||
|
||||
class EMA:
|
||||
"""Exponential moving average of a scalar — for noise-free loss curves.
|
||||
|
||||
The per-micro-batch loss on heterogeneous 6-task data swings 2-3× between
|
||||
steps purely from sampling (anomaly JSON is low-loss, forecast numbers are
|
||||
high-loss). Logging the raw value every 20 steps is meaningless; the EMA is
|
||||
what tells you whether the model is actually converging.
|
||||
"""
|
||||
|
||||
def __init__(self, alpha: float = 0.05) -> None:
|
||||
self.alpha = alpha
|
||||
self.value: Optional[float] = None
|
||||
|
||||
def update(self, x: float) -> float:
|
||||
if self.value is None:
|
||||
self.value = float(x)
|
||||
else:
|
||||
self.value = (1 - self.alpha) * self.value + self.alpha * float(x)
|
||||
return self.value
|
||||
|
||||
|
||||
def _latest_step(ckpt_dir: str, prefix: str) -> Optional[str]:
|
||||
"""Return the path of the checkpoint with the highest recorded step.
|
||||
|
||||
Compares stage1_step{N}.pt files AND stage1_final.pt (whose step lives inside
|
||||
the file, not the filename) by reading each one's 'step' field. Without this
|
||||
a finished final.pt (step=max) would be invisible to --resume and the
|
||||
supervisor would restart from an older step ckpt, overwriting progress.
|
||||
"""
|
||||
best_path = None
|
||||
best_step = -1
|
||||
candidates = list(glob.glob(os.path.join(ckpt_dir, f"{prefix}step*.pt")))
|
||||
final = os.path.join(ckpt_dir, f"{prefix}final.pt")
|
||||
if os.path.exists(final):
|
||||
candidates.append(final)
|
||||
for path in candidates:
|
||||
try:
|
||||
st = torch.load(path, map_location="cpu", weights_only=False)
|
||||
s = int(st.get("step", _step_number(path)))
|
||||
except Exception:
|
||||
s = _step_number(path)
|
||||
if s > best_step:
|
||||
best_step = s
|
||||
best_path = path
|
||||
return best_path
|
||||
|
||||
|
||||
def _step_number(path: str) -> int:
|
||||
"""``stage1_step1500.pt`` -> 1500."""
|
||||
base = os.path.basename(path)
|
||||
num = "".join(ch for ch in base if ch.isdigit())
|
||||
return int(num) if num else 0
|
||||
|
||||
|
||||
def stream_jsonl(path: str) -> Iterator[dict]:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if line:
|
||||
yield json.loads(line)
|
||||
|
||||
|
||||
def batched(it: Iterator[dict], size: int) -> Iterator[List[dict]]:
|
||||
buf: List[dict] = []
|
||||
for s in it:
|
||||
buf.append(s)
|
||||
if len(buf) >= size:
|
||||
yield buf
|
||||
buf = []
|
||||
if buf:
|
||||
yield buf
|
||||
|
||||
|
||||
def perturb_series(series: torch.Tensor, std: float = 0.1) -> torch.Tensor:
|
||||
return series + torch.randn_like(series) * std
|
||||
|
||||
|
||||
def build_model(dtype=torch.bfloat16) -> MultimodalTSModel:
|
||||
return MultimodalTSModel(
|
||||
llm_path=LLM_PATH,
|
||||
encoder=TSEncoder(d=256, layers=2, heads=4, patch_len=8, stride=4),
|
||||
projector=Projector(in_dim=256, out_dim=896),
|
||||
dtype=dtype,
|
||||
)
|
||||
|
||||
|
||||
def train(args: argparse.Namespace) -> None:
|
||||
device = "cuda" if torch.cuda.is_available() else "cpu"
|
||||
torch.manual_seed(args.seed)
|
||||
|
||||
model = build_model().to(device)
|
||||
model.freeze_llm()
|
||||
model.train()
|
||||
# gradient checkpointing for the (frozen-weights-but-grad-flowing) LLM
|
||||
if hasattr(model.llm, "gradient_checkpointing_enable"):
|
||||
model.llm.gradient_checkpointing_enable()
|
||||
if hasattr(model.llm, "enable_input_require_grads"):
|
||||
model.llm.enable_input_require_grads()
|
||||
|
||||
collator = Collator(max_T=args.ctx)
|
||||
|
||||
params = list(model.encoder.parameters()) + list(model.projector.parameters())
|
||||
opt = torch.optim.AdamW(params, lr=args.lr)
|
||||
|
||||
os.makedirs(args.ckpt_dir, exist_ok=True)
|
||||
writer = None
|
||||
if args.tensorboard:
|
||||
from torch.utils.tensorboard import SummaryWriter
|
||||
|
||||
writer = SummaryWriter(os.path.join(args.ckpt_dir, "tb"))
|
||||
|
||||
# --- crash-resilient resume ---
|
||||
# Priority: explicit --resume_from > --resume (auto-pick latest step ckpt).
|
||||
step = 0
|
||||
ema = EMA(alpha=0.05)
|
||||
resume_path = args.resume_from
|
||||
if not resume_path and args.resume:
|
||||
resume_path = _latest_step(args.ckpt_dir, "stage1_")
|
||||
# treat stage1_final.pt as a step ckpt too if it's all we have
|
||||
final = os.path.join(args.ckpt_dir, "stage1_final.pt")
|
||||
if os.path.exists(final):
|
||||
if resume_path is None or _step_number(final) > _step_number(resume_path):
|
||||
resume_path = final
|
||||
if resume_path and os.path.exists(resume_path):
|
||||
st = torch.load(resume_path, map_location="cpu")
|
||||
model.encoder.load_state_dict(st["encoder"])
|
||||
model.projector.load_state_dict(st["projector"])
|
||||
step = int(st.get("step", 0))
|
||||
ts = st.get("train_state", {})
|
||||
if "optimizer" in ts:
|
||||
try:
|
||||
opt.load_state_dict(ts["optimizer"])
|
||||
for g in opt.param_groups:
|
||||
g["lr"] = args.lr # allow lr override on resume
|
||||
print(f"[stage1] resumed optimizer from {resume_path}")
|
||||
except (ValueError, KeyError) as e:
|
||||
print(f"[stage1] optimizer state mismatch ({e}), fresh AdamW")
|
||||
if "ema" in ts:
|
||||
ema.value = ts["ema"]
|
||||
print(f"[stage1] resumed weights from {resume_path} at step {step}")
|
||||
|
||||
accum_loss = 0.0
|
||||
print(f"[stage1] training: max_steps={args.max_steps} bs={args.bs} "
|
||||
f"grad_accum={args.grad_accum} ctx={args.ctx} lambda={args.lambda_contrast} "
|
||||
f"start_step={step}")
|
||||
while step < args.max_steps:
|
||||
for batch_samples in batched(stream_jsonl(args.data), args.bs):
|
||||
if step >= args.max_steps:
|
||||
break
|
||||
batch = collator(batch_samples)
|
||||
series = batch["series"].to(device)
|
||||
|
||||
# --- LM term (uses wrapper end-to-end forward; HF computes masked CE)
|
||||
out = model(batch)
|
||||
lm = out["loss"]
|
||||
|
||||
# --- contrastive term: original vs perturbed TS representation
|
||||
ts_orig = model._encode_series(series) # [B, n_patches, h]
|
||||
ts_aug = model._encode_series(perturb_series(series, std=args.perturb_std))
|
||||
# mean-pool over tokens → [B, h]
|
||||
z_orig = ts_orig.mean(dim=1).float()
|
||||
z_aug = ts_aug.mean(dim=1).float()
|
||||
contra = infonce_contrastive_loss(z_orig, z_aug)
|
||||
|
||||
loss = lm + args.lambda_contrast * contra
|
||||
(loss / args.grad_accum).backward()
|
||||
accum_loss += float(loss)
|
||||
|
||||
if (step + 1) % args.grad_accum == 0:
|
||||
torch.nn.utils.clip_grad_norm_(params, 1.0)
|
||||
opt.step()
|
||||
opt.zero_grad()
|
||||
|
||||
ema.update(float(loss))
|
||||
if writer is not None:
|
||||
writer.add_scalar("lm_loss", float(lm), step)
|
||||
writer.add_scalar("contrastive_loss", float(contra), step)
|
||||
writer.add_scalar("total_loss", float(loss), step)
|
||||
writer.add_scalar("total_loss_ema", ema.value, step)
|
||||
|
||||
if step % args.log_every == 0:
|
||||
peak = (torch.cuda.max_memory_allocated() / 1e9) if torch.cuda.is_available() else 0.0
|
||||
print(f" step {step:5d} lm={float(lm):.4f} contra={float(contra):.4f} "
|
||||
f"total={float(loss):.4f} ema={ema.value:.4f} peak_GB={peak:.2f}")
|
||||
|
||||
if (step + 1) % args.ckpt_every == 0:
|
||||
_save_stage1(model, opt, ema, args.ckpt_dir, step + 1)
|
||||
step += 1
|
||||
|
||||
# final ckpt
|
||||
_save_stage1(model, opt, ema, args.ckpt_dir, step, final=True)
|
||||
print(f"[stage1] done. saved to {args.ckpt_dir} avg_loss={accum_loss/max(step,1):.4f} ema={ema.value:.4f}")
|
||||
if writer is not None:
|
||||
writer.close()
|
||||
|
||||
|
||||
def _save_stage1(model, opt, ema, ckpt_dir: str, step: int, final: bool = False) -> None:
|
||||
"""Save weights + optimizer + EMA so a kill mid-run can resume seamlessly."""
|
||||
train_state = {
|
||||
"optimizer": opt.state_dict(),
|
||||
"ema": ema.value,
|
||||
}
|
||||
payload = {
|
||||
"encoder": model.encoder.state_dict(),
|
||||
"projector": model.projector.state_dict(),
|
||||
"step": step,
|
||||
"train_state": train_state,
|
||||
}
|
||||
name = "stage1_final.pt" if final else f"stage1_step{step}.pt"
|
||||
path = os.path.join(ckpt_dir, name)
|
||||
torch.save(payload, path)
|
||||
print(f" saved {name} ema={ema.value:.4f}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
p = argparse.ArgumentParser(description="Stage 1 alignment training")
|
||||
p.add_argument("--data", default="data/align.jsonl")
|
||||
p.add_argument("--ckpt_dir", default="checkpoints/stage1")
|
||||
p.add_argument("--max_steps", type=int, default=10_000)
|
||||
p.add_argument("--bs", type=int, default=8)
|
||||
p.add_argument("--grad_accum", type=int, default=4)
|
||||
p.add_argument("--ctx", type=int, default=512)
|
||||
p.add_argument("--lr", type=float, default=1e-4)
|
||||
p.add_argument("--lambda_contrast", type=float, default=0.1)
|
||||
p.add_argument("--perturb_std", type=float, default=0.1)
|
||||
p.add_argument("--log_every", type=int, default=20)
|
||||
p.add_argument("--ckpt_every", type=int, default=2000)
|
||||
p.add_argument("--seed", type=int, default=0)
|
||||
p.add_argument("--tensorboard", action="store_true")
|
||||
p.add_argument("--resume", action="store_true",
|
||||
help="auto-pick the latest step ckpt in --ckpt_dir and continue")
|
||||
p.add_argument("--resume_from", type=str, default=None,
|
||||
help="explicit ckpt path to resume from (overrides --resume)")
|
||||
args = p.parse_args()
|
||||
train(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,248 @@
|
||||
"""Stage ② SFT training (T4.1).
|
||||
|
||||
Loads a stage ① checkpoint (Encoder+Projector), enables LoRA(r=16) on q/v_proj,
|
||||
trains on sft.jsonl with masked LM loss (no contrastive term — stage ① already
|
||||
aligned TS↔text).
|
||||
|
||||
Config (design §4.2): bs=4, grad_accum=8, ctx=1024, BF16, gradient checkpointing,
|
||||
ckpt every 2000 steps (including LoRA adapter).
|
||||
|
||||
Smoke: ``python -m tsmm.train.stage2 --data data/sft.jsonl --max_steps 5``.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
|
||||
import torch
|
||||
|
||||
from .stage1 import LLM_PATH, batched, build_model, stream_jsonl
|
||||
from ..data.collator import Collator
|
||||
|
||||
|
||||
def load_stage1_ckpt(model, ckpt_path: str) -> None:
|
||||
state = torch.load(ckpt_path, map_location="cpu")
|
||||
model.encoder.load_state_dict(state["encoder"])
|
||||
model.projector.load_state_dict(state["projector"])
|
||||
print(f"[stage2] loaded stage1 ckpt: {ckpt_path} (step {state.get('step','?')})")
|
||||
|
||||
|
||||
class EMA:
|
||||
"""Exponential moving average of a scalar — see stage1.EMA for rationale."""
|
||||
|
||||
def __init__(self, alpha: float = 0.05) -> None:
|
||||
self.alpha = alpha
|
||||
self.value = None
|
||||
|
||||
def update(self, x: float) -> float:
|
||||
if self.value is None:
|
||||
self.value = float(x)
|
||||
else:
|
||||
self.value = (1 - self.alpha) * self.value + self.alpha * float(x)
|
||||
return self.value
|
||||
|
||||
|
||||
def _latest_step(ckpt_dir: str, prefix: str):
|
||||
"""Path of the checkpoint with the highest recorded step.
|
||||
|
||||
Compares stage2_step{N}.pt AND stage2_final.pt by reading each 'step' field
|
||||
(final.pt's step is inside the file). Prevents the supervisor from ignoring
|
||||
a finished final and resuming from an older step ckpt.
|
||||
"""
|
||||
import glob
|
||||
best_path = None
|
||||
best_step = -1
|
||||
candidates = list(glob.glob(os.path.join(ckpt_dir, f"{prefix}step*.pt")))
|
||||
final = os.path.join(ckpt_dir, f"{prefix}final.pt")
|
||||
if os.path.exists(final):
|
||||
candidates.append(final)
|
||||
for path in candidates:
|
||||
try:
|
||||
st = torch.load(path, map_location="cpu", weights_only=False)
|
||||
s = int(st.get("step", _step_number(path)))
|
||||
except Exception:
|
||||
s = _step_number(path)
|
||||
if s > best_step:
|
||||
best_step = s
|
||||
best_path = path
|
||||
return best_path
|
||||
|
||||
|
||||
def _step_number(path: str) -> int:
|
||||
base = os.path.basename(path)
|
||||
num = "".join(ch for ch in base if ch.isdigit())
|
||||
return int(num) if num else 0
|
||||
|
||||
|
||||
def train(args: argparse.Namespace) -> None:
|
||||
device = "cuda" if torch.cuda.is_available() else "cpu"
|
||||
torch.manual_seed(args.seed)
|
||||
|
||||
# Decide resume BEFORE enabling LoRA: a stage2 resume ckpt carries its own
|
||||
# encoder/projector (so we skip the stage1 load); a fresh run loads stage1.
|
||||
resume_path = args.resume_from
|
||||
if not resume_path and args.resume:
|
||||
resume_path = _latest_step(args.ckpt_dir, "stage2_")
|
||||
final = os.path.join(args.ckpt_dir, "stage2_final.pt")
|
||||
if os.path.exists(final):
|
||||
if resume_path is None or _step_number(final) > _step_number(resume_path):
|
||||
resume_path = final
|
||||
|
||||
model = build_model().to(device)
|
||||
# When resuming, encoder/projector come from the stage2 ckpt (which already
|
||||
# absorbed stage1 + further SFT); only a fresh run bootstraps from stage1.
|
||||
if resume_path and os.path.exists(resume_path):
|
||||
st = torch.load(resume_path, map_location="cpu")
|
||||
model.encoder.load_state_dict(st["encoder"])
|
||||
model.projector.load_state_dict(st["projector"])
|
||||
resume_step = int(st.get("step", 0))
|
||||
print(f"[stage2] resumed weights from {resume_path} at step {resume_step}")
|
||||
elif args.stage1_ckpt and os.path.exists(args.stage1_ckpt):
|
||||
load_stage1_ckpt(model, args.stage1_ckpt)
|
||||
resume_step = 0
|
||||
else:
|
||||
resume_step = 0
|
||||
|
||||
model.freeze_llm()
|
||||
# Build LoRA structure the SAME way for fresh and resume (so the trainable
|
||||
# param set matches the saved optimizer state). For resume we then overlay
|
||||
# the saved adapter weights via load_adapter (which does NOT reset
|
||||
# requires_grad, unlike PeftModel.from_pretrained).
|
||||
model.enable_lora(r=args.lora_r)
|
||||
resume_lora_loaded = False
|
||||
if resume_path and os.path.exists(resume_path):
|
||||
lora_dir = os.path.join(args.ckpt_dir, "lora_adapter")
|
||||
if os.path.exists(lora_dir):
|
||||
try:
|
||||
from peft import set_peft_model_state_dict
|
||||
import json as _json
|
||||
# load adapter weights into the already-built LoRA model
|
||||
sd_path = os.path.join(lora_dir, "adapter_model.safetensors")
|
||||
if os.path.exists(sd_path):
|
||||
from safetensors.torch import load_file
|
||||
sd = load_file(sd_path)
|
||||
set_peft_model_state_dict(model.llm, sd)
|
||||
print(f"[stage2] resumed LoRA weights from {lora_dir}")
|
||||
resume_lora_loaded = True
|
||||
except Exception as e:
|
||||
print(f"[stage2] LoRA weight load failed ({e}), using fresh LoRA")
|
||||
model.train()
|
||||
if hasattr(model.llm, "gradient_checkpointing_enable"):
|
||||
model.llm.gradient_checkpointing_enable()
|
||||
if hasattr(model.llm, "enable_input_require_grads"):
|
||||
model.llm.enable_input_require_grads()
|
||||
|
||||
collator = Collator(max_T=args.ctx)
|
||||
|
||||
params = [p for p in model.parameters() if p.requires_grad]
|
||||
opt = torch.optim.AdamW(params, lr=args.lr)
|
||||
|
||||
# restore optimizer + EMA when resuming (LoRA already rebuilt above to match)
|
||||
ema = EMA(alpha=0.05)
|
||||
if resume_path and os.path.exists(resume_path):
|
||||
st = torch.load(resume_path, map_location="cpu")
|
||||
ts = st.get("train_state", {})
|
||||
if "optimizer" in ts:
|
||||
try:
|
||||
opt.load_state_dict(ts["optimizer"])
|
||||
for g in opt.param_groups:
|
||||
g["lr"] = args.lr
|
||||
print(f"[stage2] resumed optimizer from {resume_path}")
|
||||
except (ValueError, KeyError) as e:
|
||||
print(f"[stage2] optimizer state mismatch ({e}), fresh AdamW")
|
||||
if "ema" in ts:
|
||||
ema.value = ts["ema"]
|
||||
|
||||
os.makedirs(args.ckpt_dir, exist_ok=True)
|
||||
writer = None
|
||||
if args.tensorboard:
|
||||
from torch.utils.tensorboard import SummaryWriter
|
||||
writer = SummaryWriter(os.path.join(args.ckpt_dir, "tb"))
|
||||
|
||||
step = resume_step
|
||||
opt.zero_grad()
|
||||
accum = 0.0
|
||||
print(f"[stage2] training: max_steps={args.max_steps} bs={args.bs} "
|
||||
f"grad_accum={args.grad_accum} ctx={args.ctx} lora_r={args.lora_r} "
|
||||
f"start_step={step}")
|
||||
while step < args.max_steps:
|
||||
for batch_samples in batched(stream_jsonl(args.data), args.bs):
|
||||
if step >= args.max_steps:
|
||||
break
|
||||
batch = collator(batch_samples)
|
||||
out = model(batch)
|
||||
loss = out["loss"]
|
||||
(loss / args.grad_accum).backward()
|
||||
accum += float(loss)
|
||||
|
||||
if (step + 1) % args.grad_accum == 0:
|
||||
torch.nn.utils.clip_grad_norm_(params, 1.0)
|
||||
opt.step()
|
||||
opt.zero_grad()
|
||||
|
||||
ema.update(float(loss))
|
||||
if writer is not None:
|
||||
writer.add_scalar("sft_loss", float(loss), step)
|
||||
writer.add_scalar("sft_loss_ema", ema.value, step)
|
||||
|
||||
if step % args.log_every == 0:
|
||||
peak = (torch.cuda.max_memory_allocated() / 1e9) if torch.cuda.is_available() else 0.0
|
||||
print(f" step {step:5d} sft_loss={float(loss):.4f} ema={ema.value:.4f} peak_GB={peak:.2f}")
|
||||
|
||||
if (step + 1) % args.ckpt_every == 0:
|
||||
save_stage2_ckpt(model, opt, ema, args.ckpt_dir, step + 1)
|
||||
step += 1
|
||||
|
||||
save_stage2_ckpt(model, opt, ema, args.ckpt_dir, step, final=True)
|
||||
print(f"[stage2] done. saved to {args.ckpt_dir} avg_loss={accum/max(step,1):.4f} ema={ema.value:.4f}")
|
||||
if writer is not None:
|
||||
writer.close()
|
||||
|
||||
|
||||
def save_stage2_ckpt(model, opt, ema, ckpt_dir: str, step: int, final: bool = False) -> None:
|
||||
"""Save encoder/projector + LoRA adapter + optimizer + EMA for resume."""
|
||||
train_state = {
|
||||
"optimizer": opt.state_dict(),
|
||||
"ema": ema.value,
|
||||
}
|
||||
payload = {
|
||||
"encoder": model.encoder.state_dict(),
|
||||
"projector": model.projector.state_dict(),
|
||||
"step": step,
|
||||
"train_state": train_state,
|
||||
}
|
||||
name = "stage2_final.pt" if final else f"stage2_step{step}.pt"
|
||||
torch.save(payload, os.path.join(ckpt_dir, name))
|
||||
# save LoRA adapter separately (peft)
|
||||
try:
|
||||
model.llm.save_pretrained(os.path.join(ckpt_dir, "lora_adapter"))
|
||||
except Exception as e: # pragma: no cover - best-effort adapter dump
|
||||
print(f" (lora adapter save skipped: {e})")
|
||||
print(f" saved {name} ema={ema.value:.4f}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
p = argparse.ArgumentParser(description="Stage 2 SFT training (LoRA)")
|
||||
p.add_argument("--data", default="data/sft.jsonl")
|
||||
p.add_argument("--stage1_ckpt", default="checkpoints/stage1/stage1_final.pt")
|
||||
p.add_argument("--ckpt_dir", default="checkpoints/stage2")
|
||||
p.add_argument("--max_steps", type=int, default=10_000)
|
||||
p.add_argument("--bs", type=int, default=4)
|
||||
p.add_argument("--grad_accum", type=int, default=8)
|
||||
p.add_argument("--ctx", type=int, default=1024)
|
||||
p.add_argument("--lr", type=float, default=1e-4)
|
||||
p.add_argument("--lora_r", type=int, default=16)
|
||||
p.add_argument("--log_every", type=int, default=20)
|
||||
p.add_argument("--ckpt_every", type=int, default=2000)
|
||||
p.add_argument("--seed", type=int, default=0)
|
||||
p.add_argument("--tensorboard", action="store_true")
|
||||
p.add_argument("--resume", action="store_true",
|
||||
help="auto-pick latest step ckpt + lora_adapter and continue")
|
||||
p.add_argument("--resume_from", type=str, default=None,
|
||||
help="explicit ckpt path to resume from (overrides --resume)")
|
||||
args = p.parse_args()
|
||||
train(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,62 @@
|
||||
"""Tests for eval/baselines.py (T5.4)."""
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from tsmm.eval.baselines import (
|
||||
RandomBaseline,
|
||||
ConstantBaseline,
|
||||
AllReportBaseline,
|
||||
ts_to_text,
|
||||
)
|
||||
|
||||
|
||||
class TestTrivial:
|
||||
def test_random_in_range_and_shape(self):
|
||||
bl = RandomBaseline(seed=0)
|
||||
scores = bl.score(T=100)
|
||||
assert scores.shape == (100,)
|
||||
assert scores.min() >= 0.0 and scores.max() <= 1.0
|
||||
|
||||
def test_random_reproducible(self):
|
||||
a = RandomBaseline(seed=42).score(T=50)
|
||||
b = RandomBaseline(seed=42).score(T=50)
|
||||
assert np.allclose(a, b)
|
||||
|
||||
def test_constant_value(self):
|
||||
bl = ConstantBaseline(value=0.3)
|
||||
s = bl.score(T=10)
|
||||
assert s.shape == (10,)
|
||||
assert np.allclose(s, 0.3)
|
||||
|
||||
def test_all_report_is_all_ones(self):
|
||||
bl = AllReportBaseline()
|
||||
s = bl.score(T=8)
|
||||
assert s.shape == (8,)
|
||||
assert np.allclose(s, 1.0)
|
||||
|
||||
def test_all_report_vus_pr_not_inflated(self):
|
||||
from tsmm.eval.ts_metrics import vus_pr
|
||||
y = np.zeros(100, dtype=int); y[20:40] = 1
|
||||
s = AllReportBaseline().score(T=100)
|
||||
v = vus_pr(s, y)
|
||||
assert v == pytest.approx(20 / 100, abs=1e-3)
|
||||
|
||||
|
||||
class TestTsToText:
|
||||
def test_shape_and_content(self):
|
||||
series = np.arange(12).reshape(4, 3).tolist() # T=4, C=3
|
||||
text = ts_to_text(series, max_points=12)
|
||||
assert isinstance(text, str)
|
||||
assert "0" in text
|
||||
|
||||
def test_truncates_long_series(self):
|
||||
series = [[float(i)] for i in range(1000)]
|
||||
text = ts_to_text(series, max_points=50)
|
||||
# roughly bounded length
|
||||
assert len(text) < 5000
|
||||
|
||||
def test_handles_nan(self):
|
||||
import math
|
||||
series = [[1.0], [float("nan")], [3.0]]
|
||||
text = ts_to_text(series, max_points=10)
|
||||
assert "nan" in text.lower() or "missing" in text.lower()
|
||||
@@ -0,0 +1,97 @@
|
||||
"""Tests for Collator (T2.5): JSONL sample dicts → wrapper raw batch."""
|
||||
import math
|
||||
|
||||
import torch
|
||||
|
||||
from tsmm.data.collator import Collator
|
||||
|
||||
|
||||
def make_sample(T=8, C=3, with_nan=False, n_events=1, answer="no", instruction="Is there an anomaly?"):
|
||||
"""Build a dict mimicking the gen_pipeline JSONL schema."""
|
||||
series = [[float(i * C + c) for c in range(C)] for i in range(T)]
|
||||
if with_nan:
|
||||
series[1][0] = float("nan")
|
||||
return {
|
||||
"series": series, # [T, C]
|
||||
"attributes": [
|
||||
{"name": f"var{c}", "unit": "%", "freq": "1Hz"} for c in range(C)
|
||||
],
|
||||
"timestamps": [1700000000 + i for i in range(T)],
|
||||
"events": [{"t": 3, "text": "deploy"}] * n_events,
|
||||
"instruction": instruction,
|
||||
"answer": answer,
|
||||
"labels": [0] * T,
|
||||
}
|
||||
|
||||
|
||||
class TestCollator:
|
||||
def test_basic_shapes_and_types(self):
|
||||
col = Collator(max_T=8)
|
||||
batch = col([make_sample(T=8, C=3) for _ in range(4)])
|
||||
assert batch["series"].shape == (4, 8, 3)
|
||||
assert isinstance(batch["attributes"], list) and len(batch["attributes"]) == 4
|
||||
assert isinstance(batch["timestamps"], list) and len(batch["timestamps"]) == 4
|
||||
assert isinstance(batch["events"], list) and len(batch["events"]) == 4
|
||||
assert isinstance(batch["question"], list) and len(batch["question"]) == 4
|
||||
assert isinstance(batch["answer"], list) and len(batch["answer"]) == 4
|
||||
|
||||
def test_question_maps_from_instruction(self):
|
||||
col = Collator()
|
||||
batch = col([make_sample()])
|
||||
assert batch["question"][0] == "Is there an anomaly?"
|
||||
|
||||
def test_answer_passthrough(self):
|
||||
col = Collator()
|
||||
batch = col([make_sample(answer="yes")])
|
||||
assert batch["answer"][0] == "yes"
|
||||
|
||||
def test_no_nan_after_collate(self):
|
||||
col = Collator(nan_fill=0.0)
|
||||
batch = col([make_sample(T=8, C=3, with_nan=True)])
|
||||
assert not torch.isnan(batch["series"]).any()
|
||||
# the NaN position was filled with 0
|
||||
assert batch["series"][0, 1, 0].item() == 0.0
|
||||
|
||||
def test_truncates_long_T(self):
|
||||
col = Collator(max_T=16)
|
||||
batch = col([make_sample(T=32, C=2)])
|
||||
assert batch["series"].shape == (1, 16, 2)
|
||||
|
||||
def test_pads_short_T(self):
|
||||
col = Collator(max_T=64)
|
||||
batch = col([make_sample(T=16, C=2)])
|
||||
assert batch["series"].shape == (1, 64, 2)
|
||||
# padded region is zero
|
||||
assert batch["series"][0, 16:].abs().sum().item() == 0.0
|
||||
|
||||
def test_pads_variable_C_within_batch(self):
|
||||
col = Collator(max_T=16)
|
||||
batch = col([make_sample(T=16, C=2), make_sample(T=16, C=5)])
|
||||
# both padded to max_C=5
|
||||
assert batch["series"].shape == (2, 16, 5)
|
||||
|
||||
def test_attribute_string_is_descriptive(self):
|
||||
col = Collator()
|
||||
batch = col([make_sample(C=2)])
|
||||
s = batch["attributes"][0]
|
||||
assert "var0" in s and "var1" in s
|
||||
|
||||
def test_timestamps_string(self):
|
||||
col = Collator()
|
||||
batch = col([make_sample(T=8)])
|
||||
s = batch["timestamps"][0]
|
||||
assert isinstance(s, str) and "1700000000" in s
|
||||
|
||||
def test_events_string(self):
|
||||
col = Collator()
|
||||
batch = col([make_sample(n_events=2)])
|
||||
s = batch["events"][0]
|
||||
assert isinstance(s, str) and "deploy" in s
|
||||
|
||||
def test_answer_optional(self):
|
||||
# answer=None path (inference): collator should omit/None answer
|
||||
col = Collator()
|
||||
sample = make_sample()
|
||||
sample["answer"] = None
|
||||
batch = col([sample])
|
||||
assert batch.get("answer") is None or batch["answer"][0] is None
|
||||
@@ -0,0 +1,91 @@
|
||||
"""Unit tests for eval.instruct_check (T4.2) — CPU helpers only.
|
||||
|
||||
The model-driven ``evaluate`` path is exercised live in M4 exit validation.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
from tsmm.eval.instruct_check import (
|
||||
CATEGORIES,
|
||||
_forecast_list,
|
||||
load_by_category,
|
||||
)
|
||||
|
||||
|
||||
def _write_eval(tmpdir, rows):
|
||||
path = os.path.join(tmpdir, "eval.jsonl")
|
||||
with open(path, "w") as f:
|
||||
for r in rows:
|
||||
f.write(json.dumps(r) + "\n")
|
||||
return path
|
||||
|
||||
|
||||
def test_load_by_category_buckets_and_caps(tmp_path):
|
||||
rows = []
|
||||
for cat in CATEGORIES:
|
||||
for i in range(5):
|
||||
rows.append({"category": cat, "series": [[1.0]], "answer": "x", "labels": [0]})
|
||||
# plus an unknown-category row that must be ignored
|
||||
rows.append({"category": "mystery", "series": [[1.0]], "answer": "x"})
|
||||
path = _write_eval(str(tmp_path), rows)
|
||||
out = load_by_category(path, per_cat=3, seed=0)
|
||||
assert set(out.keys()) == set(CATEGORIES)
|
||||
for cat in CATEGORIES:
|
||||
assert len(out[cat]) == 3
|
||||
assert "mystery" not in out
|
||||
|
||||
|
||||
def test_load_by_category_deterministic_with_seed(tmp_path):
|
||||
rows = [{"category": "anomaly", "series": [[float(i)]], "answer": str(i), "labels": [0]}
|
||||
for i in range(20)]
|
||||
path = _write_eval(str(tmp_path), rows)
|
||||
a = load_by_category(path, per_cat=5, seed=42)
|
||||
b = load_by_category(path, per_cat=5, seed=42)
|
||||
assert [s["answer"] for s in a["anomaly"]] == [s["answer"] for s in b["anomaly"]]
|
||||
# different seed → (very likely) different selection
|
||||
c = load_by_category(path, per_cat=5, seed=7)
|
||||
assert [s["answer"] for s in a["anomaly"]] != [s["answer"] for s in c["anomaly"]]
|
||||
|
||||
|
||||
def test_load_by_category_handles_short_pool(tmp_path):
|
||||
rows = [{"category": "event", "series": [[1.0]], "answer": "x"} for _ in range(2)]
|
||||
path = _write_eval(str(tmp_path), rows)
|
||||
out = load_by_category(path, per_cat=10, seed=0)
|
||||
assert len(out["event"]) == 2
|
||||
# missing categories are empty lists, not errors
|
||||
assert out["anomaly"] == []
|
||||
|
||||
|
||||
def test_load_by_category_skips_blank_lines(tmp_path):
|
||||
path = os.path.join(str(tmp_path), "eval.jsonl")
|
||||
with open(path, "w") as f:
|
||||
f.write(json.dumps({"category": "anomaly", "series": [[1.0]], "answer": "x"}) + "\n")
|
||||
f.write("\n")
|
||||
f.write(" \n")
|
||||
f.write(json.dumps({"category": "anomaly", "series": [[2.0]], "answer": "y"}) + "\n")
|
||||
out = load_by_category(path, per_cat=5, seed=0)
|
||||
assert len(out["anomaly"]) == 2
|
||||
|
||||
|
||||
def test_forecast_list_from_nested_json():
|
||||
ref = json.dumps({"forecast": [[1.0, 2.0, 3.0]]})
|
||||
assert _forecast_list(ref) == [1.0, 2.0, 3.0]
|
||||
|
||||
|
||||
def test_forecast_list_from_flat_json():
|
||||
ref = json.dumps({"forecast": [4.0, 5.0]})
|
||||
assert _forecast_list(ref) == [4.0, 5.0]
|
||||
|
||||
|
||||
def test_forecast_list_accepts_dict_input():
|
||||
assert _forecast_list({"forecast": [[1.0, 2.0]]}) == [1.0, 2.0]
|
||||
|
||||
|
||||
def test_forecast_list_returns_empty_on_garbage():
|
||||
assert _forecast_list("not json") == []
|
||||
assert _forecast_list(json.dumps({"no_forecast": 1})) == []
|
||||
assert _forecast_list("") == []
|
||||
assert _forecast_list(None) == []
|
||||
@@ -0,0 +1,80 @@
|
||||
"""Tests for train/losses.py (T3.1)."""
|
||||
import pytest
|
||||
import torch
|
||||
|
||||
from tsmm.train.losses import lm_loss, infonce_contrastive_loss
|
||||
|
||||
|
||||
class TestLMLoss:
|
||||
def test_scalar_and_finite(self):
|
||||
logits = torch.randn(2, 10, 100)
|
||||
labels = torch.full((2, 10), -100, dtype=torch.long)
|
||||
labels[:, 5:] = torch.randint(0, 100, (2, 5))
|
||||
loss = lm_loss(logits, labels)
|
||||
assert loss.dim() == 0
|
||||
assert torch.isfinite(loss)
|
||||
|
||||
def test_all_masked_returns_zero(self):
|
||||
logits = torch.randn(2, 10, 100)
|
||||
labels = torch.full((2, 10), -100, dtype=torch.long)
|
||||
loss = lm_loss(logits, labels)
|
||||
assert float(loss) == 0.0
|
||||
|
||||
def test_shift_correctness(self):
|
||||
# CE with shift: logits[t] predicts labels[t+1]
|
||||
torch.manual_seed(0)
|
||||
V = 5
|
||||
logits = torch.zeros(1, 3, V)
|
||||
# make position 0 strongly predict token 2
|
||||
logits[0, 0, 2] = 10.0
|
||||
labels = torch.full((1, 3), -100, dtype=torch.long)
|
||||
labels[0, 1] = 2 # label at position 1 (shifted: predicted from logits[0])
|
||||
loss = lm_loss(logits, labels)
|
||||
assert float(loss) < 0.01 # near-zero when prediction is confident & correct
|
||||
|
||||
def test_gradient_flows(self):
|
||||
logits = torch.randn(2, 8, 50, requires_grad=True)
|
||||
labels = torch.full((2, 8), -100, dtype=torch.long)
|
||||
labels[:, 4:] = torch.randint(0, 50, (2, 4))
|
||||
loss = lm_loss(logits, labels)
|
||||
loss.backward()
|
||||
assert logits.grad is not None
|
||||
assert logits.grad.abs().sum() > 0
|
||||
|
||||
|
||||
class TestInfoNCE:
|
||||
def test_scalar_finite(self):
|
||||
z1 = torch.randn(4, 16)
|
||||
z2 = torch.randn(4, 16)
|
||||
loss = infonce_contrastive_loss(z1, z2)
|
||||
assert loss.dim() == 0
|
||||
assert torch.isfinite(loss)
|
||||
|
||||
def test_identical_pairs_low_loss(self):
|
||||
torch.manual_seed(0)
|
||||
z = torch.randn(8, 32)
|
||||
z = torch.nn.functional.normalize(z, dim=-1)
|
||||
loss_same = infonce_contrastive_loss(z, z, temperature=0.1)
|
||||
loss_rand = infonce_contrastive_loss(z, torch.randn(8, 32), temperature=0.1)
|
||||
assert float(loss_same) < float(loss_rand)
|
||||
|
||||
def test_symmetric(self):
|
||||
z1 = torch.randn(6, 24)
|
||||
z2 = torch.randn(6, 24)
|
||||
a = infonce_contrastive_loss(z1, z2)
|
||||
b = infonce_contrastive_loss(z2, z1)
|
||||
assert torch.allclose(a, b, atol=1e-5)
|
||||
|
||||
def test_gradient_flows(self):
|
||||
z1 = torch.randn(4, 16, requires_grad=True)
|
||||
z2 = torch.randn(4, 16, requires_grad=True)
|
||||
loss = infonce_contrastive_loss(z1, z2)
|
||||
loss.backward()
|
||||
assert z1.grad is not None and z2.grad is not None
|
||||
|
||||
def test_l2_normalized(self):
|
||||
# loss should be invariant to per-sample scale (normalize inside)
|
||||
z1 = torch.randn(4, 16)
|
||||
loss_a = infonce_contrastive_loss(z1, z1)
|
||||
loss_b = infonce_contrastive_loss(z1 * 5.0, z1 * 5.0)
|
||||
assert torch.allclose(loss_a, loss_b, atol=1e-4)
|
||||
@@ -0,0 +1,136 @@
|
||||
"""Tests for multimodal splice (T2.3).
|
||||
|
||||
Verifies the splice [属性][时间戳][TS tok][事件][问题][回答] produces:
|
||||
- inputs_embeds [1, seq, hidden] with seq_len <= max_len
|
||||
- attention_mask [1, seq] all ones (unpadded single sample)
|
||||
- labels [1, seq] with -100 everywhere EXCEPT the answer segment
|
||||
|
||||
Uses the real Qwen2.5-0.5B tokenizer (local) + a stub embedding layer so the
|
||||
splice *logic* is validated without loading the full LLM weights.
|
||||
"""
|
||||
import pytest
|
||||
import torch
|
||||
from transformers import AutoTokenizer
|
||||
|
||||
from tsmm.model.multimodal import MultimodalSplicer
|
||||
|
||||
TOKENIZER_PATH = "/home/zhangzp/models/Qwen2.5-0.5B-Instruct"
|
||||
HIDDEN = 896
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def tokenizer():
|
||||
return AutoTokenizer.from_pretrained(TOKENIZER_PATH)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def stub_embed(tokenizer):
|
||||
# mimic LLM get_input_embeddings(): Embedding(vocab, hidden).
|
||||
# Use len(tokenizer) which includes added special tokens (eos=151645 etc.);
|
||||
# tokenizer.vocab_size (151643) excludes them and is too small.
|
||||
return torch.nn.Embedding(len(tokenizer), HIDDEN)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def splicer(tokenizer, stub_embed):
|
||||
s = MultimodalSplicer(tokenizer, hidden_size=HIDDEN, max_len=1024)
|
||||
s.bind(stub_embed)
|
||||
return s
|
||||
|
||||
|
||||
def make_ts_embeds(n_patches=127, hidden=HIDDEN):
|
||||
return torch.randn(n_patches, hidden)
|
||||
|
||||
|
||||
class TestSplice:
|
||||
def test_shapes(self, splicer):
|
||||
out = splicer.splice(
|
||||
attributes="变量：CPU利用率 单位：% 采样率：1Hz",
|
||||
timestamps="时间范围：2026-01-01 00:00 ~ 00:08",
|
||||
ts_embeds=make_ts_embeds(127),
|
||||
events="",
|
||||
question="描述这段时序的整体趋势",
|
||||
answer="整体呈上升趋势。",
|
||||
)
|
||||
assert out.inputs_embeds.dim() == 3
|
||||
assert out.inputs_embeds.shape[0] == 1
|
||||
assert out.inputs_embeds.shape[2] == HIDDEN
|
||||
seq = out.inputs_embeds.shape[1]
|
||||
assert out.attention_mask.shape == (1, seq)
|
||||
assert out.labels.shape == (1, seq)
|
||||
# within context budget
|
||||
assert seq <= 1024
|
||||
|
||||
def test_attention_mask_all_ones_unpadded(self, splicer):
|
||||
out = splicer.splice(
|
||||
attributes="CPU", timestamps="t0..t1", ts_embeds=make_ts_embeds(10),
|
||||
events="", question="Q?", answer="A.",
|
||||
)
|
||||
assert torch.all(out.attention_mask == 1)
|
||||
|
||||
def test_labels_only_answer_non_masked(self, splicer, tokenizer):
|
||||
out = splicer.splice(
|
||||
attributes="CPU", timestamps="t0", ts_embeds=make_ts_embeds(10),
|
||||
events="", question="Q?", answer="上升趋势",
|
||||
)
|
||||
labels = out.labels[0]
|
||||
# there must be at least one non -100 position (the answer)
|
||||
assert (labels != -100).sum() > 0
|
||||
# all non-answer positions must be -100
|
||||
n_non_masked = int((labels != -100).sum())
|
||||
# the answer token count should match tokenizing answer (+ eos)
|
||||
ans_ids = tokenizer("上升趋势", add_special_tokens=False)["input_ids"]
|
||||
# +1 for appended eos
|
||||
assert n_non_masked == len(ans_ids) + 1
|
||||
|
||||
def test_no_answer_means_all_masked(self, splicer):
|
||||
# inference mode: no answer → all labels -100
|
||||
out = splicer.splice(
|
||||
attributes="CPU", timestamps="t0", ts_embeds=make_ts_embeds(10),
|
||||
events="", question="Q?", answer=None,
|
||||
)
|
||||
assert torch.all(out.labels == -100)
|
||||
|
||||
def test_ts_tokens_in_sequence(self, splicer):
|
||||
# the TS token block length should appear in the total seq length
|
||||
n_patches = 50
|
||||
out_no_ts = splicer.splice(
|
||||
attributes="CPU", timestamps="t0", ts_embeds=torch.zeros(0, HIDDEN),
|
||||
events="", question="Q?", answer="A.",
|
||||
)
|
||||
out_with_ts = splicer.splice(
|
||||
attributes="CPU", timestamps="t0", ts_embeds=make_ts_embeds(n_patches),
|
||||
events="", question="Q?", answer="A.",
|
||||
)
|
||||
delta = out_with_ts.inputs_embeds.shape[1] - out_no_ts.inputs_embeds.shape[1]
|
||||
assert delta == n_patches
|
||||
|
||||
def test_seq_len_within_budget(self, splicer):
|
||||
out = splicer.splice(
|
||||
attributes="CPU 内存 磁盘 网络 温度 请求量",
|
||||
timestamps="2026-01-01 00:00:00 至 2026-01-01 00:08:32 UTC",
|
||||
ts_embeds=make_ts_embeds(127),
|
||||
events="t=300 处发生部署事件：服务重启",
|
||||
question="请判断这段时序是否存在异常，并以 JSON 给出异常区间。",
|
||||
answer='{"segments": [[300, 350]], "reason": "部署后延迟突增"}',
|
||||
)
|
||||
assert out.inputs_embeds.shape[1] <= 1024
|
||||
|
||||
def test_no_nan(self, splicer):
|
||||
out = splicer.splice(
|
||||
attributes="CPU", timestamps="t0", ts_embeds=make_ts_embeds(20),
|
||||
events="", question="Q?", answer="A.",
|
||||
)
|
||||
assert not torch.isnan(out.inputs_embeds).any()
|
||||
|
||||
def test_truncation_when_overlong(self, splicer):
|
||||
# force an overlong sequence; splicer should truncate to max_len
|
||||
splicer.max_len = 40
|
||||
out = splicer.splice(
|
||||
attributes="CPU", timestamps="t0",
|
||||
ts_embeds=make_ts_embeds(127),
|
||||
events="", question="Q?", answer="A.",
|
||||
)
|
||||
assert out.inputs_embeds.shape[1] <= 40
|
||||
assert out.attention_mask.shape == (1, out.inputs_embeds.shape[1])
|
||||
assert out.labels.shape == (1, out.inputs_embeds.shape[1])
|
||||
@@ -0,0 +1,135 @@
|
||||
"""Tests for eval/parse_answer.py (T5.1)."""
|
||||
import json
|
||||
|
||||
import numpy as np
|
||||
|
||||
from tsmm.eval.parse_answer import parse_anomaly_segments, answer_to_point_scores
|
||||
|
||||
|
||||
class TestParseSegments:
|
||||
def test_clean_json(self):
|
||||
text = '{"segments": [[10, 20], [30, 40]], "type": "spike"}'
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
assert ok is True
|
||||
assert segs == [(10, 20), (30, 40)]
|
||||
|
||||
def test_json_in_prose(self):
|
||||
text = 'The anomalies are: {"segments": [[5, 8]], "type":"level_shift"} per your request.'
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
assert ok is True
|
||||
assert segs == [(5, 8)]
|
||||
|
||||
def test_nested_brackets(self):
|
||||
text = '{"segments": [[320, 360]], "type":"spike", "score":0.9}'
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
assert ok is True
|
||||
assert segs == [(320, 360)]
|
||||
|
||||
def test_multiple_json_objects(self):
|
||||
# when multiple JSON-ish blobs exist, take the one with a segments key
|
||||
text = 'some text {"foo": 1} then {"segments":[[1,2]]}'
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
assert ok is True
|
||||
assert segs == [(1, 2)]
|
||||
|
||||
def test_no_json_returns_empty_fail(self):
|
||||
text = "I cannot find any anomaly."
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
assert ok is False
|
||||
assert segs == []
|
||||
|
||||
def test_malformed_json_fallback_to_regex(self):
|
||||
# if JSON parse fails but [s,e] patterns exist, regex-extract them
|
||||
text = 'anomalies at [100, 150] and [200, 250]'
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
# ok=False because no valid JSON, but segments extracted via fallback
|
||||
assert segs == [(100, 150), (200, 250)]
|
||||
|
||||
def test_anomaly_regions_schema(self):
|
||||
# the schema data/instruct.py actually emits (nested, start/end objects)
|
||||
text = '{"anomaly_regions": [{"start": 10, "end": 93}, {"start": 245, "end": 286}], "n_regions": 2}'
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
assert ok is True
|
||||
assert segs == [(10, 93), (245, 286)]
|
||||
|
||||
def test_anomaly_regions_in_prose(self):
|
||||
text = 'Based on the series: {"anomaly_regions": [{"start": 5, "end": 8}]} done.'
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
assert ok is True
|
||||
assert segs == [(5, 8)]
|
||||
|
||||
def test_segments_with_start_end_objects(self):
|
||||
# also accept the {start,end} object form under the 'segments' key
|
||||
text = '{"segments": [{"start": 1, "end": 4}]}'
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
assert ok is True
|
||||
assert segs == [(1, 4)]
|
||||
|
||||
def test_balanced_json_handles_nested(self):
|
||||
from tsmm.eval.parse_answer import _extract_balanced_json
|
||||
text = 'x {"a": {"b": 1}, "c": [2,3]} y {"d": 4}'
|
||||
blobs = _extract_balanced_json(text)
|
||||
# outer nested object + the flat one, but NOT the inner {"b":1} alone
|
||||
assert any('"a"' in b and '"b"' in b for b in blobs)
|
||||
assert '{"d": 4}' in blobs
|
||||
|
||||
def test_non_numeric_segment_entries_are_skipped(self):
|
||||
# a pure-LLM may emit junk like ["CPU Usage", 30] — must not crash, just skip
|
||||
from tsmm.eval.parse_answer import _seg_from_entry
|
||||
assert _seg_from_entry(["CPU Usage", 30]) is None
|
||||
assert _seg_from_entry({"start": "foo", "end": "bar"}) is None
|
||||
assert _seg_from_entry("not a segment") is None
|
||||
# good entries alongside junk are still extracted
|
||||
text = '{"segments": [["CPU Usage", 30], [10, 20]]}'
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
assert ok is True
|
||||
assert segs == [(10, 20)]
|
||||
|
||||
def test_empty_segments_list(self):
|
||||
text = '{"segments": [], "type": "none"}'
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
assert ok is True
|
||||
assert segs == []
|
||||
|
||||
def test_single_int_segment(self):
|
||||
# segments might be [start] only or [[s]] — be lenient
|
||||
text = '{"segments": [[42]]}'
|
||||
segs, ok = parse_anomaly_segments(text)
|
||||
assert ok is True
|
||||
# single-element segment treated as point
|
||||
assert segs == [(42, 42)]
|
||||
|
||||
|
||||
class TestAnswerToPointScores:
|
||||
def test_basic(self):
|
||||
text = '{"segments": [[2, 5]]}'
|
||||
scores, ok = answer_to_point_scores(text, T=10)
|
||||
assert ok is True
|
||||
assert scores.shape == (10,)
|
||||
assert list(scores) == [0, 0, 1, 1, 1, 1, 0, 0, 0, 0]
|
||||
|
||||
def test_multiple_segments(self):
|
||||
text = '{"segments": [[0, 2], [7, 9]]}'
|
||||
scores, ok = answer_to_point_scores(text, T=10)
|
||||
assert ok is True
|
||||
assert list(scores) == [1, 1, 1, 0, 0, 0, 0, 1, 1, 1]
|
||||
|
||||
def test_no_anomaly_all_zero(self):
|
||||
text = '{"segments": []}'
|
||||
scores, ok = answer_to_point_scores(text, T=8)
|
||||
assert ok is True
|
||||
assert float(scores.sum()) == 0.0
|
||||
|
||||
def test_parse_fail_all_zero(self):
|
||||
text = "no idea"
|
||||
scores, ok = answer_to_point_scores(text, T=8)
|
||||
assert ok is False
|
||||
assert scores.shape == (8,)
|
||||
assert float(scores.sum()) == 0.0
|
||||
|
||||
def test_clips_to_T(self):
|
||||
# segment beyond T is clipped
|
||||
text = '{"segments": [[5, 100]]}'
|
||||
scores, ok = answer_to_point_scores(text, T=10)
|
||||
assert ok is True
|
||||
assert float(scores.sum()) == 5.0 # indices 5..9
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Tests for Projector (T2.2): TS token d → LLM hidden."""
|
||||
import torch
|
||||
|
||||
from tsmm.model.projector import Projector
|
||||
|
||||
|
||||
class TestProjector:
|
||||
def test_output_shape(self):
|
||||
proj = Projector(in_dim=256, out_dim=896)
|
||||
x = torch.randn(2, 127, 256)
|
||||
out = proj(x)
|
||||
assert out.shape == (2, 127, 896)
|
||||
|
||||
def test_aligns_qwen_hidden(self):
|
||||
# Qwen2.5-0.5B hidden = 896
|
||||
proj = Projector(in_dim=256, out_dim=896)
|
||||
out = proj(torch.randn(1, 127, 256))
|
||||
assert out.shape[-1] == 896
|
||||
|
||||
def test_has_layernorm(self):
|
||||
proj = Projector(in_dim=256, out_dim=896)
|
||||
assert any(isinstance(m, torch.nn.LayerNorm) for m in proj.modules())
|
||||
|
||||
def test_gradient_flows(self):
|
||||
proj = Projector(in_dim=256, out_dim=896)
|
||||
x = torch.randn(1, 31, 256)
|
||||
out = proj(x)
|
||||
out.sum().backward()
|
||||
assert proj.linear.weight.grad is not None
|
||||
assert proj.linear.weight.grad.abs().sum() > 0
|
||||
|
||||
def test_no_nan(self):
|
||||
proj = Projector(in_dim=256, out_dim=896)
|
||||
out = proj(torch.randn(4, 128, 256))
|
||||
assert not torch.isnan(out).any()
|
||||
@@ -0,0 +1,105 @@
|
||||
"""Tests for eval/qa_judge.py (T5.3)."""
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from tsmm.eval.qa_judge import (
|
||||
RuleJudge,
|
||||
rule_score_describe,
|
||||
rule_score_anomaly,
|
||||
rule_score_forecast,
|
||||
)
|
||||
|
||||
|
||||
class TestDescribeRule:
|
||||
def test_numerical_within_tolerance(self):
|
||||
# predicted mean within 5% of reference → 1.0
|
||||
out = rule_score_describe(
|
||||
pred='{"mean": 50.0, "std": 2.0}',
|
||||
ref={"mean": 50.5, "std": 2.1},
|
||||
rtol=0.05,
|
||||
)
|
||||
assert out["score"] == 1.0
|
||||
assert out["parsed"] is True
|
||||
|
||||
def test_outside_tolerance(self):
|
||||
out = rule_score_describe(
|
||||
pred='{"mean": 80.0}',
|
||||
ref={"mean": 50.0},
|
||||
rtol=0.05,
|
||||
)
|
||||
assert out["score"] == 0.0
|
||||
assert out["parsed"] is True
|
||||
|
||||
def test_unparseable(self):
|
||||
out = rule_score_describe(pred="the mean is around 50", ref={"mean": 50.0})
|
||||
assert out["parsed"] is False
|
||||
assert out["score"] == 0.0
|
||||
|
||||
|
||||
class TestAnomalyRule:
|
||||
def test_exact_match(self):
|
||||
pred = '{"segments": [[10, 20]]}'
|
||||
ref_segs = [(10, 20)]
|
||||
out = rule_score_anomaly(pred, ref_segs, T=50)
|
||||
assert out["score"] == 1.0
|
||||
assert out["parsed"] is True
|
||||
|
||||
def test_iou_partial(self):
|
||||
pred = '{"segments": [[10, 30]]}'
|
||||
ref_segs = [(20, 40)]
|
||||
out = rule_score_anomaly(pred, ref_segs, T=50)
|
||||
assert 0.0 < out["score"] < 1.0
|
||||
|
||||
def test_no_overlap(self):
|
||||
pred = '{"segments": [[0, 5]]}'
|
||||
ref_segs = [(20, 40)]
|
||||
out = rule_score_anomaly(pred, ref_segs, T=50)
|
||||
assert out["score"] == 0.0
|
||||
|
||||
|
||||
class TestForecastRule:
|
||||
def test_within_tolerance(self):
|
||||
pred = '{"forecast": [1.0, 2.0, 3.0]}'
|
||||
ref = [1.05, 1.95, 3.1]
|
||||
out = rule_score_forecast(pred, ref, rtol=0.1)
|
||||
assert out["score"] == 1.0
|
||||
assert out["parsed"] is True
|
||||
|
||||
def test_one_off(self):
|
||||
pred = '{"forecast": [1.0, 2.0, 9.0]}'
|
||||
ref = [1.0, 2.0, 3.0]
|
||||
out = rule_score_forecast(pred, ref, rtol=0.1)
|
||||
assert out["score"] < 1.0
|
||||
|
||||
|
||||
class TestRuleJudgeDispatch:
|
||||
def test_judge_anomaly(self):
|
||||
judge = RuleJudge()
|
||||
out = judge.judge("anomaly", '{"segments": [[5, 8]]}', ref_segs=[(5, 8)], T=20)
|
||||
assert out["score"] == 1.0
|
||||
assert out["track"] == "rule"
|
||||
|
||||
def test_judge_unknown_category_skips(self):
|
||||
judge = RuleJudge()
|
||||
out = judge.judge("root_cause", "some text answer", ref_answer="some text answer")
|
||||
assert out["track"] == "skip"
|
||||
assert out["score"] is None
|
||||
|
||||
|
||||
class TestLLMJudgeStub:
|
||||
def test_stub_returns_deterministic(self):
|
||||
from tsmm.eval.qa_judge import LLMJudge
|
||||
judge = LLMJudge(backend="stub")
|
||||
s1 = judge.score("Q?", "ref", "pred")
|
||||
s2 = judge.score("Q?", "ref", "pred")
|
||||
assert isinstance(s1, dict)
|
||||
assert "score" in s1 and 0 <= s1["score"] <= 5
|
||||
assert s1 == s2 # deterministic stub
|
||||
|
||||
def test_stub_rewards_exact_match(self):
|
||||
from tsmm.eval.qa_judge import LLMJudge
|
||||
judge = LLMJudge(backend="stub")
|
||||
same = judge.score("Q?", "正常", "正常")["score"]
|
||||
diff = judge.score("Q?", "正常", "完全不同的乱码回答xyz")["score"]
|
||||
assert same >= diff
|
||||
@@ -55,7 +55,8 @@ def test_load_windows_unknown_dataset_raises():
|
||||
|
||||
def test_load_windows_real_dataset_missing_data_raises_clear_error():
|
||||
# Real loaders must raise a helpful error when the local data path is absent,
|
||||
# rather than silently returning empty arrays.
|
||||
# rather than silently returning empty arrays. Use a registered dataset name
|
||||
# whose files are guaranteed absent (not 'smd', which may be present after
|
||||
# the real-benchmark download step).
|
||||
with pytest.raises((FileNotFoundError, RuntimeError)):
|
||||
# SMD is almost certainly not present in CI; should raise.
|
||||
real_bench.load_windows("smd", T=64, stride=32)
|
||||
real_bench.load_windows("swat", T=64, stride=32)
|
||||
|
||||
@@ -127,3 +127,22 @@ def test_add_missing_reproducible():
|
||||
a = add_missing(s, rate=0.1, seed=3)
|
||||
b = add_missing(s, rate=0.1, seed=3)
|
||||
assert np.array_equal(a, b, equal_nan=True)
|
||||
|
||||
|
||||
# ─── regression: _random_window end >= start (found during 50k generation) ─
|
||||
def test_inject_anomaly_end_ge_start_across_many_seeds():
|
||||
"""Regression: _random_window fallback used to draw end independently,
|
||||
producing end < start (ValueError in linspace). Sweep many seeds + types."""
|
||||
import numpy as np
|
||||
s0 = generate_series(T=512, C=5, seed=0)
|
||||
for seed in range(300):
|
||||
for types in (
|
||||
["drift"], ["level_shift"], ["variance_change"],
|
||||
["drift", "level_shift", "variance_change", "spike"],
|
||||
):
|
||||
s = s0.copy()
|
||||
out, labels, segments = inject_anomaly(s, types=types, seed=seed)
|
||||
for seg in segments:
|
||||
assert seg["end"] >= seg["start"], (seed, types, seg)
|
||||
assert out.shape == s.shape
|
||||
assert labels.shape == (512,)
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
"""Tests for TS Encoder (T2.1).
|
||||
|
||||
Spec clarification notes (recorded in tasks.md / commit):
|
||||
- n_patches = (T - patch_len) // stride + 1 → T=512,P=8,S=4 ⇒ 127 (design's
|
||||
"128" was an arithmetic slip; downstream is channel-wise so 1-token diff is nil).
|
||||
- "Channel-independent" = shared per-channel patch embedding + mean-pool over
|
||||
channels → output [B, n_patches, d]. Channel identity carried by [属性] text tokens.
|
||||
- Actual params ≈1.6M for a compact 2-layer d=256 encoder (design's "≈4M" was rough).
|
||||
"""
|
||||
import torch
|
||||
|
||||
from tsmm.model.ts_encoder import Patchify, TSEncoder
|
||||
|
||||
|
||||
class TestPatchify:
|
||||
def test_basic_shape(self):
|
||||
pf = Patchify(patch_len=8, stride=4)
|
||||
x = torch.randn(2, 512, 5)
|
||||
patches = pf(x)
|
||||
assert patches.shape == (2, 127, 5, 8)
|
||||
|
||||
def test_n_patches_formula(self):
|
||||
pf = Patchify(patch_len=8, stride=4)
|
||||
for T, expected in [(512, 127), (128, 31), (256, 63), (16, 3)]:
|
||||
x = torch.randn(1, T, 3)
|
||||
assert pf(x).shape[1] == expected, f"T={T}"
|
||||
|
||||
def test_non_overlapping(self):
|
||||
# stride == patch_len ⇒ non-overlapping
|
||||
pf = Patchify(patch_len=8, stride=8)
|
||||
x = torch.randn(1, 512, 4)
|
||||
assert pf(x).shape == (1, 64, 4, 8)
|
||||
|
||||
def test_values_are_correct_windows(self):
|
||||
# patch i covers x[:, i*stride : i*stride+patch_len, :]
|
||||
pf = Patchify(patch_len=4, stride=2)
|
||||
x = torch.arange(12, dtype=torch.float).view(1, 12, 1)
|
||||
patches = pf(x) # [1, n_patches=5, 1, 4]
|
||||
# patch 0 = x[0:4]
|
||||
assert torch.allclose(patches[0, 0, 0], torch.tensor([0.0, 1.0, 2.0, 3.0]))
|
||||
# patch 1 = x[2:6]
|
||||
assert torch.allclose(patches[0, 1, 0], torch.tensor([2.0, 3.0, 4.0, 5.0]))
|
||||
|
||||
|
||||
class TestTSEncoder:
|
||||
def test_output_shape(self):
|
||||
enc = TSEncoder(d=256, layers=2, heads=4, patch_len=8, stride=4)
|
||||
x = torch.randn(2, 512, 5)
|
||||
out = enc(x)
|
||||
assert out.shape == (2, 127, 256)
|
||||
|
||||
def test_different_T(self):
|
||||
enc = TSEncoder(d=256, layers=2, heads=4, patch_len=8, stride=4)
|
||||
for T, n in [(128, 31), (256, 63), (512, 127)]:
|
||||
out = enc(torch.randn(1, T, 3))
|
||||
assert out.shape == (1, n, 256)
|
||||
|
||||
def test_different_C(self):
|
||||
enc = TSEncoder(d=256, layers=2, heads=4, patch_len=8, stride=4)
|
||||
for C in [1, 5, 17]:
|
||||
out = enc(torch.randn(1, 512, C))
|
||||
assert out.shape == (1, 127, 256) # output C-free
|
||||
|
||||
def test_param_count_reasonable(self):
|
||||
enc = TSEncoder(d=256, layers=2, heads=4, patch_len=8, stride=4)
|
||||
n_params = sum(p.numel() for p in enc.parameters())
|
||||
# compact 2-layer d=256 encoder ≈ 1.6M; sanity range 0.5M–6M
|
||||
assert 500_000 < n_params < 6_000_000, f"params={n_params}"
|
||||
|
||||
def test_deterministic_in_eval(self):
|
||||
enc = TSEncoder(d=256, layers=2, heads=4, patch_len=8, stride=4).eval()
|
||||
x = torch.randn(1, 128, 3)
|
||||
with torch.no_grad():
|
||||
o1 = enc(x)
|
||||
o2 = enc(x)
|
||||
assert torch.allclose(o1, o2)
|
||||
|
||||
def test_gradient_flows(self):
|
||||
enc = TSEncoder(d=256, layers=2, heads=4, patch_len=8, stride=4)
|
||||
x = torch.randn(1, 128, 3, requires_grad=False)
|
||||
out = enc(x)
|
||||
loss = out.pow(2).sum()
|
||||
loss.backward()
|
||||
grads = [p.grad for p in enc.parameters() if p.grad is not None]
|
||||
assert all(g is not None for g in grads)
|
||||
assert any(g.abs().sum() > 0 for g in grads)
|
||||
@@ -0,0 +1,109 @@
|
||||
"""Tests for eval/ts_metrics.py (T5.2)."""
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from tsmm.eval.ts_metrics import (
|
||||
affiliation_f1,
|
||||
auc_pr,
|
||||
point_f1,
|
||||
point_f1_with_pa,
|
||||
vus_pr,
|
||||
)
|
||||
|
||||
|
||||
def labels(segs, T=100):
|
||||
y = np.zeros(T, dtype=int)
|
||||
for a, b in segs:
|
||||
y[a:b + 1] = 1
|
||||
return y
|
||||
|
||||
|
||||
class TestPointF1:
|
||||
def test_perfect(self):
|
||||
y = labels([(20, 40)], T=100)
|
||||
s = np.zeros(100); s[20:41] = 1.0
|
||||
f1, prec, rec = point_f1(s, y)
|
||||
assert f1 == pytest.approx(1.0, abs=1e-6)
|
||||
|
||||
def test_no_overlap(self):
|
||||
y = labels([(20, 40)])
|
||||
s = np.zeros(100); s[60:70] = 1.0
|
||||
f1, _, _ = point_f1(s, y)
|
||||
assert f1 == pytest.approx(0.0, abs=1e-6)
|
||||
|
||||
def test_partial(self):
|
||||
y = labels([(20, 40)])
|
||||
s = np.zeros(100); s[30:41] = 0.9
|
||||
f1, prec, rec = point_f1(s, y, threshold=0.5)
|
||||
# predicted 11 anomalous all in true region: precision 1, recall 11/21
|
||||
assert prec == pytest.approx(1.0, abs=1e-6)
|
||||
assert rec == pytest.approx(11 / 21, abs=1e-3)
|
||||
|
||||
|
||||
class TestAUCPR:
|
||||
def test_perfect_ranking(self):
|
||||
y = labels([(10, 20)])
|
||||
s = np.zeros(100); s[10:21] = 1.0
|
||||
assert auc_pr(s, y) == pytest.approx(1.0, abs=1e-6)
|
||||
|
||||
def test_inverse_ranking_low(self):
|
||||
y = labels([(10, 20)])
|
||||
s = np.ones(100); s[10:21] = 0.0 # anomalies score LOW
|
||||
assert auc_pr(s, y) < 0.5
|
||||
|
||||
|
||||
class TestPAF1:
|
||||
def test_pa_inflates(self):
|
||||
# PA: if ANY predicted point is in the true anomaly, the whole segment
|
||||
# counts as caught → much higher F1 than point-wise
|
||||
y = labels([(20, 40)])
|
||||
s = np.zeros(100); s[40] = 1.0 # single overlap point
|
||||
f1_pa, _, _ = point_f1_with_pa(s, y)
|
||||
f1_pt, _, _ = point_f1(s, y)
|
||||
assert f1_pa > f1_pt
|
||||
|
||||
|
||||
class TestAffiliationF1:
|
||||
def test_perfect(self):
|
||||
y = labels([(20, 40)])
|
||||
s = np.zeros(100); s[20:41] = 1.0
|
||||
f1, _, _ = affiliation_f1(s, y)
|
||||
assert f1 == pytest.approx(1.0, abs=1e-6)
|
||||
|
||||
def test_far_predictions_lower(self):
|
||||
y = labels([(20, 40)])
|
||||
s_near = np.zeros(100); s_near[41:45] = 1.0
|
||||
s_far = np.zeros(100); s_far[90:95] = 1.0
|
||||
f1_near, _, _ = affiliation_f1(s_near, y)
|
||||
f1_far, _, _ = affiliation_f1(s_far, y)
|
||||
assert f1_near > f1_far
|
||||
|
||||
|
||||
class TestVUSPR:
|
||||
def test_perfect_high(self):
|
||||
y = labels([(20, 40)])
|
||||
s = np.zeros(100); s[20:41] = 1.0
|
||||
v = vus_pr(s, y)
|
||||
assert v == pytest.approx(1.0, abs=1e-3)
|
||||
|
||||
def test_random_scores_moderate(self):
|
||||
rng = np.random.default_rng(0)
|
||||
y = labels([(20, 40)])
|
||||
s = rng.random(100)
|
||||
v = vus_pr(s, y)
|
||||
# random ranking → VUS-PR around the anomaly prevalence (~0.21)
|
||||
assert 0.0 < v < 0.6
|
||||
|
||||
def test_all_report_not_inflated(self):
|
||||
# spec REQUIREMENT: "全报异常" trivial baseline must NOT have inflated VUS-PR
|
||||
y = labels([(20, 40)], T=100)
|
||||
s = np.ones(100) # report everything as anomalous
|
||||
v = vus_pr(s, y)
|
||||
# prevalence = 21/100; all-ones gives AUC-PR == prevalence everywhere
|
||||
assert v == pytest.approx(21 / 100, abs=1e-3)
|
||||
|
||||
def test_returns_scalar(self):
|
||||
y = labels([(5, 10)])
|
||||
s = np.random.default_rng(1).random(100)
|
||||
v = vus_pr(s, y)
|
||||
assert np.isscalar(v) or v.shape == ()
|
||||
@@ -0,0 +1,125 @@
|
||||
"""Tests for MultimodalTSModel wrapper (T2.4).
|
||||
|
||||
Verifies the end-to-end path: series → Encoder → Projector → splice → LLM.
|
||||
- forward returns a finite scalar loss
|
||||
- generate returns text
|
||||
- freeze_llm=True (stage ①): LLM frozen, encoder+projector trainable, and the
|
||||
loss carries grad into encoder/projector (backward works)
|
||||
- enable_lora(r=16) (stage ②): base LLM weights frozen, LoRA adapters trainable
|
||||
|
||||
GPU (RTX 3060) forward/backward + peak-memory checks live here too.
|
||||
"""
|
||||
import pytest
|
||||
import torch
|
||||
|
||||
from tsmm.model.ts_encoder import TSEncoder
|
||||
from tsmm.model.projector import Projector
|
||||
from tsmm.model.wrapper import MultimodalTSModel
|
||||
|
||||
LLM_PATH = "/home/zhangzp/models/Qwen2.5-0.5B-Instruct"
|
||||
pytestmark = pytest.mark.skipif(
|
||||
not torch.cuda.is_available(), reason="needs CUDA for LLM forward"
|
||||
)
|
||||
|
||||
|
||||
def make_model():
|
||||
return MultimodalTSModel(
|
||||
llm_path=LLM_PATH,
|
||||
encoder=TSEncoder(d=256, layers=2, heads=4, patch_len=8, stride=4),
|
||||
projector=Projector(in_dim=256, out_dim=896),
|
||||
dtype=torch.bfloat16,
|
||||
)
|
||||
|
||||
|
||||
def make_raw_batch(batch_size=2, T=512, C=5):
|
||||
"""Raw batch contract (what the collator T2.5 will produce)."""
|
||||
series = torch.randn(batch_size, T, C)
|
||||
return {
|
||||
"series": series,
|
||||
"attributes": ["CPU 内存"] * batch_size,
|
||||
"timestamps": ["t0..t1"] * batch_size,
|
||||
"events": [""] * batch_size,
|
||||
"question": ["是否存在异常？"] * batch_size,
|
||||
"answer": ["正常。"] * batch_size,
|
||||
}
|
||||
|
||||
|
||||
class TestForwardGenerate:
|
||||
def test_forward_returns_finite_loss(self):
|
||||
model = make_model().cuda()
|
||||
batch = make_raw_batch(batch_size=2)
|
||||
with torch.no_grad():
|
||||
out = model(batch)
|
||||
assert "loss" in out
|
||||
assert torch.isfinite(out["loss"])
|
||||
assert out["loss"].dim() == 0
|
||||
|
||||
def test_forward_backward_flows_into_encoder_projector(self):
|
||||
# stage ①: LLM frozen, so the only grad path is encoder+projector
|
||||
model = make_model().cuda()
|
||||
model.freeze_llm()
|
||||
model.train()
|
||||
batch = make_raw_batch(batch_size=1)
|
||||
out = model(batch)
|
||||
out["loss"].backward()
|
||||
# encoder/projector must have received gradients
|
||||
for p in model.encoder.parameters():
|
||||
assert p.grad is not None
|
||||
for p in model.projector.parameters():
|
||||
assert p.grad is not None
|
||||
|
||||
def test_generate_returns_text(self):
|
||||
model = make_model().cuda()
|
||||
batch = make_raw_batch(batch_size=1)
|
||||
text = model.generate(batch, max_new_tokens=8)
|
||||
assert isinstance(text, list)
|
||||
assert len(text) == 1
|
||||
assert isinstance(text[0], str)
|
||||
|
||||
|
||||
class TestStageModes:
|
||||
def test_freeze_llm_only_trains_encoder_projector(self):
|
||||
model = make_model()
|
||||
model.freeze_llm() # stage ①
|
||||
llm_requires_grad = [p.requires_grad for p in model.llm.parameters()]
|
||||
assert all(not r for r in llm_requires_grad)
|
||||
enc_proj_requires_grad = (
|
||||
[p.requires_grad for p in model.encoder.parameters()]
|
||||
+ [p.requires_grad for p in model.projector.parameters()]
|
||||
)
|
||||
assert all(enc_proj_requires_grad)
|
||||
|
||||
def test_enable_lora_freezes_base_injects_adapters(self):
|
||||
model = make_model()
|
||||
model.freeze_llm()
|
||||
model.enable_lora(r=16) # stage ②
|
||||
# base (non-LoRA) LLM weights must stay frozen
|
||||
base_frozen = all(
|
||||
not p.requires_grad
|
||||
for n, p in model.llm.named_parameters()
|
||||
if "lora_" not in n
|
||||
)
|
||||
assert base_frozen
|
||||
# LoRA adapter params must be trainable
|
||||
lora_trainable = [
|
||||
p for n, p in model.llm.named_parameters()
|
||||
if "lora_" in n and p.requires_grad
|
||||
]
|
||||
assert len(lora_trainable) > 0
|
||||
# there must be lora-named modules
|
||||
lora_names = [n for n, _ in model.named_modules() if "lora" in n.lower()]
|
||||
assert len(lora_names) > 0
|
||||
|
||||
|
||||
class TestGPUMemory:
|
||||
def test_forward_backward_no_oom_and_reasonable_mem(self):
|
||||
torch.cuda.reset_peak_memory_stats()
|
||||
model = make_model().cuda()
|
||||
model.freeze_llm() # stage ① mode for memory check
|
||||
model.train()
|
||||
batch = make_raw_batch(batch_size=2)
|
||||
out = model(batch)
|
||||
out["loss"].backward()
|
||||
peak_gb = torch.cuda.max_memory_allocated() / 1e9
|
||||
# stage ① should be well under the design M2 target (~5GB); use 8GB headroom
|
||||
assert peak_gb < 8.0, f"peak {peak_gb:.2f} GB"
|
||||
Reference in New Issue
Block a user