Stabilize product data export flow

skills/product-data/README.md

@@ -43,7 +43,8 @@ cat input.json | python skills/product-data/scripts/<tool>.py
 | --- | --- |
 | `scripts/normalize_dataset_draft.py` | 把 dataset draft text v1 转成 canonical records |
 | `scripts/validate_dataset_records.py` | 校验 canonical records |
-| `scripts/export_dataset_records.py` | 导出紧凑 JSONL/JSON |
+| `scripts/export_dataset_records.py` | 导出紧凑 JSONL/JSON，并默认生成同目录 `records.csv` 表格 |
+| `scripts/export_dataset_table.py` | 已有 canonical records 时，只补生成同事流转表格 |
 
 ## 平台内优先级
 

skills/product-data/SKILL.md

@@ -28,6 +28,7 @@ skills/product-data/
     normalize_dataset_draft.py
     validate_dataset_records.py
     export_dataset_records.py
+    export_dataset_table.py
 ```
 
 在 ZK Data Agent 平台里，优先使用已注册的 `data_agent_*` 工具，因为它们带有平台级 review 状态、确认门禁和会话 output 路由。
@@ -164,9 +165,16 @@ review 展示必须简短清晰，不要重复解释工具和流程。每次 rev
 11. 生成 dataset draft text v1。
 12. 调用 `data_agent_normalize_dataset_draft`，必须传入 `confirmed_plan_id`。
 13. 调用 `data_agent_validate_dataset_records`。
-14. 如果用户要求落盘 canonical records，调用 `data_agent_export_dataset_records`，`output_path` 固定传 `output/records.jsonl`，默认导出紧凑 JSONL，不要用 `write_file` 手写 JSON。
+14. 如果用户要求落盘 canonical records，调用 `data_agent_export_dataset_records`，`output_path` 固定传 `output/records.jsonl`，默认导出紧凑 JSONL，并同时生成同目录 `records.csv` 表格，不要用 `write_file` 手写 JSON 或 CSV。
 15. 本阶段默认只推进到 canonical metadata records；除非用户另行要求，不做最终训练/评测格式导出。
 
+生成阶段要避免一次性把大量数据塞进工具参数：
+
+- 单次 `data_agent_normalize_dataset_draft` 最多处理 8 条 case；计划数量更多时，分批生成、分批 normalize，再汇总校验和导出。
+- dataset draft 中的 Agent target 推荐写成单引号形式，例如 `target: Agent(tag='地图导航')`。工具会规范化为 `Agent(tag="地图导航")`，这样可以降低 tool call JSON 里双引号转义失败的概率。
+- 如果 draft 已经保存在文件中，优先传 `draft_path`，不要再把大段 `draft_text` 作为工具参数传入。
+- 校验和导出 records 时，如果 records 已经保存在 JSON/JSONL 文件中，优先传 `records_path`。
+
 ## 数据生成输出格式
 
 当需要生成数据样本时，默认使用 dataset draft text v1，不要直接输出 JSON、JSONL、CSV 或最终表格格式。除非用户明确要求机器可读格式，否则优先输出便于人工 review 的文本格式。
@@ -196,6 +204,7 @@ notes: 可选，说明覆盖的问题或边界
 - `小爱:` 表示小爱回复 tts。
 - 最后一个 `用户:` 是本轮 query。
 - `target:` 是本轮 query 对应的监督标签，必须存在。
+- 为避免工具参数 JSON 转义失败，dataset draft 里 Agent 标签优先写 `target: Agent(tag='xxx')`；转换工具会统一规范为 `Agent(tag="xxx")`。
 - 单轮数据只需要写一行 `用户:`，然后写 `target:`。
 - 多轮数据需要按照时间顺序写多组 `用户:` / `小爱:`。
 - 多轮数据的最后一轮只写 `用户:` 和 `target:`，不要写最后一轮 `小爱:`，因为本轮 query 不包含 tts。
@@ -207,7 +216,7 @@ notes: 可选，说明覆盖的问题或边界
 
 `data_agent_normalize_dataset_draft` 会把 dataset draft text 转成 canonical records，并统一补齐 `record_id`、`source`、`timestamp`、`context`、`target_type` 等机械字段。
 
-canonical records 落盘必须使用 `data_agent_export_dataset_records`，默认格式是紧凑 JSONL：一行一个 canonical record，不带外层数组，不手写缩进 JSON。默认 `output_path` 固定传 `output/records.jsonl`；不要传 `output/<数据集名>/records.jsonl`，不要传 `tasks/...`，不要自定义文件名。只有用户明确要求兼容旧文件时，才使用 `output_format="json"` 导出紧凑 JSON 数组，此时工具会归一为 `output/records.json`。
+canonical records 落盘必须使用 `data_agent_export_dataset_records`，默认格式是紧凑 JSONL：一行一个 canonical record，不带外层数组，不手写缩进 JSON。默认 `output_path` 固定传 `output/records.jsonl`；不要传 `output/<数据集名>/records.jsonl`，不要传 `tasks/...`，不要自定义文件名。工具默认同时在同目录生成 `records.csv`，用于同事之间流转和表格查看。只有用户明确要求兼容旧文件时，才使用 `output_format="json"` 导出紧凑 JSON 数组，此时工具会归一为 `output/records.json`，但表格仍然是同目录 `records.csv`。
 
 当前 canonical record v1 工作格式：
 
@@ -337,12 +346,18 @@ skills/product-data/scripts/normalize_dataset_draft.py
 
 ```json
 {
-  "draft_text": "# dataset_label: 地图餐饮边界\n\n### case: 找附近美食\n用户: 附近有什么好吃的\ntarget: Agent(tag=\"餐饮服务\")",
+  "draft_text": "# dataset_label: 地图餐饮边界\n\n### case: 找附近美食\n用户: 附近有什么好吃的\ntarget: Agent(tag='餐饮服务')",
   "batch_id": "aabbccdd",
   "source_type": "generated"
 }
 ```
 
+如果草稿已经落盘，也可以传：
+
+```json
+{"draft_path": "output/dataset_draft.txt", "batch_id": "aabbccdd", "source_type": "generated"}
+```
+
 用途：把 dataset draft text v1 转成 canonical records。
 
 ### `product_data_validate_dataset_records`
@@ -387,6 +402,31 @@ skills/product-data/scripts/export_dataset_records.py
 
 用途：导出紧凑 JSONL/JSON。默认文件名是 `records.jsonl` 或 `records.json`。
 
+默认还会生成同目录 `records.csv` 表格，字段固定为：
+
+```text
+request_id,timestamp,query,prev_session,context,label,是否迁移Function,function
+```
+
+### `product_data_export_dataset_table`
+
+脚本：
+
+```text
+skills/product-data/scripts/export_dataset_table.py
+```
+
+输入支持：
+
+```json
+{
+  "records_path": "output/records.jsonl",
+  "output_dir": "output"
+}
+```
+
+用途：已有 canonical records 时，只补生成同事流转表格 `records.csv`，不改写元数据文件。主流程仍优先用 `product_data_export_dataset_records`，因为它会一次性导出元数据和表格。
+
 ## 当前可用工具
 
 - `read_file`：读取用户提供的产品定义、标签定义、样例 query 文件。
@@ -405,16 +445,16 @@ skills/product-data/scripts/export_dataset_records.py
 - `data_agent_show_generation_plan`：用户要求查看当前计划，或继续上下文时需要恢复计划详情时使用。
 - `data_agent_update_generation_plan`：用户对计划提出修改意见后使用，更新计划并重新展示。
 - `data_agent_confirm_generation_plan`：用户明确确认当前计划版本后使用，获取 `confirmed_plan_id`。
-- `data_agent_normalize_dataset_draft`：用户确认计划后，把 dataset draft text v1 转成 canonical records；必须传入 `confirmed_plan_id`。
-- `data_agent_validate_dataset_records`：对 canonical records 做结构、标签、时间戳和多轮上下文校验。
-- `data_agent_export_dataset_records`：校验 canonical records 并落盘；默认写紧凑 JSONL，一行一条，固定传 `output/records.jsonl`，不要再用 `write_file` 手写 records 文件。
+- `data_agent_normalize_dataset_draft`：用户确认计划后，把 dataset draft text v1 转成 canonical records；必须传入 `confirmed_plan_id`；支持 `draft_text` 或 `draft_path`，大草稿优先 `draft_path`。
+- `data_agent_validate_dataset_records`：对 canonical records 做结构、标签、时间戳和多轮上下文校验；支持 `records` 或 `records_path`。
+- `data_agent_export_dataset_records`：校验 canonical records 并落盘；支持 `records` 或 `records_path`；默认写紧凑 JSONL，一行一条，固定传 `output/records.jsonl`，并同时生成 `output/records.csv` 表格，不要再用 `write_file` 手写 records 或表格文件。
 
 ## 约束
 
 - 不要静默解决产品或标签歧义。
 - 如果 `ask_user_question` 不可用，使用普通回复向用户提问并停止，不要自己替用户确认。
 - canonical records 通过校验前，不要生成最终导出格式。
-- canonical records 需要落盘时，必须用 `data_agent_export_dataset_records`；不要自己拼接 JSON/JSONL；不要创建数据集子目录或自定义 records 文件名。
+- canonical records 需要落盘时，必须用 `data_agent_export_dataset_records`；不要自己拼接 JSON/JSONL/CSV；不要创建数据集子目录或自定义 records 文件名。
 - 如果使用 portable scripts，它们只负责格式转换、校验和导出，不替代 `generation_goal` / `generation_plan` 的用户 review。
 - 除非用户明确要求，否则不要把“修改标签定义”和“生成数据”混在一起做。
 - 不要因为用户说“生成一些数据”就跳过边界总结和 generation plan review。

skills/product-data/knowledge/canonical_record_v1.md

@@ -44,3 +44,14 @@ canonical record 是产品数据生成链路的中间元数据格式。它不是
 - 相邻轮时间间隔超过 5 分钟时给 warning。
 - `label.target` 必须有值。
 - `target_type` 只能是 `agent`、`function` 或 `unknown`。
+
+## 默认导出
+
+调用 `data_agent_export_dataset_records` 或 portable `export_dataset_records.py` 时，默认产物包括：
+
+- `records.jsonl`：canonical records，一行一条紧凑 JSON。
+- `records.csv`：同事流转表格，字段为 `request_id,timestamp,query,prev_session,context,label,是否迁移Function,function`。
+
+`records.csv` 中的 `prev_session` 和 `context` 是紧凑 JSON 字符串；`prev_session` 中的 `timestamp` 按历史表格习惯输出为字符串。
+
+如果已有 `records.jsonl` 或 `records.json`，只需要补表格，可以执行 portable `export_dataset_table.py` 或等价工具；默认仍输出到同目录的 `records.csv`。

skills/product-data/knowledge/dataset_draft_v1.md

@@ -9,14 +9,14 @@
 
 ### case: case名称
 用户: 本轮 query
-target: Agent(tag="xxx")
+target: Agent(tag='xxx')
 notes: 可选，说明覆盖的问题或边界
 
 ### case: 多轮 case 名称
 用户: 前一轮 query
 小爱: 前一轮 tts
 用户: 本轮 query
-target: Agent(tag="xxx")
+target: Agent(tag='xxx')
 notes: 可选，说明覆盖的问题或边界
 ```
 
