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

Add training and planning eval exports

Browse Source
This commit is contained in:
wuyang6
2026-05-11 14:50:00 +08:00
parent 8cca3793e0
commit be731caed7
15 changed files with 1074 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/product-data/README.md
+2
View File
@@ -45,6 +45,8 @@ cat input.json | python skills/product-data/scripts/<tool>.py
| `scripts/validate_dataset_records.py` | 校验 canonical records |
| `scripts/export_dataset_records.py` | 导出紧凑 JSONL/JSON，并默认生成同目录 `records.csv` 表格 |
| `scripts/export_dataset_table.py` | 已有 canonical records 时，只补生成同事流转表格 |
| `scripts/export_training_jsonl.py` | 把 canonical records 转成训练 JSONL |
| `scripts/export_planning_eval_csv.py` | 把 canonical records 转成含 `newPrompt` 的评测 CSV |
## 平台内优先级
skills/product-data/SKILL.md
+55 -3
View File
@@ -3,7 +3,7 @@ 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, 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
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, data_agent_export_training_jsonl, data_agent_export_planning_eval_csv
---
使用这个 skill 作为“产品/标签定义/手写规则/示例 query -> 输入文本化 -> generation goal 草案 -> 人工 review -> 生成计划 review -> dataset draft text -> canonical metadata records”的统一入口。
@@ -29,6 +29,8 @@ skills/product-data/
validate_dataset_records.py
export_dataset_records.py
export_dataset_table.py
export_training_jsonl.py
export_planning_eval_csv.py
```
在 ZK Data Agent 平台里，优先使用已注册的 `data_agent_*` 工具，因为它们带有平台级 review 状态、确认门禁和会话 output 路由。
@@ -148,7 +150,7 @@ review 展示必须简短清晰，不要重复解释工具和流程。每次 rev
## 必须遵守的数据生成协议
不要直接生成 canonical JSON，不要直接导出最终训练/评测格式，不要绕过人类 review。
不要直接生成 canonical JSON，不要在 canonical records 校验通过前导出训练/评测格式，不要绕过人类 review。
如果需要生成数据，必须按顺序执行：
@@ -166,7 +168,9 @@ review 展示必须简短清晰，不要重复解释工具和流程。每次 rev
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，并同时生成同目录 `records.csv` 表格，不要用 `write_file` 手写 JSON 或 CSV。
15. 本阶段默认只推进到 canonical metadata records；除非用户另行要求，不做最终训练/评测格式导出。
15. 如果用户已经明确要训练数据，调用 `data_agent_export_training_jsonl`，优先传 `records_path`，输出固定为 `output/training.jsonl`。
16. 如果用户已经明确要评测 planningPrompt 数据，调用 `data_agent_export_planning_eval_csv`，优先传 `records_path`，输出固定为 `output/eval_planning.csv`。
17. 如果用户只说“生成数据”但没有说明下游用途，生成 canonical records 和 `records.csv` 后，简短询问用户是否还需要导出训练 jsonl 或评测 csv；不要自己默认生成全部最终格式。
生成阶段要避免一次性把大量数据塞进工具参数：
@@ -218,6 +222,13 @@ notes: 可选，说明覆盖的问题或边界
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 records 转换，不要让模型手写：
- 训练数据：调用 `data_agent_export_training_jsonl`，输出 `training.jsonl`，每行包含 `system`、`instruction`、`output`。
- 评测数据：调用 `data_agent_export_planning_eval_csv`，输出 `eval_planning.csv`，字段为 `request_id,newPrompt,query,类别真实标签,code标签,complex`。
- 两个派生格式默认复用相同 prompt 构造逻辑：`[知识注入]`、`[系统状态]`、`[对话历史]`、`[当前query]`、`[function]`。
- 对话历史默认最多取 5 轮，且相邻轮间隔不超过 5 分钟；`context` 默认注入 `location` 和 `rag` 两个字段。
当前 canonical record v1 工作格式：
```json
@@ -427,6 +438,44 @@ skills/product-data/scripts/export_dataset_table.py
用途：已有 canonical records 时，只补生成同事流转表格 `records.csv`，不改写元数据文件。主流程仍优先用 `product_data_export_dataset_records`，因为它会一次性导出元数据和表格。
### `product_data_export_training_jsonl`
脚本：
```text
skills/product-data/scripts/export_training_jsonl.py
```
输入支持：
```json
{
"records_path": "output/records.jsonl",
"output_dir": "output"
}
```
用途：把 canonical records 转成训练 JSONL，默认文件名 `training.jsonl`。
### `product_data_export_planning_eval_csv`
脚本：
```text
skills/product-data/scripts/export_planning_eval_csv.py
```
输入支持：
```json
{
"records_path": "output/records.jsonl",
"output_dir": "output"
}
```
用途：把 canonical records 转成含 `newPrompt` 的评测 CSV，默认文件名 `eval_planning.csv`。
## 当前可用工具
- `read_file`：读取用户提供的产品定义、标签定义、样例 query 文件。
@@ -448,6 +497,8 @@ skills/product-data/scripts/export_dataset_table.py
- `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 或表格文件。
- `data_agent_export_training_jsonl`：把 canonical records 转成训练 JSONL；支持 `records` 或 `records_path`；默认写 `output/training.jsonl`。
- `data_agent_export_planning_eval_csv`：把 canonical records 转成评测 CSV；支持 `records` 或 `records_path`；默认写 `output/eval_planning.csv`。
## 约束
@@ -455,6 +506,7 @@ skills/product-data/scripts/export_dataset_table.py
- 如果 `ask_user_question` 不可用，使用普通回复向用户提问并停止，不要自己替用户确认。
- canonical records 通过校验前，不要生成最终导出格式。
- canonical records 需要落盘时，必须用 `data_agent_export_dataset_records`；不要自己拼接 JSON/JSONL/CSV；不要创建数据集子目录或自定义 records 文件名。
- 训练/评测派生格式必须从 canonical records 通过工具导出；不要让模型自己拼 prompt、手写 jsonl 或 csv。
- 如果使用 portable scripts，它们只负责格式转换、校验和导出，不替代 `generation_goal` / `generation_plan` 的用户 review。
- 除非用户明确要求，否则不要把“修改标签定义”和“生成数据”混在一起做。
- 不要因为用户说“生成一些数据”就跳过边界总结和 generation plan review。
skills/product-data/knowledge/canonical_record_v1.md
+9
View File
@@ -55,3 +55,12 @@ canonical record 是产品数据生成链路的中间元数据格式。它不是
`records.csv` 中的 `prev_session` 和 `context` 是紧凑 JSON 字符串；`prev_session` 中的 `timestamp` 按历史表格习惯输出为字符串。
如果已有 `records.jsonl` 或 `records.json`，只需要补表格，可以执行 portable `export_dataset_table.py` 或等价工具；默认仍输出到同目录的 `records.csv`。
## 派生格式
训练和评测格式都从 canonical records 转换，不由模型手写：
- `training.jsonl`：每行 `{"system": "...", "instruction": "...", "output": "..."}`，由 `export_training_jsonl.py` 或 `data_agent_export_training_jsonl` 生成。
- `eval_planning.csv`：字段为 `request_id,newPrompt,query,类别真实标签,code标签,complex`，由 `export_planning_eval_csv.py` 或 `data_agent_export_planning_eval_csv` 生成。
`instruction` 和 `newPrompt` 使用同一套 prompt 模板，默认包含 `[知识注入]`、`[系统状态]`、`[对话历史]`、`[当前query]`、`[function]`。历史轮次默认最多取 5 轮，且相邻时间间隔不超过 5 分钟。
skills/product-data/schemas/export_planning_eval_csv.input.schema.json
+25
View File
@@ -0,0 +1,25 @@
{
"type": "object",
"properties": {
"records": {
"type": "array",
"description": "Canonical records. Provide exactly one of records or records_path."
},
"records_path": {
"type": "string",
"description": "Path to records JSON/JSONL. Provide exactly one of records or records_path."
},
"output_path": {"type": "string"},
"output_dir": {"type": "string"},
"session_num": {"type": "integer", "minimum": 1, "default": 5},
"session_time_minutes": {"type": "integer", "minimum": 1, "default": 5},
"context_fields": {
"type": "array",
"items": {"type": "string"},
"default": ["location", "rag"]
},
"complex_default": {"type": "boolean", "default": false},
"require_validation_ok": {"type": "boolean", "default": true},
"overwrite": {"type": "boolean", "default": true}
}
}
skills/product-data/schemas/export_planning_eval_csv.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/export_training_jsonl.input.schema.json
+28
View File
@@ -0,0 +1,28 @@
{
"type": "object",
"properties": {
"records": {
"type": "array",
"description": "Canonical records. Provide exactly one of records or records_path."
},
"records_path": {
"type": "string",
"description": "Path to records JSON/JSONL. Provide exactly one of records or records_path."
},
"output_path": {"type": "string"},
"output_dir": {"type": "string"},
"session_num": {"type": "integer", "minimum": 1, "default": 5},
"session_time_minutes": {"type": "integer", "minimum": 1, "default": 5},
"context_fields": {
"type": "array",
"items": {"type": "string"},
"default": ["location", "rag"]
},
"system_prompt": {
"type": "string",
"default": "你是小爱同学，中文智能语音助手。"
},
"require_validation_ok": {"type": "boolean", "default": true},
"overwrite": {"type": "boolean", "default": true}
}
}
skills/product-data/schemas/export_training_jsonl.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/scripts/export_planning_eval_csv.py
+35
View File
@@ -0,0 +1,35 @@
from __future__ import annotations
from product_data_portable import (
default_planning_eval_output_path,
emit_error,
emit_success,
export_planning_eval_csv,
load_json_payload,
load_records,
)
def main() -> int:
try:
payload = load_json_payload()
records = load_records(payload)
result = export_planning_eval_csv(
records,
output_path=default_planning_eval_output_path(payload),
session_num=int(payload.get("session_num", 5)),
session_time_minutes=int(payload.get("session_time_minutes", 5)),
context_fields=payload.get("context_fields"),
complex_default=bool(payload.get("complex_default", False)),
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/export_training_jsonl.py
+36
View File
@@ -0,0 +1,36 @@
from __future__ import annotations
from product_data_portable import (
DEFAULT_SYSTEM_PROMPT,
default_training_output_path,
emit_error,
emit_success,
export_training_jsonl,
load_json_payload,
load_records,
)
def main() -> int:
try:
payload = load_json_payload()
records = load_records(payload)
result = export_training_jsonl(
records,
output_path=default_training_output_path(payload),
session_num=int(payload.get("session_num", 5)),
session_time_minutes=int(payload.get("session_time_minutes", 5)),
context_fields=payload.get("context_fields"),
system_prompt=str(payload.get("system_prompt") or DEFAULT_SYSTEM_PROMPT),
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
+292 -1
View File
@@ -19,6 +19,8 @@ from typing import Any
DEFAULT_REQUEST_ID = "aabbccdd"
DEFAULT_TIMESTAMP_STEP_MS = 60_000
DEFAULT_SYSTEM_PROMPT = "你是小爱同学，中文智能语音助手。"
DEFAULT_CONTEXT_FIELDS = ("location", "rag")
class ProductDataError(ValueError):
@@ -205,7 +207,7 @@ def case_to_record(
if not current_query:
raise ProductDataError(f"case {index} current 用户 line must be non-empty")
target = str(case.get("target") or "").strip()
target = canonical_target(str(case.get("target") or "").strip())
if not target:
raise ProductDataError(f"case {index} target is required")
@@ -447,6 +449,89 @@ def export_dataset_table(
}
def export_training_jsonl(
records: list[dict[str, Any]],
*,
output_path: str,
session_num: int = 5,
session_time_minutes: int = 5,
context_fields: list[str] | None = None,
system_prompt: str = DEFAULT_SYSTEM_PROMPT,
require_validation_ok: bool = True,
overwrite: bool = True,
) -> dict[str, Any]:
if session_num <= 0:
raise ProductDataError("session_num must be greater than 0")
if session_time_minutes <= 0:
raise ProductDataError("session_time_minutes must be greater than 0")
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)
fields = normalize_context_fields(context_fields)
content = "".join(
training_jsonl_line(
record,
session_num=session_num,
session_time_minutes=session_time_minutes,
context_fields=fields,
system_prompt=system_prompt,
)
+ "\n"
for record in records
)
path.write_text(content, encoding="utf-8")
return {
"output_path": str(path),
"output_format": "jsonl",
"record_count": len(records),
"bytes_written": len(content.encode("utf-8")),
"validation": validation,
}
def export_planning_eval_csv(
records: list[dict[str, Any]],
*,
output_path: str,
session_num: int = 5,
session_time_minutes: int = 5,
context_fields: list[str] | None = None,
complex_default: bool = False,
require_validation_ok: bool = True,
overwrite: bool = True,
) -> dict[str, Any]:
if session_num <= 0:
raise ProductDataError("session_num must be greater than 0")
if session_time_minutes <= 0:
raise ProductDataError("session_time_minutes must be greater than 0")
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_planning_eval_csv(
records,
session_num=session_num,
session_time_minutes=session_time_minutes,
context_fields=context_fields,
complex_default=complex_default,
)
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")
@@ -456,6 +541,185 @@ def render_dataset_records_table_csv(records: list[dict[str, Any]]) -> str:
return output.getvalue()
def render_planning_eval_csv(
records: list[dict[str, Any]],
*,
session_num: int = 5,
session_time_minutes: int = 5,
context_fields: list[str] | None = None,
complex_default: bool = False,
) -> str:
fields = normalize_context_fields(context_fields)
output = StringIO()
writer = csv.DictWriter(
output,
fieldnames=["request_id", "newPrompt", "query", "类别真实标签", "code标签", "complex"],
lineterminator="\n",
)
writer.writeheader()
for record in records:
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 {}
target = str(label.get("target") or "")
writer.writerow(
{
"request_id": str(source.get("request_id") or ""),
"newPrompt": build_training_instruction(
record,
session_num=session_num,
session_time_minutes=session_time_minutes,
context_fields=fields,
),
"query": str(turn.get("query") or ""),
"类别真实标签": category_label_from_target(target),
"code标签": target,
"complex": "true" if complex_default else "false",
}
)
return output.getvalue()
def build_training_instruction(
record: dict[str, Any],
*,
session_num: int = 5,
session_time_minutes: int = 5,
context_fields: list[str] | None = None,
) -> 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_for_export(turn.get("timestamp"))
if current_ts is None:
source = record.get("source") if isinstance(record.get("source"), dict) else {}
current_ts = optional_int_for_export(source.get("timestamp"))
session, last_tts = training_history(
prev_session,
current_ts=current_ts,
session_num=session_num,
session_time_minutes=session_time_minutes,
)
fields = normalize_context_fields(context_fields)
instruction = "请参考用户的[当前query]、[对话历史]、[知识注入]、[系统状态]识别出[当前query]的[function]结果，[function]是python的code形式。\n"
instruction += "[知识注入]\n"
instruction += f"{context_prompt_block(context, fields)}\n"
instruction += "[系统状态]\n"
instruction += "{}\n"
instruction += "[对话历史]\n"
if session:
history_parts = [f"用户: {session_query}" for session_query in session]
if last_tts is not None:
history_parts.append(f"小爱: {last_tts}")
instruction += "\n".join(history_parts)
instruction += "\n"
instruction += "[当前query]\n"
instruction += f"用户: {query}\n"
instruction += "[function]\n"
return instruction
def training_jsonl_line(
record: dict[str, Any],
*,
session_num: int,
session_time_minutes: int,
context_fields: list[str],
system_prompt: str,
) -> str:
label = record.get("label") if isinstance(record.get("label"), dict) else {}
payload = {
"system": system_prompt,
"instruction": build_training_instruction(
record,
session_num=session_num,
session_time_minutes=session_time_minutes,
context_fields=context_fields,
),
"output": str(label.get("target") or ""),
}
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
def training_history(
prev_session: list[Any],
*,
current_ts: int | None,
session_num: int,
session_time_minutes: int,
) -> tuple[list[str], str | None]:
if current_ts is None:
return [], None
session_time_ms = session_time_minutes * 60 * 1000
valid_items: list[tuple[int, str, str]] = []
for item in prev_session:
if not isinstance(item, dict):
continue
ts = optional_int_for_export(item.get("timestamp"))
query = str(item.get("query") or "")
tts = str(item.get("tts") or "")
if ts is not None and ts > 0 and query:
valid_items.append((ts, query, tts))
valid_items.sort(key=lambda item: item[0])
session: list[str] = []
last_tts: str | None = None
last_ts = current_ts
count = 0
for ts, query, tts in reversed(valid_items):
if last_ts - ts <= session_time_ms and last_ts >= ts:
session.append(query)
if count == 0:
last_tts = tts
last_ts = ts
count += 1
if count >= session_num:
break
else:
break
session.reverse()
return session, last_tts
def context_prompt_block(context: dict[str, Any], fields: list[str]) -> str:
lines = ["{"]
for index, field in enumerate(fields):
comma = "," if index < len(fields) - 1 else ""
value = str(context.get(field, ""))
lines.append(f'"{field}": {json.dumps(value, ensure_ascii=False)}{comma}')
lines.append("}")
return "\n".join(lines)
def normalize_context_fields(context_fields: list[str] | None) -> list[str]:
if context_fields is not None and not isinstance(context_fields, list):
raise ProductDataError("context_fields must be an array of strings")
fields = context_fields or list(DEFAULT_CONTEXT_FIELDS)
normalized = [field.strip() for field in fields if isinstance(field, str) and field.strip()]
return normalized or list(DEFAULT_CONTEXT_FIELDS)
def optional_int_for_export(value: Any) -> int | None:
if isinstance(value, bool) or value is None or value == "":
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def category_label_from_target(target: str) -> str:
agent_match = re.match(r'''^Agent\s*\(\s*tag\s*=\s*["']([^"']+)["']\s*\)$''', target.strip())
if agent_match:
return agent_match.group(1)
function_names = re.findall(r"\b([A-Za-z_][A-Za-z0-9_]*)\s*\(", target)
if function_names:
return function_names[-1]
return target
def dataset_table_columns() -> list[str]:
return [
"request_id",
@@ -524,6 +788,26 @@ def default_table_output_path(payload: dict[str, Any]) -> str:
return str(Path("output") / "records.csv")
def default_training_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() / "training.jsonl")
if isinstance(payload.get("records_path"), str) and payload["records_path"].strip():
return str(Path(payload["records_path"]).expanduser().with_name("training.jsonl"))
return str(Path("output") / "training.jsonl")
def default_planning_eval_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() / "eval_planning.csv")
if isinstance(payload.get("records_path"), str) and payload["records_path"].strip():
return str(Path(payload["records_path"]).expanduser().with_name("eval_planning.csv"))
return str(Path("output") / "eval_planning.csv")
def target_type(target: str) -> str:
if re.match(r"^Agent\s*\(\s*tag\s*=", target):
return "agent"
@@ -532,6 +816,13 @@ def target_type(target: str) -> str:
return "unknown"
def canonical_target(target: str) -> str:
agent_match = re.match(r'''^Agent\s*\(\s*tag\s*=\s*["']([^"']+)["']\s*\)$''', target)
if agent_match:
return f'Agent(tag="{agent_match.group(1)}")'
return target
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}"
skills/product-data/tools.yaml
+16
View File
@@ -38,3 +38,19 @@ tools:
fallback:
command: python skills/product-data/scripts/export_dataset_table.py --input <input.json>
stdin: true
- name: product_data_export_training_jsonl
description: Convert canonical records into training JSONL lines with system, instruction and output fields.
script: scripts/export_training_jsonl.py
input_schema: schemas/export_training_jsonl.input.schema.json
output_schema: schemas/export_training_jsonl.output.schema.json
fallback:
command: python skills/product-data/scripts/export_training_jsonl.py --input <input.json>
stdin: true
- name: product_data_export_planning_eval_csv
description: Convert canonical records into evaluation CSV with request_id, newPrompt, query, 类别真实标签, code标签 and complex columns.
script: scripts/export_planning_eval_csv.py
input_schema: schemas/export_planning_eval_csv.input.schema.json
output_schema: schemas/export_planning_eval_csv.output.schema.json
fallback:
command: python skills/product-data/scripts/export_planning_eval_csv.py --input <input.json>
stdin: true
src/agent_tool_specs/data_agent.py
+98
View File
@@ -547,4 +547,102 @@ def build_data_agent_tools(handlers: Mapping[str, ToolHandler]) -> list[AgentToo
},
handler=resolve_handler(handlers, 'data_agent_export_dataset_records', 'data-agent'),
),
AgentTool(
name='data_agent_export_training_jsonl',
description=(
'Convert canonical data-agent records into training JSONL. Each line has system, instruction, '
'and output. The instruction contains 知识注入, 系统状态, 对话历史, 当前query, and function sections.'
),
parameters={
'type': 'object',
'properties': {
'records': {
'oneOf': [
{'type': 'array'},
{'type': 'string'},
],
'description': 'Canonical records as an array, or a JSON string containing the array. Provide exactly one of records or records_path.',
},
'records_path': {
'type': 'string',
'description': 'Path to records JSON/JSONL. Prefer this when records are large. Provide exactly one of records or records_path.',
},
'output_path': {
'type': 'string',
'description': 'Optional requested path. Runtime normalizes to current session output/training.jsonl.',
},
'session_num': {
'type': 'integer',
'minimum': 1,
'description': 'Max history queries to include. Defaults to 5.',
},
'session_time_minutes': {
'type': 'integer',
'minimum': 1,
'description': 'Max adjacent history gap in minutes. Defaults to 5.',
},
'context_fields': {
'type': 'array',
'items': {'type': 'string'},
'description': 'Context fields to inject. Defaults to ["location", "rag"].',
},
'system_prompt': {
'type': 'string',
'description': 'Defaults to 你是小爱同学，中文智能语音助手。',
},
'require_validation_ok': {'type': 'boolean'},
'overwrite': {'type': 'boolean'},
},
},
handler=resolve_handler(handlers, 'data_agent_export_training_jsonl', 'data-agent'),
),
AgentTool(
name='data_agent_export_planning_eval_csv',
description=(
'Convert canonical data-agent records into evaluation CSV with columns: request_id, newPrompt, '
'query, 类别真实标签, code标签, complex. newPrompt uses the same prompt body as training instruction.'
),
parameters={
'type': 'object',
'properties': {
'records': {
'oneOf': [
{'type': 'array'},
{'type': 'string'},
],
'description': 'Canonical records as an array, or a JSON string containing the array. Provide exactly one of records or records_path.',
},
'records_path': {
'type': 'string',
'description': 'Path to records JSON/JSONL. Prefer this when records are large. Provide exactly one of records or records_path.',
},
'output_path': {
'type': 'string',
'description': 'Optional requested path. Runtime normalizes to current session output/eval_planning.csv.',
},
'session_num': {
'type': 'integer',
'minimum': 1,
'description': 'Max history queries to include. Defaults to 5.',
},
'session_time_minutes': {
'type': 'integer',
'minimum': 1,
'description': 'Max adjacent history gap in minutes. Defaults to 5.',
},
'context_fields': {
'type': 'array',
'items': {'type': 'string'},
'description': 'Context fields to inject. Defaults to ["location", "rag"].',
},
'complex_default': {
'type': 'boolean',
'description': 'Default complex column value. Defaults to false.',
},
'require_validation_ok': {'type': 'boolean'},
'overwrite': {'type': 'boolean'},
},
},
handler=resolve_handler(handlers, 'data_agent_export_planning_eval_csv', 'data-agent'),
),
]
src/agent_tools.py
+68
View File
@@ -34,6 +34,8 @@ from .data_agent_records import (
confirm_generation_goal,
confirm_generation_plan,
export_dataset_records,
export_planning_eval_csv,
export_training_jsonl,
get_generation_plan,
normalize_dataset_draft,
prepare_generation_goal,
@@ -1049,6 +1051,8 @@ def default_tool_registry() -> dict[str, AgentTool]:
'data_agent_normalize_dataset_draft': _normalize_dataset_draft_tool,
'data_agent_validate_dataset_records': _validate_dataset_records_tool,
'data_agent_export_dataset_records': _export_dataset_records_tool,
'data_agent_export_training_jsonl': _export_training_jsonl_tool,
'data_agent_export_planning_eval_csv': _export_planning_eval_csv_tool,
}),
]
return {tool.name: tool for tool in tools}
@@ -1550,6 +1554,70 @@ def _export_dataset_records_tool(arguments: dict[str, Any], context: ToolExecuti
return json.dumps(payload, ensure_ascii=False, indent=2)
def _export_training_jsonl_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
require_validation_ok = arguments.get('require_validation_ok', True)
if not isinstance(require_validation_ok, bool):
raise ToolExecutionError('require_validation_ok must be a boolean')
overwrite = arguments.get('overwrite', True)
if not isinstance(overwrite, bool):
raise ToolExecutionError('overwrite must be a boolean')
system_prompt = arguments.get('system_prompt', '你是小爱同学，中文智能语音助手。')
if not isinstance(system_prompt, str):
raise ToolExecutionError('system_prompt must be a string')
try:
records = _records_from_arguments(arguments, context)
payload = export_training_jsonl(
records,
root=str(context.root),
output_path=_data_agent_output_path(
_optional_string(arguments, 'output_path') or 'output/training.jsonl',
context,
canonical_filename='training.jsonl',
),
session_num=_coerce_int(arguments, 'session_num', 5),
session_time_minutes=_coerce_int(arguments, 'session_time_minutes', 5),
context_fields=_optional_string_list(arguments, 'context_fields'),
system_prompt=system_prompt,
require_validation_ok=require_validation_ok,
overwrite=overwrite,
)
except (DataRecordError, json.JSONDecodeError) as exc:
raise ToolExecutionError(str(exc)) from exc
return json.dumps(payload, ensure_ascii=False, indent=2)
def _export_planning_eval_csv_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
require_validation_ok = arguments.get('require_validation_ok', True)
if not isinstance(require_validation_ok, bool):
raise ToolExecutionError('require_validation_ok must be a boolean')
overwrite = arguments.get('overwrite', True)
if not isinstance(overwrite, bool):
raise ToolExecutionError('overwrite must be a boolean')
complex_default = arguments.get('complex_default', False)
if not isinstance(complex_default, bool):
raise ToolExecutionError('complex_default must be a boolean')
try:
records = _records_from_arguments(arguments, context)
payload = export_planning_eval_csv(
records,
root=str(context.root),
output_path=_data_agent_output_path(
_optional_string(arguments, 'output_path') or 'output/eval_planning.csv',
context,
canonical_filename='eval_planning.csv',
),
session_num=_coerce_int(arguments, 'session_num', 5),
session_time_minutes=_coerce_int(arguments, 'session_time_minutes', 5),
context_fields=_optional_string_list(arguments, 'context_fields'),
complex_default=complex_default,
require_validation_ok=require_validation_ok,
overwrite=overwrite,
)
except (DataRecordError, json.JSONDecodeError) as exc:
raise ToolExecutionError(str(exc)) from exc
return json.dumps(payload, ensure_ascii=False, indent=2)
def _draft_text_from_arguments(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
draft_text = arguments.get('draft_text')
draft_path = arguments.get('draft_path')
src/data_agent_records.py
+275
View File
@@ -18,6 +18,8 @@ from typing import Any
DEFAULT_REQUEST_ID = 'aabbccdd'
DEFAULT_TIMESTAMP_STEP_MS = 60_000
DEFAULT_SYSTEM_PROMPT = '你是小爱同学，中文智能语音助手。'
DEFAULT_CONTEXT_FIELDS = ('location', 'rag')
DEFAULT_PLAN_STATE_FILE = '.port_sessions/data_agent_generation_plans.json'
DEFAULT_GOAL_STATE_FILE = '.port_sessions/data_agent_generation_goals.json'
@@ -520,6 +522,100 @@ def export_dataset_records(
return payload
def export_training_jsonl(
records: list[dict[str, Any]],
*,
root: str,
output_path: str,
session_num: int = 5,
session_time_minutes: int = 5,
context_fields: list[str] | None = None,
system_prompt: str = DEFAULT_SYSTEM_PROMPT,
require_validation_ok: bool = True,
overwrite: bool = True,
) -> dict[str, Any]:
"""导出训练集 JSONL：system / instruction / output。"""
if session_num <= 0:
raise DataRecordError('session_num must be greater than 0')
if session_time_minutes <= 0:
raise DataRecordError('session_time_minutes must be greater than 0')
validation = validate_dataset_records(records)
if require_validation_ok and not validation['ok']:
raise DataRecordError(
f'records failed validation with {validation["error_count"]} errors'
)
path = _resolve_output_path(root, output_path)
if path.exists() and not overwrite:
raise DataRecordError(f'output_path already exists: {output_path}')
path.parent.mkdir(parents=True, exist_ok=True)
fields = _normalize_context_fields(context_fields)
content = ''.join(
_training_jsonl_line(
record,
session_num=session_num,
session_time_minutes=session_time_minutes,
context_fields=fields,
system_prompt=system_prompt,
)
+ '\n'
for record in records
)
path.write_text(content, encoding='utf-8')
return {
'output_path': str(path.relative_to(Path(root).resolve())),
'output_format': 'jsonl',
'record_count': len(records),
'bytes_written': len(content.encode('utf-8')),
'validation': validation,
}
def export_planning_eval_csv(
records: list[dict[str, Any]],
*,
root: str,
output_path: str,
session_num: int = 5,
session_time_minutes: int = 5,
context_fields: list[str] | None = None,
complex_default: bool = False,
require_validation_ok: bool = True,
overwrite: bool = True,
) -> dict[str, Any]:
"""导出评测 CSV：request_id / newPrompt / query / 类别真实标签 / code标签 / complex。"""
if session_num <= 0:
raise DataRecordError('session_num must be greater than 0')
if session_time_minutes <= 0:
raise DataRecordError('session_time_minutes must be greater than 0')
validation = validate_dataset_records(records)
if require_validation_ok and not validation['ok']:
raise DataRecordError(
f'records failed validation with {validation["error_count"]} errors'
)
path = _resolve_output_path(root, output_path)
if path.exists() and not overwrite:
raise DataRecordError(f'output_path already exists: {output_path}')
path.parent.mkdir(parents=True, exist_ok=True)
fields = _normalize_context_fields(context_fields)
content = render_planning_eval_csv(
records,
session_num=session_num,
session_time_minutes=session_time_minutes,
context_fields=fields,
complex_default=complex_default,
)
path.write_text(content, encoding='utf-8-sig')
return {
'output_path': str(path.relative_to(Path(root).resolve())),
'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:
"""把 canonical records 转成同事流转用的表格 CSV。"""
@@ -531,6 +627,185 @@ def render_dataset_records_table_csv(records: list[dict[str, Any]]) -> str:
return output.getvalue()
def render_planning_eval_csv(
records: list[dict[str, Any]],
*,
session_num: int = 5,
session_time_minutes: int = 5,
context_fields: list[str] | None = None,
complex_default: bool = False,
) -> str:
"""把 canonical records 转成含 newPrompt 的评测 CSV。"""
fields = _normalize_context_fields(context_fields)
output = StringIO()
writer = csv.DictWriter(
output,
fieldnames=['request_id', 'newPrompt', 'query', '类别真实标签', 'code标签', 'complex'],
lineterminator='\n',
)
writer.writeheader()
for record in records:
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 {}
target = str(label.get('target') or '')
writer.writerow(
{
'request_id': str(source.get('request_id') or ''),
'newPrompt': build_training_instruction(
record,
session_num=session_num,
session_time_minutes=session_time_minutes,
context_fields=fields,
),
'query': str(turn.get('query') or ''),
'类别真实标签': _category_label_from_target(target),
'code标签': target,
'complex': 'true' if complex_default else 'false',
}
)
return output.getvalue()
def build_training_instruction(
record: dict[str, Any],
*,
session_num: int = 5,
session_time_minutes: int = 5,
context_fields: list[str] | None = None,
) -> 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_for_export(turn.get('timestamp'))
if current_ts is None:
source = record.get('source') if isinstance(record.get('source'), dict) else {}
current_ts = _optional_int_for_export(source.get('timestamp'))
session, last_tts = _training_history(
prev_session,
current_ts=current_ts,
session_num=session_num,
session_time_minutes=session_time_minutes,
)
fields = _normalize_context_fields(context_fields)
instruction = '请参考用户的[当前query]、[对话历史]、[知识注入]、[系统状态]识别出[当前query]的[function]结果，[function]是python的code形式。\n'
instruction += '[知识注入]\n'
instruction += f'{_context_prompt_block(context, fields)}\n'
instruction += '[系统状态]\n'
instruction += '{}\n'
instruction += '[对话历史]\n'
if session:
history_parts = [f'用户: {session_query}' for session_query in session]
if last_tts is not None:
history_parts.append(f'小爱: {last_tts}')
instruction += '\n'.join(history_parts)
instruction += '\n'
instruction += '[当前query]\n'
instruction += f'用户: {query}\n'
instruction += '[function]\n'
return instruction
def _training_jsonl_line(
record: dict[str, Any],
*,
session_num: int,
session_time_minutes: int,
context_fields: list[str],
system_prompt: str,
) -> str:
label = record.get('label') if isinstance(record.get('label'), dict) else {}
payload = {
'system': system_prompt,
'instruction': build_training_instruction(
record,
session_num=session_num,
session_time_minutes=session_time_minutes,
context_fields=context_fields,
),
'output': str(label.get('target') or ''),
}
return json.dumps(payload, ensure_ascii=False, separators=(',', ':'))
def _training_history(
prev_session: list[Any],
*,
current_ts: int | None,
session_num: int,
session_time_minutes: int,
) -> tuple[list[str], str | None]:
if current_ts is None:
return [], None
session_time_ms = session_time_minutes * 60 * 1000
valid_items: list[tuple[int, str, str]] = []
for item in prev_session:
if not isinstance(item, dict):
continue
ts = _optional_int_for_export(item.get('timestamp'))
query = str(item.get('query') or '')
tts = str(item.get('tts') or '')
if ts is not None and ts > 0 and query:
valid_items.append((ts, query, tts))
valid_items.sort(key=lambda item: item[0])
session: list[str] = []
last_tts: str | None = None
last_ts = current_ts
count = 0
for ts, query, tts in reversed(valid_items):
if last_ts - ts <= session_time_ms and last_ts >= ts:
session.append(query)
if count == 0:
last_tts = tts
last_ts = ts
count += 1
if count >= session_num:
break
else:
break
session.reverse()
return session, last_tts
def _context_prompt_block(context: dict[str, Any], fields: list[str]) -> str:
lines = ['{']
for index, field in enumerate(fields):
comma = ',' if index < len(fields) - 1 else ''
value = str(context.get(field, ''))
lines.append(f'"{field}": {json.dumps(value, ensure_ascii=False)}{comma}')
lines.append('}')
return '\n'.join(lines)
def _normalize_context_fields(context_fields: list[str] | None) -> list[str]:
fields = context_fields or list(DEFAULT_CONTEXT_FIELDS)
normalized = [field.strip() for field in fields if isinstance(field, str) and field.strip()]
return normalized or list(DEFAULT_CONTEXT_FIELDS)
def _optional_int_for_export(value: Any) -> int | None:
if isinstance(value, bool) or value is None or value == '':
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def _category_label_from_target(target: str) -> str:
agent_match = re.match(r'''^Agent\s*\(\s*tag\s*=\s*["']([^"']+)["']\s*\)$''', target.strip())
if agent_match:
return agent_match.group(1)
function_names = re.findall(r'\b([A-Za-z_][A-Za-z0-9_]*)\s*\(', target)
if function_names:
return function_names[-1]
return target
def _dataset_table_columns() -> list[str]:
return [
'request_id',
tests/test_data_agent_records.py
+109
View File
@@ -12,11 +12,14 @@ from src.data_agent_records import (
confirm_generation_goal,
confirm_generation_plan,
export_dataset_records,
export_planning_eval_csv,
export_training_jsonl,
get_generation_plan,
normalize_dataset_draft,
prepare_generation_goal,
prepare_generation_plan,
render_dataset_records_table_csv,
render_planning_eval_csv,
update_generation_plan,
validate_dataset_records,
)
@@ -179,6 +182,90 @@ target: Agent(tag='地图导航')
self.assertEqual(prev_session[0]['timestamp'], '1755567870500')
self.assertEqual(rows[0]['function'], 'Agent(tag="地图导航")')
def test_export_training_jsonl_uses_prompt_template_and_history(self) -> None:
records = normalize_dataset_draft(
'''
# dataset_label: 地图导航边界数据
### case: 多轮顺路停车场
用户: 查一下附近停车场
小爱: 找到了附近停车场
用户: 帮我找个顺路的
target: Agent(tag='地图导航')
'''.strip(),
batch_id='demo',
base_timestamp=1_755_567_930_500,
timestamp_step_ms=60_000,
)['records']
records[0]['context'] = {'location': '北京', 'rag': '地图服务可用', 'other': '忽略'}
with tempfile.TemporaryDirectory() as tmp_dir:
payload = export_training_jsonl(
records,
root=tmp_dir,
output_path='output/training.jsonl',
)
output_path = Path(tmp_dir) / payload['output_path']
line = json.loads(output_path.read_text(encoding='utf-8').splitlines()[0])
self.assertEqual(payload['output_format'], 'jsonl')
self.assertEqual(line['system'], '你是小爱同学，中文智能语音助手。')
self.assertEqual(line['output'], 'Agent(tag="地图导航")')
self.assertIn('[知识注入]\n{\n"location": "北京",\n"rag": "地图服务可用"\n}', line['instruction'])
self.assertIn('用户: 查一下附近停车场\n小爱: 找到了附近停车场', line['instruction'])
self.assertIn('[当前query]\n用户: 帮我找个顺路的', line['instruction'])
def test_render_planning_eval_csv_uses_expected_columns(self) -> None:
records = normalize_dataset_draft(
'''
# dataset_label: 地图导航边界数据
### case: 多轮顺路停车场
用户: 查一下附近停车场
小爱: 找到了附近停车场
用户: 帮我找个顺路的
target: Agent(tag='地图导航')
'''.strip(),
batch_id='demo',
base_timestamp=1_755_567_930_500,
)['records']
rows = list(csv.DictReader(render_planning_eval_csv(records).splitlines()))
self.assertEqual(
list(rows[0].keys()),
['request_id', 'newPrompt', 'query', '类别真实标签', 'code标签', 'complex'],
)
self.assertEqual(rows[0]['request_id'], 'aabbccdd')
self.assertEqual(rows[0]['query'], '帮我找个顺路的')
self.assertEqual(rows[0]['类别真实标签'], '地图导航')
self.assertEqual(rows[0]['code标签'], 'Agent(tag="地图导航")')
self.assertEqual(rows[0]['complex'], 'false')
self.assertIn('[function]', rows[0]['newPrompt'])
def test_export_planning_eval_csv_writes_file(self) -> None:
records = normalize_dataset_draft(
'''
# dataset_label: 时间工具数据
### case: 几点
用户: 现在几点
target: CalendarQA(type="TIME")
'''.strip(),
batch_id='demo',
base_timestamp=1_755_567_930_500,
)['records']
with tempfile.TemporaryDirectory() as tmp_dir:
payload = export_planning_eval_csv(
records,
root=tmp_dir,
output_path='output/eval_planning.csv',
complex_default=True,
)
output_path = Path(tmp_dir) / payload['output_path']
rows = list(csv.DictReader(output_path.read_text(encoding='utf-8-sig').splitlines()))
self.assertEqual(payload['output_format'], 'csv')
self.assertEqual(rows[0]['类别真实标签'], 'CalendarQA')
self.assertEqual(rows[0]['code标签'], 'CalendarQA(type="TIME")')
self.assertEqual(rows[0]['complex'], 'true')
def test_normalize_online_draft_uses_real_request_metadata(self) -> None:
records = normalize_dataset_draft(
'''
@@ -641,11 +728,33 @@ target: Agent(tag='地图导航')
self.assertTrue(export_result.ok, export_result.content)
export_payload = json.loads(export_result.content)
table_exists = (root / export_payload['table_output_path']).exists()
training_result = execute_tool(
default_tool_registry(),
'data_agent_export_training_jsonl',
{'records_path': str(records_path)},
context,
)
self.assertTrue(training_result.ok, training_result.content)
training_payload = json.loads(training_result.content)
training_exists = (root / training_payload['output_path']).exists()
eval_result = execute_tool(
default_tool_registry(),
'data_agent_export_planning_eval_csv',
{'records_path': str(records_path)},
context,
)
self.assertTrue(eval_result.ok, eval_result.content)
eval_payload = json.loads(eval_result.content)
eval_exists = (root / eval_payload['output_path']).exists()
self.assertTrue(validate_result.ok, validate_result.content)
self.assertTrue(json.loads(validate_result.content)['ok'])
self.assertEqual(export_payload['record_count'], 1)
self.assertTrue(table_exists)
self.assertEqual(training_payload['output_path'], '.port_sessions/data_agent_output/training.jsonl')
self.assertTrue(training_exists)
self.assertEqual(eval_payload['output_path'], '.port_sessions/data_agent_output/eval_planning.csv')
self.assertTrue(eval_exists)
def test_tool_schemas_avoid_top_level_composition_keywords(self) -> None:
# 部分 Bedrock/Anthropic 兼容后端不接受顶层 anyOf/oneOf/allOf。