"""Convert part_06.jsonl → platform-importable JSON. Source schema (per record): question, answer, history (list of {role,content}), trace (list of steps with type ∈ {thought, tool_call, tool_result}), source_id, online_record_id, source_path, source_line, request_id, session_id, user_id, agent_id, business_tag, ability_tree_key, model_name, usage_input_tokens, usage_output_tokens, usage_total_tokens, intent_labels, capability_labels, domain_label, query_quality, query_difficulty, primary_intent, selection_group, offline_split_id, assigned_reviewer Target schema (per record, Chinese keys to match the 13 canonical files): id ← source_id (verbatim; platform PK lookup) 来源 ← business_tag (provenance) 问题 ← question 答案 ← answer 链路数据 ← trace reshaped into [{plan, tools[]}] form so the platform's _extract_tools/_infer_difficulty heuristics can still count tools (key for evaluation). 上下文 ← history (verbatim) tags ← auxiliary metadata JSON-packed into List[str] for future audit / selection-group queries (single tag per record; original fields are NOT lost — see also `source_id`/`online_record_id` round-trip). Pairing strategy for trace → 链路数据: - tool_call and tool_result steps share metadata.tool_call_id - tool_call becomes a phase with tools=[{name, input}]; if matching tool_result exists, output is attached - thought becomes a phase with plan=content, tools=[] - orphan tool_results (no matching call) become a phase with the output as a "result" tool Every source step is consumed exactly once → no data loss; the original trace is still recoverable via online_record_id. """ from __future__ import annotations import json import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "数据测试集" / "part_06.jsonl" DST = ROOT / "数据测试集" / "part_06-importable.json" # Auxiliary fields the platform TestCase ORM doesn't model — packed into # `tags` as a single JSON-encoded string per record so the row remains # importable but the data is preserved for downstream queries / audits. AUX_FIELDS = ( "agent_id", "business_tag", "ability_tree_key", "model_name", "primary_intent", "query_difficulty", "query_quality", "domain_label", "intent_labels", "capability_labels", "selection_group", "offline_split_id", "assigned_reviewer", "source_path", "source_line", "request_id", "session_id", "user_id", "online_record_id", "usage_input_tokens", "usage_output_tokens", "usage_total_tokens", ) def reshape_trace(trace): """Reshape [{type: thought|tool_call|tool_result}, ...] → [{plan, tools[]}, ...].""" if not isinstance(trace, list): return [] # Pre-build id→{name,input} and id→output maps so pairing is robust to # interleaved ordering (record 7 had 3 tool_calls → 3 tool_results). calls_by_id: dict[str, dict] = {} results_by_id: dict[str, str] = {} for step in trace: if not isinstance(step, dict): continue meta = step.get("metadata") or {} tc_id = meta.get("tool_call_id") or "" if step.get("type") == "tool_call": calls_by_id[tc_id] = { "name": step.get("title", "") or "", "input": step.get("content", "") or "", } elif step.get("type") == "tool_result": results_by_id[tc_id] = step.get("content", "") or "" phases: list[dict] = [] consumed_ids: set[str] = set() for step in trace: if not isinstance(step, dict): continue t = step.get("type") if t == "thought": phases.append({"plan": step.get("content", "") or "", "tools": []}) continue if t == "tool_call": tc_id = (step.get("metadata") or {}).get("tool_call_id") or "" tool = dict(calls_by_id.get(tc_id) or { "name": step.get("title", "") or "", "input": step.get("content", "") or "", }) if tc_id in results_by_id: tool["output"] = results_by_id[tc_id] consumed_ids.add(tc_id) phases.append({"plan": "", "tools": [tool]}) continue if t == "tool_result": tc_id = (step.get("metadata") or {}).get("tool_call_id") or "" if tc_id in consumed_ids: # Already attached to the matching tool_call — skip to avoid dup. continue if tc_id and tc_id not in calls_by_id: # Orphan result with a known id (its call was filtered out). tool = { "name": step.get("title", "") or "", "output": step.get("content", "") or "", } phases.append({"plan": "", "tools": [tool]}) continue # Result with no id at all — keep as raw content. phases.append({ "plan": "", "tools": [{ "name": "result", "output": step.get("content", "") or "", }], }) return phases def convert(rec: dict) -> dict: aux = {k: rec[k] for k in AUX_FIELDS if k in rec and rec[k] is not None} return { "id": rec.get("source_id") or "", "来源": rec.get("business_tag") or "iwencai", "问题": rec.get("question") or "", "答案": rec.get("answer") or "", "链路数据": reshape_trace(rec.get("trace") or []), "上下文": rec.get("history") or [], "tags": [json.dumps(aux, ensure_ascii=False)], } def main() -> int: if not SRC.exists(): print(f"❌ source not found: {SRC}", file=sys.stderr) return 1 records = [] with SRC.open(encoding="utf-8") as f: for lineno, line in enumerate(f, 1): line = line.strip() if not line: continue try: records.append(json.loads(line)) except json.JSONDecodeError as e: print(f"⚠️ line {lineno} skipped: {e}", file=sys.stderr) print(f"📥 read {len(records)} records from {SRC.name}") out = [convert(r) for r in records] DST.parent.mkdir(parents=True, exist_ok=True) with DST.open("w", encoding="utf-8") as f: json.dump(out, f, ensure_ascii=False, indent=2) print(f"📤 wrote {len(out)} records → {DST.relative_to(ROOT)}") print(f" size: {DST.stat().st_size:,} bytes") return 0 if __name__ == "__main__": raise SystemExit(main())