@@ -27,6 +27,7 @@ notes: 可选，说明覆盖的问题或边界
 - `小爱:` 表示小爱回复 tts。
 - 最后一个 `用户:` 是本轮 query。
 - `target:` 是本轮 query 对应的监督标签，必须存在。
+- Agent 标签推荐写单引号形式 `Agent(tag='xxx')`，转换工具会规范成 `Agent(tag="xxx")`。
 - 单轮数据只需要写一行 `用户:`，然后写 `target:`。
 - 多轮数据需要按照时间顺序写多组 `用户:` / `小爱:`。
 - 多轮数据的最后一轮只写 `用户:` 和 `target:`，不要写最后一轮 `小爱:`。

skills/product-data/schemas/export_dataset_records.input.schema.json

@@ -11,7 +11,8 @@
       "default": "jsonl"
     },
     "require_validation_ok": {"type": "boolean", "default": true},
-    "overwrite": {"type": "boolean", "default": true}
+    "overwrite": {"type": "boolean", "default": true},
+    "export_table": {"type": "boolean", "default": true}
   },
   "anyOf": [
     {"required": ["records"]},

skills/product-data/schemas/export_dataset_records.output.schema.json

@@ -4,8 +4,11 @@
     "ok": {"type": "boolean"},
     "output_path": {"type": "string"},
     "output_format": {"type": "string"},
+    "table_output_path": {"type": "string"},
+    "table_output_format": {"type": "string"},
     "record_count": {"type": "integer"},
     "bytes_written": {"type": "integer"},
+    "table_bytes_written": {"type": "integer"},
     "validation": {"type": "object"},
     "error": {"type": "string"}
   },

