diff --git a/src/agent_tools.py b/src/agent_tools.py index f0e6b8a..f7be540 100644 --- a/src/agent_tools.py +++ b/src/agent_tools.py @@ -35,6 +35,13 @@ from .data_agent_inputs import ( load_input_sources, render_source_context, ) +from .data_agent_router_sessions import ( + DataAgentRouterSessionError, + convert_router_candidates_to_records, + profile_router_sessions, + sample_router_candidates, + search_router_sessions, +) from .session_env_vars import get_session_env_vars if TYPE_CHECKING: @@ -1466,6 +1473,137 @@ def default_tool_registry() -> dict[str, AgentTool]: }, handler=_data_agent_render_source_context_tool, ), + AgentTool( + name='data_agent_profile_router_sessions', + description=( + 'Profile local router_session_parquet date partitions before mining online sessions. ' + 'Returns schema, sampled row count, distributions, and example sessions.' + ), + parameters={ + 'type': 'object', + 'properties': { + 'dates': { + 'type': 'array', + 'items': {'type': 'string'}, + 'description': 'Date partitions such as ["20260428"]. Resolved under router_session_parquet/date=YYYYMMDD.', + }, + 'paths': { + 'type': 'array', + 'items': {'type': 'string'}, + 'description': 'Workspace-relative parquet files or date directories. Use when dates is not enough.', + }, + 'max_files': {'type': 'integer', 'minimum': 1, 'maximum': 200}, + 'max_rows_per_file': {'type': 'integer', 'minimum': 1, 'maximum': 100000}, + 'examples_limit': {'type': 'integer', 'minimum': 0, 'maximum': 20}, + }, + }, + handler=_data_agent_profile_router_sessions_tool, + ), + AgentTool( + name='data_agent_search_router_sessions', + description=( + 'Search local router_session_parquet data with reviewable filters such as device, domain, intent, func, ' + 'query keywords, regex, turn count, and match scope. Returns turn-level candidates with prev turns.' + ), + parameters={ + 'type': 'object', + 'properties': { + 'dates': { + 'type': 'array', + 'items': {'type': 'string'}, + 'description': 'Date partitions such as ["20260428"]. Resolved under router_session_parquet/date=YYYYMMDD.', + }, + 'paths': { + 'type': 'array', + 'items': {'type': 'string'}, + 'description': 'Workspace-relative parquet files or date directories. Use when dates is not enough.', + }, + 'devices': {'type': 'array', 'items': {'type': 'string'}}, + 'domains': {'type': 'array', 'items': {'type': 'string'}}, + 'intents': {'type': 'array', 'items': {'type': 'string'}}, + 'funcs': {'type': 'array', 'items': {'type': 'string'}}, + 'query_keywords': {'type': 'array', 'items': {'type': 'string'}}, + 'keyword_match_mode': {'type': 'string', 'enum': ['any', 'all']}, + 'query_regex': {'type': 'string'}, + 'match_scope': {'type': 'string', 'enum': ['any_turn', 'last_turn']}, + 'turn_count_min': {'type': 'integer', 'minimum': 1}, + 'turn_count_max': {'type': 'integer', 'minimum': 1}, + 'max_files': {'type': 'integer', 'minimum': 1, 'maximum': 200}, + 'max_rows_per_file': {'type': 'integer', 'minimum': 1, 'maximum': 100000}, + 'max_candidates': {'type': 'integer', 'minimum': 1, 'maximum': 5000}, + }, + }, + handler=_data_agent_search_router_sessions_tool, + ), + AgentTool( + name='data_agent_sample_router_candidates', + description=( + 'Sample router session candidates for human review and return candidate-level summary statistics. ' + 'Use after data_agent_search_router_sessions.' + ), + parameters={ + 'type': 'object', + 'properties': { + 'candidates': { + 'oneOf': [ + {'type': 'array'}, + {'type': 'object'}, + {'type': 'string'}, + ], + 'description': 'Candidates array, search result object, or JSON string containing candidates.', + }, + 'sample_size': {'type': 'integer', 'minimum': 1, 'maximum': 1000}, + 'strategy': {'type': 'string', 'enum': ['random', 'first', 'stride']}, + 'seed': {'type': 'integer'}, + }, + 'required': ['candidates'], + }, + handler=_data_agent_sample_router_candidates_tool, + ), + AgentTool( + name='data_agent_convert_router_candidates_to_records', + description=( + 'Convert reviewed online router session candidates directly into canonical data-agent records. ' + 'Use this when the user wants to keep mined online data as samples; do not use generation-plan tools for this path.' + ), + parameters={ + 'type': 'object', + 'properties': { + 'candidates': { + 'oneOf': [ + {'type': 'array'}, + {'type': 'object'}, + {'type': 'string'}, + ], + 'description': 'Candidates array, search/sample result object, or JSON string containing candidates.', + }, + 'dataset_label': {'type': 'string'}, + 'default_target': { + 'type': 'string', + 'description': 'Target label used for included candidates unless review_decisions provides a target.', + }, + 'review_decisions': { + 'type': 'array', + 'description': 'Optional include/exclude/uncertain decisions keyed by semantic_session_id or req_id.', + 'items': { + 'type': 'object', + 'properties': { + 'semantic_session_id': {'type': 'string'}, + 'req_id': {'type': 'string'}, + 'matched_turn_index': {'type': 'integer'}, + 'decision': {'type': 'string', 'enum': ['include', 'exclude', 'uncertain']}, + 'target': {'type': 'string'}, + 'notes': {'type': 'string'}, + }, + }, + }, + 'batch_id': {'type': 'string'}, + 'include_uncertain': {'type': 'boolean'}, + }, + 'required': ['candidates', 'dataset_label'], + }, + handler=_data_agent_convert_router_candidates_to_records_tool, + ), AgentTool( name='data_agent_show_generation_plan', description='Show a data-agent generation plan for human review.', @@ -1689,6 +1827,15 @@ def _coerce_int(arguments: dict[str, Any], key: str, default: int) -> int: return value +def _optional_int(arguments: dict[str, Any], key: str) -> int | None: + value = arguments.get(key) + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int): + raise ToolExecutionError(f'{key} must be an integer') + return value + + def _coerce_float(arguments: dict[str, Any], key: str, default: float) -> float: value = arguments.get(key, default) if isinstance(value, bool) or not isinstance(value, (int, float)): @@ -1855,6 +2002,88 @@ def _data_agent_render_source_context_tool(arguments: dict[str, Any], context: T return data_agent_dumps_payload(payload) +def _data_agent_profile_router_sessions_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str: + try: + payload = profile_router_sessions( + context.root, + dates=_optional_string_list(arguments, 'dates'), + paths=_optional_string_list(arguments, 'paths'), + max_files=_coerce_int(arguments, 'max_files', 8), + max_rows_per_file=_coerce_int(arguments, 'max_rows_per_file', 2000), + examples_limit=_coerce_int(arguments, 'examples_limit', 3), + ) + except DataAgentRouterSessionError as exc: + raise ToolExecutionError(str(exc)) from exc + return data_agent_dumps_payload(payload) + + +def _data_agent_search_router_sessions_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str: + try: + payload = search_router_sessions( + context.root, + dates=_optional_string_list(arguments, 'dates'), + paths=_optional_string_list(arguments, 'paths'), + devices=_optional_string_list(arguments, 'devices'), + domains=_optional_string_list(arguments, 'domains'), + intents=_optional_string_list(arguments, 'intents'), + funcs=_optional_string_list(arguments, 'funcs'), + query_keywords=_optional_string_list(arguments, 'query_keywords'), + keyword_match_mode=_optional_string(arguments, 'keyword_match_mode') or 'any', + query_regex=_optional_string(arguments, 'query_regex'), + match_scope=_optional_string(arguments, 'match_scope') or 'any_turn', + turn_count_min=_optional_int(arguments, 'turn_count_min'), + turn_count_max=_optional_int(arguments, 'turn_count_max'), + max_files=_coerce_int(arguments, 'max_files', 20), + max_rows_per_file=_coerce_int(arguments, 'max_rows_per_file', 5000), + max_candidates=_coerce_int(arguments, 'max_candidates', 200), + ) + except DataAgentRouterSessionError as exc: + raise ToolExecutionError(str(exc)) from exc + return data_agent_dumps_payload(payload) + + +def _data_agent_sample_router_candidates_tool(arguments: dict[str, Any], _context: ToolExecutionContext) -> str: + if 'candidates' not in arguments: + raise ToolExecutionError('candidates is required') + try: + payload = sample_router_candidates( + arguments['candidates'], + sample_size=_coerce_int(arguments, 'sample_size', 100), + strategy=_optional_string(arguments, 'strategy') or 'random', + seed=_coerce_int(arguments, 'seed', 20260430), + ) + except DataAgentRouterSessionError as exc: + raise ToolExecutionError(str(exc)) from exc + return data_agent_dumps_payload(payload) + + +def _data_agent_convert_router_candidates_to_records_tool( + arguments: dict[str, Any], + _context: ToolExecutionContext, +) -> str: + if 'candidates' not in arguments: + raise ToolExecutionError('candidates is required') + review_decisions = arguments.get('review_decisions') + if review_decisions is not None and not isinstance(review_decisions, list): + raise ToolExecutionError('review_decisions must be an array') + include_uncertain = arguments.get('include_uncertain', False) + if not isinstance(include_uncertain, bool): + raise ToolExecutionError('include_uncertain must be a boolean') + try: + payload = convert_router_candidates_to_records( + arguments['candidates'], + dataset_label=_require_string(arguments, 'dataset_label'), + default_target=_optional_string(arguments, 'default_target'), + review_decisions=review_decisions, + batch_id=_optional_string(arguments, 'batch_id') or 'router', + include_uncertain=include_uncertain, + ) + payload['validation'] = validate_dataset_records(payload['records']) + except (DataAgentRouterSessionError, DataRecordError) 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( diff --git a/src/data_agent_records.py b/src/data_agent_records.py index a54ae9e..ff54299 100644 --- a/src/data_agent_records.py +++ b/src/data_agent_records.py @@ -402,8 +402,8 @@ def validate_dataset_records(records: list[dict[str, Any]]) -> dict[str, Any]: continue if not _non_empty_str(item.get('query')): errors.append(_issue(f'{item_prefix}.query', 'query is required')) - if not _non_empty_str(item.get('tts')): - errors.append(_issue(f'{item_prefix}.tts', 'tts is required')) + if 'tts' not in item or not isinstance(item.get('tts'), str): + errors.append(_issue(f'{item_prefix}.tts', 'tts must be a string')) timestamp = item.get('timestamp') if not _int_like(timestamp): errors.append(_issue(f'{item_prefix}.timestamp', 'timestamp must be an integer')) diff --git a/src/data_agent_router_sessions.py b/src/data_agent_router_sessions.py new file mode 100644 index 0000000..e9b045e --- /dev/null +++ b/src/data_agent_router_sessions.py @@ -0,0 +1,739 @@ +from __future__ import annotations + +import json +import random +import re +from collections import Counter +from pathlib import Path +from typing import Any, Iterable + + +class DataAgentRouterSessionError(ValueError): + """路由 session parquet 工具的输入或运行错误。""" + + +DEFAULT_ROUTER_SESSION_ROOT = 'router_session_parquet' + + +def profile_router_sessions( + root: str | Path, + *, + dates: list[str] | None = None, + paths: list[str] | None = None, + max_files: int = 8, + max_rows_per_file: int = 2000, + examples_limit: int = 3, +) -> dict[str, Any]: + """读取路由 session parquet 的小样本概貌，辅助后续制定挖掘策略。""" + + files = _resolve_parquet_files(root, dates=dates, paths=paths, max_files=max_files) + rows_seen = 0 + missing_req_id = 0 + devices: Counter[str] = Counter() + turn_counts: Counter[str] = Counter() + domains: Counter[str] = Counter() + intents: Counter[str] = Counter() + funcs: Counter[str] = Counter() + action_keys: Counter[str] = Counter() + examples: list[dict[str, Any]] = [] + schema = '' + + for parquet_file in files: + parquet = _import_pyarrow_parquet() + if not schema: + schema = str(parquet.ParquetFile(parquet_file).schema) + for row in _iter_parquet_rows(parquet_file, max_rows_per_file=max_rows_per_file): + rows_seen += 1 + session = _normalize_session(row, parquet_file) + if not session.get('req_id'): + missing_req_id += 1 + devices[_counter_key(session.get('device'))] += 1 + turn_counts[str(session.get('turn_count', 0))] += 1 + for turn in session.get('turns', []): + action = turn.get('action') or {} + for key in action: + action_keys[key] += 1 + domains[_counter_key(action.get('domain'))] += 1 + intents[_counter_key(action.get('intent'))] += 1 + if action.get('func') is not None: + funcs[_counter_key(action.get('func'))] += 1 + if len(examples) < examples_limit: + examples.append(_session_example(session)) + + return { + 'ok': True, + 'action': 'data_agent_profile_router_sessions', + 'source': { + 'dates': dates or [], + 'paths': [str(item) for item in files], + 'file_count': len(files), + 'max_rows_per_file': max_rows_per_file, + }, + 'schema': schema, + 'sampled_row_count': rows_seen, + 'missing_req_id_count': missing_req_id, + 'distributions': { + 'device': _top_counts(devices), + 'turn_count': _top_counts(turn_counts), + 'domain': _top_counts(domains), + 'intent': _top_counts(intents), + 'func': _top_counts(funcs), + 'action_json_keys': _top_counts(action_keys), + }, + 'examples': examples, + } + + +def search_router_sessions( + root: str | Path, + *, + dates: list[str] | None = None, + paths: list[str] | None = None, + devices: list[str] | None = None, + domains: list[str] | None = None, + intents: list[str] | None = None, + funcs: list[str] | None = None, + query_keywords: list[str] | None = None, + keyword_match_mode: str = 'any', + query_regex: str = '', + match_scope: str = 'any_turn', + turn_count_min: int | None = None, + turn_count_max: int | None = None, + max_files: int = 20, + max_rows_per_file: int = 5000, + max_candidates: int = 200, +) -> dict[str, Any]: + """按可 review 的策略召回路由 session 候选，输出 turn 级候选。""" + + if match_scope not in {'any_turn', 'last_turn'}: + raise DataAgentRouterSessionError('match_scope must be any_turn or last_turn') + if keyword_match_mode not in {'any', 'all'}: + raise DataAgentRouterSessionError('keyword_match_mode must be any or all') + regex = _compile_regex(query_regex) + files = _resolve_parquet_files(root, dates=dates, paths=paths, max_files=max_files) + candidate_rows: list[dict[str, Any]] = [] + scanned_rows = 0 + scanned_turns = 0 + + filters = { + 'devices': devices or [], + 'domains': domains or [], + 'intents': intents or [], + 'funcs': funcs or [], + 'query_keywords': query_keywords or [], + 'keyword_match_mode': keyword_match_mode, + 'query_regex': query_regex, + 'match_scope': match_scope, + 'turn_count_min': turn_count_min, + 'turn_count_max': turn_count_max, + } + + for parquet_file in files: + for row in _iter_parquet_rows(parquet_file, max_rows_per_file=max_rows_per_file): + scanned_rows += 1 + session = _normalize_session(row, parquet_file) + if not _session_matches(session, devices, turn_count_min, turn_count_max): + continue + turn_indexes = _candidate_turn_indexes(session.get('turns', []), match_scope) + for turn_index in turn_indexes: + scanned_turns += 1 + turn = session['turns'][turn_index] + if not _turn_matches( + turn, + domains=domains, + intents=intents, + funcs=funcs, + query_keywords=query_keywords, + keyword_match_mode=keyword_match_mode, + regex=regex, + ): + continue + candidate_rows.append(_candidate_from_session(session, turn_index)) + if len(candidate_rows) >= max_candidates: + return _search_payload(files, filters, scanned_rows, scanned_turns, candidate_rows, truncated=True) + + return _search_payload(files, filters, scanned_rows, scanned_turns, candidate_rows, truncated=False) + + +def sample_router_candidates( + candidates: list[dict[str, Any]] | dict[str, Any] | str, + *, + sample_size: int = 100, + strategy: str = 'random', + seed: int = 20260430, +) -> dict[str, Any]: + """从候选集中做稳定抽样，并给出用于人工 review 的基础统计。""" + + candidate_rows = _extract_candidates(candidates) + if sample_size < 1: + raise DataAgentRouterSessionError('sample_size must be greater than 0') + if strategy not in {'random', 'first', 'stride'}: + raise DataAgentRouterSessionError('strategy must be random, first, or stride') + + if strategy == 'first': + sampled = candidate_rows[:sample_size] + elif strategy == 'stride': + sampled = _stride_sample(candidate_rows, sample_size) + else: + sampled = list(candidate_rows) + random.Random(seed).shuffle(sampled) + sampled = sampled[:sample_size] + + return { + 'ok': True, + 'action': 'data_agent_sample_router_candidates', + 'total_candidate_count': len(candidate_rows), + 'sampled_count': len(sampled), + 'strategy': strategy, + 'seed': seed if strategy == 'random' else None, + 'summary': _candidate_summary(candidate_rows), + 'sample_summary': _candidate_summary(sampled), + 'candidates': sampled, + } + + +def convert_router_candidates_to_records( + candidates: list[dict[str, Any]] | dict[str, Any] | str, + *, + dataset_label: str, + default_target: str = '', + review_decisions: list[dict[str, Any]] | None = None, + batch_id: str = 'router', + include_uncertain: bool = False, +) -> dict[str, Any]: + """把 review 后的线上候选直接转换为 canonical records，不经过数据生成流程。""" + + if not dataset_label.strip(): + raise DataAgentRouterSessionError('dataset_label is required') + if not batch_id.strip(): + raise DataAgentRouterSessionError('batch_id is required') + + candidate_rows = _extract_candidates(candidates) + decisions = _normalize_review_decisions(review_decisions or []) + records: list[dict[str, Any]] = [] + skipped: list[dict[str, Any]] = [] + + for index, candidate in enumerate(candidate_rows, 1): + decision = _decision_for_candidate(decisions, candidate) + status = str(decision.get('decision') or 'include').strip().lower() + if status in {'exclude', 'drop', 'reject'}: + skipped.append(_skip_item(candidate, status, decision.get('notes'))) + continue + if status in {'uncertain', 'unknown', '不确定'} and not include_uncertain: + skipped.append(_skip_item(candidate, status, decision.get('notes'))) + continue + + target = str(decision.get('target') or default_target).strip() + if not target: + raise DataAgentRouterSessionError( + f'target is required for candidate {candidate.get("semantic_session_id")}#{candidate.get("matched_turn_index")}' + ) + records.append( + _router_candidate_to_record( + candidate, + dataset_label=str(decision.get('dataset_label') or dataset_label).strip(), + target=target, + notes=str(decision.get('notes') or '').strip(), + batch_id=batch_id, + index=len(records) + 1, + ) + ) + + return { + 'ok': True, + 'action': 'data_agent_convert_router_candidates_to_records', + 'input_candidate_count': len(candidate_rows), + 'record_count': len(records), + 'skipped_count': len(skipped), + 'skipped': skipped, + 'records': records, + } + + +def _resolve_parquet_files( + root: str | Path, + *, + dates: list[str] | None, + paths: list[str] | None, + max_files: int, +) -> list[Path]: + if max_files < 1: + raise DataAgentRouterSessionError('max_files must be greater than 0') + root_path = Path(root).resolve() + requested: list[Path] = [] + if dates: + for date in dates: + if not re.fullmatch(r'\d{8}', date): + raise DataAgentRouterSessionError(f'invalid date: {date}') + requested.append(root_path / DEFAULT_ROUTER_SESSION_ROOT / f'date={date}') + if paths: + for raw_path in paths: + if not isinstance(raw_path, str) or not raw_path.strip(): + raise DataAgentRouterSessionError('paths items must be non-empty strings') + expanded = Path(raw_path).expanduser() + requested.append(expanded if expanded.is_absolute() else root_path / expanded) + if not requested: + raise DataAgentRouterSessionError('dates or paths is required') + + files: list[Path] = [] + for path in requested: + resolved = path.resolve() + try: + resolved.relative_to(root_path) + except ValueError as exc: + raise DataAgentRouterSessionError(f'path escapes workspace root: {path}') from exc + if not resolved.exists(): + raise DataAgentRouterSessionError(f'path not found: {path}') + if resolved.is_dir(): + files.extend(sorted(item for item in resolved.glob('*.parquet') if item.is_file())) + elif resolved.suffix == '.parquet': + files.append(resolved) + else: + raise DataAgentRouterSessionError(f'path is not a parquet file or directory: {path}') + + deduped = [] + seen: set[Path] = set() + for file_path in files: + if file_path in seen: + continue + seen.add(file_path) + deduped.append(file_path) + if not deduped: + raise DataAgentRouterSessionError('no parquet files found') + return deduped[:max_files] + + +def _iter_parquet_rows(path: Path, *, max_rows_per_file: int) -> Iterable[dict[str, Any]]: + if max_rows_per_file < 1: + raise DataAgentRouterSessionError('max_rows_per_file must be greater than 0') + parquet = _import_pyarrow_parquet() + parquet_file = parquet.ParquetFile(path) + emitted = 0 + batch_size = min(1024, max_rows_per_file) + for batch in parquet_file.iter_batches(batch_size=batch_size): + for row in batch.to_pylist(): + yield row + emitted += 1 + if emitted >= max_rows_per_file: + return + + +def _import_pyarrow_parquet() -> Any: + try: + import pyarrow.parquet as parquet + except ImportError as exc: + raise DataAgentRouterSessionError('读取 router session parquet 需要安装 pyarrow') from exc + return parquet + + +def _normalize_session(row: dict[str, Any], source_path: Path) -> dict[str, Any]: + turns = row.get('turns_array') or [] + normalized_turns = [] + for turn in turns: + if not isinstance(turn, dict): + continue + action, parse_error = _parse_action(turn.get('action_json')) + normalized_turns.append( + { + 'timestamp': turn.get('timestamp'), + 'query': str(turn.get('query') or ''), + 'tts': str(turn.get('tts') or ''), + 'action': action, + 'action_parse_error': parse_error, + } + ) + return { + 'semantic_session_id': row.get('semantic_session_id'), + 'device': row.get('device'), + 'req_id': row.get('req_id'), + 'turn_count': int(row.get('turn_count') or len(normalized_turns)), + 'turns': normalized_turns, + 'source_path': str(source_path), + 'date': _date_from_path(source_path), + } + + +def _parse_action(raw: Any) -> tuple[dict[str, Any], str | None]: + if not raw: + return {}, None + if isinstance(raw, dict): + return raw, None + if not isinstance(raw, str): + return {}, f'unsupported action_json type: {type(raw).__name__}' + try: + payload = json.loads(raw) + except json.JSONDecodeError as exc: + return {}, str(exc) + if not isinstance(payload, dict): + return {}, 'action_json is not an object' + return payload, None + + +def _date_from_path(path: Path) -> str: + for part in path.parts: + if part.startswith('date='): + return part.split('=', 1)[1] + return '' + + +def _session_matches( + session: dict[str, Any], + devices: list[str] | None, + turn_count_min: int | None, + turn_count_max: int | None, +) -> bool: + if devices and session.get('device') not in set(devices): + return False + turn_count = int(session.get('turn_count') or 0) + if turn_count_min is not None and turn_count < turn_count_min: + return False + if turn_count_max is not None and turn_count > turn_count_max: + return False + return True + + +def _turn_matches( + turn: dict[str, Any], + *, + domains: list[str] | None, + intents: list[str] | None, + funcs: list[str] | None, + query_keywords: list[str] | None, + keyword_match_mode: str, + regex: re.Pattern[str] | None, +) -> bool: + action = turn.get('action') or {} + if domains and action.get('domain') not in set(domains): + return False + if intents and action.get('intent') not in set(intents): + return False + if funcs and action.get('func') not in set(funcs): + return False + query = str(turn.get('query') or '') + keywords = [keyword for keyword in query_keywords or [] if keyword] + if keywords: + hits = [keyword in query for keyword in keywords] + if keyword_match_mode == 'all' and not all(hits): + return False + if keyword_match_mode == 'any' and not any(hits): + return False + if regex and not regex.search(query): + return False + return True + + +def _compile_regex(pattern: str) -> re.Pattern[str] | None: + if not pattern: + return None + try: + return re.compile(pattern) + except re.error as exc: + raise DataAgentRouterSessionError(f'invalid query_regex: {exc}') from exc + + +def _candidate_turn_indexes(turns: list[dict[str, Any]], match_scope: str) -> list[int]: + if not turns: + return [] + if match_scope == 'last_turn': + return [len(turns) - 1] + return list(range(len(turns))) + + +def _candidate_from_session(session: dict[str, Any], turn_index: int) -> dict[str, Any]: + turns = session.get('turns') or [] + turn = turns[turn_index] + return { + 'semantic_session_id': session.get('semantic_session_id'), + 'req_id': session.get('req_id'), + 'device': session.get('device'), + 'turn_count': session.get('turn_count'), + 'matched_turn_index': turn_index, + 'matched_turn': turn, + 'prev_turns': [ + { + 'timestamp': prev.get('timestamp'), + 'query': prev.get('query'), + 'tts': prev.get('tts'), + 'action': prev.get('action'), + } + for prev in turns[max(0, turn_index - 10) : turn_index] + ], + 'source_path': session.get('source_path'), + 'date': session.get('date'), + } + + +def _search_payload( + files: list[Path], + filters: dict[str, Any], + scanned_rows: int, + scanned_turns: int, + candidates: list[dict[str, Any]], + *, + truncated: bool, +) -> dict[str, Any]: + return { + 'ok': True, + 'action': 'data_agent_search_router_sessions', + 'source': { + 'paths': [str(item) for item in files], + 'file_count': len(files), + }, + 'filters': filters, + 'scanned_row_count': scanned_rows, + 'scanned_turn_count': scanned_turns, + 'candidate_count': len(candidates), + 'truncated': truncated, + 'summary': _candidate_summary(candidates), + 'candidates': candidates, + } + + +def _extract_candidates(candidates: list[dict[str, Any]] | dict[str, Any] | str) -> list[dict[str, Any]]: + payload = candidates + if isinstance(payload, str): + try: + payload = json.loads(payload) + except json.JSONDecodeError as exc: + raise DataAgentRouterSessionError(f'candidates is not valid JSON: {exc}') from exc + if isinstance(payload, dict): + payload = payload.get('candidates') + if not isinstance(payload, list): + raise DataAgentRouterSessionError('candidates must be an array or an object with candidates') + for index, item in enumerate(payload): + if not isinstance(item, dict): + raise DataAgentRouterSessionError(f'candidates[{index}] must be an object') + return payload + + +def _normalize_review_decisions(review_decisions: list[dict[str, Any]]) -> dict[tuple[str, str], dict[str, Any]]: + decisions: dict[tuple[str, str], dict[str, Any]] = {} + if not isinstance(review_decisions, list): + raise DataAgentRouterSessionError('review_decisions must be an array') + for index, item in enumerate(review_decisions): + if not isinstance(item, dict): + raise DataAgentRouterSessionError(f'review_decisions[{index}] must be an object') + key = _decision_key_from_item(item) + if key == ('', ''): + raise DataAgentRouterSessionError( + f'review_decisions[{index}] must include semantic_session_id or req_id' + ) + decisions[key] = item + return decisions + + +def _decision_key_from_item(item: dict[str, Any]) -> tuple[str, str]: + session_id = str(item.get('semantic_session_id') or item.get('session_id') or item.get('req_id') or '').strip() + turn_index = item.get('matched_turn_index') + return session_id, '' if turn_index is None else str(turn_index) + + +def _candidate_key(candidate: dict[str, Any]) -> tuple[str, str]: + session_id = str(candidate.get('semantic_session_id') or candidate.get('req_id') or '').strip() + turn_index = candidate.get('matched_turn_index') + exact_key = (session_id, '' if turn_index is None else str(turn_index)) + if exact_key in {('', '')}: + return exact_key + return exact_key + + +def _decision_for_candidate( + decisions: dict[tuple[str, str], dict[str, Any]], + candidate: dict[str, Any], +) -> dict[str, Any]: + turn_index = candidate.get('matched_turn_index') + turn_key = '' if turn_index is None else str(turn_index) + keys = [ + (str(candidate.get('semantic_session_id') or '').strip(), turn_key), + (str(candidate.get('semantic_session_id') or '').strip(), ''), + (str(candidate.get('req_id') or '').strip(), turn_key), + (str(candidate.get('req_id') or '').strip(), ''), + ] + for key in keys: + if key in decisions: + return decisions[key] + return {} + + +def _router_candidate_to_record( + candidate: dict[str, Any], + *, + dataset_label: str, + target: str, + notes: str, + batch_id: str, + index: int, +) -> dict[str, Any]: + matched_turn = candidate.get('matched_turn') + if not isinstance(matched_turn, dict): + raise DataAgentRouterSessionError(f'candidate {index} matched_turn is required') + query = str(matched_turn.get('query') or '').strip() + if not query: + raise DataAgentRouterSessionError(f'candidate {index} matched_turn.query is required') + timestamp = _int_or_zero(matched_turn.get('timestamp')) + if timestamp <= 0: + raise DataAgentRouterSessionError(f'candidate {index} matched_turn.timestamp is required') + request_id = str(candidate.get('req_id') or '').strip() + if not request_id: + raise DataAgentRouterSessionError(f'candidate {index} req_id is required') + + action = matched_turn.get('action') if isinstance(matched_turn.get('action'), dict) else {} + prev_session = _prev_turns_to_prev_session(candidate.get('prev_turns')) + return { + 'record_id': f'online_{_safe_token(batch_id)}_{_safe_token(request_id)}_{index:06d}', + 'source': { + 'type': 'online', + 'request_id': request_id, + 'timestamp': timestamp, + }, + 'turn': { + 'query': query, + 'timestamp': timestamp, + }, + 'prev_session': prev_session, + 'context': { + 'semantic_session_id': candidate.get('semantic_session_id'), + 'device': candidate.get('device'), + 'date': candidate.get('date'), + 'turn_count': candidate.get('turn_count'), + 'matched_turn_index': candidate.get('matched_turn_index'), + 'source_path': candidate.get('source_path'), + 'domain': action.get('domain'), + 'intent': action.get('intent'), + 'func': action.get('func'), + 'norm_code': action.get('norm_code'), + }, + 'label': { + 'dataset_label': dataset_label, + 'target': target, + 'target_type': _target_type(target), + }, + 'meta': { + 'case_name': '', + 'notes': notes, + }, + } + + +def _prev_turns_to_prev_session(value: Any) -> list[dict[str, Any]]: + if not isinstance(value, list): + return [] + prev_session: list[dict[str, Any]] = [] + for item in value[-10:]: + if not isinstance(item, dict): + continue + query = str(item.get('query') or '').strip() + if not query: + continue + prev_session.append( + { + 'query': query, + 'tts': str(item.get('tts') or ''), + 'timestamp': _int_or_zero(item.get('timestamp')), + } + ) + return prev_session + + +def _skip_item(candidate: dict[str, Any], decision: str, notes: Any) -> dict[str, Any]: + return { + 'semantic_session_id': candidate.get('semantic_session_id'), + 'req_id': candidate.get('req_id'), + 'matched_turn_index': candidate.get('matched_turn_index'), + 'decision': decision, + 'notes': str(notes or ''), + } + + +def _target_type(target: str) -> str: + if target.startswith('Agent('): + return 'agent' + if target: + return 'function' + return 'unknown' + + +def _safe_token(value: str) -> str: + token = re.sub(r'[^A-Za-z0-9_.-]+', '_', value.strip()) + return token.strip('_.-') or 'item' + + +def _int_or_zero(value: Any) -> int: + if isinstance(value, bool): + return 0 + if isinstance(value, int): + return value + if isinstance(value, float) and value.is_integer(): + return int(value) + if isinstance(value, str) and value.isdigit(): + return int(value) + return 0 + + +def _candidate_summary(candidates: list[dict[str, Any]]) -> dict[str, Any]: + devices: Counter[str] = Counter() + turn_counts: Counter[str] = Counter() + domains: Counter[str] = Counter() + intents: Counter[str] = Counter() + funcs: Counter[str] = Counter() + top_queries: Counter[str] = Counter() + for item in candidates: + devices[_counter_key(item.get('device'))] += 1 + turn_counts[str(item.get('turn_count') or 0)] += 1 + matched_turn = item.get('matched_turn') or {} + if isinstance(matched_turn, dict): + query = str(matched_turn.get('query') or '') + if query: + top_queries[query] += 1 + action = matched_turn.get('action') or {} + if isinstance(action, dict): + domains[_counter_key(action.get('domain'))] += 1 + intents[_counter_key(action.get('intent'))] += 1 + if action.get('func') is not None: + funcs[_counter_key(action.get('func'))] += 1 + return { + 'device': _top_counts(devices), + 'turn_count': _top_counts(turn_counts), + 'domain': _top_counts(domains), + 'intent': _top_counts(intents), + 'func': _top_counts(funcs), + 'top_queries': _top_counts(top_queries, limit=20), + } + + +def _stride_sample(items: list[dict[str, Any]], sample_size: int) -> list[dict[str, Any]]: + if sample_size >= len(items): + return list(items) + if sample_size == 1: + return [items[0]] + step = (len(items) - 1) / (sample_size - 1) + return [items[round(index * step)] for index in range(sample_size)] + + +def _session_example(session: dict[str, Any]) -> dict[str, Any]: + return { + 'semantic_session_id': session.get('semantic_session_id'), + 'req_id': session.get('req_id'), + 'device': session.get('device'), + 'turn_count': session.get('turn_count'), + 'date': session.get('date'), + 'turns': [ + { + 'timestamp': turn.get('timestamp'), + 'query': turn.get('query'), + 'tts': turn.get('tts'), + 'action': turn.get('action'), + } + for turn in session.get('turns', [])[:5] + ], + } + + +def _top_counts(counter: Counter[str], *, limit: int = 20) -> list[dict[str, Any]]: + return [{'value': value, 'count': count} for value, count in counter.most_common(limit)] + + +def _counter_key(value: Any) -> str: + if value is None or value == '': + return '' + return str(value) diff --git a/src/skills/bundled/online-badcase-mining/SKILL.md b/src/skills/bundled/online-badcase-mining/SKILL.md index 4f8c11e..edbf615 100644 --- a/src/skills/bundled/online-badcase-mining/SKILL.md +++ b/src/skills/bundled/online-badcase-mining/SKILL.md @@ -3,11 +3,43 @@ name: online-badcase-mining description: 分析 badcase，并通过可 review 的策略迭代挖掘线上相似 case。 when_to_use: 当用户提供线上 badcase 或产品/标签定义，并希望查找相似线上问题或构建专项集时使用。 aliases: badcase-mining, online-mining -allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_user_question, mcp_list_tools, mcp_call_tool +allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_user_question, data_agent_profile_router_sessions, data_agent_search_router_sessions, data_agent_sample_router_candidates, data_agent_convert_router_candidates_to_records, data_agent_export_dataset_records --- 使用这个 skill 处理“badcase 或标签定义 -> 挖掘策略 -> 候选召回 -> 抽样 review -> 策略迭代 -> mined dataset”的工作流。 +## 两条分支 + +线上挖掘后必须先判断用户要的是哪条分支。 + +### 分支 A：线上候选直接作为数据样本 + +当用户说“把筛选出的数据变成样本”“拿这些线上数据做评测集”“保留这批线上 case”“导出候选样本”时，走这个分支。 + +流程： + +1. 用 `data_agent_profile_router_sessions` 查看数据概貌。 +2. 用 `data_agent_search_router_sessions` 按策略召回线上候选。 +3. 用 `data_agent_sample_router_candidates` 抽样展示给用户 review。 +4. 根据用户 review 意见形成 include/exclude/uncertain 决策和目标标签。 +5. 用 `data_agent_convert_router_candidates_to_records` 把线上候选直接转换为 canonical records。 +6. 如果用户要求落盘，用 `data_agent_export_dataset_records` 导出紧凑 JSONL。 + +这个分支不生成新 query，不调用数据生成计划工具，不调用 dataset draft 归一化工具。 + +### 分支 B：基于线上问题再生成补充数据 + +当用户明确说“生成/扩写/构造/造一批类似 case/补充训练数据”时，才走这个分支。 + +流程： + +1. 先完成线上召回、抽样和 review。 +2. 总结线上问题模式和需要覆盖的边界。 +3. 再切换到数据生成流程，创建待 review 的 generation goal 和 generation plan。 +4. 用户确认后才生成 draft，并转换为 canonical records。 + +如果用户只是要求“把线上数据作为样本”，不要进入这个分支。 + ## 输入假设 用户可能会提供： @@ -51,15 +83,16 @@ allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_u - `grep_search` - `glob_search` - `ask_user_question` +- `data_agent_profile_router_sessions`：读取 `router_session_parquet/date=YYYYMMDD/` 的小样本概貌，查看 schema、设备、轮次、domain、intent、func 等分布。 +- `data_agent_search_router_sessions`：按日期、设备、domain、intent、func、query 关键词/正则、轮次数等条件召回线上 session 候选；输出命中 turn、前文 turn、req_id 和 action_json 解析结果。 +- `data_agent_sample_router_candidates`：对召回候选做稳定抽样，并输出 review 需要的基础统计。 +- `data_agent_convert_router_candidates_to_records`：把 review 后的线上候选直接转换为 canonical records；用于“线上数据作为样本”的分支。 +- `data_agent_export_dataset_records`：后续把 review 后的线上候选转换为 canonical records 后落盘；默认紧凑 JSONL。 -如果线上挖掘工具通过 MCP 暴露，先用 `mcp_list_tools` 查看可用工具，再决定是否调用 `mcp_call_tool`。 - -TODO：规划中的专用工具： +TODO：后续规划中的专用工具： - `analyze_badcases` - `build_mining_strategy` -- `search_online_sessions` -- `sample_online_candidates` - `create_annotation_batch` - `read_annotation_result` - `evaluate_mining_precision` @@ -86,3 +119,5 @@ tasks/{task_id}/memory/failed_attempts.md - 除非批准的工具输出已经脱敏，否则不要导出敏感线上原始字段。 - 不要把第一版挖掘策略当成最终策略。 - 始终让筛选条件和抽样决策可 review。 +- 只有用户明确要求生成、扩写或构造新数据时，才进入数据生成分支。 +- 如果用户要求把筛选出的线上候选变成样本，必须先使用 `data_agent_convert_router_candidates_to_records`，不要改走 generation goal/plan。 diff --git a/tests/test_data_agent_router_sessions.py b/tests/test_data_agent_router_sessions.py new file mode 100644 index 0000000..a108f8d --- /dev/null +++ b/tests/test_data_agent_router_sessions.py @@ -0,0 +1,229 @@ +from __future__ import annotations + +import importlib.util +import json +import tempfile +import unittest +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_router_sessions import ( + convert_router_candidates_to_records, + profile_router_sessions, + sample_router_candidates, + search_router_sessions, +) + + +HAS_PYARROW = importlib.util.find_spec('pyarrow') is not None + + +class DataAgentRouterSessionTests(unittest.TestCase): + def test_sample_router_candidates_accepts_search_payload(self) -> None: + payload = { + 'candidates': [ + _candidate('s1', 'phone', 'QA', 'unknown', '怎么查找设备'), + _candidate('s2', 'car', 'mapCopilot', '导航', '导航去公司'), + _candidate('s3', 'phone', 'QA', 'unknown', '手机丢了怎么办'), + ] + } + + result = sample_router_candidates(payload, sample_size=2, strategy='stride') + + self.assertEqual(result['total_candidate_count'], 3) + self.assertEqual(result['sampled_count'], 2) + self.assertEqual(result['summary']['device'][0], {'value': 'phone', 'count': 2}) + self.assertEqual(result['candidates'][0]['semantic_session_id'], 's1') + self.assertEqual(result['candidates'][1]['semantic_session_id'], 's3') + + def test_sample_router_candidates_accepts_json_string(self) -> None: + result = sample_router_candidates( + json.dumps([_candidate('s1', 'phone', 'QA', 'unknown', '怎么查找设备')], ensure_ascii=False), + sample_size=1, + ) + + self.assertEqual(result['sampled_count'], 1) + self.assertEqual(result['candidates'][0]['matched_turn']['query'], '怎么查找设备') + + def test_convert_router_candidates_to_records_keeps_online_metadata(self) -> None: + candidate = _candidate('s1', 'phone', 'QA', 'unknown', '总结一下') + candidate['turn_count'] = 2 + candidate['matched_turn_index'] = 1 + candidate['matched_turn']['timestamp'] = 20 + candidate['prev_turns'] = [{'timestamp': 10, 'query': '这篇文章讲了什么', 'tts': ''}] + + result = convert_router_candidates_to_records( + [candidate], + dataset_label='总结类边界评测集', + default_target='Summarize', + batch_id='summary', + ) + + self.assertEqual(result['record_count'], 1) + record = result['records'][0] + self.assertEqual(record['source']['type'], 'online') + self.assertEqual(record['source']['request_id'], 's1-rid') + self.assertEqual(record['turn']['query'], '总结一下') + self.assertEqual(record['prev_session'], [{'query': '这篇文章讲了什么', 'tts': '', 'timestamp': 10}]) + self.assertEqual(record['context']['domain'], 'QA') + self.assertEqual(record['label']['target'], 'Summarize') + + def test_convert_router_candidates_to_records_applies_review_decisions(self) -> None: + candidates = [ + _candidate('s1', 'phone', 'QA', 'unknown', '总结一下'), + _candidate('s2', 'phone', 'QA', 'unknown', '总结一下红楼梦'), + ] + + result = convert_router_candidates_to_records( + candidates, + dataset_label='总结类边界评测集', + review_decisions=[ + { + 'semantic_session_id': 's1', + 'matched_turn_index': 0, + 'decision': 'include', + 'target': 'Summarize', + 'notes': '前文有可总结内容', + }, + { + 'semantic_session_id': 's2', + 'matched_turn_index': 0, + 'decision': 'exclude', + 'notes': '开放知识对象', + }, + ], + ) + + self.assertEqual(result['record_count'], 1) + self.assertEqual(result['skipped_count'], 1) + self.assertEqual(result['records'][0]['meta']['notes'], '前文有可总结内容') + + @unittest.skipUnless(HAS_PYARROW, 'pyarrow is required for parquet tests') + def test_profile_and_search_router_sessions_read_parquet(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + _write_router_parquet(Path(tmp_dir)) + + profile = profile_router_sessions(tmp_dir, dates=['20260428'], max_files=1) + search = search_router_sessions( + tmp_dir, + dates=['20260428'], + domains=['mapCopilot'], + query_keywords=['导航'], + max_files=1, + max_candidates=5, + ) + + self.assertEqual(profile['sampled_row_count'], 2) + self.assertEqual(profile['missing_req_id_count'], 0) + self.assertTrue(any(item['value'] == 'phone' for item in profile['distributions']['device'])) + self.assertEqual(search['candidate_count'], 1) + candidate = search['candidates'][0] + self.assertEqual(candidate['req_id'], 'rid-1') + self.assertEqual(candidate['matched_turn']['query'], '导航去公司') + self.assertEqual(candidate['prev_turns'][0]['query'], '你好小爱') + + @unittest.skipUnless(HAS_PYARROW, 'pyarrow is required for parquet tests') + def test_router_session_tools_execute_against_registry(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + _write_router_parquet(Path(tmp_dir)) + context = build_tool_context(AgentRuntimeConfig(cwd=Path(tmp_dir))) + registry = default_tool_registry() + + search_result = execute_tool( + registry, + 'data_agent_search_router_sessions', + {'dates': ['20260428'], 'query_regex': '查找设备|手机丢了', 'max_files': 1}, + context, + ) + self.assertTrue(search_result.ok, search_result.content) + sample_result = execute_tool( + registry, + 'data_agent_sample_router_candidates', + {'candidates': json.loads(search_result.content), 'sample_size': 1, 'strategy': 'first'}, + context, + ) + convert_result = execute_tool( + registry, + 'data_agent_convert_router_candidates_to_records', + { + 'candidates': json.loads(sample_result.content), + 'dataset_label': '查找设备评测集', + 'default_target': 'QA', + 'batch_id': 'find_device', + }, + context, + ) + + self.assertTrue(sample_result.ok, sample_result.content) + self.assertEqual(json.loads(sample_result.content)['sampled_count'], 1) + self.assertTrue(convert_result.ok, convert_result.content) + self.assertTrue(json.loads(convert_result.content)['validation']['ok']) + + +def _candidate(session_id: str, device: str, domain: str, intent: str, query: str) -> dict[str, object]: + return { + 'semantic_session_id': session_id, + 'req_id': f'{session_id}-rid', + 'device': device, + 'turn_count': 1, + 'matched_turn_index': 0, + 'matched_turn': { + 'timestamp': 1, + 'query': query, + 'tts': '', + 'action': {'domain': domain, 'intent': intent}, + }, + 'prev_turns': [], + 'source_path': 'router_session_parquet/date=20260428/part-0.parquet', + 'date': '20260428', + } + + +def _write_router_parquet(root: Path) -> None: + import pyarrow as pa + import pyarrow.parquet as pq + + partition = root / 'router_session_parquet' / 'date=20260428' + partition.mkdir(parents=True) + rows = [ + { + 'semantic_session_id': 'phone_20260428_1', + 'device': 'phone', + 'req_id': 'rid-1', + 'turn_count': 2, + 'turns_array': [ + { + 'timestamp': 1, + 'query': '你好小爱', + 'action_json': '{"domain":"dialogCopilot","intent":"问候"}', + 'tts': '你好', + }, + { + 'timestamp': 2, + 'query': '导航去公司', + 'action_json': '{"domain":"mapCopilot","intent":"导航","func":"start_navigation"}', + 'tts': '开始导航', + }, + ], + }, + { + 'semantic_session_id': 'phone_20260428_2', + 'device': 'phone', + 'req_id': 'rid-2', + 'turn_count': 1, + 'turns_array': [ + { + 'timestamp': 3, + 'query': '手机丢了怎么查找设备', + 'action_json': '{"domain":"QA","intent":"unknown"}', + 'tts': '', + } + ], + }, + ] + pq.write_table(pa.Table.from_pylist(rows), partition / 'part-0.parquet') + + +if __name__ == '__main__': + unittest.main()