In this tutorial, we implement an end-to-end supervised fine-tuning pipeline for the XYZ-Aquila-SFT dataset, leveraging 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.
Prerequisites and Setup
Before diving in, ensure your environment meets the following requirements:
- Python 3.9+: We recommend using a virtual environment or a cloud-based notebook (e.g., Google Colab) for reproducibility.
- Hardware: A GPU with at least 16GB of VRAM is recommended for fine-tuning. For CPU-only usage, reduce
MAXSTEPSandNSTREAMto avoid long training times. - Dependencies: The pipeline installs the required libraries automatically via
pip(see code block below).
Configuration and Installation
The pipeline is controlled by a central configuration dictionary, allowing you to adjust hyperparameters, paths, and flags without modifying the core logic. The following code sets up the configuration, installs dependencies, and initializes the environment.
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 dataset...")
Streaming and Inspecting the Dataset
We use Hugging Face's datasets library to stream the XYZ-Aquila-SFT dataset, which contains thousands of multi-turn conversations where an assistant interacts with external tools. Streaming allows us to iterate through the data without downloading the entire corpus at once, making the pipeline memory-efficient.
# Load the dataset in streaming mode
ds = load_dataset(CFG["REPO"], split="train", streaming=True)
stream_iter = iter(ds.take(CFG["N_STREAM"]))
# Inspect the first few samples to understand the structure
for i, sample in enumerate(stream_iter):
if i >= 3:
break
print(f"\nSample {i}:")
print(sample)
Each sample typically contains a messages list with roles such as system, user, assistant, and tool. Embedded in the assistant messages are tool call requests and observations, which we will extract and structure for training.
Parsing Tool-Use Trajectories
The core of fine-tuning a tool-calling model lies in accurately parsing the conversation history. We extract structured tool calls from the raw messages, preserving the context of each call, the arguments passed, and the resulting observations. This step ensures the model learns not only to generate tool calls but also to interpret their outcomes.
# Parse a message list into a trajectory of tool calls and observations
def parse_trajectory(messages: List[Dict]) -> List[Dict]:
trajectory = []
for msg in messages:
if msg["role"] == "assistant":
# Extract tool calls if present (e.g., in function_call format)
if "function_call" in msg:
trajectory.append({
"type": "tool_call",
"name": msg["function_call"]["name"],
"arguments": msg["function_call"]["arguments"]
})
elif msg["role"] == "tool":
trajectory.append({
"type": "observation",
"content": msg["content"]
})
else:
trajectory.append({"type": "message", "role": msg["role"], "content": msg["content"]})
return trajectory
# Apply to the streamed samples
samples = list(stream_iter)
for sample in samples[:2]:
traj = parse_trajectory(sample["messages"])
print(json.dumps(traj, indent=2))
Analyzing Corpus Characteristics
To ensure the model is well-suited for tool interactions, we analyze the corpus in terms of tool-call frequency, argument complexity, and the balance between reasoning and tool use. This informs preprocessing decisions, such as whether to truncate or pad sequences and how to weight loss masking.
# Count tool call types and measure argument lengths
call_types = Counter()
arg_lengths = []
for sample in samples:
traj = parse_trajectory(sample["messages"])
for item in traj:
if item["type"] == "tool_call":
call_types[item["name"]] += 1
arg_lengths.append(len(item["arguments"]))
print("Tool call frequency:", call_types.most_common(10))
print(f"Argument length - mean: {stats.mean(arg_lengths):.1f}, median: {stats.median(arg_lengths)}")
Converting Tool Schemas and Rendering ChatML
Qwen3 models expect a specific ChatML format for fine-tuning. We convert the parsed trajectories into a single text string, complete with special tokens for system, user, assistant, and tool messages. Critically, we apply loss masking to ensure the model only learns to predict assistant responses, not user inputs or system prompts.
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained(CFG["MODEL_ID"], trust_remote_code=True)
# Define ChatML conversion
def render_chatml(messages: List[Dict]) -> str:
chat_str = ""
for msg in messages:
role = msg["role"]
content = msg["content"]
if role == "system":
chat_str += f"<|im_start|>system\n{content}<|im_end|>\n"
elif role == "user":
chat_str += f"<|im_start|>user\n{content}<|im_end|>\n"
elif role == "assistant":
chat_str += f"<|im_start|>assistant\n{content}<|im_end|>\n"
elif role == "tool":
chat_str += f"<|im_start|>tool\n{content}<|im_end|>\n"
return chat_str
# For a single sample
sample = samples[0]
chat_text = render_chatml(sample["messages"])
print(chat_text[:500])
Preparing Custom Dataset and Collator
We implement a custom PyTorch Dataset and a dynamic collator that tokenizes the ChatML strings, applies loss masking to non-assistant tokens, and truncates sequences to MAXSEQLEN. This ensures consistency during training and evaluation.
@dataclass
class ToolDataset(torch.utils.data.Dataset):
samples: List[Dict]
tokenizer: Any
max_seq_len: int = 2048
def __len__(self):
return len(self.samples)
def __getitem__(self, idx):
sample = self.samples[idx]
chat_text = render_chatml(sample["messages"])
enc = self.tokenizer(chat_text, truncation=True, max_length=self.max_seq_len)
return {"input_ids": enc["input_ids"], "attention_mask": enc["attention_mask"]}
def collate_fn(batch):
input_ids = [item["input_ids"] for item in batch]
attention_mask = [item["attention_mask"] for item in batch]
# Pad sequences to the longest in the batch
padded = tokenizer.pad({"input_ids": input_ids, "attention_mask": attention_mask}, return_tensors="pt")
# Create labels with loss masking (we'll refine this later)
labels = padded["input_ids"].clone()
return {"input_ids": padded["input_ids"], "attention_mask": padded["attention_mask"], "labels": labels}
dataset = ToolDataset(samples, tokenizer, CFG["MAX_SEQ_LEN"])
dataloader = torch.utils.data.DataLoader(dataset, batch_size=2, collate_fn=collate_fn, shuffle=True)
Fine-Tuning with LoRA
We fine-tune Qwen3-0.6B using Low-Rank Adaptation (LoRA) to efficiently adapt the model to tool-calling tasks without updating all parameters. This reduces memory usage and training time while maintaining performance.
from peft import LoraConfig, get_peft_model, TaskType
lora_config = LoraConfig(
task_type=TaskType.CAUSAL_LM,
r=CFG["LORA_R"],
lora_alpha=CFG["LORA_R"] * 2,
target_modules=["q_proj", "v_proj"],
lora_dropout=0.1,
)
model = AutoModelForCausalLM.from_pretrained(CFG["MODEL_ID"], torch_dtype=torch.float16 if BF16 else torch.float32)
model = get_peft_model(model, lora_config)
model.to(DEV)
optimizer = torch.optim.AdamW(model.parameters(), lr=CFG["LR"])
scheduler = get_cosine_schedule_with_warmup(optimizer, num_warmup_steps=10, num_training_steps=CFG["MAX_STEPS"])
# Training loop
model.train()
for step, batch in enumerate(dataloader):
if step >= CFG["MAX_STEPS"]:
break
batch = {k: v.to(DEV) for k, v in batch.items()}
outputs = model(**batch)
loss = outputs.loss / CFG["GRAD_ACCUM"]
loss.backward()
if (step + 1) % CFG["GRAD_ACCUM"] == 0:
optimizer.step()
scheduler.step()
optimizer.zero_grad()
if step % 10 == 0:
print(f"Step {step}: loss = {loss.item():.4f}")
Evaluating Tool-Call Prediction
After training, we evaluate the model's ability to generate accurate tool calls. We use a set of held-out probes, comparing the predicted tool name and arguments against ground truth using a simple metric (e.g., exact match or F1 score).
from transformers import pipeline
def evaluate(model, tokenizer, probes):
model.eval()
correct = 0
total = 0
for probe in probes:
prompt = probe["prompt"]
expected_call = probe["expected_call"]
inputs = tokenizer(prompt, return_tensors="pt").to(DEV)
with torch.no_grad():
outputs = model.generate(**inputs, max_new_tokens=50, pad_token_id=tokenizer.eos_token_id)
generated = tokenizer.decode(outputs[0], skip_special_tokens=True)
# Simple check: does the generated text contain the expected function call?
if expected_call in generated:
correct += 1
total += 1
return correct / total
# Create a few probes from test samples
probes = [
{"prompt": render_chatml(sample["messages"][:-1]), "expected_call": "get_weather"} for sample in samples[:CFG["N_EVAL_PROBES"]]
]
score = evaluate(model, tokenizer, probes)
print(f"Evaluation accuracy: {score:.2f}")
Exporting Results
Finally, we save the trained model, the transformed dataset, and the corpus statistics for future use or sharing.
# Save model
model.save_pretrained(f"{CFG['OUT_DIR']}/lora_model")
tokenizer.save_pretrained(f"{CFG['OUT_DIR']}/lora_model")
# Export transformed dataset
with open(f"{CFG['OUT_DIR']}/transformed_data.jsonl", "w") as f:
for sample in samples:
f.write(json.dumps(sample) + "\n")
# Save statistics
with open(f"{CFG['OUT_DIR']}/stats.json", "w") as f:
json.dump({"tool_call_frequency": dict(call_types), "arg_lengths": arg_lengths}, f, indent=2)
print("Pipeline completed. All outputs saved to", CFG["OUT_DIR"])
Conclusion and Next Steps
In this tutorial, you built a complete fine-tuning pipeline for tool-calling language models using XYZ-Aquila-SFT and Qwen3. You learned how to parse trajectories, convert to ChatML, apply loss masking, and fine-tune with LoRA. With 2026's evolving landscape, we recommend exploring the following to further enhance your models:
- Multi-turn reasoning: Extend the training to emphasize reasoning steps before tool calls, improving interpretability.
- Tool selection: Incorporate a selection mechanism to handle large tool catalogs dynamically.
- Quantization: Use QLoRA (or similar) to reduce memory footprint further, enabling larger base models.
- Evaluation suites: Implement more robust metrics like tool-call accuracy and argument validity using online platforms (e.g., Berkeley Function Calling Leaderboard).
All code is available in the associated repository; feel free to adapt the configuration to your specific needs. Happy fine-tuning!
via MarkTechPost