skills/product-data/schemas/export_dataset_table.input.schema.json

@@ -0,0 +1,15 @@
+{
+  "type": "object",
+  "properties": {
+    "records": {"type": "array"},
+    "records_path": {"type": "string"},
+    "output_path": {"type": "string"},
+    "output_dir": {"type": "string"},
+    "require_validation_ok": {"type": "boolean", "default": true},
+    "overwrite": {"type": "boolean", "default": true}
+  },
+  "anyOf": [
+    {"required": ["records"]},
+    {"required": ["records_path"]}
+  ]
+}

skills/product-data/schemas/export_dataset_table.output.schema.json

@@ -0,0 +1,13 @@
+{
+  "type": "object",
+  "properties": {
+    "ok": {"type": "boolean"},
+    "output_path": {"type": "string"},
+    "output_format": {"type": "string"},
+    "record_count": {"type": "integer"},
+    "bytes_written": {"type": "integer"},
+    "validation": {"type": "object"},
+    "error": {"type": "string"}
+  },
+  "required": ["ok"]
+}

skills/product-data/scripts/export_dataset_records.py

@@ -21,6 +21,7 @@ def main() -> int:
             output_format=output_format,
             require_validation_ok=bool(payload.get("require_validation_ok", True)),
             overwrite=bool(payload.get("overwrite", True)),
