Build an LLM Training Data Pipeline
Generate a curated SFT dataset from Python code using Groq's LLM API.
Introduction
30 Second Summary
Every large language model you interact with was trained on carefully curated data. Behind the scenes, AI teams build pipelines that manufacture this training data at scale, transforming raw code and text into structured datasets that teach models how to respond.
In this project, you will build a Python pipeline that takes code snippets as input, uses an LLM to generate instruction-response pairs, and filters them through quality gates to produce a clean dataset ready for fine-tuning.
What You'll Build
A complete training data pipeline that distills Python code snippets into a curated JSONL dataset suitable for supervised fine-tuning of LLMs.
By the end of this project, you'll have:
- A working data generation engine that calls Groq's Llama 3.3 70B model to produce diverse instruction-response pairs from code snippets.
- A multi-stage quality filter that deduplicates, validates schema, and enforces length constraints on generated data.
- A clean JSONL dataset formatted for HuggingFace TRL SFTTrainer, ready for supervised fine-tuning.
- Secret Mission: Add a diversity scorer that uses TF-IDF cosine similarity to catch semantically similar pairs that slip past exact deduplication.
Are there any prerequisites?
You should be comfortable with Python basics like functions, file I/O, and working with JSON. Familiarity with making API calls is also helpful.
Prior projects in this series (Code Verification Agent, Property-Based Testing) provide useful context but are not required.
Set Up Your Environment
Before building a data pipeline, you need a clean, isolated Python environment with the right dependencies installed.
In this step, you'll create a project folder, install the Groq SDK, and confirm your API key connects successfully.
In this step, get ready to:
- Create a project folder with a Python virtual environment.
- Install the groq package.
- Verify API connectivity with a test call.
Create a project folder and set up a virtual environment
A virtual environment gives your project its own isolated set of Python packages. This prevents conflicts with other projects on your machine.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir training-data-pipeline
cd training-data-pipeline- Confirm the folder exists by running:
lsYou should see the training-data-pipeline folder listed in your Desktop directory. You're now inside it.
- Create and activate your virtual environment using the instructions for your operating system:
macOS
python3 -m venv venv
source venv/bin/activateWindows
python -m venv venv
venv\Scripts\activateYou should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.
Don't see (venv) in your prompt?
Make sure you ran both lines (creation and activation). On Windows, if you see a permissions error, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell first, then try the activate command again.
help me activate my Python virtual environment
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type VS Code and press Enter to open it.
- In VS Code, click File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop and select the training-data-pipeline folder, then click Open.
Why use VS Code for this project?
VS Code gives you a file explorer and integrated terminal in one window. You'll write your pipeline code in the editor and run it from the terminal without switching applications.
Install the groq package and verify the installation
The groq package is the official Python SDK for calling Groq's LLM API. Your pipeline will use it to generate training data.
- Switch back to your terminal and install the package by running:
pip install groq==1.2.0What does this command do?
pip install groq==1.2.0 installs version 1.2.0 of the Groq SDK into your virtual environment. Pinning the version ensures everyone running this project gets the same behavior.
- Verify the installation succeeded by running:
pip show groqYou should see output showing Name: groq and Version: 1.2.0 near the top.
Seeing 'No matching distribution found'?
Make sure your virtual environment is active (you should see (venv) in your prompt). If you accidentally closed your terminal, navigate back to the project folder and re-activate with source venv/bin/activate (macOS) or venv\Scripts\activate (Windows).
help me install the groq Python package
Set the GROQ_API_KEY environment variable and verify connectivity
The Groq SDK reads your API key from an environment variable. Setting it in your terminal session keeps the key out of your source code.
- Set your API key as an environment variable using the command for your operating system:
macOS
export GROQ_API_KEY=your-api-key-hereReplace your-api-key-here with the actual API key from your Groq account.
Windows
set GROQ_API_KEY=your-api-key-hereReplace your-api-key-here with the actual API key from your Groq account.
Where do I find my Groq API key?
Visit the Groq Console and copy your existing key. If you completed the Code Verification Agent project, you already have one.
- Test your connection to Groq by running this command:
python -c "from groq import Groq; import os; client = Groq(api_key=os.environ.get('GROQ_API_KEY')); r = client.chat.completions.create(messages=[{'role': 'user', 'content': 'Say hello in one word'}], model='llama-3.3-70b-versatile'); print(r.choices[0].message.content)"What does this test do?
This one-liner imports the Groq SDK, reads your API key from the environment variable, and sends a simple prompt to the llama-3.3-70b-versatile model. If everything is configured correctly, you'll see a short greeting printed in your terminal.
You should see a response like Hello or Hi printed in your terminal. This confirms your API key works and the Groq SDK is installed correctly.
Seeing an authentication error?
Double-check that you pasted the full API key with no extra spaces. Run echo $GROQ_API_KEY (macOS) or echo %GROQ_API_KEY% (Windows) to confirm the variable is set correctly.
If you see a rate limit error, wait a minute and try again. The Groq free tier allows 30 requests per minute.
help me connect to the Groq API
Your environment is ready. Next up, you'll create the source code snippets that your pipeline will transform into training data.
Create Source Code Snippets
Your environment is wired up and ready to call Groq. Now you need something to feed into the pipeline.
Every training data pipeline starts with raw input material. In this step, you will create a curated set of Python code snippets that represent different coding patterns. These snippets become the "seed" that your pipeline will distill into structured training data.
In this step, get ready to:
- Create a source snippets file with 5 diverse Python code examples.
- Structure each snippet with metadata fields for traceability.
- Add a loader function that the pipeline can call to fetch snippets.
Create the source snippets file
The pipeline needs real code to generate training data from. You will create 5 snippets covering different Python patterns: a decorator, a class with error handling, a generator, a context manager, and a recursive algorithm.
- In VS Code, create a new file called source_snippets.py in your project folder.
- Paste the following code into source_snippets.py:
"""Source code snippets for training data generation."""
SNIPPETS = [
{
"id": "snippet_01",
"category": "decorator",
"code": '''def retry(max_attempts=3, delay=1):
"""Decorator that retries a function on exception."""
import time
import functools
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
attempts = 0
while attempts < max_attempts:
try:
return func(*args, **kwargs)
except Exception as e:
attempts += 1
if attempts == max_attempts:
raise
time.sleep(delay * attempts)
return wrapper
return decorator'''
},
{
"id": "snippet_02",
"category": "class_error_handling",
"code": '''class ConnectionPool:
"""A simple connection pool with health checking."""
def __init__(self, max_connections=10):
self._pool = []
self._max = max_connections
self._in_use = set()
def acquire(self):
if self._pool:
conn = self._pool.pop()
self._in_use.add(id(conn))
return conn
if len(self._in_use) >= self._max:
raise RuntimeError("Connection pool exhausted")
conn = self._create_connection()
self._in_use.add(id(conn))
return conn
def release(self, conn):
if id(conn) not in self._in_use:
raise ValueError("Connection not from this pool")
self._in_use.discard(id(conn))
self._pool.append(conn)
def _create_connection(self):
return {"status": "connected", "id": id(object())}'''
},
{
"id": "snippet_03",
"category": "generator",
"code": '''def chunked_reader(file_path, chunk_size=1024):
"""Generator that reads a file in fixed-size chunks."""
with open(file_path, "rb") as f:
while True:
chunk = f.read(chunk_size)
if not chunk:
break
yield chunk
def batch_processor(items, batch_size=32):
"""Generator that yields items in batches."""
batch = []
for item in items:
batch.append(item)
if len(batch) == batch_size:
yield batch
batch = []
if batch:
yield batch'''
},
{
"id": "snippet_04",
"category": "context_manager",
"code": '''import time
class Timer:
"""Context manager for timing code blocks."""
def __init__(self, label=""):
self.label = label
self.elapsed = 0.0
def __enter__(self):
self._start = time.perf_counter()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.elapsed = time.perf_counter() - self._start
if self.label:
print(f"{self.label}: {self.elapsed:.4f}s")
return False'''
},
{
"id": "snippet_05",
"category": "recursive_algorithm",
"code": '''def flatten(nested):
"""Recursively flatten a nested list structure."""
result = []
for item in nested:
if isinstance(item, (list, tuple)):
result.extend(flatten(item))
else:
result.append(item)
return result
def tree_paths(tree, path=None):
"""Find all root-to-leaf paths in a dict-based tree."""
if path is None:
path = []
if not isinstance(tree, dict) or not tree:
return [path]
paths = []
for key, subtree in tree.items():
paths.extend(tree_paths(subtree, path + [key]))
return paths'''
},
]
def load_snippets():
"""Load and return all source snippets."""
return SNIPPETSWhat does this code do?
This file defines the raw material for your training data pipeline. Each snippet is a dictionary with three fields:
- id gives each snippet a unique identifier so you can trace generated training pairs back to their source.
- category labels the coding pattern (decorator, generator, etc.) for tracking what kinds of data the pipeline produces.
- code holds the actual Python source that the LLM will read and generate training pairs from.
The load_snippets() function at the bottom returns the full list. The pipeline will call this function to fetch the snippets rather than importing the list directly. This keeps the interface clean if you later swap in a database or API as the snippet source.
Why 5 different coding patterns?
Diversity in your source material leads to diversity in the generated training data. Each snippet category exercises a different part of the LLM's code understanding. A decorator tests knowledge of closures and function wrapping. A class tests object-oriented reasoning. A generator tests iteration protocols.
In production pipelines, you would source hundreds or thousands of snippets. Five is enough to demonstrate the full pipeline without hitting API rate limits.
- Save source_snippets.py.
- Verify the file is valid by running this in your terminal:
python source_snippets.pyYou should see no output and no errors. The script defines data but does not print anything when run directly.
Seeing a SyntaxError?
The triple-quoted strings ('''...''') inside each snippet's code field are easy to break if a quote gets misplaced during copy-paste. Make sure each code value starts and ends with exactly three single quotes.
Also check that the indentation inside each triple-quoted string is preserved exactly. Python is whitespace-sensitive, and the code stored in these strings will be sent to the LLM as-is.
help me debug a SyntaxError in my source_snippets.py file
- Confirm the snippets load correctly by running this quick test:
python -c "from source_snippets import load_snippets; print(len(load_snippets()), 'snippets loaded')"You should see 5 snippets loaded printed in your terminal.
Seeing ModuleNotFoundError?
Make sure your terminal is in the same folder as source_snippets.py. Run ls (macOS/Linux) or dir (Windows) and confirm you see the file listed.
help me fix a ModuleNotFoundError for source_snippets
✔️ Awesome, I've got everything!
Great. Double check you have saved source_snippets.py and that the import test prints 5 snippets loaded.
ⓧ I'd like to double check the full code
Here is the complete source_snippets.py file for comparison:
"""Source code snippets for training data generation."""
SNIPPETS = [
{
"id": "snippet_01",
"category": "decorator",
"code": '''def retry(max_attempts=3, delay=1):
"""Decorator that retries a function on exception."""
import time
import functools
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
attempts = 0
while attempts < max_attempts:
try:
return func(*args, **kwargs)
except Exception as e:
attempts += 1
if attempts == max_attempts:
raise
time.sleep(delay * attempts)
return wrapper
return decorator'''
},
{
"id": "snippet_02",
"category": "class_error_handling",
"code": '''class ConnectionPool:
"""A simple connection pool with health checking."""
def __init__(self, max_connections=10):
self._pool = []
self._max = max_connections
self._in_use = set()
def acquire(self):
if self._pool:
conn = self._pool.pop()
self._in_use.add(id(conn))
return conn
if len(self._in_use) >= self._max:
raise RuntimeError("Connection pool exhausted")
conn = self._create_connection()
self._in_use.add(id(conn))
return conn
def release(self, conn):
if id(conn) not in self._in_use:
raise ValueError("Connection not from this pool")
self._in_use.discard(id(conn))
self._pool.append(conn)
def _create_connection(self):
return {"status": "connected", "id": id(object())}'''
},
{
"id": "snippet_03",
"category": "generator",
"code": '''def chunked_reader(file_path, chunk_size=1024):
"""Generator that reads a file in fixed-size chunks."""
with open(file_path, "rb") as f:
while True:
chunk = f.read(chunk_size)
if not chunk:
break
yield chunk
def batch_processor(items, batch_size=32):
"""Generator that yields items in batches."""
batch = []
for item in items:
batch.append(item)
if len(batch) == batch_size:
yield batch
batch = []
if batch:
yield batch'''
},
{
"id": "snippet_04",
"category": "context_manager",
"code": '''import time
class Timer:
"""Context manager for timing code blocks."""
def __init__(self, label=""):
self.label = label
self.elapsed = 0.0
def __enter__(self):
self._start = time.perf_counter()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.elapsed = time.perf_counter() - self._start
if self.label:
print(f"{self.label}: {self.elapsed:.4f}s")
return False'''
},
{
"id": "snippet_05",
"category": "recursive_algorithm",
"code": '''def flatten(nested):
"""Recursively flatten a nested list structure."""
result = []
for item in nested:
if isinstance(item, (list, tuple)):
result.extend(flatten(item))
else:
result.append(item)
return result
def tree_paths(tree, path=None):
"""Find all root-to-leaf paths in a dict-based tree."""
if path is None:
path = []
if not isinstance(tree, dict) or not tree:
return [path]
paths = []
for key, subtree in tree.items():
paths.extend(tree_paths(subtree, path + [key]))
return paths'''
},
]
def load_snippets():
"""Load and return all source snippets."""
return SNIPPETSYour source data is ready. Next up, you will build the generation engine that feeds these snippets to an LLM and gets structured training pairs back.
Build the Data Generation Engine
Your source snippets are ready. Now you need a way to transform raw code into structured training data that an LLM can learn from.
In this step, you will build the core generation engine. It uses Groq's Llama 3.3 70B model to read each code snippet and produce diverse (instruction, response) pairs. This is data distillation in action. You feed knowledge from a large model into a structured format that a smaller model can be fine-tuned on.
In this step, get ready to:
- Create pipeline.py with a carefully engineered system prompt for training data generation.
- Build a function that calls Groq's API and parses structured JSON responses.
- Plan rate-limit-aware processing to stay within the free tier.
Create the pipeline file and system prompt
The quality of your training data depends entirely on how well you instruct the LLM to generate it. A good system prompt constrains the model's output format and guides it toward diverse, useful pairs.
- Create a new file called pipeline.py in your project folder.
- Add the following imports and system prompt by pasting this code:
"""Training data pipeline: generate and filter SFT pairs from code snippets."""
import hashlib
import json
import os
import time
from groq import Groq
from source_snippets import load_snippets
SYSTEM_PROMPT = """You are a training data generator for a code-focused LLM. Given a Python code snippet, generate exactly 3 diverse (instruction, response) pairs that could be used to fine-tune a coding assistant.
Requirements:
- Each pair must be about the PROVIDED code snippet
- Generate diverse instruction types: one "explain" (ask what the code does), one "debug/improve" (ask to find issues or improve), one "extend" (ask to add a feature or modify)
- Instructions should be natural questions a developer would ask
- Responses should be detailed, accurate, and reference the specific code
- Return ONLY a JSON array with exactly 3 objects, each having "instruction" and "response" keys
Example output format:
[
{"instruction": "What does this function do?", "response": "This function..."},
{"instruction": "How can I improve the error handling?", "response": "You could improve..."},
{"instruction": "How would I add logging to this?", "response": "To add logging..."}
]"""What does this code do?
The imports bring in everything the pipeline will need. hashlib is for deduplication later, json for parsing LLM responses, os for environment variables, and time for rate limiting.
SYSTEM_PROMPT is the instruction that shapes every LLM response. It tells the model to generate exactly 3 pairs with specific diversity requirements (explain, debug/improve, extend) and return structured JSON. The example output format anchors the model's response structure.
Why three instruction types?
Real SFT datasets need variety. If every instruction asks "what does this code do?", the fine-tuned model only learns to explain. By forcing one explain, one debug/improve, and one extend pair per snippet, the dataset covers multiple developer workflows from the same source material.
- Save pipeline.py.
- Confirm the file exists by running:
lsYou should see both pipeline.py and source_snippets.py listed in the output.
Don't see pipeline.py?
Make sure you saved the file in your project folder, not a different directory. Check that your terminal is in the same folder by looking at the path in your prompt.
Help me find where my file was saved.
Build the generate_pairs function
This function is the heart of the pipeline. It sends each code snippet to the LLM, receives a JSON response, and extracts validated (instruction, response) pairs.
- In pipeline.py, add the following function below the SYSTEM_PROMPT constant:
def generate_pairs(client, code_snippet):
"""Generate (instruction, response) pairs from a code snippet using Groq."""
user_message = f"Generate training pairs for this Python code:\n\n```python\n{code_snippet}\n```"
try:
chat_completion = client.chat.completions.create(
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
model="llama-3.3-70b-versatile",
temperature=0.7,
max_tokens=2000,
response_format={"type": "json_object"},
)
response_text = chat_completion.choices[0].message.content
parsed = json.loads(response_text)
if isinstance(parsed, list):
pairs = parsed
elif isinstance(parsed, dict):
for value in parsed.values():
if isinstance(value, list):
pairs = value
break
else:
return []
else:
return []
valid_pairs = []
for pair in pairs:
if isinstance(pair, dict) and "instruction" in pair and "response" in pair:
valid_pairs.append(pair)
return valid_pairs
except json.JSONDecodeError:
print(" Warning: Failed to parse LLM response as JSON")
return []
except Exception as e:
print(f" Warning: API call failed: {e}")
return []What does this function do?
- response_format={"type": "json_object"} tells Groq to constrain the model's output to valid JSON. This dramatically reduces parsing failures compared to free-form text.
- The parsing section handles two response shapes. Some models return a raw JSON array, while others wrap it in an object like {"pairs": [...]}. The code handles both gracefully.
- The validation loop ensures each pair is a dictionary with both instruction and response keys present. Malformed entries are silently dropped.
- The try/except block catches JSON parse errors and API failures separately. Instead of crashing the whole pipeline, it returns an empty list so processing can continue with the next snippet.
- Save pipeline.py.
You won't see output yet. The next step adds quality filters, and Step 5 wires everything into a runnable pipeline.
Seeing a syntax error on save?
Check that the triple-quoted f-string for user_message uses regular quotes, not curly quotes. Also verify the response_format dictionary uses straight double quotes around json_object.
Help me fix a syntax error in my generate_pairs function.
Plan for rate limiting
Groq's free tier allows 30 requests per minute for the llama-3.3-70b-versatile model. Your pipeline will process 5 snippets, making 5 API calls total.
Why a 2.5-second delay?
At 30 RPM, you can safely make one request every 2 seconds. A 2.5-second delay adds a small buffer to avoid hitting the limit if requests cluster. The import time at the top of your file provides the time.sleep() function you will use for this delay.
You will add the actual time.sleep(2.5) call in Step 5 when you wire up the main loop that iterates over all snippets. For now, the import is in place and ready.
✔️ Awesome, I've got everything!
Great. Double check you have saved pipeline.py with both the SYSTEM_PROMPT constant and the generate_pairs function.
ⓧ I'd like to double check the full code
Here is the complete pipeline.py so far:
"""Training data pipeline: generate and filter SFT pairs from code snippets."""
import hashlib
import json
import os
import time
from groq import Groq
from source_snippets import load_snippets
SYSTEM_PROMPT = """You are a training data generator for a code-focused LLM. Given a Python code snippet, generate exactly 3 diverse (instruction, response) pairs that could be used to fine-tune a coding assistant.
Requirements:
- Each pair must be about the PROVIDED code snippet
- Generate diverse instruction types: one "explain" (ask what the code does), one "debug/improve" (ask to find issues or improve), one "extend" (ask to add a feature or modify)
- Instructions should be natural questions a developer would ask
- Responses should be detailed, accurate, and reference the specific code
- Return ONLY a JSON array with exactly 3 objects, each having "instruction" and "response" keys
Example output format:
[
{"instruction": "What does this function do?", "response": "This function..."},
{"instruction": "How can I improve the error handling?", "response": "You could improve..."},
{"instruction": "How would I add logging to this?", "response": "To add logging..."}
]"""
def generate_pairs(client, code_snippet):
"""Generate (instruction, response) pairs from a code snippet using Groq."""
user_message = f"Generate training pairs for this Python code:\n\n```python\n{code_snippet}\n```"
try:
chat_completion = client.chat.completions.create(
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
model="llama-3.3-70b-versatile",
temperature=0.7,
max_tokens=2000,
response_format={"type": "json_object"},
)
response_text = chat_completion.choices[0].message.content
parsed = json.loads(response_text)
if isinstance(parsed, list):
pairs = parsed
elif isinstance(parsed, dict):
for value in parsed.values():
if isinstance(value, list):
pairs = value
break
else:
return []
else:
return []
valid_pairs = []
for pair in pairs:
if isinstance(pair, dict) and "instruction" in pair and "response" in pair:
valid_pairs.append(pair)
return valid_pairs
except json.JSONDecodeError:
print(" Warning: Failed to parse LLM response as JSON")
return []
except Exception as e:
print(f" Warning: API call failed: {e}")
return []Your generation engine is ready to produce raw training pairs. Next up, you will build the quality filters that separate good data from noise.
Implement Quality Filters
In the previous step, you built the data generation engine that calls Groq to produce (instruction, response) pairs from code snippets.
Raw LLM output is messy. Some pairs might have one-word responses, broken JSON, or instructions that just echo the code back. In production data pipelines, a quality filtering layer sits between generation and the final dataset to catch these problems. In this step, you will build a multi-stage filter that rejects bad data before it reaches your training file.
In this step, get ready to:
- Build a QualityFilter class with length, format, and content checks.
- Add deduplication using MD5 hashing.
- Implement a batch filter method that runs all checks and tracks statistics.
Build the QualityFilter class with validation methods
Every pair the LLM generates needs to pass three checks before it earns a spot in the final dataset. Length constraints prevent one-word or absurdly long entries. Format validation proves the pair can convert to the HuggingFace TRL conversational schema. Content checks catch low-quality responses that just echo the instruction.
- In pipeline.py, scroll to the bottom of the file (after the generate_pairs function).
- Add the QualityFilter class with its three validation methods by pasting this code at the bottom of the file:
class QualityFilter:
"""Multi-stage quality filter for generated training pairs."""
def __init__(self):
self._seen_hashes = set()
self.stats = {
"length_filtered": 0,
"format_filtered": 0,
"content_filtered": 0,
"duplicate_filtered": 0,
}
def check_length(self, pair):
"""Check instruction and response length constraints."""
instruction = pair["instruction"]
response = pair["response"]
if len(instruction) < 10 or len(instruction) > 500:
return False
if len(response) < 50 or len(response) > 2000:
return False
return True
def check_format(self, pair):
"""Validate the pair can be converted to TRL conversational format."""
try:
messages = [
{"role": "user", "content": pair["instruction"]},
{"role": "assistant", "content": pair["response"]},
]
json.dumps({"messages": messages})
return True
except (TypeError, ValueError):
return False
def check_content(self, pair):
"""Reject low-quality content (echoes, too similar instruction/response)."""
instruction = pair["instruction"].lower().strip()
response = pair["response"].lower().strip()
# Reject if response is too short relative to instruction
if len(response) < len(instruction):
return False
# Reject if response just echoes the instruction
if instruction in response[:len(instruction) + 50]:
return False
return TrueWhat does this code do?
- __init__ creates a _seen_hashes set for deduplication (used later) and a stats dictionary to track how many pairs each filter removes.
- check_length enforces character limits: instructions must be 10-500 characters and responses must be 50-2000 characters. Too short means low quality. Too long means the model rambled.
- check_format builds the TRL conversational messages structure and confirms it serializes to valid JSON. If the pair contains un-serializable data, this catch prevents corrupt entries.
- check_content rejects pairs where the response is shorter than the instruction (lazy answers) or where the response starts by echoing the instruction back (a common LLM failure mode).
You will not see a change yet. The next chunk adds deduplication, and the final chunk wires all filters together.
Add deduplication with MD5 hashing
LLMs sometimes generate near-identical instructions across different snippets. Deduplication catches exact matches by hashing the normalized instruction text and rejecting any instruction it has seen before.
- Still inside the QualityFilter class, add the check_duplicate method below check_content:
def check_duplicate(self, pair):
"""Deduplicate using MD5 hash of normalized instruction."""
normalized = pair["instruction"].lower().strip()
hash_val = hashlib.md5(normalized.encode()).hexdigest()
if hash_val in self._seen_hashes:
return False
self._seen_hashes.add(hash_val)
return TrueWhat does this code do?
The method lowercases and strips the instruction, then generates an MD5 hash of the result. If that hash already exists in _seen_hashes, the pair is a duplicate and gets rejected. Otherwise, the hash is stored and the pair passes through.
Normalizing before hashing means "What does this do?" and "what does this do?" count as the same instruction, catching duplicates that differ only in casing or whitespace.
You will not see a change yet. The next chunk adds the orchestrator method that runs all four checks in sequence.
Implement the batch filter method
Each individual check returns a simple True/False. The batch filter ties them together into a pipeline, running every pair through all four gates in order and tracking which filter caught what.
- Still inside the QualityFilter class, add the filter_batch method below check_duplicate:
def filter_batch(self, pairs):
"""Run all quality checks on a batch of pairs."""
passed = []
for pair in pairs:
if not self.check_length(pair):
self.stats["length_filtered"] += 1
continue
if not self.check_format(pair):
self.stats["format_filtered"] += 1
continue
if not self.check_content(pair):
self.stats["content_filtered"] += 1
continue
if not self.check_duplicate(pair):
self.stats["duplicate_filtered"] += 1
continue
passed.append(pair)
return passedWhat does this code do?
The method loops through every pair and runs the four checks in order: length, format, content, then duplicate. If any check fails, the stats counter for that filter increments and the pair is skipped with continue. Only pairs that pass all four checks get appended to the passed list.
The ordering matters. Length and format checks are cheap. Content and dedup are slightly more expensive. Running cheap filters first means fewer pairs reach the expensive checks.
- Save pipeline.py.
- Verify the class works by running this quick import test in your terminal:
python -c "from pipeline import QualityFilter; qf = QualityFilter(); print(qf.stats)"You should see the stats dictionary printed with all counters at zero:
{'length_filtered': 0, 'format_filtered': 0, 'content_filtered': 0, 'duplicate_filtered': 0}Seeing an ImportError or SyntaxError?
- Make sure the QualityFilter class is defined at the module level (no extra indentation). It should start at column 0, just like generate_pairs.
- Check that all methods inside the class are indented one level (4 spaces). Python is strict about consistent indentation.
- Confirm you saved the file before running the import command.
Why is my QualityFilter class not importing correctly?
✔️ Awesome, I've got everything!
Great. Double check you have saved pipeline.py and the import test printed the stats dictionary with all zeros.
ⓧ I'd like to double check the full code
Here is the complete pipeline.py at this point in the project:
"""Training data pipeline: generate and filter SFT pairs from code snippets."""
import hashlib
import json
import os
import time
from groq import Groq
from source_snippets import load_snippets
SYSTEM_PROMPT = """You are a training data generator for a code-focused LLM. Given a Python code snippet, generate exactly 3 diverse (instruction, response) pairs that could be used to fine-tune a coding assistant.
Requirements:
- Each pair must be about the PROVIDED code snippet
- Generate diverse instruction types: one "explain" (ask what the code does), one "debug/improve" (ask to find issues or improve), one "extend" (ask to add a feature or modify)
- Instructions should be natural questions a developer would ask
- Responses should be detailed, accurate, and reference the specific code
- Return ONLY a JSON array with exactly 3 objects, each having "instruction" and "response" keys
Example output format:
[
{"instruction": "What does this function do?", "response": "This function..."},
{"instruction": "How can I improve the error handling?", "response": "You could improve..."},
{"instruction": "How would I add logging to this?", "response": "To add logging..."}
]"""
def generate_pairs(client, code_snippet):
"""Generate (instruction, response) pairs from a code snippet using Groq."""
user_message = f"Generate training pairs for this Python code:\n\n```python\n{code_snippet}\n```"
try:
chat_completion = client.chat.completions.create(
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
model="llama-3.3-70b-versatile",
temperature=0.7,
max_tokens=2000,
response_format={"type": "json_object"},
)
response_text = chat_completion.choices[0].message.content
parsed = json.loads(response_text)
if isinstance(parsed, list):
pairs = parsed
elif isinstance(parsed, dict):
for value in parsed.values():
if isinstance(value, list):
pairs = value
break
else:
return []
else:
return []
valid_pairs = []
for pair in pairs:
if isinstance(pair, dict) and "instruction" in pair and "response" in pair:
valid_pairs.append(pair)
return valid_pairs
except json.JSONDecodeError:
print(" Warning: Failed to parse LLM response as JSON")
return []
except Exception as e:
print(f" Warning: API call failed: {e}")
return []
class QualityFilter:
"""Multi-stage quality filter for generated training pairs."""
def __init__(self):
self._seen_hashes = set()
self.stats = {
"length_filtered": 0,
"format_filtered": 0,
"content_filtered": 0,
"duplicate_filtered": 0,
}
def check_length(self, pair):
"""Check instruction and response length constraints."""
instruction = pair["instruction"]
response = pair["response"]
if len(instruction) < 10 or len(instruction) > 500:
return False
if len(response) < 50 or len(response) > 2000:
return False
return True
def check_format(self, pair):
"""Validate the pair can be converted to TRL conversational format."""
try:
messages = [
{"role": "user", "content": pair["instruction"]},
{"role": "assistant", "content": pair["response"]},
]
json.dumps({"messages": messages})
return True
except (TypeError, ValueError):
return False
def check_content(self, pair):
"""Reject low-quality content (echoes, too similar instruction/response)."""
instruction = pair["instruction"].lower().strip()
response = pair["response"].lower().strip()
if len(response) < len(instruction):
return False
if instruction in response[:len(instruction) + 50]:
return False
return True
def check_duplicate(self, pair):
"""Deduplicate using MD5 hash of normalized instruction."""
normalized = pair["instruction"].lower().strip()
hash_val = hashlib.md5(normalized.encode()).hexdigest()
if hash_val in self._seen_hashes:
return False
self._seen_hashes.add(hash_val)
return True
def filter_batch(self, pairs):
"""Run all quality checks on a batch of pairs."""
passed = []
for pair in pairs:
if not self.check_length(pair):
self.stats["length_filtered"] += 1
continue
if not self.check_format(pair):
self.stats["format_filtered"] += 1
continue
if not self.check_content(pair):
self.stats["content_filtered"] += 1
continue
if not self.check_duplicate(pair):
self.stats["duplicate_filtered"] += 1
continue
passed.append(pair)
return passedYour quality gates are in place. Next up, you will wire the generator and filters together into a single pipeline that produces a clean JSONL dataset file.
Assemble the Pipeline and Generate Your Dataset
You have the two core engines of your pipeline built: a generation function that calls Groq to produce training pairs, and a quality filter that catches bad data. Now you need to connect them into a single runnable script.
In this step, you will wire together a main() function that loads your source snippets, generates pairs for each one, filters the batch, and writes the clean results to a JSONL file ready for fine-tuning.
In this step, get ready to:
- Wire the generation engine and quality filter into a single main() function.
- Add pipeline statistics and write the output to dataset.jsonl.
- Run the pipeline and verify your training dataset.
Add the format converter and generation loop
The HuggingFace TRL SFTTrainer expects each training example as a JSON object with a messages array containing user and assistant roles. You need a helper function to convert your raw pairs into this format.
- In pipeline.py, scroll to the bottom of the file (after the QualityFilter class) and add this format converter function:
def to_sft_format(pair):
"""Convert a pair to HuggingFace TRL conversational SFT format."""
return {
"messages": [
{"role": "user", "content": pair["instruction"]},
{"role": "assistant", "content": pair["response"]},
]
}What does this code do?
This function takes a raw pair dictionary (with instruction and response keys) and wraps it in the conversational format that TRL SFTTrainer expects.
The output looks like: {"messages": [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]}. This is the same format used by OpenAI fine-tuning and Axolotl.
Now you need the main() function that orchestrates everything. It initializes the Groq client, loops through each snippet, and collects raw pairs with a rate-limit delay between calls.
- Below to_sft_format, add the first part of your main() function:
def main():
"""Run the full training data pipeline."""
print("=" * 60)
print("Training Data Pipeline - SFT Dataset Generator")
print("=" * 60)
# Initialize Groq client
client = Groq(api_key=os.environ.get("GROQ_API_KEY"))
# Load source snippets
snippets = load_snippets()
print(f"\nLoaded {len(snippets)} source code snippets")
# Generate pairs from each snippet
all_pairs = []
for i, snippet in enumerate(snippets):
print(f"\nProcessing snippet {i + 1}/{len(snippets)}: {snippet['id']} ({snippet['category']})")
pairs = generate_pairs(client, snippet["code"])
print(f" Generated {len(pairs)} pairs")
# Tag pairs with source info
for pair in pairs:
pair["source_id"] = snippet["id"]
pair["category"] = snippet["category"]
all_pairs.extend(pairs)
# Rate limit delay (stay within 30 RPM free tier)
if i < len(snippets) - 1:
time.sleep(2.5)What does this code do?
- Groq(api_key=os.environ.get("GROQ_API_KEY")) creates the API client using your environment variable.
- The loop calls generate_pairs for each snippet, then tags each pair with source_id and category so you can trace where each training example came from.
- time.sleep(2.5) pauses 2.5 seconds between API calls. Groq's free tier allows 30 requests per minute, so this keeps you safely under the limit.
Don't save yet. You still need to add the filtering and output logic below this code.
Add quality filtering, statistics, and JSONL output
The generation loop collects raw pairs, but you need to filter them, report what was removed, and write the survivors to a file. This second half of main() ties everything together.
- Still inside the main() function (at the same indentation level as the loop above), add the filtering and output code:
print(f"\n{'=' * 60}")
print(f"Total raw pairs generated: {len(all_pairs)}")
# Apply quality filters
quality_filter = QualityFilter()
filtered_pairs = quality_filter.filter_batch(all_pairs)
print(f"Pairs passing quality filters: {len(filtered_pairs)}")
print(f"\nFilter breakdown:")
for filter_name, count in quality_filter.stats.items():
print(f" {filter_name}: {count} removed")
# Write to JSONL
output_file = "dataset.jsonl"
with open(output_file, "w", encoding="utf-8") as f:
for pair in filtered_pairs:
sft_entry = to_sft_format(pair)
f.write(json.dumps(sft_entry, ensure_ascii=False) + "\n")
print(f"\nDataset written to: {output_file}")
print(f"Total training examples: {len(filtered_pairs)}")
print(f"{'=' * 60}")
# Show a sample entry
if filtered_pairs:
print("\nSample entry from dataset.jsonl:")
sample = to_sft_format(filtered_pairs[0])
print(json.dumps(sample, indent=2, ensure_ascii=False)[:500])
if __name__ == "__main__":
main()What does this code do?
- QualityFilter().filter_batch(all_pairs) runs every pair through the length, format, content, and deduplication checks you built in the previous step.
- The filter breakdown prints exactly how many pairs each filter rejected, so you can see which quality gate catches the most issues.
- The JSONL writer calls to_sft_format on each passing pair and writes one JSON object per line. This is the standard format for feeding data into fine-tuning tools.
- if __name__ == "__main__": ensures main() only runs when you execute the script directly, not when other files import from it.
- Save pipeline.py.
Seeing indentation errors?
Make sure the second block of code (starting with the print line) is indented one level inside main(). It continues the same function from the previous chunk. The only line that should be at the top level is if __name__ == "__main__":.
Help me fix my indentation in pipeline.py
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
Here is the complete pipeline.py file at this point:
"""Training data pipeline: generate and filter SFT pairs from code snippets."""
import hashlib
import json
import os
import time
from groq import Groq
from source_snippets import load_snippets
SYSTEM_PROMPT = """You are a training data generator for a code-focused LLM. Given a Python code snippet, generate exactly 3 diverse (instruction, response) pairs that could be used to fine-tune a coding assistant.
Requirements:
- Each pair must be about the PROVIDED code snippet
- Generate diverse instruction types: one "explain" (ask what the code does), one "debug/improve" (ask to find issues or improve), one "extend" (ask to add a feature or modify)
- Instructions should be natural questions a developer would ask
- Responses should be detailed, accurate, and reference the specific code
- Return ONLY a JSON array with exactly 3 objects, each having "instruction" and "response" keys
Example output format:
[
{"instruction": "What does this function do?", "response": "This function..."},
{"instruction": "How can I improve the error handling?", "response": "You could improve..."},
{"instruction": "How would I add logging to this?", "response": "To add logging..."}
]"""
def generate_pairs(client, code_snippet):
"""Generate (instruction, response) pairs from a code snippet using Groq."""
user_message = f"Generate training pairs for this Python code:\n\n```python\n{code_snippet}\n```"
try:
chat_completion = client.chat.completions.create(
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
model="llama-3.3-70b-versatile",
temperature=0.7,
max_tokens=2000,
response_format={"type": "json_object"},
)
response_text = chat_completion.choices[0].message.content
parsed = json.loads(response_text)
# Handle both direct array and wrapped object responses
if isinstance(parsed, list):
pairs = parsed
elif isinstance(parsed, dict):
# Look for an array value in the dict
for value in parsed.values():
if isinstance(value, list):
pairs = value
break
else:
return []
else:
return []
# Validate each pair has the required keys
valid_pairs = []
for pair in pairs:
if isinstance(pair, dict) and "instruction" in pair and "response" in pair:
valid_pairs.append(pair)
return valid_pairs
except json.JSONDecodeError:
print(" Warning: Failed to parse LLM response as JSON")
return []
except Exception as e:
print(f" Warning: API call failed: {e}")
return []
class QualityFilter:
"""Multi-stage quality filter for generated training pairs."""
def __init__(self):
self._seen_hashes = set()
self.stats = {
"length_filtered": 0,
"format_filtered": 0,
"content_filtered": 0,
"duplicate_filtered": 0,
}
def check_length(self, pair):
"""Check instruction and response length constraints."""
instruction = pair["instruction"]
response = pair["response"]
if len(instruction) < 10 or len(instruction) > 500:
return False
if len(response) < 50 or len(response) > 2000:
return False
return True
def check_format(self, pair):
"""Validate the pair can be converted to TRL conversational format."""
try:
messages = [
{"role": "user", "content": pair["instruction"]},
{"role": "assistant", "content": pair["response"]},
]
# Verify it's JSON-serializable
json.dumps({"messages": messages})
return True
except (TypeError, ValueError):
return False
def check_content(self, pair):
"""Reject low-quality content (echoes, too similar instruction/response)."""
instruction = pair["instruction"].lower().strip()
response = pair["response"].lower().strip()
# Reject if response is too short relative to instruction
if len(response) < len(instruction):
return False
# Reject if response just echoes the instruction
if instruction in response[:len(instruction) + 50]:
return False
return True
def check_duplicate(self, pair):
"""Deduplicate using MD5 hash of normalized instruction."""
normalized = pair["instruction"].lower().strip()
hash_val = hashlib.md5(normalized.encode()).hexdigest()
if hash_val in self._seen_hashes:
return False
self._seen_hashes.add(hash_val)
return True
def filter_batch(self, pairs):
"""Run all quality checks on a batch of pairs."""
passed = []
for pair in pairs:
if not self.check_length(pair):
self.stats["length_filtered"] += 1
continue
if not self.check_format(pair):
self.stats["format_filtered"] += 1
continue
if not self.check_content(pair):
self.stats["content_filtered"] += 1
continue
if not self.check_duplicate(pair):
self.stats["duplicate_filtered"] += 1
continue
passed.append(pair)
return passed
def to_sft_format(pair):
"""Convert a pair to HuggingFace TRL conversational SFT format."""
return {
"messages": [
{"role": "user", "content": pair["instruction"]},
{"role": "assistant", "content": pair["response"]},
]
}
def main():
"""Run the full training data pipeline."""
print("=" * 60)
print("Training Data Pipeline - SFT Dataset Generator")
print("=" * 60)
# Initialize Groq client
client = Groq(api_key=os.environ.get("GROQ_API_KEY"))
# Load source snippets
snippets = load_snippets()
print(f"\nLoaded {len(snippets)} source code snippets")
# Generate pairs from each snippet
all_pairs = []
for i, snippet in enumerate(snippets):
print(f"\nProcessing snippet {i + 1}/{len(snippets)}: {snippet['id']} ({snippet['category']})")
pairs = generate_pairs(client, snippet["code"])
print(f" Generated {len(pairs)} pairs")
# Tag pairs with source info
for pair in pairs:
pair["source_id"] = snippet["id"]
pair["category"] = snippet["category"]
all_pairs.extend(pairs)
# Rate limit delay (stay within 30 RPM free tier)
if i < len(snippets) - 1:
time.sleep(2.5)
print(f"\n{'=' * 60}")
print(f"Total raw pairs generated: {len(all_pairs)}")
# Apply quality filters
quality_filter = QualityFilter()
filtered_pairs = quality_filter.filter_batch(all_pairs)
print(f"Pairs passing quality filters: {len(filtered_pairs)}")
print(f"\nFilter breakdown:")
for filter_name, count in quality_filter.stats.items():
print(f" {filter_name}: {count} removed")
# Write to JSONL
output_file = "dataset.jsonl"
with open(output_file, "w", encoding="utf-8") as f:
for pair in filtered_pairs:
sft_entry = to_sft_format(pair)
f.write(json.dumps(sft_entry, ensure_ascii=False) + "\n")
print(f"\nDataset written to: {output_file}")
print(f"Total training examples: {len(filtered_pairs)}")
print(f"{'=' * 60}")
# Show a sample entry
if filtered_pairs:
print("\nSample entry from dataset.jsonl:")
sample = to_sft_format(filtered_pairs[0])
print(json.dumps(sample, indent=2, ensure_ascii=False)[:500])
if __name__ == "__main__":
main()Run the pipeline and verify the output
Your pipeline is complete. Time to run it and produce your training dataset. The script will take about 15 seconds to finish because of the rate-limit delays between API calls.
- Run the pipeline by executing this command in your terminal:
python pipeline.pyWhat should I see?
The pipeline prints progress as it processes each snippet. You should see output like this:
- A header banner with the pipeline name.
- Five "Processing snippet" lines, each followed by "Generated 3 pairs".
- A summary showing total raw pairs (likely 15), how many passed filters, and a breakdown of what each filter removed.
- A sample JSON entry from the output file showing the conversational format.
- Verify that dataset.jsonl was created by running:
head -n 1 dataset.jsonl- You should see a single JSON line with a messages array containing a user role and an assistant role. This confirms your dataset is in the correct HuggingFace TRL conversational format.
Pipeline failing with an API error?
- If you see "API call failed" for every snippet, check that your GROQ_API_KEY environment variable is still set in this terminal session. If you opened a new terminal, you need to export it again.
- If you see a rate limit error (429), wait 60 seconds and try again. The free tier allows 30 requests per minute.
- If dataset.jsonl is empty, check your terminal output for "Warning: Failed to parse LLM response as JSON" messages. This can happen occasionally. Run the pipeline again.
Help me debug my pipeline run
You just built a complete training data pipeline from scratch. You have a clean, deduplicated JSONL dataset that could be fed directly into a fine-tuning job. Next up, you will extend this pipeline with a diversity scorer that catches semantically similar pairs the deduplication filter misses.
Secret mission
Add a Diversity Scorer
Your quality filters catch bad data, but what about data that's technically valid yet repetitive? Two instructions like 'What does this function do?' and 'What does this code do?' both pass every check, yet they teach the model the same thing. Can you build a semantic similarity detector that catches these hidden duplicates?
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally with the Groq free tier, so there are no ongoing costs.
Resources you used:
- Project folder on your Desktop (contains source_snippets.py, pipeline.py, dataset.jsonl, and your Python virtual environment).
- GROQ_API_KEY environment variable set in your terminal session.
- Groq account with a free-tier API key.
Keep everything running
No action needed. Choose this if you plan to extend the pipeline with more snippets, experiment with different prompts, or add more quality filters.
- Your project folder stays on your Desktop with all files intact.
- Your GROQ_API_KEY environment variable will disappear when you close the terminal session (unless you added it to your shell profile).
- Your Groq account remains active with free-tier access.
Pause - I'll come back to this later
There are no running processes to stop. The pipeline runs once and exits, so there is nothing consuming resources between runs.
- Your project folder stays on your Desktop ready for next time.
- When you return, reactivate your virtual environment and re-export your GROQ_API_KEY before running the pipeline again.
Delete - I don't want to use this again
Remove all project files and clean up your environment completely.
- Deactivate your virtual environment (if it is still active) by running:
deactivate- Unset the GROQ_API_KEY environment variable by running:
unset GROQ_API_KEY- Delete the project folder from your Desktop by running:
cd ~/Desktop
rm -rf [[PROJECT_FOLDER="your-project-folder-name"]]Windows users
Use this command instead to delete the project folder:
cd ~\Desktop
rmdir /s /q [[PROJECT_FOLDER="your-project-folder-name"]]- Your Groq account is free and does not need to be deleted. No charges will ever be applied to it.
Nice Work!
Nice Work!
You just built a complete training data pipeline for LLM fine-tuning. This is the exact data curation and distillation workflow used in production AI engineering roles.
Here's what you learned:
- Built an LLM-powered synthetic data generation engine using Groq's API and Llama 3.3 70B to distill code snippets into structured training pairs.
- Implemented multi-stage quality filtering with length constraints, schema validation, content checks, and MD5 deduplication to reject low-quality data before it reaches a model.
- Produced a real JSONL dataset in HuggingFace TRL SFTTrainer conversational format, ready to plug directly into a fine-tuning job.
- Secret Mission: Added semantic diversity scoring with TF-IDF cosine similarity to catch non-obvious duplicates that exact-match deduplication misses.
Ready to quiz yourself?