跳到主内容
@wquguru
精选85MarkTechPost（RSS）技巧与观点

用 XYZ-Aquila-SFT 与 Qwen3 微调工具调用 LLM 完整指南

Fine-Tuning Tool-Calling LLMs: A Complete Guide Using XYZ-Aquila-SFT and Qwen3

原文
推荐理由

做工具调用微调的同学必看，这份指南从数据解析到 LoRA 训练再到评估全流程可照做，赶紧收藏跑一遍。

In this tutorial, we implement an end-to-end supervised fine-tuning pipeline for the XYZ-Aquila-SFT dataset, Hugging Face Transformers, PyTorch, and PEFT. We stream and inspect the dataset, parse multi-turn tool-use trajectories, extract structured tool calls, analyze corpus characteristics, and preserve embedded reasoning and observation patterns. We then convert tool schemas between message-embedded and structured formats, render Qwen-compatible ChatML with assistant-only loss masking, prepare a custom PyTorch dataset and collator, and fine-tune Qwen3-0.6B with LoRA. Finally, we evaluate tool-call prediction before and after training and export both the transformed dataset and corpus statistics for further experimentation.

在本教程中，我们为XYZ-Aquila-SFT数据集、Hugging Face Transformers、PyTorch和PEFT实现了一个端到端的监督微调流水线。我们流式读取并检查数据集，解析多轮工具使用轨迹，提取结构化工具调用，分析语料特征，并保留嵌入的推理和观察模式。然后，我们将工具模式在消息嵌入和结构化格式之间转换，渲染兼容Qwen的ChatML格式，并带有仅助手损失的掩码，准备自定义的PyTorch数据集和collator，并使用LoRA微调Qwen3-0.6B。最后，我们评估训练前后的工具调用预测，并导出转换后的数据集和语料统计信息以供进一步实验。

代码 · 45 行
import os, sys, subprocess
CFG = dict(
   REPO            = "XYZAILab/XYZ-Aquila-SFT",
   LANG            = "en",
   N_STREAM        = 400,
   N_EVAL          = 40,
   MODEL_ID        = "Qwen/Qwen3-0.6B",
   MAX_SEQ_LEN     = 2048,
   LENGTH_POLICY   = "truncate",
   RUN_TRAINING    = True,
   MAX_STEPS       = 30,
   GRAD_ACCUM      = 8,
   LR              = 1e-4,
   LORA_R          = 16,
   RUN_EVAL        = True,
   N_EVAL_PROBES   = 24,
   OUT_DIR         = "/content/aquila_out",
   SEED            = 0,
)
os.makedirs(CFG["OUT_DIR"], exist_ok=True)
def pip(*pkgs):
   subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-U", *pkgs], check=False)
pip("datasets>=3.0.0", "transformers>=4.51.0", "peft>=0.13.0", "accelerate>=1.0.0")
import json, re, math, random, statistics as stats
from collections import Counter, defaultdict
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
import torch
import matplotlib.pyplot as plt
from datasets import load_dataset
from transformers import AutoTokenizer, AutoModelForCausalLM, get_cosine_schedule_with_warmup
random.seed(CFG["SEED"]); torch.manual_seed(CFG["SEED"])
DEV = "cuda" if torch.cuda.is_available() else "cpu"
BF16 = DEV == "cuda" and torch.cuda.is_bf16_supported()
print(f"device={DEV}  bf16={BF16}  torch={torch.__version__}")
print(f"\n[1] streaming {CFG['REPO']}:{CFG['LANG']} ...")
stream = load_dataset(CFG["REPO"], CFG["LANG"], split="train", streaming=True)
RAW: List[Dict[str, Any]] = list(stream.take(CFG["N_STREAM"]))
print(f"    pulled {len(RAW)} rows; keys = {list(RAW[0].keys())}")
_r = RAW[0]
print(f"    question[:110]        : {_r['question'][:110]}...")
print(f"    answer                : {_r['answer'][:80]}")
print(f"    number of tool calls  : {_r['number of tool calls']}")
print(f"    trajectory len        : {len(_r['trajectory'])} msgs")
print(f"    role sequence (first8): {[m['role'] for m in _r['trajectory'][:8]]}")
代码 · 45 行
import os, sys, subprocess
CFG = dict(
   REPO            = "XYZAILab/XYZ-Aquila-SFT",
   LANG            = "en",
   N_STREAM        = 400,
   N_EVAL          = 40,
   MODEL_ID        = "Qwen/Qwen3-0.6B",
   MAX_SEQ_LEN     = 2048,
   LENGTH_POLICY   = "truncate",
   RUN_TRAINING    = True,
   MAX_STEPS       = 30,
   GRAD_ACCUM      = 8,
   LR              = 1e-4,
   LORA_R          = 16,
   RUN_EVAL        = True,
   N_EVAL_PROBES   = 24,
   OUT_DIR         = "/content/aquila_out",
   SEED            = 0,
)
os.makedirs(CFG["OUT_DIR"], exist_ok=True)
def pip(*pkgs):
   subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-U", *pkgs], check=False)