+            export_table=bool(payload.get("export_table", True)),
         )
         emit_success(result)
         return 0

skills/product-data/scripts/export_dataset_table.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from product_data_portable import (
+    default_table_output_path,
+    emit_error,
+    emit_success,
+    export_dataset_table,
+    load_json_payload,
+    load_records,
+)
+
+
+def main() -> int:
+    try:
+        payload = load_json_payload()
+        records = load_records(payload)
+        result = export_dataset_table(
+            records,
+            output_path=default_table_output_path(payload),
+            require_validation_ok=bool(payload.get("require_validation_ok", True)),
+            overwrite=bool(payload.get("overwrite", True)),
+        )
+        emit_success(result)
+        return 0
+    except Exception as exc:  # noqa: BLE001 - CLI 需要把错误稳定转成 JSON
+        emit_error(exc)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

skills/product-data/scripts/product_data_portable.py

@@ -7,10 +7,12 @@ from __future__ import annotations
 """
 
 import argparse
+import csv
 import json
 import re
 import sys
 import time
+from io import StringIO
 from pathlib import Path
 from typing import Any
 
@@ -88,6 +90,8 @@ def read_records_file(path: str) -> list[dict[str, Any]]:
             records.append(item)
         return records
     decoded = json.loads(text)
+    if isinstance(decoded, dict) and isinstance(decoded.get("records"), list):
+        decoded = decoded["records"]
     if not isinstance(decoded, list):
         raise ProductDataError(f"{path} must contain a JSON array")
     return decoded
@@ -374,6 +378,7 @@ def export_dataset_records(
     output_format: str = "jsonl",
     require_validation_ok: bool = True,
     overwrite: bool = True,
+    export_table: bool = True,
 ) -> dict[str, Any]:
     if output_format not in {"jsonl", "json"}:
         raise ProductDataError("output_format must be jsonl or json")
@@ -382,8 +387,11 @@ def export_dataset_records(
         raise ProductDataError(f"records failed validation with {validation['error_count']} errors")
 
     path = Path(output_path).expanduser()
+    table_path = path.with_name("records.csv")
     if path.exists() and not overwrite:
         raise ProductDataError(f"output_path already exists: {output_path}")
+    if export_table and table_path.exists() and not overwrite:
+        raise ProductDataError("table output_path already exists: records.csv")
     path.parent.mkdir(parents=True, exist_ok=True)
 
     if output_format == "jsonl":
@@ -394,13 +402,106 @@ def export_dataset_records(
     else:
         content = json.dumps(records, ensure_ascii=False, separators=(",", ":")) + "\n"
     path.write_text(content, encoding="utf-8")
-    return {
+    payload = {
         "output_path": str(path),
         "output_format": output_format,
         "record_count": len(records),
         "bytes_written": len(content.encode("utf-8")),
         "validation": validation,
     }
+    if export_table:
+        table_content = render_dataset_records_table_csv(records)
+        table_path.write_text(table_content, encoding="utf-8-sig")
+        payload.update(
+            {
+                "table_output_path": str(table_path),
+                "table_output_format": "csv",
+                "table_bytes_written": len(table_content.encode("utf-8-sig")),
+            }
+        )
+    return payload
+
+
+def export_dataset_table(
+    records: list[dict[str, Any]],
+    *,
+    output_path: str,
+    require_validation_ok: bool = True,
+    overwrite: bool = True,
+) -> dict[str, Any]:
+    validation = validate_dataset_records(records)
+    if require_validation_ok and not validation["ok"]:
+        raise ProductDataError(f"records failed validation with {validation['error_count']} errors")
+    path = Path(output_path).expanduser()
+    if path.exists() and not overwrite:
+        raise ProductDataError(f"output_path already exists: {output_path}")
+    path.parent.mkdir(parents=True, exist_ok=True)
+    content = render_dataset_records_table_csv(records)
+    path.write_text(content, encoding="utf-8-sig")
+    return {
+        "output_path": str(path),
+        "output_format": "csv",
+        "record_count": len(records),
+        "bytes_written": len(content.encode("utf-8-sig")),
+        "validation": validation,
+    }
+
+
+def render_dataset_records_table_csv(records: list[dict[str, Any]]) -> str:
+    output = StringIO()
+    writer = csv.DictWriter(output, fieldnames=dataset_table_columns(), lineterminator="\n")
+    writer.writeheader()
+    for record in records:
+        writer.writerow(dataset_record_table_row(record))
+    return output.getvalue()
+
+
+def dataset_table_columns() -> list[str]:
+    return [
+        "request_id",
+        "timestamp",
+        "query",
+        "prev_session",
+        "context",
+        "label",
+        "是否迁移Function",
+        "function",
+    ]
+
+
+def dataset_record_table_row(record: dict[str, Any]) -> dict[str, str]:
+    source = record.get("source") if isinstance(record.get("source"), dict) else {}
+    turn = record.get("turn") if isinstance(record.get("turn"), dict) else {}
+    label = record.get("label") if isinstance(record.get("label"), dict) else {}
+    context = record.get("context") if isinstance(record.get("context"), dict) else {}
+    prev_session = record.get("prev_session")
+    if not isinstance(prev_session, list):
+        prev_session = []
+    return {
+        "request_id": str(source.get("request_id") or ""),
+        "timestamp": str(source.get("timestamp") or turn.get("timestamp") or ""),
+        "query": str(turn.get("query") or ""),
+        "prev_session": json.dumps(table_prev_session(prev_session), ensure_ascii=False, separators=(",", ":")),
+        "context": json.dumps(context, ensure_ascii=False, separators=(",", ":")),
+        "label": str(label.get("dataset_label") or ""),
+        "是否迁移Function": "",
+        "function": str(label.get("target") or ""),
+    }
+
+
+def table_prev_session(prev_session: list[Any]) -> list[dict[str, str]]:
+    rows: list[dict[str, str]] = []
+    for item in prev_session[-10:]:
+        if not isinstance(item, dict):
+            continue
+        rows.append(
+            {
+                "query": str(item.get("query") or ""),
+                "tts": str(item.get("tts") or ""),
+                "timestamp": str(item.get("timestamp") or ""),
+            }
+        )
+    return rows
 
 
 def default_output_path(payload: dict[str, Any]) -> str:
@@ -413,6 +514,16 @@ def default_output_path(payload: dict[str, Any]) -> str:
     return str(Path("output") / filename)
 
 
+def default_table_output_path(payload: dict[str, Any]) -> str:
+    if isinstance(payload.get("output_path"), str) and payload["output_path"].strip():
+        return payload["output_path"].strip()
+    if isinstance(payload.get("output_dir"), str) and payload["output_dir"].strip():
+        return str(Path(payload["output_dir"]).expanduser() / "records.csv")
+    if isinstance(payload.get("records_path"), str) and payload["records_path"].strip():
+        return str(Path(payload["records_path"]).expanduser().with_name("records.csv"))
+    return str(Path("output") / "records.csv")
+
+
 def target_type(target: str) -> str:
     if re.match(r"^Agent\s*\(\s*tag\s*=", target):
         return "agent"

skills/product-data/tools.yaml

@@ -23,10 +23,18 @@ tools:
       command: python skills/product-data/scripts/validate_dataset_records.py --input <input.json>
       stdin: true
   - name: product_data_export_dataset_records
-    description: Validate canonical records and write compact JSONL or JSON files without asking the model to hand-write JSON.
+    description: Validate canonical records and write compact JSONL or JSON files plus a sibling records.csv sharing table without asking the model to hand-write files.
     script: scripts/export_dataset_records.py
     input_schema: schemas/export_dataset_records.input.schema.json
     output_schema: schemas/export_dataset_records.output.schema.json
     fallback:
       command: python skills/product-data/scripts/export_dataset_records.py --input <input.json>
       stdin: true
+  - name: product_data_export_dataset_table
+    description: Convert existing canonical records into the shared records.csv table format without rewriting the metadata file.
+    script: scripts/export_dataset_table.py
+    input_schema: schemas/export_dataset_table.input.schema.json
+    output_schema: schemas/export_dataset_table.output.schema.json
+    fallback:
+      command: python skills/product-data/scripts/export_dataset_table.py --input <input.json>
+      stdin: true