tao-route-visual-changenet-samples

TAO VCN 样例路由技能

在 VCN AOI SDA 流水线中，你是差距分析与增强模块之间的调度器。每个增强模块只能处理其已知的标签：

• k-NN Mining 只能为 源池 CSV 中已有的标签挖掘真实图像邻居。如果池中没有 SHIFT 行，就没必要查找 SHIFT 邻居。
• AnomalyGen（Cosmos SDG）只能为其推理流水线支持的类别生成合成异常：PASS、EXCESS_SOLDER、MISSING、BRIDGE。标签不在此集合中的弱样本无法路由到 AnomalyGen。

此技能在每次 SDA 迭代中于差距分析完成后立即运行一次。它会将差距分析 parquet 拆分为每个模块各自的一个过滤后 parquet，以便每个模块仅处理符合条件的子集，并生成一份易读摘要，列出各标签的路由判定。

这项工作本身很简单：读取一个 parquet，执行两次 .isin(...) 过滤，写入两个 parquet，再写入一份摘要。此技能旨在让这些判定可供审计——摘要中必须列出每个标签，并给出每个模块的是／否判定，以便下游审核者发现某个标签因没有模块接收而被悄然丢弃的情况。

---

输入

1. gaps_parquet — 差距分析输出（通常为 tao-analyze-gaps-visual-changenet 生成的 <exp_dir>/rca_results/<timestamp>/gaps.parquet）。必需列：filepath、label。其他列（siamese_score、weakness）会原样保留。
2. source_pool_csv — VCN 格式的挖掘源池 CSV，包含 label 列。允许为空字符串或不存在的路径；在这种情况下，挖掘子集将为空。
3. 输出目录 — 写入两个路由后的 parquet、摘要和报告的位置。默认值：差距分析结果目录下带时间戳的文件夹：<rca_result_dir>/routing_results/<timestamp>/。
4. anomalygen_supported_labels （可选） — 覆盖默认的 AnomalyGen 合格标签集合。默认值：{"PASS", "EXCESS_SOLDER", "MISSING", "BRIDGE"}。**警告：**此处必须与 mdo-kratos-workflows/pipelines/sda/routing.py 中的 ANOMALYGEN_SUPPORTED_LABELS 以及 AnomalyGen 集成实际支持的生成类别保持同步。向 AnomalyGen 添加新的缺陷类别时，也必须在此处添加。

---

方法

整个技能只需对大写形式的标签列执行两次 .isin(...) 掩码过滤。

步骤 1 — 加载并转换为大写

df = pd.read_parquet(gaps_parquet)
labels_upper = df["label"].astype(str).str.upper()

两个模块的检查均不区分大小写。输出 parquet 中的原始 label 列保持不变——只有用于比较的键会转换为大写。

步骤 2 — 挖掘子集

if source_pool_csv and os.path.isfile(source_pool_csv):
    pool_df = pd.read_csv(source_pool_csv)
    pool_labels = {str(l).upper() for l in pool_df["label"].unique()}
    mn_mask = labels_upper.isin(pool_labels)
    mn_df = df[mn_mask]
else:
    pool_missing = True
    pool_labels = set()
    mn_df = df.iloc[0:0]   # empty, but with the same schema
mn_df.to_parquet(mining_gaps_parquet, index=False)

如果池 CSV 缺失或为空，挖掘子集将是一个与输入列完全相同的空 DataFrame，以免下游读取器因模式不匹配而崩溃。请在摘要中标记此情况。

步骤 3 — AnomalyGen 子集

ANOMALYGEN_SUPPORTED = {"PASS", "EXCESS_SOLDER", "MISSING", "BRIDGE"}
ag_mask = labels_upper.isin(ANOMALYGEN_SUPPORTED)
ag_df = df[ag_mask]
ag_df.to_parquet(anomalygen_gaps_parquet, index=False)

标签属于 AnomalyGen 支持集合的行会原样写入 anomalygen_gaps.parquet。其模式与输入 parquet 完全一致——下游 AnomalyGen（Cosmos SDG）无需做其他改动。

步骤 4 — 按标签列出路由明细

对于输入 gaps parquet 中每个不同的标签（转换为大写），记录：

• count — 此标签对应的行数
• mining — 如果标签属于 pool_labels，则为 yes；否则为 no
• anomalygen — 如果标签属于 ANOMALYGEN_SUPPORTED，则为 yes；否则为 no

