diff --git a/skills/online-mining/SKILL.md b/skills/online-mining/SKILL.md index be3dc26..f3828cf 100644 --- a/skills/online-mining/SKILL.md +++ b/skills/online-mining/SKILL.md @@ -1,11 +1,14 @@ --- name: online-mining -description: 分析 badcase，并通过可 review 的策略迭代挖掘线上相似 case。 -when_to_use: 当用户提供线上 badcase 或产品/标签定义，并希望查找相似线上问题或构建专项集时使用。 +description: 历史版线上挖掘 skill，已停用；新任务请使用 online-mining-v2。 +when_to_use: 不再用于新任务；保留此文件仅用于查看旧 parquet 挖掘流程说明。 aliases: badcase-mining, router-mining -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 +allowed_tools: read_file, grep_search, glob_search --- +> 这个 skill 是历史版实现，依赖的 router-session 平台工具已经下线。 +> 新的线上挖掘任务请使用 `online-mining-v2`，通过 skill 内脚本和 `elk-fetch` 能力完成。 + 使用这个 skill 处理“badcase 或标签定义 -> 挖掘策略 -> 候选召回 -> 抽样 review -> 策略迭代 -> mined dataset”的工作流。 ## 输出目录约束 diff --git a/src/agent_tool_specs/data_agent.py b/src/agent_tool_specs/data_agent.py index 3b0f1cb..9bb45b4 100644 --- a/src/agent_tool_specs/data_agent.py +++ b/src/agent_tool_specs/data_agent.py @@ -237,142 +237,6 @@ def build_data_agent_tools(handlers: Mapping[str, ToolHandler]) -> list[AgentToo }, handler=resolve_handler(handlers, 'data_agent_render_source_context', 'data-agent'), ), - 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': 'Parquet files or date directories. Relative paths resolve from workspace root; absolute paths are allowed for external online data.', - }, - '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=resolve_handler(handlers, 'data_agent_profile_router_sessions', 'data-agent'), - ), - 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': 'Parquet files or date directories. Relative paths resolve from workspace root; absolute paths are allowed for external online data.', - }, - '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=resolve_handler(handlers, 'data_agent_search_router_sessions', 'data-agent'), - ), - 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=resolve_handler(handlers, 'data_agent_sample_router_candidates', 'data-agent'), - ), - 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.', - }, - 'default_complex': { - 'type': 'boolean', - 'description': 'Default complex dimension for included candidates unless review_decisions provides complex.', - }, - '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'}, - 'complex': {'type': 'boolean'}, - 'notes': {'type': 'string'}, - }, - }, - }, - 'batch_id': {'type': 'string'}, - 'include_uncertain': {'type': 'boolean'}, - }, - 'required': ['candidates', 'dataset_label'], - }, - handler=resolve_handler(handlers, 'data_agent_convert_router_candidates_to_records', 'data-agent'), - ), AgentTool( name='data_agent_show_generation_plan', description='Show a data-agent generation plan for human review.', diff --git a/src/agent_tools.py b/src/agent_tools.py index 51f288c..026110f 100644 --- a/src/agent_tools.py +++ b/src/agent_tools.py @@ -51,13 +51,6 @@ 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 @@ -1041,10 +1034,6 @@ def default_tool_registry() -> dict[str, AgentTool]: 'data_agent_load_input_sources': _data_agent_load_input_sources_tool, 'data_agent_extract_case_evidence': _data_agent_extract_case_evidence_tool, 'data_agent_render_source_context': _data_agent_render_source_context_tool, - 'data_agent_profile_router_sessions': _data_agent_profile_router_sessions_tool, - 'data_agent_search_router_sessions': _data_agent_search_router_sessions_tool, - 'data_agent_sample_router_candidates': _data_agent_sample_router_candidates_tool, - 'data_agent_convert_router_candidates_to_records': _data_agent_convert_router_candidates_to_records_tool, 'data_agent_show_generation_plan': _data_agent_show_generation_plan_tool, 'data_agent_update_generation_plan': _data_agent_update_generation_plan_tool, 'data_agent_confirm_generation_plan': _data_agent_confirm_generation_plan_tool, @@ -1350,92 +1339,6 @@ 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') - default_complex = arguments.get('default_complex', False) - if not isinstance(default_complex, bool): - raise ToolExecutionError('default_complex 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'), - default_complex=default_complex, - 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_router_sessions.py b/src/data_agent_router_sessions.py deleted file mode 100644 index 2459fed..0000000 --- a/src/data_agent_router_sessions.py +++ /dev/null @@ -1,775 +0,0 @@ -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 = '', - default_complex: bool = False, - 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")}' - ) - raw_target, prefixed_complex = _split_complex_prefixed_target(target) - complex_value = _parse_complex( - decision.get('complex'), - default=default_complex if prefixed_complex is None else prefixed_complex, - ) - records.append( - _router_candidate_to_record( - candidate, - dataset_label=str(decision.get('dataset_label') or dataset_label).strip(), - target=raw_target, - complex_value=complex_value, - 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() - 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, - complex_value: bool, - 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), - }, - 'dimensions': { - 'complex': complex_value, - }, - '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: - target = _strip_complex_prefix(target) - if target.startswith('Agent('): - return 'agent' - if target: - return 'function' - return 'unknown' - - -def _strip_complex_prefix(target: str) -> str: - stripped_target, _complex_value = _split_complex_prefixed_target(target) - return stripped_target - - -def _split_complex_prefixed_target(target: str) -> tuple[str, bool | None]: - lines = target.strip().splitlines() - if not lines: - return '', None - match = re.fullmatch(r'complex\s*=\s*(.+)', lines[0].strip(), flags=re.IGNORECASE) - if not match: - return target.strip(), None - return '\n'.join(lines[1:]).strip(), _parse_complex(match.group(1), default=False) - - -def _parse_complex(value: Any, *, default: bool) -> bool: - if value is None or value == '': - return default - if isinstance(value, bool): - return value - text = str(value).strip().lower() - if text in {'true', '1', 'yes', 'y', '是', '复杂', 'complex'}: - return True - if text in {'false', '0', 'no', 'n', '否', '不复杂', '简单', 'simple'}: - return False - raise DataAgentRouterSessionError('complex must be a boolean value such as true/false') - - -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)