From 87dd67c76ab6f71da53bf8b65f891fa4b2783bd3 Mon Sep 17 00:00:00 2001 From: wuyang6 Date: Thu, 14 May 2026 17:42:08 +0800 Subject: [PATCH] Use prompt history for online mining records --- skills/online-mining-v2/README.md | 10 +- skills/online-mining-v2/SKILL.md | 16 ++- skills/online-mining-v2/knowledge/工作流.md | 7 +- .../knowledge/数据源字段说明.md | 17 +-- .../scripts/build_online_records.py | 105 ++++++++++++++---- .../scripts/elk_fetch_by_request_ids.py | 16 ++- .../scripts/elk_join_request_logs.py | 25 ++++- .../scripts/elk_profile_index.py | 14 ++- .../scripts/elk_search_cases.py | 10 +- .../scripts/online_mining_common.py | 96 ++++++++-------- 10 files changed, 219 insertions(+), 97 deletions(-) diff --git a/skills/online-mining-v2/README.md b/skills/online-mining-v2/README.md index a9a9e19..a3fa519 100644 --- a/skills/online-mining-v2/README.md +++ b/skills/online-mining-v2/README.md @@ -34,7 +34,7 @@ ```json { "request_ids": [""], - "date": "YYYYMMDD", + "lookback_days": 7, "dataset_label": "线上挖掘样本", "target": "Agent(tag=\"xxx\")", "complex": false, @@ -50,10 +50,16 @@ - `request_id`：主表 `request_id` / 前处理表 `requestId` - `timestamp`：优先主表 `timestamp` - `query`：优先前处理 query，缺失时用主表 query -- `prev_session`：主表同 `session_id` 且早于当前 timestamp 的最多 10 轮 +- `prev_session`：从前处理 `promptModel` 的 `[对话历史]` 抽取，这是 planning 模型真实看到的 session - `target`：优先用户输入，否则用前处理 `code/planningOriginalResult`，再用主表仲裁信息兜底 - `complex`：优先用户输入，默认 false +时间筛选支持： + +- `date`: 单日，格式 `YYYYMMDD` 或 `YYYY-MM-DD` +- `date_from` + `date_to`: 起止日期，包含两端日期 +- `lookback_days`: 最近 N 天，例如用户说“一周内”就传 `7` + 如果需要拆开调试，后处理仍可复用 `skills/product-data/scripts/`： 1. `normalize_dataset_draft.py` diff --git a/skills/online-mining-v2/SKILL.md b/skills/online-mining-v2/SKILL.md index 6e108b8..f0cab1e 100644 --- a/skills/online-mining-v2/SKILL.md +++ b/skills/online-mining-v2/SKILL.md @@ -34,7 +34,7 @@ skills/online-mining-v2/ 默认数据源： - `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`。 +- `arch-flat-nlp-log-f-*`：主 NLP 日志，适合拿最终 `query`、`domain`、`func`、`request_id`、`device_id`、`device`、`tts/text/to_speak`。 后处理全部复用 `product-data`： @@ -89,6 +89,14 @@ skills/online-mining-v2/ } ``` +时间筛选支持三种写法： + +- 单日：`"date": "20260512"` +- 日期范围：`"date_from": "20260508", "date_to": "20260514"` +- 最近 N 天：`"lookback_days": 7` + +用户说“一周内”“最近 7 天”时，优先传 `lookback_days: 7`；用户给明确起止日期时，传 `date_from/date_to`。 + ### 3. 搜索候选 按策略搜索： @@ -135,7 +143,7 @@ skills/online-mining-v2/ `build_online_records.py`。这个脚本会完成： - 主表 `arch-flat-nlp-log-f-*` 和前处理表 `pre-processing*` 双表拉取。 -- 用主表 `session_id + timestamp` 补齐当前请求前最多 10 轮 session。 +- 从前处理 `promptModel` 的 `[对话历史]` 抽取模型真实输入 session，作为 `prev_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`。 @@ -147,7 +155,7 @@ skills/online-mining-v2/ "script_path": "skills/online-mining-v2/scripts/build_online_records.py", "stdin": { "request_ids": ["6656271eedfe4b478ac716448a3ad310"], - "date": "20260514", + "lookback_days": 7, "dataset_label": "线上挖掘样本", "target": "Agent(tag=\"地图导航\")", "complex": false, @@ -164,8 +172,10 @@ skills/online-mining-v2/ 注意： - 老 rid 超出近 48 小时时必须让用户补日期，传 `date=YYYYMMDD`。 +- 如果用户说“一周内/最近 7 天”，可以直接传 `lookback_days: 7`，不必逐日循环。 - 如果用户没有指定 `target`，脚本可以从线上模型输出推断，但最终仍建议展示 `target_source` 给用户确认。 - 如果脚本返回 `case has no query`，说明该 rid 在当前日期窗口没有命中主表/前处理表，不要继续伪造元数据。 +- 不要用主表 `session_id` 去拼训练/评测 session；它不是 planning 模型输入 session 的同义概念。 ### 6. 给用户 review diff --git a/skills/online-mining-v2/knowledge/工作流.md b/skills/online-mining-v2/knowledge/工作流.md index c65789e..3f0edd3 100644 --- a/skills/online-mining-v2/knowledge/工作流.md +++ b/skills/online-mining-v2/knowledge/工作流.md @@ -31,8 +31,8 @@ 流程： -1. 整理 `request_ids`、`dataset_label`、`target`、`complex`、日期。 -2. 如果没有日期，先按近 48 小时查；查不到要问用户补日期。 +1. 整理 `request_ids`、`dataset_label`、`target`、`complex`、时间范围。 +2. 如果用户说“一周内/最近 7 天”，传 `lookback_days: 7`；如果给起止日期，传 `date_from/date_to`；如果只给某一天，传 `date`。 3. 调用 `build_online_records.py`。 4. 检查返回的 `summaries`： - `query` @@ -46,6 +46,9 @@ - `output/records.csv` - 用户要求训练/评测时，再导出 `training.jsonl` / `eval_planning.csv` +`prev_session` 以前处理 `promptModel` 的 `[对话历史]` 为准。不要用主表 `session_id` +重建模型输入历史；它更适合排查同设备/同链路日志，不适合作为训练/评测 prompt 历史。 + ## 分支 B：基于线上问题补充生成数据 适用表达： diff --git a/skills/online-mining-v2/knowledge/数据源字段说明.md b/skills/online-mining-v2/knowledge/数据源字段说明.md index 99cdd28..22294e8 100644 --- a/skills/online-mining-v2/knowledge/数据源字段说明.md +++ b/skills/online-mining-v2/knowledge/数据源字段说明.md @@ -13,7 +13,7 @@ | `query` | 当前 query。 | | `domain` | 最终 domain。 | | `func` / `func_name` | 最终 function 或函数名。 | -| `session_id` | 会话 id，可用于后续重建多轮。 | +| `session_id` | 主表会话 id，只用于排查，不等同于 planning 模型输入 session。 | | `device_id` | 设备 id。 | | `device` | 设备 JSON，包含设备类型、经纬度等。 | | `text` / `display_text` / `to_speak` | 小爱回复文本，可作为上一轮 tts。 | @@ -90,10 +90,10 @@ join 后优先使用： | `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` | +| `prev_session` | 前处理 `promptModel` 的 `[对话历史]`，这是 planning 模型真实看到的 session | +| `prev_session[].query` | `[对话历史]` 中的 `用户:` 行 | +| `prev_session[].tts` | `[对话历史]` 中紧随其后的 `小爱:` 行，没有内容时保留空字符串 | +| `prev_session[].timestamp` | 线上 prompt 不带历史时间戳，工具按当前轮 timestamp 往前每轮 1 分钟补齐，保证顺序和 5 分钟拼接约束 | | `context` | 先置 `{}`，后续由专用工具补充 | | `label.dataset_label` | 用户输入的批次标签 | | `label.target` | 用户输入 `target`；否则依次用前处理 `code`、`planningOriginalResult`、主表 `llm_agent_info.agentType`、主表 `domain` 推断 | @@ -104,8 +104,11 @@ join 后优先使用： 当用户说“把这个 rid 的数据拉下来作为测试数据”时： -1. 如果用户没给日期，先用近 48 小时查；查不到就问用户日期，不要继续生成空数据。 -2. 调用 `build_online_records.py`，传 `request_ids`、`dataset_label`、必要时传 `target/complex/date`。 +1. 如果用户给“一周内/最近 7 天”，传 `lookback_days: 7`；给起止日期时传 `date_from/date_to`；只给某一天时传 `date`。 +2. 调用 `build_online_records.py`，传 `request_ids`、`dataset_label`、必要时传 `target/complex/date/lookback_days/date_from/date_to`。 3. 检查返回的 `summaries[].target_source` 和 `prev_session_count`。 4. 展示 query、target、target_source、prev_session_count 给用户确认。 5. 产物默认在当前会话 `output/records.jsonl` 和 `output/records.csv`。 + +注意：不要用主表 `session_id` 拼接 `prev_session`。中控 planning 模型的输入历史以 +前处理表 `dispatchLargeModelInput.promptModel` 中的 `[对话历史]` 为准。 diff --git a/skills/online-mining-v2/scripts/build_online_records.py b/skills/online-mining-v2/scripts/build_online_records.py index 03e11e9..b08d164 100644 --- a/skills/online-mining-v2/scripts/build_online_records.py +++ b/skills/online-mining-v2/scripts/build_online_records.py @@ -4,7 +4,7 @@ from __future__ import annotations 这个脚本承接 online-mining-v2 和 product-data： 1. 按 request_id 拉主 NLP 表和前处理表。 -2. 用主表 session_id 补齐当前请求前最多 10 轮上下文。 +2. 优先从前处理表 promptModel 的 [对话历史] 抽取模型真实输入 session。 3. 从用户指定 target 或线上 planning/code 结果推断标签。 4. 直接产出 canonical records，并可同步导出流通表格、训练 jsonl、评测 csv。 """ @@ -22,7 +22,6 @@ from online_mining_common import ( emit_success, extract_case, fetch_one_by_request_id, - fetch_session_turns, load_json_payload, resolve_portable_path, string_values, @@ -73,11 +72,32 @@ def merge_case(request_id: str, main_case: dict[str, Any] | None, pre_case: dict } -def fetch_cases_by_request_ids(request_ids: list[str], *, date: str | None) -> list[dict[str, Any]]: +def fetch_cases_by_request_ids( + request_ids: list[str], + *, + date: str | None, + date_from: str | None = None, + date_to: str | None = None, + lookback_days: int | None = 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_doc = fetch_one_by_request_id( + "main", + request_id, + date, + date_from=date_from, + date_to=date_to, + lookback_days=lookback_days, + ) + pre_doc = fetch_one_by_request_id( + "pre_processing", + request_id, + date, + date_from=date_from, + date_to=date_to, + lookback_days=lookback_days, + ) 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)) @@ -92,24 +112,55 @@ def case_value(case: dict[str, Any], key: str) -> Any: 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]]: +def enrich_prev_session(case: dict[str, Any], *, 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: + # 前处理 promptModel 是中控 planning 模型真实看到的输入，里面的 [对话历史] + # 才是训练/评测应复现的 session。主表 session_id 不是同一概念，不默认使用。 + prompt_history = parse_prompt_history(str(case_value(case, "prompt_model") or "")) + if prompt_history and timestamp is not None: + step_ms = 60_000 + start_ts = timestamp - len(prompt_history) * step_ms + for index, item in enumerate(prompt_history): + item["timestamp"] = start_ts + index * step_ms + return prompt_history[-limit:] + + +def parse_prompt_history(prompt: str) -> list[dict[str, Any]]: + if not prompt: 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:] + history_marker = "[对话历史]" + current_marker = "[当前query]" + history_start = prompt.rfind(history_marker) + if history_start < 0: + return [] + current_start = prompt.find(current_marker, history_start) + if current_start < 0: + return [] + block = prompt[history_start + len(history_marker) : current_start].strip() + if not block: + return [] + rows: list[dict[str, Any]] = [] + pending_user: str | None = None + for raw_line in block.splitlines(): + line = raw_line.strip() + if not line: + continue + user_match = re.match(r"^用户\s*[:：]\s*(.*)$", line) + if user_match: + if pending_user is not None: + rows.append({"query": pending_user, "tts": "", "timestamp": 0}) + pending_user = user_match.group(1).strip() + continue + assistant_match = re.match(r"^小爱\s*[:：]\s*(.*)$", line) + if assistant_match and pending_user is not None: + rows.append({"query": pending_user, "tts": assistant_match.group(1).strip(), "timestamp": 0}) + pending_user = None + if pending_user is not None: + rows.append({"query": pending_user, "tts": "", "timestamp": 0}) + return [row for row in rows if row["query"]] def normalize_prev_session(items: list[Any]) -> list[dict[str, Any]]: @@ -166,7 +217,6 @@ def case_to_record( 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() @@ -180,7 +230,7 @@ def case_to_record( 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) + prev_session = enrich_prev_session(case, 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 = { @@ -270,6 +320,10 @@ def main() -> int: 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 + date_from = str(payload.get("date_from") or "").strip() or None + date_to = str(payload.get("date_to") or "").strip() or None + lookback_days = payload.get("lookback_days") + lookback_days = int(lookback_days) if lookback_days not in (None, "") else 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") @@ -277,7 +331,13 @@ def main() -> int: 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) + cases = fetch_cases_by_request_ids( + request_ids, + date=date, + date_from=date_from, + date_to=date_to, + lookback_days=lookback_days, + ) if not cases: raise OnlineMiningError("request_ids, cases or cases_path is required") @@ -292,7 +352,6 @@ def main() -> int: target=target, complex_value=complex_value, index=index, - date=date, session_limit=session_limit, ) except Exception as exc: # noqa: BLE001 @@ -334,7 +393,7 @@ def main() -> int: emit_success( { - "date": date or "past-48h", + "date": date or (f"{date_from}..{date_to}" if date_from and date_to else f"past-{lookback_days}d" if lookback_days else "past-48h"), "record_count": len(records), "records_path": records_path, "validation": validation, diff --git a/skills/online-mining-v2/scripts/elk_fetch_by_request_ids.py b/skills/online-mining-v2/scripts/elk_fetch_by_request_ids.py index a5f9a8d..d527c6f 100644 --- a/skills/online-mining-v2/scripts/elk_fetch_by_request_ids.py +++ b/skills/online-mining-v2/scripts/elk_fetch_by_request_ids.py @@ -20,17 +20,28 @@ def main() -> int: raise ValueError("request_ids is required") sources = payload.get("sources") or ["main", "pre_processing"] date = payload.get("date") + date_from = payload.get("date_from") + date_to = payload.get("date_to") + lookback_days = payload.get("lookback_days") + lookback_days = int(lookback_days) if lookback_days not in (None, "") else None rows = [] for request_id in request_ids: item = {"request_id": request_id} for source in sources: - doc = fetch_one_by_request_id(str(source), request_id, date) + doc = fetch_one_by_request_id( + str(source), + request_id, + date, + date_from=date_from, + date_to=date_to, + lookback_days=lookback_days, + ) item[str(source)] = extract_case(str(source), doc) if doc else None rows.append(item) output_path = write_optional_jsonl(str(payload.get("output_path") or ""), rows) emit_success( { - "date": date or "past-48h", + "date": date or (f"{date_from}..{date_to}" if date_from and date_to else f"past-{lookback_days}d" if lookback_days else "past-48h"), "count": len(rows), "output_path": output_path, "cases": [ @@ -58,4 +69,3 @@ def main() -> int: if __name__ == "__main__": raise SystemExit(main()) - diff --git a/skills/online-mining-v2/scripts/elk_join_request_logs.py b/skills/online-mining-v2/scripts/elk_join_request_logs.py index 0079d33..8b9eaea 100644 --- a/skills/online-mining-v2/scripts/elk_join_request_logs.py +++ b/skills/online-mining-v2/scripts/elk_join_request_logs.py @@ -58,17 +58,35 @@ def main() -> int: if not request_ids: raise ValueError("request_ids, cases or cases_path is required") date = payload.get("date") + date_from = payload.get("date_from") + date_to = payload.get("date_to") + lookback_days = payload.get("lookback_days") + lookback_days = int(lookback_days) if lookback_days not in (None, "") else None 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_doc = fetch_one_by_request_id( + "main", + request_id, + date, + date_from=date_from, + date_to=date_to, + lookback_days=lookback_days, + ) + pre_doc = fetch_one_by_request_id( + "pre_processing", + request_id, + date, + date_from=date_from, + date_to=date_to, + lookback_days=lookback_days, + ) 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)) output_path = write_optional_jsonl(str(payload.get("output_path") or ""), rows) emit_success( { - "date": date or "past-48h", + "date": date or (f"{date_from}..{date_to}" if date_from and date_to else f"past-{lookback_days}d" if lookback_days else "past-48h"), "count": len(rows), "output_path": output_path, "cases": [ @@ -95,4 +113,3 @@ def main() -> int: if __name__ == "__main__": raise SystemExit(main()) - diff --git a/skills/online-mining-v2/scripts/elk_profile_index.py b/skills/online-mining-v2/scripts/elk_profile_index.py index 06698db..356e4eb 100644 --- a/skills/online-mining-v2/scripts/elk_profile_index.py +++ b/skills/online-mining-v2/scripts/elk_profile_index.py @@ -18,6 +18,10 @@ def main() -> int: payload = load_json_payload() sources = payload.get("sources") or ["main", "pre_processing"] date = payload.get("date") + date_from = payload.get("date_from") + date_to = payload.get("date_to") + lookback_days = payload.get("lookback_days") + lookback_days = int(lookback_days) if lookback_days not in (None, "") else None sample_size = int(payload.get("sample_size") or 5) result: dict[str, object] = {} for source in sources: @@ -25,7 +29,14 @@ def main() -> int: client = build_client(str(source)) caps = client.field_caps(index=str(config["index"]), fields="*") fields = caps.get("fields", {}) - docs = search_docs(source=str(source), date=date, size=sample_size) + docs = search_docs( + source=str(source), + date=date, + size=sample_size, + date_from=date_from, + date_to=date_to, + lookback_days=lookback_days, + ) cases = [extract_case(str(source), deep_parse(doc)) for doc in docs] interesting = [ name @@ -72,4 +83,3 @@ def main() -> int: if __name__ == "__main__": raise SystemExit(main()) - diff --git a/skills/online-mining-v2/scripts/elk_search_cases.py b/skills/online-mining-v2/scripts/elk_search_cases.py index a599833..3de39b2 100644 --- a/skills/online-mining-v2/scripts/elk_search_cases.py +++ b/skills/online-mining-v2/scripts/elk_search_cases.py @@ -38,6 +38,10 @@ def main() -> int: payload = load_json_payload() source = str(payload.get("source") or "pre_processing") date = payload.get("date") + date_from = payload.get("date_from") + date_to = payload.get("date_to") + lookback_days = payload.get("lookback_days") + lookback_days = int(lookback_days) if lookback_days not in (None, "") else None size = int(payload.get("size") or 50) scan_size = int(payload.get("scan_size") or max(size * 5, size)) filters = payload.get("filters") or {} @@ -48,6 +52,9 @@ def main() -> int: date=date, size=scan_size, query_filters=build_index_filters(source, filters), + date_from=date_from, + date_to=date_to, + lookback_days=lookback_days, ) cases = [] for doc in docs: @@ -61,7 +68,7 @@ def main() -> int: emit_success( { "source": source, - "date": date or "past-48h", + "date": date or (f"{date_from}..{date_to}" if date_from and date_to else f"past-{lookback_days}d" if lookback_days else "past-48h"), "scanned": len(docs), "matched": len(cases), "output_path": output_path, @@ -83,4 +90,3 @@ def main() -> int: 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 c117e5c..a210004 100644 --- a/skills/online-mining-v2/scripts/online_mining_common.py +++ b/skills/online-mining-v2/scripts/online_mining_common.py @@ -89,9 +89,32 @@ def source_config(source: str) -> dict[str, Any]: return SOURCE_CONFIGS[source] -def date_range_ms(date_str: str | None) -> tuple[int, int]: +def parse_date(value: str) -> datetime: + text = str(value).strip() + for fmt in ("%Y%m%d", "%Y-%m-%d"): + try: + return datetime.strptime(text, fmt) + except ValueError: + continue + raise OnlineMiningError("date must be YYYYMMDD or YYYY-MM-DD") + + +def date_range_ms(date_str: str | None, *, date_from: str | None = None, date_to: str | None = None, lookback_days: int | None = None) -> tuple[int, int]: + if date_from or date_to: + if not date_from or not date_to: + raise OnlineMiningError("date_from and date_to must be provided together") + start_obj = parse_date(date_from) + end_obj = parse_date(date_to) + timedelta(days=1) + if end_obj <= start_obj: + raise OnlineMiningError("date_to must be greater than or equal to date_from") + return int(start_obj.timestamp() * 1000), int(end_obj.timestamp() * 1000) - 1 + if lookback_days is not None: + if lookback_days <= 0 or lookback_days > 31: + raise OnlineMiningError("lookback_days must be between 1 and 31") + end = datetime.now() + return int((end - timedelta(days=lookback_days)).timestamp() * 1000), int(end.timestamp() * 1000) if date_str: - date_obj = datetime.strptime(str(date_str), "%Y%m%d") + date_obj = parse_date(date_str) start_ms = int(date_obj.timestamp() * 1000) end_ms = int((date_obj + timedelta(days=1)).timestamp() * 1000) - 1 else: @@ -266,10 +289,13 @@ def search_docs( date: str | None, size: int, query_filters: list[dict[str, Any]] | None = None, + date_from: str | None = None, + date_to: str | None = None, + lookback_days: int | None = None, ) -> list[dict[str, Any]]: config = source_config(source) client = build_client(source) - start_ms, end_ms = date_range_ms(date) + start_ms, end_ms = date_range_ms(date, date_from=date_from, date_to=date_to, lookback_days=lookback_days) filters = [{"range": {str(config["time_field"]): {"gte": start_ms, "lte": end_ms}}}] filters.extend(deepcopy(query_filters or [])) body = { @@ -281,7 +307,15 @@ def search_docs( return [deep_parse(hit.get("_source", {})) for hit in response.get("hits", {}).get("hits", [])] -def fetch_one_by_request_id(source: str, request_id: str, date: str | None = None) -> dict[str, Any] | None: +def fetch_one_by_request_id( + source: str, + request_id: str, + date: str | None = None, + *, + date_from: str | None = None, + date_to: str | None = None, + lookback_days: int | None = None, +) -> dict[str, Any] | None: config = source_config(source) id_field = str(config["id_field"]) # 精确查优先用 .keyword；部分索引字段本身就是 keyword，所以再兜底原字段。 @@ -292,56 +326,20 @@ def fetch_one_by_request_id(source: str, request_id: str, date: str | None = Non {"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]) + docs = search_docs( + source=source, + date=date, + size=1, + query_filters=[query_filter], + date_from=date_from, + date_to=date_to, + lookback_days=lookback_days, + ) 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 [] - docs: list[dict[str, Any]] = [] - # arch-flat-nlp-log-f-* 里 session_id 在部分索引没有 .keyword 子字段， - # 所以必须先试 keyword，再兜底原字段。 - for session_field in ("session_id.keyword", "session_id"): - filters = [ - {"term": {session_field: session_id}}, - {"range": {"timestamp": {"lt": before_timestamp}}}, - ] - docs = search_docs(source="main", date=date, size=max(limit * 3, limit), query_filters=filters) - if docs: - break - 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 字段时使用。"""