一个标签可以同时路由到两个模块（例如，PASS 行会路由到 AnomalyGen；如果源池也包含 PASS 行，它们也会路由到 Mining）。标签也可能路由到两个模块都不接收的情况——请标记这些标签，因为它们会被悄然丢弃，并且可能意味着配置不匹配。

将明细写入 routing_summary.txt。格式与参考组件完全一致：

Weak-sample routing summary
Total weak samples: <N>
Mining subset:      <N_mn> -> <mining_gaps_parquet>
AnomalyGen subset:  <N_ag> -> <anomalygen_gaps_parquet>

[If pool missing:]
No source pool CSV at '<path>'; mining subset is empty.

Per-label breakdown (count, mining, anomalygen):
  PASS: 50 (mining=yes, anomalygen=yes)
  MISSING: 32 (mining=no, anomalygen=yes)
  SHIFT: 14 (mining=yes, anomalygen=no)
  EXCESS_SOLDER: 9 (mining=yes, anomalygen=yes)
  ...

第 5 步 — 完整性检查

写入两个子集后，请验证：

• 子集大小之和不必等于 len(df)——允许重叠（某个标签可以同时路由到两个模块）。关键是，每个输入行都至少出现在一个子集中，或者出现在“none”列表中并附有明确原因。
• 如果 len(mn_df) == 0 且 len(ag_df) == 0，说明存在问题——请在报告中醒目标出。
• 如果某个完整的标签组未路由到任何模块，Recommended Actions 部分必须指出这一点，以便用户可以为源池加入该标签，或扩展 AnomalyGen 的支持集。

---

Python 参考流程

这是从 mdo-kratos-workflows/pipelines/sda/routing.py 中提取的完整计算流程。请通过 Bash 将其作为单个 Python 脚本运行；它会生成除报告以外的所有产物。

import os
import pandas as pd

ANOMALYGEN_SUPPORTED = {"PASS", "EXCESS_SOLDER", "MISSING", "BRIDGE"}

df = pd.read_parquet(gaps_parquet)
labels_upper = df["label"].astype(str).str.upper()

# Mining subset
pool_missing = False
if source_pool_csv and os.path.isfile(source_pool_csv):
    pool_df = pd.read_csv(source_pool_csv)
    pool_labels = {str(l).upper() for l in pool_df["label"].unique()}
    mn_mask = labels_upper.isin(pool_labels)
    mn_df = df[mn_mask]
else:
    pool_missing = True
    pool_labels = set()
    mn_df = df.iloc[0:0]
os.makedirs(os.path.dirname(mining_gaps_parquet) or ".", exist_ok=True)
mn_df.to_parquet(mining_gaps_parquet, index=False)

# AnomalyGen subset
ag_mask = labels_upper.isin(ANOMALYGEN_SUPPORTED)
ag_df = df[ag_mask]
os.makedirs(os.path.dirname(anomalygen_gaps_parquet) or ".", exist_ok=True)
ag_df.to_parquet(anomalygen_gaps_parquet, index=False)

# Per-label breakdown
summary_lines = [
    "Weak-sample routing summary",
    f"Total weak samples: {len(df)}",
    f"Mining subset:      {len(mn_df)} -> {mining_gaps_parquet}",
    f"AnomalyGen subset:  {len(ag_df)} -> {anomalygen_gaps_parquet}",
    "",
]
if pool_missing:
    summary_lines.append(f"No source pool CSV at {source_pool_csv!r}; mining subset is empty.")
    summary_lines.append("")
summary_lines.append("Per-label breakdown (count, mining, anomalygen):")
label_counts = labels_upper.value_counts()
for label, count in label_counts.items():
    in_mn = (not pool_missing) and label in pool_labels
    in_ag = label in ANOMALYGEN_SUPPORTED
    summary_lines.append(
        f"  {label}: {count} "
        f"(mining={'yes' if in_mn else 'no'}, "
        f"anomalygen={'yes' if in_ag else 'no'})"
    )
summary_text = "\n".join(summary_lines) + "\n"

os.makedirs(logs_dir, exist_ok=True)
with open(os.path.join(logs_dir, "routing_summary.txt"), "w", encoding="utf-8") as f:
    f.write(summary_text)
