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

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.

Copy CodeCopiedUse a different Browser

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.

Copy CodeCopiedUse a different Browser

TOOLS_BLOCK_RE = re.compile(r"\s*(.*?)\s*", re.S) THINK_RE = re.compile(r"(.*?)", re.S) TOOL_RESP_RE = re.compile(r"\s*(.*?)\s*", 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: 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"", 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: {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()

We define nesting-safe utilities for extracting JSON tool calls, reasoning blocks, observations, and embedded tool schemas from each conversation. We convert every raw dataset row into a structured trajectory object and verify that the parsed tool-call counts match the values declared by the dataset. We then calculate corpus-level statistics and visualize the distributions of tool calls, message depth, trajectory size, and tool usage frequency.

Copy CodeCopiedUse a different Browser

QWEN3_TOOLS_TMPL = ( "You are provided with function signatures within XML tags:\n\n" "{lines}\n\n\nFor each function call, return a json object with function name " "and arguments within XML tags:\n\n" '{{"name": , "arguments": }}\n' ) def extract_tools(t: Trajectory) -> Dict[str, Any]: """message-embedded schemas -> {'messages': [...], 'tools': [...]}""" msgs = [dict(m) for m in t.messages] if msgs and msgs[0]["role"] == "system": msgs[0]["content"] = t.system_core return {"messages": msgs, "tools": t.tools, "question": t.question, "answer": t.answer} def render_tools(rec: Dict[str, Any]) -> List[Dict[str, str]]: """inverse: structured tools -> schemas re-embedded in the system message""" msgs = [dict(m) for m in rec["messages"]] if rec["tools"] and msgs and msgs[0]["role"] == "system": lines = "\n".join(json.dumps(x, ensure_ascii=False) for x in rec["tools"]) msgs[0]["content"] = msgs[0]["content"] + QWEN3_TOOLS_TMPL.format(lines=lines) return msgs _rt = render_tools(extract_tools(t0)) exact = _rt[0]["content"] == t0.messages[0]["content"] print(f"\n[4] extract->render byte-exact: {exact}") if not exact: print(" template drift detected -> using verbatim tools_suffix for render()") a, b = t0.messages[0]["content"], _rt[0]["content"] i = next((i for i in range(min(len(a), len(b))) if a[i] != b[i]), min(len(a), len(b))) print(f" first divergence @{i}: {a[i:i+70]!r} vs {b[i:i+70]!r}") tok = AutoTokenizer.from_pretrained(CFG["MODEL_ID"]) if tok.pad_token is None: tok.pad_token = tok.eos_token IM_START, IM_END, NL = "", "", "\n" def render_and_mask(t: Trajectory, max_len: int, policy: str): """Manual ChatML so we control masking token-exactly. WHY NOT apply_chat_template(): Qwen3's template deletes ... from every assistant turn except the last. On this dataset that silently destroys most of the reasoning supervision you are paying to train on. """ ids, labels = [], [] for m in t.messages: head = tok(f"{IM_START}{m['role']}{NL}", add_special_tokens=False).input_ids body = tok(m["content"], add_special_tokens=False).input_ids tail = tok(f"{IM_END}{NL}", add_special_tokens=False).input_ids seg = head + body + tail if m["role"] == "assistant": lab = [-100] * len(head) + body + tail else: lab = [-100] * len(seg) ids += seg; labels += lab if len(ids) > max_len: if policy == "drop": return None ids, labels = ids[:max_len], labels[:max_len] if all(l == -100 for l in labels): return None return {"input_ids": ids, "labels": labels} _probe = [{"role": "system", "content": "S"}, {"role": "user", "content": "U"}, {"role": "assistant", "content": "A"}] _mine = "".join(f"{IM_START}{m['role']}{NL}{m['content']}{IM_END}{NL}" for m in _probe) _theirs = tok.apply_chat_template(_probe, tokenize=False, add_generation_prompt=False) print(f"\n[5] manual ChatML == chat_template on tool-free probe: {_mine == _theirs}") if _mine != _theirs: print(f" mine : {_mine!r}\n theirs: {_theirs!r} (informational only)") ENC = [e for e in (render_and_mask(t, CFG["MAX_SEQ_LEN"], CFG["LENGTH_POLICY"]) for t in TRAJ) if e] sup = [sum(1 for x in e["labels"] if x != -100) / len(e["labels"]) for e in ENC] print(f" encoded {len(ENC)}/{len(TRAJ)} examples") print(f" supervised-token ratio: mean {stats.mean(sup):.3f} p10 {q(sup,.1):.3f} p90 {q(sup,.9):.3f}") over = sum(1 for t in TRAJ if sum(len(tok(m['content'], add_special_tokens=False).input_ids) for m in t.messages[:3]) > CFG["MAX_SEQ_LEN"]) print(f" trajectories whose first 3 msgs alone exceed MAX_SEQ_LEN: {over}") SPLIT = len(ENC) - min(CFG["N_EVAL"], len(ENC)//5) TRAIN_ENC, EVAL_TRAJ = ENC[:SPLIT], TRAJ[SPLIT:] class SFTSet(torch.utils.data.Dataset): def init(self, rows): self.rows = rows def len(self): return len(self.rows) def getitem(self, i): return self.rows[i] def collate(batch): L = max(len(b["input_ids"]) for b in batch) pad = tok.pad_token_id return { "input_ids": torch.tensor([b["input_ids"] + [pad]*(L-len(b["input_ids"])) for b in batch]), "labels": torch.tensor([b["labels"] + [-100]*(L-len(b["labels"])) for b in batch]), "attention_mask": torch.tensor([[1]*len(b["input_ids"]) + [0]*(L-len(b["input_ids"])) for b in batch]), } loader = torch.utils.data.DataLoader(SFTSet(TRAIN_ENC), batch_size=1, shuffle=True, collate_fn=collate) print(f"\n[6] train={len(TRAIN_ENC)} eval_trajectories={len(EVAL_TRAJ)}")

We extract embedded tool definitions into a structured format and reconstruct them to test whether the conversion preserves the original system message. We manually render each tra

[truncated for AI cost control]