pip("datasets>=3.0.0", "transformers>=4.51.0", "peft>=0.13.0", "accelerate>=1.0.0")
import json, re, math, random, statistics as stats
from collections import Counter, defaultdict
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
import torch
import matplotlib.pyplot as plt
from datasets import load_dataset
from transformers import AutoTokenizer, AutoModelForCausalLM, get_cosine_schedule_with_warmup
random.seed(CFG["SEED"]); torch.manual_seed(CFG["SEED"])
DEV = "cuda" if torch.cuda.is_available() else "cpu"
BF16 = DEV == "cuda" and torch.cuda.is_bf16_supported()
print(f"device={DEV}  bf16={BF16}  torch={torch.__version__}")
print(f"\n[1] streaming {CFG['REPO']}:{CFG['LANG']} ...")
stream = load_dataset(CFG["REPO"], CFG["LANG"], split="train", streaming=True)
RAW: List[Dict[str, Any]] = list(stream.take(CFG["N_STREAM"]))
print(f"    pulled {len(RAW)} rows; keys = {list(RAW[0].keys())}")
_r = RAW[0]
print(f"    question[:110]        : {_r['question'][:110]}...")
print(f"    answer                : {_r['answer'][:80]}")
print(f"    number of tool calls  : {_r['number of tool calls']}")
print(f"    trajectory len        : {len(_r['trajectory'])} msgs")
print(f"    role sequence (first8): {[m['role'] for m in _r['trajectory'][:8]]}")

We configure the dataset, model, training parameters, output directory, and reproducibility settings for the complete workflow. We install the required Hugging Face, PEFT, Accelerate, and PyTorch-related dependencies and detect whether a CUDA GPU and BF16 support are available. We then stream a limited number of XYZ-Aquila-SFT examples, inspect the dataset schema, and examine the structure of the first tool-use trajectory.

我们为整个工作流配置数据集、模型、训练参数、输出目录和可复现性设置。我们安装所需的Hugging Face、PEFT、Accelerate和PyTorch相关依赖，并检测CUDA GPU和BF16支持是否可用。然后，我们流式读取有限数量的XYZ-Aquila-SFT示例，检查数据集模式，并检查第一个工具使用轨迹的结构。

代码 · 101 行
TOOLS_BLOCK_RE = re.compile(r"<tools>\s*(.*?)\s*</tools>", re.S)
THINK_RE       = re.compile(r"<think>(.*?)</think>", re.S)
TOOL_RESP_RE   = re.compile(r"<tool_response>\s*(.*?)\s*</tool_response>", re.S)
TOOLS_HDR_RE   = re.compile(r"\n\n# Tools\n\n")
def iter_json_objects(text: str, limit: int = 1):
   """Nesting-safe JSON scanner. Regex like r'\\{.*?\\}' breaks on nested
   `arguments` objects, which every real tool call has."""
   dec, i, n, out = json.JSONDecoder(), 0, len(text), []
   while i < n and len(out) < limit:
       while i < n and text[i] not in "{[":
           i += 1
       if i >= n:
           break
       try:
           obj, end = dec.raw_decode(text, i)
       except json.JSONDecodeError:
           i += 1
           continue
       out.append(obj); i = end
   return out
def parse_tool_calls(content: str) -> List[Dict[str, Any]]:
   calls = []
   for m in re.finditer(r"<tool_call>", content):
       got = iter_json_objects(content[m.end():], limit=1)
       if got:
           calls.append(got[0])
   return calls
@dataclass
class Trajectory:
   question: str
   answer: str
   declared_calls: int
   messages: List[Dict[str, str]]
   system_core: str = ""
   tools: List[Dict[str, Any]] = field(default_factory=list)
   tools_suffix: str = ""
   calls: List[Dict[str, Any]] = field(default_factory=list)
   n_observations: int = 0
   n_think: int = 0
   @property
   def tool_names(self):  return [c.get("name", "?") for c in self.calls]
   @property
   def depth(self):       return len(self.messages)