print(summary_text.strip())

---

输出

将所有内容写入带时间戳的文件夹。在写入 Routing_Report.md 时，打包钩子会自动复制 routing_config/ 和 claude_session.jsonl。

<output_dir>/routing_results/YYYY-MM-DD_HHMMSS/
├── Routing_Report.md           # Full routing report
├── mining_gaps.parquet         # Subset routed to k-NN Mining
├── anomalygen_gaps.parquet     # Subset routed to AnomalyGen (Cosmos SDG)
├── routing_summary.txt         # Plain-text per-label breakdown
├── routing_config/             # Auto-copied by hook
└── claude_session.jsonl        # Auto-copied by hook

运行开始时，通过在 Bash 中运行 date +%Y-%m-%d_%H%M%S 获取实际时间戳。如果用户指定了自定义输出路径，请直接使用该路径，但保持内部目录结构。

---

报告结构

报告应简短（400–800 字）。路由是确定性决策；其价值在于让决策可审计，而非进行叙述。

# VCN Routing Report: <Iteration / Experiment Name>

## 1. Verdict
- Total weak samples in: <N>
- Mining subset:     <N_mn> rows  →  `mining_gaps.parquet`
- AnomalyGen subset: <N_ag> rows  →  `anomalygen_gaps.parquet`
- Source pool present? <yes/no — and the path>
- One-line headline: "<X> labels routed, <Y> labels dropped (no module accepted)"

## 2. Inputs
| Input | Path | Notes |
|-------|------|-------|
| gaps_parquet     | … | rows=<N>, columns=<col list> |
| source_pool_csv  | … | rows=<M> or "not provided" / "missing" |

## 3. Per-Label Routing Decisions
| Label | Count in gaps | In source pool? | Mining? | AnomalyGen? | Routed To |
|-------|----------------|------------------|----------|--------------|-----------|

(One row per distinct label in `gaps_parquet`, uppercased. `Routed To` is one of:
`mining only`, `anomalygen only`, `mining+anomalygen`, `neither (DROPPED)`.
Use `neither (DROPPED)` whenever no module accepted the label. Sort by count descending.)

## 4. Module-Level Summaries
### 4.1 k-NN Mining
- Pool labels (from source_pool_csv): <list, or "pool missing">
- Labels accepted from input: <list>
- Total rows routed: <N_mn>
- Per-label row counts: <breakdown>

### 4.2 AnomalyGen (Cosmos SDG)
- Eligible labels (configured): PASS, EXCESS_SOLDER, MISSING, BRIDGE
- Labels accepted from input: <list>
- Total rows routed: <N_ag>
- Per-label row counts: <breakdown>

## 5. Dropped Labels (routed to NEITHER module)
| Label | Count | Why dropped | Suggested fix |
|-------|-------|-------------|----------------|

(Empty table is OK and means no labels were dropped. If non-empty, every row needs a
"why" — typically one of: "not in source pool AND not in AnomalyGen supported set",
"source pool missing entirely AND label not in AnomalyGen set", "label name doesn't
match any module's expected canonicalization".)

## 6. Recommended Actions
1. **If any labels are dropped**: seed the source pool with that label, OR extend
   `ANOMALYGEN_SUPPORTED_LABELS` (and the AnomalyGen generator coverage).
2. **If source pool is missing**: provide `source_pool_csv` to enable the Mining branch.
   Without it, half of the augmentation pipeline is dark.
3. **If AnomalyGen subset is empty**: gap analysis only surfaced labels AnomalyGen cannot
   generate; rely on Mining for this iteration, or extend the AnomalyGen integration.
4. **If both subsets are empty**: stop the SDA iteration. Nothing downstream can run.

---

执行顺序

1. 运行 date +%Y-%m-%d_%H%M%S 获取时间戳；创建 <output_dir>/routing_results/<timestamp>/。
2. 运行 Python 流程（步骤 1–4），生成 mining_gaps.parquet、anomalygen_gaps.parquet 和 routing_summary.txt。将汇总统计打印到 stdout，以便脚本检查钩子验证脚本已运行。
3. 读取两个 parquet 文件，并按标签计算路由判定，构建逐标签决策表。
4. 最后写入 Routing_Report.md——写入该文件会触发打包钩子，将会话日志和技能配置一并复制。