Merge remote-tracking branch 'origin/main' into wsh_dev

# Conflicts: # backend/api/server.py
2026-05-25 11:46:44 +08:00
parent f068311cac c40d59901b
commit d8741a3489
77 changed files with 11155 additions and 2259 deletions
@@ -0,0 +1,136 @@
+---
+name: intervention-data
+description: 生成和校验单句精确干预、正则干预候选 TSV，复用标签大师的 target 语法校验，并从历史干预表推断 code 到下发 domain 的固定映射。
+when_to_use: 当用户希望新增、检查或整理干预规则，例如给某个 query 或多轮正则生成干预条目、确认设备和标签是否合法、推断下发 domain 时使用。
+aliases: intervention, rule-intervention, 干预数据
+allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_user_question, python_exec
+---
+
+使用这个 skill 处理“query/多轮表达 -> 干预 TSV 候选 -> 合法性校验”的任务。
+
+干预数据目前有两类：
+
+- **单句精确干预**：一行四列，格式为 `设备\tquery\tcode\t下发domain`。
+- **正则干预**：一行三列，格式为 `设备\t正则\t标签`。
+
+其中单句精确干预的 `code` 和正则干预的 `标签` 使用同一套 target 语法，和 `label-master` 的输出一致，但不包含复杂度判断。校验时必须复用 `skills/label-master/scripts/validate_label_output.py` 的逻辑。
+
+## 关键约定
+
+- `unify` 表示全部设备。
+- 设备必须优先从历史干预文件中归纳，不要随手发明设备名。
+- 单句精确干预如果用户没有提供 `下发domain`，必须从历史精确干预表中按 `code -> 下发domain` 的多数映射推断。
+- 正则干预支持多轮。多轮正则使用 `beforeQuery#上一轮#query#当前轮` 这种片段表达，最终通常包在 `^...$` 中。
+- 默认只生成候选 TSV 和校验结果，不直接追加到历史干预文件。只有用户明确要求“追加到某个文件末尾”或“修改某个文件”时，才使用通用文件编辑能力。
+- 所有产物优先写到当前会话的 `output/` 或 `scratchpad/` 下，不要写到项目根目录的 `output/`。
+
+## 推荐流程
+
+1. 先确认用户要哪种干预方式：
+   - 单句精确干预：query 必须完全命中。
+   - 正则干预：适合一类表达、多轮上下文或需要泛化的高频 case。
+2. 确认设备。用户没说设备时，必须问设备；可以提示 `unify` 表示全部设备。
+3. 确认目标标签/code。它必须是完整 target，例如 `Agent(tag="地图导航")`、`QA()`、`Chat()` 或合法 function 调用。
+4. 如果是单句精确干预：
+   - 确认 `query`。
+   - 如果没有 `下发domain`，调用脚本根据历史数据推断。
+5. 如果是正则干预：
+   - 确认当前轮 query。
+   - 如果有多轮，确认前序 query 顺序。
+   - 用户给了原始正则就保留；没给时由脚本生成 `^query#...$` 或 `^beforeQuery#...#query#...$`。
+6. 调用 `scripts/generate_intervention_candidate.py` 生成候选 TSV 和校验报告。
+7. 展示时只展示必要信息：干预类型、设备、候选 TSV、校验是否通过、需要用户确认的问题。
+
+## 脚本能力
+
+```text
+skills/intervention-data/
+  SKILL.md
+  knowledge/
+    type_match_agent.json
+  scripts/
+    intervention_common.py
+    analyze_intervention_sources.py
+    audit_intervention_sources.py
+    generate_intervention_candidate.py
+    validate_intervention_records.py
+```
+
+### analyze_intervention_sources.py
+
+读取历史干预文件，输出设备枚举、`code -> 下发domain` 多数映射、常见标签，以及 `knowledge/type_match_agent.json` 中的下发表。
+
+示例：
+
+```bash
+python skills/intervention-data/scripts/analyze_intervention_sources.py
+```
+
+### generate_intervention_candidate.py
+
+从 JSON 输入生成候选 TSV，并自动校验。输入可以来自文件或 stdin。
+
+### audit_intervention_sources.py
+
+审计现有两个历史干预文件，只输出摘要，不修改文件。适合验证当前 skill 的校验规则和存量数据是否贴合。
+
+示例：
+
+```bash
+python skills/intervention-data/scripts/audit_intervention_sources.py
+```
+
+单句精确干预示例：
+
+```json
+{
+  "mode": "exact",
+  "device": "unify",
+  "query": "播放周杰伦的歌",
+  "target": "Agent(tag=\"音乐\")"
+}
+```
+
+正则多轮干预示例：
+
+```json
+{
+  "mode": "regex",
+  "device": "miCar",
+  "before_queries": ["第一个"],
+  "query": "红绿灯少的路线",
+  "target": "Agent(tag=\"地图导航\")"
+}
+```
+
+### validate_intervention_records.py
+
+校验候选 TSV。支持 `--mode exact` 或 `--mode regex`。
+
+规则包括：
+
+- 列数正确。
+- 设备合法。
+- `code` / `标签` 优先通过 label-master 校验；如果 label-master 未收录但历史干预表中已经稳定出现，允许作为候选通过，并给出 warning 让用户确认知识库是否需要补齐。
+- 单句精确干预的 `下发domain` 与历史多数映射一致，默认不一致为 warning，`--strict-domain` 时为 error。
+- 单句精确干预还会用 `type_match_agent.json` 做配置校对；配置值可以是更宽泛前缀，例如 `音乐 -> contentCopilot` 可以兼容历史里的 `contentCopilot|music`。
+- 正则可编译，且包含 `query#` 片段。
+- 与历史文件重复或冲突时给出 warning/error。
+
+## 展示格式
+
+生成候选后，优先用这个格式回复：
+
+```text
+我生成了一条候选干预，先不写入历史文件。
+- 类型：单句精确干预 / 正则干预
+- 设备：xxx
+- 标签：xxx
+- 校验：通过 / 有问题
+
+候选 TSV：`设备	query	code	下发domain`
+
+需要确认：是否写入某个文件，或是否调整设备/标签/正则。
+```
+
+如果校验失败，先解释失败原因，不要要求用户直接落盘。
@@ -0,0 +1,211 @@
+{
+  "音乐": "contentCopilot",
+  "音乐#tvV2": "contentCopilot|video",
+  "音乐#miCar": "contentCopilot|station",
+  "音乐#screenSoundbox": "contentCopilot|music",
+  "音乐#glass": "contentCopilot|ancientPoem",
+  "电台": "contentCopilot",
+  "电台#tvV2": "contentCopilot|video",
+  "视频": "contentCopilot",
+  "古诗": "contentCopilot",
+  "歌单": "contentCopilot|controlCopilot|soundboxControl|smartApp",
+  "应用播放": "contentCopilot|controlCopilot|soundboxControl|smartApp",
+  "新闻": "contentCopilot",
+  "笑话": "contentCopilot",
+  "内容控制": "contentCopilot|soundboxControl|controlCopilot",
+  "内容控制|系统控制": "contentCopilot|soundboxControl|controlCopilot",
+  "内容控制|应用控制": "contentCopilot|smartApp|controlCopilot",
+  "播放状态查询": "contentCopilot|soundboxControl|controlCopilot",
+  "内容问答": "contentCopilot|qabot|dialogCopilot",
+  "内容问答#tvV2": "contentCopilot|qabot|dialogCopilot|video",
+  "内容问答#soundbox": "contentCopilot|qabot|dialogCopilot|michat",
+  "赛事播放": "contentCopilot|qabot|dialogCopilot|sports",
+  "赛事预约": "contentCopilot|qabot|dialogCopilot|sports",
+  "声音": "contentCopilot",
+  "人物": "dialogCopilot",
+  "人物#tvV2": "dialogCopilot|video",
+  "问答": "dialogCopilot|open-class|shopping|productAgent",
+  "问答#tvV2": "dialogCopilot|open-class|shopping|productAgent|video",
+  "百科": "dialogCopilot",
+  "百科#tvV2": "dialogCopilot|video",
+  "医疗": "dialogCopilot",
+  "星座": "dialogCopilot",
+  "菜谱": "dialogCopilot",
+  "民俗": "dialogCopilot",
+  "运动": "contentCopilot|dialogCopilot|sports|qabot",
+  "体育赛事问答": "contentCopilot|dialogCopilot|sports|qabot",
+  "词典": "dialogCopilot",
+  "彩票": "dialogCopilot",
+  "投资": "dialogCopilot",
+  "股票": "dialogCopilot",
+  "闲聊": "dialogCopilot|feedback|scenes|fitnessHealth",
+  "闲聊#tvV2": "dialogCopilot|video|feedback|scenes|help|fitnessHealth|shopping|productAgent",
+  "用户反馈": "dialogCopilot|feedback",
+  "角色扮演": "dialogCopilot",
+  "重说": "dialogCopilot",
+  "图片问答": "dialogCopilot|imageLU|mapCopilot",
+  "屏幕问答": "dialogCopilot|imageLU|mapCopilot|smartApp|controlCopilot|productAgent|auto|lifeCopilot|lifeAgent",
+  "拍照问答": "dialogCopilot|imageLU|mapCopilot",
+  "搜索": "dialogCopilot|search",
+  "搜索#tvV2": "dialogCopilot|search|video|contentCopilot",
+  "搜索#soundbox": "dialogCopilot|search|music|contentCopilot",
+  "健康监控#glass": "dialogCopilot|toolsCopilot",
+  "地图导航": "mapCopilot|lifeCopilot|lifeAgent",
+  "地图导航#tvV2": "mapCopilot|lifeCopilot|lifeAgent|mapApp",
+  "地图问答": "mapCopilot|qabot|dialogCopilot",
+  "走哪问哪": "mapCopilot|qabot|dialogCopilot",
+  "地图设置": "mapCopilot|soundboxControl|controlCopilot|toolsCopilot|todolist",
+  "违章查询": "mapCopilot",
+  "限号": "mapCopilot",
+  "地址设置": "toolsCopilot|mapCopilot|mapApp",
+  "图片": "aicreativeCopilot|dialogCopilot|qabot",
+  "图像创作": "aicreativeCopilot|dialogCopilot|qabot",
+  "视频创作": "aicreativeCopilot|dialogCopilot|qabot",
+  "作文": "aicreativeCopilot",
+  "文本创作": "aicreativeCopilot|dialogCopilot|toolsCopilot",
+  "代码创作": "aicreativeCopilot",
+  "图像编辑": "aicreativeCopilot",
+  "时间": "toolsCopilot",
+  "天气": "toolsCopilot|dialogCopilot|qabot",
+  "数学": "toolsCopilot",
+  "数学应用题": "toolsCopilot",
+  "电话": "toolsCopilot",
+  "看图打电话": "toolsCopilot",
+  "翻译": "toolsCopilot",
+  "翻译#glass": "toolsCopilot|dialogCopilot|qabot",
+  "闹钟": "toolsCopilot",
+  "备忘录": "toolsCopilot|dialogCopilot|michat",
+  "记忆": "toolsCopilot",
+  "结构记忆": "toolsCopilot|dialogCopilot|michat",
+  "视频收藏": "toolsCopilot|controlCopilot|smartApp",
+  "电话问答": "toolsCopilot|dialogCopilot|qabot",
+  "课程表": "toolsCopilot",
+  "内容总结": "toolsCopilot|aicreativeCopilot|dialogCopilot|qabot",
+  "留言": "toolsCopilot|message",
+  "健康监控": "toolsCopilot|fitnessHealth|todolist|dialogCopilot|qabot",
+  "健康控制": "toolsCopilot|fitnessHealth|todolist|controlCopilot|soundboxControl|smartApp",
+  "声纹": "controlCopilot|voiceprint",
+  "家庭传声": "controlCopilot|iotCopilot|smartMiot|dialogCopilot|qabot|michat|contentCopilot|music|station",
+  "系统控制": "controlCopilot|iotCopilot|productAgent",
+  "萌宠控制": "controlCopilot|iotCopilot|productAgent",
+  "屏幕操作": "controlCopilot|iotCopilot|productAgent",
+  "音色切换": "controlCopilot|iotCopilot|productAgent",
+  "车载控制": "controlCopilot|iotCopilot|productAgent",
+  "车载设备状态查询": "controlCopilot|iotCopilot|productAgent",
+  "系统查询": "controlCopilot|iotCopilot|productAgent",
+  "设备控制": "controlCopilot|iotCopilot",
+  "家用设备状态查询": "controlCopilot|iotCopilot|productAgent",
+  "应用控制": "controlCopilot|iotCopilot|lifeCopilot|lifeAgent|fitnessHealth",
+  "生活缴费": "controlCopilot",
+  "应用窗口控制": "controlCopilot",
+  "应用控制#tvV2": "controlCopilot|iotCopilot|lifeCopilot|lifeAgent|fitnessHealth|video|soundboxControl|phonecall",
+  "应用控制#soundbox": "controlCopilot|contentCopilot|music|station|dialogCopilot|michat|qabot|video|fitnessHealth|lifeCopilot",
+  "应用控制搜索": "controlCopilot|smartApp",
+  "浏览器搜索": "controlCopilot|search|smartApp",
+  "相机": "controlCopilot",
+  "设备查找": "controlCopilot|smartMiot|michat",
+  "系统定时控制": "controlCopilot|iotCopilot|productAgent",
+  "设备定时控制": "controlCopilot|iotCopilot",
+  "应用定时控制": "controlCopilot|iotCopilot|lifeCopilot|lifeAgent|fitnessHealth",
+  "帮助": "productAgent|soundboxControl|controlCopilot",
+  "汽车手册": "productAgent|dialogCopilot|qabot",
+  "客服问答": "productAgent|soundboxControl|controlCopilot",
+  "购物": "productAgent|soundboxControl|controlCopilot",
+  "产品问答": "productAgent",
+  "手车互联": "productAgent|soundboxControl|controlCopilot",
+  "产品推荐": "productAgent",
+  "产品购买": "productAgent",
+  "外卖": "lifeCopilot",
+  "外卖|应用": "lifeCopilot|controlCopilot|smartApp",
+  "美食": "lifeCopilot",
+  "美食|应用": "lifeCopilot|controlCopilot|smartApp",
+  "景点": "lifeCopilot",
+  "景点|应用": "lifeCopilot|controlCopilot|smartApp",
+  "购票": "lifeCopilot",
+  "购票|应用": "lifeCopilot|controlCopilot|smartApp",
+  "电影": "lifeCopilot",
+  "电影|应用": "lifeCopilot|controlCopilot|smartApp",
+  "酒店": "lifeCopilot",
+  "酒店|应用": "lifeCopilot|controlCopilot|smartApp",
+  "生活休闲": "lifeCopilot",
+  "车服": "lifeCopilot",
+  "汽车服务": "lifeCopilot",
+  "团购": "lifeCopilot",
+  "打车": "lifeCopilot",
+  "航班查询": "lifeCopilot",
+  "通讯服务": "lifeCopilot",
+  "快递服务": "lifeCopilot|productAgent|shopping",
+  "产品推荐|参数对比": "lifeCopilot|productAgent|shopping",
+  "产品推荐|信息对比": "lifeCopilot|productAgent|shopping",
+  "平台比价": "productAgent|shopping",
+  "找同款": "lifeCopilot|productAgent|shopping",
+  "订单查询": "lifeCopilot|productAgent",
+  "订座排号": "lifeCopilot",
+  "前车识别": "dialogCopilot|imageLU|qabot",
+  "频道": "contentCopilot",
+  "频道#tvV2": "contentCopilot|controlCopilot|smartMiot|iotCopilot",
+  "频道#soundbox": "controlCopilot|smartMiot|iotCopilot",
+  "地图|搜索": "mapCopilot|controlCopilot|smartApp",
+  "生活服务|搜索": "lifeCopilot|lifeAgent|controlCopilot|smartApp",
+  "应用|购买": "controlCopilot|smartApp",
+  "地图控制": "mapCopilot|soundboxControl|controlCopilot|toolsCopilot|todolist",
+  "限行": "mapCopilot",
+  "音乐播放": "contentCopilot",
+  "音乐播放#tvV2": "contentCopilot|video",
+  "音乐播放#miCar": "contentCopilot|station",
+  "音乐播放#screenSoundbox": "contentCopilot|music",
+  "音乐播放#glass": "contentCopilot|ancientPoem",
+  "电台播放": "contentCopilot",
+  "电台播放#tvV2": "contentCopilot|video",
+  "视频播放": "contentCopilot",
+  "电视频道": "contentCopilot",
+  "电视频道#tvV2": "contentCopilot|controlCopilot|smartMiot|iotCopilot",
+  "电视频道#soundbox": "controlCopilot|smartMiot|iotCopilot",
+  "媒体资源问答": "contentCopilot|qabot|dialogCopilot",
+  "媒体资源问答#tvV2": "contentCopilot|qabot|dialogCopilot|video",
+  "媒体资源问答#soundbox": "contentCopilot|qabot|dialogCopilot|michat",
+  "媒体应用播放": "contentCopilot|controlCopilot|soundboxControl|smartApp",
+  "媒体资源切换": "contentCopilot|soundboxControl|controlCopilot",
+  "播放器控制": "contentCopilot|soundboxControl|controlCopilot",
+  "体育赛事播放": "contentCopilot|qabot|dialogCopilot|sports",
+  "体育赛事预约": "contentCopilot|qabot|dialogCopilot|sports",
+  "古诗词播放": "contentCopilot",
+  "古诗词问答": "contentCopilot",
+  "讲笑话": "contentCopilot",
+  "声音博物馆": "contentCopilot",
+  "闹钟计时器": "toolsCopilot",
+  "简单数学问题": "toolsCopilot",
+  "打电话": "toolsCopilot",
+  "发短信": "toolsCopilot",
+  "通讯录": "toolsCopilot",
+  "电话号码查询": "toolsCopilot|dialogCopilot|qabot",
+  "外语翻译": "toolsCopilot",
+  "翻译控制": "toolsCopilot",
+  "外语词典查询": "toolsCopilot",
+  "外语问答": "toolsCopilot",
+  "外语翻译#glass": "toolsCopilot|dialogCopilot|qabot",
+  "外语问答#glass": "toolsCopilot|dialogCopilot|qabot",
+  "外语词典查询#glass": "toolsCopilot|dialogCopilot|qabot",
+  "翻译控制#glass": "toolsCopilot|dialogCopilot|qabot",
+  "提醒": "toolsCopilot|dialogCopilot|michat",
+  "信息记忆": "toolsCopilot",
+  "记忆查询": "toolsCopilot|dialogCopilot|michat",
+  "个人信息": "toolsCopilot|dialogCopilot|michat",
+  "声纹设置": "controlCopilot|voiceprint",
+  "相机控制": "controlCopilot",
+  "系统状态查询": "controlCopilot|iotCopilot|productAgent",
+  "小爱帮助": "productAgent|soundboxControl|controlCopilot",
+  "小米产品帮助": "productAgent",
+  "商品购买": "productAgent",
+  "餐饮服务": "lifeCopilot",
+  "餐饮服务|应用": "lifeCopilot|controlCopilot|smartApp",
+  "旅游": "lifeCopilot",
+  "生活服务": "lifeCopilot",
+  "交通购票": "lifeCopilot",
+  "电影票购买": "lifeCopilot",
+  "电影票购买|应用": "lifeCopilot|controlCopilot|smartApp",
+  "航班信息查询": "lifeCopilot",
+  "话费流量服务": "lifeCopilot",
+  "购物订单查询": "lifeCopilot|productAgent",
+  "餐厅订座排号": "lifeCopilot"
+}
@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+"""分析历史干预文件，归纳设备枚举和 code 到下发 domain 的映射。"""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from intervention_common import DEFAULT_EXACT_SOURCE, DEFAULT_REGEX_SOURCE, build_source_summary, json_dumps
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="分析历史干预 TSV")
+    parser.add_argument("--exact-source", default=str(DEFAULT_EXACT_SOURCE), help="单句精确干预 TSV 路径")
+    parser.add_argument("--regex-source", default=str(DEFAULT_REGEX_SOURCE), help="正则干预 TSV 路径")
+    parser.add_argument("--top-mapping", type=int, default=30, help="输出前 N 个 code-domain 映射")
+    args = parser.parse_args()
+
+    summary = build_source_summary(Path(args.exact_source), Path(args.regex_source))
+    mapping_items = sorted(
+        summary["code_domain_mapping"].items(),
+        key=lambda item: item[1]["count"],
+        reverse=True,
+    )
+    summary["code_domain_mapping"] = dict(mapping_items[: args.top_mapping])
+    print(json_dumps(summary))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,247 @@
+#!/usr/bin/env python3
+"""审计现有干预文件和当前规则的贴合度。
+
+这个脚本用于回答两个问题：
+1. 现有单句精确干预/正则干预文件自身是否存在明显格式问题。
+2. 当前 label-master 校验、type_match_agent 下发表和历史数据是否匹配。
+
+它不会修改任何文件，只输出 JSON 摘要和少量样例。
+"""
+
+from __future__ import annotations
+
+import argparse
+import re
+from collections import Counter, defaultdict
+from pathlib import Path
+from typing import Any
+
+from intervention_common import (
+    DEFAULT_EXACT_SOURCE,
+    DEFAULT_REGEX_SOURCE,
+    build_source_summary,
+    domain_compatible,
+    infer_configured_dispatch_domain,
+    json_dumps,
+    load_exact_records,
+    load_label_validator,
+    load_regex_records,
+)
+
+
+def sample_append(bucket: list[dict[str, Any]], item: dict[str, Any], limit: int) -> None:
+    if len(bucket) < limit:
+        bucket.append(item)
+
+
+def validate_unique_targets(targets: set[str]) -> dict[str, dict[str, Any]]:
+    validator = load_label_validator()
+    return {target: validator.validate(target) for target in sorted(targets)}
+
+
+def audit_exact_records(summary: dict[str, Any], target_results: dict[str, dict[str, Any]], sample_limit: int) -> dict[str, Any]:
+    records = load_exact_records(Path(summary["source_files"]["exact"]))
+    empty_fields = Counter()
+    target_invalid = Counter()
+    configured_mismatch = Counter()
+    duplicate_keys: dict[tuple[str, str], list[Any]] = defaultdict(list)
+    samples = {
+        "empty_fields": [],
+        "invalid_targets": [],
+        "configured_domain_mismatch": [],
+        "conflicts": [],
+    }
+
+    for record in records:
+        if not record.device:
+            empty_fields["device"] += 1
+        if not record.query:
+            empty_fields["query"] += 1
+        if not record.target:
+            empty_fields["target"] += 1
+        if not record.dispatch_domain:
+            empty_fields["dispatch_domain"] += 1
+        if any(not value for value in (record.device, record.query, record.target, record.dispatch_domain)):
+            sample_append(samples["empty_fields"], record.__dict__, sample_limit)
+
+        result = target_results.get(record.target)
+        if result and not result.get("valid"):
+            target_invalid[record.target] += 1
+            sample_append(
+                samples["invalid_targets"],
+                {
+                    "line_no": record.line_no,
+                    "target": record.target,
+                    "errors": result.get("errors", []),
+                },
+                sample_limit,
+            )
+
+        configured = infer_configured_dispatch_domain(record.target, record.device, summary)
+        if configured and not domain_compatible(record.dispatch_domain, configured["dispatch_domain"]):
+            configured_mismatch[configured["matched_key"]] += 1
+            sample_append(
+                samples["configured_domain_mismatch"],
+                {
+                    "line_no": record.line_no,
+                    "device": record.device,
+                    "query": record.query,
+                    "target": record.target,
+                    "dispatch_domain": record.dispatch_domain,
+                    "configured": configured,
+                },
+                sample_limit,
+            )
+
+        duplicate_keys[(record.device, record.query)].append(record)
+
+    conflicts = []
+    duplicate_count = 0
+    for (device, query), items in duplicate_keys.items():
+        if len(items) <= 1:
+            continue
+        duplicate_count += len(items)
+        variants = {(item.target, item.dispatch_domain) for item in items}
+        if len(variants) > 1:
+            conflict = {
+                "device": device,
+                "query": query,
+                "lines": [item.line_no for item in items[:10]],
+                "variants": sorted([{"target": target, "dispatch_domain": domain} for target, domain in variants], key=str),
+            }
+            conflicts.append(conflict)
+            sample_append(samples["conflicts"], conflict, sample_limit)
+
+    return {
+        "total": len(records),
+        "empty_fields": dict(empty_fields),
+        "invalid_target_total": sum(target_invalid.values()),
+        "invalid_target_top": dict(target_invalid.most_common(20)),
+        "configured_domain_mismatch_total": sum(configured_mismatch.values()),
+        "configured_domain_mismatch_top": dict(configured_mismatch.most_common(20)),
+        "duplicate_row_count_by_device_query": duplicate_count,
+        "conflict_key_count": len(conflicts),
+        "samples": samples,
+    }
+
+
+def audit_regex_records(target_results: dict[str, dict[str, Any]], summary: dict[str, Any], sample_limit: int) -> dict[str, Any]:
+    records = load_regex_records(Path(summary["source_files"]["regex"]))
+    empty_fields = Counter()
+    regex_errors = Counter()
+    anchor_warnings = 0
+    missing_query_fragment = 0
+    target_invalid = Counter()
+    duplicate_keys: dict[tuple[str, str], list[Any]] = defaultdict(list)
+    samples = {
+        "empty_fields": [],
+        "regex_errors": [],
+        "missing_query_fragment": [],
+        "invalid_targets": [],
+        "conflicts": [],
+    }
+
+    for record in records:
+        if not record.device:
+            empty_fields["device"] += 1
+        if not record.pattern:
+            empty_fields["pattern"] += 1
+        if not record.target:
+            empty_fields["target"] += 1
+        if any(not value for value in (record.device, record.pattern, record.target)):
+            sample_append(samples["empty_fields"], record.__dict__, sample_limit)
+
+        if record.pattern:
+            try:
+                re.compile(record.pattern)
+            except re.error as exc:
+                regex_errors[str(exc)] += 1
+                sample_append(
+                    samples["regex_errors"],
+                    {"line_no": record.line_no, "pattern": record.pattern, "error": str(exc)},
+                    sample_limit,
+                )
+            if "query#" not in record.pattern:
+                missing_query_fragment += 1
+                sample_append(samples["missing_query_fragment"], record.__dict__, sample_limit)
+            if not record.pattern.startswith("^") or not record.pattern.endswith("$"):
+                anchor_warnings += 1
+
+        result = target_results.get(record.target)
+        if result and not result.get("valid"):
+            target_invalid[record.target] += 1
+            sample_append(
+                samples["invalid_targets"],
+                {
+                    "line_no": record.line_no,
+                    "target": record.target,
+                    "errors": result.get("errors", []),
+                },
+                sample_limit,
+            )
+
+        duplicate_keys[(record.device, record.pattern)].append(record)
+
+    conflicts = []
+    duplicate_count = 0
+    for (device, pattern), items in duplicate_keys.items():
+        if len(items) <= 1:
+            continue
+        duplicate_count += len(items)
+        variants = {item.target for item in items}
+        if len(variants) > 1:
+            conflict = {
+                "device": device,
+                "pattern": pattern,
+                "lines": [item.line_no for item in items[:10]],
+                "variants": sorted(variants),
+            }
+            conflicts.append(conflict)
+            sample_append(samples["conflicts"], conflict, sample_limit)
+
+    return {
+        "total": len(records),
+        "empty_fields": dict(empty_fields),
+        "regex_error_total": sum(regex_errors.values()),
+        "regex_error_top": dict(regex_errors.most_common(20)),
+        "missing_query_fragment": missing_query_fragment,
+        "anchor_warning_count": anchor_warnings,
+        "invalid_target_total": sum(target_invalid.values()),
+        "invalid_target_top": dict(target_invalid.most_common(20)),
+        "duplicate_row_count_by_device_pattern": duplicate_count,
+        "conflict_key_count": len(conflicts),
+        "samples": samples,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="审计现有干预文件")
+    parser.add_argument("--exact-source", default=str(DEFAULT_EXACT_SOURCE), help="单句精确干预 TSV 路径")
+    parser.add_argument("--regex-source", default=str(DEFAULT_REGEX_SOURCE), help="正则干预 TSV 路径")
+    parser.add_argument("--sample-limit", type=int, default=5, help="每类问题最多输出样例数")
+    args = parser.parse_args()
+
+    summary = build_source_summary(Path(args.exact_source), Path(args.regex_source))
+    exact_records = load_exact_records(Path(args.exact_source))
+    regex_records = load_regex_records(Path(args.regex_source))
+    target_results = validate_unique_targets({record.target for record in exact_records + regex_records if record.target})
+
+    payload = {
+        "source_files": summary["source_files"],
+        "devices": summary["devices"],
+        "counts": summary["counts"],
+        "target_validation": {
+            "unique_targets": len(target_results),
+            "valid_unique_targets": sum(1 for item in target_results.values() if item.get("valid")),
+            "invalid_unique_targets": sum(1 for item in target_results.values() if not item.get("valid")),
+        },
+        "exact": audit_exact_records(summary, target_results, args.sample_limit),
+        "regex": audit_regex_records(target_results, summary, args.sample_limit),
+    }
+    print(json_dumps(payload))
+    has_hard_errors = bool(payload["exact"]["empty_fields"] or payload["regex"]["regex_error_total"])
+    return 1 if has_hard_errors else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,141 @@
+#!/usr/bin/env python3
+"""生成单句精确干预或正则干预候选 TSV，并返回校验结果。"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+from intervention_common import (
+    DEFAULT_EXACT_SOURCE,
+    DEFAULT_REGEX_SOURCE,
+    build_query_pattern,
+    build_source_summary,
+    exact_tsv_line,
+    infer_configured_dispatch_domain,
+    infer_dispatch_domain_for_device,
+    json_dumps,
+    normalize_device,
+    normalize_target,
+    regex_tsv_line,
+)
+
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+VALIDATE_SCRIPT = SCRIPT_DIR / "validate_intervention_records.py"
+
+
+def load_payload(path: str | None) -> dict[str, Any]:
+    text = Path(path).read_text(encoding="utf-8") if path else sys.stdin.read()
+    if not text.strip():
+        raise ValueError("输入 JSON 不能为空")
+    return json.loads(text)
+
+
+def run_validator(mode: str, tsv: str, exact_source: str, regex_source: str) -> dict[str, Any]:
+    proc = subprocess.run(
+        [
+            sys.executable,
+            str(VALIDATE_SCRIPT),
+            "--mode",
+            mode,
+            "--exact-source",
+            exact_source,
+            "--regex-source",
+            regex_source,
+        ],
+        input=tsv,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=False,
+    )
+    if not proc.stdout.strip():
+        return {
+            "valid": False,
+            "errors": [proc.stderr.strip() or f"校验脚本退出：{proc.returncode}"],
+        }
+    return json.loads(proc.stdout)
+
+
+def generate_exact(payload: dict[str, Any], exact_source: str, regex_source: str) -> dict[str, Any]:
+    device = normalize_device(str(payload.get("device", "")))
+    query = str(payload.get("query", "")).strip()
+    target = normalize_target(str(payload.get("target") or payload.get("code") or ""))
+    dispatch_domain = str(payload.get("dispatch_domain") or payload.get("domain") or "").strip()
+
+    summary = build_source_summary(Path(exact_source), Path(regex_source))
+    inferred = None
+    configured = infer_configured_dispatch_domain(target, device, summary)
+    if not dispatch_domain:
+        inferred = infer_dispatch_domain_for_device(target, device, summary)
+        if inferred:
+            dispatch_domain = inferred["dispatch_domain"]
+
+    tsv = exact_tsv_line(device, query, target, dispatch_domain)
+    validation = run_validator("exact", tsv, exact_source, regex_source)
+    return {
+        "mode": "exact",
+        "tsv": tsv,
+        "record": {
+            "device": device,
+            "query": query,
+            "target": target,
+            "dispatch_domain": dispatch_domain,
+        },
+        "inferred_dispatch_domain": inferred,
+        "configured_dispatch_domain": configured,
+        "validation": validation,
+    }
+
+
+def generate_regex(payload: dict[str, Any], exact_source: str, regex_source: str) -> dict[str, Any]:
+    device = normalize_device(str(payload.get("device", "")))
+    target = normalize_target(str(payload.get("target") or payload.get("label") or ""))
+    raw_pattern = str(payload.get("pattern") or payload.get("regex") or "").strip()
+    query = str(payload.get("query") or payload.get("current_query") or "").strip()
+    before_queries = payload.get("before_queries") or []
+    if not isinstance(before_queries, list):
+        raise ValueError("before_queries 必须是字符串数组")
+    pattern = raw_pattern or build_query_pattern(query, [str(item) for item in before_queries])
+
+    tsv = regex_tsv_line(device, pattern, target)
+    validation = run_validator("regex", tsv, exact_source, regex_source)
+    return {
+        "mode": "regex",
+        "tsv": tsv,
+        "record": {
+            "device": device,
+            "pattern": pattern,
+            "target": target,
+        },
+        "validation": validation,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="生成干预候选 TSV")
+    parser.add_argument("--input", help="输入 JSON 文件；不传则读取 stdin")
+    parser.add_argument("--exact-source", default=str(DEFAULT_EXACT_SOURCE), help="历史单句精确干预 TSV")
+    parser.add_argument("--regex-source", default=str(DEFAULT_REGEX_SOURCE), help="历史正则干预 TSV")
+    args = parser.parse_args()
+
+    payload = load_payload(args.input)
+    mode = str(payload.get("mode", "")).strip().lower()
+    if mode == "exact":
+        result = generate_exact(payload, args.exact_source, args.regex_source)
+    elif mode == "regex":
+        result = generate_regex(payload, args.exact_source, args.regex_source)
+    else:
+        raise ValueError("mode 必须是 exact 或 regex")
+
+    print(json_dumps(result))
+    return 0 if result.get("validation", {}).get("valid") else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,330 @@
+#!/usr/bin/env python3
+"""干预数据脚本的公共能力。
+
+这里尽量只做确定性处理：
+- 读取历史干预 TSV。
+- 归纳设备枚举和 code 到下发 domain 的多数映射。
+- 复用 label-master 校验 target 语法。
+- 校验候选干预条目是否和历史数据重复或冲突。
+"""
+
+from __future__ import annotations
+
+import csv
+import importlib.util
+import json
+import re
+import sys
+from collections import Counter, defaultdict
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+SKILL_ROOT = SCRIPT_DIR.parent
+REPO_ROOT = SKILL_ROOT.parent.parent
+DEFAULT_EXACT_SOURCE = REPO_ROOT / "干预skill" / "20260134"
+DEFAULT_REGEX_SOURCE = REPO_ROOT / "干预skill" / "20260207"
+LABEL_VALIDATOR_PATH = REPO_ROOT / "skills" / "label-master" / "scripts" / "validate_label_output.py"
+CONFIG_MAPPING_PATH = SKILL_ROOT / "knowledge" / "type_match_agent.json"
+AGENT_TARGET_RE = re.compile(r'^Agent\(\s*tag\s*=\s*["\'](.+?)["\']\s*\)$')
+
+
+@dataclass(frozen=True)
+class ExactRecord:
+    device: str
+    query: str
+    target: str
+    dispatch_domain: str
+    line_no: int
+
+
+@dataclass(frozen=True)
+class RegexRecord:
+    device: str
+    pattern: str
+    target: str
+    line_no: int
+
+
+def normalize_device(device: str) -> str:
+    """归一化历史数据里的设备写法，但保留常见大小写。"""
+
+    value = device.strip()
+    if value.startswith("#"):
+        value = value[1:].strip()
+    aliases = {
+        "micar": "miCar",
+        "miai": "miai",
+        "tVV2": "tvV2",
+        "tvv2": "tvV2",
+        "unify": "unify",
+    }
+    return aliases.get(value, value)
+
+
+def normalize_target(target: str) -> str:
+    return target.replace("“", '"').replace("”", '"').replace("‘", "'").replace("’", "'").strip()
+
+
+def read_tsv_rows(path: Path) -> list[list[str]]:
+    with path.open("r", encoding="utf-8-sig", newline="") as handle:
+        return list(csv.reader(handle, delimiter="\t"))
+
+
+def load_exact_records(path: Path = DEFAULT_EXACT_SOURCE) -> list[ExactRecord]:
+    rows = read_tsv_rows(path)
+    records: list[ExactRecord] = []
+    for index, row in enumerate(rows, start=1):
+        if index == 1 and row[:4] == ["#设备名", "query", "code", "下发domain"]:
+            continue
+        if not row or all(not cell.strip() for cell in row):
+            continue
+        padded = [*row, "", "", "", ""]
+        records.append(
+            ExactRecord(
+                device=normalize_device(padded[0]),
+                query=padded[1].strip(),
+                target=normalize_target(padded[2]),
+                dispatch_domain=padded[3].strip(),
+                line_no=index,
+            )
+        )
+    return records
+
+
+def load_regex_records(path: Path = DEFAULT_REGEX_SOURCE) -> list[RegexRecord]:
+    rows = read_tsv_rows(path)
+    records: list[RegexRecord] = []
+    for index, row in enumerate(rows, start=1):
+        if not row or all(not cell.strip() for cell in row):
+            continue
+        padded = [*row, "", ""]
+        records.append(
+            RegexRecord(
+                device=normalize_device(padded[0]),
+                pattern=padded[1].strip(),
+                target=normalize_target(padded[2]),
+                line_no=index,
+            )
+        )
+    return records
+
+
+def build_source_summary(
+    exact_path: Path = DEFAULT_EXACT_SOURCE,
+    regex_path: Path = DEFAULT_REGEX_SOURCE,
+) -> dict[str, Any]:
+    exact_records = load_exact_records(exact_path) if exact_path.exists() else []
+    regex_records = load_regex_records(regex_path) if regex_path.exists() else []
+
+    exact_devices = Counter(record.device for record in exact_records)
+    regex_devices = Counter(record.device for record in regex_records)
+    devices = sorted(set(exact_devices) | set(regex_devices))
+
+    code_domain_counts: dict[str, Counter[str]] = defaultdict(Counter)
+    for record in exact_records:
+        if record.target and record.dispatch_domain:
+            code_domain_counts[record.target][record.dispatch_domain] += 1
+
+    code_domain_mapping: dict[str, dict[str, Any]] = {}
+    for target, counts in code_domain_counts.items():
+        domain, count = counts.most_common(1)[0]
+        total = sum(counts.values())
+        code_domain_mapping[target] = {
+            "dispatch_domain": domain,
+            "count": count,
+            "total": total,
+            "confidence": round(count / total, 4) if total else 0,
+            "alternatives": [
+                {"dispatch_domain": item_domain, "count": item_count}
+                for item_domain, item_count in counts.most_common(5)
+            ],
+        }
+
+    config_mapping = load_type_match_agent_mapping()
+
+    return {
+        "source_files": {
+            "exact": str(exact_path),
+            "regex": str(regex_path),
+        },
+        "counts": {
+            "exact_records": len(exact_records),
+            "regex_records": len(regex_records),
+        },
+        "devices": devices,
+        "device_counts": {
+            "exact": dict(exact_devices.most_common()),
+            "regex": dict(regex_devices.most_common()),
+        },
+        "code_domain_mapping": code_domain_mapping,
+        "configured_code_domain_mapping": config_mapping,
+        "historical_targets": sorted({record.target for record in exact_records + regex_records if record.target}),
+        "top_exact_targets": dict(Counter(record.target for record in exact_records).most_common(50)),
+        "top_regex_targets": dict(Counter(record.target for record in regex_records).most_common(50)),
+    }
+
+
+def historical_targets(summary: dict[str, Any]) -> set[str]:
+    targets = set(summary.get("top_exact_targets", {})) | set(summary.get("top_regex_targets", {}))
+    targets.update(summary.get("code_domain_mapping", {}))
+    targets.update(summary.get("historical_targets", []))
+    return {normalize_target(target) for target in targets if target}
+
+
+def infer_dispatch_domain(target: str, summary: dict[str, Any]) -> dict[str, Any] | None:
+    normalized = normalize_target(target)
+    historical = summary.get("code_domain_mapping", {}).get(normalized)
+    if historical:
+        return {**historical, "source": "historical"}
+    configured = infer_configured_dispatch_domain(normalized, "unify", summary)
+    if configured:
+        return {
+            "dispatch_domain": configured["dispatch_domain"],
+            "count": 0,
+            "total": 0,
+            "confidence": None,
+            "alternatives": [],
+            "source": "type_match_agent",
+            "matched_key": configured["matched_key"],
+        }
+    return None
+
+
+def infer_dispatch_domain_for_device(target: str, device: str, summary: dict[str, Any]) -> dict[str, Any] | None:
+    """按设备推断推荐下发 domain。
+
+    优先级：
+    1. `标签#设备` 这种明确配置。
+    2. 历史精确干预表里的全局多数映射。
+    3. `标签` 通用配置。
+    """
+
+    normalized = normalize_target(target)
+    configured = infer_configured_dispatch_domain(normalized, device, summary)
+    if configured and "#" in configured["matched_key"]:
+        return {
+            "dispatch_domain": configured["dispatch_domain"],
+            "count": 0,
+            "total": 0,
+            "confidence": None,
+            "alternatives": [],
+            "source": "type_match_agent",
+            "matched_key": configured["matched_key"],
+        }
+
+    historical = summary.get("code_domain_mapping", {}).get(normalized)
+    if historical:
+        return {**historical, "source": "historical"}
+
+    if configured:
+        return {
+            "dispatch_domain": configured["dispatch_domain"],
+            "count": 0,
+            "total": 0,
+            "confidence": None,
+            "alternatives": [],
+            "source": "type_match_agent",
+            "matched_key": configured["matched_key"],
+        }
+    return None
+
+
+def load_type_match_agent_mapping(path: Path = CONFIG_MAPPING_PATH) -> dict[str, str]:
+    if not path.exists():
+        return {}
+    data = json.loads(path.read_text(encoding="utf-8"))
+    return {str(key): str(value) for key, value in data.items()}
+
+
+def extract_agent_tag(target: str) -> str | None:
+    match = AGENT_TARGET_RE.match(normalize_target(target))
+    if not match:
+        return None
+    return match.group(1)
+
+
+def infer_configured_dispatch_domain(target: str, device: str, summary: dict[str, Any]) -> dict[str, str] | None:
+    tag = extract_agent_tag(target)
+    if not tag:
+        return None
+    mapping = summary.get("configured_code_domain_mapping", {})
+    normalized_device = normalize_device(device)
+    for key in (f"{tag}#{normalized_device}", tag):
+        if key in mapping:
+            return {"matched_key": key, "dispatch_domain": mapping[key]}
+    return None
+
+
+def split_domain(domain: str) -> list[str]:
+    return [part for part in domain.split("|") if part]
+
+
+def domain_compatible(actual: str, expected: str) -> bool:
+    """判断实际下发 domain 是否和配置基线兼容。
+
+    下发表里有些是宽泛入口，例如 `音乐 -> contentCopilot`；
+    历史干预里常见更细路径，例如 `contentCopilot|music`。
+    只要二者一方的管道 token 是另一方的子集，就认为兼容。
+    """
+
+    actual_parts = set(split_domain(actual))
+    expected_parts = set(split_domain(expected))
+    if not actual_parts or not expected_parts:
+        return False
+    return actual_parts.issubset(expected_parts) or expected_parts.issubset(actual_parts)
+
+
+def load_label_validator() -> Any:
+    spec = importlib.util.spec_from_file_location("label_master_validate_label_output", LABEL_VALIDATOR_PATH)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"无法加载 label-master 校验器：{LABEL_VALIDATOR_PATH}")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    manifest_path = module.DEFAULT_OUTPUT
+    validator = module.LabelOutputValidator(module.load_manifest(manifest_path))
+    return validator
+
+
+def validate_target(target: str) -> dict[str, Any]:
+    validator = load_label_validator()
+    # 历史 TSV 中多行 function/Agent 目标常以字面量 `\n` 存在，
+    # label-master 校验器需要真实换行才能按多行程序解析。
+    return validator.validate(normalize_target(target).replace("\\n", "\n"))
+
+
+def escape_regex_literal(text: str) -> str:
+    return re.escape(text.strip())
+
+
+def build_query_pattern(query: str, before_queries: list[str] | None = None) -> str:
+    before_queries = before_queries or []
+    parts: list[str] = []
+    for before_query in before_queries:
+        if before_query.strip():
+            parts.append(f"beforeQuery#{escape_regex_literal(before_query)}")
+    parts.append(f"query#{escape_regex_literal(query)}")
+    return "^" + "#".join(parts) + "$"
+
+
+def exact_tsv_line(device: str, query: str, target: str, dispatch_domain: str) -> str:
+    return "\t".join([normalize_device(device), query.strip(), normalize_target(target), dispatch_domain.strip()])
+
+
+def regex_tsv_line(device: str, pattern: str, target: str) -> str:
+    return "\t".join([normalize_device(device), pattern.strip(), normalize_target(target)])
+
+
+def parse_candidate_lines(text: str) -> list[list[str]]:
+    rows = []
+    for line in text.splitlines():
+        if line.strip():
+            rows.append(next(csv.reader([line], delimiter="\t")))
+    return rows
+
+
+def json_dumps(payload: Any) -> str:
+    return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)
@@ -0,0 +1,220 @@
+#!/usr/bin/env python3
+"""校验单句精确干预和正则干预候选 TSV。"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from pathlib import Path
+from typing import Any
+
+from intervention_common import (
+    DEFAULT_EXACT_SOURCE,
+    DEFAULT_REGEX_SOURCE,
+    build_source_summary,
+    domain_compatible,
+    historical_targets,
+    infer_configured_dispatch_domain,
+    infer_dispatch_domain_for_device,
+    load_exact_records,
+    load_regex_records,
+    normalize_device,
+    normalize_target,
+    parse_candidate_lines,
+    validate_target,
+    json_dumps,
+)
+
+
+def load_input_text(path: str | None) -> str:
+    if path:
+        return Path(path).read_text(encoding="utf-8")
+    return sys.stdin.read()
+
+
+def validate_exact_rows(rows: list[list[str]], args: argparse.Namespace) -> dict[str, Any]:
+    summary = build_source_summary(Path(args.exact_source), Path(args.regex_source))
+    devices = set(summary["devices"])
+    known_historical_targets = historical_targets(summary)
+    existing: dict[tuple[str, str], list[Any]] = {}
+    for record in load_exact_records(Path(args.exact_source)):
+        existing.setdefault((record.device, record.query), []).append(record)
+
+    results = []
+    for index, row in enumerate(rows, start=1):
+        errors: list[str] = []
+        warnings: list[str] = []
+        if len(row) != 4:
+            errors.append(f"单句精确干预必须是 4 列，当前 {len(row)} 列")
+            results.append({"line": index, "valid": False, "errors": errors, "warnings": warnings, "row": row})
+            continue
+
+        device, query, target, dispatch_domain = [cell.strip() for cell in row]
+        device = normalize_device(device)
+        target = normalize_target(target)
+        if device not in devices:
+            errors.append(f"设备不在历史枚举中：{device}，可选：{', '.join(sorted(devices))}")
+        if not query:
+            errors.append("query 不能为空")
+        if "\t" in query or "\n" in query or "\r" in query:
+            errors.append("query 不能包含制表符或换行")
+        target_result = validate_target(target)
+        if not target_result["valid"]:
+            if target in known_historical_targets:
+                warnings.append("target 未通过 label-master 知识库校验，但历史干预文件中已存在；建议后续补齐标签知识库")
+                target_result["normalized_output"] = target
+            else:
+                errors.extend([f"target 不合法：{message}" for message in target_result.get("errors", [])])
+        if not dispatch_domain:
+            errors.append("下发domain 不能为空")
+
+        inferred = infer_dispatch_domain_for_device(target, device, summary)
+        if inferred and dispatch_domain and dispatch_domain != inferred["dispatch_domain"]:
+            message = (
+                f"下发domain 与推断映射不一致：当前 {dispatch_domain}，"
+                f"推断为 {inferred['dispatch_domain']}，来源 {inferred.get('source', 'historical')}，置信度 {inferred.get('confidence')}"
+            )
+            if args.strict_domain:
+                errors.append(message)
+            else:
+                warnings.append(message)
+        if not inferred:
+            warnings.append("未在历史精确干预表中找到该 target 的下发domain 映射，需要人工确认")
+
+        configured = infer_configured_dispatch_domain(target, device, summary)
+        if configured and dispatch_domain and not domain_compatible(dispatch_domain, configured["dispatch_domain"]):
+            warnings.append(
+                f"下发domain 与 type_match_agent 配置不兼容：当前 {dispatch_domain}，"
+                f"配置 {configured['matched_key']} -> {configured['dispatch_domain']}"
+            )
+        if not configured:
+            warnings.append("type_match_agent 下发表中未找到该 Agent tag 的设备/通用映射")
+
+        conflicts = existing.get((device, query), [])
+        for item in conflicts:
+            if item.target == target and item.dispatch_domain == dispatch_domain:
+                warnings.append(f"历史文件已存在相同单句干预：line {item.line_no}")
+            else:
+                errors.append(
+                    f"历史文件存在同设备同 query 的不同干预：line {item.line_no}，"
+                    f"{item.target} / {item.dispatch_domain}"
+                )
+
+        results.append(
+            {
+                "line": index,
+                "valid": not errors,
+                "errors": errors,
+                "warnings": warnings,
+                "normalized": {
+                    "device": device,
+                    "query": query,
+                    "target": target_result.get("normalized_output", target),
+                    "dispatch_domain": dispatch_domain,
+                },
+                "configured_dispatch_domain": configured,
+            }
+        )
+    return summarize_results(results)
+
+
+def validate_regex_rows(rows: list[list[str]], args: argparse.Namespace) -> dict[str, Any]:
+    summary = build_source_summary(Path(args.exact_source), Path(args.regex_source))
+    devices = set(summary["devices"])
+    known_historical_targets = historical_targets(summary)
+    existing: dict[tuple[str, str], list[Any]] = {}
+    for record in load_regex_records(Path(args.regex_source)):
+        existing.setdefault((record.device, record.pattern), []).append(record)
+
+    results = []
+    for index, row in enumerate(rows, start=1):
+        errors: list[str] = []
+        warnings: list[str] = []
+        if len(row) != 3:
+            errors.append(f"正则干预必须是 3 列，当前 {len(row)} 列")
+            results.append({"line": index, "valid": False, "errors": errors, "warnings": warnings, "row": row})
+            continue
+
+        device, pattern, target = [cell.strip() for cell in row]
+        device = normalize_device(device)
+        target = normalize_target(target)
+        if device not in devices:
+            errors.append(f"设备不在历史枚举中：{device}，可选：{', '.join(sorted(devices))}")
+        if not pattern:
+            errors.append("正则不能为空")
+        else:
+            try:
+                re.compile(pattern)
+            except re.error as exc:
+                errors.append(f"正则无法编译：{exc}")
+            if "query#" not in pattern:
+                errors.append("正则必须包含 query# 片段")
+            if not pattern.startswith("^") or not pattern.endswith("$"):
+                warnings.append("历史常见正则通常使用 ^...$ 完整锚定，建议确认是否需要锚定")
+
+        target_result = validate_target(target)
+        if not target_result["valid"]:
+            if target in known_historical_targets:
+                warnings.append("target 未通过 label-master 知识库校验，但历史干预文件中已存在；建议后续补齐标签知识库")
+                target_result["normalized_output"] = target
+            else:
+                errors.extend([f"target 不合法：{message}" for message in target_result.get("errors", [])])
+
+        conflicts = existing.get((device, pattern), [])
+        for item in conflicts:
+            if item.target == target:
+                warnings.append(f"历史文件已存在相同正则干预：line {item.line_no}")
+            else:
+                errors.append(
+                    f"历史文件存在同设备同正则的不同干预：line {item.line_no}，{item.target}"
+                )
+
+        results.append(
+            {
+                "line": index,
+                "valid": not errors,
+                "errors": errors,
+                "warnings": warnings,
+                "normalized": {
+                    "device": device,
+                    "pattern": pattern,
+                    "target": target_result.get("normalized_output", target),
+                },
+            }
+        )
+    return summarize_results(results)
+
+
+def summarize_results(results: list[dict[str, Any]]) -> dict[str, Any]:
+    return {
+        "valid": all(item["valid"] for item in results),
+        "total": len(results),
+        "invalid": sum(1 for item in results if not item["valid"]),
+        "warning_count": sum(len(item.get("warnings", [])) for item in results),
+        "results": results,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="校验干预 TSV 候选")
+    parser.add_argument("--mode", choices=["exact", "regex"], required=True, help="干预类型")
+    parser.add_argument("--input", help="候选 TSV 文件路径；不传则读取 stdin")
+    parser.add_argument("--exact-source", default=str(DEFAULT_EXACT_SOURCE), help="历史单句精确干预 TSV")
+    parser.add_argument("--regex-source", default=str(DEFAULT_REGEX_SOURCE), help="历史正则干预 TSV")
+    parser.add_argument("--strict-domain", action="store_true", help="下发domain 和历史多数映射不一致时直接报错")
+    args = parser.parse_args()
+
+    text = load_input_text(args.input)
+    rows = parse_candidate_lines(text)
+    if args.mode == "exact":
+        payload = validate_exact_rows(rows, args)
+    else:
+        payload = validate_regex_rows(rows, args)
+    print(json_dumps(payload))
+    return 0 if payload["valid"] else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,35 @@
+# model-labeling
+
+对已有数据调用线上模型接口批量打标。
+
+典型输入：
+
+- `output/records.jsonl`
+- `output/training.jsonl`
+- `output/eval_planning.csv`
+- `output/records.csv`
+
+典型流程：
+
+1. 确认输入文件路径。
+2. 确认线上模型 `generate` URL。
+3. 用 `batch_label_model.py` 的 `dry_run=true` 识别格式。
+4. 批量请求模型，输出 `output/model_predictions.jsonl`。
+
+执行示例：
+
+```json
+{
+  "script_path": "skills/model-labeling/scripts/batch_label_model.py",
+  "stdin": {
+    "input_path": "output/records.jsonl",
+    "model_url": "http://example/generate",
+    "output_path": "output/model_predictions.jsonl",
+    "parameters": {
+      "max_new_tokens": 64
+    }
+  },
+  "timeout_seconds": 600,
+  "max_output_chars": 20000
+}
+```
@@ -0,0 +1,159 @@
+---
+name: model-labeling
+description: 对已有数据集调用线上模型 generate 接口批量打标，支持 canonical records、训练 jsonl、评测 CSV 和同事流转表格的统一归一化。
+when_to_use: 当用户已有一份数据，或刚通过 product-data / online-mining-v2 生成数据后，希望指定线上模型 URL 批量请求模型、得到预测标签、对比真实标签或产出标注结果时使用。
+aliases: batch-labeling, model-annotation, online-model-labeling, 模型打标
+allowed_tools: read_file, write_file, grep_search, glob_search, ask_user_question, python_exec
+---
+
+# Model Labeling
+
+使用这个 skill 处理“已有数据 -> 统一样本格式 -> 调线上模型接口批量打标 -> 输出预测结果”的流程。
+
+线上模型 URL 经常变化，**不要猜 URL**。如果用户没有明确提供 `http://.../generate` 或等价接口地址，必须先询问用户。
+
+## 能力组织
+
+```text
+skills/model-labeling/
+  SKILL.md
+  knowledge/
+    input_formats.md
+  scripts/
+    batch_label_model.py
+```
+
+`batch_label_model.py` 是 portable script，只依赖 Python 标准库。它会：
+
+- 识别 canonical records JSON/JSONL。
+- 识别 product-data 导出的训练 JSONL。
+- 识别 eval/planning CSV，优先使用 `newPrompt`。
+- 识别同事流转 CSV，使用 `query`、`prev_session`、`context`、`function`。
+- 对未知格式返回结构化错误，要求用户提供字段映射，不要擅自转换。
+- 调用用户提供的模型接口，默认请求体为：
+
+```json
+{
+  "inputs": "<prompt>",
+  "parameters": {
+    "max_new_tokens": 64
+  }
+}
+```
+
+## 交互规则
+
+开始执行前必须确认：
+
+1. **输入数据路径**：用户给出的文件路径，或上一轮产物路径。
+2. **模型接口 URL**：必须是用户明确提供的 URL；没有就问。
+3. **输入格式是否可识别**：canonical records、训练 jsonl、eval CSV、同事流转表格可以直接处理。
+4. **未知格式的字段映射**：如果脚本提示 unknown format，需要问用户：
+   - 哪列是 query？
+   - 哪列是 prompt？
+   - 哪列是真实标签？
+   - 是否有历史上下文、context？
+5. **输出路径**：默认写到当前 session 的 `output/model_predictions.jsonl`。
+
+不要在没有 URL 的情况下开始打标。不要把临时结果写到项目根目录。所有输出优先放当前 session 的 `output/`。
+
+## 推荐流程
+
+### 1. 找到输入文件
+
+如果用户说“刚才生成的数据”“这份数据”，先用会话文件面板或 `glob_search` / `read_file` 找到实际路径。常见路径：
+
+```text
+output/records.jsonl
+output/training.jsonl
+output/eval_planning.csv
+output/records.csv
+```
+
+### 2. 确认模型 URL
+
+如果用户没有提供 URL，直接问：
+
+```text
+请提供这次要调用的线上模型 generate 接口 URL，例如 http://.../generate。
+```
+
+如果 `ask_user_question` 可用，优先使用；不可用就普通回复提问并停止。
+
+### 3. 先 dry run 识别格式
+
+先执行一次 `dry_run=true`，只识别格式和样例，不请求模型：
+
+```json
+{
+  "script_path": "skills/model-labeling/scripts/batch_label_model.py",
+  "stdin": {
+    "input_path": "output/records.jsonl",
+    "model_url": "http://example/generate",
+    "dry_run": true,
+    "max_records": 3
+  },
+  "timeout_seconds": 60,
+  "max_output_chars": 20000
+}
+```
+
+如果返回 `ok=false` 且 `needs_mapping=true`，必须把错误和已识别字段展示给用户，让用户说明映射。
+
+### 4. 批量请求模型
+
+确认格式后执行：
+
+```json
+{
+  "script_path": "skills/model-labeling/scripts/batch_label_model.py",
+  "stdin": {
+    "input_path": "output/records.jsonl",
+    "model_url": "http://example/generate",
+    "output_path": "output/model_predictions.jsonl",
+    "parameters": {
+      "max_new_tokens": 64
+    },
+    "timeout_seconds": 60
+  },
+  "timeout_seconds": 600,
+  "max_output_chars": 20000
+}
+```
+
+`output_path` 使用相对 `output/...`，平台会路由到当前 session output 目录。
+
+### 5. 展示结果
+
+执行完成后，简短展示：
+
+- 输入格式。
+- 处理条数、成功数、失败数。
+- 输出路径。
+- 抽 3 条预测样例。
+
+如果存在真实标签，说明输出里包含 `gold_label`，后续可以继续做准确率或错误分析。
+
+## 输出格式
+
+默认输出 JSONL，一行一条紧凑 JSON：
+
+```json
+{"index":0,"request_id":"...","query":"...","gold_label":"complex=false\nAgent(tag=\"地图导航\")","prediction":"...","ok":true,"latency_ms":123}
+```
+
+如果请求失败：
+
+```json
+{"index":0,"query":"...","gold_label":"...","prediction":"","ok":false,"error":"HTTP 500 ...","latency_ms":123}
+```
+
+默认不把完整 prompt 写入结果，避免文件过大。如确实需要排查，可传 `include_prompt=true`。
+
+## 注意事项
+
+- URL、鉴权 header、特殊请求体字段都以用户提供为准。
+- 默认接口字段是 `inputs` 和 `parameters`；如果用户说明接口不同，需要在脚本输入里传 `request_template`。
+- 大批量请求前先小样本 dry run。
+- 打标脚本只负责请求模型和记录预测结果，不负责修改原始数据。
+- 后续准确率统计、错误聚类、补数计划可以再交给其他 skill。
@@ -0,0 +1,81 @@
+# Model Labeling 输入格式
+
+本 skill 的脚本会把不同来源的数据统一成 sample：
+
+```json
+{
+  "index": 0,
+  "request_id": "",
+  "query": "",
+  "prompt": "",
+  "gold_label": "",
+  "source_format": ""
+}
+```
+
+## canonical records
+
+识别条件：
+
+- JSONL 每行是对象，或 JSON 数组 / `{ "records": [...] }`。
+- 对象包含 `turn.query`。
+
+字段来源：
+
+- `query`：`turn.query`
+- `request_id`：`source.request_id`
+- `gold_label`：`complex=true/false` + `label.target`
+- `prompt`：按 product-data 的 planning prompt 规则生成
+
+## training jsonl
+
+识别条件：
+
+- 每行是对象。
+- 包含 `instruction` 或 `system`。
+
+字段来源：
+
+- `prompt`：`system + instruction` 包成 chat template；如果已有 `prompt` 则直接使用。
+- `gold_label`：`output`
+- `query`：尽力从 `[当前query]` 后的 `用户:` 抽取。
+
+## eval/planning CSV
+
+识别条件：
+
+- CSV 表头包含 `newPrompt` 或 `query`。
+
+字段来源：
+
+- `prompt`：优先 `newPrompt`
+- `query`：`query`
+- `gold_label`：`code标签`，如果有 `complex` 列则组合成两行
+
+## 同事流转 CSV
+
+识别条件：
+
+- CSV 表头包含 `query` 和 `function`。
+
+字段来源：
+
+- `query`：`query`
+- `request_id`：`request_id`
+- `gold_label`：`function`
+- `prompt`：根据 `query`、`prev_session`、`context` 生成 planning prompt
+
+## 未知格式
+
+如果不能识别，脚本返回：
+
+```json
+{
+  "ok": false,
+  "needs_mapping": true,
+  "columns": ["..."],
+  "error": "..."
+}
+```
+
+此时必须询问用户字段映射，不要猜。
@@ -0,0 +1,600 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+"""批量调用线上模型 generate 接口给数据打标。
+
+输入通过 stdin 或 --input 传 JSON 对象，输出稳定 JSON 对象到 stdout。
+脚本只依赖 Python 标准库，便于在本地、Linux runtime 用户或远程工作区迁移执行。
+"""
+
+import argparse
+import csv
+import json
+import os
+import re
+import sys
+import time
+import urllib.error
+import urllib.request
+from io import StringIO
+from pathlib import Path
+from typing import Any
+
+
+DEFAULT_SYSTEM_PROMPT = "你是小爱同学，中文智能语音助手。"
+DEFAULT_PARAMETERS = {"max_new_tokens": 64}
+
+
+class LabelingError(ValueError):
+    """输入参数或数据格式错误。"""
+
+
+def main() -> int:
+    try:
+        payload = load_payload()
+        result = run(payload)
+        emit({"ok": True, **result})
+        return 0
+    except LabelingError as exc:
+        emit({"ok": False, "error": str(exc), **getattr(exc, "extra", {})})
+        return 1
+    except Exception as exc:  # noqa: BLE001 - CLI 需要稳定 JSON 错误
+        emit({"ok": False, "error": str(exc)})
+        return 1
+
+
+def load_payload() -> dict[str, Any]:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--input", "-i", help="JSON 参数文件；不传则从 stdin 读取")
+    args = parser.parse_args()
+    text = Path(args.input).read_text(encoding="utf-8") if args.input else sys.stdin.read()
+    if not text.strip():
+        raise LabelingError("input JSON is required")
+    payload = json.loads(text)
+    if not isinstance(payload, dict):
+        raise LabelingError("input JSON must be an object")
+    return payload
+
+
+def emit(payload: dict[str, Any]) -> None:
+    print(json.dumps(payload, ensure_ascii=False, separators=(",", ":")))
+
+
+def run(payload: dict[str, Any]) -> dict[str, Any]:
+    input_path = require_string(payload, "input_path")
+    model_url = require_string(payload, "model_url")
+    dry_run = bool(payload.get("dry_run", False))
+    include_prompt = bool(payload.get("include_prompt", False))
+    max_records = optional_int(payload.get("max_records"))
+    start_index = int(payload.get("start_index") or 0)
+    timeout_seconds = float(payload.get("timeout_seconds") or 60)
+    output_path = str(payload.get("output_path") or "output/model_predictions.jsonl")
+    parameters = payload.get("parameters")
+    if parameters is None:
+        parameters = dict(DEFAULT_PARAMETERS)
+    if not isinstance(parameters, dict):
+        raise LabelingError("parameters must be an object")
+    headers = payload.get("headers")
+    if headers is None:
+        headers = {"Content-Type": "application/json"}
+    if not isinstance(headers, dict):
+        raise LabelingError("headers must be an object")
+    request_template = payload.get("request_template")
+    if request_template is not None and not isinstance(request_template, dict):
+        raise LabelingError("request_template must be an object")
+
+    source = read_input_file(input_path)
+    samples = normalize_samples(
+        source,
+        field_mapping=payload.get("field_mapping"),
+        system_prompt=str(payload.get("system_prompt") or DEFAULT_SYSTEM_PROMPT),
+        session_num=int(payload.get("session_num") or 5),
+        session_time_minutes=int(payload.get("session_time_minutes") or 5),
+    )
+    if start_index:
+        samples = samples[start_index:]
+    if max_records is not None:
+        samples = samples[:max_records]
+
+    summary = {
+        "input_path": input_path,
+        "source_format": source["format"],
+        "total_samples": len(samples),
+        "sample_preview": preview_samples(samples),
+    }
+    if dry_run:
+        return {**summary, "dry_run": True}
+
+    if not model_url.startswith(("http://", "https://")):
+        raise LabelingError("model_url must start with http:// or https://")
+    results: list[dict[str, Any]] = []
+    ok_count = 0
+    output = resolve_runtime_path(output_path)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    with output.open("w", encoding="utf-8") as file:
+        for sample in samples:
+            item = request_one(
+                sample,
+                model_url=model_url,
+                parameters=parameters,
+                headers={str(k): str(v) for k, v in headers.items()},
+                timeout_seconds=timeout_seconds,
+                request_template=request_template,
+                include_prompt=include_prompt,
+            )
+            if item.get("ok"):
+                ok_count += 1
+            results.append(item)
+            file.write(json.dumps(item, ensure_ascii=False, separators=(",", ":")) + "\n")
+
+    return {
+        **summary,
+        "dry_run": False,
+        "output_path": str(output),
+        "success_count": ok_count,
+        "failure_count": len(results) - ok_count,
+        "result_preview": results[:3],
+    }
+
+
+def request_one(
+    sample: dict[str, Any],
+    *,
+    model_url: str,
+    parameters: dict[str, Any],
+    headers: dict[str, str],
+    timeout_seconds: float,
+    request_template: dict[str, Any] | None,
+    include_prompt: bool,
+) -> dict[str, Any]:
+    prompt = str(sample["prompt"])
+    body = build_request_body(prompt, parameters, request_template)
+    started = time.time()
+    base = {
+        "index": sample["index"],
+        "request_id": sample.get("request_id", ""),
+        "query": sample.get("query", ""),
+        "gold_label": sample.get("gold_label", ""),
+        "source_format": sample.get("source_format", ""),
+    }
+    if include_prompt:
+        base["prompt"] = prompt
+    try:
+        request = urllib.request.Request(
+            model_url,
+            data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
+            headers=headers,
+            method="POST",
+        )
+        with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
+            text = response.read().decode("utf-8", errors="replace")
+            status = getattr(response, "status", 200)
+        latency_ms = int((time.time() - started) * 1000)
+        decoded = try_json(text)
+        return {
+            **base,
+            "prediction": extract_prediction(decoded, text),
+            "ok": 200 <= int(status) < 300,
+            "status": int(status),
+            "latency_ms": latency_ms,
+            "response": decoded if decoded is not None else text,
+        }
+    except urllib.error.HTTPError as exc:
+        text = exc.read().decode("utf-8", errors="replace")
+        return {
+            **base,
+            "prediction": extract_prediction(try_json(text), text),
+            "ok": False,
+            "status": exc.code,
+            "latency_ms": int((time.time() - started) * 1000),
+            "error": f"HTTP {exc.code}: {text[:500]}",
+        }
+    except Exception as exc:  # noqa: BLE001 - 单条失败不中断整体批次
+        return {
+            **base,
+            "prediction": "",
+            "ok": False,
+            "latency_ms": int((time.time() - started) * 1000),
+            "error": str(exc),
+        }
+
+
+def build_request_body(
+    prompt: str,
+    parameters: dict[str, Any],
+    request_template: dict[str, Any] | None,
+) -> dict[str, Any]:
+    if request_template is None:
+        return {"inputs": prompt, "parameters": parameters}
+    return replace_placeholders(request_template, {"prompt": prompt, "parameters": parameters})
+
+
+def replace_placeholders(value: Any, variables: dict[str, Any]) -> Any:
+    if isinstance(value, str):
+        if value == "{{prompt}}":
+            return variables["prompt"]
+        if value == "{{parameters}}":
+            return variables["parameters"]
+        return value.replace("{{prompt}}", str(variables["prompt"]))
+    if isinstance(value, list):
+        return [replace_placeholders(item, variables) for item in value]
+    if isinstance(value, dict):
+        return {key: replace_placeholders(item, variables) for key, item in value.items()}
+    return value
+
+
+def read_input_file(path: str) -> dict[str, Any]:
+    file_path = resolve_runtime_path(path)
+    if not file_path.exists():
+        raise LabelingError(f"input_path not found: {path}")
+    suffix = file_path.suffix.lower()
+    text = file_path.read_text(encoding="utf-8", errors="replace")
+    if suffix in {".csv", ".tsv"}:
+        delimiter = "\t" if suffix == ".tsv" else ","
+        rows = list(csv.DictReader(StringIO(text), delimiter=delimiter))
+        return {"format": "table", "path": str(file_path), "rows": rows, "columns": list(rows[0].keys()) if rows else []}
+    if suffix == ".jsonl":
+        rows = [json.loads(line) for line in text.splitlines() if line.strip()]
+        return {"format": "jsonl", "path": str(file_path), "rows": rows}
+    decoded = json.loads(text)
+    if isinstance(decoded, dict) and isinstance(decoded.get("records"), list):
+        decoded = decoded["records"]
+    if isinstance(decoded, list):
+        return {"format": "json", "path": str(file_path), "rows": decoded}
+    raise_with_mapping("JSON input must be an array or {records:[...]}", columns=list(decoded.keys()) if isinstance(decoded, dict) else [])
+
+
+def normalize_samples(
+    source: dict[str, Any],
+    *,
+    field_mapping: Any,
+    system_prompt: str,
+    session_num: int,
+    session_time_minutes: int,
+) -> list[dict[str, Any]]:
+    rows = source.get("rows")
+    if not isinstance(rows, list) or not rows:
+        raise LabelingError("input file contains no rows")
+    if isinstance(field_mapping, dict):
+        return samples_from_mapping(rows, field_mapping, source["format"], system_prompt)
+    first = rows[0]
+    if not isinstance(first, dict):
+        raise_with_mapping("rows must be objects")
+    if is_canonical_record(first):
+        return [
+            sample_from_record(index, row, system_prompt, session_num, session_time_minutes)
+            for index, row in enumerate(rows)
+            if isinstance(row, dict)
+        ]
+    if is_training_row(first):
+        return [sample_from_training(index, row, system_prompt) for index, row in enumerate(rows) if isinstance(row, dict)]
+    if source["format"] == "table":
+        columns = list(first.keys())
+        if "newPrompt" in columns:
+            return [sample_from_eval_row(index, row) for index, row in enumerate(rows) if isinstance(row, dict)]
+        if "query" in columns and "function" in columns:
+            return [sample_from_flow_row(index, row, system_prompt) for index, row in enumerate(rows) if isinstance(row, dict)]
+        raise_with_mapping("unrecognized table format", columns=columns)
+    raise_with_mapping("unrecognized JSON/JSONL format", columns=list(first.keys()))
+
+
+def sample_from_record(
+    index: int,
+    record: dict[str, Any],
+    system_prompt: str,
+    session_num: int,
+    session_time_minutes: int,
+) -> dict[str, Any]:
+    source = record.get("source") if isinstance(record.get("source"), dict) else {}
+    turn = record.get("turn") if isinstance(record.get("turn"), dict) else {}
+    return {
+        "index": index,
+        "request_id": str(source.get("request_id") or ""),
+        "query": str(turn.get("query") or ""),
+        "prompt": build_planning_prompt(record, system_prompt, session_num, session_time_minutes),
+        "gold_label": combined_label(record),
+        "source_format": "canonical_record_v1",
+    }
+
+
+def sample_from_training(index: int, row: dict[str, Any], system_prompt: str) -> dict[str, Any]:
+    prompt = str(row.get("prompt") or "")
+    if not prompt:
+        system = str(row.get("system") or system_prompt)
+        instruction = str(row.get("instruction") or row.get("input") or "")
+        prompt = wrap_chat_prompt(system, instruction)
+    return {
+        "index": index,
+        "request_id": str(row.get("request_id") or ""),
+        "query": extract_query_from_prompt(prompt),
+        "prompt": prompt,
+        "gold_label": str(row.get("output") or row.get("target") or ""),
+        "source_format": "training_jsonl",
+    }
+
+
+def sample_from_eval_row(index: int, row: dict[str, Any]) -> dict[str, Any]:
+    gold = str(row.get("code标签") or row.get("function") or row.get("target") or "")
+    complex_value = row.get("complex")
+    if complex_value not in (None, "") and not gold.startswith("complex="):
+        gold = f"complex={normalize_bool_literal(complex_value)}\n{gold}".rstrip()
+    prompt = str(row.get("newPrompt") or row.get("prompt") or "")
+    query = str(row.get("query") or "")
+    if not prompt:
+        prompt = build_planning_prompt(minimal_record(query, {}, [], gold), DEFAULT_SYSTEM_PROMPT, 5, 5)
+    return {
+        "index": index,
+        "request_id": str(row.get("request_id") or ""),
+        "query": query,
+        "prompt": prompt,
+        "gold_label": gold,
+        "source_format": "eval_csv",
+    }
+
+
+def sample_from_flow_row(index: int, row: dict[str, Any], system_prompt: str) -> dict[str, Any]:
+    record = record_from_flow_row(row)
+    return {
+        "index": index,
+        "request_id": str(row.get("request_id") or ""),
+        "query": str(row.get("query") or ""),
+        "prompt": build_planning_prompt(record, system_prompt, 5, 5),
+        "gold_label": str(row.get("function") or ""),
+        "source_format": "flow_csv",
+    }
+
+
+def samples_from_mapping(
+    rows: list[Any],
+    mapping: dict[str, Any],
+    source_format: str,
+    system_prompt: str,
+) -> list[dict[str, Any]]:
+    query_field = str(mapping.get("query") or "")
+    prompt_field = str(mapping.get("prompt") or "")
+    label_field = str(mapping.get("label") or mapping.get("target") or "")
+    request_id_field = str(mapping.get("request_id") or "")
+    if not query_field and not prompt_field:
+        raise LabelingError("field_mapping must provide query or prompt")
+    samples: list[dict[str, Any]] = []
+    for index, row in enumerate(rows):
+        if not isinstance(row, dict):
+            continue
+        prompt = str(row.get(prompt_field) or "")
+        query = str(row.get(query_field) or "")
+        if not prompt:
+            record = minimal_record(query, {}, [], str(row.get(label_field) or ""))
+            prompt = build_planning_prompt(record, system_prompt, 5, 5)
+        samples.append(
+            {
+                "index": index,
+                "request_id": str(row.get(request_id_field) or ""),
+                "query": query or extract_query_from_prompt(prompt),
+                "prompt": prompt,
+                "gold_label": str(row.get(label_field) or ""),
+                "source_format": f"{source_format}_mapped",
+            }
+        )
+    return samples
+
+
+def is_canonical_record(row: dict[str, Any]) -> bool:
+    return isinstance(row.get("turn"), dict) and bool(row["turn"].get("query"))
+
+
+def is_training_row(row: dict[str, Any]) -> bool:
+    return any(key in row for key in ("instruction", "system", "output", "prompt"))
+
+
+def build_planning_prompt(
+    record: dict[str, Any],
+    system_prompt: str,
+    session_num: int,
+    session_time_minutes: int,
+) -> str:
+    instruction = build_training_instruction(record, session_num, session_time_minutes)
+    return wrap_chat_prompt(system_prompt, instruction)
+
+
+def wrap_chat_prompt(system_prompt: str, instruction: str) -> str:
+    return (
+        f"<|im_start|>system\n{system_prompt}<|im_end|>\n"
+        f"<|im_start|>user\n{instruction}<|im_end|>\n"
+        "<|im_start|>assistant\n"
+    )
+
+
+def build_training_instruction(record: dict[str, Any], session_num: int, session_time_minutes: int) -> str:
+    turn = record.get("turn") if isinstance(record.get("turn"), dict) else {}
+    query = str(turn.get("query") or "")
+    context = record.get("context") if isinstance(record.get("context"), dict) else {}
+    prev_session = record.get("prev_session") if isinstance(record.get("prev_session"), list) else []
+    current_ts = optional_int(turn.get("timestamp"))
+    history = render_history(prev_session, current_ts, session_num, session_time_minutes)
+    return (
+        "请参考用户的[当前query]、[对话历史]、[知识注入]、[系统状态]识别出[当前query]的[function]结果，[function]是python的code形式。\n"
+        "[知识注入]\n"
+        f"{json.dumps({'location': str(context.get('location') or ''), 'rag': str(context.get('rag') or '')}, ensure_ascii=False, indent=0)}\n"
+        "[系统状态]\n"
+        "{}\n"
+        "[对话历史]\n"
+        f"{history}"
+        "[当前query]\n"
+        f"用户: {query}\n"
+        "[function]\n"
+    )
+
+
+def render_history(prev_session: list[Any], current_ts: int | None, session_num: int, session_time_minutes: int) -> str:
+    usable: list[dict[str, Any]] = []
+    for item in prev_session[-session_num:]:
+        if not isinstance(item, dict):
+            continue
+        ts = optional_int(item.get("timestamp"))
+        if current_ts is not None and ts is not None:
+            if abs(current_ts - ts) > session_time_minutes * 60_000:
+                continue
+        usable.append(item)
+    if not usable:
+        return ""
+    lines: list[str] = []
+    for item in usable:
+        query = str(item.get("query") or "").strip()
+        tts = str(item.get("tts") or "").strip()
+        if query:
+            lines.append(f"用户: {query}")
+        if tts:
+            lines.append(f"小爱: {tts}")
+    return "\n".join(lines) + ("\n" if lines else "")
+
+
+def record_from_flow_row(row: dict[str, Any]) -> dict[str, Any]:
+    prev_session = parse_json_cell(row.get("prev_session"), default=[])
+    context = parse_json_cell(row.get("context"), default={})
+    return minimal_record(
+        str(row.get("query") or ""),
+        context if isinstance(context, dict) else {},
+        prev_session if isinstance(prev_session, list) else [],
+        str(row.get("function") or ""),
+        request_id=str(row.get("request_id") or ""),
+        timestamp=optional_int(row.get("timestamp")),
+    )
+
+
+def minimal_record(
+    query: str,
+    context: dict[str, Any],
+    prev_session: list[Any],
+    target: str,
+    *,
+    request_id: str = "",
+    timestamp: int | None = None,
+) -> dict[str, Any]:
+    return {
+        "source": {"request_id": request_id, "timestamp": timestamp},
+        "turn": {"query": query, "timestamp": timestamp},
+        "prev_session": prev_session,
+        "context": context,
+        "label": {"target": target},
+        "dimensions": {},
+    }
+
+
+def combined_label(record: dict[str, Any]) -> str:
+    target = ""
+    label = record.get("label") if isinstance(record.get("label"), dict) else {}
+    if isinstance(label, dict):
+        target = str(label.get("target") or "")
+    dimensions = record.get("dimensions") if isinstance(record.get("dimensions"), dict) else {}
+    complex_value = dimensions.get("complex") if isinstance(dimensions, dict) else None
+    if isinstance(complex_value, bool):
+        return f"complex={'true' if complex_value else 'false'}\n{target}".rstrip()
+    return target
+
+
+def extract_query_from_prompt(prompt: str) -> str:
+    match = re.search(r"\[当前query\]\s*\n用户[:：]\s*(.+)", prompt)
+    return match.group(1).strip() if match else ""
+
+
+def normalize_bool_literal(value: Any) -> str:
+    text = str(value).strip().lower()
+    return "true" if text in {"true", "1", "yes", "y", "是", "复杂"} else "false"
+
+
+def parse_json_cell(value: Any, default: Any) -> Any:
+    if value in (None, ""):
+        return default
+    if isinstance(value, (dict, list)):
+        return value
+    try:
+        return json.loads(str(value))
+    except json.JSONDecodeError:
+        return default
+
+
+def resolve_runtime_path(path: str) -> Path:
+    raw = Path(path).expanduser()
+    if raw.is_absolute():
+        return raw
+    scratchpad = Path(str(Path.cwd()))
+    if os.environ.get("PYTHON_EXEC_SCRATCHPAD"):
+        scratchpad = Path(os.environ["PYTHON_EXEC_SCRATCHPAD"]).expanduser()
+    parts = raw.parts
+    if parts and parts[0] in {"output", "outputs"}:
+        return scratchpad.parent / "output" / Path(*parts[1:])
+    if parts and parts[0] in {"input", "inputs"}:
+        return scratchpad.parent / "input" / Path(*parts[1:])
+    if parts and parts[0] in {"scratchpad", "scratch"}:
+        return scratchpad / Path(*parts[1:])
+    return raw
+
+
+def optional_int(value: Any) -> int | None:
+    if value in (None, ""):
+        return None
+    try:
+        return int(float(str(value)))
+    except (TypeError, ValueError):
+        return None
+
+
+def extract_prediction(decoded: Any, text: str) -> str:
+    if isinstance(decoded, dict):
+        for key in ("generated_text", "text", "output", "response", "result"):
+            value = decoded.get(key)
+            if isinstance(value, str):
+                return value.strip()
+        outputs = decoded.get("outputs")
+        if isinstance(outputs, list) and outputs:
+            first = outputs[0]
+            if isinstance(first, str):
+                return first.strip()
+            if isinstance(first, dict):
+                return extract_prediction(first, json.dumps(first, ensure_ascii=False))
+        choices = decoded.get("choices")
+        if isinstance(choices, list) and choices:
+            first = choices[0]
+            if isinstance(first, dict):
+                message = first.get("message")
+                if isinstance(message, dict) and isinstance(message.get("content"), str):
+                    return message["content"].strip()
+                if isinstance(first.get("text"), str):
+                    return first["text"].strip()
+    return text.strip()
+
+
+def try_json(text: str) -> Any:
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        return None
+
+
+def preview_samples(samples: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    return [
+        {
+            "index": item.get("index"),
+            "query": item.get("query"),
+            "gold_label": item.get("gold_label"),
+            "prompt_preview": str(item.get("prompt") or "")[:200],
+        }
+        for item in samples[:3]
+    ]
+
+
+def require_string(payload: dict[str, Any], key: str) -> str:
+    value = payload.get(key)
+    if not isinstance(value, str) or not value.strip():
+        raise LabelingError(f"{key} is required")
+    return value.strip()
+
+
+def raise_with_mapping(message: str, columns: list[str] | None = None) -> None:
+    exc = LabelingError(message)
+    exc.extra = {"needs_mapping": True, "columns": columns or []}  # type: ignore[attr-defined]
+    raise exc
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())