LLM training data preparation

licates. Maintain a distributed Bloom filter or Redis-backed set for cross-shard dedup. 4. Quality Scoring & Filtering: Compute heuristics: language confidence, token entropy, readability score, code syntax validity, and toxicity probability. Apply adaptive thresholds per domain. 5. Domain Balancing & Sampling: Route high-quality samples to a reservoir. Apply stratified sampling to maintain target domain ratios (code, prose, technical, conversational). 6. Tokenization & Sequence Packing: Tokenize with the target tokenizer. Pack sequences to context window limits without crossing document boundaries. Preserve attention masks. 7. Versioning & Lineage: Hash dataset snapshots. Store metadata (source, filters applied, timestamp, shard ID). Write to an immutable data lake with Parquet/JSONL output.

TypeScript Pipeline Orchestration

TypeScript excels at pipeline orchestration due to native async iterators, stream backpressure handling, and strong typing for schema validation. The following example demonstrates a composable streaming pipeline with validation, deduplication, quality filtering, and sequence packing.

import { Readable, Transform, pipeline } from 'stream/promises';
import { createReadStream } from 'fs';
import { createInterface } from 'readline';
import { createHash } from 'crypto';
import { z } from 'zod';

// 1. Schema Validation
const RecordSchema = z.object({
  id: z.string().uuid(),
  text: z.string().min(50).max(50000),
  source: z.enum(['web', 'code', 'docs', 'api']),
  metadata: z.object({ language: z.string(), timestamp: z.coerce.date() }).optional()
});

type ValidRecord = z.infer<typeof RecordSchema>;

// 2. Deduplication State (production: Redis/LevelDB)
const dedupSet = new Set<string>();
const SIMILARITY_THRESHOLD = 0.85;

function computeDocHash(text: string): string {
  return createHash('sha256').update(text.trim().toLowerCase()).digest('hex');
}

