diff --git a/skills/online-mining-v2/README.md b/skills/online-mining-v2/README.md index 92808fa..a9a9e19 100644 --- a/skills/online-mining-v2/README.md +++ b/skills/online-mining-v2/README.md @@ -16,23 +16,48 @@ | `scripts/elk_profile_index.py` | 采样索引 schema 和关键字段，帮助 Agent 判断字段可用性。 | | `scripts/elk_search_cases.py` | 按 query/domain/prompt/model output/candidate domain 等条件搜索候选 case。 | | `scripts/elk_fetch_by_request_ids.py` | 批量按 request id 拉取 main/pre_processing 原始摘要。 | -| `scripts/elk_join_request_logs.py` | 把 `arch-flat-nlp-log-f-*` 和 `pre-processing-info*` 按 request id 合并。 | +| `scripts/elk_join_request_logs.py` | 把 `arch-flat-nlp-log-f-*` 和 `pre-processing*` 按 request id 合并。 | | `scripts/build_dataset_draft.py` | 把 review 后的线上 case 转成 product-data dataset draft text。 | +| `scripts/build_online_records.py` | 把 rid/候选 case 直接转成 canonical records，并可导出 records.csv、training.jsonl、eval_planning.csv。 | ## 数据源 - `main`：`arch-flat-nlp-log-f-*` -- `pre_processing`：`pre-processing-info*` +- `pre_processing`：`pre-processing*` 字段说明见 `knowledge/数据源字段说明.md`。 ## 后处理 -后处理复用 `skills/product-data/scripts/`： +优先使用 `scripts/build_online_records.py` 一步完成线上候选到标准数据： + +```json +{ + "request_ids": [""], + "date": "YYYYMMDD", + "dataset_label": "线上挖掘样本", + "target": "Agent(tag=\"xxx\")", + "complex": false, + "output_dir": "output", + "export_records": true, + "export_training": false, + "export_eval": false +} +``` + +它会补齐： + +- `request_id`：主表 `request_id` / 前处理表 `requestId` +- `timestamp`：优先主表 `timestamp` +- `query`：优先前处理 query，缺失时用主表 query +- `prev_session`：主表同 `session_id` 且早于当前 timestamp 的最多 10 轮 +- `target`：优先用户输入，否则用前处理 `code/planningOriginalResult`，再用主表仲裁信息兜底 +- `complex`：优先用户输入，默认 false + +如果需要拆开调试，后处理仍可复用 `skills/product-data/scripts/`： 1. `normalize_dataset_draft.py` 2. `validate_dataset_records.py` 3. `export_dataset_records.py` 4. 可选 `export_training_jsonl.py` 5. 可选 `export_planning_eval_csv.py` - diff --git a/skills/online-mining-v2/SKILL.md b/skills/online-mining-v2/SKILL.md index 0d340f2..6e108b8 100644 --- a/skills/online-mining-v2/SKILL.md +++ b/skills/online-mining-v2/SKILL.md @@ -28,11 +28,12 @@ skills/online-mining-v2/ elk_fetch_by_request_ids.py elk_join_request_logs.py build_dataset_draft.py + build_online_records.py ``` 默认数据源： -- `pre-processing-info*`：前处理日志，适合拿模型 prompt、模型输出、候选 domain、`excellent_domains_result`。 +- `pre-processing*`：前处理日志，适合拿模型 prompt、模型输出、候选 domain、`excellent_domains_result`。 - `arch-flat-nlp-log-f-*`：主 NLP 日志，适合拿最终 `query`、`domain`、`func`、`session_id`、`device_id`、`device`、`tts/text/to_speak`。 后处理全部复用 `product-data`： @@ -128,12 +129,51 @@ skills/online-mining-v2/ } ``` -### 5. 给用户 review +### 5. 单条 rid 或已确认候选直接转元数据 + +如果用户给了明确 rid，或已经确认一批线上候选要直接作为样本，优先使用 +`build_online_records.py`。这个脚本会完成： + +- 主表 `arch-flat-nlp-log-f-*` 和前处理表 `pre-processing*` 双表拉取。 +- 用主表 `session_id + timestamp` 补齐当前请求前最多 10 轮 session。 +- 优先使用用户指定 `target`；没有指定时，按 `pre_processing.code` -> `planningOriginalResult` -> 主表 `llm_agent_info.agentType` -> 主表 `domain` 推断标签。 +- 生成 product-data canonical records。 +- 默认导出 `output/records.jsonl` 和 `output/records.csv`，可选导出 `training.jsonl` 和 `eval_planning.csv`。 + +示例： + +```json +{ + "script_path": "skills/online-mining-v2/scripts/build_online_records.py", + "stdin": { + "request_ids": ["6656271eedfe4b478ac716448a3ad310"], + "date": "20260514", + "dataset_label": "线上挖掘样本", + "target": "Agent(tag=\"地图导航\")", + "complex": false, + "output_dir": "output", + "export_records": true, + "export_training": false, + "export_eval": false + }, + "timeout_seconds": 300, + "max_output_chars": 30000 +} +``` + +注意： + +- 老 rid 超出近 48 小时时必须让用户补日期，传 `date=YYYYMMDD`。 +- 如果用户没有指定 `target`，脚本可以从线上模型输出推断，但最终仍建议展示 `target_source` 给用户确认。 +- 如果脚本返回 `case has no query`，说明该 rid 在当前日期窗口没有命中主表/前处理表，不要继续伪造元数据。 + +### 6. 给用户 review 展示候选样本时只展示必要字段： - request_id - query +- target / target_source（如果已经转换） - main.domain / main.func - pre.planning_result - pre.hit_rules @@ -143,9 +183,10 @@ skills/online-mining-v2/ 不要一次贴大量 prompt。需要看 prompt 时只展示截断摘要，或保存到会话 output 目录。 -### 6. 用户确认后转成 product-data draft +### 7. 用户确认后转成 product-data draft -线上候选直接作为样本时，先生成 dataset draft text： +一般情况下不要再手动接 normalize/export。只有当用户明确要看 draft text， +或需要和 `product-data` 的人工生成流程混合时，才先生成 dataset draft text： ```json { @@ -162,7 +203,7 @@ skills/online-mining-v2/ } ``` -然后必须调用 product-data： +然后再调用 product-data： ```json { @@ -179,7 +220,9 @@ skills/online-mining-v2/ } ``` -再执行校验和导出： +更推荐直接使用第 5 步的 `build_online_records.py`，减少模型手动接线出错。 + +如果使用 draft 分支，再执行校验和导出： - `validate_dataset_records.py` - `export_dataset_records.py`，输出固定 `output/records.jsonl` @@ -201,4 +244,3 @@ skills/online-mining-v2/ - canonical records：通过 product-data 导出到 `output/records.jsonl` 不要写项目根目录、`tasks/`、`src/` 或 `skills/`。 - diff --git a/skills/online-mining-v2/knowledge/工作流.md b/skills/online-mining-v2/knowledge/工作流.md index b1622c1..c65789e 100644 --- a/skills/online-mining-v2/knowledge/工作流.md +++ b/skills/online-mining-v2/knowledge/工作流.md @@ -16,11 +16,36 @@ 3. 用 `elk_search_cases.py` 拉候选。 4. 用 `elk_join_request_logs.py` 补全 main/pre_processing 字段。 5. 展示小批候选给用户 review。 -6. 用户确认后，用 `build_dataset_draft.py` 转 dataset draft。 -7. 用 `product-data` 脚本 normalize、validate、export。 +6. 用户确认后，优先用 `build_online_records.py` 直接生成 canonical records 和流通表格。 +7. 只有需要人工查看 draft 或混合生成数据时，才用 `build_dataset_draft.py` 转 dataset draft，再接 product-data。 这个分支不生成新 query。 +### rid 快速转样本流程 + +适用表达： + +- “把这个 rid 的数据拉下来，作为测试数据” +- “这几个 request_id 转成我们的元数据” +- “线上命中的这批样本直接导出” + +流程： + +1. 整理 `request_ids`、`dataset_label`、`target`、`complex`、日期。 +2. 如果没有日期，先按近 48 小时查；查不到要问用户补日期。 +3. 调用 `build_online_records.py`。 +4. 检查返回的 `summaries`： + - `query` + - `target` + - `target_source` + - `prev_session_count` + - `main_domain` + - `planning_result` +5. 告诉用户输出路径： + - `output/records.jsonl` + - `output/records.csv` + - 用户要求训练/评测时，再导出 `training.jsonl` / `eval_planning.csv` + ## 分支 B：基于线上问题补充生成数据 适用表达： @@ -51,4 +76,3 @@ tts: ``` 不要默认展示完整 prompt。prompt 很长时只展示命中片段。 - diff --git a/skills/online-mining-v2/knowledge/数据源字段说明.md b/skills/online-mining-v2/knowledge/数据源字段说明.md index 701f3fb..99cdd28 100644 --- a/skills/online-mining-v2/knowledge/数据源字段说明.md +++ b/skills/online-mining-v2/knowledge/数据源字段说明.md @@ -28,7 +28,7 @@ - `session_id`、`device_id`、设备类型。 - `intention.intent_arbitrator_info.llm_agent_info.agentType`。 -## pre_processing：`pre-processing-info*` +## pre_processing：`pre-processing*` 适合拿前处理和规划模型判断过程。 @@ -65,6 +65,13 @@ main.request_id == pre_processing.requestId ``` +精确查询时优先使用 keyword 字段： + +```text +main.request_id.keyword +pre_processing.requestId.keyword +``` + join 后优先使用： - query：优先 `pre_processing` 的 query，缺失时用 `main.query`。 @@ -72,3 +79,33 @@ join 后优先使用： - tts：优先 `main.to_speak`，再 `main.text` 或 `display_text`。 - label 参考：模型输出用 `pre_processing.planning_result`，线上最终结果用 `main.domain` / `main.func`。 +## 转 canonical record 的字段映射 + +`build_online_records.py` 固定执行以下映射，模型不要手写转换逻辑： + +| canonical 字段 | 来源 | +| --- | --- | +| `source.type` | 固定 `online` | +| `source.request_id` | 主表 `request_id` 或前处理表 `requestId` | +| `source.timestamp` | 优先主表 `timestamp`，缺失时用前处理表 `timestamp` | +| `turn.query` | 优先前处理 `responseBoby.nodes.0.core.query.query`，缺失时用主表 `query` | +| `turn.timestamp` | 同 `source.timestamp` | +| `prev_session` | 主表同 `session_id` 且 timestamp 早于当前请求的最多 10 轮，按时间升序排列 | +| `prev_session[].query` | 主表历史轮 `query` | +| `prev_session[].tts` | 主表历史轮 `to_speak/text/display_text`，没有时保留空字符串 | +| `prev_session[].timestamp` | 主表历史轮 `timestamp` | +| `context` | 先置 `{}`，后续由专用工具补充 | +| `label.dataset_label` | 用户输入的批次标签 | +| `label.target` | 用户输入 `target`；否则依次用前处理 `code`、`planningOriginalResult`、主表 `llm_agent_info.agentType`、主表 `domain` 推断 | +| `dimensions.complex` | 用户输入，默认 false | +| `meta` | 保存 `session_id/device_id/device/main_domain/main_func/planning_result/planning_code/hit_rules/candidate_domains/target_source` | + +## 按 rid 转数据的固定流程 + +当用户说“把这个 rid 的数据拉下来作为测试数据”时： + +1. 如果用户没给日期，先用近 48 小时查；查不到就问用户日期，不要继续生成空数据。 +2. 调用 `build_online_records.py`，传 `request_ids`、`dataset_label`、必要时传 `target/complex/date`。 +3. 检查返回的 `summaries[].target_source` 和 `prev_session_count`。 +4. 展示 query、target、target_source、prev_session_count 给用户确认。 +5. 产物默认在当前会话 `output/records.jsonl` 和 `output/records.csv`。 diff --git a/skills/online-mining-v2/scripts/build_dataset_draft.py b/skills/online-mining-v2/scripts/build_dataset_draft.py index adc7758..284a144 100644 --- a/skills/online-mining-v2/scripts/build_dataset_draft.py +++ b/skills/online-mining-v2/scripts/build_dataset_draft.py @@ -35,6 +35,69 @@ def case_value(case: dict, key: str): return case.get(key) or pre.get(key) or main.get(key) +def build_dataset_draft_text( + *, + dataset_label: str, + target: str, + complex_value: bool, + cases: list[dict], +) -> str: + """把 review 后的线上 case 转成 product-data dataset draft text。 + + 这个 draft 是中间格式，不是最终数据。后续必须再经过 + product-data 的 normalize/validate/export。 + """ + + lines = [f"# dataset_label: {dataset_label}", ""] + for index, case in enumerate(cases, start=1): + query = str(case_value(case, "query") or "").strip() + if not query: + continue + request_id = str(case_value(case, "request_id") or "").strip() + timestamp = case_value(case, "timestamp") + tts = str(case_value(case, "tts") or "").strip() + planning_result = str(case_value(case, "planning_result") or "").strip() + case_target = str(case_value(case, "target") or target).strip() + case_complex = case_value(case, "complex") + if isinstance(case_complex, str): + case_complex_text = case_complex.strip().lower() + item_complex = case_complex_text in {"true", "1", "yes", "y", "是", "复杂", "complex"} + elif isinstance(case_complex, bool): + item_complex = case_complex + else: + item_complex = complex_value + notes = [] + if planning_result: + notes.append(f"线上模型输出: {planning_result}") + main_domain = case_value(case, "domain") or case.get("main_domain") + if main_domain: + notes.append(f"线上 domain: {main_domain}") + lines.append(f"### case: online_{index:04d}") + prev_session = case_value(case, "prev_session") + if isinstance(prev_session, list): + for item in prev_session[-10:]: + if not isinstance(item, dict): + continue + prev_query = str(item.get("query") or "").strip() + prev_tts = str(item.get("tts") or "").strip() + if prev_query and prev_tts: + lines.append(f"用户: {prev_query}") + lines.append(f"小爱: {prev_tts}") + lines.append(f"用户: {query}") + lines.append(f"complex: {'true' if item_complex else 'false'}") + lines.append(f"target: {case_target}") + if request_id: + lines.append(f"request_id: {request_id}") + if timestamp: + lines.append(f"timestamp: {timestamp}") + if tts: + lines.append(f"notes: tts={tts}" + (f"；{'；'.join(notes)}" if notes else "")) + elif notes: + lines.append(f"notes: {'；'.join(notes)}") + lines.append("") + return "\n".join(lines).strip() + "\n" + + def main() -> int: try: payload = load_json_payload() @@ -49,36 +112,12 @@ def main() -> int: if not cases: raise ValueError("cases must not be empty") - lines = [f"# dataset_label: {dataset_label}", ""] - for index, case in enumerate(cases, start=1): - query = str(case_value(case, "query") or "").strip() - if not query: - continue - request_id = str(case_value(case, "request_id") or "").strip() - timestamp = case_value(case, "timestamp") - tts = str(case_value(case, "tts") or "").strip() - planning_result = str(case_value(case, "planning_result") or "").strip() - notes = [] - if planning_result: - notes.append(f"线上模型输出: {planning_result}") - main_domain = case_value(case, "domain") or case.get("main_domain") - if main_domain: - notes.append(f"线上 domain: {main_domain}") - lines.append(f"### case: online_{index:04d}") - lines.append(f"用户: {query}") - lines.append(f"complex: {'true' if complex_value else 'false'}") - lines.append(f"target: {target}") - if request_id: - lines.append(f"request_id: {request_id}") - if timestamp: - lines.append(f"timestamp: {timestamp}") - if tts: - lines.append(f"notes: tts={tts}" + (f"；{'；'.join(notes)}" if notes else "")) - elif notes: - lines.append(f"notes: {'；'.join(notes)}") - lines.append("") - - draft_text = "\n".join(lines).strip() + "\n" + draft_text = build_dataset_draft_text( + dataset_label=dataset_label, + target=target, + complex_value=complex_value, + cases=cases, + ) output_path = None if payload.get("output_path"): output = resolve_portable_path(str(payload["output_path"])) @@ -94,4 +133,3 @@ def main() -> int: if __name__ == "__main__": raise SystemExit(main()) - diff --git a/skills/online-mining-v2/scripts/build_online_records.py b/skills/online-mining-v2/scripts/build_online_records.py new file mode 100644 index 0000000..03e11e9 --- /dev/null +++ b/skills/online-mining-v2/scripts/build_online_records.py @@ -0,0 +1,353 @@ +from __future__ import annotations + +"""把线上 ELK case 直接转换为 product-data canonical records。 + +这个脚本承接 online-mining-v2 和 product-data： +1. 按 request_id 拉主 NLP 表和前处理表。 +2. 用主表 session_id 补齐当前请求前最多 10 轮上下文。 +3. 从用户指定 target 或线上 planning/code 结果推断标签。 +4. 直接产出 canonical records，并可同步导出流通表格、训练 jsonl、评测 csv。 +""" + +import json +import re +import sys +from pathlib import Path +from typing import Any + +from online_mining_common import ( + OnlineMiningError, + compact_text, + emit_error, + emit_success, + extract_case, + fetch_one_by_request_id, + fetch_session_turns, + load_json_payload, + resolve_portable_path, + string_values, +) + +PRODUCT_DATA_SCRIPT_DIR = Path(__file__).resolve().parents[1].parent / "product-data" / "scripts" +sys.path.insert(0, str(PRODUCT_DATA_SCRIPT_DIR)) + +from product_data_portable import ( # noqa: E402 + canonical_target, + export_dataset_records, + export_planning_eval_csv, + export_training_jsonl, + record_id, + target_type, + validate_dataset_records, +) + + +def load_cases(payload: dict[str, Any]) -> list[dict[str, Any]]: + if isinstance(payload.get("cases"), list): + return [item for item in payload["cases"] if isinstance(item, dict)] + cases_path = payload.get("cases_path") + if cases_path: + rows = [] + text = resolve_portable_path(str(cases_path)).read_text(encoding="utf-8") + for line in text.splitlines(): + if not line.strip(): + continue + item = json.loads(line) + if isinstance(item, dict): + rows.append(item) + return rows + return [] + + +def merge_case(request_id: str, main_case: dict[str, Any] | None, pre_case: dict[str, Any] | None) -> dict[str, Any]: + return { + "request_id": request_id, + "query": (pre_case or {}).get("query") or (main_case or {}).get("query"), + "timestamp": (main_case or {}).get("timestamp") or (pre_case or {}).get("timestamp"), + "session_id": (main_case or {}).get("session_id"), + "device_id": (main_case or {}).get("device_id"), + "device": (main_case or {}).get("device"), + "tts": (main_case or {}).get("text"), + "main": main_case, + "pre_processing": pre_case, + } + + +def fetch_cases_by_request_ids(request_ids: list[str], *, date: str | None) -> list[dict[str, Any]]: + rows = [] + for request_id in request_ids: + main_doc = fetch_one_by_request_id("main", request_id, date) + pre_doc = fetch_one_by_request_id("pre_processing", request_id, date) + main_case = extract_case("main", main_doc) if main_doc else None + pre_case = extract_case("pre_processing", pre_doc) if pre_doc else None + rows.append(merge_case(request_id, main_case, pre_case)) + return rows + + +def case_value(case: dict[str, Any], key: str) -> Any: + if key in case: + return case.get(key) + main = case.get("main") if isinstance(case.get("main"), dict) else {} + pre = case.get("pre_processing") if isinstance(case.get("pre_processing"), dict) else {} + return case.get(key) or pre.get(key) or main.get(key) + + +def enrich_prev_session(case: dict[str, Any], *, date: str | None, limit: int) -> list[dict[str, Any]]: + explicit = case.get("prev_session") + if isinstance(explicit, list): + return normalize_prev_session(explicit)[-limit:] + session_id = str(case_value(case, "session_id") or "").strip() + timestamp = optional_int(case_value(case, "timestamp")) + if not session_id or timestamp is None: + return [] + turns = fetch_session_turns(session_id=session_id, before_timestamp=timestamp, date=date, limit=limit) + return [ + { + "query": str(turn.get("query") or ""), + "tts": str(turn.get("text") or ""), + "timestamp": optional_int(turn.get("timestamp")) or 0, + } + for turn in turns + if str(turn.get("query") or "").strip() + ][-limit:] + + +def normalize_prev_session(items: list[Any]) -> list[dict[str, Any]]: + rows = [] + for item in items: + if not isinstance(item, dict): + continue + query = str(item.get("query") or "").strip() + if not query: + continue + rows.append( + { + "query": query, + "tts": str(item.get("tts") or ""), + "timestamp": optional_int(item.get("timestamp")) or 0, + } + ) + rows.sort(key=lambda item: int(item.get("timestamp") or 0)) + return rows + + +def infer_target(case: dict[str, Any], explicit_target: str) -> tuple[str, str]: + if explicit_target.strip(): + return canonical_target(explicit_target.strip()), "input.target" + case_target = str(case_value(case, "target") or "").strip() + if case_target: + return canonical_target(case_target), "case.target" + + pre = case.get("pre_processing") if isinstance(case.get("pre_processing"), dict) else {} + main = case.get("main") if isinstance(case.get("main"), dict) else {} + for key in ("code", "planning_result"): + value = str(pre.get(key) or "").strip() + if value: + return canonical_target(value), f"pre_processing.{key}" + + llm_agent_info = main.get("llm_agent_info") + if isinstance(llm_agent_info, dict): + agent_type = str(llm_agent_info.get("agentType") or "").strip() + if agent_type: + return f'Agent(tag="{agent_type}")', "main.intention.intent_arbitrator_info.llm_agent_info.agentType" + + domain = str(main.get("domain") or case_value(case, "domain") or "").strip() + if domain: + if re.fullmatch(r"[A-Z][A-Za-z0-9_]*", domain): + return f"{domain}()", "main.domain" + return f'Agent(tag="{domain}")', "main.domain" + raise OnlineMiningError("target is required because online logs do not contain code/planning_result/domain") + + +def case_to_record( + case: dict[str, Any], + *, + dataset_label: str, + target: str, + complex_value: bool, + index: int, + date: str | None, + session_limit: int, +) -> tuple[dict[str, Any], dict[str, Any]]: + query = str(case_value(case, "query") or "").strip() + if not query: + raise OnlineMiningError(f"case {index} has no query") + request_id = str(case_value(case, "request_id") or "").strip() + if not request_id: + raise OnlineMiningError(f"case {index} has no request_id") + timestamp = optional_int(case_value(case, "timestamp")) + if timestamp is None: + raise OnlineMiningError(f"case {index} has no timestamp") + + final_target, target_source = infer_target(case, target) + prev_session = enrich_prev_session(case, date=date, limit=session_limit) + main = case.get("main") if isinstance(case.get("main"), dict) else {} + pre = case.get("pre_processing") if isinstance(case.get("pre_processing"), dict) else {} + record = { + "record_id": record_id(source_type="online", batch_id="online", request_id=request_id, index=index), + "source": { + "type": "online", + "request_id": request_id, + "timestamp": timestamp, + }, + "turn": { + "query": query, + "timestamp": timestamp, + }, + "prev_session": prev_session, + "context": {}, + "label": { + "dataset_label": dataset_label, + "target": final_target, + "target_type": target_type(final_target), + }, + "dimensions": { + "complex": parse_complex(case_value(case, "complex"), default=complex_value), + }, + "meta": { + "session_id": str(case_value(case, "session_id") or ""), + "device_id": str(case_value(case, "device_id") or ""), + "device": compact_text(case_value(case, "device"), 500), + "online_tts": str(case_value(case, "tts") or ""), + "main_domain": str(main.get("domain") or ""), + "main_func": str(main.get("func") or ""), + "planning_result": str(pre.get("planning_result") or ""), + "planning_code": str(pre.get("code") or ""), + "hit_rules": pre.get("hit_rules") or main.get("hit_rules") or [], + "candidate_domains": pre.get("candidate_domains") or [], + "target_source": target_source, + }, + } + summary = { + "request_id": request_id, + "query": query, + "target": final_target, + "target_source": target_source, + "prev_session_count": len(prev_session), + "main_domain": record["meta"]["main_domain"], + "planning_result": record["meta"]["planning_result"], + } + return record, summary + + +def optional_int(value: Any) -> int | None: + if value is None or value == "": + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +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 OnlineMiningError("complex must be true/false") + + +def write_records_jsonl(records: list[dict[str, Any]], path: str) -> str: + output = resolve_portable_path(path) + output.parent.mkdir(parents=True, exist_ok=True) + with output.open("w", encoding="utf-8") as fh: + for record in records: + fh.write(json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n") + return str(output) + + +def main() -> int: + try: + payload = load_json_payload() + dataset_label = str(payload.get("dataset_label") or "").strip() + if not dataset_label: + raise OnlineMiningError("dataset_label is required") + target = str(payload.get("target") or "").strip() + complex_value = parse_complex(payload.get("complex"), default=False) + date = str(payload.get("date") or "").strip() or None + session_limit = int(payload.get("session_limit") or 10) + if session_limit < 0 or session_limit > 10: + raise OnlineMiningError("session_limit must be between 0 and 10") + + request_ids = string_values(payload.get("request_ids")) + cases = load_cases(payload) + if request_ids: + cases = fetch_cases_by_request_ids(request_ids, date=date) + if not cases: + raise OnlineMiningError("request_ids, cases or cases_path is required") + + records = [] + summaries = [] + skipped = [] + for index, case in enumerate(cases, start=1): + try: + record, summary = case_to_record( + case, + dataset_label=dataset_label, + target=target, + complex_value=complex_value, + index=index, + date=date, + session_limit=session_limit, + ) + except Exception as exc: # noqa: BLE001 + skipped.append({"index": index, "request_id": case.get("request_id"), "error": str(exc)}) + continue + records.append(record) + summaries.append(summary) + + if not records: + raise OnlineMiningError(f"no records built; skipped={skipped}") + + records_path = write_records_jsonl(records, str(payload.get("records_output_path") or "scratchpad/online_records.jsonl")) + validation = validate_dataset_records(records) + output_dir = str(payload.get("output_dir") or "output") + exports: dict[str, Any] = {} + if bool(payload.get("export_records", True)): + exports["records"] = export_dataset_records( + records, + output_path=str(resolve_portable_path(output_dir) / "records.jsonl"), + output_format="jsonl", + require_validation_ok=bool(payload.get("require_validation_ok", True)), + overwrite=bool(payload.get("overwrite", True)), + export_table=bool(payload.get("export_table", True)), + ) + if bool(payload.get("export_training", False)): + exports["training"] = export_training_jsonl( + records, + output_path=str(resolve_portable_path(output_dir) / "training.jsonl"), + require_validation_ok=bool(payload.get("require_validation_ok", True)), + overwrite=bool(payload.get("overwrite", True)), + ) + if bool(payload.get("export_eval", False)): + exports["eval"] = export_planning_eval_csv( + records, + output_path=str(resolve_portable_path(output_dir) / "eval_planning.csv"), + require_validation_ok=bool(payload.get("require_validation_ok", True)), + overwrite=bool(payload.get("overwrite", True)), + ) + + emit_success( + { + "date": date or "past-48h", + "record_count": len(records), + "records_path": records_path, + "validation": validation, + "exports": exports, + "summaries": summaries, + "skipped": skipped, + } + ) + return 0 + except Exception as exc: # noqa: BLE001 + emit_error(exc) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/online-mining-v2/scripts/online_mining_common.py b/skills/online-mining-v2/scripts/online_mining_common.py index d4496ba..776aaf3 100644 --- a/skills/online-mining-v2/scripts/online_mining_common.py +++ b/skills/online-mining-v2/scripts/online_mining_common.py @@ -8,6 +8,7 @@ from __future__ import annotations import argparse import importlib.util import json +import os import re import sys from copy import deepcopy @@ -35,7 +36,7 @@ SOURCE_CONFIGS: dict[str, dict[str, Any]] = { }, "pre_processing": { "profile": "default", - "index": "pre-processing-info*", + "index": "pre-processing*", "id_field": "requestId", "time_field": "timestamp", }, @@ -282,21 +283,70 @@ def search_docs( def fetch_one_by_request_id(source: str, request_id: str, date: str | None = None) -> dict[str, Any] | None: config = source_config(source) - docs = search_docs( - source=source, - date=date, - size=1, - query_filters=[{"term": {str(config["id_field"]): request_id}}], - ) - if not docs: - # 部分 request id 会带后缀，兜底用 wildcard。 - docs = search_docs( - source=source, - date=date, - size=1, - query_filters=[{"wildcard": {str(config["id_field"]): {"value": f"{request_id}*"}}}], - ) - return docs[0] if docs else None + id_field = str(config["id_field"]) + # 精确查优先用 .keyword；部分索引字段本身就是 keyword，所以再兜底原字段。 + attempts = [ + {"term": {f"{id_field}.keyword": request_id}}, + {"term": {id_field: request_id}}, + {"wildcard": {f"{id_field}.keyword": {"value": f"{request_id}*"}}}, + {"wildcard": {id_field: {"value": f"{request_id}*"}}}, + ] + for query_filter in attempts: + docs = search_docs(source=source, date=date, size=1, query_filters=[query_filter]) + if docs: + return docs[0] + return None + + +def fetch_session_turns( + *, + session_id: str, + before_timestamp: int, + date: str | None = None, + limit: int = 10, +) -> list[dict[str, Any]]: + """按主表 session_id 拉当前请求之前的多轮上下文。""" + + if not session_id or not before_timestamp: + return [] + filters = [ + {"term": {"session_id.keyword": session_id}}, + {"range": {"timestamp": {"lt": before_timestamp}}}, + ] + docs = search_docs(source="main", date=date, size=max(limit * 3, limit), query_filters=filters) + cases = [extract_main_case(doc) for doc in docs] + cases = [ + case + for case in cases + if str(case.get("session_id") or "") == session_id + and isinstance(case.get("timestamp"), (int, str)) + and int(case.get("timestamp") or 0) < before_timestamp + and str(case.get("query") or "").strip() + ] + cases.sort(key=lambda item: int(item.get("timestamp") or 0)) + return cases[-limit:] + + +def fetch_session_turns_for_case(case: dict[str, Any], *, date: str | None = None, limit: int = 10) -> list[dict[str, Any]]: + session_id = str(case.get("session_id") or "").strip() + try: + timestamp = int(case.get("timestamp") or 0) + except (TypeError, ValueError): + timestamp = 0 + return fetch_session_turns(session_id=session_id, before_timestamp=timestamp, date=date, limit=limit) + + +def fetch_by_terms(source: str, *, field: str, value: str, date: str | None = None, size: int = 10) -> list[dict[str, Any]]: + """小范围精确字段查询，供脚本探测 session/request 字段时使用。""" + + filters = [{"term": {field: value}}] + return search_docs(source=source, date=date, size=size, query_filters=filters) + + +def fetch_one_by_request_id_legacy(source: str, request_id: str, date: str | None = None) -> dict[str, Any] | None: + """保留旧函数名兼容外部脚本；新代码请用 fetch_one_by_request_id。""" + + return fetch_one_by_request_id(source, request_id, date) def write_optional_jsonl(path: str | None, rows: list[dict[str, Any]]) -> str | None: @@ -311,8 +361,27 @@ def write_optional_jsonl(path: str | None, rows: list[dict[str, Any]]) -> str | def resolve_portable_path(path: str) -> Path: + """解析 online-mining-v2 脚本路径，并优先映射到当前会话目录。 + + 在 ZK Data Agent 中，python_exec 会注入 PYTHON_EXEC_SCRATCHPAD。 + 用户和 skill 文档里写的 scratchpad/、output/、input/ 都应该落在 + 当前会话下，不能误写到项目根目录。 + """ + raw = Path(path).expanduser() if raw.is_absolute(): return raw + scratchpad = os.environ.get("PYTHON_EXEC_SCRATCHPAD") + if scratchpad: + session_root = Path(scratchpad).expanduser().parent + parts = raw.parts + if parts: + head, *tail = parts + tail_path = Path(*tail) if tail else Path() + if head in {"scratchpad", "scratch"}: + return (Path(scratchpad).expanduser() / tail_path).resolve() + if head in {"output", "outputs"}: + return (session_root / "output" / tail_path).resolve() + if head in {"input", "inputs"}: + return (session_root / "input" / tail_path).resolve() return (REPO_ROOT / raw).resolve() -