Official docs: Prepare your dataset
Good fine-tuning starts with good data. This lesson covers the JSONL format, quality requirements, validation, and augmentation techniques.
Azure OpenAI expects training data as JSONL (JSON Lines) — one JSON object per line using the chat completions message format.
Each line contains a conversation with a system prompt, a user message, and the ideal assistant response.
{"messages": [{"role": "system", "content": "You are a customer service agent for Contoso Electronics."}, {"role": "user", "content": "What is your return policy?"}, {"role": "assistant", "content": "Contoso Electronics offers a 30-day return policy for all items in original packaging. Opened electronics can be returned within 14 days with a 15% restocking fee."}]}
{"messages": [{"role": "system", "content": "You are a customer service agent for Contoso Electronics."}, {"role": "user", "content": "Do you ship internationally?"}, {"role": "assistant", "content": "Yes, Contoso ships to over 40 countries. Standard international shipping takes 7-14 business days. Express shipping (3-5 days) is available for an additional fee."}]}
For conversational fine-tuning, include multiple turns in a single example:
{"messages": [{"role": "system", "content": "You are a technical support agent."}, {"role": "user", "content": "My laptop won't turn on."}, {"role": "assistant", "content": "I'm sorry to hear that. Let's troubleshoot. Is the charging LED lit when you plug in the power adapter?"}, {"role": "user", "content": "No, there's no light at all."}, {"role": "assistant", "content": "That suggests a power delivery issue. Please try these steps:\n1. Unplug the adapter and try a different wall outlet.\n2. If you have a spare adapter, try that instead.\n3. Hold the power button for 15 seconds, then release and try powering on.\n\nDid any of those steps help?"}]}
weight to Control TrainingYou can add a weight field (0 or 1) to assistant messages to exclude certain responses from the loss calculation. This is useful in multi-turn conversations where only the final response matters:
{"messages": [{"role": "system", "content": "You are a math tutor."}, {"role": "user", "content": "What is 2+2?"}, {"role": "assistant", "content": "Can you try solving it first?", "weight": 0}, {"role": "user", "content": "I think it's 4."}, {"role": "assistant", "content": "Correct! 2+2 = 4. Great job!", "weight": 1}]}
| # | Requirement | Details |
|---|---|---|
| 1 | Minimum 10 examples | 10 is the minimum; 50-100+ is recommended for meaningful improvement |
| 2 | Consistent format | Every example must follow the same message structure |
| 3 | Accurate assistant responses | The model learns to mimic these — errors will be replicated |
| 4 | Diverse inputs | Cover edge cases, different phrasings, varying complexity |
| 5 | Appropriate length | Assistant responses should match your desired output length |
| 6 | System prompt consistency | Use the same system prompt you will use at inference time |
| 7 | No PII or sensitive data | Training data is stored on Azure; follow your org’s data policies |
| 8 | Valid JSON per line | Even one malformed line will cause the upload to fail |
| 9 | UTF-8 encoding | Save files with UTF-8 encoding (no BOM) |
| 10 | No duplicate examples | Duplicates waste training budget and can bias the model |
Tip: Quality > Quantity. 50 excellent examples often outperform 500 mediocre ones.
Splitting your data into training and validation sets lets you detect overfitting. Azure OpenAI also accepts a separate validation_file parameter.
"""Split a JSONL file into training and validation sets."""
import json
import random
from pathlib import Path
def split_jsonl(
input_path: str,
train_path: str = "train.jsonl",
val_path: str = "val.jsonl",
val_ratio: float = 0.2,
seed: int = 42,
) -> dict:
"""Split JSONL into train/val sets.
Args:
input_path: Path to the source JSONL file.
train_path: Output path for training data.
val_path: Output path for validation data.
val_ratio: Fraction of data reserved for validation (0.0-1.0).
seed: Random seed for reproducibility.
Returns:
Dict with counts for train, val, and total examples.
"""
with open(input_path, "r", encoding="utf-8") as f:
examples = [json.loads(line) for line in f if line.strip()]
random.seed(seed)
random.shuffle(examples)
split_idx = int(len(examples) * (1 - val_ratio))
train_data = examples[:split_idx]
val_data = examples[split_idx:]
for path, data in [(train_path, train_data), (val_path, val_data)]:
with open(path, "w", encoding="utf-8") as f:
for example in data:
f.write(json.dumps(example, ensure_ascii=False) + "\n")
result = {"train": len(train_data), "val": len(val_data), "total": len(examples)}
print(f"Split complete: {result['train']} train / {result['val']} val "
f"(total: {result['total']})")
return result
if __name__ == "__main__":
split_jsonl("training_data.jsonl")
Run this before uploading to catch issues early:
"""Validate a JSONL fine-tuning dataset: format, token counts, cost estimate."""
import json
import sys
from collections import defaultdict
from pathlib import Path
try:
import tiktoken
except ImportError:
sys.exit("Install tiktoken: pip install tiktoken")
# ---------- configuration ----------
MODEL = "gpt-4o-mini-2024-07-18"
ENCODING_NAME = "o200k_base" # encoding for gpt-4o family
TRAINING_COST_PER_1K = 0.008 # USD — adjust per model/pricing page
MAX_TOKENS = 128_000 # context window for gpt-4o-mini
MIN_EXAMPLES = 10
RECOMMENDED_EXAMPLES = 50
VALID_ROLES = {"system", "user", "assistant"}
def validate_jsonl(filepath: str, encoding_name: str = ENCODING_NAME) -> None:
"""Validate a JSONL dataset and print a detailed report."""
enc = tiktoken.get_encoding(encoding_name)
errors: list[str] = []
warnings: list[str] = []
token_counts: list[int] = []
role_counts: dict[str, int] = defaultdict(int)
seen_hashes: set[int] = set()
n_duplicates = 0
path = Path(filepath)
if not path.exists():
sys.exit(f"File not found: {filepath}")
with open(path, "r", encoding="utf-8") as f:
lines = f.readlines()
if len(lines) == 0:
sys.exit("File is empty.")
for i, line in enumerate(lines, start=1):
line = line.strip()
if not line:
continue
# --- JSON parse ---
try:
obj = json.loads(line)
except json.JSONDecodeError as e:
errors.append(f"Line {i}: Invalid JSON -- {e}")
continue
# --- top-level structure ---
if "messages" not in obj:
errors.append(f"Line {i}: Missing 'messages' key.")
continue
messages = obj["messages"]
if not isinstance(messages, list) or len(messages) < 2:
errors.append(f"Line {i}: 'messages' must be a list with >= 2 items.")
continue
# --- role / content checks ---
has_assistant = False
line_tokens = 0
for j, msg in enumerate(messages):
role = msg.get("role")
content = msg.get("content", "")
if role not in VALID_ROLES:
errors.append(f"Line {i}, msg {j}: Invalid role '{role}'.")
role_counts[role] += 1
if not content and role != "system":
warnings.append(f"Line {i}, msg {j}: Empty content for role '{role}'.")
if role == "assistant":
has_assistant = True
line_tokens += len(enc.encode(str(content)))
if not has_assistant:
errors.append(f"Line {i}: No assistant message found.")
# --- token limit ---
if line_tokens > MAX_TOKENS:
errors.append(
f"Line {i}: {line_tokens:,} tokens exceeds {MAX_TOKENS:,} limit."
)
token_counts.append(line_tokens)
# --- duplicate detection ---
line_hash = hash(line)
if line_hash in seen_hashes:
n_duplicates += 1
warnings.append(f"Line {i}: Possible duplicate.")
seen_hashes.add(line_hash)
# ---------- report ----------
n_examples = len(token_counts)
total_tokens = sum(token_counts)
print("=" * 60)
print(f" Dataset Validation Report: {path.name}")
print("=" * 60)
print(f" Examples : {n_examples}")
print(f" Errors : {len(errors)}")
print(f" Warnings : {len(warnings)}")
print(f" Duplicates : {n_duplicates}")
print("-" * 60)
print(f" Total tokens : {total_tokens:,}")
if token_counts:
print(f" Avg tokens/ex : {total_tokens // n_examples:,}")
print(f" Min tokens/ex : {min(token_counts):,}")
print(f" Max tokens/ex : {max(token_counts):,}")
print("-" * 60)
print(" Role distribution:")
for role in ["system", "user", "assistant"]:
print(f" {role:12s} : {role_counts.get(role, 0)}")
print("-" * 60)
# --- cost estimate (multiple epoch counts) ---
for epochs in [1, 2, 3, 5]:
cost = (total_tokens * epochs / 1000) * TRAINING_COST_PER_1K
print(f" Est. training cost ({epochs} epochs): ${cost:,.2f}")
print("-" * 60)
if n_examples < MIN_EXAMPLES:
errors.append(
f"Only {n_examples} examples; minimum is {MIN_EXAMPLES}."
)
if n_examples < RECOMMENDED_EXAMPLES:
warnings.append(
f"Only {n_examples} examples; {RECOMMENDED_EXAMPLES}+ recommended."
)
if errors:
print("\nERRORS (must fix):")
for e in errors:
print(f" - {e}")
if warnings:
print("\nWARNINGS (should review):")
for w in warnings:
print(f" - {w}")
if not errors:
print("\nDataset is valid and ready for upload.")
sys.exit(1 if errors else 0)
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python validate_dataset.py <path-to-jsonl>")
sys.exit(1)
validate_jsonl(sys.argv[1])
When you have limited examples, use the base model to generate paraphrases:
"""Augment training data by paraphrasing user messages."""
import json
from openai import AzureOpenAI
client = AzureOpenAI(
azure_endpoint="https://<resource>.openai.azure.com/",
api_key="<key>",
api_version="2025-03-01-preview",
)
def paraphrase_user_message(message: str, n_variants: int = 3) -> list[str]:
"""Generate paraphrased versions of a user message."""
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": (
"Generate exactly {n} paraphrased versions of the user message. "
"Keep the same meaning and intent. Return one paraphrase per line, "
"no numbering or bullets."
).format(n=n_variants),
},
{"role": "user", "content": message},
],
temperature=0.9,
)
return [
line.strip()
for line in response.choices[0].message.content.strip().split("\n")
if line.strip()
]
def augment_dataset(input_path: str, output_path: str, n_variants: int = 3) -> None:
"""Read a JSONL file, paraphrase user messages, write augmented dataset."""
augmented = []
with open(input_path, "r", encoding="utf-8") as f:
examples = [json.loads(line) for line in f if line.strip()]
for example in examples:
augmented.append(example) # keep the original
messages = example["messages"]
for i, msg in enumerate(messages):
if msg["role"] != "user":
continue
paraphrases = paraphrase_user_message(msg["content"], n_variants)
for para in paraphrases:
new_messages = [m.copy() for m in messages]
new_messages[i] = {"role": "user", "content": para}
augmented.append({"messages": new_messages})
with open(output_path, "w", encoding="utf-8") as f:
for ex in augmented:
f.write(json.dumps(ex, ensure_ascii=False) + "\n")
print(f"Augmented {len(examples)} -> {len(augmented)} examples -> {output_path}")
if __name__ == "__main__":
augment_dataset("training_data.jsonl", "training_data_augmented.jsonl")
If you have internal docs, you can generate Q&A pairs suitable for fine-tuning:
"""Generate fine-tuning examples from source documents."""
import json
from pathlib import Path
from openai import AzureOpenAI
client = AzureOpenAI(
azure_endpoint="https://<resource>.openai.azure.com/",
api_key="<key>",
api_version="2025-03-01-preview",
)
SYSTEM_PROMPT = "You are a helpful assistant for Contoso product documentation."
GENERATION_PROMPT = """\
Given the following document excerpt, generate {n} realistic question-answer pairs.
Each pair should be a question a user might ask and the ideal assistant answer
based ONLY on the document content.
Return valid JSON -- an array of objects with "question" and "answer" keys.
---
{document}
---
"""
def generate_from_document(
doc_text: str, n_pairs: int = 5, model: str = "gpt-4o"
) -> list[dict]:
"""Generate fine-tuning examples from a document chunk."""
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "You generate training data in JSON format."},
{
"role": "user",
"content": GENERATION_PROMPT.format(n=n_pairs, document=doc_text),
},
],
temperature=0.7,
response_format={"type": "json_object"},
)
content = response.choices[0].message.content
data = json.loads(content)
# Handle both {"pairs": [...]} and [...] formats
pairs = data if isinstance(data, list) else data.get("pairs", data.get("questions", []))
return [
{
"messages": [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": p["question"]},
{"role": "assistant", "content": p["answer"]},
]
}
for p in pairs
]
def process_documents(
docs_dir: str, output_path: str, n_pairs: int = 5, chunk_size: int = 2000
) -> None:
"""Process all .txt/.md files in a directory into fine-tuning JSONL."""
all_examples = []
for file_path in Path(docs_dir).glob("**/*"):
if file_path.suffix not in (".txt", ".md"):
continue
text = file_path.read_text(encoding="utf-8")
# Simple chunking by character count
chunks = [text[i : i + chunk_size] for i in range(0, len(text), chunk_size)]
for chunk in chunks:
if len(chunk.strip()) < 100:
continue
examples = generate_from_document(chunk, n_pairs=n_pairs)
all_examples.extend(examples)
print(f" {file_path.name}: generated {len(examples)} examples from chunk")
with open(output_path, "w", encoding="utf-8") as f:
for ex in all_examples:
f.write(json.dumps(ex, ensure_ascii=False) + "\n")
print(f"\nTotal: {len(all_examples)} examples -> {output_path}")
if __name__ == "__main__":
process_documents("./source_docs", "generated_training_data.jsonl")
Warning: Always review generated training data manually. LLM-generated answers can contain hallucinations that, if trained on, will be replicated by your fine-tuned model.
| Step | Action | Tool |
|---|---|---|
| 1 | Collect or generate examples | Manual / generate_from_document() |
| 2 | Format as JSONL | Text editor / script |
| 3 | Validate format and tokens | validate_jsonl() |
| 4 | Remove duplicates and PII | Script / manual review |
| 5 | Augment if needed | augment_dataset() |
| 6 | Split train / validation | split_jsonl() |
| 7 | Upload to Azure OpenAI | SDK client.files.create() |
Next –> Lesson 2: Fine-Tuning Workflow