def parse_row(row: Dict[str, Any]) -> Trajectory:
   msgs = [{"role": m["role"], "content": m["content"]} for m in row["trajectory"]]
   t = Trajectory(row["question"], row["answer"], row["number of tool calls"], msgs)
   if msgs and msgs[0]["role"] == "system":
       sysmsg = msgs[0]["content"]
       split = TOOLS_HDR_RE.search(sysmsg)
       if split:
           t.system_core   = sysmsg[:split.start()]
           t.tools_suffix  = sysmsg[split.start():]
       else:
           t.system_core = sysmsg
       blk = TOOLS_BLOCK_RE.search(sysmsg)
       if blk:
           t.tools = iter_json_objects(blk.group(1), limit=64)
   for m in msgs:
       if m["role"] == "assistant":
           t.calls += parse_tool_calls(m["content"])
           t.n_think += len(THINK_RE.findall(m["content"]))
       else:
           t.n_observations += len(TOOL_RESP_RE.findall(m["content"]))
   return t
TRAJ = [parse_row(r) for r in RAW]
t0 = TRAJ[0]
print(f"\n[2] parsed {len(TRAJ)} trajectories")
print(f"    tool schemas found : {[fn.get('function', fn).get('name') for fn in t0.tools]}")
print(f"    parsed calls       : {len(t0.calls)}  (declared {t0.declared_calls})")
print(f"    observations       : {t0.n_observations}   think blocks: {t0.n_think}")
if t0.calls:
   print(f"    sample call        : {json.dumps(t0.calls[0], ensure_ascii=False)[:200]}")
agree = sum(len(t.calls) == t.declared_calls for t in TRAJ)
print(f"    parser vs 'number of tool calls': {agree}/{len(TRAJ)} exact match")
calls_per   = [len(t.calls) for t in TRAJ]
depth_per   = [t.depth for t in TRAJ]
chars_per   = [sum(len(m["content"]) for m in t.messages) for t in TRAJ]
name_freq   = Counter(n for t in TRAJ for n in t.tool_names)
argkey_freq = defaultdict(Counter)
for t in TRAJ:
   for c in t.calls:
       args = c.get("arguments", {})
       if isinstance(args, dict):
           for k in args: argkey_freq[c.get("name", "?")][k] += 1
def q(xs, p):
   xs = sorted(xs); return xs[min(len(xs) - 1, int(p * len(xs)))]
print("\n[3] corpus statistics")
print(f"    tool calls / traj : mean {stats.mean(calls_per):.1f}  p50 {q(calls_per,.5)}  "
     f"p90 {q(calls_per,.9)}  max {max(calls_per)}")
print(f"    messages / traj   : mean {stats.mean(depth_per):.1f}  p90 {q(depth_per,.9)}  max {max(depth_per)}")
print(f"    chars / traj      : mean {stats.mean(chars_per):,.0f}  p90 {q(chars_per,.9):,}")
print(f"    tool distribution : {dict(name_freq)}")
for k, v in argkey_freq.items():
   print(f"      {k:<24} arg keys -> {dict(v.most_common(6))}")
