From 5bdc0ffb0122bc5206e560e854aca474889f26f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=AD=A6=E9=98=B3?= Date: Thu, 30 Apr 2026 15:11:37 +0800 Subject: [PATCH] Add data agent input and export workflow --- .gitignore | 1 + docs/data_agent_skill_map.md | 9 +- src/agent_tools.py | 369 ++++++++++- src/data_agent_inputs.py | 615 ++++++++++++++++++ src/data_agent_records.py | 258 ++++++++ .../bundled/data-record-format/SKILL.md | 181 ------ .../SKILL.md | 315 +++++++-- tests/test_data_agent_inputs.py | 142 ++++ tests/test_data_agent_records.py | 134 ++++ 9 files changed, 1787 insertions(+), 237 deletions(-) create mode 100644 src/data_agent_inputs.py delete mode 100644 src/skills/bundled/data-record-format/SKILL.md create mode 100644 tests/test_data_agent_inputs.py diff --git a/.gitignore b/.gitignore index 53c246e..56a2445 100644 --- a/.gitignore +++ b/.gitignore @@ -18,6 +18,7 @@ archive/ .claude/ .claude.json .port_sessions/ +tasks/ # Environment files .env diff --git a/docs/data_agent_skill_map.md b/docs/data_agent_skill_map.md index b08481f..93f2534 100644 --- a/docs/data_agent_skill_map.md +++ b/docs/data_agent_skill_map.md @@ -6,24 +6,19 @@ | Skill | 主要用途 | 当前状态 | |---|---|---| -| `data-record-format` | 讨论和起草 canonical record 规则、样例记录、格式相关 open questions | 草案；canonical record v1 仍待确定 | | `eval-error-data-generation` | 分析评测错误，并规划针对性训练/评测数据生成 | 草案；数据工具仍待实现 | -| `product-definition-data-generation` | 从产品或标签定义文档生成边界感知的数据计划和候选记录 | 草案；定义抽取工具仍待实现 | +| `product-definition-data-generation` | 从产品/标签定义、手写规则或示例 query 生成边界感知的数据计划、dataset draft text 和 canonical records | 草案；已合并数据格式协议，定义抽取工具仍待实现 | | `online-badcase-mining` | 分析 badcase，并起草可迭代的线上挖掘策略 | 草案；线上挖掘工具仍待实现 | | `tool-smoke-test` | 验证目录化 skill 加载和基础工具 trace | 可用 smoke test | ## 示例触发 Query -```text -Use the data-record-format skill. 帮我 review 这个样例文件：test_data/data_agent/sample.jsonl -``` - ```text Use the eval-error-data-generation skill. 输入文件：tasks/debug/context/eval_errors.csv。请先起草错误分析和数据生成计划。 ``` ```text -Use the product-definition-data-generation skill. 输入文件：tasks/debug/context/product_definition.md。请总结标签边界并起草数据生成计划。 +Use the product-definition-data-generation skill. 输入文件：tasks/debug/context/product_definition.md。请总结标签边界并起草数据生成计划，确认后生成 dataset draft text 并转成 canonical records。 ``` ```text diff --git a/src/agent_tools.py b/src/agent_tools.py index 3265c27..f0e6b8a 100644 --- a/src/agent_tools.py +++ b/src/agent_tools.py @@ -17,14 +17,24 @@ from typing import TYPE_CHECKING, Any, Callable, Iterator, Union from .agent_types import AgentPermissions, AgentRuntimeConfig, ToolExecutionResult from .data_agent_records import ( DataRecordError, + confirm_generation_goal, confirm_generation_plan, + export_dataset_records, get_generation_plan, normalize_dataset_draft, + prepare_generation_goal, prepare_generation_plan, records_from_tool_argument, update_generation_plan, validate_dataset_records, ) +from .data_agent_inputs import ( + DataAgentInputError, + dumps_payload as data_agent_dumps_payload, + extract_case_evidence, + load_input_sources, + render_source_context, +) from .session_env_vars import get_session_env_vars if TYPE_CHECKING: @@ -1252,14 +1262,92 @@ def default_tool_registry() -> dict[str, AgentTool]: handler=_execute_skill, ), AgentTool( - name='data_agent_prepare_generation_plan', + name='data_agent_prepare_generation_goal', description=( - 'Create a pending data-agent generation plan. Use this before generating dataset draft text; ' - 'show the returned plan to the user and wait for confirmation.' + 'Create a pending data-agent generation goal from source evidence or user rules. ' + 'Use this before data_agent_prepare_generation_plan; show the returned goal to the user and wait for confirmation.' ), parameters={ 'type': 'object', 'properties': { + 'dataset_label': {'type': 'string'}, + 'goal_summary': {'type': 'string'}, + 'target': { + 'type': 'string', + 'description': 'Single target label. For multi-target boundary tasks, use target_definitions.', + }, + 'target_definitions': { + 'type': 'array', + 'items': { + 'type': 'object', + 'properties': { + 'name': {'type': 'string'}, + 'target': {'type': 'string'}, + 'rule': {'type': 'string'}, + }, + 'required': ['target'], + }, + }, + 'plan_hint': { + 'type': 'string', + 'description': ( + 'Optional plain-language hint for the later plan, such as ' + '"建议先生成 50 条单轮，输出到 tasks/.../records.jsonl". ' + 'Structured total_count, turn_mix, and output_path belong to data_agent_prepare_generation_plan.' + ), + }, + 'coverage': {'type': 'string'}, + 'exclusions': {'type': 'string'}, + 'open_questions': { + 'type': 'array', + 'items': {'type': 'string'}, + }, + 'source_refs': { + 'type': 'array', + 'items': {'type': 'string'}, + }, + 'notes': {'type': 'string'}, + }, + 'required': ['dataset_label', 'goal_summary', 'coverage', 'exclusions'], + }, + handler=_data_agent_prepare_generation_goal_tool, + ), + AgentTool( + name='data_agent_confirm_generation_goal', + description=( + 'Mark a pending data-agent generation goal as confirmed after the user explicitly approves it. ' + 'The returned confirmed_goal_id is required by data_agent_prepare_generation_plan.' + ), + parameters={ + 'type': 'object', + 'properties': { + 'goal_id': {'type': 'string'}, + 'confirmation': { + 'type': 'string', + 'description': 'The user approval text, such as 确认, 开始生成, or approve.', + }, + 'reviewed_revision': { + 'type': 'integer', + 'description': 'The goal revision shown to the user before confirmation.', + }, + }, + 'required': ['goal_id', 'confirmation'], + }, + handler=_data_agent_confirm_generation_goal_tool, + ), + AgentTool( + name='data_agent_prepare_generation_plan', + description=( + 'Create a pending data-agent generation plan. Use this before generating dataset draft text; ' + 'requires confirmed_goal_id from data_agent_confirm_generation_goal; show the returned plan to the user and wait for confirmation.' + ), + parameters={ + 'type': 'object', + 'properties': { + 'confirmed_goal_id': { + 'type': 'string', + 'description': 'Required; returned by data_agent_confirm_generation_goal after the user reviews the generation goal.', + }, 'dataset_label': {'type': 'string'}, 'target': { 'type': 'string', @@ -1286,6 +1374,7 @@ def default_tool_registry() -> dict[str, AgentTool]: 'notes': {'type': 'string'}, }, 'required': [ + 'confirmed_goal_id', 'dataset_label', 'total_count', 'turn_mix', @@ -1296,6 +1385,87 @@ def default_tool_registry() -> dict[str, AgentTool]: }, handler=_data_agent_prepare_generation_plan_tool, ), + AgentTool( + name='data_agent_load_input_sources', + description=( + 'Load data-agent input files or directories and extract structured paragraphs and table previews ' + 'from xlsx, csv, docx, pdf, txt, md, json, or jsonl sources.' + ), + parameters={ + 'type': 'object', + 'properties': { + 'paths': { + 'type': 'array', + 'items': {'type': 'string'}, + 'description': 'Workspace-relative files or directories to load.', + }, + 'max_files': {'type': 'integer', 'minimum': 1, 'maximum': 100}, + 'max_paragraphs_per_file': {'type': 'integer', 'minimum': 1, 'maximum': 300}, + 'max_tables_per_file': {'type': 'integer', 'minimum': 1, 'maximum': 100}, + 'max_rows_per_table': {'type': 'integer', 'minimum': 1, 'maximum': 200}, + 'max_cell_chars': {'type': 'integer', 'minimum': 20, 'maximum': 2000}, + }, + 'required': ['paths'], + }, + handler=_data_agent_load_input_sources_tool, + ), + AgentTool( + name='data_agent_extract_case_evidence', + description=( + 'Extract query/badcase evidence from loaded input sources or source paths. ' + 'Profiles table columns first and returns required questions when query or expected label columns are ambiguous.' + ), + parameters={ + 'type': 'object', + 'properties': { + 'paths': { + 'type': 'array', + 'items': {'type': 'string'}, + 'description': 'Workspace-relative files or directories. Omit when loaded_sources is provided.', + }, + 'loaded_sources': { + 'type': 'object', + 'description': 'Output from data_agent_load_input_sources.', + }, + 'field_mapping': { + 'type': 'object', + 'description': 'Optional role-to-column mapping, e.g. {"query":"query","expected_label":"预期domain"}.', + }, + 'max_cases': {'type': 'integer', 'minimum': 1, 'maximum': 1000}, + }, + }, + handler=_data_agent_extract_case_evidence_tool, + ), + AgentTool( + name='data_agent_render_source_context', + description=( + 'Render loaded data-agent sources into LLM-readable evidence text with source refs. ' + 'Use this after data_agent_load_input_sources before asking the model to structure product semantics.' + ), + parameters={ + 'type': 'object', + 'properties': { + 'paths': { + 'type': 'array', + 'items': {'type': 'string'}, + 'description': 'Workspace-relative files or directories. Omit when loaded_sources is provided.', + }, + 'loaded_sources': { + 'type': 'object', + 'description': 'Output from data_agent_load_input_sources.', + }, + 'max_chars': {'type': 'integer', 'minimum': 1000, 'maximum': 200000}, + 'max_tables_per_source': {'type': 'integer', 'minimum': 1, 'maximum': 100}, + 'max_rows_per_table': {'type': 'integer', 'minimum': 1, 'maximum': 200}, + 'focus_keywords': { + 'type': 'array', + 'items': {'type': 'string'}, + 'description': 'Optional keywords used to keep only matching paragraphs/table rows.', + }, + }, + }, + handler=_data_agent_render_source_context_tool, + ), AgentTool( name='data_agent_show_generation_plan', description='Show a data-agent generation plan for human review.', @@ -1417,6 +1587,44 @@ def default_tool_registry() -> dict[str, AgentTool]: }, handler=_validate_dataset_records_tool, ), + AgentTool( + name='data_agent_export_dataset_records', + description=( + 'Validate canonical data-agent records and write them to a workspace file. ' + 'Defaults to compact JSONL, one canonical record per line, so the model does not hand-write JSON output.' + ), + parameters={ + 'type': 'object', + 'properties': { + 'records': { + 'oneOf': [ + {'type': 'array'}, + {'type': 'string'}, + ], + 'description': 'Canonical records as an array, or a JSON string containing the array.', + }, + 'output_path': { + 'type': 'string', + 'description': 'Workspace-relative path to write, usually ending in .jsonl.', + }, + 'output_format': { + 'type': 'string', + 'enum': ['jsonl', 'json'], + 'description': 'Defaults to jsonl. json writes a compact JSON array.', + }, + 'require_validation_ok': { + 'type': 'boolean', + 'description': 'When true, refuse to write records with validation errors.', + }, + 'overwrite': { + 'type': 'boolean', + 'description': 'When false, refuse to overwrite an existing file.', + }, + }, + 'required': ['records', 'output_path'], + }, + handler=_export_dataset_records_tool, + ), ] return {tool.name: tool for tool in tools} @@ -1488,6 +1696,64 @@ def _coerce_float(arguments: dict[str, Any], key: str, default: float) -> float: return float(value) +def _optional_string_list(arguments: dict[str, Any], key: str) -> list[str] | None: + value = arguments.get(key) + if value is None: + return None + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + raise ToolExecutionError(f'{key} must be an array of strings') + return value + + +def _data_agent_prepare_generation_goal_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str: + notes = arguments.get('notes', '') + if not isinstance(notes, str): + raise ToolExecutionError('notes must be a string') + try: + payload = prepare_generation_goal( + root=str(context.root), + dataset_label=_require_string(arguments, 'dataset_label'), + goal_summary=_require_string(arguments, 'goal_summary'), + target=_optional_string(arguments, 'target'), + target_definitions=_optional_target_definitions(arguments.get('target_definitions')), + plan_hint=_optional_string(arguments, 'plan_hint'), + coverage=_require_string(arguments, 'coverage'), + exclusions=_require_string(arguments, 'exclusions'), + open_questions=_optional_string_list(arguments, 'open_questions'), + source_refs=_optional_string_list(arguments, 'source_refs'), + notes=notes, + ) + except DataRecordError as exc: + raise ToolExecutionError(str(exc)) from exc + return ( + json.dumps(payload, ensure_ascii=False, indent=2), + { + 'action': 'data_agent_prepare_generation_goal', + 'requires_user_review': True, + 'goal_id': payload.get('goal', {}).get('goal_id'), + 'goal_revision': payload.get('goal', {}).get('revision'), + }, + ) + + +def _data_agent_confirm_generation_goal_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str: + reviewed_revision = arguments.get('reviewed_revision') + if reviewed_revision is not None and ( + isinstance(reviewed_revision, bool) or not isinstance(reviewed_revision, int) + ): + raise ToolExecutionError('reviewed_revision must be an integer') + try: + payload = confirm_generation_goal( + root=str(context.root), + goal_id=_require_string(arguments, 'goal_id'), + confirmation=_require_string(arguments, 'confirmation'), + reviewed_revision=reviewed_revision, + ) + except DataRecordError as exc: + raise ToolExecutionError(str(exc)) from exc + return json.dumps(payload, ensure_ascii=False, indent=2) + + def _data_agent_prepare_generation_plan_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str: total_count = _coerce_int(arguments, 'total_count', 0) notes = arguments.get('notes', '') @@ -1505,6 +1771,7 @@ def _data_agent_prepare_generation_plan_tool(arguments: dict[str, Any], context: output_path=_require_string(arguments, 'output_path'), target_definitions=_optional_target_definitions(arguments.get('target_definitions')), notes=notes, + confirmed_goal_id=_require_string(arguments, 'confirmed_goal_id'), ) except DataRecordError as exc: raise ToolExecutionError(str(exc)) from exc @@ -1519,6 +1786,75 @@ def _data_agent_prepare_generation_plan_tool(arguments: dict[str, Any], context: ) +def _data_agent_load_input_sources_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str: + paths = arguments.get('paths') + if not isinstance(paths, list) or not all(isinstance(item, str) for item in paths): + raise ToolExecutionError('paths must be an array of strings') + try: + payload = load_input_sources( + context.root, + paths, + max_files=_coerce_int(arguments, 'max_files', 20), + max_paragraphs_per_file=_coerce_int(arguments, 'max_paragraphs_per_file', 80), + max_tables_per_file=_coerce_int(arguments, 'max_tables_per_file', 20), + max_rows_per_table=_coerce_int(arguments, 'max_rows_per_table', 30), + max_cell_chars=_coerce_int(arguments, 'max_cell_chars', 240), + ) + except DataAgentInputError as exc: + raise ToolExecutionError(str(exc)) from exc + return data_agent_dumps_payload(payload) + + +def _data_agent_extract_case_evidence_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str: + paths = arguments.get('paths') + if paths is not None and (not isinstance(paths, list) or not all(isinstance(item, str) for item in paths)): + raise ToolExecutionError('paths must be an array of strings') + loaded_sources = arguments.get('loaded_sources') + if loaded_sources is not None and not isinstance(loaded_sources, dict): + raise ToolExecutionError('loaded_sources must be an object') + field_mapping = arguments.get('field_mapping') + if field_mapping is not None and not isinstance(field_mapping, dict): + raise ToolExecutionError('field_mapping must be an object') + try: + payload = extract_case_evidence( + context.root, + paths=paths, + loaded_sources=loaded_sources, + field_mapping=field_mapping, + max_cases=_coerce_int(arguments, 'max_cases', 200), + ) + except DataAgentInputError as exc: + raise ToolExecutionError(str(exc)) from exc + return data_agent_dumps_payload(payload) + + +def _data_agent_render_source_context_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str: + paths = arguments.get('paths') + if paths is not None and (not isinstance(paths, list) or not all(isinstance(item, str) for item in paths)): + raise ToolExecutionError('paths must be an array of strings') + loaded_sources = arguments.get('loaded_sources') + if loaded_sources is not None and not isinstance(loaded_sources, dict): + raise ToolExecutionError('loaded_sources must be an object') + focus_keywords = arguments.get('focus_keywords') + if focus_keywords is not None and ( + not isinstance(focus_keywords, list) or not all(isinstance(item, str) for item in focus_keywords) + ): + raise ToolExecutionError('focus_keywords must be an array of strings') + try: + payload = render_source_context( + context.root, + paths=paths, + loaded_sources=loaded_sources, + max_chars=_coerce_int(arguments, 'max_chars', 40_000), + max_tables_per_source=_coerce_int(arguments, 'max_tables_per_source', 12), + max_rows_per_table=_coerce_int(arguments, 'max_rows_per_table', 20), + focus_keywords=focus_keywords, + ) + except DataAgentInputError as exc: + raise ToolExecutionError(str(exc)) from exc + return data_agent_dumps_payload(payload) + + def _data_agent_show_generation_plan_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str: try: payload = get_generation_plan( @@ -1609,6 +1945,33 @@ def _validate_dataset_records_tool(arguments: dict[str, Any], _context: ToolExec return json.dumps(payload, ensure_ascii=False, indent=2) +def _export_dataset_records_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str: + if 'records' not in arguments: + raise ToolExecutionError('records is required') + output_format = arguments.get('output_format', 'jsonl') + if not isinstance(output_format, str): + raise ToolExecutionError('output_format must be a string') + 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') + try: + records = records_from_tool_argument(arguments['records']) + payload = export_dataset_records( + records, + root=str(context.root), + output_path=_require_string(arguments, 'output_path'), + output_format=output_format, + 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 _resolve_path(raw_path: str, context: ToolExecutionContext, *, allow_missing: bool = True) -> Path: expanded = Path(raw_path).expanduser() candidate = expanded if expanded.is_absolute() else context.root / expanded diff --git a/src/data_agent_inputs.py b/src/data_agent_inputs.py new file mode 100644 index 0000000..3bba184 --- /dev/null +++ b/src/data_agent_inputs.py @@ -0,0 +1,615 @@ +from __future__ import annotations + +"""数据智能体输入文件解析工具。 + +这里的工具只负责把产品定义、badcase 表格、走查文档等输入稳定抽成 +结构化证据。语义判断和生成策略仍交给 skill/模型完成，避免工具里写死业务规则。 +""" + +import csv +import json +import re +import zipfile +from pathlib import Path +from typing import Any +from xml.etree import ElementTree + + +class DataAgentInputError(ValueError): + """输入文件无法解析或参数结构不符合预期。""" + + +SUPPORTED_SUFFIXES = {'.csv', '.docx', '.json', '.jsonl', '.md', '.pdf', '.txt', '.xlsx'} +QUERY_ROLE = 'query' +EXPECTED_LABEL_ROLE = 'expected_label' +PREDICTED_LABEL_ROLE = 'predicted_label' +CONTEXT_ROLE = 'context' +NOTES_ROLE = 'notes' +CASE_TYPE_ROLE = 'case_type' + + +def load_input_sources( + root: str | Path, + paths: list[str], + *, + max_files: int = 20, + max_paragraphs_per_file: int = 80, + max_tables_per_file: int = 20, + max_rows_per_table: int = 30, + max_cell_chars: int = 240, +) -> dict[str, Any]: + resolved_paths = _resolve_input_paths(root, paths, max_files=max_files) + sources = [ + _load_one_source( + path, + root=Path(root).resolve(), + max_paragraphs=max_paragraphs_per_file, + max_tables=max_tables_per_file, + max_rows=max_rows_per_table, + max_cell_chars=max_cell_chars, + ) + for path in resolved_paths + ] + return {'sources': sources, 'source_count': len(sources)} + + +def extract_case_evidence( + root: str | Path, + *, + paths: list[str] | None = None, + loaded_sources: dict[str, Any] | None = None, + field_mapping: dict[str, str] | None = None, + max_cases: int = 200, +) -> dict[str, Any]: + sources = _sources_from_args(root, paths, loaded_sources) + evidence: list[dict[str, Any]] = [] + table_profiles: list[dict[str, Any]] = [] + required_questions: list[str] = [] + + for source in sources: + for table in source.get('tables', []): + headers = _table_headers(table) + header_index = _table_header_index(table) + if not headers: + continue + rows = table.get('rows') + if not isinstance(rows, list): + continue + profile = _profile_headers(headers, rows, field_mapping or {}) + profile.update({'source_ref': table.get('source_ref'), 'title': table.get('title', '')}) + table_profiles.append(profile) + questions = _questions_for_profile(profile) + required_questions.extend(questions) + if questions: + continue + evidence.extend( + _rows_to_case_evidence( + source, + table, + rows, + profile['role_to_column'], + header_index=header_index, + max_cases=max(0, max_cases - len(evidence)), + ) + ) + if len(evidence) >= max_cases: + break + if len(evidence) >= max_cases: + break + + return { + 'evidence': evidence, + 'evidence_count': len(evidence), + 'table_profiles': table_profiles, + 'required_questions': _dedupe(required_questions), + } + + +def render_source_context( + root: str | Path, + *, + paths: list[str] | None = None, + loaded_sources: dict[str, Any] | None = None, + max_chars: int = 40_000, + max_tables_per_source: int = 12, + max_rows_per_table: int = 20, + focus_keywords: list[str] | None = None, +) -> dict[str, Any]: + sources = _sources_from_args(root, paths, loaded_sources) + sections: list[str] = [] + truncated = False + keywords = [keyword for keyword in focus_keywords or [] if isinstance(keyword, str) and keyword.strip()] + for source in sources: + source_lines = [ + f'# Source: {source.get("path", "")}', + f'kind: {source.get("kind", "")}', + f'summary: {source.get("summary", "")}', + ] + warnings = source.get('warnings') + if warnings: + source_lines.append('warnings: ' + '; '.join(str(item) for item in warnings)) + for paragraph in source.get('paragraphs', []): + text = str(paragraph.get('text') or '').strip() + if not text or not _matches_focus(text, keywords): + continue + source_lines.append('') + source_lines.append(f'## Paragraph {paragraph.get("source_ref", "")}') + source_lines.append(text) + for table in source.get('tables', [])[:max_tables_per_source]: + headers = _table_headers(table) + rows = table.get('rows') + if not isinstance(rows, list): + continue + rendered_rows = [] + for row_index, row in enumerate(rows[:max_rows_per_table], start=1): + if not isinstance(row, list): + continue + row_text = _render_table_row(headers, row) if headers else ' | '.join(str(cell) for cell in row) + if not row_text or not _matches_focus(row_text, keywords): + continue + rendered_rows.append(f'[{table.get("source_ref")}:row{row_index}] {row_text}') + if rendered_rows: + source_lines.append('') + source_lines.append(f'## Table {table.get("title", "")} {table.get("source_ref", "")}') + if headers: + source_lines.append('columns: ' + ' | '.join(headers)) + source_lines.extend(rendered_rows) + section = '\n'.join(source_lines).strip() + if section: + sections.append(section) + + rendered = '\n\n'.join(sections) + if len(rendered) > max_chars: + rendered = rendered[: max_chars - 80].rstrip() + '\n\n...[truncated: source context exceeded max_chars]...' + truncated = True + return { + 'context_text': rendered, + 'char_count': len(rendered), + 'truncated': truncated, + 'source_count': len(sources), + 'instructions': '请让大模型只基于 context_text 抽取和用户 query 语义、功能点、边界、target/open questions 相关的信息，并保留 source refs。', + } + + +def _resolve_input_paths(root: str | Path, paths: list[str], *, max_files: int) -> list[Path]: + if not isinstance(paths, list) or not paths: + raise DataAgentInputError('paths must be a non-empty array') + root_path = Path(root).resolve() + resolved: list[Path] = [] + for raw in paths: + if not isinstance(raw, str) or not raw.strip(): + raise DataAgentInputError('paths items must be non-empty strings') + path = Path(raw).expanduser() + candidate = path if path.is_absolute() else root_path / path + candidate = candidate.resolve() + try: + candidate.relative_to(root_path) + except ValueError as exc: + raise DataAgentInputError(f'path escapes workspace root: {raw}') from exc + if not candidate.exists(): + raise DataAgentInputError(f'path not found: {raw}') + if candidate.is_dir(): + for child in sorted(candidate.rglob('*')): + if child.is_file() and child.suffix.lower() in SUPPORTED_SUFFIXES: + resolved.append(child.resolve()) + elif candidate.is_file(): + if candidate.suffix.lower() in SUPPORTED_SUFFIXES: + resolved.append(candidate) + if len(resolved) >= max_files: + break + return _dedupe_paths(resolved)[:max_files] + + +def _load_one_source( + path: Path, + *, + root: Path, + max_paragraphs: int, + max_tables: int, + max_rows: int, + max_cell_chars: int, +) -> dict[str, Any]: + suffix = path.suffix.lower() + rel_path = path.relative_to(root).as_posix() + source: dict[str, Any] = { + 'path': rel_path, + 'kind': suffix.lstrip('.'), + 'size_bytes': path.stat().st_size, + 'paragraphs': [], + 'tables': [], + 'warnings': [], + } + try: + if suffix == '.xlsx': + _load_xlsx(path, source, max_tables=max_tables, max_rows=max_rows, max_cell_chars=max_cell_chars) + elif suffix == '.csv': + _load_csv(path, source, max_rows=max_rows, max_cell_chars=max_cell_chars) + elif suffix == '.docx': + _load_docx(path, source, max_paragraphs=max_paragraphs, max_tables=max_tables, max_rows=max_rows, max_cell_chars=max_cell_chars) + elif suffix == '.pdf': + _load_pdf(path, source, max_paragraphs=max_paragraphs) + elif suffix in {'.txt', '.md', '.json', '.jsonl'}: + _load_text(path, source, max_paragraphs=max_paragraphs) + else: + source['warnings'].append(f'暂不支持的文件类型: {suffix}') + except Exception as exc: + source['warnings'].append(f'解析失败: {type(exc).__name__}: {exc}') + source['summary'] = _source_summary(source) + return source + + +def _load_xlsx(path: Path, source: dict[str, Any], *, max_tables: int, max_rows: int, max_cell_chars: int) -> None: + try: + import openpyxl # type: ignore[import-not-found] + except ImportError as exc: + raise DataAgentInputError('解析 xlsx 需要 openpyxl') from exc + workbook = openpyxl.load_workbook(path, data_only=True, read_only=True) + for sheet_index, sheet in enumerate(workbook.worksheets[:max_tables]): + # 有些 xlsx 的维度元数据不完整，read_only 模式会误判成 A1:A1。 + # reset_dimensions 后让 openpyxl 从实际行流里重新推断列数。 + if hasattr(sheet, 'reset_dimensions'): + sheet.reset_dimensions() + rows = [] + for row in sheet.iter_rows(max_row=max_rows, values_only=True): + rows.append([_clean_cell(value, max_cell_chars=max_cell_chars) for value in row]) + source['tables'].append( + { + 'source_ref': f'{source["path"]}#sheet={sheet.title}', + 'title': sheet.title, + 'row_count': sheet.max_row, + 'column_count': sheet.max_column, + 'rows': rows, + } + ) + if sheet_index + 1 >= max_tables: + break + + +def _load_csv(path: Path, source: dict[str, Any], *, max_rows: int, max_cell_chars: int) -> None: + rows: list[list[str]] = [] + with path.open('r', encoding='utf-8-sig', errors='replace', newline='') as handle: + reader = csv.reader(handle) + for index, row in enumerate(reader): + if index >= max_rows: + break + rows.append([_clean_cell(cell, max_cell_chars=max_cell_chars) for cell in row]) + source['tables'].append( + { + 'source_ref': source['path'], + 'title': path.name, + 'row_count': len(rows), + 'column_count': max((len(row) for row in rows), default=0), + 'rows': rows, + } + ) + + +def _load_docx( + path: Path, + source: dict[str, Any], + *, + max_paragraphs: int, + max_tables: int, + max_rows: int, + max_cell_chars: int, +) -> None: + with zipfile.ZipFile(path) as archive: + document_xml = archive.read('word/document.xml') + root = ElementTree.fromstring(document_xml) + namespace = {'w': 'http://schemas.openxmlformats.org/wordprocessingml/2006/main'} + paragraphs: list[dict[str, str]] = [] + tables: list[dict[str, Any]] = [] + body = root.find('w:body', namespace) + if body is None: + return + for element in list(body): + if element.tag.endswith('}p') and len(paragraphs) < max_paragraphs: + text = _docx_text(element, namespace) + if text: + paragraphs.append({'text': text, 'source_ref': f'{source["path"]}#p{len(paragraphs) + 1}'}) + elif element.tag.endswith('}tbl') and len(tables) < max_tables: + rows = _docx_table_rows(element, namespace, max_rows=max_rows, max_cell_chars=max_cell_chars) + tables.append( + { + 'source_ref': f'{source["path"]}#table={len(tables)}', + 'title': f'table_{len(tables)}', + 'row_count': len(rows), + 'column_count': max((len(row) for row in rows), default=0), + 'rows': rows, + } + ) + source['paragraphs'].extend(paragraphs) + source['tables'].extend(tables) + + +def _load_pdf(path: Path, source: dict[str, Any], *, max_paragraphs: int) -> None: + try: + from pypdf import PdfReader # type: ignore[import-not-found] + except ImportError: + source['warnings'].append('当前 Python 环境缺少 pypdf，无法抽取 PDF 文本。优先使用同名 docx。') + return + reader = PdfReader(str(path)) + for index, page in enumerate(reader.pages[:max_paragraphs]): + text = ' '.join((page.extract_text() or '').split()) + if text: + source['paragraphs'].append({'text': text, 'source_ref': f'{source["path"]}#page={index + 1}'}) + + +def _load_text(path: Path, source: dict[str, Any], *, max_paragraphs: int) -> None: + text = path.read_text(encoding='utf-8', errors='replace') + for index, block in enumerate(re.split(r'\n\s*\n', text)): + if index >= max_paragraphs: + break + block = ' '.join(block.split()) + if block: + source['paragraphs'].append({'text': block, 'source_ref': f'{source["path"]}#block={index + 1}'}) + + +def _docx_text(element: ElementTree.Element, namespace: dict[str, str]) -> str: + parts = [node.text or '' for node in element.findall('.//w:t', namespace)] + return ' '.join(''.join(parts).split()) + + +def _docx_table_rows( + table: ElementTree.Element, + namespace: dict[str, str], + *, + max_rows: int, + max_cell_chars: int, +) -> list[list[str]]: + rows: list[list[str]] = [] + for row in table.findall('./w:tr', namespace): + if len(rows) >= max_rows: + break + cells = [] + for cell in row.findall('./w:tc', namespace): + cells.append(_clean_cell(_docx_text(cell, namespace), max_cell_chars=max_cell_chars)) + rows.append(cells) + return rows + + +def _sources_from_args( + root: str | Path, + paths: list[str] | None, + loaded_sources: dict[str, Any] | None, +) -> list[dict[str, Any]]: + if loaded_sources is not None: + sources = loaded_sources.get('sources') if isinstance(loaded_sources, dict) else None + if not isinstance(sources, list): + raise DataAgentInputError('loaded_sources.sources must be an array') + return [source for source in sources if isinstance(source, dict)] + if paths is None: + raise DataAgentInputError('paths or loaded_sources is required') + return load_input_sources(root, paths)['sources'] + + +def _table_headers(table: dict[str, Any]) -> list[str]: + rows = table.get('rows') + if not isinstance(rows, list) or not rows: + return [] + first = rows[_table_header_index(table)] + if not isinstance(first, list): + return [] + headers = [] + for index, cell in enumerate(first): + text = str(cell).strip() + headers.append(text or f'column_{index + 1}') + return headers + + +def _table_header_index(table: dict[str, Any]) -> int: + rows = table.get('rows') + if not isinstance(rows, list): + return 0 + best_index = 0 + best_score = -1 + for index, row in enumerate(rows[:12]): + if not isinstance(row, list): + continue + score = 0 + for cell in [str(value).strip().lower() for value in row]: + if any(token in cell for token in ('query', '用户query', '预期', 'label', 'domain', '类型', '类别', '场景', '意图')): + score += 2 + if any(token in cell for token in ('备注', 'case问题', '高置信', '低置信', '用户主要诉求')): + score += 1 + if score > best_score: + best_score = score + best_index = index + return best_index if best_score > 0 else 0 + + +def _profile_headers( + headers: list[str], + rows: list[Any], + field_mapping: dict[str, str], +) -> dict[str, Any]: + column_profiles: list[dict[str, Any]] = [] + role_to_column: dict[str, str] = {} + explicit = {str(role): str(column) for role, column in field_mapping.items()} + for header in headers: + if not header: + continue + samples = _sample_values(header, headers, rows) + role, confidence = _detect_column_role(header, samples) + for explicit_role, explicit_column in explicit.items(): + if explicit_column == header: + role = explicit_role + confidence = 1.0 + if role and role not in role_to_column: + role_to_column[role] = header + column_profiles.append( + { + 'name': header, + 'detected_role': role or 'unknown', + 'confidence': confidence, + 'sample_values': samples[:5], + } + ) + return {'columns': column_profiles, 'role_to_column': role_to_column} + + +def _detect_column_role(header: str, samples: list[str]) -> tuple[str, float]: + name = header.strip().lower() + joined = ' '.join(samples[:8]).lower() + if any(token in name for token in ('query', 'utterance', '用户query', '请求', '问题')): + return QUERY_ROLE, 0.95 + if any(token in name for token in ('预期', 'expected', '正确', '人工', 'gold', 'target', 'label')): + return EXPECTED_LABEL_ROLE, 0.85 + if any(token in name for token in ('pred', '预测', '模型', 'prev-domain', '当前domain', '实际domain', '命中')): + return PREDICTED_LABEL_ROLE, 0.8 + if any(token in name for token in ('context', '上下文', 'prev_session', 'session')): + return CONTEXT_ROLE, 0.85 + if any(token in name for token in ('备注', 'note', '原因', '反馈', 'case问题', '说明')): + return NOTES_ROLE, 0.8 + if any(token in name for token in ('type', '类型', '类别', '场景', '子类')): + return CASE_TYPE_ROLE, 0.75 + if any(marker in joined for marker in ('agent(', 'function', 'productagent', 'complex_task')): + return EXPECTED_LABEL_ROLE, 0.45 + if name.startswith('column_') and any(len(sample) >= 10 for sample in samples): + return NOTES_ROLE, 0.35 + return '', 0.0 + + +def _questions_for_profile(profile: dict[str, Any]) -> list[str]: + role_to_column = profile.get('role_to_column') if isinstance(profile.get('role_to_column'), dict) else {} + title = profile.get('title') or profile.get('source_ref') or '表格' + questions = [] + if QUERY_ROLE not in role_to_column: + questions.append(f'{title} 没有稳定识别到 query 列，请确认哪一列是用户 query。') + if EXPECTED_LABEL_ROLE not in role_to_column: + questions.append(f'{title} 没有稳定识别到预期标签列，请确认哪一列是正确 label/target。') + return questions + + +def _rows_to_case_evidence( + source: dict[str, Any], + table: dict[str, Any], + rows: list[Any], + role_to_column: dict[str, str], + *, + header_index: int, + max_cases: int, +) -> list[dict[str, Any]]: + if max_cases <= 0: + return [] + headers = _table_headers(table) + evidence = [] + for row_index, row in enumerate(rows[header_index + 1 :], start=header_index + 2): + if not isinstance(row, list): + continue + query = _cell_by_header(row, headers, role_to_column.get(QUERY_ROLE)) + expected = _cell_by_header(row, headers, role_to_column.get(EXPECTED_LABEL_ROLE)) + if not query: + continue + evidence.append( + { + 'query': _clean_query_text(query), + 'expected_label': expected, + 'predicted_label': _cell_by_header(row, headers, role_to_column.get(PREDICTED_LABEL_ROLE)), + 'context': _cell_by_header(row, headers, role_to_column.get(CONTEXT_ROLE)), + 'case_type': _cell_by_header(row, headers, role_to_column.get(CASE_TYPE_ROLE)), + 'notes': _cell_by_header(row, headers, role_to_column.get(NOTES_ROLE)), + 'source_ref': f'{table.get("source_ref") or source.get("path")}:row{row_index}', + } + ) + if len(evidence) >= max_cases: + break + return evidence + + +def _cell_by_header(row: list[Any], headers: list[str], header: str | None) -> str: + if not header: + return '' + try: + index = headers.index(header) + except ValueError: + return '' + if index >= len(row): + return '' + return str(row[index] or '').strip() + + +def _sample_values(header: str, headers: list[str], rows: list[Any]) -> list[str]: + samples = [] + start_index = 0 + for index, row in enumerate(rows[:12]): + if isinstance(row, list) and _looks_like_header_row(row, headers): + start_index = index + 1 + break + for row in rows[start_index : start_index + 12]: + if not isinstance(row, list): + continue + value = _cell_by_header(row, headers, header) + if value: + samples.append(value[:120]) + return samples + + +def _looks_like_header_row(row: list[Any], headers: list[str]) -> bool: + return [str(cell).strip() for cell in row] == headers + + +def _clean_query_text(text: str) -> str: + text = re.sub(r'^[QA]\s*[:：]\s*', '', text.strip()) + return ' / '.join(part.strip() for part in re.split(r'\s*/\s*', text) if part.strip()) + + +def _render_table_row(headers: list[str], row: list[Any]) -> str: + pairs = [] + for index, cell in enumerate(row): + text = str(cell or '').strip() + if not text: + continue + header = headers[index] if index < len(headers) else f'column_{index + 1}' + pairs.append(f'{header}: {text}') + return ' | '.join(pairs) + + +def _matches_focus(text: str, keywords: list[str]) -> bool: + if not keywords: + return True + return any(keyword in text for keyword in keywords) + + +def _clean_cell(value: Any, *, max_cell_chars: int) -> str: + if value is None: + return '' + text = ' / '.join(str(value).split()) + if len(text) > max_cell_chars: + return text[: max_cell_chars - 3] + '...' + return text + + +def _source_summary(source: dict[str, Any]) -> str: + return ( + f'{source.get("kind")} 文件，' + f'{len(source.get("paragraphs", []))} 个段落，' + f'{len(source.get("tables", []))} 个表格，' + f'{len(source.get("warnings", []))} 个警告' + ) + + +def _dedupe(values: list[str]) -> list[str]: + seen = set() + output = [] + for value in values: + if value in seen: + continue + seen.add(value) + output.append(value) + return output + + +def _dedupe_paths(paths: list[Path]) -> list[Path]: + seen = set() + output = [] + for path in paths: + if path in seen: + continue + seen.add(path) + output.append(path) + return output + + +def dumps_payload(payload: dict[str, Any]) -> str: + return json.dumps(payload, ensure_ascii=False, indent=2) diff --git a/src/data_agent_records.py b/src/data_agent_records.py index 06edb0b..a54ae9e 100644 --- a/src/data_agent_records.py +++ b/src/data_agent_records.py @@ -10,18 +10,109 @@ Skill 应该先让模型生成紧凑、便于人工 review 的 dataset draft tex import json import re import time +from pathlib import Path from typing import Any DEFAULT_REQUEST_ID = 'aabbccdd' DEFAULT_TIMESTAMP_STEP_MS = 60_000 DEFAULT_PLAN_STATE_FILE = '.port_sessions/data_agent_generation_plans.json' +DEFAULT_GOAL_STATE_FILE = '.port_sessions/data_agent_generation_goals.json' class DataRecordError(ValueError): """Raised when dataset draft text or records are structurally invalid.""" +def prepare_generation_goal( + *, + root: str, + dataset_label: str, + goal_summary: str, + coverage: str, + exclusions: str, + target: str = '', + target_definitions: list[dict[str, Any]] | None = None, + plan_hint: str = '', + open_questions: list[str] | None = None, + source_refs: list[str] | None = None, + notes: str = '', +) -> dict[str, Any]: + missing = _missing_plan_fields( + { + 'dataset_label': dataset_label, + 'goal_summary': goal_summary, + 'coverage': coverage, + 'exclusions': exclusions, + } + ) + if missing: + raise DataRecordError('missing required goal fields: ' + ', '.join(missing)) + normalized_targets = _normalize_target_definitions(target_definitions) + state = _load_goal_state(root) + sequence = int(state.get('next_sequence') or 1) + goal_id = f'data_goal_{sequence:06d}' + goal = { + 'goal_id': goal_id, + 'status': 'pending_confirmation', + 'dataset_label': dataset_label.strip(), + 'goal_summary': goal_summary.strip(), + 'target': target.strip(), + 'target_definitions': normalized_targets, + 'plan_hint': plan_hint.strip(), + 'coverage': coverage.strip(), + 'exclusions': exclusions.strip(), + 'open_questions': _normalize_string_list(open_questions, 'open_questions'), + 'source_refs': _normalize_string_list(source_refs, 'source_refs'), + 'notes': notes.strip(), + 'revision': 1, + 'created_at_ms': int(time.time() * 1000), + } + goals = state.setdefault('goals', {}) + if not isinstance(goals, dict): + goals = {} + state['goals'] = goals + goals[goal_id] = goal + state['next_sequence'] = sequence + 1 + _save_goal_state(root, state) + return { + 'goal': goal, + 'message': 'generation goal 已创建，状态为 pending_confirmation。请向用户展示 goal，并等待用户确认后再创建 generation plan。', + } + + +def confirm_generation_goal( + *, + root: str, + goal_id: str, + confirmation: str, + reviewed_revision: int | None = None, +) -> dict[str, Any]: + if not goal_id.strip(): + raise DataRecordError('goal_id is required') + if not _is_confirmation_text(confirmation): + raise DataRecordError('confirmation must clearly approve the generation goal') + state = _load_goal_state(root) + goal = _get_goal(state, goal_id) + current_revision = int(goal.get('revision') or 1) + if reviewed_revision is not None and reviewed_revision != current_revision: + raise DataRecordError( + f'reviewed_revision must match current goal revision {current_revision}' + ) + if not _allowed_targets_for_goal(goal): + raise DataRecordError('confirmed generation goal must include target or target_definitions') + goal['status'] = 'confirmed' + goal['confirmed_at_ms'] = int(time.time() * 1000) + goal['confirmation'] = confirmation.strip() + goal['confirmed_revision'] = current_revision + _save_goal_state(root, state) + return { + 'goal': goal, + 'confirmed_goal_id': goal_id, + 'message': 'generation goal 已确认，可以把 confirmed_goal_id 传给 data_agent_prepare_generation_plan。', + } + + def normalize_dataset_draft( draft_text: str, *, @@ -80,7 +171,11 @@ def prepare_generation_plan( output_path: str, target_definitions: list[dict[str, Any]] | None = None, notes: str = '', + confirmed_goal_id: str | None = None, ) -> dict[str, Any]: + confirmed_goal: dict[str, Any] | None = None + if confirmed_goal_id is not None: + confirmed_goal = _require_confirmed_generation_goal(root, confirmed_goal_id) missing = _missing_plan_fields( { 'dataset_label': dataset_label, @@ -97,6 +192,13 @@ def prepare_generation_plan( raise DataRecordError('missing required plan fields: ' + ', '.join(missing)) if total_count <= 0: raise DataRecordError('total_count must be greater than 0') + if confirmed_goal is not None: + _validate_plan_against_goal( + dataset_label=dataset_label, + target=target, + target_definitions=normalized_targets, + goal=confirmed_goal, + ) state = _load_plan_state(root) sequence = int(state.get('next_sequence') or 1) @@ -115,6 +217,7 @@ def prepare_generation_plan( 'output_path': resolved_output_path, 'requested_output_path': output_path.strip(), 'notes': notes.strip(), + 'confirmed_goal_id': confirmed_goal_id or '', 'revision': 1, 'review_history': [], 'created_at_ms': int(time.time() * 1000), @@ -342,6 +445,56 @@ def validate_dataset_records(records: list[dict[str, Any]]) -> dict[str, Any]: } +def export_dataset_records( + records: list[dict[str, Any]], + *, + root: str, + output_path: str, + output_format: str = 'jsonl', + require_validation_ok: bool = True, + overwrite: bool = True, +) -> dict[str, Any]: + """校验并导出 canonical records。 + + 这个函数用于收口数据智能体的最终落盘动作，避免模型手写 JSON/JSONL。 + 默认导出紧凑 JSONL：一行一个 canonical record，方便 diff、抽样和增量处理。 + """ + + if output_format not in {'jsonl', 'json'}: + raise DataRecordError('output_format must be jsonl or json') + if not output_path.strip(): + raise DataRecordError('output_path is required') + + 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) + + if output_format == 'jsonl': + content = ''.join( + json.dumps(record, ensure_ascii=False, separators=(',', ':')) + '\n' + for record in records + ) + else: + content = json.dumps(records, ensure_ascii=False, separators=(',', ':')) + '\n' + + path.write_text(content, encoding='utf-8') + rel_path = str(path.relative_to(Path(root).resolve())) + return { + 'output_path': rel_path, + 'output_format': output_format, + 'record_count': len(records), + 'bytes_written': len(content.encode('utf-8')), + 'validation': validation, + } + + def _parse_draft_text(draft_text: str) -> dict[str, Any]: dataset_label = '' cases: list[dict[str, Any]] = [] @@ -418,6 +571,25 @@ def _normalize_target_definitions(value: list[dict[str, Any]] | None) -> list[di return normalized +def _normalize_string_list(value: list[str] | None, field_name: str) -> list[str]: + if value is None: + return [] + if not isinstance(value, list): + raise DataRecordError(f'{field_name} must be a list') + normalized: list[str] = [] + for index, item in enumerate(value): + if not isinstance(item, str): + raise DataRecordError(f'{field_name}[{index}] must be a string') + text = item.strip() + if text: + normalized.append(text) + return normalized + + +def _allowed_targets_for_goal(goal: dict[str, Any]) -> set[str]: + return _allowed_targets_for_plan(goal) + + def _allowed_targets_for_plan(plan: dict[str, Any]) -> set[str]: target_definitions = plan.get('target_definitions') allowed: set[str] = set() @@ -431,6 +603,33 @@ def _allowed_targets_for_plan(plan: dict[str, Any]) -> set[str]: return allowed +def _validate_plan_against_goal( + *, + dataset_label: str, + target: str, + target_definitions: list[dict[str, str]], + goal: dict[str, Any], +) -> None: + goal_label = str(goal.get('dataset_label') or '').strip() + if goal_label and dataset_label.strip() != goal_label: + raise DataRecordError( + f'plan dataset_label must match confirmed generation goal: {goal_label}' + ) + goal_targets = _allowed_targets_for_goal(goal) + plan_targets = set() + if target.strip(): + plan_targets.add(target.strip()) + for item in target_definitions: + item_target = item.get('target', '').strip() + if item_target: + plan_targets.add(item_target) + unexpected = sorted(plan_targets - goal_targets) + if unexpected: + raise DataRecordError( + 'plan targets must match confirmed generation goal: ' + ', '.join(unexpected) + ) + + def _validate_records_against_plan(records: list[dict[str, Any]], plan: dict[str, Any]) -> None: allowed_targets = _allowed_targets_for_plan(plan) if not allowed_targets: @@ -475,6 +674,12 @@ def _plan_state_path(root: str): return Path(root).resolve() / DEFAULT_PLAN_STATE_FILE +def _goal_state_path(root: str): + from pathlib import Path + + return Path(root).resolve() / DEFAULT_GOAL_STATE_FILE + + def _load_plan_state(root: str) -> dict[str, Any]: path = _plan_state_path(root) if not path.exists(): @@ -492,12 +697,35 @@ def _load_plan_state(root: str) -> dict[str, Any]: return payload +def _load_goal_state(root: str) -> dict[str, Any]: + path = _goal_state_path(root) + if not path.exists(): + return {'next_sequence': 1, 'goals': {}} + try: + payload = json.loads(path.read_text(encoding='utf-8')) + except json.JSONDecodeError as exc: + raise DataRecordError(f'invalid generation goal state file: {path}') from exc + if not isinstance(payload, dict): + raise DataRecordError(f'invalid generation goal state file: {path}') + if not isinstance(payload.get('goals'), dict): + payload['goals'] = {} + if not isinstance(payload.get('next_sequence'), int): + payload['next_sequence'] = 1 + return payload + + def _save_plan_state(root: str, state: dict[str, Any]) -> None: path = _plan_state_path(root) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding='utf-8') +def _save_goal_state(root: str, state: dict[str, Any]) -> None: + path = _goal_state_path(root) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding='utf-8') + + def _get_plan(state: dict[str, Any], plan_id: str) -> dict[str, Any]: plans = state.get('plans') if not isinstance(plans, dict): @@ -508,6 +736,16 @@ def _get_plan(state: dict[str, Any], plan_id: str) -> dict[str, Any]: return plan +def _get_goal(state: dict[str, Any], goal_id: str) -> dict[str, Any]: + goals = state.get('goals') + if not isinstance(goals, dict): + raise DataRecordError('generation goal state is invalid') + goal = goals.get(goal_id) + if not isinstance(goal, dict): + raise DataRecordError(f'generation goal not found: {goal_id}') + return goal + + def _require_confirmed_generation_plan(root: str, plan_id: str) -> dict[str, Any]: state = _load_plan_state(root) plan = _get_plan(state, plan_id) @@ -516,6 +754,14 @@ def _require_confirmed_generation_plan(root: str, plan_id: str) -> dict[str, Any return plan +def _require_confirmed_generation_goal(root: str, goal_id: str) -> dict[str, Any]: + state = _load_goal_state(root) + goal = _get_goal(state, goal_id) + if goal.get('status') != 'confirmed': + raise DataRecordError(f'generation goal is not confirmed: {goal_id}') + return goal + + def _case_to_record( case: dict[str, Any], *, @@ -622,6 +868,18 @@ def _safe_token(value: str) -> str: return token.strip('_') or 'unknown' +def _resolve_output_path(root: str, output_path: str) -> Path: + root_path = Path(root).resolve() + candidate = Path(output_path).expanduser() + path = candidate if candidate.is_absolute() else root_path / candidate + resolved = path.resolve(strict=False) + try: + resolved.relative_to(root_path) + except ValueError as exc: + raise DataRecordError(f'output_path escapes workspace root: {output_path}') from exc + return resolved + + def _optional_int(value: Any) -> int | None: if value is None or value == '': return None diff --git a/src/skills/bundled/data-record-format/SKILL.md b/src/skills/bundled/data-record-format/SKILL.md deleted file mode 100644 index 714741e..0000000 --- a/src/skills/bundled/data-record-format/SKILL.md +++ /dev/null @@ -1,181 +0,0 @@ ---- -name: data-record-format -description: 定义、检查、校验并准备标准数据记录，作为导出前的统一中间格式。 -when_to_use: 当用户想讨论、创建、校验、归一化或转换标准数据记录格式时使用。 -aliases: canonical-record, record-format -allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_user_question, 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 ---- - -使用这个 skill 处理共享的数据格式层。目标是让模型先输出便于人工 review 的 dataset draft text，再通过工具进入统一 canonical record，最后再导出为训练、评测、prompt 或展示格式。 - -## 交互门禁 - -默认不要一步到位生成数据。除非用户已经明确说“开始生成”“确认计划”“按这个计划生成”或同义表达，否则只能做目标对齐、计划草案和问题确认。 - -开始生成前必须确认这些信息： - -- `dataset_label`：数据集或专题名称。 -- `target` / `target_definitions`：最终监督标签。单标签任务用 `target`，多标签边界任务必须用 `target_definitions` 列出每个标签和判定规则。 -- 生成数量：总条数，以及单轮/多轮数量或比例。 -- 覆盖范围：需要覆盖哪些 query 类型、意图边界或错误类型。 -- 负例/排除项：哪些表达不要生成，或哪些边界容易误判。 -- 落盘路径：draft、canonical records、validation result 写到哪里。 - -如果任一信息缺失，不要生成数据，不要调用 `data_agent_normalize_dataset_draft`，不要调用 `data_agent_validate_dataset_records`，只向用户提出需要确认的问题。 - -信息完整后，先调用 `data_agent_prepare_generation_plan` 创建 pending 计划，并把返回的计划展示给用户。此时必须停止，等待用户 review。 - -在真正生成前，必须先给用户展示一份简短生成计划，包含： - -```text -dataset_label: -target 或 target_definitions: -生成数量: -单轮/多轮: -覆盖范围: -负例/排除项: -落盘路径: -``` - -展示计划后停下来等用户 review。 - -- 如果用户提出修改意见，调用 `data_agent_update_generation_plan` 更新计划，然后重新展示计划并继续等待 review。 -- 如果用户要求查看当前计划，调用 `data_agent_show_generation_plan`。 -- 只有用户明确确认当前计划版本后，才能调用 `data_agent_confirm_generation_plan`。确认时传入刚展示给用户的 `reviewed_revision`。 - -用户确认后，拿到 `confirmed_plan_id`，才能生成 dataset draft text，并继续调用工具。 - -`data_agent_normalize_dataset_draft` 对生成数据有代码级门禁：没有 `confirmed_plan_id`，或者计划未确认，会拒绝执行。 - -多标签边界数据不要拆成多个互不相关的单标签计划。应该创建一个计划，并在 `target_definitions` 中列出所有候选标签。例如： - -```json -[ - { - "name": "餐饮服务", - "target": "Agent(tag=\"餐饮服务\")", - "rule": "找附近的美食、奶茶、餐厅等，没有明确要求导航。" - }, - { - "name": "地图导航", - "target": "Agent(tag=\"地图导航\")", - "rule": "明确出现导航去某地点、带我去某地点、路线规划等。" - } -] -``` - -生成 draft text 时，每条 case 的 `target:` 必须从已确认计划的 `target` 或 `target_definitions[*].target` 中选择。不要临时发明新 target。 - -## 数据生成输出格式 - -当需要生成数据样本时，默认使用 dataset draft text v1，不要直接输出 JSON、JSONL、CSV 或最终表格格式。除非用户明确要求机器可读格式，否则优先输出便于人工 review 的文本格式。 - -### 格式 - -```text -# dataset_label: 数据集或专题名称 - -### case: case名称 -用户: 本轮 query -target: Agent(tag="xxx") -notes: 可选，说明覆盖的问题或边界 - -### case: 多轮 case 名称 -用户: 前一轮 query -小爱: 前一轮 tts -用户: 本轮 query -target: Agent(tag="xxx") -notes: 可选，说明覆盖的问题或边界 -``` - -### 规则 - -- 每条数据用一个 `### case:` 开始。 -- `用户:` 表示用户 query。 -- `小爱:` 表示小爱回复 tts。 -- 最后一个 `用户:` 是本轮 query。 -- `target:` 是本轮 query 对应的监督标签，必须存在。 -- 单轮数据只需要写一行 `用户:`，然后写 `target:`。 -- 多轮数据需要按照时间顺序写多组 `用户:` / `小爱:`。 -- 多轮数据的最后一轮只写 `用户:` 和 `target:`，不要写最后一轮 `小爱:`，因为本轮 query 不包含 tts。 -- 如果 `target` 无法确定，不要编造，必须向用户确认。 -- 不要手写 `record_id`、`request_id`、`timestamp`、`context`。 -- 线上挖掘数据如有真实 `request_id` 和 `timestamp`，可以附加在 case 中；没有则不写。 - -## Canonical Record - -`data_agent_normalize_dataset_draft` 会把 dataset draft text 转成 canonical records，并统一补齐 `record_id`、`source`、`timestamp`、`context`、`target_type` 等机械字段。 - -当前 canonical record v1 工作格式： - -```json -{ - "record_id": "gen_aabbccdd_000001", - "source": { - "type": "generated", - "request_id": "aabbccdd", - "timestamp": 1755567930500 - }, - "turn": { - "query": "本轮 query", - "timestamp": 1755567930500 - }, - "prev_session": [ - { - "query": "前一轮 query", - "tts": "前一轮 tts", - "timestamp": 1755567870500 - } - ], - "context": {}, - "label": { - "dataset_label": "数据集或专题名称", - "target": "Agent(tag=\"xxx\")", - "target_type": "agent" - }, - "meta": { - "case_name": "case名称", - "notes": "" - } -} -``` - -## 必要工作循环 - -1. 先判断用户讨论的是模型生成草稿、canonical record 层，还是下游导出格式层。 -2. 如果用户提供了文件，先检查文件结构，再提出转换或校验方案。 -3. 如果生成目标没有确认完整，按“交互门禁”提问，不要继续生成。 -4. 信息完整后，调用 `data_agent_prepare_generation_plan` 创建 pending 计划，展示计划并等待用户 review。 -5. 用户提出修改意见时，调用 `data_agent_update_generation_plan`，然后重新展示计划。 -6. 用户明确确认当前计划版本后，调用 `data_agent_confirm_generation_plan` 获取 `confirmed_plan_id`。 -7. 让模型优先生成 dataset draft text，而不是直接生成 canonical JSON。 -8. 调用 `data_agent_normalize_dataset_draft`，并传入 `confirmed_plan_id`，转成 canonical records。 -9. 调用 `data_agent_validate_dataset_records` 校验字段、标签、时间戳和多轮结构。 -10. 明确记录必填字段中的不确定点，尤其是 `target`、标签类型、多轮结构和来源元数据。 -11. 如果当前 schema 信息不足以继续，提出简短问题，或把问题写入任务笔记。 -12. 在 canonical records 存在之前，不要直接创建最终训练、评测或展示格式。 - -## 工具使用建议 - -- 用 `read_file` 读取用户提供的样例或 schema 笔记。 -- 用 `write_file` 或 `edit_file` 起草 schema 笔记、样例记录或 open questions。 -- 用 `grep_search` 和 `glob_search` 查找已有 data-agent 文档和样例。 -- 用 `data_agent_prepare_generation_plan` 创建待确认生成计划。 -- 用 `data_agent_show_generation_plan` 查看当前计划。 -- 用 `data_agent_update_generation_plan` 根据用户 review 意见修改计划。 -- 用 `data_agent_confirm_generation_plan` 记录用户确认，并获取 `confirmed_plan_id`。 -- 用 `data_agent_normalize_dataset_draft` 把 dataset draft text 转成 canonical records，生成数据必须传入 `confirmed_plan_id`。 -- 用 `data_agent_validate_dataset_records` 校验 canonical records。 -- 后续导出阶段再使用 `convert_dataset_format` 或 `export_dataset`。 - -## 推荐产物 - -优先把过程沉淀到这些路径： - -```text -tasks/{task_id}/memory/open_questions.md -tasks/{task_id}/artifacts/canonical_record_notes.md -tasks/{task_id}/artifacts/sample_records.jsonl -``` - -最终回复保持简短，并引用创建或更新过的文件路径。 diff --git a/src/skills/bundled/product-definition-data-generation/SKILL.md b/src/skills/bundled/product-definition-data-generation/SKILL.md index 82f26c8..958a242 100644 --- a/src/skills/bundled/product-definition-data-generation/SKILL.md +++ b/src/skills/bundled/product-definition-data-generation/SKILL.md @@ -1,70 +1,293 @@ --- name: product-definition-data-generation -description: 从产品或标签定义文档中提取边界，并生成边界感知的 canonical data records。 -when_to_use: 当用户提供产品定义、标签规则或路由边界文档，并希望生成训练/评测数据时使用。 +description: 从产品/标签定义、手写边界规则或示例 query 中提取标签边界，并生成可 review 的数据计划、dataset draft text 和 canonical metadata records。 +when_to_use: 当用户提供产品定义、标签规则、路由边界文档、示例 query、手写标签边界，并希望生成训练/评测/专项数据时使用。 aliases: definition-data-generation, label-definition-generation -allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_user_question +allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_user_question, data_agent_load_input_sources, data_agent_render_source_context, data_agent_extract_case_evidence, data_agent_prepare_generation_goal, data_agent_confirm_generation_goal, data_agent_prepare_generation_plan, data_agent_show_generation_plan, data_agent_update_generation_plan, data_agent_confirm_generation_plan, data_agent_normalize_dataset_draft, data_agent_validate_dataset_records, data_agent_export_dataset_records --- -使用这个 skill 处理“产品/标签定义 -> 边界总结 -> review 后的数据生成计划 -> canonical records -> 导出”的工作流。 +使用这个 skill 作为“产品/标签定义/手写规则/示例 query -> 输入文本化 -> generation goal 草案 -> 人工 review -> 生成计划 review -> dataset draft text -> canonical metadata records”的统一入口。 + +本 skill 内置数据记录生成协议。其他数据开发 skill 如果需要生成或整理标准数据，可以复用这里的“交互门禁、dataset draft text v1、canonical record v1”规则。 ## 输入假设 用户可能会提供： -- 产品定义文档 -- 标签定义文档 -- 正例和反例 -- 路由、agent 或 function 选择规则 -- 已知边界冲突 +- 产品定义文档、标签定义文档、路由规则文档。 +- 表格、Markdown、JSON、CSV 或普通文本里的标签定义。 +- 手写的标签边界规则。 +- 一组 example query、badcase、正例或反例。 +- 已知边界冲突，例如某类 query 应该进入哪个 Agent/function。 -文档可能不完整或存在歧义。保留不确定性，不要自行发明隐藏规则。 +输入可能不完整或存在歧义。保留不确定性，不要自行发明隐藏规则。 -## 必要工作流 +## 任务定位 -1. 读取用户提供的定义文档；如果没有路径，先询问。 -2. 总结目标标签/类型及其预期边界。 -3. 提取： - - 正向规则 - - 负向规则 - - 模糊边界 - - 示例和反例 - - 缺失假设 -4. 写出可 review 的边界总结。 -5. 对重要的模糊边界，先请用户确认，再进行大规模生成。 -6. 创建生成计划，覆盖： - - 直接正例 - - hard negative - - 边界样例 - - 必要时包含单轮和多轮样例 -7. 只有在边界总结和计划清楚后，才生成 canonical records。 -8. TODO：工具实现后，校验 records、生成预览并导出目标格式。 +先判断用户输入属于哪一类： -## 当前工具状态 +1. **文件定义型**：用户提供文件路径、文档、表格或粘贴的大段定义内容。 +2. **手写规则型**：用户直接描述标签边界，例如“找附近美食是餐饮服务，导航去某地是地图导航”。 +3. **示例归纳型**：用户只给 query/example/badcase，需要先归纳边界和标签倾向。 -当前先使用已有工具完成文件检查和产物写入： - -- `read_file` -- `write_file` -- `edit_file` -- `grep_search` -- `glob_search` -- `ask_user_question` - -规划中的定义类工具，例如 `extract_label_definition` 和 `validate_label_coverage`，不一定已经实现。 - -## 推荐产物 +三类输入最后都要统一产出： ```text -tasks/{task_id}/context/product_definition.* -tasks/{task_id}/artifacts/label_boundary_summary.md -tasks/{task_id}/artifacts/generation_plan.md -tasks/{task_id}/artifacts/generated_candidates.jsonl -tasks/{task_id}/memory/open_questions.md +dataset_label: +target 或 target_definitions: +plan_hint: +coverage: +exclusions: +open_questions: +source_refs: ``` +这一步称为 `generation_goal`。它是模型基于输入资料整理出的数据生成目标草案，不是最终生成计划。 + +`generation_goal` 必须先展示给用户 review。只有用户确认 generation goal 后，才能进入本 skill 内置的 generation plan review 流程。 + +## 交互门禁 + +默认不要一步到位生成数据。除非用户已经明确说“开始生成”“确认计划”“按这个计划生成”或同义表达，否则只能做目标对齐、计划草案和问题确认。 + +开始生成前必须确认这些信息： + +- `dataset_label`：数据集或专题名称。 +- `target` / `target_definitions`：最终监督标签。单标签任务用 `target`，多标签边界任务必须用 `target_definitions` 列出每个标签和判定规则。 +- 生成数量：总条数，以及单轮/多轮数量或比例。 +- 覆盖范围：需要覆盖哪些 query 类型、意图边界或错误类型。 +- 负例/排除项：哪些表达不要生成，或哪些边界容易误判。 +- 落盘路径：draft、canonical records、validation result 写到哪里。 + +如果任一信息缺失，不要生成数据，不要调用 `data_agent_prepare_generation_plan`，不要调用 `data_agent_normalize_dataset_draft`，不要调用 `data_agent_validate_dataset_records`，只向用户提出需要确认的问题。 + +信息完整后，调用 `data_agent_prepare_generation_goal` 创建 pending goal。这个工具会暂停本轮，必须把返回的 `generation_goal` 展示给用户 review。用户确认 goal 之后，调用 `data_agent_confirm_generation_goal` 获取 `confirmed_goal_id`，再调用 `data_agent_prepare_generation_plan` 创建 pending plan。创建 plan 后也会暂停本轮，必须等待用户 review。 + +用户确认后，拿到 `confirmed_plan_id`，才能生成 dataset draft text，并继续调用工具。 + +`data_agent_normalize_dataset_draft` 对生成数据有代码级门禁：没有 `confirmed_plan_id`，或者计划未确认，会拒绝执行。 + +多标签边界数据不要拆成多个互不相关的单标签计划。应该创建一个计划，并在 `target_definitions` 中列出所有候选标签。例如： + +```json +[ + { + "name": "餐饮服务", + "target": "Agent(tag=\"餐饮服务\")", + "rule": "找附近的美食、奶茶、餐厅等，没有明确要求导航。" + }, + { + "name": "地图导航", + "target": "Agent(tag=\"地图导航\")", + "rule": "明确出现导航去某地点、带我去某地点、路线规划等。" + } +] +``` + +生成 draft text 时，每条 case 的 `target:` 必须从已确认计划的 `target` 或 `target_definitions[*].target` 中选择。不要临时发明新 target。 + +## 必须遵守的数据生成协议 + +不要直接生成 canonical JSON，不要直接导出最终训练/评测格式，不要绕过人类 review。 + +如果需要生成数据，必须按顺序执行： + +1. 用 `data_agent_load_input_sources` 读取文件。 +2. 用 `data_agent_render_source_context` 把输入渲染成大模型可读 evidence text。 +3. 大模型只基于 evidence text 抽取 `generation_goal`，包括 `dataset_label`、`target_definitions`、`plan_hint`、`coverage`、`exclusions`、`open_questions`、`source_refs`。 +4. 如果目标标签、边界或字段含义不清楚，先用普通回复向用户提问并停止。 +5. 信息足够时，调用 `data_agent_prepare_generation_goal` 创建 pending goal，并停止等待用户 review。 +6. 用户确认 generation goal 后，调用 `data_agent_confirm_generation_goal` 获取 `confirmed_goal_id`。 +7. 调用 `data_agent_prepare_generation_plan` 创建 pending 计划，必须传入 `confirmed_goal_id`。 +8. 展示计划后停止本轮，等待用户 review。 +9. 用户提出修改意见时，调用 `data_agent_update_generation_plan`，再展示计划。 +10. 用户明确确认当前计划版本后，调用 `data_agent_confirm_generation_plan`。 +11. 生成 dataset draft text v1。 +12. 调用 `data_agent_normalize_dataset_draft`，必须传入 `confirmed_plan_id`。 +13. 调用 `data_agent_validate_dataset_records`。 +14. 如果用户要求落盘 canonical records，调用 `data_agent_export_dataset_records`，默认导出紧凑 JSONL，不要用 `write_file` 手写 JSON。 +15. 本阶段默认只推进到 canonical metadata records；除非用户另行要求，不做最终训练/评测格式导出。 + +## 数据生成输出格式 + +当需要生成数据样本时，默认使用 dataset draft text v1，不要直接输出 JSON、JSONL、CSV 或最终表格格式。除非用户明确要求机器可读格式，否则优先输出便于人工 review 的文本格式。 + +### 格式 + +```text +# dataset_label: 数据集或专题名称 + +### case: case名称 +用户: 本轮 query +target: Agent(tag="xxx") +notes: 可选，说明覆盖的问题或边界 + +### case: 多轮 case 名称 +用户: 前一轮 query +小爱: 前一轮 tts +用户: 本轮 query +target: Agent(tag="xxx") +notes: 可选，说明覆盖的问题或边界 +``` + +### 规则 + +- 每条数据用一个 `### case:` 开始。 +- `用户:` 表示用户 query。 +- `小爱:` 表示小爱回复 tts。 +- 最后一个 `用户:` 是本轮 query。 +- `target:` 是本轮 query 对应的监督标签，必须存在。 +- 单轮数据只需要写一行 `用户:`，然后写 `target:`。 +- 多轮数据需要按照时间顺序写多组 `用户:` / `小爱:`。 +- 多轮数据的最后一轮只写 `用户:` 和 `target:`，不要写最后一轮 `小爱:`，因为本轮 query 不包含 tts。 +- 如果 `target` 无法确定，不要编造，必须向用户确认。 +- 不要手写 `record_id`、`request_id`、`timestamp`、`context`。 +- 线上挖掘数据如有真实 `request_id` 和 `timestamp`，可以附加在 case 中；没有则不写。 + +## Canonical Record + +`data_agent_normalize_dataset_draft` 会把 dataset draft text 转成 canonical records，并统一补齐 `record_id`、`source`、`timestamp`、`context`、`target_type` 等机械字段。 + +canonical records 落盘必须使用 `data_agent_export_dataset_records`，默认格式是紧凑 JSONL：一行一个 canonical record，不带外层数组，不手写缩进 JSON。只有用户明确要求兼容旧文件时，才使用 `output_format="json"` 导出紧凑 JSON 数组。 + +当前 canonical record v1 工作格式： + +```json +{ + "record_id": "gen_aabbccdd_000001", + "source": { + "type": "generated", + "request_id": "aabbccdd", + "timestamp": 1755567930500 + }, + "turn": { + "query": "本轮 query", + "timestamp": 1755567930500 + }, + "prev_session": [ + { + "query": "前一轮 query", + "tts": "前一轮 tts", + "timestamp": 1755567870500 + } + ], + "context": {}, + "label": { + "dataset_label": "数据集或专题名称", + "target": "Agent(tag=\"xxx\")", + "target_type": "agent" + }, + "meta": { + "case_name": "case名称", + "notes": "" + } +} +``` + +## 场景工作流 + +### 文件定义型 + +1. 使用 `data_agent_load_input_sources` 读取用户提供的文件。 +2. 如果用户只描述了文件名或主题但没给路径，先询问路径，不要猜。 +3. 使用 `data_agent_render_source_context` 文本化输入资料，必要时用 `focus_keywords` 缩小到 query、功能点、标签、badcase 相关内容。 +4. 大模型基于 evidence text 总结 query 语义、功能点、边界、正例、反例、冲突点和缺失假设。 +5. 把标签边界整理为 `generation_goal.target_definitions`。 +6. 如果文档里没有明确 target 格式，向用户确认，例如 `Agent(tag="xxx")` 还是 function 调用格式。 +7. 调用 `data_agent_prepare_generation_goal`，由工具暂停等待用户 review；用户确认后再调用 `data_agent_confirm_generation_goal` 和 `data_agent_prepare_generation_plan`。 + +### 手写规则型 + +1. 直接从用户描述里抽取标签边界。 +2. 多标签边界必须使用 `target_definitions`，不要拆成多个无关单标签计划。 +3. 识别规则中的冲突词、优先级和反例。 +4. 整理为 `generation_goal` 并调用 `data_agent_prepare_generation_goal`。 +5. 对不明确的 target、数量、单轮/多轮比例、输出路径提出问题。 +6. 用户确认 generation goal 后，调用 `data_agent_confirm_generation_goal`，再调用 `data_agent_prepare_generation_plan`。 + +### 示例归纳型 + +1. 先把 example query / badcase 按意图和可能标签分组。 +2. 输出边界归纳和不确定点，不要马上生成数据。 +3. 如果 query 没有明确正确标签，必须向用户确认标签或允许的 target 集合。 +4. 用户确认后，整理为 `generation_goal.target_definitions` 和 `generation_goal.coverage`。 +5. 调用 `data_agent_prepare_generation_goal` 展示 `generation_goal` 给用户 review；用户确认后再调用 `data_agent_confirm_generation_goal` 和 `data_agent_prepare_generation_plan`。 + +## Generation Goal 草案格式 + +大模型完成产品定义或 badcase 分析后，先输出下面的草案给用户 review： + +```json +{ + "dataset_label": "数据集或专题名称", + "goal_summary": "这批数据要解决什么问题", + "target_definitions": [ + { + "name": "标签名称", + "target": "Agent(tag=\"xxx\") 或 function 调用", + "rule": "哪些 query 应该进入这个标签", + "positive_examples": [], + "negative_examples": [], + "boundary_notes": [], + "source_refs": [] + } + ], + "plan_hint": "建议先生成 50 条单轮，输出到 tasks/<数据集名称>/records.jsonl；具体数量、轮次和路径在 generation plan 中确认。", + "coverage": "需要覆盖的 query 语义、功能点、错误类型", + "exclusions": "不要生成或需要排除的表达", + "open_questions": [], + "source_refs": [] +} +``` + +`generation_goal` 只确认“做什么数据、为什么做、标签边界是什么”。不要在 goal 中维护结构化的 `total_count`、`turn_mix` 或 `output_path`；这些字段属于后续 `generation_plan`。如果需要在 goal review 阶段提示执行方向，只写一句 `plan_hint`，例如“建议先生成 50 条单轮，输出到 tasks/.../records.jsonl；具体数量、轮次和路径在 generation plan 中确认”。 + +如果未来增加 `data_agent_validate_generation_goal`，它只做结构校验和缺失字段提示，不做语义判断，不替代用户 review，也不替代 `data_agent_prepare_generation_plan`。 + +现在已经有代码级 goal review 门禁：不要手写 goal 后直接进入 plan，必须先调用 `data_agent_prepare_generation_goal`，并在用户确认后用 `data_agent_confirm_generation_goal` 取得 `confirmed_goal_id`。 + +## 计划中的专用工具 + +当前前链路只保留三个工具。不要再假设有 `data_agent_load_definition_source`、`data_agent_extract_target_definitions` 这类更细工具。 + +### `data_agent_load_input_sources` + +读取目录或文件，统一抽取 `xlsx/csv/docx/pdf/txt/md/json/jsonl` 的段落、表格预览、行数据样例和 source refs。 + +### `data_agent_render_source_context` + +把 `data_agent_load_input_sources` 的结构化结果渲染成大模型可读的 evidence text。产品定义/PRD/走查文档的语义理解应该基于这个文本由大模型完成，不要依赖程序规则直接抽语义。 + +### `data_agent_extract_case_evidence` + +从 badcase、评测表、走查表里识别 query、上下文、预期标签、模型预测、类型和备注。字段不明确时，它会返回 `required_questions`，此时必须向用户确认字段含义。 + +## 当前可用工具 + +- `read_file`：读取用户提供的产品定义、标签定义、样例 query 文件。 +- `write_file`：在用户确认后落盘 draft、校验结果或说明文档；不要用它手写 canonical records 文件。 +- `edit_file`：修改已有的计划、说明文档或生成结果文件。 +- `grep_search`：在项目中搜索已有标签定义、历史数据样例或相关文档。 +- `glob_search`：按路径模式查找定义文件、样例文件或历史产物。 +- `ask_user_question`：需要用户明确选择或补充关键信息时使用；如果不可用，就用普通回复提问并停止。 +- `data_agent_load_input_sources`：读取用户给的目录或文件，把 docx/xlsx/pdf 等输入统一抽成段落、表格和 source refs。 +- `data_agent_render_source_context`：把结构化输入渲染成大模型可读文本，支持 `max_chars`、表格行数和关键词过滤，用于后续模型语义抽取。 +- `data_agent_extract_case_evidence`：从 badcase/评测/走查表中抽取 query、预期标签、模型预测、上下文和备注；字段歧义会返回需要确认的问题。 +- `data_agent_prepare_generation_goal`：在已经整理出 `dataset_label`、`target_definitions`、`plan_hint`、`coverage`、`exclusions`、`source_refs` 后，创建待 review 的 generation goal；调用后本轮会暂停等待用户 review。 +- `data_agent_confirm_generation_goal`：用户明确确认 generation goal 后使用，获取 `confirmed_goal_id`。 +- `data_agent_prepare_generation_plan`：在 generation goal 已确认后创建待 review 的生成计划；必须传入 `confirmed_goal_id`，调用后本轮会暂停等待用户 review。 +- `data_agent_show_generation_plan`：用户要求查看当前计划，或继续上下文时需要恢复计划详情时使用。 +- `data_agent_update_generation_plan`：用户对计划提出修改意见后使用，更新计划并重新展示。 +- `data_agent_confirm_generation_plan`：用户明确确认当前计划版本后使用，获取 `confirmed_plan_id`。 +- `data_agent_normalize_dataset_draft`：用户确认计划后，把 dataset draft text v1 转成 canonical records；必须传入 `confirmed_plan_id`。 +- `data_agent_validate_dataset_records`：对 canonical records 做结构、标签、时间戳和多轮上下文校验。 +- `data_agent_export_dataset_records`：校验 canonical records 并落盘；默认写紧凑 JSONL，一行一条，不要再用 `write_file` 手写 records 文件。 + ## 约束 - 不要静默解决产品或标签歧义。 +- 如果 `ask_user_question` 不可用，使用普通回复向用户提问并停止，不要自己替用户确认。 - canonical records 通过校验前，不要生成最终导出格式。 +- canonical records 需要落盘时，必须用 `data_agent_export_dataset_records`；不要自己拼接 JSON/JSONL。 - 除非用户明确要求，否则不要把“修改标签定义”和“生成数据”混在一起做。 +- 不要因为用户说“生成一些数据”就跳过边界总结和 generation plan review。 diff --git a/tests/test_data_agent_inputs.py b/tests/test_data_agent_inputs.py new file mode 100644 index 0000000..12aab24 --- /dev/null +++ b/tests/test_data_agent_inputs.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +import zipfile +from pathlib import Path + +import openpyxl + +from src.agent_tools import build_tool_context, default_tool_registry, execute_tool +from src.agent_types import AgentRuntimeConfig +from src.data_agent_inputs import ( + extract_case_evidence, + load_input_sources, + render_source_context, +) + + +class DataAgentInputTests(unittest.TestCase): + def test_load_input_sources_reads_xlsx_tables(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / 'cases.xlsx' + _write_xlsx(path) + + payload = load_input_sources(tmp_dir, ['cases.xlsx']) + + self.assertEqual(payload['source_count'], 1) + table = payload['sources'][0]['tables'][0] + self.assertEqual(table['title'], 'Sheet') + self.assertEqual(table['rows'][0], ['query', '预期domain', '0106-prev-domain', 'type', '备注']) + + def test_extract_case_evidence_profiles_and_extracts_rows(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / 'cases.xlsx' + _write_xlsx(path) + + payload = extract_case_evidence(tmp_dir, paths=['cases.xlsx']) + + self.assertEqual(payload['required_questions'], []) + self.assertEqual(payload['evidence_count'], 2) + self.assertEqual(payload['evidence'][0]['query'], '怎么开启查找设备') + self.assertEqual(payload['evidence'][0]['expected_label'], 'productAgent') + self.assertEqual(payload['evidence'][0]['predicted_label'], 'QA') + + def test_extract_case_evidence_asks_when_expected_label_is_missing(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / 'ambiguous.xlsx' + workbook = openpyxl.Workbook() + sheet = workbook.active + sheet.append(['query', '备注']) + sheet.append(['怎么开启查找设备', '需要产品问答']) + workbook.save(path) + + payload = extract_case_evidence(tmp_dir, paths=['ambiguous.xlsx']) + + self.assertEqual(payload['evidence_count'], 0) + self.assertTrue(any('预期标签列' in question for question in payload['required_questions'])) + + def test_render_source_context_reads_docx_tables(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / 'prd.docx' + _write_minimal_docx(path) + + payload = render_source_context(tmp_dir, paths=['prd.docx']) + + self.assertFalse(payload['truncated']) + self.assertIn('complex_task(tag="设备控制")', payload['context_text']) + self.assertIn('query示例: 我到家了布置下客厅灯光', payload['context_text']) + + def test_tools_execute_against_registry(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / 'cases.xlsx' + _write_xlsx(path) + context = build_tool_context(AgentRuntimeConfig(cwd=Path(tmp_dir))) + registry = default_tool_registry() + + load_result = execute_tool( + registry, + 'data_agent_load_input_sources', + {'paths': ['cases.xlsx'], 'max_rows_per_table': 5}, + context, + ) + self.assertTrue(load_result.ok, load_result.content) + loaded = json.loads(load_result.content) + + evidence_result = execute_tool( + registry, + 'data_agent_extract_case_evidence', + {'loaded_sources': loaded}, + context, + ) + render_result = execute_tool( + registry, + 'data_agent_render_source_context', + {'loaded_sources': loaded, 'max_chars': 2000}, + context, + ) + + self.assertTrue(evidence_result.ok, evidence_result.content) + evidence = json.loads(evidence_result.content) + self.assertEqual(evidence['evidence_count'], 2) + self.assertTrue(render_result.ok, render_result.content) + rendered = json.loads(render_result.content) + self.assertIn('怎么开启查找设备', rendered['context_text']) + + +def _write_xlsx(path: Path) -> None: + workbook = openpyxl.Workbook() + sheet = workbook.active + sheet.append(['query', '预期domain', '0106-prev-domain', 'type', '备注']) + sheet.append(['怎么开启查找设备', 'productAgent', 'QA', '设置问答', '预期落产品问答']) + sheet.append(['查找设备打开了吗', 'productAgent', 'smartApp', '状态问答', '']) + workbook.save(path) + + +def _write_minimal_docx(path: Path) -> None: + document_xml = ''' + + + 中控tag：complex_task(tag="设备控制") + + + 意图类别 + query示例 + 高置信品类 + + + 设备控制 + 我到家了布置下客厅灯光 + light + + + + +''' + with zipfile.ZipFile(path, 'w') as archive: + archive.writestr('word/document.xml', document_xml) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_data_agent_records.py b/tests/test_data_agent_records.py index bfcdd73..28982ec 100644 --- a/tests/test_data_agent_records.py +++ b/tests/test_data_agent_records.py @@ -8,9 +8,12 @@ from pathlib import Path from src.agent_tools import build_tool_context, default_tool_registry, execute_tool from src.agent_types import AgentRuntimeConfig from src.data_agent_records import ( + confirm_generation_goal, confirm_generation_plan, + export_dataset_records, get_generation_plan, normalize_dataset_draft, + prepare_generation_goal, prepare_generation_plan, update_generation_plan, validate_dataset_records, @@ -94,6 +97,32 @@ target: Agent(tag="life_service") self.assertFalse(result['ok']) self.assertEqual(result['errors'][0]['path'], 'records[0].turn.tts') + def test_export_dataset_records_writes_compact_jsonl(self) -> None: + records = normalize_dataset_draft( + ''' +# dataset_label: 地图和生活边界数据 +### case: 附近餐饮查询 +用户: 附近有什么好吃的 +target: Agent(tag="life_service") +'''.strip(), + batch_id='demo', + base_timestamp=1_755_567_930_500, + )['records'] + with tempfile.TemporaryDirectory() as tmp_dir: + payload = export_dataset_records( + records, + root=tmp_dir, + output_path='tasks/demo/records.jsonl', + ) + output_path = Path(tmp_dir) / payload['output_path'] + lines = output_path.read_text(encoding='utf-8').splitlines() + + self.assertEqual(payload['output_format'], 'jsonl') + self.assertEqual(payload['record_count'], 1) + self.assertEqual(len(lines), 1) + self.assertNotIn(': ', lines[0]) + self.assertEqual(json.loads(lines[0])['turn']['query'], '附近有什么好吃的') + def test_normalize_online_draft_uses_real_request_metadata(self) -> None: records = normalize_dataset_draft( ''' @@ -151,6 +180,55 @@ target: Agent(tag="life_service") self.assertEqual(records[0]['label']['dataset_label'], '地图和生活边界数据') + def test_generation_goal_can_gate_plan_creation(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + goal_payload = prepare_generation_goal( + root=tmp_dir, + dataset_label='地图和生活边界数据', + goal_summary='生成附近生活服务边界数据', + target='Agent(tag="life_service")', + plan_hint='建议先生成 2 条单轮，输出到 tasks/demo/records.jsonl', + coverage='附近吃喝玩乐', + exclusions='不要生成导航路线类 query', + source_refs=['manual:user'], + ) + goal_id = goal_payload['goal']['goal_id'] + with self.assertRaisesRegex(Exception, 'generation goal is not confirmed'): + prepare_generation_plan( + root=tmp_dir, + confirmed_goal_id=goal_id, + dataset_label='地图和生活边界数据', + target='Agent(tag="life_service")', + total_count=2, + turn_mix='2 条单轮', + coverage='附近吃喝玩乐', + exclusions='不要生成导航路线类 query', + output_path='tasks/demo/artifacts', + ) + confirmed_goal = confirm_generation_goal( + root=tmp_dir, + goal_id=goal_id, + confirmation='确认 goal', + reviewed_revision=1, + ) + plan_payload = prepare_generation_plan( + root=tmp_dir, + confirmed_goal_id=confirmed_goal['confirmed_goal_id'], + dataset_label='地图和生活边界数据', + target='Agent(tag="life_service")', + total_count=2, + turn_mix='2 条单轮', + coverage='附近吃喝玩乐', + exclusions='不要生成导航路线类 query', + output_path='tasks/demo/artifacts', + ) + + self.assertEqual(plan_payload['plan']['confirmed_goal_id'], goal_id) + self.assertEqual( + goal_payload['goal']['plan_hint'], + '建议先生成 2 条单轮，输出到 tasks/demo/records.jsonl', + ) + def test_generation_plan_dedupes_existing_task_output_path(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: existing = Path(tmp_dir) / 'tasks' / 'demo' / 'artifacts' @@ -290,10 +368,48 @@ target: Agent(tag="天气") def test_tools_execute_against_registry(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: context = build_tool_context(AgentRuntimeConfig(cwd=Path(tmp_dir))) + goal_result = execute_tool( + default_tool_registry(), + 'data_agent_prepare_generation_goal', + { + 'dataset_label': '地图和生活边界数据', + 'goal_summary': '生成附近生活服务边界数据', + 'target': 'Agent(tag="life_service")', + 'coverage': '附近吃喝玩乐', + 'exclusions': '不要生成导航路线类 query', + 'source_refs': ['manual:user'], + }, + context, + ) + self.assertTrue(goal_result.ok, goal_result.content) + goal_id = json.loads(goal_result.content)['goal']['goal_id'] + blocked_plan_result = execute_tool( + default_tool_registry(), + 'data_agent_prepare_generation_plan', + { + 'dataset_label': '地图和生活边界数据', + 'target': 'Agent(tag="life_service")', + 'total_count': 1, + 'turn_mix': '1 条单轮', + 'coverage': '附近吃喝玩乐', + 'exclusions': '不要生成导航路线类 query', + 'output_path': 'tasks/demo/artifacts', + }, + context, + ) + self.assertFalse(blocked_plan_result.ok) + confirmed_goal_result = execute_tool( + default_tool_registry(), + 'data_agent_confirm_generation_goal', + {'goal_id': goal_id, 'confirmation': '确认 goal', 'reviewed_revision': 1}, + context, + ) + self.assertTrue(confirmed_goal_result.ok, confirmed_goal_result.content) plan_result = execute_tool( default_tool_registry(), 'data_agent_prepare_generation_plan', { + 'confirmed_goal_id': goal_id, 'dataset_label': '地图和生活边界数据', 'target': 'Agent(tag="life_service")', 'total_count': 1, @@ -350,9 +466,27 @@ target: Agent(tag="life_service") {'records': records}, context, ) + export_result = execute_tool( + default_tool_registry(), + 'data_agent_export_dataset_records', + { + 'records': records, + 'output_path': 'tasks/demo/records.jsonl', + }, + context, + ) + export_payload = json.loads(export_result.content) if export_result.ok else {} + exported_lines = ( + (Path(tmp_dir) / export_payload['output_path']).read_text(encoding='utf-8').splitlines() + if export_result.ok + else [] + ) self.assertTrue(validate_result.ok) self.assertTrue(json.loads(validate_result.content)['ok']) + self.assertTrue(export_result.ok, export_result.content) + self.assertEqual(export_payload['record_count'], 1) + self.assertEqual(len(exported_lines), 1) if __name__ == '__main__':