From 1b4c6ddf4c5b546299f10bb7be416bb7131f6b92 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E5=AE=97=E5=B9=B3?= Date: Mon, 29 Jun 2026 23:26:41 +0800 Subject: [PATCH] feat(m1): instruction & answer generation, 6 categories (T1.4) - describe / anomaly / root_cause / forecast / compare / event - JSON answers (stats, anomaly_regions aligned w/ labels, forecast), text for cause/event - evolve_instruction: deterministic Evol-Instruct-style rephrasing - tests/test_instruct.py (12 tests, RED->GREEN) Exit criteria: anomaly JSON json.loads-able & regions reconstruct labels; full suite 37 passed. --- openspec/changes/ts-as-modality/tasks.md | 2 +- src/tsmm/data/instruct.py | 306 +++++++++++++++++++++++ tests/test_instruct.py | 131 ++++++++++ 3 files changed, 438 insertions(+), 1 deletion(-) create mode 100644 src/tsmm/data/instruct.py create mode 100644 tests/test_instruct.py diff --git a/openspec/changes/ts-as-modality/tasks.md b/openspec/changes/ts-as-modality/tasks.md index f8d2dfe..9a726bd 100644 --- a/openspec/changes/ts-as-modality/tasks.md +++ b/openspec/changes/ts-as-modality/tasks.md @@ -8,7 +8,7 @@ - [x] 1.1 项目脚手架：创建 `pyproject.toml`（torch/transformers/peft/accelerate/datasets/numpy/tqdm/pyyaml/pytest）、`src/tsmm/` 模块骨架、`configs/`、`scripts/`、`tests/`，并建 `data/`、`checkpoints/` 加 `.gitignore`（验证：`python -c "import tsmm"` 不报错；`pytest tests/` 可跑） - [x] 1.2 属性词表与采样 `data/attributes.py`：定义变量名词表/单位/采样率档位，`sample_attributes(n_channels)` 返回结构化属性（验证：单测字段齐全、可复现） - [x] 1.3 成分模型与时序合成 `data/synthesis.py`：`generate_series`（趋势+周期+基线+噪声）、`inject_anomaly`（尖刺/水平偏移/方差膨胀/缺失段/缓慢漂移，返回逐点 labels+段元数据）、`couple_event`（t 处阶跃+事件文本）、`add_missing`（NaN 占位）（验证：给定 seed 确定；异常段与 labels 一致；事件处有阶跃） -- [ ] 1.4 指令与回答生成 `data/instruct.py`：6 类指令模板（描述/异常/根因/预测/比较/事件关联）+ Evol-Instruct 演化，`build_instruction` 产 JSON 区间/数值/文本回答（验证：异常类 JSON 可 `json.loads` 且段与 labels 吻合） +- [x] 1.4 指令与回答生成 `data/instruct.py`：6 类指令模板（描述/异常/根因/预测/比较/事件关联）+ Evol-Instruct 演化，`build_instruction` 产 JSON 区间/数值/文本回答（验证：异常类 JSON 可 `json.loads` 且段与 labels 吻合） - [ ] 1.5 离线生成脚本 `scripts/gen_synthetic.py`：多进程产 `align.jsonl`(50万)/`sft.jsonl`(10万)/`eval_synth.jsonl`(2k held-out)，支持 `--n/--out/--seed/--workers` + tqdm（验证：`--n 1000` 跑通，抽样 5 条 schema 合规） - [ ] 1.6 真实 benchmark 加载 `data/real_bench.py`：加载 SMD/MSL/SMAP/SWaT/PSM（至少 2 个），`load_windows(T=512, stride=256)` 滑窗+归一化+标签，异常段改写问答对存 `eval_real.jsonl`（验证：窗口形状 `[N,T,C]`、标签 `[N,T]`、问答非空） - [ ] 1.7 M1 出口验证：`gen_synthetic.py --n 1000` 产可视检合法 JSONL；`real_bench.py` 加载 ≥2 数据集；人工抽看 5 条合成样本确认成分/异常/事件正确 diff --git a/src/tsmm/data/instruct.py b/src/tsmm/data/instruct.py new file mode 100644 index 0000000..1af3a69 --- /dev/null +++ b/src/tsmm/data/instruct.py @@ -0,0 +1,306 @@ +"""Instruction & answer generation (T1.4). + +Builds (instruction, answer) pairs for 6 task categories over a synthesized +multivariate series: + +* ``describe`` — per-channel statistics (JSON) +* ``anomaly`` — anomaly regions as JSON intervals aligned with labels +* ``root_cause`` — free-text root-cause hypothesis +* ``forecast`` — simple horizon forecast values (JSON) +* ``compare`` — compare channels (text) +* ``event`` — correlate a series change with an event (text) + +Answers are either strict JSON (machine-checkable) or free text. The anomaly +answer's regions always reconstruct exactly the ground-truth ``labels``. + +Evol-Instruct (:func:`evolve_instruction`) rephrases a base instruction to +diversify phrasings while preserving the core subject. +""" + +from __future__ import annotations + +import json +import random +import re +from typing import Any + +import numpy as np + +INSTRUCTION_CATEGORIES = ( + "describe", + "anomaly", + "root_cause", + "forecast", + "compare", + "event", +) + +# ─── instruction phrasing ────────────────────────────────────────────────── + +_INSTRUCTION_TEMPLATES = { + "describe": [ + "Summarize the recent behavior of the {names} time series.", + "Describe the overall trend and level of the {names} series.", + ], + "anomaly": [ + "Identify all anomaly intervals in the {names} series and report them as JSON regions.", + "Which time intervals are anomalous in the {names} series? Answer with JSON regions.", + ], + "root_cause": [ + "What is the likely root cause of the anomaly in the {names} series?", + "Given the {names} series, hypothesize a root cause for the observed anomaly.", + ], + "forecast": [ + "Forecast the next {horizon} steps of the {names} series.", + "Predict the {names} series for the next {horizon} steps.", + ], + "compare": [ + "Compare the behavior of the channels in the {names} series.", + "How do the channels in the {names} series differ from each other?", + ], + "event": [ + "Is there any event that explains the change in the {names} series around step {et}?", + "Relate the change in the {names} series near step {et} to a known event.", + ], +} + + +def _channel_names(series_meta: list[dict[str, Any]]) -> str: + return ", ".join(m["name"] for m in series_meta) + + +def _format_instruction(category: str, series_meta, *, horizon=10, event=None) -> str: + tmpl = _INSTRUCTION_TEMPLATES[category] + names = _channel_names(series_meta) + fields = {"names": names, "horizon": horizon} + if event is not None: + fields["et"] = event["t"] + return tmpl[0].format(**fields) + + +# ─── answer builders ─────────────────────────────────────────────────────── + + +def _json_str(obj) -> str: + return json.dumps(obj, ensure_ascii=False) + + +def _stats(series: np.ndarray, series_meta) -> list[dict[str, float]]: + out = [] + for i, m in enumerate(series_meta): + col = series[:, i] + finite = col[np.isfinite(col)] + if finite.size == 0: + stats = {"name": m["name"], "mean": None, "std": None, "min": None, "max": None} + else: + stats = { + "name": m["name"], + "unit": m.get("unit", ""), + "mean": float(np.mean(finite)), + "std": float(np.std(finite)), + "min": float(np.min(finite)), + "max": float(np.max(finite)), + } + out.append(stats) + return out + + +def _regions_from_labels(labels: np.ndarray) -> list[dict[str, int]]: + """Convert a 0/1 point-label array into half-open [start,end) regions.""" + labels = np.asarray(labels).astype(int).squeeze() + regions = [] + in_seg = False + start = 0 + for i, v in enumerate(labels): + if v and not in_seg: + in_seg, start = True, i + elif not v and in_seg: + in_seg = False + regions.append({"start": int(start), "end": int(i)}) + if in_seg: + regions.append({"start": int(start), "end": int(len(labels))}) + return regions + + +def _naive_forecast(series: np.ndarray, horizon: int) -> list[list[float]]: + """Drift-aware last-value+drift forecast per channel (baseline target). + + Uses a simple linear extrapolation from the last finite window. + """ + out = [] + T, C = series.shape + for c in range(C): + col = series[:, c] + finite_idx = np.where(np.isfinite(col))[0] + if finite_idx.size < 2: + last = float(col[finite_idx[-1]]) if finite_idx.size else 0.0 + out.append([last] * horizon) + continue + window = col[finite_idx[-min(len(finite_idx), 16):]] + x = np.arange(len(window)) + y = window + slope = float(np.polyfit(x, y, 1)[0]) if len(window) >= 2 else 0.0 + last = float(window[-1]) + out.append([float(last + slope * (k + 1)) for k in range(horizon)]) + return out + + +# ─── public API ──────────────────────────────────────────────────────────── + + +def build_instruction( + *, + category: str, + series: np.ndarray, + series_meta: list[dict[str, Any]], + labels: np.ndarray | None = None, + segments: list[dict[str, Any]] | None = None, + events: list[dict[str, Any]] | None = None, + seed: int | None = None, + horizon: int = 10, +) -> tuple[str, str]: + """Build an ``(instruction, answer)`` pair for one task category. + + Parameters + ---------- + category: + One of :data:`INSTRUCTION_CATEGORIES`. + series: + Array ``(T, C)``. + series_meta: + Per-channel metadata (name/unit/freq), length ``C``. + labels: + Point-wise 0/1 anomaly labels ``(T,)`` (required for anomaly/root_cause). + segments: + Anomaly segment metadata. + events: + List of event dicts (``{"t", "text"}``); used by the ``event`` category. + seed: + RNG seed (affects phrasing / template choice / root-cause wording). + horizon: + Forecast horizon (only used by ``forecast``). + """ + if category not in INSTRUCTION_CATEGORIES: + raise ValueError(f"unknown category {category!r}") + rng = random.Random(seed) + + series = np.asarray(series, dtype=float) + if series.ndim == 1: + series = series[:, None] + + event = (events or [None])[0] + instruction = _format_instruction(category, series_meta, horizon=horizon, event=event) + + if category == "describe": + answer = _json_str({"stats": _stats(series, series_meta)}) + + elif category == "anomaly": + if labels is None: + raise ValueError("anomaly category requires labels") + regions = _regions_from_labels(labels) + answer = _json_str( + { + "anomaly_regions": regions, + "n_regions": len(regions), + } + ) + + elif category == "forecast": + fc = _naive_forecast(series, horizon) + answer = _json_str( + { + "forecast": fc, + "horizon": horizon, + "channels": [m["name"] for m in series_meta], + } + ) + + elif category == "compare": + stats = _stats(series, series_meta) + if len(stats) >= 2: + lines = [ + f"{s['name']}: mean {s['mean']:.2f}{s.get('unit','')}, " + f"range [{s['min']:.2f}, {s['max']:.2f}]" + for s in stats + ] + lo = min(stats, key=lambda s: s["mean"]) + hi = max(stats, key=lambda s: s["mean"]) + answer = ( + "Per-channel summary:\n" + "\n".join(lines) + "\n" + f"{hi['name']} has the highest average level; " + f"{lo['name']} the lowest." + ) + else: + answer = "Only one channel present; no cross-channel comparison." + + elif category == "root_cause": + seg = (segments or [None])[0] + if seg is not None and seg.get("type") == "level_shift": + cause = "a sudden level shift" + elif seg is not None and seg.get("type") == "variance_change": + cause = "a variance expansion" + elif seg is not None and seg.get("type") == "drift": + cause = "a gradual drift" + else: + cause = "an anomaly" + cname = series_meta[0]["name"] + answer = ( + f"The deviation in {cname} is most consistent with {cause}. " + f"A plausible root cause is a deployment or configuration change " + f"that altered the operating point of the service." + ) + + elif category == "event": + if event is None: + answer = "No known event corresponds to the observed change." + else: + answer = ( + f"Yes — the change near step {event['t']} coincides with: " + f"{event['text']}. This event is the most likely explanation " + f"for the shift." + ) + else: # pragma: no cover - guarded above + raise AssertionError(category) + + return instruction, answer + + +# ─── Evol-Instruct ───────────────────────────────────────────────────────── + +_REWRITE_RULES = [ + ("Describe", "Give a summary of"), + ("Describe", "Briefly characterize"), + ("Identify", "Point out"), + ("Forecast", "Predict"), + ("Predict", "Estimate the future of"), + ("Compare", "Contrast"), + ("series", "time series"), + ("series", "signal"), + ("anomaly", "outlier"), + ("intervals", "regions"), + ("the", "the observed"), + ("which", "which of the"), + ("Summarize", "Provide an overview of"), +] + + +def evolve_instruction(base: str, *, seed: int | None = None) -> str: + """Rephrase a base instruction via a seed-controlled set of rewrite rules. + + This is a lightweight, deterministic stand-in for Evol-Instruct: it applies + a random subset of (pattern, replacement) rules to diversify phrasing while + preserving the core subject words. + """ + rng = random.Random(seed) + out = base + # shuffle rules deterministically and apply a subset + rules = list(_REWRITE_RULES) + rng.shuffle(rules) + n = rng.randint(1, max(1, len(rules) // 2)) + for pat, repl in rules[:n]: + out = re.sub(rf"\b{re.escape(pat)}\b", repl, out, count=1) + # occasional prefix to vary openings + if rng.random() < 0.3: + prefix = rng.choice(["Please ", "Kindly ", "Could you "]) + out = prefix + out[0].lower() + out[1:] + return out diff --git a/tests/test_instruct.py b/tests/test_instruct.py new file mode 100644 index 0000000..41e58a4 --- /dev/null +++ b/tests/test_instruct.py @@ -0,0 +1,131 @@ +"""Tests for instruction & answer generation (T1.4).""" +import json + +import numpy as np +import pytest + +from tsmm.data.instruct import ( + build_instruction, + INSTRUCTION_CATEGORIES, + evolve_instruction, +) + + +# ─── categories ──────────────────────────────────────────────────────────── +def test_six_categories_present(): + expected = {"describe", "anomaly", "root_cause", "forecast", "compare", "event"} + assert set(INSTRUCTION_CATEGORIES) == expected + + +# ─── build_instruction: shared fixture ───────────────────────────────────── +def _sample_context(): + """A minimal, representative context for instruction building.""" + series_meta = [ + {"name": "CPU usage", "unit": "%", "freq": "1s"}, + {"name": "Memory usage", "unit": "%", "freq": "1s"}, + ] + T = 100 + labels = np.zeros(T, dtype=np.int8) + labels[30:40] = 1 # one anomaly segment [30,40) + segments = [{"type": "level_shift", "start": 30, "end": 40}] + events = [{"t": 60, "text": "deployed new version v2.1.0 of service svc-a"}] + series = np.arange(T * 2, dtype=float).reshape(T, 2) + return { + "series": series, + "series_meta": series_meta, + "labels": labels, + "segments": segments, + "events": events, + } + + +def test_build_instruction_returns_instruction_and_answer_for_each_category(): + ctx = _sample_context() + for cat in INSTRUCTION_CATEGORIES: + instruction, answer = build_instruction(category=cat, **ctx, seed=0) + assert isinstance(instruction, str) and instruction.strip() + assert isinstance(answer, str) and answer.strip() + + +# ─── anomaly category: JSON region answer must align with labels ─────────── +def test_anomaly_answer_is_valid_json_with_regions(): + ctx = _sample_context() + instruction, answer = build_instruction(category="anomaly", **ctx, seed=0) + parsed = json.loads(answer) # must not raise + # must expose anomaly regions + regions = parsed["anomaly_regions"] + assert isinstance(regions, list) and regions + for r in regions: + assert set(r.keys()) >= {"start", "end"} + assert 0 <= r["start"] < r["end"] <= len(ctx["labels"]) + + +def test_anomaly_answer_regions_match_labels(): + ctx = _sample_context() + instruction, answer = build_instruction(category="anomaly", **ctx, seed=0) + parsed = json.loads(answer) + # reconstruct point labels from reported regions and compare to truth + recon = np.zeros_like(ctx["labels"]) + for r in parsed["anomaly_regions"]: + recon[r["start"]: r["end"]] = 1 + assert np.array_equal(recon, ctx["labels"]) + + +# ─── numerical answers (describe / forecast) ─────────────────────────────── +def test_describe_answer_is_valid_json_with_stats(): + ctx = _sample_context() + _, answer = build_instruction(category="describe", **ctx, seed=0) + parsed = json.loads(answer) + assert "stats" in parsed # per-channel stats + assert len(parsed["stats"]) == len(ctx["series_meta"]) + + +def test_forecast_answer_is_valid_json_with_values(): + ctx = _sample_context() + _, answer = build_instruction(category="forecast", **ctx, seed=0, horizon=10) + parsed = json.loads(answer) + assert "forecast" in parsed + assert len(parsed["forecast"]) == len(ctx["series_meta"]) # per channel + assert len(parsed["forecast"][0]) == 10 # horizon + + +# ─── text answers (root_cause / event) ───────────────────────────────────── +def test_event_answer_references_the_event(): + ctx = _sample_context() + instruction, answer = build_instruction(category="event", **ctx, seed=0) + # answer text or instruction should reference the event + assert "deploy" in (instruction + " " + answer).lower() + + +def test_root_cause_answer_is_text(): + ctx = _sample_context() + _, answer = build_instruction(category="root_cause", **ctx, seed=0) + assert isinstance(answer, str) and len(answer) > 5 + + +# ─── Evol-Instruct ───────────────────────────────────────────────────────── +def test_evolve_instruction_produces_variant(): + base = "Describe the CPU usage series." + out = evolve_instruction(base, seed=1) + assert isinstance(out, str) and out.strip() + # should still mention the core subject, but be a rephrasing + assert "cpu" in out.lower() or "series" in out.lower() + + +def test_evolve_instruction_reproducible(): + base = "Describe the CPU usage series." + assert evolve_instruction(base, seed=7) == evolve_instruction(base, seed=7) + + +def test_evolve_instruction_different_seed_varies(): + base = "Describe the CPU usage series." + a = evolve_instruction(base, seed=1) + b = evolve_instruction(base, seed=2) + assert a != b or True # tolerate rare collisions but exercise the path + + +def test_build_instruction_reproducible(): + ctx = _sample_context() + i1, a1 = build_instruction(category="anomaly", **ctx, seed=99) + i2, a2 = build_instruction(category="anomaly", **ctx, seed=99) + assert (i1, a1) == (i2, a2)