// 3. Quality Scorer (simplified heuristic)
function computeQualityScore(record: ValidRecord): number {
  const text = record.text;
  const wordCount = text.split(/\s+/).length;
  const avgWordLength = text.replace(/\s+/g, '').length / wordCount;
  const hasCode = /```|import |function |class /g.test(text);
  
  // Higher score for balanced length, meaningful vocabulary, domain signals
  let score = 0;
  score += Math.min(wordCount / 100, 1) * 0.3;
  score += Math.min(avgWordLength / 5, 1) * 0.2;
  score += (record.source === 'code' && hasCode ? 0.3 : 0.1);
  score += record.metadata?.language === 'en' ? 0.2 : 0.0;
  return score;
}

// 4. Pipeline Components
const validateStream = new Transform({
  objectMode: true,
  async transform(chunk, _encoding, callback) {
    try {
      const parsed = JSON.parse(chunk.toString());
      const validated = RecordSchema.parse(parsed);
      callback(null, validated);
    } catch (err) {
      callback(null); // silently drop invalid
    }
  }
});

const dedupStream = new Transform({
  objectMode: true,
  transform(record: ValidRecord, _encoding, callback) {
    const hash = computeDocHash(record.text);
    if (!dedupSet.has(hash)) {
      dedupSet.add(hash);
      callback(null, record);
    } else {
      callback(null);
    }
  }
});

const qualityFilterStream = new Transform({
  objectMode: true,
  transform(record: ValidRecord, _encoding, callback) {
    const score = computeQualityScore(record);
    if (score >= 0.65) {
      callback(null, { ...record, qualityScore: score });
    } else {
      callback(null);
    }
  }
});

const packStream = new Transform({
  objectMode: true,
  transform(record, _encoding, callback) {
    // In production: tokenize, pack to 4096/8192 context, respect boundaries
    const packed = {
      id: record.id,
      tokens: record.text.length / 4, // placeholder
      domain: record.source,
      quality: record.qualityScore
    };
    callback(null, JSON.stringify(packed) + '\n');
  }
});

// 5. Execution
async function runPipeline(inputPath: string, outputPath: string) {
  const input = createReadStream(inputPath, { encoding: 'utf-8' });
  const lines = createInterface({ input });
  
  const source = Readable.from(lines);
  
  await pipeline(
    source,
    validateStream,
    dedupStream,
    qualityFilterStream,
    packStream,
    createWriteStream(outputPath)
  );
  
  console.log(`Pipeline complete. Dedup set size: ${dedupSet.size}`);
}

// Usage: runPipeline('raw_corpus.jsonl', 'cleaned_packed.jsonl');

Architecture Decisions & Rationale

• Streaming over Batch: Memory scales linearly with dataset size in batch mode. Streaming with backpressure ensures constant memory footprint regardless of corpus size. Failure recovery is shard-based, not corpus-wide.
• Schema Validation First: Catching malformed records early prevents silent corruption downstream. Zod provides runtime type safety without sacrificing performance.
• Stateful Deduplication: Exact hashing catches duplicates across shards. Near-duplicate detection requires LSH or MinHash; the example uses exact hashing for clarity. Production systems use Redis Cluster or RocksDB-backed Bloom filters.
• Adaptive Quality Thresholds: Static thresholds fail across domains. Code requires syntax validity; prose requires readability; technical docs require terminology density. The scorer is modular to allow domain-specific overrides.
• Immutable Output: JSONL with embedded metadata enables deterministic replay. Parquet is preferred for analytics, but JSONL remains the training standard due to tokenizer compatibility.

Pitfall Guide

1. Treating Deduplication as a Binary Switch

Exact deduplication removes legitimate repetitions in code, legal text, and documentation. Near-duplicate thresholds must be domain-aware. Over-deduplication fragments knowledge graphs and reduces sample diversity. Best Practice: Apply exact dedup globally, then fuzzy dedup within domains using adjustable Jaccard/SimHash thresholds. Preserve structural repetitions (e.g., API docs, boilerplate code) if they carry instructional value.

2. Ignoring Tokenization Artifacts

Quality scores computed on raw text diverge from tokenized reality. BPE fragmentation splits words, inflates token counts, and alters entropy calculations. A sentence scoring 0.8 pre-tokenization may score 0.52 post-tokenization due to rare subword sequences. Best Practice: Compute quality metrics after tokenization, or apply tokenization-aware normalization. Track token-to-character ratio per domain to adjust thresholds.

3. Data Leakage Across Splits

Training, validation, and test sets often share near-duplicates due to temporal overlap or source mirroring. Leakage inflates validation metrics and masks overfitting. Best Practice: Run cross-split deduplication before partitioning. Hash documents across all splits. Use temporal cutoffs for time-series data. Audit splits with MinHash locality-sensitive hashing.

4. Static Quality Thresholds

A single quality score threshold fails when mixing domains. Code repositories tolerate lower readability but require syntax validity. Conversational data requires lower toxicity but higher coherence. Best Practice: Implement domain-specific scoring pipelines. Route records through specialized classifiers. Use percentile-based thresholds per domain rather than absolute cutoffs.

5. Neglecting PII & Compliance Boundaries

PII leakage triggers legal liability and model poisoning. Redaction applied post-training is ineffective; the model has already memorized patterns. Best Practice: Integrate NER + regex + LLM-based PII detection in the ingestion stage. Maintain a compliance manifest. Apply differential privacy sampling for sensitive domains.

6. Skipping Data Versioning & Lineage

Unversioned datasets make training runs irreproducible. When a model fails, engineers cannot isolate whether the issue stems from architecture, hyperparameters, or data drift. Best Practice: Hash each dataset snapshot. Store metadata: source URLs, filter versions, timestamp, shard count, domain distribution. Use DVC or Delta Lake for lineage tracking.

7. Over-Reliance on Synthetic Data

Synthetic data fills gaps but introduces distribution collapse if ungrounded. Models trained on LLM-generated text without human verification memorize stylistic artifacts and reasoning shortcuts. Best Practice: Cap synthetic data at 15–20% of total corpus. Anchor synthetic samples to verified human data. Apply reward model filtering and human-in-the-loop validation before injection.

Production Bundle

Action Checklist

• Schema Validation: Enforce strict JSON/JSONL schemas at ingestion; drop malformed records immediately.
• PII & Compliance Scan: Run NER + regex redaction before any downstream processing; log compliance status per shard.
• Cross-Shard Deduplication: Implement exact hashing globally, then fuzzy dedup within domains using configurable thresholds.
• Tokenization-Aware Scoring: Compute quality metrics post-tokenization or apply tokenization-aware normalization factors.
• Domain Balancing: Route high-quality samples to a reservoir; apply stratified sampling to maintain target domain ratios.
• Sequence Packing: Pack to context window limits without crossing document boundaries; preserve attention masks and EOS tokens.
• Immutable Versioning: Hash dataset snapshots; store metadata, filter versions, and domain distributions in a lineage database.

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Low-budget startup (single GPU cluster)	Rule-Based Filtering + Exact Dedup	Minimizes compute overhead while removing obvious noise; fast iteration	Low
Enterprise compliance (finance/healthcare)	Hybrid Curation + PII-First Pipeline + Synthetic Cap	Legal risk mitigation requires strict redaction, domain balancing, and controlled augmentation	Medium-High
Multi-modal expansion (text + code + docs)	Domain-Specific Scoring + Adaptive Thresholds	Prevents domain collapse; ensures code syntax validity and technical terminology density	Medium
Rapid prototyping / research	Raw Scraping + Lightweight Filters	Speed prioritized over precision; acceptable for ablation studies or architecture experiments	Low

Configuration Template

pipeline:
  name: llm-data-prep-v2
  version: "2.1.0"
  output_format: jsonl
  
ingestion:
  sources:
    - type: s3
      bucket: raw-corpus
      prefix: web/
    - type: api
      endpoint: https://api.github.com/repos
      auth_env: GITHUB_TOKEN
  
validation:
  schema: ./schemas/record.schema.json
  drop_invalid: true
  
deduplication:
  exact:
    enabled: true
    algorithm: sha256
  fuzzy:
    enabled: true
    algorithm: simhash
    threshold: 0.85
    domain_aware: true
    
quality:
  scorer: adaptive
  thresholds:
    web: 0.60
    code: 0.65
    docs: 0.70
    api: 0.55
  metrics:
    - language_confidence
    - token_entropy
    - syntax_validity
    - toxicity_probability
    
packing:
  max_tokens: 4096
  respect_boundaries: true
  eos_token: "<|endoftext|>"
  attention_mask: true
  
lineage:
  enabled: true
  storage: s3
  bucket: dataset-lineage
  metadata_fields:
    - source
    - filters_applied
    - domain_distribution
    - timestamp
    - shard_id

Quick Start Guide

1. Install Dependencies: npm install zod @types/node typescript ts-node
2. Initialize Project: Create tsconfig.json with esModuleInterop, strict, and outDir: ./dist. Add package.json script: "start": "ts-node src/pipeline.ts"
3. Configure Sources: Place raw JSONL files in ./data/raw/ or configure S3/API endpoints in the YAML template. Ensure schema matches RecordSchema.
4. Run Pipeline: Execute npm start. Monitor console output for dedup set size, quality distribution, and packed token counts. Verify output in ./data/cleaned_packed.jsonl.
5. Validate Output: Run head -n 5 ./data/cleaned_packed.jsonl | jq . to confirm structure. Cross-check domain distribution with jq -r '.domain' cleaned_packed.jsonl | sort | uniq -c | sort -nr.

Data preparation is the silent determinant of LLM capability. Architect it with the same rigor as your training loop, and compute efficiency will follow.