wuyang/zk-data-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make product-data skill portable

Browse Source
This commit is contained in:
wuyang6
2026-05-11 10:58:35 +08:00
parent d7744f026c
commit ca2f6f36fc
17 changed files with 964 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/product-data/README.md
+52
View File
@@ -0,0 +1,52 @@
# product-data
`product-data` 是产品/标签定义到标准数据记录的可迁移 skill 包。
设计原则：
- `SKILL.md` 描述流程、review 门禁和 Agent 使用方式。
- `knowledge/` 保存数据格式、生成协议和可维护知识。
- `scripts/` 保存可独立执行的脚本能力。
- `tools.yaml` 描述脚本如何被平台注册成工具；没有注册能力的 Agent 也可以直接调用脚本。
## 可迁移工具约定
脚本能力是本体，平台注册是适配层。
每个脚本都支持两种输入：
```bash
python skills/product-data/scripts/<tool>.py --input input.json
```
或：
```bash
cat input.json | python skills/product-data/scripts/<tool>.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。
skills/product-data/SKILL.md
+103 -2
View File
@@ -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。
skills/product-data/knowledge/canonical_record_v1.md
+46
View File
@@ -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`。
skills/product-data/knowledge/dataset_draft_v1.md
+34
View File
@@ -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`。
skills/product-data/knowledge/portable_skill_contract.md
+65
View File
@@ -0,0 +1,65 @@
# Portable Skill Contract
本约定用于让 skill 能在不同 Agent 平台之间迁移。
## 核心原则
能力归 skill，运行适配归平台。
- 脚本必须能独立执行。
- 平台工具注册只做薄封装，不承载业务逻辑。
- 没有工具注册能力的 Agent，可以阅读 `SKILL.md` 并直接执行脚本。
- 脚本输入输出都使用 JSON，避免自然语言解析。
## 脚本输入
脚本必须支持：
```bash
python scripts/<name>.py --input input.json
```
也必须支持从 stdin 读取 JSON：
```bash
cat input.json | python scripts/<name>.py
```
## 脚本输出
stdout 只输出一个 JSON 对象。
成功：
```json
{
"ok": true,
"result": {}
}
```
失败：
```json
{
"ok": false,
"error": "错误原因"
}
```
## 输出路径
默认写入当前会话 `output/`，或调用方明确传入的输出目录。
脚本不得写项目源码目录、随机临时目录或用户未确认的外部路径。
## 平台注册
`tools.yaml` 是平台适配层：
- `name` 是平台注册后的工具名。
- `script` 指向真正执行能力。
- `input_schema` / `output_schema` 供平台生成工具 schema。
- `fallback` 告诉无注册环境如何手动调用。
如果平台不支持 `tools.yaml`，不影响脚本独立使用。
skills/product-data/requirements.txt
+1
View File
@@ -0,0 +1 @@
# 当前 portable scripts 只依赖 Python 标准库。
skills/product-data/schemas/export_dataset_records.input.schema.json
+20
View File
@@ -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"]}
]
}
skills/product-data/schemas/export_dataset_records.output.schema.json
+13
View File
@@ -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/schemas/normalize_dataset_draft.input.schema.json
+20
View File
@@ -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"]}
]
}
skills/product-data/schemas/normalize_dataset_draft.output.schema.json
+10
View File
@@ -0,0 +1,10 @@
{
"type": "object",
"properties": {
"ok": {"type": "boolean"},
"records": {"type": "array"},
"warnings": {"type": "array", "items": {"type": "string"}},
"error": {"type": "string"}
},
"required": ["ok"]
}
skills/product-data/schemas/validate_dataset_records.input.schema.json
+11
View File
@@ -0,0 +1,11 @@
{
"type": "object",
"properties": {
"records": {"type": "array"},
"records_path": {"type": "string"}
},
"anyOf": [
{"required": ["records"]},
{"required": ["records_path"]}
]
}
skills/product-data/schemas/validate_dataset_records.output.schema.json
+12
View File
@@ -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"]
}
skills/product-data/scripts/export_dataset_records.py
+33
View File
@@ -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())
skills/product-data/scripts/normalize_dataset_draft.py
+37
View File
@@ -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())
skills/product-data/scripts/product_data_portable.py
+457
View File
@@ -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}
skills/product-data/scripts/validate_dataset_records.py
+18
View File
@@ -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())
skills/product-data/tools.yaml
+32
View File
@@ -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 <input.json>
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 <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.
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