Heuristic Filtering

1
from nemo_curator.pipeline import Pipeline
2
from nemo_curator.stages.text.io.reader import JsonlReader
3
from nemo_curator.stages.text.io.writer import JsonlWriter
4
from nemo_curator.stages.text.modules import ScoreFilter
5
from nemo_curator.stages.text.filters import (
6
    WordCountFilter,
7
    RepeatingTopNGramsFilter,
8
    PunctuationFilter
9
)
10
11
# Create pipeline
12
pipeline = Pipeline(name="heuristic_filtering")
13
14
# Load your dataset
15
reader = JsonlReader(
16
    file_paths="input_data/",
17
    fields=["text", "id"]
18
)
19
pipeline.add_stage(reader)
20
21
# Add filter stages
22
pipeline.add_stage(ScoreFilter(
23
    filter_obj=WordCountFilter(min_words=80),
24
    text_field="text",
25
    score_field="word_count"
26
))
27
pipeline.add_stage(ScoreFilter(
28
    filter_obj=PunctuationFilter(max_num_sentences_without_endmark_ratio=0.85),
29
    text_field="text"
30
))
31
pipeline.add_stage(ScoreFilter(
32
    filter_obj=RepeatingTopNGramsFilter(n=2, max_repeating_ngram_ratio=0.2),
33
    text_field="text"
34
))
35
pipeline.add_stage(ScoreFilter(
36
    filter_obj=RepeatingTopNGramsFilter(n=3, max_repeating_ngram_ratio=0.18),
37
    text_field="text"
38
))
39
pipeline.add_stage(ScoreFilter(
40
    filter_obj=RepeatingTopNGramsFilter(n=4, max_repeating_ngram_ratio=0.16),
41
    text_field="text"
42
))
43
44
# Add output stage
45
writer = JsonlWriter(path="high_quality_output/")
46
pipeline.add_stage(writer)
47
48
# Execute pipeline
49
results = pipeline.run()

1
# Efficient ordering - place fast filters first
2
from nemo_curator.pipeline import Pipeline
3
from nemo_curator.stages.text.modules import ScoreFilter
4
from nemo_curator.stages.text.filters import WordCountFilter, UrlsFilter, RepeatingTopNGramsFilter
5
6
pipeline = Pipeline(name="efficient_filtering")
7
# Fast filters first
8
pipeline.add_stage(ScoreFilter(filter_obj=WordCountFilter(min_words=50), text_field="text"))
9
# Medium complexity filters
10
pipeline.add_stage(ScoreFilter(filter_obj=UrlsFilter(), text_field="text"))
11
# Slow filters last
12
pipeline.add_stage(ScoreFilter(filter_obj=RepeatingTopNGramsFilter(), text_field="text"))

1
import pandas as pd
2
3
# Load scores from filter run
4
scores = pd.read_json("output/scores/scores.jsonl", lines=True)
5
6
# Analyze rejection reasons
7
rejection_counts = scores[scores["rejected"] == True].groupby("rejected_by").size()
8
print(f"Documents rejected by filter:\n{rejection_counts}")
9
10
# Analyze score distributions
11
import matplotlib.pyplot as plt
12
scores.hist(column="word_count", bins=50)
13
plt.title("Word Count Distribution")
14
plt.savefig("word_count_hist.png")

1
# Process large datasets efficiently using pipeline streaming
2
3
# Configure for streaming processing
4
executor_config = {
5
    "execution_mode": "streaming",
6
    "cpu_allocation_percentage": 0.8,
7
    "logging_interval": 60
8
}
9
10
# Use custom configuration for large datasets
11
executor = XennaExecutor(config=executor_config)
12
results = pipeline.run(executor)
13
14
# Default configuration works for most cases
15
# results = pipeline.run()