From be731caed76884c7c592430850a45f19cf85e630 Mon Sep 17 00:00:00 2001 From: wuyang6 Date: Mon, 11 May 2026 14:50:00 +0800 Subject: [PATCH] Add training and planning eval exports --- skills/product-data/README.md | 2 + skills/product-data/SKILL.md | 58 +++- .../knowledge/canonical_record_v1.md | 9 + ...export_planning_eval_csv.input.schema.json | 25 ++ ...xport_planning_eval_csv.output.schema.json | 13 + .../export_training_jsonl.input.schema.json | 28 ++ .../export_training_jsonl.output.schema.json | 13 + .../scripts/export_planning_eval_csv.py | 35 +++ .../scripts/export_training_jsonl.py | 36 +++ .../scripts/product_data_portable.py | 293 +++++++++++++++++- skills/product-data/tools.yaml | 16 + src/agent_tool_specs/data_agent.py | 98 ++++++ src/agent_tools.py | 68 ++++ src/data_agent_records.py | 275 ++++++++++++++++ tests/test_data_agent_records.py | 109 +++++++ 15 files changed, 1074 insertions(+), 4 deletions(-) create mode 100644 skills/product-data/schemas/export_planning_eval_csv.input.schema.json create mode 100644 skills/product-data/schemas/export_planning_eval_csv.output.schema.json create mode 100644 skills/product-data/schemas/export_training_jsonl.input.schema.json create mode 100644 skills/product-data/schemas/export_training_jsonl.output.schema.json create mode 100644 skills/product-data/scripts/export_planning_eval_csv.py create mode 100644 skills/product-data/scripts/export_training_jsonl.py diff --git a/skills/product-data/README.md b/skills/product-data/README.md index 2ee960d..fe7d506 100644 --- a/skills/product-data/README.md +++ b/skills/product-data/README.md @@ -45,6 +45,8 @@ cat input.json | python skills/product-data/scripts/.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 | ## 平台内优先级 diff --git a/skills/product-data/SKILL.md b/skills/product-data/SKILL.md index 0f48579..f7ba5fb 100644 --- a/skills/product-data/SKILL.md +++ b/skills/product-data/SKILL.md @@ -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。 diff --git a/skills/product-data/knowledge/canonical_record_v1.md b/skills/product-data/knowledge/canonical_record_v1.md index 041f9f7..b38014d 100644 --- a/skills/product-data/knowledge/canonical_record_v1.md +++ b/skills/product-data/knowledge/canonical_record_v1.md @@ -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 分钟。 diff --git a/skills/product-data/schemas/export_planning_eval_csv.input.schema.json b/skills/product-data/schemas/export_planning_eval_csv.input.schema.json new file mode 100644 index 0000000..bd793e4 --- /dev/null +++ b/skills/product-data/schemas/export_planning_eval_csv.input.schema.json @@ -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} + } +} diff --git a/skills/product-data/schemas/export_planning_eval_csv.output.schema.json b/skills/product-data/schemas/export_planning_eval_csv.output.schema.json new file mode 100644 index 0000000..7ddb6f3 --- /dev/null +++ b/skills/product-data/schemas/export_planning_eval_csv.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/export_training_jsonl.input.schema.json b/skills/product-data/schemas/export_training_jsonl.input.schema.json new file mode 100644 index 0000000..0dc73fe --- /dev/null +++ b/skills/product-data/schemas/export_training_jsonl.input.schema.json @@ -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} + } +} diff --git a/skills/product-data/schemas/export_training_jsonl.output.schema.json b/skills/product-data/schemas/export_training_jsonl.output.schema.json new file mode 100644 index 0000000..7ddb6f3 --- /dev/null +++ b/skills/product-data/schemas/export_training_jsonl.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/scripts/export_planning_eval_csv.py b/skills/product-data/scripts/export_planning_eval_csv.py new file mode 100644 index 0000000..df1f3a7 --- /dev/null +++ b/skills/product-data/scripts/export_planning_eval_csv.py @@ -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()) diff --git a/skills/product-data/scripts/export_training_jsonl.py b/skills/product-data/scripts/export_training_jsonl.py new file mode 100644 index 0000000..2c3fdf4 --- /dev/null +++ b/skills/product-data/scripts/export_training_jsonl.py @@ -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()) diff --git a/skills/product-data/scripts/product_data_portable.py b/skills/product-data/scripts/product_data_portable.py index b5a1cf7..26b4944 100644 --- a/skills/product-data/scripts/product_data_portable.py +++ b/skills/product-data/scripts/product_data_portable.py @@ -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}" diff --git a/skills/product-data/tools.yaml b/skills/product-data/tools.yaml index ff458f5..3707081 100644 --- a/skills/product-data/tools.yaml +++ b/skills/product-data/tools.yaml @@ -38,3 +38,19 @@ tools: fallback: command: python skills/product-data/scripts/export_dataset_table.py --input 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 + 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 + stdin: true diff --git a/src/agent_tool_specs/data_agent.py b/src/agent_tool_specs/data_agent.py index dd316cb..d3545e6 100644 --- a/src/agent_tool_specs/data_agent.py +++ b/src/agent_tool_specs/data_agent.py @@ -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'), + ), ] diff --git a/src/agent_tools.py b/src/agent_tools.py index 450f187..14abb2d 100644 --- a/src/agent_tools.py +++ b/src/agent_tools.py @@ -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') diff --git a/src/data_agent_records.py b/src/data_agent_records.py index 48c8369..d6b6c42 100644 --- a/src/data_agent_records.py +++ b/src/data_agent_records.py @@ -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', diff --git a/tests/test_data_agent_records.py b/tests/test_data_agent_records.py index ad73f6c..d9b9c2f 100644 --- a/tests/test_data_agent_records.py +++ b/tests/test_data_agent_records.py @@ -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。