From ca2f6f36fc4d2659d2dde08d119433804812ec16 Mon Sep 17 00:00:00 2001 From: wuyang6 Date: Mon, 11 May 2026 10:58:35 +0800 Subject: [PATCH] Make product-data skill portable --- skills/product-data/README.md | 52 ++ skills/product-data/SKILL.md | 105 +++- .../knowledge/canonical_record_v1.md | 46 ++ .../knowledge/dataset_draft_v1.md | 34 ++ .../knowledge/portable_skill_contract.md | 65 +++ skills/product-data/requirements.txt | 1 + .../export_dataset_records.input.schema.json | 20 + .../export_dataset_records.output.schema.json | 13 + .../normalize_dataset_draft.input.schema.json | 20 + ...normalize_dataset_draft.output.schema.json | 10 + ...validate_dataset_records.input.schema.json | 11 + ...alidate_dataset_records.output.schema.json | 12 + .../scripts/export_dataset_records.py | 33 ++ .../scripts/normalize_dataset_draft.py | 37 ++ .../scripts/product_data_portable.py | 457 ++++++++++++++++++ .../scripts/validate_dataset_records.py | 18 + skills/product-data/tools.yaml | 32 ++ 17 files changed, 964 insertions(+), 2 deletions(-) create mode 100644 skills/product-data/README.md create mode 100644 skills/product-data/knowledge/canonical_record_v1.md create mode 100644 skills/product-data/knowledge/dataset_draft_v1.md create mode 100644 skills/product-data/knowledge/portable_skill_contract.md create mode 100644 skills/product-data/requirements.txt create mode 100644 skills/product-data/schemas/export_dataset_records.input.schema.json create mode 100644 skills/product-data/schemas/export_dataset_records.output.schema.json create mode 100644 skills/product-data/schemas/normalize_dataset_draft.input.schema.json create mode 100644 skills/product-data/schemas/normalize_dataset_draft.output.schema.json create mode 100644 skills/product-data/schemas/validate_dataset_records.input.schema.json create mode 100644 skills/product-data/schemas/validate_dataset_records.output.schema.json create mode 100644 skills/product-data/scripts/export_dataset_records.py create mode 100644 skills/product-data/scripts/normalize_dataset_draft.py create mode 100644 skills/product-data/scripts/product_data_portable.py create mode 100644 skills/product-data/scripts/validate_dataset_records.py create mode 100644 skills/product-data/tools.yaml diff --git a/skills/product-data/README.md b/skills/product-data/README.md new file mode 100644 index 0000000..4ec997b --- /dev/null +++ b/skills/product-data/README.md @@ -0,0 +1,52 @@ +# product-data + +`product-data` 是产品/标签定义到标准数据记录的可迁移 skill 包。 + +设计原则： + +- `SKILL.md` 描述流程、review 门禁和 Agent 使用方式。 +- `knowledge/` 保存数据格式、生成协议和可维护知识。 +- `scripts/` 保存可独立执行的脚本能力。 +- `tools.yaml` 描述脚本如何被平台注册成工具；没有注册能力的 Agent 也可以直接调用脚本。 + +## 可迁移工具约定 + +脚本能力是本体，平台注册是适配层。 + +每个脚本都支持两种输入： + +```bash +python skills/product-data/scripts/.py --input input.json +``` + +或： + +```bash +cat input.json | python skills/product-data/scripts/.py +``` + +脚本统一输出 JSON： + +```json +{"ok":true} +``` + +失败时输出： + +```json +{"ok":false,"error":"..."} +``` + +## 当前脚本 + +| 脚本 | 用途 | +| --- | --- | +| `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 | + +## 平台内优先级 + +在 ZK Data Agent 里，仍优先使用已注册的 `data_agent_*` 工具，因为它们带有平台级 review 状态门禁和 session 输出路由。 + +当运行环境没有这些工具，或其他 Agent 迁移使用本 skill 时，再使用 `scripts/` 里的 portable scripts。 diff --git a/skills/product-data/SKILL.md b/skills/product-data/SKILL.md index 5fc0a74..dce614d 100644 --- a/skills/product-data/SKILL.md +++ b/skills/product-data/SKILL.md @@ -3,13 +3,46 @@ name: product-data description: 从产品/标签定义、手写边界规则或示例 query 中提取标签边界，并生成可 review 的数据计划、dataset draft text 和 canonical metadata records。 when_to_use: 当用户提供产品定义、标签规则、路由边界文档、示例 query、手写标签边界，并希望生成训练/评测/专项数据时使用。 aliases: definition-data, label-data -allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_user_question, data_agent_load_input_sources, data_agent_render_source_context, data_agent_extract_case_evidence, data_agent_prepare_generation_goal, data_agent_confirm_generation_goal, data_agent_prepare_generation_plan, data_agent_show_generation_plan, data_agent_update_generation_plan, data_agent_confirm_generation_plan, data_agent_normalize_dataset_draft, data_agent_validate_dataset_records, data_agent_export_dataset_records +allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_user_question, python_exec, data_agent_load_input_sources, data_agent_render_source_context, data_agent_extract_case_evidence, data_agent_prepare_generation_goal, data_agent_confirm_generation_goal, data_agent_prepare_generation_plan, data_agent_show_generation_plan, data_agent_update_generation_plan, data_agent_confirm_generation_plan, data_agent_normalize_dataset_draft, data_agent_validate_dataset_records, data_agent_export_dataset_records --- 使用这个 skill 作为“产品/标签定义/手写规则/示例 query -> 输入文本化 -> generation goal 草案 -> 人工 review -> 生成计划 review -> dataset draft text -> canonical metadata records”的统一入口。 本 skill 内置数据记录生成协议。其他数据开发 skill 如果需要生成或整理标准数据，可以复用这里的“交互门禁、dataset draft text v1、canonical record v1”规则。 +## 能力组织 + +本 skill 按 portable skill contract 组织，能力本体在 skill 目录内： + +```text +skills/product-data/ + SKILL.md + tools.yaml + requirements.txt + knowledge/ + dataset_draft_v1.md + canonical_record_v1.md + portable_skill_contract.md + schemas/ + scripts/ + normalize_dataset_draft.py + validate_dataset_records.py + export_dataset_records.py +``` + +在 ZK Data Agent 平台里，优先使用已注册的 `data_agent_*` 工具，因为它们带有平台级 review 状态、确认门禁和会话 output 路由。 + +迁移到其他 Agent 或没有平台注册能力时，可以直接执行 `scripts/` 下的 portable scripts。脚本支持 `--input input.json` 或 stdin JSON，stdout 只输出一个 JSON 对象。`tools.yaml` 描述了这些脚本如何被平台注册为工具；如果平台不支持 `tools.yaml`，不影响脚本独立使用。 + +脚本能力和平台注册的关系是： + +```text +脚本能力是本体 +平台注册是适配层 +``` + +portable scripts 不包含人类 review 状态机。生成式数据仍必须先由 Agent 按本 skill 的 review 门禁获得用户确认，再调用脚本做格式转换、校验和导出。 + ## 输入假设 用户可能会提供： @@ -272,7 +305,7 @@ canonical records 落盘必须使用 `data_agent_export_dataset_records`，默 现在已经有代码级 goal review 门禁：不要手写 goal 后直接进入 plan，必须先调用 `data_agent_prepare_generation_goal`，并在用户确认后用 `data_agent_confirm_generation_goal` 取得 `confirmed_goal_id`。 -## 计划中的专用工具 +## 输入文本化平台工具 当前前链路只保留三个工具。不要再假设有 `data_agent_load_definition_source`、`data_agent_extract_target_definitions` 这类更细工具。 @@ -288,6 +321,72 @@ canonical records 落盘必须使用 `data_agent_export_dataset_records`，默 从 badcase、评测表、走查表里识别 query、上下文、预期标签、模型预测、类型和备注。字段不明确时，它会返回 `required_questions`，此时必须向用户确认字段含义。 +## Portable Scripts + +当平台注册工具不可用时，使用 `python_exec` 执行这些脚本。不要用 `bash` 执行 Python。 + +### `product_data_normalize_dataset_draft` + +脚本： + +```text +skills/product-data/scripts/normalize_dataset_draft.py +``` + +输入示例： + +```json +{ + "draft_text": "# dataset_label: 地图餐饮边界\n\n### case: 找附近美食\n用户: 附近有什么好吃的\ntarget: Agent(tag=\"餐饮服务\")", + "batch_id": "aabbccdd", + "source_type": "generated" +} +``` + +用途：把 dataset draft text v1 转成 canonical records。 + +### `product_data_validate_dataset_records` + +脚本： + +```text +skills/product-data/scripts/validate_dataset_records.py +``` + +输入支持： + +```json +{"records": []} +``` + +或： + +```json +{"records_path": "output/records.jsonl"} +``` + +用途：校验 canonical records。 + +### `product_data_export_dataset_records` + +脚本： + +```text +skills/product-data/scripts/export_dataset_records.py +``` + +输入支持： + +```json +{ + "records": [], + "output_dir": "output", + "output_format": "jsonl" +} +``` + +用途：导出紧凑 JSONL/JSON。默认文件名是 `records.jsonl` 或 `records.json`。 + ## 当前可用工具 - `read_file`：读取用户提供的产品定义、标签定义、样例 query 文件。 @@ -296,6 +395,7 @@ canonical records 落盘必须使用 `data_agent_export_dataset_records`，默 - `grep_search`：在项目中搜索已有标签定义、历史数据样例或相关文档。 - `glob_search`：按路径模式查找定义文件、样例文件或历史产物。 - `ask_user_question`：需要用户明确选择或补充关键信息时使用；如果不可用，就用普通回复提问并停止。 +- `python_exec`：仅在已注册 `data_agent_*` 工具不可用、或需要迁移验证 portable scripts 时使用；优先执行 `skills/product-data/scripts/` 下脚本。 - `data_agent_load_input_sources`：读取用户给的目录或文件，把 docx/xlsx/pdf 等输入统一抽成段落、表格和 source refs。 - `data_agent_render_source_context`：把结构化输入渲染成大模型可读文本，支持 `max_chars`、表格行数和关键词过滤，用于后续模型语义抽取。 - `data_agent_extract_case_evidence`：从 badcase/评测/走查表中抽取 query、预期标签、模型预测、上下文和备注；字段歧义会返回需要确认的问题。 @@ -315,5 +415,6 @@ canonical records 落盘必须使用 `data_agent_export_dataset_records`，默 - 如果 `ask_user_question` 不可用，使用普通回复向用户提问并停止，不要自己替用户确认。 - canonical records 通过校验前，不要生成最终导出格式。 - canonical records 需要落盘时，必须用 `data_agent_export_dataset_records`；不要自己拼接 JSON/JSONL；不要创建数据集子目录或自定义 records 文件名。 +- 如果使用 portable scripts，它们只负责格式转换、校验和导出，不替代 `generation_goal` / `generation_plan` 的用户 review。 - 除非用户明确要求，否则不要把“修改标签定义”和“生成数据”混在一起做。 - 不要因为用户说“生成一些数据”就跳过边界总结和 generation plan review。 diff --git a/skills/product-data/knowledge/canonical_record_v1.md b/skills/product-data/knowledge/canonical_record_v1.md new file mode 100644 index 0000000..b43f6b1 --- /dev/null +++ b/skills/product-data/knowledge/canonical_record_v1.md @@ -0,0 +1,46 @@ +# Canonical Record v1 + +canonical record 是产品数据生成链路的中间元数据格式。它不是最终训练、评测或展示格式。 + +## 示例 + +```json +{ + "record_id": "gen_aabbccdd_000001", + "source": { + "type": "generated", + "request_id": "aabbccdd", + "timestamp": 1755567930500 + }, + "turn": { + "query": "本轮 query", + "timestamp": 1755567930500 + }, + "prev_session": [ + { + "query": "前一轮 query", + "tts": "前一轮 tts", + "timestamp": 1755567870500 + } + ], + "context": {}, + "label": { + "dataset_label": "数据集或专题名称", + "target": "Agent(tag=\"xxx\")", + "target_type": "agent" + }, + "meta": { + "case_name": "case名称", + "notes": "" + } +} +``` + +## 校验重点 + +- `turn.query` 必须有值。 +- 当前轮 `turn` 不包含 `tts`。 +- `prev_session` 最多 10 轮，且按时间从早到晚。 +- 相邻轮时间间隔超过 5 分钟时给 warning。 +- `label.target` 必须有值。 +- `target_type` 只能是 `agent`、`function` 或 `unknown`。 diff --git a/skills/product-data/knowledge/dataset_draft_v1.md b/skills/product-data/knowledge/dataset_draft_v1.md new file mode 100644 index 0000000..b23ea57 --- /dev/null +++ b/skills/product-data/knowledge/dataset_draft_v1.md @@ -0,0 +1,34 @@ +# Dataset Draft Text v1 + +模型生成数据样本时，先写便于人工 review 的 draft text，不直接写 canonical JSON。 + +## 格式 + +```text +# dataset_label: 数据集或专题名称 + +### case: case名称 +用户: 本轮 query +target: Agent(tag="xxx") +notes: 可选，说明覆盖的问题或边界 + +### case: 多轮 case 名称 +用户: 前一轮 query +小爱: 前一轮 tts +用户: 本轮 query +target: Agent(tag="xxx") +notes: 可选，说明覆盖的问题或边界 +``` + +## 规则 + +- 每条数据用一个 `### case:` 开始。 +- `用户:` 表示用户 query。 +- `小爱:` 表示小爱回复 tts。 +- 最后一个 `用户:` 是本轮 query。 +- `target:` 是本轮 query 对应的监督标签，必须存在。 +- 单轮数据只需要写一行 `用户:`，然后写 `target:`。 +- 多轮数据需要按照时间顺序写多组 `用户:` / `小爱:`。 +- 多轮数据的最后一轮只写 `用户:` 和 `target:`，不要写最后一轮 `小爱:`。 +- 如果 `target` 无法确定，不要编造，必须向用户确认。 +- 不要手写 `record_id`、`request_id`、`timestamp`、`context`。 diff --git a/skills/product-data/knowledge/portable_skill_contract.md b/skills/product-data/knowledge/portable_skill_contract.md new file mode 100644 index 0000000..f2d7226 --- /dev/null +++ b/skills/product-data/knowledge/portable_skill_contract.md @@ -0,0 +1,65 @@ +# Portable Skill Contract + +本约定用于让 skill 能在不同 Agent 平台之间迁移。 + +## 核心原则 + +能力归 skill，运行适配归平台。 + +- 脚本必须能独立执行。 +- 平台工具注册只做薄封装，不承载业务逻辑。 +- 没有工具注册能力的 Agent，可以阅读 `SKILL.md` 并直接执行脚本。 +- 脚本输入输出都使用 JSON，避免自然语言解析。 + +## 脚本输入 + +脚本必须支持： + +```bash +python scripts/.py --input input.json +``` + +也必须支持从 stdin 读取 JSON： + +```bash +cat input.json | python scripts/.py +``` + +## 脚本输出 + +stdout 只输出一个 JSON 对象。 + +成功： + +```json +{ + "ok": true, + "result": {} +} +``` + +失败： + +```json +{ + "ok": false, + "error": "错误原因" +} +``` + +## 输出路径 + +默认写入当前会话 `output/`，或调用方明确传入的输出目录。 + +脚本不得写项目源码目录、随机临时目录或用户未确认的外部路径。 + +## 平台注册 + +`tools.yaml` 是平台适配层： + +- `name` 是平台注册后的工具名。 +- `script` 指向真正执行能力。 +- `input_schema` / `output_schema` 供平台生成工具 schema。 +- `fallback` 告诉无注册环境如何手动调用。 + +如果平台不支持 `tools.yaml`，不影响脚本独立使用。 diff --git a/skills/product-data/requirements.txt b/skills/product-data/requirements.txt new file mode 100644 index 0000000..fe6714f --- /dev/null +++ b/skills/product-data/requirements.txt @@ -0,0 +1 @@ +# 当前 portable scripts 只依赖 Python 标准库。 diff --git a/skills/product-data/schemas/export_dataset_records.input.schema.json b/skills/product-data/schemas/export_dataset_records.input.schema.json new file mode 100644 index 0000000..8b7fa06 --- /dev/null +++ b/skills/product-data/schemas/export_dataset_records.input.schema.json @@ -0,0 +1,20 @@ +{ + "type": "object", + "properties": { + "records": {"type": "array"}, + "records_path": {"type": "string"}, + "output_path": {"type": "string"}, + "output_dir": {"type": "string"}, + "output_format": { + "type": "string", + "enum": ["jsonl", "json"], + "default": "jsonl" + }, + "require_validation_ok": {"type": "boolean", "default": true}, + "overwrite": {"type": "boolean", "default": true} + }, + "anyOf": [ + {"required": ["records"]}, + {"required": ["records_path"]} + ] +} diff --git a/skills/product-data/schemas/export_dataset_records.output.schema.json b/skills/product-data/schemas/export_dataset_records.output.schema.json new file mode 100644 index 0000000..7ddb6f3 --- /dev/null +++ b/skills/product-data/schemas/export_dataset_records.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"] +} diff --git a/skills/product-data/schemas/normalize_dataset_draft.input.schema.json b/skills/product-data/schemas/normalize_dataset_draft.input.schema.json new file mode 100644 index 0000000..c127ab2 --- /dev/null +++ b/skills/product-data/schemas/normalize_dataset_draft.input.schema.json @@ -0,0 +1,20 @@ +{ + "type": "object", + "properties": { + "draft_text": {"type": "string"}, + "draft_path": {"type": "string"}, + "batch_id": {"type": "string", "default": "aabbccdd"}, + "source_type": { + "type": "string", + "enum": ["generated", "online", "manual", "mixed"], + "default": "generated" + }, + "base_timestamp": {"type": "integer"}, + "timestamp_step_ms": {"type": "integer", "minimum": 1, "default": 60000}, + "default_request_id": {"type": "string", "default": "aabbccdd"} + }, + "anyOf": [ + {"required": ["draft_text"]}, + {"required": ["draft_path"]} + ] +} diff --git a/skills/product-data/schemas/normalize_dataset_draft.output.schema.json b/skills/product-data/schemas/normalize_dataset_draft.output.schema.json new file mode 100644 index 0000000..0f453b8 --- /dev/null +++ b/skills/product-data/schemas/normalize_dataset_draft.output.schema.json @@ -0,0 +1,10 @@ +{ + "type": "object", + "properties": { + "ok": {"type": "boolean"}, + "records": {"type": "array"}, + "warnings": {"type": "array", "items": {"type": "string"}}, + "error": {"type": "string"} + }, + "required": ["ok"] +} diff --git a/skills/product-data/schemas/validate_dataset_records.input.schema.json b/skills/product-data/schemas/validate_dataset_records.input.schema.json new file mode 100644 index 0000000..b93494b --- /dev/null +++ b/skills/product-data/schemas/validate_dataset_records.input.schema.json @@ -0,0 +1,11 @@ +{ + "type": "object", + "properties": { + "records": {"type": "array"}, + "records_path": {"type": "string"} + }, + "anyOf": [ + {"required": ["records"]}, + {"required": ["records_path"]} + ] +} diff --git a/skills/product-data/schemas/validate_dataset_records.output.schema.json b/skills/product-data/schemas/validate_dataset_records.output.schema.json new file mode 100644 index 0000000..43348ea --- /dev/null +++ b/skills/product-data/schemas/validate_dataset_records.output.schema.json @@ -0,0 +1,12 @@ +{ + "type": "object", + "properties": { + "ok": {"type": "boolean"}, + "error_count": {"type": "integer"}, + "warning_count": {"type": "integer"}, + "errors": {"type": "array"}, + "warnings": {"type": "array"}, + "error": {"type": "string"} + }, + "required": ["ok"] +} diff --git a/skills/product-data/scripts/export_dataset_records.py b/skills/product-data/scripts/export_dataset_records.py new file mode 100644 index 0000000..0d970b4 --- /dev/null +++ b/skills/product-data/scripts/export_dataset_records.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +from product_data_portable import ( + default_output_path, + emit_error, + emit_success, + export_dataset_records, + load_json_payload, + load_records, +) + + +def main() -> int: + try: + payload = load_json_payload() + records = load_records(payload) + output_format = str(payload.get("output_format") or "jsonl") + result = export_dataset_records( + records, + output_path=default_output_path(payload), + output_format=output_format, + 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()) diff --git a/skills/product-data/scripts/normalize_dataset_draft.py b/skills/product-data/scripts/normalize_dataset_draft.py new file mode 100644 index 0000000..5d163f7 --- /dev/null +++ b/skills/product-data/scripts/normalize_dataset_draft.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from pathlib import Path + +from product_data_portable import ( + DEFAULT_REQUEST_ID, + DEFAULT_TIMESTAMP_STEP_MS, + emit_error, + emit_success, + load_json_payload, + normalize_dataset_draft, +) + + +def main() -> int: + try: + payload = load_json_payload() + draft_text = str(payload.get("draft_text") or "") + if not draft_text and payload.get("draft_path"): + draft_text = Path(str(payload["draft_path"])).expanduser().read_text(encoding="utf-8") + result = normalize_dataset_draft( + draft_text, + batch_id=str(payload.get("batch_id") or DEFAULT_REQUEST_ID), + source_type=str(payload.get("source_type") or "generated"), + base_timestamp=payload.get("base_timestamp"), + timestamp_step_ms=int(payload.get("timestamp_step_ms") or DEFAULT_TIMESTAMP_STEP_MS), + default_request_id=str(payload.get("default_request_id") or DEFAULT_REQUEST_ID), + ) + emit_success(result) + return 0 + except Exception as exc: # noqa: BLE001 - CLI 需要把错误稳定转成 JSON + emit_error(exc) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/product-data/scripts/product_data_portable.py b/skills/product-data/scripts/product_data_portable.py new file mode 100644 index 0000000..3311ba5 --- /dev/null +++ b/skills/product-data/scripts/product_data_portable.py @@ -0,0 +1,457 @@ +from __future__ import annotations + +"""product-data 可迁移脚本的共享逻辑。 + +这里不依赖 ZK Data Agent 的运行时，便于把整个 skill 复制到其他 Agent +或普通 Python 环境中使用。平台注册工具时，也应该只薄封装这些脚本。 +""" + +import argparse +import json +import re +import sys +import time +from pathlib import Path +from typing import Any + + +DEFAULT_REQUEST_ID = "aabbccdd" +DEFAULT_TIMESTAMP_STEP_MS = 60_000 + + +class ProductDataError(ValueError): + """数据格式或参数错误。""" + + +def load_json_payload(argv: list[str] | None = None) -> dict[str, Any]: + parser = argparse.ArgumentParser(add_help=True) + parser.add_argument("--input", "-i", help="JSON 参数文件；不传则从 stdin 读取") + args = parser.parse_args(argv) + + if args.input: + text = Path(args.input).expanduser().read_text(encoding="utf-8") + else: + text = sys.stdin.read() + if not text.strip(): + raise ProductDataError("input JSON is required") + payload = json.loads(text) + if not isinstance(payload, dict): + raise ProductDataError("input JSON must be an object") + return payload + + +def emit_success(payload: dict[str, Any]) -> None: + result = {"ok": True, **payload} + print(json.dumps(result, ensure_ascii=False, separators=(",", ":"))) + + +def emit_error(exc: BaseException) -> None: + print( + json.dumps( + {"ok": False, "error": str(exc)}, + ensure_ascii=False, + separators=(",", ":"), + ) + ) + + +def load_records(payload: dict[str, Any]) -> list[dict[str, Any]]: + if "records" in payload: + records = payload["records"] + elif "records_path" in payload: + records = read_records_file(str(payload["records_path"])) + else: + raise ProductDataError("records or records_path is required") + if isinstance(records, str): + records = json.loads(records) + if not isinstance(records, list): + raise ProductDataError("records must be an array") + normalized: list[dict[str, Any]] = [] + for index, item in enumerate(records): + if not isinstance(item, dict): + raise ProductDataError(f"records[{index}] must be an object") + normalized.append(item) + return normalized + + +def read_records_file(path: str) -> list[dict[str, Any]]: + file_path = Path(path).expanduser() + text = file_path.read_text(encoding="utf-8") + if file_path.suffix.lower() == ".jsonl": + records: list[dict[str, Any]] = [] + for line_number, line in enumerate(text.splitlines(), start=1): + if not line.strip(): + continue + item = json.loads(line) + if not isinstance(item, dict): + raise ProductDataError(f"{path}:{line_number} must be a JSON object") + records.append(item) + return records + decoded = json.loads(text) + if not isinstance(decoded, list): + raise ProductDataError(f"{path} must contain a JSON array") + return decoded + + +def normalize_dataset_draft( + draft_text: str, + *, + batch_id: str = DEFAULT_REQUEST_ID, + source_type: str = "generated", + base_timestamp: int | None = None, + timestamp_step_ms: int = DEFAULT_TIMESTAMP_STEP_MS, + default_request_id: str = DEFAULT_REQUEST_ID, +) -> dict[str, Any]: + if source_type not in {"generated", "online", "manual", "mixed"}: + raise ProductDataError("source_type is invalid") + if not draft_text.strip(): + raise ProductDataError("draft_text must be non-empty") + if timestamp_step_ms <= 0: + raise ProductDataError("timestamp_step_ms must be greater than 0") + + base = int(time.time() * 1000) if base_timestamp is None else int(base_timestamp) + parsed = parse_draft_text(draft_text) + records: list[dict[str, Any]] = [] + warnings: list[str] = [] + + for index, case in enumerate(parsed["cases"], start=1): + record, case_warnings = case_to_record( + case, + global_dataset_label=str(parsed.get("dataset_label") or ""), + index=index, + batch_id=batch_id, + source_type=source_type, + base_timestamp=base, + timestamp_step_ms=timestamp_step_ms, + default_request_id=default_request_id, + ) + records.append(record) + warnings.extend(case_warnings) + + return {"records": records, "warnings": warnings} + + +def parse_draft_text(draft_text: str) -> dict[str, Any]: + dataset_label = "" + cases: list[dict[str, Any]] = [] + current: dict[str, Any] | None = None + + for raw_line in draft_text.splitlines(): + line = raw_line.strip() + if not line: + continue + dataset_match = re.match(r"^#\s*dataset_label\s*[:：]\s*(.+)$", line, re.I) + if dataset_match: + dataset_label = dataset_match.group(1).strip() + continue + case_match = re.match(r"^#{3,}\s*case\s*[:：]\s*(.+)$", line, re.I) + if case_match: + current = {"case_name": case_match.group(1).strip(), "turns": []} + cases.append(current) + continue + if current is None: + continue + + field_match = re.match( + r"^(用户|小爱|target|notes|dataset_label|request_id|timestamp)\s*[:：]\s*(.*)$", + line, + re.I, + ) + if not field_match: + continue + key = field_match.group(1).lower() + value = field_match.group(2).strip() + if key == "用户": + current["turns"].append({"role": "user", "text": value}) + elif key == "小爱": + current["turns"].append({"role": "assistant", "text": value}) + else: + current[key] = value + + if not cases: + raise ProductDataError('draft_text must contain at least one "### case:" block') + return {"dataset_label": dataset_label, "cases": cases} + + +def case_to_record( + case: dict[str, Any], + *, + global_dataset_label: str, + index: int, + batch_id: str, + source_type: str, + base_timestamp: int, + timestamp_step_ms: int, + default_request_id: str, +) -> tuple[dict[str, Any], list[str]]: + warnings: list[str] = [] + turns = case.get("turns") + if not isinstance(turns, list): + raise ProductDataError(f"case {index} has invalid turns") + + user_indexes = [ + turn_index + for turn_index, turn in enumerate(turns) + if isinstance(turn, dict) and turn.get("role") == "user" + ] + if not user_indexes: + raise ProductDataError(f"case {index} must contain at least one 用户 line") + final_user_index = user_indexes[-1] + current_query = str(turns[final_user_index].get("text") or "").strip() + if not current_query: + raise ProductDataError(f"case {index} current 用户 line must be non-empty") + + target = str(case.get("target") or "").strip() + if not target: + raise ProductDataError(f"case {index} target is required") + + prev_session = build_prev_session(turns[:final_user_index], case_index=index) + if len(prev_session) > 10: + warnings.append(f"case {index} prev_session has more than 10 turns; keeping the latest 10") + prev_session = prev_session[-10:] + + request_id = str(case.get("request_id") or default_request_id).strip() + turn_timestamp = optional_int(case.get("timestamp")) + if turn_timestamp is None: + turn_timestamp = base_timestamp + (index - 1) * timestamp_step_ms * 20 + first_prev_timestamp = turn_timestamp - len(prev_session) * timestamp_step_ms + for prev_index, item in enumerate(prev_session): + item["timestamp"] = first_prev_timestamp + prev_index * timestamp_step_ms + + dataset_label = str(case.get("dataset_label") or global_dataset_label or "").strip() + record = { + "record_id": record_id( + source_type=source_type, + batch_id=batch_id, + request_id=request_id, + index=index, + ), + "source": { + "type": source_type, + "request_id": request_id, + "timestamp": turn_timestamp, + }, + "turn": { + "query": current_query, + "timestamp": turn_timestamp, + }, + "prev_session": prev_session, + "context": {}, + "label": { + "dataset_label": dataset_label, + "target": target, + "target_type": target_type(target), + }, + "meta": { + "case_name": str(case.get("case_name") or "").strip(), + "notes": str(case.get("notes") or "").strip(), + }, + } + return record, warnings + + +def build_prev_session(turns: list[dict[str, Any]], *, case_index: int) -> list[dict[str, Any]]: + prev_session: list[dict[str, Any]] = [] + index = 0 + while index < len(turns): + turn = turns[index] + if not isinstance(turn, dict) or turn.get("role") != "user": + raise ProductDataError(f"case {case_index} prev_session must start with 用户 before 小爱") + if index + 1 >= len(turns) or turns[index + 1].get("role") != "assistant": + raise ProductDataError(f"case {case_index} each previous 用户 line must be followed by 小爱") + query = str(turn.get("text") or "").strip() + tts = str(turns[index + 1].get("text") or "").strip() + if not query or not tts: + raise ProductDataError(f"case {case_index} previous 用户/小爱 lines must be non-empty") + prev_session.append({"query": query, "tts": tts}) + index += 2 + return prev_session + + +def validate_dataset_records(records: list[dict[str, Any]]) -> dict[str, Any]: + errors: list[dict[str, str]] = [] + warnings: list[dict[str, str]] = [] + seen_ids: set[str] = set() + + for index, record in enumerate(records): + prefix = f"records[{index}]" + if not isinstance(record, dict): + errors.append(issue(prefix, "record must be an object")) + continue + + record_id_value = record.get("record_id") + if not non_empty_str(record_id_value): + errors.append(issue(f"{prefix}.record_id", "record_id is required")) + elif record_id_value in seen_ids: + errors.append(issue(f"{prefix}.record_id", "record_id must be unique")) + else: + seen_ids.add(str(record_id_value)) + + source = record.get("source") + if not isinstance(source, dict): + errors.append(issue(f"{prefix}.source", "source is required")) + else: + if source.get("type") not in {"generated", "online", "manual", "mixed"}: + errors.append(issue(f"{prefix}.source.type", "source.type is invalid")) + if not non_empty_str(source.get("request_id")): + errors.append(issue(f"{prefix}.source.request_id", "source.request_id is required")) + if not int_like(source.get("timestamp")): + errors.append(issue(f"{prefix}.source.timestamp", "source.timestamp must be an integer")) + + turn = record.get("turn") + if not isinstance(turn, dict): + errors.append(issue(f"{prefix}.turn", "turn is required")) + turn_timestamp = None + else: + if not non_empty_str(turn.get("query")): + errors.append(issue(f"{prefix}.turn.query", "turn.query is required")) + if "tts" in turn: + errors.append(issue(f"{prefix}.turn.tts", "current turn must not contain tts")) + turn_timestamp = turn.get("timestamp") + if not int_like(turn_timestamp): + errors.append(issue(f"{prefix}.turn.timestamp", "turn.timestamp must be an integer")) + + prev_session = record.get("prev_session") + prev_timestamps: list[int] = [] + if not isinstance(prev_session, list): + errors.append(issue(f"{prefix}.prev_session", "prev_session must be a list")) + else: + if len(prev_session) > 10: + errors.append(issue(f"{prefix}.prev_session", "prev_session must contain at most 10 turns")) + for prev_index, item in enumerate(prev_session): + item_prefix = f"{prefix}.prev_session[{prev_index}]" + if not isinstance(item, dict): + errors.append(issue(item_prefix, "prev_session item must be an object")) + continue + if not non_empty_str(item.get("query")): + errors.append(issue(f"{item_prefix}.query", "query is required")) + if "tts" not in item or not isinstance(item.get("tts"), str): + errors.append(issue(f"{item_prefix}.tts", "tts must be a string")) + timestamp = item.get("timestamp") + if not int_like(timestamp): + errors.append(issue(f"{item_prefix}.timestamp", "timestamp must be an integer")) + else: + prev_timestamps.append(int(timestamp)) + + if prev_timestamps != sorted(prev_timestamps): + errors.append(issue(f"{prefix}.prev_session", "timestamps must be sorted from early to late")) + if int_like(turn_timestamp): + all_timestamps = [*prev_timestamps, int(turn_timestamp)] + for left, right in zip(all_timestamps, all_timestamps[1:]): + if right <= left: + errors.append(issue(f"{prefix}.timestamp", "turn timestamps must be strictly increasing")) + break + if right - left > 300_000: + warnings.append(issue(f"{prefix}.timestamp", "adjacent turns are more than 5 minutes apart")) + + if not isinstance(record.get("context"), dict): + errors.append(issue(f"{prefix}.context", "context must be an object")) + + label = record.get("label") + if not isinstance(label, dict): + errors.append(issue(f"{prefix}.label", "label is required")) + else: + if not non_empty_str(label.get("dataset_label")): + warnings.append(issue(f"{prefix}.label.dataset_label", "dataset_label is empty")) + if not non_empty_str(label.get("target")): + errors.append(issue(f"{prefix}.label.target", "label.target is required")) + if label.get("target_type") not in {"agent", "function", "unknown"}: + errors.append(issue(f"{prefix}.label.target_type", "label.target_type is invalid")) + + return { + "ok": not errors, + "error_count": len(errors), + "warning_count": len(warnings), + "errors": errors, + "warnings": warnings, + } + + +def export_dataset_records( + records: list[dict[str, Any]], + *, + output_path: str, + output_format: str = "jsonl", + require_validation_ok: bool = True, + overwrite: bool = True, +) -> dict[str, Any]: + if output_format not in {"jsonl", "json"}: + raise ProductDataError("output_format must be jsonl or json") + 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) + + if output_format == "jsonl": + content = "".join( + json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n" + for record in records + ) + else: + content = json.dumps(records, ensure_ascii=False, separators=(",", ":")) + "\n" + path.write_text(content, encoding="utf-8") + return { + "output_path": str(path), + "output_format": output_format, + "record_count": len(records), + "bytes_written": len(content.encode("utf-8")), + "validation": validation, + } + + +def default_output_path(payload: dict[str, Any]) -> str: + output_format = str(payload.get("output_format") or "jsonl") + filename = "records.json" if output_format == "json" else "records.jsonl" + 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() / filename) + return str(Path("output") / filename) + + +def target_type(target: str) -> str: + if re.match(r"^Agent\s*\(\s*tag\s*=", target): + return "agent" + if target: + return "function" + return "unknown" + + +def record_id(*, source_type: str, batch_id: str, request_id: str, index: int) -> str: + if source_type == "generated": + return f"gen_{safe_token(batch_id)}_{index:06d}" + if source_type == "online": + return f"online_{safe_token(request_id)}_{index:06d}" + return f"{safe_token(source_type)}_{safe_token(batch_id)}_{index:06d}" + + +def safe_token(value: str) -> str: + token = re.sub(r"[^A-Za-z0-9_.-]+", "_", value.strip()) + return token.strip("_") or "unknown" + + +def optional_int(value: Any) -> int | None: + if value is None or value == "": + return None + if isinstance(value, bool): + raise ProductDataError("timestamp must be an integer") + try: + return int(value) + except (TypeError, ValueError) as exc: + raise ProductDataError("timestamp must be an integer") from exc + + +def non_empty_str(value: Any) -> bool: + return isinstance(value, str) and bool(value.strip()) + + +def int_like(value: Any) -> bool: + return isinstance(value, int) and not isinstance(value, bool) + + +def issue(path: str, message: str) -> dict[str, str]: + return {"path": path, "message": message} diff --git a/skills/product-data/scripts/validate_dataset_records.py b/skills/product-data/scripts/validate_dataset_records.py new file mode 100644 index 0000000..70944ca --- /dev/null +++ b/skills/product-data/scripts/validate_dataset_records.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from product_data_portable import emit_error, emit_success, load_json_payload, load_records, validate_dataset_records + + +def main() -> int: + try: + payload = load_json_payload() + records = load_records(payload) + emit_success(validate_dataset_records(records)) + return 0 + except Exception as exc: # noqa: BLE001 - CLI 需要把错误稳定转成 JSON + emit_error(exc) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/product-data/tools.yaml b/skills/product-data/tools.yaml new file mode 100644 index 0000000..8e17632 --- /dev/null +++ b/skills/product-data/tools.yaml @@ -0,0 +1,32 @@ +version: 1 +skill: product-data +description: Product-data portable tool manifest. Scripts are the source of capability; platform tool registration is only an adapter. +runtime: + language: python + min_python: "3.10" + dependencies: [] +tools: + - name: product_data_normalize_dataset_draft + description: Parse dataset draft text v1 into canonical records with generated ids, timestamps, source metadata, context and labels. + script: scripts/normalize_dataset_draft.py + input_schema: schemas/normalize_dataset_draft.input.schema.json + output_schema: schemas/normalize_dataset_draft.output.schema.json + fallback: + command: python skills/product-data/scripts/normalize_dataset_draft.py --input + stdin: true + - name: product_data_validate_dataset_records + description: Validate canonical product-data records for required fields, labels, timestamp order and prompt stitching constraints. + script: scripts/validate_dataset_records.py + input_schema: schemas/validate_dataset_records.input.schema.json + output_schema: schemas/validate_dataset_records.output.schema.json + fallback: + command: python skills/product-data/scripts/validate_dataset_records.py --input + 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. + 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 + stdin: true