wuyang/zk-data-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove legacy online mining tools

Browse Source
This commit is contained in:
wuyang6
2026-05-12 18:03:39 +08:00
parent 6dad2cc4b3
commit a48c86501a
4 changed files with 6 additions and 1011 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/online-mining/SKILL.md
+6 -3
View File
@@ -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”的工作流。
## 输出目录约束
src/agent_tool_specs/data_agent.py
-136
View File
@@ -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.',
src/agent_tools.py
-97
View File
@@ -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(
src/data_agent_router_sessions.py
-775
View File
@@ -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 '<empty>'
return str(value)