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)