Open-SWE-Traces：代理软件工程轨迹分析与SFT构建
文章摘要
该教程围绕Open - SWE - Traces数据集展开，介绍AI驱动的软件工程轨迹研究与微调。先进行环境依赖安装与配置，定义轨迹解析工具函数，流式加载数据并检查。随后构建分析数据框，进行可视化，分析Token预算、工具使用情况，最后筛选构建精选监督微调子集并导出文件。
本教程将围绕Open-SWE-Traces数据集展开，这一开源资源专为AI驱动的软件工程轨迹研究与微调准备而设计。我们将采用流式加载的方式直接从开源数据集平台获取数据，无需在本地下载完整的庞大数据集，即可在云端开发环境中高效开展工作。整个流程涵盖单条数据记录的详细检查、多轮代理对话的标准化处理、最终代码补丁的解析、元数据提取，以及构建分析数据框以研究轨迹长度、工具使用情况、补丁规模、语言分布和任务解决结果等核心特征。基于这些分析结果，我们将筛选出符合要求的高质量轨迹数据，构建用于监督微调的精选子集，筛选标准包括任务成功标签、Token长度限制、语言过滤以及有效补丁的可用性。
环境依赖安装与基础配置
首先我们需要安装并导入流式加载、数据解析、分析与可视化所需的核心依赖库。为了确保在云端环境中表格和图表的可读性，我们会对pandas和matplotlib的展示参数进行配置。同时我们会预先定义数据集标识、代理与模型组合、采样规模以及监督微调的过滤规则，这些配置将贯穿整个教程的后续步骤。
import subprocess, sys
def _pip(*pkgs):
subprocess.run([sys.executable, "-m", "pip", "install", "-q", *pkgs], check=False)
_pip("-U", "datasets", "huggingface_hub")
_pip("tiktoken", "pandas", "matplotlib")
import json
import re
import textwrap
from itertools import islice
from collections import Counter
import pandas as pd
import matplotlib.pyplot as plt
from datasets import load_dataset
pd.set_option("display.max_columns", 50)
pd.set_option("display.width", 160)
plt.rcParams.update({
"figure.figsize": (9, 4.6),
"figure.dpi": 110,
"axes.grid": True,
"grid.alpha": 0.25,
"axes.spines.top": False,
"axes.spines.right": False,
"font.size": 11,
"axes.titlesize": 13,
"axes.titleweight": "bold",
})
BLUE, ORANGE, GREEN, RED = "#4C72B0", "#DD8452", "#55A868", "#C44E52"
def banner(title):
line = "=" * 78
print(f"\n{line}\n {title}\n{line}")
DATASET = "nvidia/Open-SWE-Traces"
AGENTS = ["openhands", "sweagent"]
MODELS = ["minimax_m25", "qwen35_122b"]
SAMPLE_ALL = True
PER_COMBO = 400
N_SINGLE = 1500
MAX_SFT_TOKENS = 32000
SFT_REQUIRE_RESOLVED = True
SFT_LANGUAGES = None
轨迹解析工具函数定义
为了让数据集处理更便捷，即使字段格式存在差异也能稳定运行，我们将定义一系列辅助工具函数。这些函数包括轨迹标准化、消息文本提取、角色计数、工具使用检测、代码补丁解析以及Token长度估算等功能。我们采用防御性编程的方式构建这些工具，确保在处理大规模流式数据集时，即使数据结构存在细微差异，分析流程也能保持稳定。
def message_text(msg):
if not isinstance(msg, dict):
return ""
content = msg.get("content", "")
if content is None:
return ""
if isinstance(content, str):
return content
if isinstance(content, list):
parts = []
for block in content:
if isinstance(block, dict):
parts.append(block.get("text") or block.get("content") or "")
elif isinstance(block, str):
parts.append(block)
return "\n".join(p for p in parts if p)
return str(content)
def normalize_trajectory(traj):
if traj is None:
return []
if isinstance(traj, str):
try:
traj = json.loads(traj)
except Exception:
return []
norm = []
for msg in traj:
if isinstance(msg, str):
try:
msg = json.loads(msg)
except Exception:
msg = {"role": "unknown", "content": msg}
if isinstance(msg, dict):
norm.append(msg)
return norm
def normalize_metadata(meta):
if isinstance(meta, str):
try:
return json.loads(meta)
except Exception:
return {}
return meta if isinstance(meta, dict) else {}
def role_counts(trajectory):
c = Counter()
for msg in trajectory or []:
if isinstance(msg, dict):
c[msg.get("role", "unknown")] += 1
return c
_FUNC_XML = re.compile(r"<function\s*=\s*([a-zA-Z0-9_\-]+)", re.IGNORECASE)
_EXEC_TAG = re.compile(r"<(execute_[a-z]+)>", re.IGNORECASE)
_BASH_FENCE = re.compile(r"```(?:bash|sh|shell)\b", re.IGNORECASE)
def extract_tool_names(trajectory):
names = Counter()
for msg in trajectory or []:
if not isinstance(msg, dict):
continue
for call in msg.get("tool_calls") or []:
fn = (call or {}).get("function", {}) if isinstance(call, dict) else {}
if fn.get("name"):
names[fn["name"]] += 1
if msg.get("role") == "tool" and msg.get("name"):
names[msg["name"]] += 1
if msg.get("role") == "assistant":
text = message_text(msg)
for m in _FUNC_XML.findall(text):
names[m.lower()] += 1
for m in _EXEC_TAG.findall(text):
names[m.lower()] += 1
if _BASH_FENCE.search(text):
names["bash_block"] += 1
return names
def parse_patch(diff_text):
if not diff_text or not isinstance(diff_text, str):
return 0, 0, 0, [], Counter()
files, exts = [], Counter()
additions = deletions = 0
for line in diff_text.splitlines():
if line.startswith("diff --git"):
parts = line.split()
if len(parts) >= 3:
path = parts[2][2:] if parts[2].startswith("a/") else parts[2]
files.append(path)
base = path.split("/")[-1]
if "." in base:
exts[base.rsplit(".", 1)[-1].lower()] += 1
elif line.startswith("+") and not line.startswith("+++"):
additions += 1
elif line.startswith("-") and not line.startswith("---"):
deletions += 1
return len(files), additions, deletions, files, exts
def make_token_counter():
try:
import tiktoken
enc = tiktoken.get_encoding("cl100k_base")
return lambda s: len(enc.encode(s, disallowed_special=()))
except Exception:
return lambda s: max(1, len(s) // 4)
count_tokens = make_token_counter()
流式加载与轨迹数据检查
我们将直接从开源数据集平台流式加载Open-SWE-Traces的采样数据，而非下载完整数据集。我们会收集不同代理与模型组合下的示例数据，随后详细检查单条记录的结构，逐一浏览轨迹的前几条对话消息，并预览最终的代码补丁，以此理解每个训练示例的完整构成。
def stream_take(agent, model, n):
ds = load_dataset(DATASET, agent, split=model, streaming=True)
rows = []
for ex in islice(ds, n):
ex = dict(ex)
ex["_agent"], ex["_model"] = agent, model
rows.append(ex)
return rows
banner("STEP 1 — Streaming trajectories from the Hub")
raw_rows = []
if SAMPLE_ALL:
combos = [(a, m) for a in AGENTS for m in MODELS]
for agent, model in combos:
try:
part = stream_take(agent, model, PER_COMBO)
raw_rows.extend(part)
print(f" ✓ {agent:<10} / {model:<12} -> {len(part):>4} rows")
except Exception as e:
print(f" ✗ {agent}/{model} failed: {type(e).__name__}: {e}")
else:
raw_rows = stream_take(AGENTS[0], MODELS[0], N_SINGLE)
print(f" ✓ {AGENTS[0]} / {MODELS[0]} -> {len(raw_rows)} rows")
print(f"\n Total rows pulled into memory: {len(raw_rows)}")
assert raw_rows, "No rows streamed — check your internet connection and retry."
banner("STEP 2 — Anatomy of a single record")
sample = raw_rows[0]
print("Top-level fields :", list(sample.keys()))
print("instance_id :", sample.get("instance_id"))
print("repo / language :", sample.get("repo"), "/", sample.get("language"))
print("license :", sample.get("license"))
print("resolved (1/0/-1):", sample.get("resolved"))
print("metadata :", normalize_metadata(sample.get("metadata")))
traj0 = normalize_trajectory(sample.get("trajectory"))
print(f"\nTrajectory has {len(traj0)} messages. Role histogram: {dict(role_counts(traj0))}")
print("\n--- Trajectory walkthrough (each message truncated to 240 chars) ---")
for i, msg in enumerate(traj0[:8]):
role = msg.get("role", "unknown").upper()
body = " ".join(message_text(msg).split())
print(f"\n[{i}] {role}")
print(textwrap.fill(body[:240] + ("…" if len(body) > 240 else ""),
width=92, subsequent_indent=" "))
if len(traj0) > 8:
print(f"\n… (+{len(traj0) - 8} more messages)")
print("\n--- Final patch (model_patch), first 25 lines ---")
print("\n".join((sample.get("model_patch") or "").splitlines()[:25]) or "(empty)")
构建分析用数据框
我们会将流式加载的原始数据转换为结构化的pandas数据框，以方便后续分析。这一步会提取轨迹层面的各类特征，包括消息数量、角色分布、补丁变更规模、Token估算值、元数据字段以及工具使用计数器等。同时我们会创建任务解决状态标记，用于对比成功与未成功的软件工程轨迹。
banner("STEP 3 — Building the analysis DataFrame")
def process_example(ex):
traj = normalize_trajectory(ex.get("trajectory"))
rc = role_counts(traj)
nf, add, dele, _files, _exts = parse_patch(ex.get("model_patch"))
meta = normalize_metadata(ex.get("metadata"))
full_text = "\n".join(message_text(m) for m in traj)
return {
"instance_id": ex.get("instance_id"),
"repo": ex.get("repo"),
"language": (ex.get("language") or "unknown").lower(),
"license": ex.get("license"),
"resolved": ex.get("resolved"),
"agent": ex.get("_agent"),
"model": ex.get("_model"),
"n_messages": len(traj),
"n_system": rc.get("system", 0),
"n_user": rc.get("user", 0),
"n_assistant": rc.get("assistant", 0),
"n_tool": rc.get("tool", 0),
"patch_files": nf,
"patch_add": add,
"patch_del": dele,
"patch_churn": add + dele,
"traj_tokens": count_tokens(full_text),
"category": meta.get("category"),
"meta_files": meta.get("num_modified_files"),
"meta_lines": meta.get("num_modified_lines"),
"_tools": extract_tool_names(traj),
}
records = [process_example(ex) for ex in raw_rows]
df = pd.DataFrame(records)
df["is_resolved"] = (df["resolved"] == 1)
df["known_label"] = df["resolved"].isin([0, 1])
print(f"DataFrame: {df.shape[0]} rows x {df.shape[1]} cols")
print("\nNumeric summary:")
print(df[["n_messages", "n_assistant", "n_tool",
"patch_files", "patch_churn", "traj_tokens"]].describe().round(1))
轨迹数据分布可视化
我们将通过多维度的可视化图表探索数据集特征，包括语言分布、任务解决率、代理与模型组合对比、消息长度分布、补丁规模分布以及Token预算分析。我们会可视化轨迹长度、Token大小和工具使用情况在采样数据中的变化规律，以此判断在不同上下文窗口限制下哪些示例适合用于微调训练。
banner("STEP 4 — Distributions & visualizations")
lang_counts = df["language"].value_counts()
print("Trajectories per language:\n", lang_counts.to_string())
ax = lang_counts.plot(kind="bar", color=BLUE)
ax.set_title("Trajectories per language (sample)")
ax.set_xlabel(""); ax.set_ylabel("count")
plt.tight_layout(); plt.show()
known = df[df["known_label"]]
by_lang = (known.groupby("language")["is_resolved"]
.agg(rate="mean", n="size")
.query("n >= 25")
.sort_values("rate", ascending=False))
print("\nResolution rate by language (n>=25):\n", by_lang.round(3).to_string())
if not by_lang.empty:
ax = by_lang["rate"].plot(kind="bar", color=GREEN)
ax.set_title("Resolution rate by language")
ax.set_xlabel(""); ax.set_ylabel("fraction resolved"); ax.set_ylim(0, 1)
plt.tight_layout(); plt.show()
if known["agent"].nunique() > 1 or known["model"].nunique() > 1:
pivot = (known.groupby(["agent", "model"])["is_resolved"].mean().unstack())
print("\nResolution rate by scaffold x model:\n", pivot.round(3).to_string())
ax = pivot.plot(kind="bar", color=[BLUE, ORANGE])
ax.set_title("Resolution rate: scaffold x model")
ax.set_xlabel("agent"); ax.set_ylabel("fraction resolved"); ax.set_ylim(0, 1)
ax.legend(title="model"); plt.tight_layout(); plt.show()
ax = df["n_messages"].plot(kind="hist", bins=40, color=BLUE, alpha=0.85)
ax.set_title("Messages per trajectory")
ax.set_xlabel("number of messages"); ax.set_ylabel("trajectories")
plt.tight_layout(); plt.show()
churn = df["patch_churn"].clip(upper=df["patch_churn"].quantile(0.97))
ax = churn.plot(kind="hist", bins=40, color=ORANGE, alpha=0.85)
ax.set_title("Patch size — lines changed (clipped at p97)")
ax.set_xlabel("added + deleted lines"); ax.set_ylabel("trajectories")
plt.tight_layout(); plt.show()
if known["is_resolved"].nunique() > 1:
fig, ax = plt.subplots()
for flag, color, lab in [(True, GREEN, "resolved"), (False, RED, "unresolved")]:
sub = known[known["is_resolved"] == flag]
ax.scatter(sub["n_messages"], sub["traj_tokens"],
s=10, alpha=0.4, color=color, label=lab)
ax.set_title("Trajectory length vs. token size, by outcome")
ax.set_xlabel("messages"); ax.set_ylabel("estimated tokens")
ax.legend(); plt.tight_layout(); plt.show()
Token预算需求分析
这一步我们会分析每条轨迹的Token使用情况，帮助确定微调任务所需的模型上下文窗口大小。我们会计算不同分位数下的Token消耗，并统计不同上下文窗口能够覆盖的轨迹比例，以此为选择合适的微调模型提供参考依据。
banner("STEP 5 — Token budget (what context window do you need?)")
tok = df["traj_tokens"]
print("Estimated tokens per trajectory — percentiles:")
for p in [50, 75, 90, 95, 99]:
print(f" p{p:<2}: {int(tok.quantile(p/100)):>8,}")
print(f" max: {int(tok.max()):>8,}")
windows = [8_192, 16_384, 32_768, 65_536, 131_072]
print("\nFraction of trajectories that fit in a given context window:")
for w in windows:
frac = (tok <= w).mean()
print(f" {w:>7,} tokens : {frac*100:5.1f}%")
ax = tok.clip(upper=tok.quantile(0.99)).plot(kind="hist", bins=50,
color=BLUE, alpha=0.85)
for w, c in zip([8_192, 32_768, 131_072], [GREEN, ORANGE, RED]):
if w <= tok.quantile(0.99):
ax.axvline(w, color=c, ls="--", lw=1.5, label=f"{w//1024}k ctx")
ax.set_title("Trajectory token-length distribution (clipped at p99)")
ax.set_xlabel("estimated tokens"); ax.set_ylabel("trajectories")
ax.legend(); plt.tight_layout(); plt.show()
代理工具使用情况统计
我们会统计代理在轨迹中使用的各类工具与操作，分析高频使用的动作类型，并对比成功与未成功任务中的工具使用差异，以此理解不同任务场景下的代理行为模式。
banner("STEP 6 — Which tools/actions do the agents use?")
tool_totals = Counter()
for t in df["_tools"]:
tool_totals.update(t)
top_tools = tool_totals.most_common(12)
if top_tools:
print("Most frequent agent actions (across the sample):")
for name, cnt in top_tools:
print(f" {name:<24} {cnt:>7,}")
labels, vals = zip(*top_tools)
fig, ax = plt.subplots(figsize=(9, 5))
ax.barh(range(len(labels)), vals, color=BLUE)
ax.set_yticks(range(len(labels))); ax.set_yticklabels(labels)
ax.invert_yaxis()
ax.set_title("Top agent actions / tool invocations")
ax.set_xlabel("count"); plt.tight_layout(); plt.show()
else:
print("No tool actions detected with the current heuristics.")
if known["is_resolved"].nunique() > 1:
print("\nMean 'tool' (environment) turns by outcome:")
print(known.groupby("is_resolved")["n_tool"].mean().round(2).to_string())
构建精选监督微调子集
我们会将筛选后的轨迹转换为标准化的监督微调格式，包括使用标准消息字典和可选的ChatML风格文本表示。我们会根据任务解决状态、Token预算、语言过滤和有效补丁可用性等规则筛选示例，确保精选子集的质量。最后我们会将分析结果导出为CSV文件，将微调数据集导出为JSONL文件，方便后续训练使用。
banner("STEP 7 — Building a curated SFT subset")
def to_chatml(trajectory):
out = []
for m in trajectory:
role = m.get("role", "unknown")
out.append(f"<|im_start|>{role}\n{message_text(m).strip()}<|im_end|>")
return "\n".join(out)
def passes_filters(rec, raw):
if SFT_REQUIRE_RESOLVED and rec["resolved"] != 1:
return False
if rec["traj_tokens"] > MAX_SFT_TOKENS:
return False
if SFT_LANGUAGES is not None and rec["language"] not in SFT_LANGUAGES:
return False
if not (raw.get("model_patch") or "").strip():
return False
return True
sft_examples = []
for rec, raw in zip(records, raw_rows):
if not passes_filters(rec, raw):
continue
messages = [{"role": m.get("role"), "content": message_text(m)}
for m in normalize_trajectory(raw.get("trajectory"))]
sft_examples.append({
"instance_id": rec["instance_id"],
"repo": rec["repo"],
"language": rec["language"],
"agent": rec["agent"],
"model": rec["model"],
"messages": messages,
"text": to_chatml(messages),
"model_patch": raw.get("model_patch"),
"approx_tokens": rec["traj_tokens"],
})
print(f"Kept {len(sft_examples)} / {len(records)} trajectories after filtering")
print(f" filters -> resolved_only={SFT_REQUIRE_RESOLVED}, "
f"max_tokens={MAX_SFT_TOKENS:,}, languages={SFT_LANGUAGES or 'all'}")
if sft_examples:
kept = pd.DataFrame(sft_examples)
print("\nCurated subset by language:\n", kept["language"].value_counts().to_string())
print("\n--- One formatted SFT example (ChatML, truncated) ---")
print(sft_examples[0]["text"][:600], "…")
banner("STEP 8 — Exporting artifacts")
csv_path = "open_swe_traces_analysis.csv"
df.drop(columns=["_tools"]).to_csv(csv_path, index=False)
print(f" Wrote analysis table -> {csv_path} ({len(df)} rows)")
jsonl_path = "open_swe_sft.jsonl"
with open(jsonl_path, "w", encoding="utf-8") as f:
for ex in sft_examples:
f.write(json.dumps(ex, ensure_ascii=False) + "\n")
print(f" Wrote SFT dataset -> {jsonl_path} ({len(sft_examples)} rows)")
print("\nDone. In Colab, open the Files pane (folder icon, left) to download both.")
print("To load the SFT file later: datasets.load_dataset('json', "
"data_files='open_swe_sft.jsonl')")
总结
本教程完整实现了从原始Open-SWE-Traces数据集到结构化分析数据与可用于监督微调的训练数据的转换流程。我们学习了如何流式加载轨迹数据、检查代理行为、测量Token预算、对比不同代理与模型的表现、分析补丁特征，以及导出分析表格与JSONL训练文件。最终我们得到了一个可复用的框架，可以扩展用于更大规模的采样、特定语言的微调、更深入的工具使用分析以及针对特定模型的对话模板格式化。
塔猴是一个专注于为用户提供系统学习、内容创作与商业连接的AIGC综合服务平台，致力于为每一位AI探索者打造理想的创作、成长家园。在塔猴，你不仅可以学习众多AIGC类实战课程，获得与时俱进的AIGC技能和视野，还有机会获得长期商业合作和接单机会！点击进入：https://www.tahou.com/
AI生成内容提示：本文由人工智能辅助创作，内容仅供参考，不代表平台观点。请注意核实信息的准确性，并理性判断。
以上内容不代表本平台立场，仅供读者参考
