From f06e573b042da0b58f3c094c39536e74036b4b4b Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 24 Jun 2026 22:34:04 +0800 Subject: [PATCH] Organize Open-SWE-Traces data prep project --- .gitignore | 10 + README.md | 147 ++++++ SKILL.md | 110 +++++ data/.gitkeep | 1 + .../legacy_subproblem_decomposition_README.md | 241 ++++++++++ pyproject.toml | 29 ++ runs/.gitkeep | 1 + scripts/filtering/audit_native_traces.py | 373 ++++++++++++++++ scripts/probing/analyze_and_decompose.py | 407 +++++++++++++++++ scripts/probing/check_tokenizer_env.py | 8 + scripts/probing/compare_qwen_tokenizers.py | 88 ++++ scripts/probing/count_qwen_tokens_exact.py | 165 +++++++ .../count_qwen_tokens_exact_parallel.py | 180 ++++++++ .../probing/estimate_native_trace_tokens.py | 83 ++++ scripts/probing/inspect_sample.py | 32 ++ scripts/probing/load_qwen_tokenizer.py | 16 + scripts/probing/query_qwen_model_ids.py | 19 + scripts/probing/sample_qwen_token_ratios.py | 105 +++++ .../build_swift_training_probe_5k.py | 259 +++++++++++ .../repurposing/build_swift_validation_500.py | 132 ++++++ .../repurposing/build_training_probe_5k.py | 171 +++++++ scripts/repurposing/coarse_decompose.py | 204 +++++++++ .../repurposing/convert_openswe_to_pi_mono.py | 419 ++++++++++++++++++ src/ti_coding_agent_data_prep/__init__.py | 6 + .../openswe/__init__.py | 6 + .../openswe/constants.py | 32 ++ .../openswe/paths.py | 20 + 27 files changed, 3264 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 SKILL.md create mode 100644 data/.gitkeep create mode 100644 docs/legacy_subproblem_decomposition_README.md create mode 100644 pyproject.toml create mode 100644 runs/.gitkeep create mode 100644 scripts/filtering/audit_native_traces.py create mode 100755 scripts/probing/analyze_and_decompose.py create mode 100644 scripts/probing/check_tokenizer_env.py create mode 100644 scripts/probing/compare_qwen_tokenizers.py create mode 100644 scripts/probing/count_qwen_tokens_exact.py create mode 100644 scripts/probing/count_qwen_tokens_exact_parallel.py create mode 100644 scripts/probing/estimate_native_trace_tokens.py create mode 100644 scripts/probing/inspect_sample.py create mode 100644 scripts/probing/load_qwen_tokenizer.py create mode 100644 scripts/probing/query_qwen_model_ids.py create mode 100644 scripts/probing/sample_qwen_token_ratios.py create mode 100644 scripts/repurposing/build_swift_training_probe_5k.py create mode 100644 scripts/repurposing/build_swift_validation_500.py create mode 100644 scripts/repurposing/build_training_probe_5k.py create mode 100644 scripts/repurposing/coarse_decompose.py create mode 100644 scripts/repurposing/convert_openswe_to_pi_mono.py create mode 100644 src/ti_coding_agent_data_prep/__init__.py create mode 100644 src/ti_coding_agent_data_prep/openswe/__init__.py create mode 100644 src/ti_coding_agent_data_prep/openswe/constants.py create mode 100644 src/ti_coding_agent_data_prep/openswe/paths.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..682018d --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +.venv/ +.ruff_cache/ + +/data/* +!/data/.gitkeep +/runs/* +!/runs/.gitkeep diff --git a/README.md b/README.md new file mode 100644 index 0000000..6b04dcd --- /dev/null +++ b/README.md @@ -0,0 +1,147 @@ +# TI Coding Agent Data Prep + +这个仓库用于整理 `nvidia/Open-SWE-Traces` 的数据准备流程，目标是把原先分散在 probe 目录里的脚本项目化，形成三个清晰阶段： + +1. `probing`：检查数据结构、统计唯一 problem/repo、比较 tokenizer、统计 token。 +2. `filtering`：对原始 trajectory 做 hard filter/audit，筛掉明显不适合作为 SFT 训练样本的轨迹。 +3. `repurposing`：把筛选后的轨迹拆解成阶段化子任务，或导出为 ModelScope-SWIFT / pi-mono 相关格式。 + +## 目录结构 + +```text +. +├── data/ # 本地数据目录，不进 git +│ └── Open-SWE-Traces/ # 建议放 nvidia/Open-SWE-Traces clone 或下载结果 +├── docs/ +│ └── legacy_subproblem_decomposition_README.md +├── runs/ # 所有脚本输出目录，不进 git +├── scripts/ +│ ├── probing/ # 探查、统计、tokenizer/token 相关脚本 +│ ├── filtering/ # native trace audit / hard filter +│ └── repurposing/ # 子问题拆解、SWIFT 导出、pi-mono 格式转换 +├── src/ti_coding_agent_data_prep/ +│ └── openswe/ # 共享常量和路径 helper +├── pyproject.toml +├── README.md +└── SKILL.md +``` + +## 环境部署 + +建议只使用 repo-local 环境，避免污染共享机器： + +```bash +cd /ssd/workspace/yi/ti_coding_agent_data_prep +export http_proxy=http://100.72.0.101:8888 +export https_proxy=http://100.72.0.101:8888 +export HTTP_PROXY=http://100.72.0.101:8888 +export HTTPS_PROXY=http://100.72.0.101:8888 +export HF_ENDPOINT=https://hf-mirror.com + +uv venv .venv --python 3.10 +source .venv/bin/activate +uv pip install -e '.[dev]' +``` + +如果不用 `uv`，也可以用普通 venv 后执行 `pip install -e '.[dev]'`。 + +## 数据准备 + +默认所有脚本都从 `data/Open-SWE-Traces` 读取数据。可以把已有数据软链进来： + +```bash +cd /ssd/workspace/yi/ti_coding_agent_data_prep +ln -s /ssd/workspace/yi/openswetraces_probe/Open-SWE-Traces data/Open-SWE-Traces +``` + +## Probing 入口 + +检查样本结构： + +```bash +python scripts/probing/inspect_sample.py +``` + +统计 unique instance/repo，并随机拆 20 条 fine-grained subproblem： + +```bash +python scripts/probing/analyze_and_decompose.py +``` + +比较 Qwen tokenizer： + +```bash +python scripts/probing/compare_qwen_tokenizers.py +``` + +精确统计 Qwen token 数，输出会持续写入 `runs/native_trace_audit/qwen_exact_token_count.json`： + +```bash +python scripts/probing/count_qwen_tokens_exact_parallel.py \ + --input-root data/Open-SWE-Traces \ + --output runs/native_trace_audit/qwen_exact_token_count.json \ + --model Qwen/Qwen3-32B \ + --workers 12 +``` + +## Filtering 入口 + +对 native trajectory 做 hard filter/audit： + +```bash +python scripts/filtering/audit_native_traces.py \ + --input-root data/Open-SWE-Traces \ + --output-dir runs/native_trace_audit +``` + +主要 hard filter 覆盖： + +- malformed tool call JSON +- tool call/result 对不上 +- tool 名或参数被模型输出污染 +- final patch 为空或不是合理 diff +- patch 文件和 trajectory 明显不一致 +- trajectory 过长 +- 重复、无推进的 tool loop +- unresolved 且明显走偏 + +## Repurposing 入口 + +把 fine-grained 子问题合并为粗阶段： + +```bash +python scripts/repurposing/coarse_decompose.py +``` + +构建 5k ModelScope-SWIFT training probe： + +```bash +python scripts/repurposing/build_swift_training_probe_5k.py +``` + +构建 500 条 validation split： + +```bash +python scripts/repurposing/build_swift_validation_500.py +``` + +SWIFT 导出的关键策略： + +- MiniMax 样本按 thinking 模式处理，`reasoning_content` 会包成 `...` 放入 assistant content。 +- Qwen 样本按 non-thinking 模式处理，不主动加入 reasoning 内容；异常非空 reasoning 会计数。 +- `system`、`user`、`tool` message 标记为 `loss=false`，只让 assistant 输出参与 loss。 + +尝试转换到 pi-mono 风格消息： + +```bash +python scripts/repurposing/convert_openswe_to_pi_mono.py \ + --input-root data/Open-SWE-Traces \ + --output-root runs/pi_mono_converted +``` + +注意：pi-mono 转换脚本保留为研究/兼容入口。由于不同 scaffold 的 system prompt、tool schema、tool call 语义不同，最安全的训练路径仍是优先保留 native trace 格式并筛掉坏样本。 + +## 输出产物 + +`runs/` 和 `data/` 默认不进 git。推荐把所有大文件、parquet、jsonl、token 统计、audit report 都留在这两个目录下。 + diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..825eccc --- /dev/null +++ b/SKILL.md @@ -0,0 +1,110 @@ +# TI Coding Agent Data Prep Skill + +Use this repo for Open-SWE-Traces data preparation before coding-agent SFT experiments. + +## Purpose + +This project organizes the existing Open-SWE-Traces probing, native-trace filtering, and trajectory repurposing scripts into a Python project. It is intended for work on `nvidia/Open-SWE-Traces`, especially: + +- inspecting dataset structure and source splits, +- auditing trajectories for SFT-quality hard filters, +- measuring Qwen tokenizer/token statistics, +- decomposing trajectories into task-oriented phases, +- exporting balanced ModelScope-SWIFT training/validation probes, +- experimenting with pi-mono-style conversion. + +## Repository Layout + +- `scripts/probing/`: dataset inspection, unique-count reports, tokenizer comparison, and token counting. +- `scripts/filtering/`: native trajectory audit and hard filtering. +- `scripts/repurposing/`: subproblem decomposition, coarse task-stage conversion, SWIFT export, and pi-mono conversion. +- `src/ti_coding_agent_data_prep/openswe/`: shared constants and path helpers. +- `data/`: local datasets; ignored by git. +- `runs/`: script outputs; ignored by git. +- `docs/`: retained legacy notes from the original probe workspace. + +## Expected Data Location + +By default scripts read: + +```text +data/Open-SWE-Traces +``` + +On B300, prefer a symlink instead of copying the dataset: + +```bash +ln -s /ssd/workspace/yi/openswetraces_probe/Open-SWE-Traces data/Open-SWE-Traces +``` + +## Environment + +Use a repo-local environment. On B300, set proxy variables before installing or downloading: + +```bash +export http_proxy=http://100.72.0.101:8888 +export https_proxy=http://100.72.0.101:8888 +export HTTP_PROXY=http://100.72.0.101:8888 +export HTTPS_PROXY=http://100.72.0.101:8888 +export HF_ENDPOINT=https://hf-mirror.com +uv venv .venv --python 3.10 +source .venv/bin/activate +uv pip install -e '.[dev]' +``` + +## Main Entrypoints + +Probe one sample per split: + +```bash +python scripts/probing/inspect_sample.py +``` + +Generate unique-count and random-20 decomposition reports: + +```bash +python scripts/probing/analyze_and_decompose.py +``` + +Run native hard-filter audit: + +```bash +python scripts/filtering/audit_native_traces.py --input-root data/Open-SWE-Traces --output-dir runs/native_trace_audit +``` + +Run exact Qwen token counting: + +```bash +python scripts/probing/count_qwen_tokens_exact_parallel.py --input-root data/Open-SWE-Traces --output runs/native_trace_audit/qwen_exact_token_count.json --model Qwen/Qwen3-32B --workers 12 +``` + +Build the balanced 5k SWIFT training probe: + +```bash +python scripts/repurposing/build_swift_training_probe_5k.py +``` + +Build the balanced 500-row validation probe: + +```bash +python scripts/repurposing/build_swift_validation_500.py +``` + +Try pi-mono-style conversion: + +```bash +python scripts/repurposing/convert_openswe_to_pi_mono.py --input-root data/Open-SWE-Traces --output-root runs/pi_mono_converted +``` + +## Filtering Semantics + +The hard filter rejects trajectories with malformed tool-call JSON, tool-call/tool-result mismatch, polluted tool names or arguments, empty or inconsistent final patches, excessive length, repeated no-progress loops, and unresolved off-track traces. + +## Training Format Notes + +For SWIFT exports: + +- MiniMax trajectories are treated as thinking-mode data and `reasoning_content` is wrapped with `...`. +- Qwen trajectories are treated as non-thinking-mode data. +- `system`, `user`, and `tool` messages use `loss=false`; assistant messages use `loss=true`. + diff --git a/data/.gitkeep b/data/.gitkeep new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/data/.gitkeep @@ -0,0 +1 @@ + diff --git a/docs/legacy_subproblem_decomposition_README.md b/docs/legacy_subproblem_decomposition_README.md new file mode 100644 index 0000000..e084390 --- /dev/null +++ b/docs/legacy_subproblem_decomposition_README.md @@ -0,0 +1,241 @@ +# Open-SWE-Traces Subproblem Decomposition Probe + +## Dataset Counts + +Full dataset location: + +```text +/ssd/workspace/yi/openswetraces_probe/Open-SWE-Traces +``` + +Probe outputs: + +```text +/ssd/workspace/yi/openswetraces_probe/subproblem_decomposition +``` + +Global counts from lightweight parquet column scan: + +```text +total trajectories: 207,489 +unique problems: 22,320 +unique repositories: 2,654 +``` + +Per config: + +| config | thinking mode | scaffold | rows | unique problems | unique repos | +|---|---:|---|---:|---:|---:| +| minimax_m25_openhands | thinking | OpenHands | 49,948 | 20,098 | 2,606 | +| minimax_m25_sweagent | thinking | SWE-agent | 57,268 | 20,791 | 2,582 | +| qwen35_openhands | non-thinking | OpenHands | 55,488 | 20,362 | 2,589 | +| qwen35_sweagent | non-thinking | SWE-agent | 44,785 | 18,211 | 2,529 | + +Problem coverage across the four configs: + +```text +appears in 4 configs: 15,605 problems +appears in 3 configs: 4,364 problems +appears in 2 configs: 1,599 problems +appears in 1 config: 752 problems +``` + +The data construction assumption is reflected in the probe: MiniMax rows are treated as thinking-mode trajectories; Qwen rows are treated as non-thinking trajectories. + +## Decomposition Logic + +The script `analyze_and_decompose.py` treats a trajectory as alternating assistant tool calls and tool observations. It assigns each assistant/tool pair to one of these phases: + +```text +understand_task +explore_repo +locate_relevant_code +inspect_code +reproduce_or_probe +edit_solution +verify_solution +cleanup +review_diff +submit +other +``` + +The phase classifier is mostly rule-based: + +- `find`, `ls`, `tree`, `rg --files` -> repository exploration / code location. +- `grep`, `rg`, symbol search -> locate relevant code. +- `view`, `cat`, `sed -n`, `head`, `tail` -> inspect code. +- `python repro`, `test_reproduce`, ad-hoc scripts -> reproduce or probe behavior. +- `str_replace_editor create/insert/str_replace` -> edit solution. +- `pytest`, `go test`, `npm test`, `cargo test`, `mvn test`, etc. -> verify solution. +- `git diff`, `git status` -> review final patch. +- `rm`, cleanup language, temporary file deletion -> cleanup. +- `finish` / `submit` -> submit. + +It also extracts files, commands, short observations, and constructs candidate QA pairs for every segment. + +## Random-20 Probe Result + +Random seed: + +```text +20260623 +``` + +Files produced: + +```text +unique_counts.json +random20_decomposed.json +report.json +``` + +Random 20 summary: + +```text +items: 20 +usable_for_subproblem_training: 20 / 20 +avg_segments_per_trace: 58.8 +``` + +Phase counts across the 20 traces: + +```text +understand_task: 20 +explore_repo: 34 +locate_relevant_code: 143 +inspect_code: 211 +reproduce_or_probe: 317 +edit_solution: 151 +verify_solution: 161 +cleanup: 30 +review_diff: 80 +submit: 29 +``` + +Interpretation: all 20 traces contain enough structure to create subproblem training data. The current segmentation is intentionally low-level and over-segments long agent loops. It is good for action-level supervision, but high-level SFT should merge adjacent phases into coarser task-oriented chunks. + +## Recommended Subproblem Formats + +### QA Format + +Use this when training repository-aware reasoning without replaying every tool result. + +```json +{ + "question": "In repo owner/name, how can we locate the code responsible for ?", + "answer": { + "phase": "locate_relevant_code", + "files": ["..."], + "commands": ["rg ...", "find ..."], + "evidence": ["symbol X is defined in file Y"], + "next_step": "inspect_code" + } +} +``` + +Best phases for QA: + +```text +locate_relevant_code +inspect_code +reproduce_or_probe +verify_solution +review_diff +``` + +### Question-Trajectory Format + +Use this when training tool-use policy or pi-mono-like agent behavior. + +```json +{ + "question": "In repo owner/name, implement the edit that fixes .", + "trajectory": [ + {"role": "assistant", "content": "...", "tool_calls": [...]}, + {"role": "tool", "content": "..."} + ], + "target_patch": "diff --git ...", + "phase": "edit_solution" +} +``` + +Best phases for Question-Trajectory: + +```text +explore_repo -> locate_relevant_code -> inspect_code +inspect_code -> edit_solution +edit_solution -> verify_solution +verify_solution -> review_diff -> submit +``` + +## Conversion Notes for pi-mono + +Tool dialect normalization is required: + +| Source tool | pi-mono target | +|---|---| +| `execute_bash`, `bash` | `bash` | +| `str_replace_editor` `view` directory | `ls` | +| `str_replace_editor` `view` file/range | `read` | +| `str_replace_editor` `create` | `write` | +| `str_replace_editor` `str_replace` / `insert` | `edit` | +| `finish`, `submit` | final answer / stop | + +Path normalization is required: + +```text +/testbed -> repo root +/workspace/__1.0 -> repo root +absolute file paths -> repo-relative paths where possible +``` + +For Qwen3.6 27B SFT: + +- Qwen traces should be used as non-thinking examples. +- MiniMax traces should be used as thinking examples, with `reasoning_content` mapped to the thinking channel. +- For long traces, prefer subproblem chunks over full-trajectory examples. +- Use resolved=1 as the highest-quality subset, but unresolved traces can still train exploration, localization, and verification behavior if segmented before the bad final edit. + +## Next Improvements + +1. Add a second-pass merger that combines adjacent low-level segments into 6-10 high-level chunks per trace. +2. Add semantic labels using issue text and touched files, e.g. `api_contract_update`, `test_regression_reproduction`, `parser_edge_case_fix`. +3. Score each segment for trainability: contains objective, command/action, evidence, and outcome. +4. Export two datasets: + - `subproblem_qa.jsonl` + - `subproblem_question_trajectory.jsonl` + +## Coarse Task-Category Decomposition + +A second script, `coarse_decompose.py`, aggregates fine segments into six coarse task categories per trace: + +```text +understand +locate +diagnose +fix +verify +finalize +``` + +For the random-20 sample, this reduced 1,176 fine segments to 120 coarse segments: + +```text +avg fine segments per trace: 58.8 +avg coarse segments per trace: 6.0 +``` + +This coarse output is better for SFT subproblem construction. Each coarse segment contains deduplicated files, commands, observations, assistant intents, and two training views: + +```text +training_views.qa +training_views.question_trajectory +``` + +Main output: + +```text +random20_coarse_decomposed.json +coarse_report.json +``` diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..28efb89 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,29 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "ti-coding-agent-data-prep" +version = "0.1.0" +description = "Open-SWE-Traces probing, filtering, and repurposing utilities for coding-agent SFT data preparation." +readme = "README.md" +requires-python = ">=3.10" +dependencies = [ + "datasets>=2.19.0", + "huggingface-hub>=0.23.0", + "pyarrow>=15.0.0", + "transformers>=4.52.0", +] + +[project.optional-dependencies] +dev = [ + "ruff>=0.5.0", +] + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.ruff] +line-length = 120 +target-version = "py310" + diff --git a/runs/.gitkeep b/runs/.gitkeep new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/runs/.gitkeep @@ -0,0 +1 @@ + diff --git a/scripts/filtering/audit_native_traces.py b/scripts/filtering/audit_native_traces.py new file mode 100644 index 0000000..d920e1d --- /dev/null +++ b/scripts/filtering/audit_native_traces.py @@ -0,0 +1,373 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import re +from collections import Counter, defaultdict, deque +from pathlib import Path +from typing import Any + +import pyarrow.parquet as pq + + +CONFIGS = [ + "minimax_m25_openhands_trajectories", + "minimax_m25_sweagent_trajectories", + "qwen35_openhands_trajectories", + "qwen35_sweagent_trajectories", +] + +KNOWN_TOOLS = { + "bash", + "execute_bash", + "str_replace_editor", + "finish", + "submit", + "think", + # Rare OpenHands/web tools seen in traces. They are not necessarily bad, + # but should stay visible in reports for downstream filtering decisions. + "fetch", +} + +NO_RESULT_TOOLS = {"finish", "submit", "think"} + +SOURCE_TOOL_COMMANDS = { + "str_replace_editor": { + "view", + "create", + "str_replace", + "insert", + "undo_edit", + } +} + +MALFORMED_MARKERS = [ + " argparse.Namespace: + parser = argparse.ArgumentParser(description="Audit native Open-SWE trajectories for SFT filtering.") + parser.add_argument("--input-root", type=Path, default=Path("data/Open-SWE-Traces")) + parser.add_argument("--output-dir", type=Path, default=Path("runs/native_trace_audit")) + parser.add_argument("--limit", type=int, default=0, help="Optional row limit for smoke tests.") + parser.add_argument("--batch-size", type=int, default=512) + parser.add_argument("--max-bad-examples", type=int, default=200) + parser.add_argument("--long-turn-threshold", type=int, default=300) + parser.add_argument("--long-char-threshold", type=int, default=900_000) + parser.add_argument("--repeat-window", type=int, default=24) + parser.add_argument("--repeat-threshold", type=int, default=10) + return parser.parse_args() + + +def json_dumps_compact(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str) + + +def parse_tool_arguments(raw: Any) -> tuple[Any, str | None]: + if raw is None: + return {}, None + if isinstance(raw, dict): + return raw, None + if not isinstance(raw, str): + return None, "tool_arguments_not_string_or_object" + try: + parsed = json.loads(raw) + except Exception: + return None, "malformed_tool_call_json" + if not isinstance(parsed, dict): + return parsed, "tool_arguments_json_not_object" + return parsed, None + + +def tool_call_signature(call: dict[str, Any]) -> str: + function = call.get("function") or {} + name = function.get("name") or call.get("name") + args_raw = function.get("arguments", call.get("arguments")) + args, _ = parse_tool_arguments(args_raw) + return json_dumps_compact({"name": name, "arguments": args if args is not None else args_raw}) + + +def content_text(msg: dict[str, Any]) -> str: + parts = [] + for key in ("content", "reasoning_content"): + value = msg.get(key) + if isinstance(value, str): + parts.append(value) + return "\n".join(parts) + + +def patch_empty_or_inconsistent(row: dict[str, Any]) -> list[str]: + issues = [] + patch = row.get("model_patch") or "" + resolved = int(row.get("resolved") or 0) + if not patch.strip(): + issues.append("final_patch_empty") + if resolved == 1: + issues.append("resolved_but_patch_empty") + return issues + if "diff --git " not in patch and not re.search(r"^--- .+\n\+\+\+ ", patch, re.M): + issues.append("final_patch_not_unified_diff") + trajectory_text = "\n".join(content_text(m) for m in row.get("trajectory") or []) + patch_files = set(re.findall(r"^diff --git a/(.*?) b/(.*?)$", patch, re.M)) + flat_patch_files = {p for pair in patch_files for p in pair} + if flat_patch_files: + mentioned = sum(1 for path in flat_patch_files if path and path in trajectory_text) + if mentioned == 0: + issues.append("patch_files_not_mentioned_in_trajectory") + return issues + + +def detect_unproductive_loop(trajectory: list[dict[str, Any]], repeat_window: int, repeat_threshold: int) -> bool: + recent: deque[str] = deque(maxlen=repeat_window) + counts = Counter() + for msg in trajectory: + if msg.get("role") != "assistant": + continue + for call in msg.get("tool_calls") or []: + sig = tool_call_signature(call) + recent.append(sig) + counts = Counter(recent) + if counts[sig] >= repeat_threshold: + return True + return False + + +def audit_row(row: dict[str, Any], args: argparse.Namespace) -> tuple[list[str], Counter]: + issues: list[str] = [] + details = Counter() + trajectory = row.get("trajectory") or [] + pending = 0 + skipped_tool_results = 0 + tool_calls = 0 + tool_results = 0 + assistant_turns = 0 + total_chars = 0 + edit_calls = 0 + verify_calls = 0 + repeated_signatures = Counter() + + for msg in trajectory: + role = msg.get("role") + total_chars += len(content_text(msg)) + if role == "assistant": + assistant_turns += 1 + calls = msg.get("tool_calls") or [] + tool_calls += len(calls) + for call in calls: + repeated_signatures[tool_call_signature(call)] += 1 + function = call.get("function") or {} + name = function.get("name") or call.get("name") + if not name or not isinstance(name, str): + issues.append("tool_name_missing_or_invalid") + continue + if name in NO_RESULT_TOOLS: + skipped_tool_results += 1 + else: + pending += 1 + if name not in KNOWN_TOOLS: + issues.append(f"unknown_tool_name:{truncate(name)}") + raw_args = function.get("arguments", call.get("arguments")) + parsed_args, parse_issue = parse_tool_arguments(raw_args) + if parse_issue: + issues.append(parse_issue) + raw_args_text = raw_args if isinstance(raw_args, str) else json_dumps_compact(raw_args) + if any(marker in raw_args_text for marker in MALFORMED_MARKERS): + issues.append("tool_arguments_polluted_by_markup_or_text") + if isinstance(parsed_args, dict) and name in SOURCE_TOOL_COMMANDS: + command = parsed_args.get("command") + if command not in SOURCE_TOOL_COMMANDS[name]: + issues.append(f"unknown_{name}_command:{truncate(command)}") + if name in {"bash", "execute_bash"} and isinstance(parsed_args, dict): + command = str(parsed_args.get("command") or "") + if re.search(r"\b(pytest|go test|npm test|pnpm test|yarn test|cargo test|mvn test|gradle test|tox)\b", command): + verify_calls += 1 + if re.search(r"\b(apply_patch|python\s+-|perl\s+-0pi|sed\s+-i)\b", command): + edit_calls += 1 + if name == "str_replace_editor" and isinstance(parsed_args, dict): + if parsed_args.get("command") in {"create", "str_replace", "insert", "undo_edit"}: + edit_calls += 1 + elif role == "tool": + tool_results += 1 + if pending <= 0: + if skipped_tool_results > 0: + skipped_tool_results -= 1 + continue + issues.append("tool_result_without_pending_tool_call") + else: + pending -= 1 + + if pending: + issues.append("tool_call_without_tool_result") + if tool_calls == 0: + issues.append("no_tool_calls") + if len(trajectory) >= args.long_turn_threshold or total_chars >= args.long_char_threshold: + issues.append("trajectory_too_long") + if detect_unproductive_loop(trajectory, args.repeat_window, args.repeat_threshold): + issues.append("repeated_tool_call_loop") + + resolved = int(row.get("resolved") or 0) + if resolved != 1: + issues.append("unresolved") + if edit_calls > 0 and verify_calls == 0: + issues.append("unresolved_edited_without_verification") + if "repeated_tool_call_loop" in issues or "trajectory_too_long" in issues: + issues.append("unresolved_likely_off_track") + + issues.extend(patch_empty_or_inconsistent(row)) + details.update( + { + "tool_calls": tool_calls, + "tool_results": tool_results, + "assistant_turns": assistant_turns, + "total_chars": total_chars, + "edit_calls": edit_calls, + "verify_calls": verify_calls, + "trajectory_len": len(trajectory), + } + ) + return sorted(set(issues)), details + + +def truncate(value: Any, limit: int = 120) -> str: + text = str(value) + return text if len(text) <= limit else text[:limit] + "..." + + +def audit_file( + file: Path, + config: str, + args: argparse.Namespace, + remaining: int | None, + bad_examples: list[dict[str, Any]], +) -> tuple[Counter, int | None]: + stats = Counter() + issue_counts = Counter() + parquet = pq.ParquetFile(file) + for batch in parquet.iter_batches(batch_size=args.batch_size): + rows = batch.to_pylist() + if remaining is not None: + rows = rows[:remaining] + for row in rows: + issues, details = audit_row(row, args) + hard_issues = hard_filter_issues(issues) + stats["rows"] += 1 + stats[f"resolved_{int(row.get('resolved') or 0)}"] += 1 + if hard_issues: + stats["filtered_rows"] += 1 + issue_counts.update(hard_issues) + if len(bad_examples) < args.max_bad_examples: + bad_examples.append( + { + "config": config, + "instance_id": row.get("instance_id"), + "repo": row.get("repo"), + "trajectory_id": row.get("trajectory_id"), + "resolved": int(row.get("resolved") or 0), + "hard_filter_issues": hard_issues, + "all_flags": issues, + "details": dict(details), + "model_patch_head": (row.get("model_patch") or "")[:1000], + } + ) + if remaining is not None: + remaining -= len(rows) + if remaining <= 0: + break + stats["kept_rows"] = stats["rows"] - stats["filtered_rows"] + for issue, count in issue_counts.items(): + stats[f"issue:{issue}"] = count + return stats, remaining + + +def hard_filter_issues(issues: list[str]) -> list[str]: + hard = [] + prefixes = ( + "unknown_tool_name:", + "unknown_str_replace_editor_command:", + ) + exact = { + "malformed_tool_call_json", + "tool_arguments_not_string_or_object", + "tool_arguments_json_not_object", + "tool_arguments_polluted_by_markup_or_text", + "tool_name_missing_or_invalid", + "tool_result_without_pending_tool_call", + "tool_call_without_tool_result", + "final_patch_empty", + "resolved_but_patch_empty", + "final_patch_not_unified_diff", + "patch_files_not_mentioned_in_trajectory", + "trajectory_too_long", + "repeated_tool_call_loop", + "unresolved_likely_off_track", + } + for issue in issues: + if issue in exact or issue.startswith(prefixes): + hard.append(issue) + return hard + + +def main() -> int: + args = parse_args() + args.output_dir.mkdir(parents=True, exist_ok=True) + data_root = args.input_root / "data" + report: dict[str, Any] = { + "input_root": str(args.input_root), + "output_dir": str(args.output_dir), + "filters": { + "long_turn_threshold": args.long_turn_threshold, + "long_char_threshold": args.long_char_threshold, + "repeat_window": args.repeat_window, + "repeat_threshold": args.repeat_threshold, + }, + "configs": {}, + } + bad_examples: list[dict[str, Any]] = [] + remaining = args.limit or None + + total = Counter() + for config in CONFIGS: + config_stats = Counter() + for file in sorted((data_root / config).glob("*.parquet")): + if remaining is not None and remaining <= 0: + break + file_stats, remaining = audit_file(file, config, args, remaining, bad_examples) + config_stats.update(file_stats) + print(json.dumps({"file": str(file), "stats": dict(file_stats)}, ensure_ascii=False), flush=True) + report["configs"][config] = summarize(config_stats) + total.update(config_stats) + report["total"] = summarize(total) + + (args.output_dir / "native_trace_audit_report.json").write_text( + json.dumps(report, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + (args.output_dir / "bad_examples_sample.json").write_text( + json.dumps(bad_examples, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + print(json.dumps({"report": str(args.output_dir / "native_trace_audit_report.json")}, ensure_ascii=False)) + return 0 + + +def summarize(stats: Counter) -> dict[str, Any]: + rows = stats.get("rows", 0) + filtered = stats.get("filtered_rows", 0) + summary = dict(stats) + summary["filtered_pct"] = round(100 * filtered / rows, 4) if rows else 0 + summary["kept_pct"] = round(100 * (rows - filtered) / rows, 4) if rows else 0 + top_issues = [] + for key, value in stats.most_common(): + if key.startswith("issue:"): + top_issues.append({"issue": key[len("issue:") :], "count": value}) + summary["top_issues"] = top_issues[:50] + return summary + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/probing/analyze_and_decompose.py b/scripts/probing/analyze_and_decompose.py new file mode 100755 index 0000000..5b44dd2 --- /dev/null +++ b/scripts/probing/analyze_and_decompose.py @@ -0,0 +1,407 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import json +import random +import re +from collections import Counter, defaultdict +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import pyarrow.parquet as pq + + +DATASET_DIR = Path("data/Open-SWE-Traces") +OUT_DIR = Path("runs/subproblem_decomposition") +SEED = 20260623 + + +CONFIGS = { + "minimax_m25_openhands": { + "path": "data/minimax_m25_openhands_trajectories", + "scaffold": "openhands", + "model_family": "minimax_m25", + "thinking_mode": "thinking", + }, + "minimax_m25_sweagent": { + "path": "data/minimax_m25_sweagent_trajectories", + "scaffold": "sweagent", + "model_family": "minimax_m25", + "thinking_mode": "thinking", + }, + "qwen35_openhands": { + "path": "data/qwen35_openhands_trajectories", + "scaffold": "openhands", + "model_family": "qwen35_122b", + "thinking_mode": "non_thinking", + }, + "qwen35_sweagent": { + "path": "data/qwen35_sweagent_trajectories", + "scaffold": "sweagent", + "model_family": "qwen35_122b", + "thinking_mode": "non_thinking", + }, +} + + +PHASE_ORDER = [ + "understand_task", + "explore_repo", + "locate_relevant_code", + "inspect_code", + "reproduce_or_probe", + "edit_solution", + "verify_solution", + "cleanup", + "review_diff", + "submit", + "other", +] + + +def read_tool_call(message: dict[str, Any]) -> tuple[str | None, dict[str, Any]]: + tool_calls = message.get("tool_calls") or [] + if not tool_calls: + return None, {} + call = tool_calls[0] or {} + fn = call.get("function") or {} + name = fn.get("name") + args = fn.get("arguments") or "{}" + try: + parsed = json.loads(args) if isinstance(args, str) and args else {} + except json.JSONDecodeError: + parsed = {"_raw": args} + return name, parsed + + +def command_text(tool_name: str | None, args: dict[str, Any]) -> str: + if not tool_name: + return "" + if "command" in args: + return str(args["command"]) + if "path" in args: + return f"{tool_name} {args.get('command', '')} {args.get('path', '')}".strip() + if "message" in args: + return str(args["message"]) + return json.dumps(args, ensure_ascii=False) + + +def classify_tool(tool_name: str | None, args: dict[str, Any], assistant_text: str = "") -> str: + cmd = command_text(tool_name, args).lower() + text = assistant_text.lower() + joined = f"{cmd}\n{text}" + + if tool_name in {"finish", "submit"}: + return "submit" + if re.search(r"\b(git diff|git status|diff\b)", cmd): + return "review_diff" + if re.search(r"\b(rm|git restore|git checkout --)\b", cmd) or "clean up" in text: + return "cleanup" + if re.search(r"\b(pytest|go test|npm test|pnpm test|yarn test|cargo test|mvn test|gradle test|phpunit|rspec|tox|unittest)\b", cmd): + return "verify_solution" + if re.search(r"\bpython .*repro|reproduce|test_reproduce|final_verification|verify|run.*script\b", joined): + return "reproduce_or_probe" + if tool_name == "str_replace_editor": + editor_cmd = str(args.get("command", "")).lower() + if editor_cmd in {"str_replace", "insert", "create"} or "file_text" in args or "new_str" in args: + return "edit_solution" + if editor_cmd == "view": + path = str(args.get("path", "")).lower() + if path.endswith((".py", ".go", ".ts", ".tsx", ".js", ".jsx", ".rs", ".java", ".php", ".c", ".cc", ".cpp", ".h", ".hpp", ".md", ".rst")): + return "inspect_code" + return "explore_repo" + if re.search(r"\b(sed -n|cat |less |head |tail |nl |grep -n|rg |ripgrep)\b", cmd): + return "inspect_code" + if re.search(r"\b(find |ls |tree |grep -r|rg --files)\b", cmd): + return "locate_relevant_code" + if re.search(r"\b(grep |rg )", cmd): + return "locate_relevant_code" + if tool_name in {"execute_bash", "bash"}: + return "reproduce_or_probe" + return "other" + + +def task_prompt(user_content: str) -> str: + markers = [ + ("", ""), + ("", ""), + ] + for start, end in markers: + if start in user_content and end in user_content: + return user_content.split(start, 1)[1].split(end, 1)[0].strip() + return user_content[:2000].strip() + + +def phase_goal(phase: str, repo: str, task: str, files: list[str], commands: list[str]) -> str: + file_hint = f" in {', '.join(files[:3])}" if files else "" + if phase == "understand_task": + return f"Understand the requested fix for {repo}." + if phase == "explore_repo": + return f"Map the repository structure and identify likely implementation areas for the task." + if phase == "locate_relevant_code": + return f"Search for symbols, files, or modules related to the issue." + if phase == "inspect_code": + return f"Read the relevant code{file_hint} and infer current behavior." + if phase == "reproduce_or_probe": + return "Run focused commands or reproduction snippets to observe the bug or current behavior." + if phase == "edit_solution": + return f"Apply code changes{file_hint} that address the issue." + if phase == "verify_solution": + return "Run targeted tests or validation commands to check the fix." + if phase == "cleanup": + return "Remove temporary reproduction files or revert unrelated artifacts." + if phase == "review_diff": + return "Inspect the final diff and working tree before submission." + if phase == "submit": + return "Submit or finish the solution with a concise summary." + return "Continue task-specific investigation." + + +def extract_paths(text: str) -> list[str]: + candidates = re.findall(r"(?:(?:/workspace|/testbed)[^\s,:;`'\")]+|[\w./-]+\.(?:py|go|ts|tsx|js|jsx|rs|java|php|c|cc|cpp|h|hpp|md|rst))", text) + cleaned = [] + for c in candidates: + c = c.rstrip(".,)]}") + if c not in cleaned: + cleaned.append(c) + return cleaned[:20] + + +def compact_text(value: Any, limit: int = 500) -> str: + if value is None: + return "" + text = str(value).strip() + return text[:limit] + ("..." if len(text) > limit else "") + + +def decompose(row: dict[str, Any], config_name: str, config: dict[str, str]) -> dict[str, Any]: + trajectory = row["trajectory"] or [] + user = next((m for m in trajectory if m.get("role") == "user"), {}) + task = task_prompt(user.get("content") or "") + + segments: list[dict[str, Any]] = [] + current: dict[str, Any] | None = None + + def close(end_idx: int) -> None: + nonlocal current + if not current: + return + current["turn_range"][1] = end_idx + current["goal"] = phase_goal( + current["phase"], + row["repo"], + task, + current["files"], + current["commands"], + ) + segments.append(current) + current = None + + for idx, msg in enumerate(trajectory): + role = msg.get("role") + if role in {"system", "user"}: + phase = "understand_task" + assistant_text = compact_text(msg.get("content"), 400) + tool_name, args = None, {} + elif role == "assistant": + assistant_text = compact_text(msg.get("content") or msg.get("reasoning_content"), 700) + tool_name, args = read_tool_call(msg) + phase = classify_tool(tool_name, args, assistant_text) + else: + if current: + obs = compact_text(msg.get("content"), 700) + if obs: + current["observations"].append(obs) + current["files"].extend(p for p in extract_paths(obs) if p not in current["files"]) + continue + + if current is None or phase != current["phase"]: + close(idx - 1) + current = { + "phase": phase, + "goal": "", + "turn_range": [idx, idx], + "commands": [], + "tool_names": [], + "files": [], + "assistant_intents": [], + "observations": [], + } + + current["turn_range"][1] = idx + if assistant_text: + current["assistant_intents"].append(assistant_text) + if tool_name: + current["tool_names"].append(tool_name) + cmd = command_text(tool_name, args) + if cmd: + current["commands"].append(compact_text(cmd, 600)) + current["files"].extend(p for p in extract_paths(cmd) if p not in current["files"]) + + close(len(trajectory) - 1) + + # Merge very small adjacent "other" segments into neighbors where possible. + merged: list[dict[str, Any]] = [] + for seg in segments: + if seg["phase"] == "other" and merged: + prev = merged[-1] + prev["turn_range"][1] = seg["turn_range"][1] + prev["commands"].extend(seg["commands"]) + prev["tool_names"].extend(seg["tool_names"]) + prev["files"].extend(p for p in seg["files"] if p not in prev["files"]) + prev["assistant_intents"].extend(seg["assistant_intents"]) + prev["observations"].extend(seg["observations"]) + else: + merged.append(seg) + + qa_pairs = [] + for seg in merged: + qa_pairs.append( + { + "question": f"In repository {row['repo']}, how can we {seg['goal'][0].lower() + seg['goal'][1:]}", + "answer_outline": { + "phase": seg["phase"], + "files": seg["files"][:8], + "commands": seg["commands"][:8], + "observations": seg["observations"][:4], + }, + } + ) + + return { + "config": config_name, + "scaffold": config["scaffold"], + "model_family": config["model_family"], + "thinking_mode": config["thinking_mode"], + "instance_id": row["instance_id"], + "repo": row["repo"], + "language": row["language"], + "license": row["license"], + "trajectory_id": row["trajectory_id"], + "resolved": row["resolved"], + "metadata": row.get("metadata"), + "trajectory_len": len(trajectory), + "task": task, + "model_patch_head": compact_text(row.get("model_patch"), 1200), + "segments": merged, + "qa_pairs": qa_pairs, + "quality_flags": quality_flags(merged, row), + } + + +def quality_flags(segments: list[dict[str, Any]], row: dict[str, Any]) -> dict[str, Any]: + phases = {s["phase"] for s in segments} + return { + "has_locate_or_inspect": bool(phases & {"locate_relevant_code", "inspect_code", "explore_repo"}), + "has_edit": "edit_solution" in phases, + "has_verify": "verify_solution" in phases or "reproduce_or_probe" in phases, + "has_submit": "submit" in phases, + "has_patch": bool(row.get("model_patch")), + "usable_for_subproblem_training": bool( + phases & {"locate_relevant_code", "inspect_code", "edit_solution"} + ) + and bool(row.get("model_patch")), + } + + +def collect_unique_counts() -> dict[str, Any]: + global_instances: set[str] = set() + global_repos: set[str] = set() + by_config = {} + by_instance_configs: dict[str, set[str]] = defaultdict(set) + + for name, cfg in CONFIGS.items(): + inst: set[str] = set() + repos: set[str] = set() + resolved = Counter() + languages = Counter() + rows = 0 + for file in sorted((DATASET_DIR / cfg["path"]).glob("*.parquet")): + table = pq.read_table(file, columns=["instance_id", "repo", "language", "resolved"]) + for r in table.to_pylist(): + rows += 1 + inst.add(r["instance_id"]) + repos.add(r["repo"]) + global_instances.add(r["instance_id"]) + global_repos.add(r["repo"]) + by_instance_configs[r["instance_id"]].add(name) + resolved[str(r["resolved"])] += 1 + languages[r["language"]] += 1 + by_config[name] = { + "rows": rows, + "unique_instances": len(inst), + "unique_repos": len(repos), + "resolved_counts": dict(resolved), + "language_counts": dict(languages), + "thinking_mode": cfg["thinking_mode"], + "scaffold": cfg["scaffold"], + "model_family": cfg["model_family"], + } + + overlap = Counter(len(v) for v in by_instance_configs.values()) + return { + "total_rows": sum(v["rows"] for v in by_config.values()), + "global_unique_instances": len(global_instances), + "global_unique_repos": len(global_repos), + "instance_coverage_across_configs": dict(overlap), + "by_config": by_config, + } + + +def load_random_rows(total: int = 20) -> list[tuple[str, dict[str, str], dict[str, Any]]]: + rng = random.Random(SEED) + per_config = total // len(CONFIGS) + remainder = total % len(CONFIGS) + output = [] + for idx, (name, cfg) in enumerate(CONFIGS.items()): + need = per_config + (1 if idx < remainder else 0) + files = sorted((DATASET_DIR / cfg["path"]).glob("*.parquet")) + row_counts = [pq.ParquetFile(f).metadata.num_rows for f in files] + total_rows = sum(row_counts) + targets = sorted(rng.sample(range(total_rows), need)) + cursor = 0 + for file, count in zip(files, row_counts): + local_targets = [t - cursor for t in targets if cursor <= t < cursor + count] + if local_targets: + table = pq.read_table(file) + rows = table.to_pylist() + for local in local_targets: + output.append((name, cfg, rows[local])) + cursor += count + rng.shuffle(output) + return output + + +def main() -> int: + OUT_DIR.mkdir(parents=True, exist_ok=True) + counts = collect_unique_counts() + (OUT_DIR / "unique_counts.json").write_text(json.dumps(counts, indent=2, ensure_ascii=False)) + + decomposed = [decompose(row, name, cfg) for name, cfg, row in load_random_rows(20)] + (OUT_DIR / "random20_decomposed.json").write_text(json.dumps(decomposed, indent=2, ensure_ascii=False)) + + phase_counts = Counter() + usable = 0 + for item in decomposed: + usable += int(item["quality_flags"]["usable_for_subproblem_training"]) + for seg in item["segments"]: + phase_counts[seg["phase"]] += 1 + + report = { + "seed": SEED, + "unique_counts": counts, + "random20_summary": { + "items": len(decomposed), + "usable_for_subproblem_training": usable, + "phase_counts": dict(phase_counts), + "avg_segments_per_trace": sum(len(x["segments"]) for x in decomposed) / len(decomposed), + }, + } + (OUT_DIR / "report.json").write_text(json.dumps(report, indent=2, ensure_ascii=False)) + print(json.dumps(report, indent=2, ensure_ascii=False)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/probing/check_tokenizer_env.py b/scripts/probing/check_tokenizer_env.py new file mode 100644 index 0000000..e524bee --- /dev/null +++ b/scripts/probing/check_tokenizer_env.py @@ -0,0 +1,8 @@ +#!/usr/bin/env python3 +mods = ["transformers", "tokenizers", "modelscope", "sentencepiece"] +for name in mods: + try: + mod = __import__(name) + print(name, "OK", getattr(mod, "__version__", "")) + except Exception as exc: + print(name, "NO", type(exc).__name__, str(exc)[:120]) diff --git a/scripts/probing/compare_qwen_tokenizers.py b/scripts/probing/compare_qwen_tokenizers.py new file mode 100644 index 0000000..823e447 --- /dev/null +++ b/scripts/probing/compare_qwen_tokenizers.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +from huggingface_hub import snapshot_download + + +MODELS = ["Qwen/Qwen3.5-27B", "Qwen/Qwen3.6-27B"] +ALLOW = [ + "tokenizer.json", + "tokenizer_config.json", + "vocab.json", + "merges.txt", + "special_tokens_map.json", + "config.json", + "generation_config.json", +] + + +def sha256(path: Path) -> str: + h = hashlib.sha256() + with path.open("rb") as f: + for chunk in iter(lambda: f.read(1024 * 1024), b""): + h.update(chunk) + return h.hexdigest() + + +def main() -> int: + base = Path("runs/qwen_tokenizer_compare") + base.mkdir(parents=True, exist_ok=True) + snapshots = {} + for model in MODELS: + local = snapshot_download( + model, + allow_patterns=ALLOW, + local_dir=base / model.replace("/", "__"), + local_dir_use_symlinks=False, + ) + snapshots[model] = Path(local) + print("DOWNLOADED", model, local) + + report = {"models": {}, "comparisons": {}} + for model, path in snapshots.items(): + files = {} + for name in ALLOW: + f = path / name + if f.exists(): + files[name] = {"sha256": sha256(f), "size": f.stat().st_size} + tokenizer_config = {} + tc = path / "tokenizer_config.json" + if tc.exists(): + tokenizer_config = json.loads(tc.read_text(encoding="utf-8")) + report["models"][model] = { + "path": str(path), + "files": files, + "tokenizer_class": tokenizer_config.get("tokenizer_class"), + "chat_template": tokenizer_config.get("chat_template"), + "eos_token": tokenizer_config.get("eos_token"), + "pad_token": tokenizer_config.get("pad_token"), + "additional_special_tokens": tokenizer_config.get("additional_special_tokens"), + } + + a, b = MODELS + a_files = report["models"][a]["files"] + b_files = report["models"][b]["files"] + all_files = sorted(set(a_files) | set(b_files)) + for name in all_files: + report["comparisons"][name] = { + "same": a_files.get(name, {}).get("sha256") == b_files.get(name, {}).get("sha256"), + a: a_files.get(name), + b: b_files.get(name), + } + report["same_tokenizer_core"] = all( + report["comparisons"].get(name, {}).get("same") for name in ["tokenizer.json", "vocab.json", "merges.txt"] + ) + report["same_tokenizer_config"] = report["comparisons"].get("tokenizer_config.json", {}).get("same") + report["same_chat_template"] = report["models"][a].get("chat_template") == report["models"][b].get("chat_template") + out = base / "comparison_report.json" + out.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8") + print(json.dumps(report, indent=2, ensure_ascii=False)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/probing/count_qwen_tokens_exact.py b/scripts/probing/count_qwen_tokens_exact.py new file mode 100644 index 0000000..4ff20a7 --- /dev/null +++ b/scripts/probing/count_qwen_tokens_exact.py @@ -0,0 +1,165 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import sys +import time +from collections import Counter +from pathlib import Path + +import pyarrow.parquet as pq +from transformers import AutoTokenizer + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT / "scripts" / "filtering")) +import audit_native_traces as audit # noqa: E402 + + +class AuditArgs: + long_turn_threshold = 300 + long_char_threshold = 900_000 + repeat_window = 24 + repeat_threshold = 10 + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Exact Qwen tokenizer count for kept Open-SWE traces.") + parser.add_argument("--input-root", type=Path, default=Path("data/Open-SWE-Traces")) + parser.add_argument("--output", type=Path, default=Path("runs/native_trace_audit/qwen_exact_token_count.json")) + parser.add_argument("--model", default="Qwen/Qwen3-32B") + parser.add_argument("--batch-size", type=int, default=512) + parser.add_argument("--tokenize-batch-size", type=int, default=2048) + parser.add_argument("--flush-interval", type=int, default=25_000) + return parser.parse_args() + + +def message_text(message: dict) -> str: + text = "" + for key in ("content", "reasoning_content"): + value = message.get(key) + if isinstance(value, str): + text += value + if message.get("tool_calls"): + text += json.dumps(message.get("tool_calls"), ensure_ascii=False) + return text + + +def add_tokenized_batch(tokenizer, texts: list[str], roles: list[str], stats: Counter) -> None: + if not texts: + return + encoded = tokenizer(texts, add_special_tokens=False, return_attention_mask=False, return_token_type_ids=False) + for role, ids in zip(roles, encoded["input_ids"], strict=True): + n = len(ids) + stats["tokens_total"] += n + stats[f"tokens_{role}"] += n + + +def update_output(path: Path, report: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8") + tmp.replace(path) + + +def summarize(stats: Counter) -> dict: + out = dict(stats) + kept = stats.get("kept", 0) + total = stats.get("tokens_total", 0) + tool = stats.get("tokens_tool", 0) + out["loss_tokens_if_tool_masked"] = total - tool + out["tool_response_token_share"] = tool / total if total else 0 + out["avg_total_tokens_per_trace"] = total / kept if kept else 0 + out["avg_loss_tokens_per_trace_if_tool_masked"] = (total - tool) / kept if kept else 0 + return out + + +def main() -> int: + args = parse_args() + tokenizer = AutoTokenizer.from_pretrained(args.model, trust_remote_code=True) + audit_args = AuditArgs() + data_root = args.input_root / "data" + total = Counter() + configs: dict[str, Counter] = {} + start = time.time() + processed_since_flush = 0 + + for config in audit.CONFIGS: + cfg_stats = configs.setdefault(config, Counter()) + for file in sorted((data_root / config).glob("*.parquet")): + file_stats = Counter() + parquet = pq.ParquetFile(file) + pending_texts: list[str] = [] + pending_roles: list[str] = [] + + def flush_tokens() -> None: + add_tokenized_batch(tokenizer, pending_texts, pending_roles, file_stats) + pending_texts.clear() + pending_roles.clear() + + for record_batch in parquet.iter_batches(batch_size=args.batch_size): + for row in record_batch.to_pylist(): + issues, _details = audit.audit_row(row, audit_args) + hard = audit.hard_filter_issues(issues) + if hard: + file_stats["filtered"] += 1 + continue + file_stats["kept"] += 1 + for message in row.get("trajectory") or []: + role = message.get("role") or "unknown" + text = message_text(message) + file_stats["messages"] += 1 + file_stats[f"messages_{role}"] += 1 + file_stats["chars_total"] += len(text) + file_stats[f"chars_{role}"] += len(text) + if text: + pending_texts.append(text) + pending_roles.append(role) + if len(pending_texts) >= args.tokenize_batch_size: + flush_tokens() + processed_since_flush += 1 + if processed_since_flush >= args.flush_interval: + flush_tokens() + processed_since_flush = 0 + elapsed = time.time() - start + partial_total = Counter() + for cstats in configs.values(): + partial_total.update(cstats) + partial_total.update(total) + report = { + "model": args.model, + "status": "running", + "elapsed_seconds": elapsed, + "configs": {k: summarize(v) for k, v in configs.items()}, + "total": summarize(partial_total), + } + update_output(args.output, report) + flush_tokens() + cfg_stats.update(file_stats) + print( + json.dumps( + { + "file": str(file), + "stats": summarize(file_stats), + "elapsed_seconds": round(time.time() - start, 2), + }, + ensure_ascii=False, + ), + flush=True, + ) + total.update(cfg_stats) + + report = { + "model": args.model, + "status": "complete", + "elapsed_seconds": time.time() - start, + "configs": {k: summarize(v) for k, v in configs.items()}, + "total": summarize(total), + } + update_output(args.output, report) + print(json.dumps({"output": str(args.output), "total": report["total"]}, ensure_ascii=False)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/probing/count_qwen_tokens_exact_parallel.py b/scripts/probing/count_qwen_tokens_exact_parallel.py new file mode 100644 index 0000000..7ca0308 --- /dev/null +++ b/scripts/probing/count_qwen_tokens_exact_parallel.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import os +import sys +import time +from collections import Counter +from concurrent.futures import ProcessPoolExecutor, as_completed +from pathlib import Path +from typing import Any + +import pyarrow.parquet as pq +from transformers import AutoTokenizer + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT / "scripts" / "filtering")) +import audit_native_traces as audit # noqa: E402 + + +TOKENIZER = None +MODEL_NAME = None + + +class AuditArgs: + long_turn_threshold = 300 + long_char_threshold = 900_000 + repeat_window = 24 + repeat_threshold = 10 + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Parallel exact Qwen tokenizer count for Open-SWE traces.") + parser.add_argument("--input-root", type=Path, default=Path("data/Open-SWE-Traces")) + parser.add_argument("--output", type=Path, default=Path("runs/native_trace_audit/qwen_exact_token_count.json")) + parser.add_argument("--model", default="Qwen/Qwen3-32B") + parser.add_argument("--workers", type=int, default=12) + parser.add_argument("--batch-size", type=int, default=512) + parser.add_argument("--tokenize-batch-size", type=int, default=1024) + return parser.parse_args() + + +def init_worker(model_name: str) -> None: + global TOKENIZER, MODEL_NAME + MODEL_NAME = model_name + os.environ.setdefault("TOKENIZERS_PARALLELISM", "false") + TOKENIZER = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True) + + +def message_text(message: dict[str, Any]) -> str: + text = "" + for key in ("content", "reasoning_content"): + value = message.get(key) + if isinstance(value, str): + text += value + if message.get("tool_calls"): + text += json.dumps(message.get("tool_calls"), ensure_ascii=False) + return text + + +def add_tokenized_batch(texts: list[str], roles: list[str], stats: Counter) -> None: + if not texts: + return + encoded = TOKENIZER(texts, add_special_tokens=False, return_attention_mask=False, return_token_type_ids=False) + for role, ids in zip(roles, encoded["input_ids"], strict=True): + n = len(ids) + stats["tokens_total"] += n + stats[f"tokens_{role}"] += n + + +def count_file(task: tuple[str, str, int, int]) -> dict[str, Any]: + config, file_name, batch_size, tokenize_batch_size = task + file = Path(file_name) + audit_args = AuditArgs() + stats = Counter() + pending_texts: list[str] = [] + pending_roles: list[str] = [] + + def flush() -> None: + add_tokenized_batch(pending_texts, pending_roles, stats) + pending_texts.clear() + pending_roles.clear() + + parquet = pq.ParquetFile(file) + for record_batch in parquet.iter_batches(batch_size=batch_size): + for row in record_batch.to_pylist(): + issues, _details = audit.audit_row(row, audit_args) + hard = audit.hard_filter_issues(issues) + if hard: + stats["filtered"] += 1 + continue + stats["kept"] += 1 + for message in row.get("trajectory") or []: + role = message.get("role") or "unknown" + text = message_text(message) + stats["messages"] += 1 + stats[f"messages_{role}"] += 1 + stats["chars_total"] += len(text) + stats[f"chars_{role}"] += len(text) + if text: + pending_texts.append(text) + pending_roles.append(role) + if len(pending_texts) >= tokenize_batch_size: + flush() + flush() + return {"config": config, "file": str(file), "stats": dict(stats)} + + +def summarize(stats: Counter) -> dict[str, Any]: + out = dict(stats) + kept = stats.get("kept", 0) + total = stats.get("tokens_total", 0) + tool = stats.get("tokens_tool", 0) + out["loss_tokens_if_tool_masked"] = total - tool + out["tool_response_token_share"] = tool / total if total else 0 + out["avg_total_tokens_per_trace"] = total / kept if kept else 0 + out["avg_loss_tokens_per_trace_if_tool_masked"] = (total - tool) / kept if kept else 0 + return out + + +def write_report(path: Path, report: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8") + tmp.replace(path) + + +def main() -> int: + args = parse_args() + data_root = args.input_root / "data" + tasks = [] + for config in audit.CONFIGS: + for file in sorted((data_root / config).glob("*.parquet")): + tasks.append((config, str(file), args.batch_size, args.tokenize_batch_size)) + + start = time.time() + configs = {config: Counter() for config in audit.CONFIGS} + total = Counter() + completed = 0 + with ProcessPoolExecutor(max_workers=args.workers, initializer=init_worker, initargs=(args.model,)) as pool: + futures = [pool.submit(count_file, task) for task in tasks] + for future in as_completed(futures): + result = future.result() + completed += 1 + config = result["config"] + file_stats = Counter(result["stats"]) + configs[config].update(file_stats) + total.update(file_stats) + elapsed = time.time() - start + print( + json.dumps( + { + "completed": completed, + "total_files": len(tasks), + "file": result["file"], + "stats": summarize(file_stats), + "elapsed_seconds": round(elapsed, 2), + }, + ensure_ascii=False, + ), + flush=True, + ) + report = { + "model": args.model, + "status": "running" if completed < len(tasks) else "complete", + "workers": args.workers, + "completed_files": completed, + "total_files": len(tasks), + "elapsed_seconds": elapsed, + "configs": {key: summarize(value) for key, value in configs.items()}, + "total": summarize(total), + } + write_report(args.output, report) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/probing/estimate_native_trace_tokens.py b/scripts/probing/estimate_native_trace_tokens.py new file mode 100644 index 0000000..a979ee5 --- /dev/null +++ b/scripts/probing/estimate_native_trace_tokens.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import json +import sys +from collections import Counter +from pathlib import Path + +import pyarrow.parquet as pq + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT / "scripts" / "filtering")) +import audit_native_traces as audit # noqa: E402 + + +class Args: + long_turn_threshold = 300 + long_char_threshold = 900_000 + repeat_window = 24 + repeat_threshold = 10 + + +def main() -> int: + args = Args() + base = Path("data/Open-SWE-Traces/data") + stats = Counter() + for cfg in audit.CONFIGS: + c = Counter() + for file in sorted((base / cfg).glob("*.parquet")): + parquet = pq.ParquetFile(file) + for batch in parquet.iter_batches(batch_size=512): + for row in batch.to_pylist(): + issues, _details = audit.audit_row(row, args) + hard = audit.hard_filter_issues(issues) + if hard: + c["filtered"] += 1 + stats["filtered"] += 1 + continue + c["kept"] += 1 + stats["kept"] += 1 + trajectory = row.get("trajectory") or [] + for message in trajectory: + role = message.get("role") or "unknown" + text = "" + for key in ("content", "reasoning_content"): + value = message.get(key) + if isinstance(value, str): + text += value + if message.get("tool_calls"): + text += json.dumps(message.get("tool_calls"), ensure_ascii=False) + chars = len(text) + c[f"chars_{role}"] += chars + stats[f"chars_{role}"] += chars + c["chars_total"] += chars + stats["chars_total"] += chars + c["messages"] += len(trajectory) + stats["messages"] += len(trajectory) + print("CFG", cfg, json.dumps(dict(c), ensure_ascii=False), flush=True) + print("TOTAL", json.dumps(dict(stats), ensure_ascii=False)) + for ratio in (3.0, 3.5, 4.0): + total_tokens = stats["chars_total"] / ratio + tool_tokens = stats["chars_tool"] / ratio + loss_tokens = (stats["chars_total"] - stats["chars_tool"]) / ratio + print( + "TOKENS_RATIO", + ratio, + json.dumps( + { + "total_tokens": round(total_tokens), + "tool_response_tokens": round(tool_tokens), + "loss_tokens_if_tool_masked": round(loss_tokens), + "avg_total_tokens_per_row": round(total_tokens / stats["kept"]), + "avg_loss_tokens_per_row": round(loss_tokens / stats["kept"]), + "tool_response_token_share": round(tool_tokens / total_tokens, 4), + }, + ensure_ascii=False, + ), + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/probing/inspect_sample.py b/scripts/probing/inspect_sample.py new file mode 100644 index 0000000..f6515f7 --- /dev/null +++ b/scripts/probing/inspect_sample.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python3 +import json +from pathlib import Path + +import pyarrow.parquet as pq + + +BASE = Path("data/Open-SWE-Traces/data") +CONFIGS = [ + "minimax_m25_openhands_trajectories", + "minimax_m25_sweagent_trajectories", + "qwen35_openhands_trajectories", + "qwen35_sweagent_trajectories", +] + + +def main() -> int: + for cfg in CONFIGS: + file = sorted((BASE / cfg).glob("*.parquet"))[0] + row = pq.read_table(file).slice(0, 1).to_pylist()[0] + print("CFG", cfg) + print("cols", list(row.keys())) + trajectory = row["trajectory"] + print("len", len(trajectory)) + for index, message in enumerate(trajectory[:10]): + print("MSG", index, json.dumps(message, ensure_ascii=False)[:1600]) + print("---") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/probing/load_qwen_tokenizer.py b/scripts/probing/load_qwen_tokenizer.py new file mode 100644 index 0000000..669a085 --- /dev/null +++ b/scripts/probing/load_qwen_tokenizer.py @@ -0,0 +1,16 @@ +#!/usr/bin/env python3 +from transformers import AutoTokenizer + + +MODEL = "Qwen/Qwen3-32B" + + +def main() -> int: + tokenizer = AutoTokenizer.from_pretrained(MODEL, trust_remote_code=True) + print(type(tokenizer).__name__, tokenizer.vocab_size, tokenizer.eos_token_id) + print(tokenizer.encode("hello world")) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/probing/query_qwen_model_ids.py b/scripts/probing/query_qwen_model_ids.py new file mode 100644 index 0000000..bfb6819 --- /dev/null +++ b/scripts/probing/query_qwen_model_ids.py @@ -0,0 +1,19 @@ +#!/usr/bin/env python3 +from huggingface_hub import HfApi + + +def main() -> int: + api = HfApi(endpoint="https://hf-mirror.com") + for query in ["Qwen3.5", "Qwen3.6", "Qwen3.5-27B", "Qwen3.6-27B", "Qwen3.6-35B"]: + print("QUERY", query) + try: + models = list(api.list_models(search=query, limit=50)) + for model in models: + print(model.modelId) + except Exception as exc: + print("ERR", type(exc).__name__, str(exc)[:300]) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/probing/sample_qwen_token_ratios.py b/scripts/probing/sample_qwen_token_ratios.py new file mode 100644 index 0000000..8b662ad --- /dev/null +++ b/scripts/probing/sample_qwen_token_ratios.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import sys +from collections import Counter +from pathlib import Path + +import pyarrow.parquet as pq +from transformers import AutoTokenizer + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT / "scripts" / "filtering")) +import audit_native_traces as audit # noqa: E402 + + +class AuditArgs: + long_turn_threshold = 300 + long_char_threshold = 900_000 + repeat_window = 24 + repeat_threshold = 10 + + +def parse_args(): + parser = argparse.ArgumentParser() + parser.add_argument("--model", default="Qwen/Qwen3-32B") + parser.add_argument("--samples-per-config", type=int, default=2000) + parser.add_argument("--output", type=Path, default=Path("runs/native_trace_audit/qwen_token_ratio_sample.json")) + return parser.parse_args() + + +def msg_text(message): + text = "" + for key in ("content", "reasoning_content"): + value = message.get(key) + if isinstance(value, str): + text += value + if message.get("tool_calls"): + text += json.dumps(message.get("tool_calls"), ensure_ascii=False) + return text + + +def main() -> int: + args = parse_args() + tokenizer = AutoTokenizer.from_pretrained(args.model, trust_remote_code=True) + audit_args = AuditArgs() + base = Path("data/Open-SWE-Traces/data") + report = {"model": args.model, "configs": {}} + total = Counter() + + for cfg in audit.CONFIGS: + stats = Counter() + for file in sorted((base / cfg).glob("*.parquet")): + if stats["rows"] >= args.samples_per_config: + break + parquet = pq.ParquetFile(file) + for batch in parquet.iter_batches(batch_size=256): + for row in batch.to_pylist(): + issues, _ = audit.audit_row(row, audit_args) + if audit.hard_filter_issues(issues): + continue + stats["rows"] += 1 + for message in row.get("trajectory") or []: + role = message.get("role") or "unknown" + text = msg_text(message) + chars = len(text) + if not chars: + continue + toks = len(tokenizer.encode(text, add_special_tokens=False)) + stats[f"chars_{role}"] += chars + stats[f"tokens_{role}"] += toks + stats["chars_total"] += chars + stats["tokens_total"] += toks + if stats["rows"] >= args.samples_per_config: + break + if stats["rows"] >= args.samples_per_config: + break + report["configs"][cfg] = summarize(stats) + total.update(stats) + print("CFG", cfg, json.dumps(report["configs"][cfg], ensure_ascii=False), flush=True) + + report["total_sample"] = summarize(total) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8") + print("REPORT", args.output) + return 0 + + +def summarize(stats: Counter) -> dict: + out = dict(stats) + if stats["chars_total"]: + out["tokens_per_char_total"] = stats["tokens_total"] / stats["chars_total"] + out["chars_per_token_total"] = stats["chars_total"] / stats["tokens_total"] + for role in ("system", "user", "assistant", "tool"): + c = stats.get(f"chars_{role}", 0) + t = stats.get(f"tokens_{role}", 0) + if c and t: + out[f"tokens_per_char_{role}"] = t / c + out[f"chars_per_token_{role}"] = c / t + return out + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/repurposing/build_swift_training_probe_5k.py b/scripts/repurposing/build_swift_training_probe_5k.py new file mode 100644 index 0000000..3beb8c3 --- /dev/null +++ b/scripts/repurposing/build_swift_training_probe_5k.py @@ -0,0 +1,259 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import gzip +import json +import sys +from collections import Counter +from pathlib import Path +from typing import Any + +import pyarrow as pa +import pyarrow.parquet as pq + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT / "scripts" / "filtering")) +import audit_native_traces as audit # noqa: E402 + + +TARGET_PER_CONFIG = 1250 +NO_RESULT_TOOLS = {"finish", "submit", "think"} + +CONFIG_META = { + "minimax_m25_openhands_trajectories": ("minimax_m25", "openhands", "thinking"), + "minimax_m25_sweagent_trajectories": ("minimax_m25", "sweagent", "thinking"), + "qwen35_openhands_trajectories": ("qwen35", "openhands", "non_thinking"), + "qwen35_sweagent_trajectories": ("qwen35", "sweagent", "non_thinking"), +} + + +class AuditArgs: + long_turn_threshold = 300 + long_char_threshold = 900_000 + repeat_window = 24 + repeat_threshold = 10 + + +def normalize_json(value: Any) -> Any: + return json.loads(json.dumps(value, ensure_ascii=False, default=str)) + + +def parse_tool_args(raw: Any) -> Any: + if isinstance(raw, dict): + return raw + if isinstance(raw, str): + try: + return json.loads(raw) + except Exception: + return raw + return raw + + +def convert_tool_calls(tool_calls: list[dict[str, Any]] | None) -> list[dict[str, Any]]: + converted = [] + for call in tool_calls or []: + function = call.get("function") or {} + converted.append( + { + "id": call.get("id"), + "type": call.get("type", "function"), + "function": { + "name": function.get("name") or call.get("name"), + "arguments": parse_tool_args(function.get("arguments", call.get("arguments"))), + }, + } + ) + return converted + + +def tool_calls_to_text(tool_calls: list[dict[str, Any]]) -> str: + if not tool_calls: + return "" + return "\n" + json.dumps(tool_calls, ensure_ascii=False, separators=(",", ":")) + "\n" + + +def tool_requires_result(tool_call: dict[str, Any]) -> bool: + name = (tool_call.get("function") or {}).get("name") + return name not in NO_RESULT_TOOLS + + +def convert_messages(trajectory: list[dict[str, Any]], thinking_mode: str, stats: Counter) -> list[dict[str, Any]]: + messages: list[dict[str, Any]] = [] + pending_tool_calls: list[dict[str, Any]] = [] + for source in trajectory: + role = source.get("role") + if role == "system": + content = source.get("content") or "" + if content: + messages.append({"role": "system", "content": content, "loss": False}) + continue + if role == "user": + messages.append({"role": "user", "content": source.get("content") or "", "loss": False}) + continue + if role == "assistant": + content = source.get("content") or "" + reasoning = source.get("reasoning_content") or "" + think_value = source.get("think") + if thinking_mode == "thinking": + if reasoning: + content = f"\n{reasoning}\n\n{content}" + stats["assistant_thinking_messages"] += 1 if reasoning else 0 + else: + if reasoning: + stats["qwen_non_thinking_reasoning_nonempty"] += 1 + if think_value not in (None, False): + stats[f"qwen_unexpected_think:{think_value}"] += 1 + tool_calls = convert_tool_calls(source.get("tool_calls")) + tool_text = tool_calls_to_text(tool_calls) + if tool_text: + content = (content + "\n" + tool_text).strip() + msg = {"role": "assistant", "content": content, "loss": True} + if tool_calls: + msg["tool_calls"] = tool_calls + pending_tool_calls.extend([tool_call for tool_call in tool_calls if tool_requires_result(tool_call)]) + messages.append(msg) + continue + if role == "tool": + tool_call = pending_tool_calls.pop(0) if pending_tool_calls else {} + messages.append( + { + "role": "tool", + "content": source.get("content") or "", + "tool_call_id": tool_call.get("id"), + "name": (tool_call.get("function") or {}).get("name"), + "loss": False, + } + ) + continue + stats[f"unknown_role:{role}"] += 1 + if pending_tool_calls: + stats["pending_tool_calls_after_conversion"] += len(pending_tool_calls) + return messages + + +def main() -> int: + input_root = Path("data/Open-SWE-Traces") + output_dir = Path("runs/training_probe_5k_swift") + output_dir.mkdir(parents=True, exist_ok=True) + audit_args = AuditArgs() + rows: list[dict[str, Any]] = [] + stats = Counter() + + for config in audit.CONFIGS: + model_family, scaffold, thinking_mode = CONFIG_META[config] + kept_for_config = 0 + for file in sorted((input_root / "data" / config).glob("*.parquet")): + if kept_for_config >= TARGET_PER_CONFIG: + break + table = pq.read_table(file) + for row in table.to_pylist(): + stats[f"seen:{config}"] += 1 + issues, details = audit.audit_row(row, audit_args) + hard = audit.hard_filter_issues(issues) + if hard: + stats[f"filtered:{config}"] += 1 + continue + messages = convert_messages(row.get("trajectory") or [], thinking_mode, stats) + item = { + "messages": messages, + "source_dataset": "nvidia/Open-SWE-Traces", + "source_config": config, + "model_family": model_family, + "scaffold": scaffold, + "thinking_mode": thinking_mode, + "instance_id": row.get("instance_id"), + "repo": row.get("repo"), + "language": row.get("language"), + "license": row.get("license"), + "trajectory_id": row.get("trajectory_id"), + "resolved": int(row.get("resolved") or 0), + "model_patch": row.get("model_patch") or "", + "metadata": normalize_json(row.get("metadata") or {}), + "audit_flags": issues, + "audit_details": dict(details), + } + rows.append(item) + kept_for_config += 1 + stats[f"kept:{config}"] += 1 + stats[f"resolved:{config}:{item['resolved']}"] += 1 + if kept_for_config >= TARGET_PER_CONFIG: + break + + jsonl = output_dir / "train.jsonl" + with jsonl.open("w", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n") + + jsonl_gz = output_dir / "train.jsonl.gz" + with gzip.open(jsonl_gz, "wt", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n") + + parquet_path = output_dir / "train.parquet" + parquet_rows = [] + for row in rows: + parquet_rows.append( + { + **{k: v for k, v in row.items() if k not in {"messages", "metadata", "audit_flags", "audit_details"}}, + "messages": json.dumps(row["messages"], ensure_ascii=False), + "metadata": json.dumps(row["metadata"], ensure_ascii=False), + "audit_flags": json.dumps(row["audit_flags"], ensure_ascii=False), + "audit_details": json.dumps(row["audit_details"], ensure_ascii=False), + } + ) + pq.write_table(pa.Table.from_pylist(parquet_rows), parquet_path, compression="zstd") + + metadata = { + "name": "open-swe-traces-swift-probe-5k", + "rows": len(rows), + "selection": f"{TARGET_PER_CONFIG} hard-filter-kept rows per source config", + "source_dataset": "nvidia/Open-SWE-Traces", + "format": "modelscope-swift messages JSONL", + "stats": dict(stats), + "thinking_policy": { + "minimax": "reasoning_content is wrapped as ... in assistant content", + "qwen": "non-thinking export; reasoning_content is not emitted; unexpected nonempty reasoning is counted", + "tool_response_mask": "tool messages have loss=false", + }, + "files": ["train.jsonl", "train.jsonl.gz", "train.parquet", "README.md", "metadata.json"], + } + (output_dir / "metadata.json").write_text(json.dumps(metadata, indent=2, ensure_ascii=False), encoding="utf-8") + (output_dir / "README.md").write_text(build_readme(metadata), encoding="utf-8") + print(json.dumps(metadata, indent=2, ensure_ascii=False)) + return 0 + + +def build_readme(metadata: dict[str, Any]) -> str: + return f"""--- +license: other +task_categories: +- text-generation +language: +- code +pretty_name: Open-SWE-Traces Swift Probe 5K +size_categories: +- 1K...`. +- Qwen rows are exported as non-thinking examples; `reasoning_content` is not emitted. +- Tool responses are included as `role: "tool"` messages with `loss: false`. + +Stats: + +```json +{json.dumps(metadata["stats"], indent=2, ensure_ascii=False)} +``` +""" + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/repurposing/build_swift_validation_500.py b/scripts/repurposing/build_swift_validation_500.py new file mode 100644 index 0000000..4ca8942 --- /dev/null +++ b/scripts/repurposing/build_swift_validation_500.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import gzip +import json +import random +import sys +from collections import Counter +from pathlib import Path +from typing import Any + +import pyarrow as pa +import pyarrow.parquet as pq + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT / "scripts" / "filtering")) +import audit_native_traces as audit # noqa: E402 +sys.path.insert(0, str(REPO_ROOT / "scripts" / "repurposing")) +import build_swift_training_probe_5k as train_builder # noqa: E402 + + +TARGET_PER_CONFIG = 125 +SEED = 20260624 + + +def normalize_json(value: Any) -> Any: + return json.loads(json.dumps(value, ensure_ascii=False, default=str)) + + +def load_train_ids(train_path: Path) -> set[str]: + ids = set() + with train_path.open("r", encoding="utf-8") as handle: + for line in handle: + row = json.loads(line) + ids.add(row["trajectory_id"]) + return ids + + +def main() -> int: + input_root = Path("data/Open-SWE-Traces") + output_dir = Path("runs/training_probe_5k_swift") + train_ids = load_train_ids(output_dir / "train.jsonl") + audit_args = train_builder.AuditArgs() + rng = random.Random(SEED) + rows: list[dict[str, Any]] = [] + stats = Counter() + + for config in audit.CONFIGS: + candidates = [] + model_family, scaffold, thinking_mode = train_builder.CONFIG_META[config] + for file in sorted((input_root / "data" / config).glob("*.parquet")): + table = pq.read_table(file) + for row in table.to_pylist(): + stats[f"seen:{config}"] += 1 + trajectory_id = row.get("trajectory_id") + if trajectory_id in train_ids: + stats[f"skip_train_overlap:{config}"] += 1 + continue + issues, details = audit.audit_row(row, audit_args) + if audit.hard_filter_issues(issues): + stats[f"filtered:{config}"] += 1 + continue + candidates.append((row, issues, details, model_family, scaffold, thinking_mode)) + + chosen = rng.sample(candidates, TARGET_PER_CONFIG) + for row, issues, details, model_family, scaffold, thinking_mode in chosen: + messages = train_builder.convert_messages(row.get("trajectory") or [], thinking_mode, stats) + item = { + "messages": messages, + "source_dataset": "nvidia/Open-SWE-Traces", + "source_config": config, + "model_family": model_family, + "scaffold": scaffold, + "thinking_mode": thinking_mode, + "instance_id": row.get("instance_id"), + "repo": row.get("repo"), + "language": row.get("language"), + "license": row.get("license"), + "trajectory_id": row.get("trajectory_id"), + "resolved": int(row.get("resolved") or 0), + "model_patch": row.get("model_patch") or "", + "metadata": normalize_json(row.get("metadata") or {}), + "audit_flags": issues, + "audit_details": dict(details), + } + rows.append(item) + stats[f"kept:{config}"] += 1 + stats[f"resolved:{config}:{item['resolved']}"] += 1 + + jsonl = output_dir / "validation.jsonl" + with jsonl.open("w", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n") + + with gzip.open(output_dir / "validation.jsonl.gz", "wt", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n") + + parquet_rows = [] + for row in rows: + parquet_rows.append( + { + **{k: v for k, v in row.items() if k not in {"messages", "metadata", "audit_flags", "audit_details"}}, + "messages": json.dumps(row["messages"], ensure_ascii=False), + "metadata": json.dumps(row["metadata"], ensure_ascii=False), + "audit_flags": json.dumps(row["audit_flags"], ensure_ascii=False), + "audit_details": json.dumps(row["audit_details"], ensure_ascii=False), + } + ) + pq.write_table(pa.Table.from_pylist(parquet_rows), output_dir / "validation.parquet", compression="zstd") + + metadata_path = output_dir / "metadata.json" + metadata = json.loads(metadata_path.read_text(encoding="utf-8")) + metadata["validation"] = { + "rows": len(rows), + "selection": f"random {TARGET_PER_CONFIG} hard-filter-kept rows per source config, excluding train trajectory_id", + "seed": SEED, + "stats": dict(stats), + "files": ["validation.jsonl", "validation.jsonl.gz", "validation.parquet"], + } + metadata_path.write_text(json.dumps(metadata, indent=2, ensure_ascii=False), encoding="utf-8") + + readme = output_dir / "README.md" + text = readme.read_text(encoding="utf-8") + text += f"\n\n## Validation Split\n\nRandom balanced validation split, seed `{SEED}`, 125 rows per config, excluding train trajectory ids.\n\n```json\n{json.dumps(metadata['validation']['stats'], indent=2, ensure_ascii=False)}\n```\n" + readme.write_text(text, encoding="utf-8") + print(json.dumps(metadata["validation"], indent=2, ensure_ascii=False)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/repurposing/build_training_probe_5k.py b/scripts/repurposing/build_training_probe_5k.py new file mode 100644 index 0000000..56ced09 --- /dev/null +++ b/scripts/repurposing/build_training_probe_5k.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import gzip +import json +import sys +from collections import Counter +from pathlib import Path +from typing import Any + +import pyarrow as pa +import pyarrow.parquet as pq + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT / "scripts" / "filtering")) +import audit_native_traces as audit # noqa: E402 + + +class AuditArgs: + long_turn_threshold = 300 + long_char_threshold = 900_000 + repeat_window = 24 + repeat_threshold = 10 + + +CONFIG_META = { + "minimax_m25_openhands_trajectories": ("minimax_m25", "openhands", "thinking"), + "minimax_m25_sweagent_trajectories": ("minimax_m25", "sweagent", "thinking"), + "qwen35_openhands_trajectories": ("qwen35", "openhands", "non_thinking"), + "qwen35_sweagent_trajectories": ("qwen35", "sweagent", "non_thinking"), +} + + +def normalize_for_json(value: Any) -> Any: + return json.loads(json.dumps(value, ensure_ascii=False, default=str)) + + +def main() -> int: + input_root = Path("data/Open-SWE-Traces") + output_dir = Path("runs/training_probe_5k") + target = 5000 + output_dir.mkdir(parents=True, exist_ok=True) + rows: list[dict[str, Any]] = [] + stats = Counter() + audit_args = AuditArgs() + + for config in audit.CONFIGS: + model_family, scaffold, thinking_mode = CONFIG_META[config] + for file in sorted((input_root / "data" / config).glob("*.parquet")): + table = pq.read_table(file) + for row in table.to_pylist(): + issues, details = audit.audit_row(row, audit_args) + hard = audit.hard_filter_issues(issues) + stats["seen"] += 1 + if hard: + stats["filtered"] += 1 + continue + item = { + "probe_index": len(rows), + "source_dataset": "nvidia/Open-SWE-Traces", + "source_config": config, + "model_family": model_family, + "scaffold": scaffold, + "thinking_mode": thinking_mode, + "instance_id": row.get("instance_id"), + "repo": row.get("repo"), + "license": row.get("license"), + "language": row.get("language"), + "trajectory_id": row.get("trajectory_id"), + "trajectory": normalize_for_json(row.get("trajectory") or []), + "model_patch": row.get("model_patch") or "", + "resolved": int(row.get("resolved") or 0), + "metadata": normalize_for_json(row.get("metadata") or {}), + "audit_flags": issues, + "hard_filter_issues": hard, + "audit_details": dict(details), + } + rows.append(item) + stats["kept"] += 1 + stats[f"config:{config}"] += 1 + stats[f"resolved:{item['resolved']}"] += 1 + if len(rows) >= target: + break + if len(rows) >= target: + break + if len(rows) >= target: + break + + jsonl_gz = output_dir / "train.jsonl.gz" + with gzip.open(jsonl_gz, "wt", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n") + + parquet_path = output_dir / "train.parquet" + table = pa.Table.from_pylist(rows) + pq.write_table(table, parquet_path, compression="zstd") + + metadata = { + "name": "open-swe-traces-native-probe-5k", + "source_dataset": "nvidia/Open-SWE-Traces", + "selection": "first 5000 hard-filter-kept traces in config/shard order", + "rows": len(rows), + "stats": dict(stats), + "hard_filter_policy": { + "script": "scripts/filtering/audit_native_traces.py", + "long_turn_threshold": audit_args.long_turn_threshold, + "long_char_threshold": audit_args.long_char_threshold, + "repeat_window": audit_args.repeat_window, + "repeat_threshold": audit_args.repeat_threshold, + }, + "files": { + "jsonl_gz": str(jsonl_gz), + "parquet": str(parquet_path), + }, + } + (output_dir / "metadata.json").write_text(json.dumps(metadata, indent=2, ensure_ascii=False), encoding="utf-8") + (output_dir / "README.md").write_text(build_readme(metadata), encoding="utf-8") + print(json.dumps(metadata, indent=2, ensure_ascii=False)) + return 0 + + +def build_readme(metadata: dict[str, Any]) -> str: + return f"""--- +license: other +task_categories: +- text-generation +language: +- code +pretty_name: Open-SWE-Traces Native Probe 5K +size_categories: +- 1K list[Any]: + seen = set() + out = [] + for item in items: + key = json.dumps(item, ensure_ascii=False, sort_keys=True) if isinstance(item, (dict, list)) else str(item) + if key in seen: + continue + seen.add(key) + out.append(item) + if limit and len(out) >= limit: + break + return out + + +def merge_segments(segments: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Aggregate fine segments into a small fixed set of task stages. + + Fine-grained traces often alternate between locate/diagnose/fix many times. + For SFT subproblem construction we want coarse, purposeful stages rather + than exact chronological micro-steps, so this groups by task category and + preserves the first-to-last turn span for each category. + """ + + by_category: dict[str, dict[str, Any]] = {} + + def start(seg: dict[str, Any], name: str) -> dict[str, Any]: + name = COARSE_MAP.get(seg["phase"], "diagnose") + return { + "category": name, + "goal": COARSE_GOALS[name], + "turn_range": list(seg["turn_range"]), + "source_phases": [seg["phase"]], + "commands": list(seg.get("commands", [])), + "tool_names": list(seg.get("tool_names", [])), + "files": list(seg.get("files", [])), + "assistant_intents": list(seg.get("assistant_intents", [])), + "observations": list(seg.get("observations", [])), + } + + for seg in segments: + name = COARSE_MAP.get(seg["phase"], "diagnose") + current = by_category.get(name) + if current is None: + by_category[name] = start(seg, name) + continue + + current["turn_range"][0] = min(current["turn_range"][0], seg["turn_range"][0]) + current["turn_range"][1] = max(current["turn_range"][1], seg["turn_range"][1]) + current["source_phases"].append(seg["phase"]) + current["commands"].extend(seg.get("commands", [])) + current["tool_names"].extend(seg.get("tool_names", [])) + current["files"].extend(seg.get("files", [])) + current["assistant_intents"].extend(seg.get("assistant_intents", [])) + current["observations"].extend(seg.get("observations", [])) + + ordered = [] + for name in ["understand", "locate", "diagnose", "fix", "verify", "finalize"]: + if name in by_category: + ordered.append(finish(by_category[name])) + return ordered + + +def finish(seg: dict[str, Any]) -> dict[str, Any]: + seg["source_phases"] = dedupe(seg["source_phases"]) + seg["commands"] = dedupe(seg["commands"], 12) + seg["tool_names"] = dedupe(seg["tool_names"], 12) + seg["files"] = dedupe(seg["files"], 12) + seg["assistant_intents"] = dedupe(seg["assistant_intents"], 6) + seg["observations"] = dedupe(seg["observations"], 6) + return seg + + +def build_training_views(item: dict[str, Any], coarse_segments: list[dict[str, Any]]) -> dict[str, Any]: + qa = [] + question_trajectory = [] + for seg in coarse_segments: + qa.append( + { + "question": f"In repository {item['repo']}, how should an agent {seg['goal'][0].lower() + seg['goal'][1:]}", + "answer": { + "category": seg["category"], + "files": seg["files"], + "commands": seg["commands"], + "evidence": seg["observations"], + "next_action_hint": next_action_hint(seg["category"]), + }, + } + ) + question_trajectory.append( + { + "question": f"For task {item['instance_id']} in {item['repo']}, perform the {seg['category']} stage.", + "category": seg["category"], + "turn_range": seg["turn_range"], + "source_phases": seg["source_phases"], + "trajectory_outline": { + "intents": seg["assistant_intents"], + "tools": seg["tool_names"], + "commands": seg["commands"], + "observations": seg["observations"], + }, + } + ) + return {"qa": qa, "question_trajectory": question_trajectory} + + +def next_action_hint(category: str) -> str: + return { + "understand": "Start repository exploration.", + "locate": "Inspect the most relevant code paths.", + "diagnose": "Decide the minimal edit needed to fix the root cause.", + "fix": "Run targeted verification.", + "verify": "Review diff and clean temporary files.", + "finalize": "Stop or submit the final solution.", + }[category] + + +def main() -> int: + items = json.loads(IN_FILE.read_text()) + output = [] + category_counts = Counter() + before = 0 + after = 0 + for item in items: + coarse_segments = merge_segments(item["segments"]) + before += len(item["segments"]) + after += len(coarse_segments) + category_counts.update(seg["category"] for seg in coarse_segments) + converted = { + **{k: item[k] for k in [ + "config", + "scaffold", + "model_family", + "thinking_mode", + "instance_id", + "repo", + "language", + "license", + "trajectory_id", + "resolved", + "trajectory_len", + "task", + "model_patch_head", + ]}, + "coarse_segments": coarse_segments, + "training_views": build_training_views(item, coarse_segments), + } + output.append(converted) + + report = { + "items": len(items), + "fine_segments": before, + "coarse_segments": after, + "avg_fine_segments_per_trace": before / len(items), + "avg_coarse_segments_per_trace": after / len(items), + "category_counts": dict(category_counts), + "categories": COARSE_GOALS, + } + OUT_FILE.write_text(json.dumps(output, indent=2, ensure_ascii=False)) + REPORT_FILE.write_text(json.dumps(report, indent=2, ensure_ascii=False)) + print(json.dumps(report, indent=2, ensure_ascii=False)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/repurposing/convert_openswe_to_pi_mono.py b/scripts/repurposing/convert_openswe_to_pi_mono.py new file mode 100644 index 0000000..1a32d7e --- /dev/null +++ b/scripts/repurposing/convert_openswe_to_pi_mono.py @@ -0,0 +1,419 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import gzip +import json +import re +from collections import Counter +from pathlib import Path +from typing import Any + +import pyarrow.parquet as pq + + +CONFIGS = { + "minimax_m25_openhands_trajectories": { + "model_family": "minimax_m25", + "scaffold": "openhands", + "thinking_mode": "thinking", + }, + "minimax_m25_sweagent_trajectories": { + "model_family": "minimax_m25", + "scaffold": "sweagent", + "thinking_mode": "thinking", + }, + "qwen35_openhands_trajectories": { + "model_family": "qwen35", + "scaffold": "openhands", + "thinking_mode": "non_thinking", + }, + "qwen35_sweagent_trajectories": { + "model_family": "qwen35", + "scaffold": "sweagent", + "thinking_mode": "non_thinking", + }, +} + + +PI_TOOL_NAMES = {"bash", "read", "edit", "write", "grep", "find", "ls"} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Convert Open-SWE-Traces to pi-mono message JSONL.") + parser.add_argument("--input-root", type=Path, default=Path("data/Open-SWE-Traces")) + parser.add_argument("--output-root", type=Path, default=Path("runs/pi_mono_converted")) + parser.add_argument("--limit", type=int, default=0, help="Optional total row limit for smoke tests.") + parser.add_argument("--sample", type=int, default=20, help="Number of converted examples to save as sample JSON.") + parser.add_argument("--batch-size", type=int, default=256) + return parser.parse_args() + + +def repo_workspace_name(repo: str) -> str: + return repo.replace("/", "__") + "__1.0" + + +def root_patterns(row: dict[str, Any]) -> list[str]: + repo = row.get("repo") or "" + patterns = ["/testbed"] + if repo: + patterns.append("/workspace/" + repo_workspace_name(repo)) + uploaded = extract_uploaded_path(row.get("trajectory") or []) + if uploaded: + patterns.append(uploaded.rstrip("/")) + return dedupe([p for p in patterns if p]) + + +def extract_uploaded_path(trajectory: list[dict[str, Any]]) -> str | None: + for msg in trajectory: + if msg.get("role") != "user": + continue + content = msg.get("content") or "" + match = re.search(r"\s*(.*?)\s*", content, re.S) + if match: + return match.group(1).strip().splitlines()[0].rstrip("/") + return None + + +def dedupe(items: list[Any]) -> list[Any]: + seen = set() + out = [] + for item in items: + key = json.dumps(item, sort_keys=True, ensure_ascii=False) if isinstance(item, (dict, list)) else str(item) + if key in seen: + continue + seen.add(key) + out.append(item) + return out + + +def normalize_text_paths(text: str, roots: list[str]) -> str: + if not text: + return text + out = text + for root in sorted(roots, key=len, reverse=True): + escaped = re.escape(root.rstrip("/")) + out = re.sub(escaped + r"(?=/|\b|$)", ".", out) + out = re.sub(r"\s*\.\s*", "\n.\n", out) + return out + + +def normalize_path(path: Any, roots: list[str]) -> str: + if not isinstance(path, str) or not path: + return "." + cleaned = normalize_text_paths(path.strip(), roots) + cleaned = cleaned.rstrip("/") + if cleaned in {"", "."}: + return "." + if cleaned.startswith("./"): + return cleaned[2:] or "." + return cleaned + + +def parse_tool_args(raw_args: Any, warnings: list[str]) -> dict[str, Any]: + if isinstance(raw_args, dict): + return raw_args + if not isinstance(raw_args, str): + warnings.append("tool_arguments_not_string_or_dict") + return {} + try: + parsed = json.loads(raw_args) + if isinstance(parsed, dict): + return parsed + warnings.append("tool_arguments_json_not_object") + return {} + except Exception: + warnings.append("tool_arguments_json_parse_failed") + return {} + + +def next_tool_observation(trajectory: list[dict[str, Any]], index: int) -> str: + for msg in trajectory[index + 1 :]: + if msg.get("role") == "tool": + return msg.get("content") or "" + if msg.get("role") == "assistant": + return "" + return "" + + +def looks_like_file_view(path: str, args: dict[str, Any], observation: str) -> bool: + if args.get("view_range"): + return True + if "cat -n" in observation or "result of running" in observation: + return True + if "files and directories" in observation: + return False + name = path.rsplit("/", 1)[-1] + if "." in name: + return True + known_files = {"Makefile", "Dockerfile", "LICENSE", "README", "Rakefile", "Gemfile", "Cargo.toml", "go.mod"} + return name in known_files + + +def line_range_to_offset_limit(view_range: Any) -> dict[str, int]: + if not isinstance(view_range, list) or len(view_range) != 2: + return {} + start, end = view_range + if not isinstance(start, int) or not isinstance(end, int): + return {} + offset = max(start - 1, 0) + if end < start: + return {"offset": offset} + return {"offset": offset, "limit": end - start + 1} + + +def convert_tool_call( + call: dict[str, Any], + roots: list[str], + observation: str, + warnings: list[str], +) -> tuple[dict[str, Any] | None, str | None, bool]: + function = call.get("function") or {} + source_name = function.get("name") or call.get("name") or "" + source_args = parse_tool_args(function.get("arguments", call.get("arguments")), warnings) + call_id = str(call.get("id") or f"call_{abs(hash(json.dumps(call, sort_keys=True, default=str))) % 10**12}") + + if source_name in {"execute_bash", "bash"}: + command = normalize_text_paths(str(source_args.get("command") or ""), roots) + return {"type": "toolCall", "id": call_id, "name": "bash", "arguments": {"command": command}}, None, True + + if source_name == "str_replace_editor": + command = source_args.get("command") + path = normalize_path(source_args.get("path"), roots) + if command == "view": + if looks_like_file_view(path, source_args, observation): + args = {"path": path} + args.update(line_range_to_offset_limit(source_args.get("view_range"))) + return {"type": "toolCall", "id": call_id, "name": "read", "arguments": args}, None, True + return {"type": "toolCall", "id": call_id, "name": "ls", "arguments": {"path": path}}, None, True + if command == "create": + content = normalize_text_paths(str(source_args.get("file_text") or ""), roots) + return {"type": "toolCall", "id": call_id, "name": "write", "arguments": {"path": path, "content": content}}, None, True + if command == "str_replace": + old_text = normalize_text_paths(str(source_args.get("old_str") or ""), roots) + new_text = normalize_text_paths(str(source_args.get("new_str") or ""), roots) + return { + "type": "toolCall", + "id": call_id, + "name": "edit", + "arguments": {"path": path, "edits": [{"oldText": old_text, "newText": new_text}]}, + }, None, True + if command == "insert": + warnings.append("str_replace_editor_insert_mapped_to_bash") + insert_line = source_args.get("insert_line") + new_text = source_args.get("new_str") or source_args.get("insert_str") or "" + py = ( + "python - <<'PY'\n" + "from pathlib import Path\n" + f"path = Path({path!r})\n" + f"line = {int(insert_line) if isinstance(insert_line, int) else 0}\n" + f"text = {new_text!r}\n" + "lines = path.read_text().splitlines(True)\n" + "idx = max(0, min(line, len(lines)))\n" + "lines[idx:idx] = [text if text.endswith('\\n') else text + '\\n']\n" + "path.write_text(''.join(lines))\n" + "PY" + ) + return {"type": "toolCall", "id": call_id, "name": "bash", "arguments": {"command": py}}, None, True + warnings.append(f"unsupported_str_replace_editor_command:{command}") + return None, None, True + + if source_name in {"finish", "submit"}: + text = source_args.get("answer") or source_args.get("summary") or source_args.get("message") or "" + return None, normalize_text_paths(str(text), roots), True + + if source_name == "think": + thought = source_args.get("thought") or source_args.get("content") or "" + return None, normalize_text_paths(str(thought), roots), True + + warnings.append(f"unsupported_tool:{source_name}") + return None, None, True + + +def clean_tool_result(content: str, roots: list[str]) -> str: + text = normalize_text_paths(content or "", roots) + text = re.sub(r"^OBSERVATION:\s*\n?", "", text) + return text + + +def convert_row(row: dict[str, Any], config_name: str) -> dict[str, Any]: + config = CONFIGS[config_name] + warnings: list[str] = [] + roots = root_patterns(row) + trajectory = row.get("trajectory") or [] + pending: list[dict[str, Any]] = [] + skipped_tool_results = 0 + messages: list[dict[str, Any]] = [] + system_messages: list[str] = [] + + for index, msg in enumerate(trajectory): + role = msg.get("role") + if role == "system": + content = normalize_text_paths(msg.get("content") or "", roots) + if content: + system_messages.append(content) + continue + + if role == "user": + messages.append({"role": "user", "content": normalize_text_paths(msg.get("content") or "", roots)}) + continue + + if role == "assistant": + blocks: list[dict[str, Any]] = [] + reasoning = msg.get("reasoning_content") or "" + if config["thinking_mode"] == "thinking" and reasoning.strip(): + blocks.append({"type": "thinking", "thinking": normalize_text_paths(reasoning, roots)}) + visible = msg.get("content") or "" + if visible.strip(): + blocks.append({"type": "text", "text": normalize_text_paths(visible, roots)}) + + final_texts: list[str] = [] + observation = next_tool_observation(trajectory, index) + for call in msg.get("tool_calls") or []: + converted, final_text, consumes_result = convert_tool_call(call, roots, observation, warnings) + if converted is not None: + blocks.append(converted) + pending.append(converted) + elif consumes_result: + skipped_tool_results += 1 + if final_text: + final_texts.append(final_text) + for text in final_texts: + if text.strip(): + blocks.append({"type": "text", "text": text}) + if blocks: + stop_reason = "toolUse" if any(block.get("type") == "toolCall" for block in blocks) else "stop" + messages.append({"role": "assistant", "content": blocks, "stopReason": stop_reason}) + continue + + if role == "tool": + if not pending: + if skipped_tool_results > 0: + skipped_tool_results -= 1 + continue + warnings.append("tool_result_without_pending_call") + continue + call = pending.pop(0) + messages.append( + { + "role": "toolResult", + "toolCallId": call["id"], + "toolName": call["name"], + "content": [{"type": "text", "text": clean_tool_result(msg.get("content") or "", roots)}], + "isError": infer_tool_error(msg.get("content") or ""), + } + ) + continue + + warnings.append(f"unsupported_role:{role}") + + if pending: + warnings.append(f"pending_tool_calls_unmatched:{len(pending)}") + + return { + "instance_id": row.get("instance_id"), + "repo": row.get("repo"), + "license": row.get("license"), + "language": row.get("language"), + "trajectory_id": row.get("trajectory_id"), + "config": config_name, + "scaffold": config["scaffold"], + "model_family": config["model_family"], + "thinking_mode": config["thinking_mode"], + "resolved": int(row.get("resolved") or 0), + "metadata": row.get("metadata") or {}, + "repo_roots": roots, + "system_messages": system_messages, + "pi_mono_messages": messages, + "model_patch": normalize_text_paths(row.get("model_patch") or "", roots), + "conversion_warnings": dedupe(warnings), + } + + +def infer_tool_error(content: str) -> bool: + text = content or "" + error_markers = [ + "Traceback (most recent call last)", + "Command failed", + "exit code 1", + "Invalid `view_range`", + "No such file or directory", + ] + return any(marker in text for marker in error_markers) + + +def write_jsonl_gz(path: Path, rows: list[dict[str, Any]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with gzip.open(path, "at", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n") + + +def convert_file(file: Path, out_file: Path, config_name: str, batch_size: int, remaining: int | None) -> dict[str, Any]: + stats = Counter() + samples: list[dict[str, Any]] = [] + parquet = pq.ParquetFile(file) + if out_file.exists(): + out_file.unlink() + for batch in parquet.iter_batches(batch_size=batch_size): + rows = batch.to_pylist() + if remaining is not None: + rows = rows[:remaining] + converted = [convert_row(row, config_name) for row in rows] + write_jsonl_gz(out_file, converted) + for item in converted: + stats["rows"] += 1 + stats["messages"] += len(item["pi_mono_messages"]) + stats["warnings"] += len(item["conversion_warnings"]) + stats[f"resolved_{item['resolved']}"] += 1 + for warning in item["conversion_warnings"]: + stats[f"warning:{warning}"] += 1 + if len(samples) < 3: + samples.append(item) + if remaining is not None: + remaining -= len(rows) + if remaining <= 0: + break + return {"stats": dict(stats), "samples": samples} + + +def main() -> int: + args = parse_args() + data_root = args.input_root / "data" + args.output_root.mkdir(parents=True, exist_ok=True) + all_stats: dict[str, Any] = { + "input_root": str(args.input_root), + "output_root": str(args.output_root), + "configs": {}, + } + samples: list[dict[str, Any]] = [] + remaining = args.limit or None + + for config_name in CONFIGS: + config_dir = data_root / config_name + out_dir = args.output_root / config_name + out_dir.mkdir(parents=True, exist_ok=True) + config_stats = Counter() + for file in sorted(config_dir.glob("*.parquet")): + if remaining is not None and remaining <= 0: + break + out_file = out_dir / (file.stem + ".pi_mono.jsonl.gz") + result = convert_file(file, out_file, config_name, args.batch_size, remaining) + file_stats = Counter(result["stats"]) + config_stats.update(file_stats) + samples.extend(result["samples"]) + if remaining is not None: + remaining -= file_stats["rows"] + print(json.dumps({"file": str(file), "out": str(out_file), "stats": dict(file_stats)}, ensure_ascii=False)) + all_stats["configs"][config_name] = dict(config_stats) + + report_path = args.output_root / "conversion_report.json" + report_path.write_text(json.dumps(all_stats, indent=2, ensure_ascii=False), encoding="utf-8") + sample_path = args.output_root / "sample_converted.json" + sample_path.write_text(json.dumps(samples[: args.sample], indent=2, ensure_ascii=False), encoding="utf-8") + print(json.dumps({"report": str(report_path), "sample": str(sample_path)}, ensure_ascii=False)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/ti_coding_agent_data_prep/__init__.py b/src/ti_coding_agent_data_prep/__init__.py new file mode 100644 index 0000000..0b87515 --- /dev/null +++ b/src/ti_coding_agent_data_prep/__init__.py @@ -0,0 +1,6 @@ +"""Utilities for coding-agent data preparation.""" + +__all__ = ["__version__"] + +__version__ = "0.1.0" + diff --git a/src/ti_coding_agent_data_prep/openswe/__init__.py b/src/ti_coding_agent_data_prep/openswe/__init__.py new file mode 100644 index 0000000..0a608fe --- /dev/null +++ b/src/ti_coding_agent_data_prep/openswe/__init__.py @@ -0,0 +1,6 @@ +"""Open-SWE-Traces data-preparation helpers.""" + +from .constants import CONFIG_META, CONFIGS + +__all__ = ["CONFIGS", "CONFIG_META"] + diff --git a/src/ti_coding_agent_data_prep/openswe/constants.py b/src/ti_coding_agent_data_prep/openswe/constants.py new file mode 100644 index 0000000..e842d73 --- /dev/null +++ b/src/ti_coding_agent_data_prep/openswe/constants.py @@ -0,0 +1,32 @@ +"""Shared Open-SWE-Traces configuration names and metadata.""" + +CONFIGS = [ + "minimax_m25_openhands_trajectories", + "minimax_m25_sweagent_trajectories", + "qwen35_openhands_trajectories", + "qwen35_sweagent_trajectories", +] + +CONFIG_META = { + "minimax_m25_openhands_trajectories": { + "model_family": "minimax_m25", + "scaffold": "openhands", + "thinking_mode": "thinking", + }, + "minimax_m25_sweagent_trajectories": { + "model_family": "minimax_m25", + "scaffold": "sweagent", + "thinking_mode": "thinking", + }, + "qwen35_openhands_trajectories": { + "model_family": "qwen35", + "scaffold": "openhands", + "thinking_mode": "non_thinking", + }, + "qwen35_sweagent_trajectories": { + "model_family": "qwen35", + "scaffold": "sweagent", + "thinking_mode": "non_thinking", + }, +} + diff --git a/src/ti_coding_agent_data_prep/openswe/paths.py b/src/ti_coding_agent_data_prep/openswe/paths.py new file mode 100644 index 0000000..fdd1bfa --- /dev/null +++ b/src/ti_coding_agent_data_prep/openswe/paths.py @@ -0,0 +1,20 @@ +"""Path helpers for repo-local Open-SWE-Traces workflows.""" + +from __future__ import annotations + +from pathlib import Path + + +def repo_root_from_script(script_file: str | Path) -> Path: + """Return the project root for files under scripts//.""" + + return Path(script_file).resolve().parents[2] + + +def default_dataset_root(repo_root: Path) -> Path: + return repo_root / "data" / "Open-SWE-Traces" + + +def default_runs_root(repo_root: Path) -> Path: + return repo_root / "runs" +