tot = sum(chars_per); top = sum(sorted(chars_per)[-max(1, len(chars_per)//10):])
print(f"    top-10% longest trajectories hold {100*top/tot:.1f}% of all characters")
fig, ax = plt.subplots(1, 3, figsize=(15, 3.6))
ax[0].hist(calls_per, bins=40); ax[0].set_yscale("log"); ax[0].set_title("tool calls / trajectory")
ax[1].hist(depth_per, bins=40); ax[1].set_yscale("log"); ax[1].set_title("messages / trajectory")
ax[2].bar(list(name_freq), list(name_freq.values())); ax[2].set_title("tool usage"); ax[2].tick_params(axis="x", rotation=20)
plt.tight_layout(); plt.show()
代码 · 101 行
TOOLS_BLOCK_RE = re.compile(r"<tools>\s*(.*?)\s*</tools>", re.S)
THINK_RE       = re.compile(r"<think>(.*?)</think>", re.S)
TOOL_RESP_RE   = re.compile(r"<tool_response>\s*(.*?)\s*</tool_response>", re.S)
TOOLS_HDR_RE   = re.compile(r"\n\n# Tools\n\n")
def iter_json_objects(text: str, limit: int = 1):
   """Nesting-safe JSON scanner. Regex like r'\\{.*?\\}' breaks on nested
   `arguments` objects, which every real tool call has."""
   dec, i, n, out = json.JSONDecoder(), 0, len(text), []
   while i < n and len(out) < limit:
       while i < n and text[i] not in "{[":
           i += 1
       if i >= n:
           break
       try:
           obj, end = dec.raw_decode(text, i)
       except json.JSONDecodeError:
           i += 1
           continue
       out.append(obj); i = end
   return out
def parse_tool_calls(content: str) -> List[Dict[str, Any]]:
   calls = []
   for m in re.finditer(r"<tool_call>", content):
       got = iter_json_objects(content[m.end():], limit=1)
       if got:
           calls.append(got[0])
   return calls
@dataclass
class Trajectory:
   question: str
   answer: str
   declared_calls: int
   messages: List[Dict[str, str]]
   system_core: str = ""
   tools: List[Dict[str, Any]] = field(default_factory=list)
   tools_suffix: str = ""
   calls: List[Dict[str, Any]] = field(default_factory=list)
   n_observations: int = 0
   n_think: int = 0
   @property
   def tool_names(self):  return [c.get("name", "?") for c in self.calls]
   @property
   def depth(self):       return len(self.messages)
def parse_row(row: Dict[str, Any]) -> Trajectory:
   msgs = [{"role": m["role"], "content": m["content"]} for m in row["trajectory"]]
   t = Trajectory(row["question"], row["answer"], row["number of tool calls"], msgs)
   if msgs and msgs[0]["role"] == "system":
       sysmsg = msgs[0]["content"]
       split = TOOLS_HDR_RE.search(sysmsg)
       if split:
           t.system_core   = sysmsg[:split.start()]
           t.tools_suffix  = sysmsg[split.start():]
       else:
           t.system_core = sysmsg
       blk = TOOLS_BLOCK_RE.search(sysmsg)
       if blk:
           t.tools = iter_json_objects(blk.group(1), limit=64)
   for m in msgs:
       if m["role"] == "assistant":
           t.calls += parse_tool_calls(m["content"])
           t.n_think += len(THINK_RE.findall(m["content"]))
       else:
           t.n_observations += len(TOOL_RESP_RE.findall(m["content"]))
   return t
TRAJ = [parse_row(r) for r in RAW]
t0 = TRAJ[0]
print(f"\n[2] parsed {len(TRAJ)} trajectories")
print(f"    tool schemas found : {[fn.get('function', fn).get('name') for fn in t0.tools]}")
print(f"    parsed calls       : {len(t0.calls)}  (declared {t0.declared_calls})")
print(f"    observations       : {t0.n_observations}   think blocks: {t0.n_think}")
if t0.calls:
   print(f"    sample call        : {json.dumps(t0.calls[0], ensure_ascii=False)[:200]}")
agree = sum(len(t.calls) == t.declared_calls for t in TRAJ)
print(f"    parser vs 'number of tool calls': {agree}/{len(TRAJ)} exact match")
calls_per   = [len(t.calls) for t in TRAJ]
depth_per   = [t.depth for t in TRAJ]
chars_per   = [sum(len(m["content"]) for m in t.messages) for t in TRAJ]
name_freq   = Counter(n for t in TRAJ for n in t.tool_names)
argkey_freq = defaultdict(Counter)
for t in TRAJ:
   for c in t.calls:
       args = c.get("arguments", {})
       if isinstance(args, dict):
           for k in args: argkey_freq[c.get("name", "?")][k] += 1
def q(xs, p):
   xs = sorted(xs); return xs[min(len(xs) - 1, int(p * len(xs)))]
print("\n[3] corpus statistics")
print(f"    tool calls / traj : mean {stats.mean(calls_per):.1f}  p50 {q(calls_per,.5)}  "
     f"p90 {q(calls_per,.9)}  max {max(calls_per)}")
print(f"    messages / traj   : mean {stats.mean(depth_per):.1f}  p90 {q(depth_per,.9)}  max {max(depth_per)}")
print(f"    chars / traj      : mean {stats.mean(chars_per):,.0f}  p90 {q(chars_per,.9):,}")
print(f"    tool distribution : {dict(name_freq)}")
for k, v in argkey_freq.items():
   print(f"      {k:<24} arg keys -> {dict(v.most_common(6))}")
tot = sum(chars_per); top = sum(sorted(chars_per)[-max(1, len(chars_per)//10):])
print(f"    top-10% longest trajectories hold {100*top/tot:.1f}% of all characters")
fig, ax = plt.subplots(1, 3, figsize=(15, 3.6))
ax[0].hist(calls_per, bins=40); ax[0].set_yscale("log"); ax[0].set_title("tool calls / trajectory")
ax[1].hist(depth_per, bins=40); ax[1].set_yscale("log"); ax[1].set_title("messages / trajectory")
ax[2].bar(list(name_freq), list(name_freq.values())); ax[2].set_title("tool usage"); ax[2].tick_params(axis="x", rotation=20)
plt.tight_layout(); plt.show()

更进一步：量化金融体系

看懂新闻只是起点——沿量化金融路径，把它变成能交付的工程能力

进入量化体系 →
话题#工具调用#微调#Qwen3#LoRA#教程

相似阅读

另一事件，读法相近