Code LLM Training › Implementation: Pretraining Data Pipeline
Language AI HandbookAll chapters Code LLM Training › Implementation: Pretraining Data PipelineOne section · Sections in this chapter7 / 13 ↓
Reading tools Definitions & inline links
Choose your expertise level to adjust how many terms are explained. Beginners see more tooltips, experts see fewer to maintain reading flow. Hover over underlined terms for instant definitions.
Article links
Make inline references clickable
Let's build a simplified data pipeline that demonstrates the key steps: loading code files, applying filters, and constructing a FIM training objective. We'll use a small in-memory dataset to keep execution fast, but the patterns generalize directly to large-scale pipelines.
import random
import numpy as np
random.seed(42)
np.random.seed(42)
# Simulated small corpus of Python code snippets
RAW_CODE_CORPUS = [
"""
import numpy as np
def compute_mean(values):
\"\"\"Compute the arithmetic mean of a list.\"\"\"
return np.mean(values)
result = compute_mean([1, 2, 3, 4, 5])
print(result)
""",
"""
def factorial(n):
if n == 0:
return 1
return n * factorial(n - 1)
for i in range(10):
print(factorial(i))
""",
"""
class BinaryTree:
def __init__(self, value):
self.value = value
self.left = None
self.right = None
def insert(self, val):
if val < self.value:
if self.left is None:
self.left = BinaryTree(val)
else:
self.left.insert(val)
else:
if self.right is None:
self.right = BinaryTree(val)
else:
self.right.insert(val)
""",
"""
import os
import sys
def read_config(path):
config = {}
with open(path) as f:
for line in f:
key, _, value = line.partition('=')
config[key.strip()] = value.strip()
return config
""",
"""
def bubble_sort(arr):
n = len(arr)
for i in range(n):
for j in range(0, n - i - 1):
if arr[j] > arr[j + 1]:
arr[j], arr[j + 1] = arr[j + 1], arr[j]
return arr
""",
"""
# Auto-generated code
# DO NOT EDIT
x=1;y=2;z=3;a=4;b=5;c=6;d=7;e=8;f=9;g=10
x=1;y=2;z=3;a=4;b=5;c=6;d=7;e=8;f=9;g=10
x=1;y=2;z=3;a=4;b=5;c=6;d=7;e=8;f=9;g=10
""",
"""
# LICENCE: MIT
def add(a, b):
return a + b
""",
# Duplicate of the factorial snippet (simulating copy-paste)
"""
def factorial(n):
if n == 0:
return 1
return n * factorial(n - 1)
for i in range(10):
print(factorial(i))
""",
]Now let's implement the quality filters. Each filter targets a specific signal of low-quality code:
def compute_line_stats(code: str) -> dict:
"""Compute quality metrics for a code file."""
lines = code.strip().split("\n")
non_empty_lines = [l for l in lines if l.strip()]
if not non_empty_lines:
return {
"avg_line_len": 0,
"alpha_ratio": 0,
"unique_ratio": 0,
"num_lines": 0,
}
avg_line_len = np.mean([len(l) for l in non_empty_lines])
all_chars = "".join(non_empty_lines)
alpha_ratio = sum(c.isalpha() for c in all_chars) / max(len(all_chars), 1)
unique_ratio = len(set(non_empty_lines)) / len(non_empty_lines)
return {
"avg_line_len": avg_line_len,
"alpha_ratio": alpha_ratio,
"unique_ratio": unique_ratio,
"num_lines": len(non_empty_lines),
}
def passes_quality_filter(code: str) -> tuple[bool, str]:
"""
Return (passes, reason). Applies basic code quality heuristics.
"""
stats = compute_line_stats(code)
# Too few lines
if stats["num_lines"] < 3:
return False, "too_short"
# Likely minified (very long lines)
if stats["avg_line_len"] > 150:
return False, "likely_minified"
# Very low alphabetic ratio (likely binary or machine-generated)
if stats["alpha_ratio"] < 0.15:
return False, "low_alpha_ratio"
# Very low unique line ratio (repetitive boilerplate)
if stats["unique_ratio"] < 0.4:
return False, "high_repetition"
return True, "passed"
# Apply filters to the corpus
filter_results = [
(code, *passes_quality_filter(code)) for code in RAW_CODE_CORPUS
]
passed_files = [(code, reason) for code, ok, reason in filter_results if ok]
filtered_files = [
(code, reason) for code, ok, reason in filter_results if not ok
]Total files: 8 Passed filter: 8 Filtered out: 0 Filtered files and reasons:
The too_short file is the two-line add function, which is too brief to provide useful training signal. The high_repetition file is the auto-generated code with repeated assignments. Both were correctly identified as low-quality.
def exact_deduplicate(corpus: list[str]) -> list[str]:
"""Remove exact duplicate documents using normalized string hashing."""
seen = set()
unique = []
for doc in corpus:
key = doc.strip()
if key not in seen:
seen.add(key)
unique.append(doc)
return unique
# Extract code strings from (code, reason) tuples
passed_code = [code for code, _ in passed_files]
deduped_corpus = exact_deduplicate(passed_code)Before deduplication: 8 files After deduplication: 7 files Duplicates removed: 1
After filtering and deduplication, our corpus is smaller but higher quality. In production, these steps typically reduce a raw GitHub scrape by 70 to 90 percent, while the remaining data is cleaner and more representative of real, idiomatic code.