Tutorial Cleanlab: Meningkatkan Kualitas Data untuk Machine Learning

Kualitas data adalah fondasi dari setiap proyek machine learning yang sukses. Tidak peduli seberapa canggih model yang Anda gunakan, jika data training mengandung label yang salah atau noise yang berlebihan, performa model akan menurun drastis. Cleanlab hadir sebagai solusi open-source yang powerful untuk mendeteksi dan memperbaiki masalah kualitas data secara otomatis.

Dalam tutorial ini, kita akan mempelajari cara menggunakan Cleanlab untuk menemukan label errors, mengidentifikasi outlier, dan membangun dataset yang lebih bersih sehingga model machine learning Anda dapat mencapai performa optimal.

Apa Itu Cleanlab?

Cleanlab adalah library Python open-source yang mengimplementasikan algoritma Confident Learning untuk mendeteksi masalah dalam dataset. Library ini dapat menemukan:

• Label errors: Data yang memiliki label salah dalam dataset training
• Outliers: Data point yang tidak biasa atau anomali
• Near-duplicates: Data yang hampir identik yang bisa menyebabkan data leakage
• Class overlap: Area di mana kelas-kelas saling tumpang tindih

Cleanlab bekerja dengan model machine learning apa pun — scikit-learn, PyTorch, TensorFlow, XGBoost, dan lainnya. Anda cukup menyediakan predicted probabilities dari model Anda, dan Cleanlab akan menganalisis hasilnya.

Instalasi

Instalasi Dasar

pip install cleanlab

Instalasi dengan Dependensi Tambahan

pip install "cleanlab[datalab]"

Instalasi untuk Development

pip install "cleanlab[all]"

Verifikasi Instalasi

import cleanlab
print(cleanlab.version)

Setup Environment

Pastikan Anda memiliki dependensi berikut:

pip install scikit-learn pandas numpy matplotlib seaborn

Basic Usage: Menemukan Label Errors

Contoh 1: Dataset Klasifikasi Sederhana

Mari kita mulai dengan contoh sederhana menggunakan dataset Iris:

import numpy as np
import pandas as pd
from sklearn.datasets import loadiris

from sklearn.modelselection import crossvalpredict
from sklearn.linearmodel import LogisticRegression

from cleanlab.filter import findlabelissues


iris = loadiris()
X = iris.data
y = iris.target

np.random.seed(42)
noisyindices = np.random.choice(len(y), size=15, replace=False)

ynoisy = y.copy()
for idx in noisyindices:

    possiblelabels = [l for l in range(3) if l != y[idx]]
    ynoisy[idx] = np.random.choice(possiblelabels)

print(f"Jumlah label yang sengaja dikacaukan: {len(noisyindices)}")

print(f"Indeks yang dikacaukan: {sorted(noisyindices)}")

Mendapatkan Predicted Probabilities

Cleanlab membutuhkan out-of-sample predicted probabilities. Cara terbaik adalah menggunakan cross-validation:

model = LogisticRegression(maxiter=500, randomstate=42)

predprobs = crossvalpredict(

    model, X, ynoisy,
    cv=5,
    method="predictproba"

)

print(f"Shape predprobs: {predprobs.shape}")

print(f"Contoh probabilitas untuk data pertama: {predprobs[0]}")

Menemukan Label Issues

labelissues = findlabelissues(
    labels=ynoisy,
    predprobs=predprobs,
    returnindicesrankedby="selfconfidence"
)

print(f"\nJumlah label issues yang ditemukan: {len(labelissues)}")

print(f"Indeks yang terdeteksi: {sorted(labelissues)}")

detected = set(labelissues)

actual = set(noisyindices)
truepositives = detected & actual

precision = len(truepositives) / len(detected) if detected else 0
recall = len(truepositives) / len(actual) if actual else 0


print(f"\nPrecision: {precision:.2f}")
print(f"Recall: {recall:.2f}")

from cleanlab.classification import CleanLearning

cl = CleanLearning(
    clf=LogisticRegression(maxiter=500, randomstate=42),

    seed=42
)

cl.fit(X, ynoisy)

labelissuesdf = cl.getlabelissues()
print(labelissuesdf.head(10))

print(f"\nKolom yang tersedia: {labelissuesdf.columns.tolist()}")
print(f"Jumlah label issues: {labelissuesdf['islabelissue'].sum()}")

Membandingkan Performa

from sklearn.modelselection import crossvalscore

baselinemodel = LogisticRegression(maxiter=500, randomstate=42)
baselinescores = crossvalscore(baselinemodel, X, ynoisy, cv=5)

print(f"Akurasi dengan data noisy: {baselinescores.mean():.4f} (+/- {baselinescores.std():.4f})")


cleanmask = ~labelissuesdf["islabelissue"].values
Xclean = X[cleanmask]
yclean = ynoisy[cleanmask]


cleanmodel = LogisticRegression(maxiter=500, randomstate=42)
cleanscores = crossvalscore(cleanmodel, Xclean, yclean, cv=5)
print(f"Akurasi dengan data bersih: {cleanscores.mean():.4f} (+/- {cleanscores.std():.4f})")

Menggunakan Datalab untuk Analisis Komprehensif

from cleanlab import Datalab

lab = Datalab(
    data=pd.DataFrame(X, columns=iris.featurenames),

    labelname="label",
    task="classification"
)

lab.findissues(predprobs=predprobs, features=X)


lab.report()

issuesummary = lab.getissuesummary()
print(issuesummary)


labelissues = lab.getissues("label")

print("\nTop 10 data dengan label issue score terendah:")
print(labelissues.nsmallest(10, "labelscore"))


outlierissues = lab.getissues("outlier")

print("\nTop 10 outlier:")
print(outlierissues.nsmallest(10, "outlierscore"))

duplicateissues = lab.getissues("nearduplicate")
neardupes = duplicateissues[duplicateissues["isnearduplicateissue"]]
print(f"\nJumlah near-duplicates: {len(neardupes)}")


if len(neardupes) > 0:
    for idx in neardupes.index[:5]:

        neardupset = duplicateissues.loc[idx, "nearduplicatesets"]
        score = duplicateissues.loc[idx, "nearduplicatescore"]

        print(f"  Index {idx} (score: {score:.4f}): near-duplicate dengan {neardupset}")

import pandas as pd
from sklearn.featureextraction.text import TfidfVectorizer
from sklearn.naivebayes import MultinomialNB

from sklearn.modelselection import crossvalpredict
from cleanlab import Datalab

texts = [
    "Harga saham naik tajam hari ini",
    "Tim sepakbola memenangkan pertandingan final",
    "Resep masakan nasi goreng spesial",
    "Ekonomi Indonesia tumbuh 5 persen",
    "Pemain basket mencetak rekor baru",
    "Cara membuat kue coklat yang lezat",
    "Bank sentral menaikkan suku bunga",
    "Turnamen tenis dimulai minggu depan",
    "Tips memasak rendang yang empuk",
    "Pasar modal mengalami koreksi",
    "Pertandingan bulu tangkis sangat seru",
    "Tutorial membuat pizza rumahan",
    "Investasi emas semakin diminati",
    "Jadwal liga sepakbola musim baru",
    "Resep smoothie buah segar",
    "Harga saham meningkat drastis",
    "Atlet renang memecahkan rekor dunia",
    "Cara memasak sup ayam kampung",
    "Rupiah menguat terhadap dolar",
    "Kompetisi e-sports semakin populer",
]

labels = [0, 1, 2, 0, 1, 2, 0, 1, 2, 0, 1, 2, 0, 1, 2, 0, 1, 2, 0, 1]
labels[3] = 1   # Sengaja salah: ekonomi -> olahraga
labels[11] = 0  # Sengaja salah: masakan -> bisnis
labels[16] = 2  # Sengaja salah: olahraga -> masakan

labelnames = {0: "bisnis", 1: "olahraga", 2: "masakan"}

df = pd.DataFrame({"text": texts, "label": labels})

vectorizer = TfidfVectorizer(maxfeatures=100)
Xtfidf = vectorizer.fittransform(df["text"]).toarray()

model = MultinomialNB()
predprobs = crossvalpredict(model, Xtfidf, df["label"], cv=3, method="predictproba")


lab = Datalab(data=df, labelname="label", task="classification")
lab.findissues(predprobs=predprobs, features=Xtfidf)
lab.report()

issues = lab.getissues("label")

problematic = issues[issues["islabelissue"]].sortvalues("labelscore")


for idx in problematic.index:
    text = df.loc[idx, "text"]
    givenlabel = labelnames[df.loc[idx, "label"]]

    predicted = labelnames[issues.loc[idx, "predictedlabel"]]

    score = issues.loc[idx, "labelscore"]
    print(f"  [{idx}] '{text}'")
    print(f"       Label: {givenlabel} -> Prediksi: {predicted} (score: {score:.4f})")

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
from sklearn.modelselection import StratifiedKFold
from cleanlab.filter import findlabelissues
import numpy as np

class SimpleClassifier(nn.Module):
    def init(self, inputdim, numclasses):
        super().init()
        self.network = nn.Sequential(
            nn.Linear(inputdim, 128),

            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(64, numclasses)
        )

    def forward(self, x):
        return self.network(x)

def getcrossvalpredprobs(X, y, nsplits=5, epochs=50):

    """Mendapatkan out-of-sample predicted probabilities menggunakan cross-validation."""
    skf = StratifiedKFold(nsplits=nsplits, shuffle=True, randomstate=42)
    predprobs = np.zeros((len(y), len(np.unique(y))))


    for fold, (trainidx, validx) in enumerate(skf.split(X, y)):

        Xtrain = torch.FloatTensor(X[trainidx])

        ytrain = torch.LongTensor(y[trainidx])

        Xval = torch.FloatTensor(X[validx])


        traindataset = TensorDataset(Xtrain, ytrain)
        trainloader = DataLoader(traindataset, batchsize=32, shuffle=True)


        model = SimpleClassifier(X.shape[1], len(np.unique(y)))
        optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
        criterion = nn.CrossEntropyLoss()

        model.train()
        for epoch in range(epochs):
            for batchX, batchy in trainloader:
                optimizer.zerograd()

                outputs = model(batchX)
                loss = criterion(outputs, batchy)

                loss.backward()
                optimizer.step()

        model.eval()
        with torch.nograd():
            valoutputs = model(Xval)
            valprobs = torch.softmax(valoutputs, dim=1).numpy()
            predprobs[validx] = valprobs


        print(f"  Fold {fold+1}/{nsplits} selesai")

    return predprobs


from sklearn.datasets import makeclassification

X, y = makeclassification(

    nsamples=1000, nfeatures=20, ninformative=15,
    nclasses=5, nclustersperclass=1, randomstate=42

)

noiserate = 0.1
nnoisy = int(len(y)  noiserate)

noisyidx = np.random.choice(len(y), size=nnoisy, replace=False)
ynoisy = y.copy()

for idx in noisyidx:
    possible = [l for l in range(5) if l != y[idx]]
    ynoisy[idx] = np.random.choice(possible)


print("Menjalankan cross-validation untuk mendapatkan predicted probabilities...")
predprobs = getcrossvalpredprobs(X, ynoisy, nsplits=5, epochs=30)

issues = findlabelissues(
    labels=ynoisy,

    predprobs=predprobs,

    returnindicesrankedby="selfconfidence"

)

print(f"\nDitemukan {len(issues)} label issues")
print(f"Label noisy yang sebenarnya: {len(noisyidx)}")

overlap = len(set(issues[:nnoisy]) & set(noisyidx))
print(f"True positives (dalam top-{nnoisy}): {overlap}")

print(f"Recall: {overlap/nnoisy:.2%}")

import xgboost as xgb
from sklearn.modelselection import crossvalpredict

from cleanlab.classification import CleanLearning
from sklearn.datasets import loadbreastcancer


data = loadbreastcancer()

X, y = data.data, data.target

np.random.seed(42)
noiseidx = np.random.choice(len(y), size=30, replace=False)
ynoisy = y.copy()

ynoisy[noiseidx] = 1 - ynoisy[noiseidx]


xgbmodel = xgb.XGBClassifier(
    nestimators=100,

    maxdepth=3,
    learningrate=0.1,

    randomstate=42,
    evalmetric="logloss"

)

cl = CleanLearning(clf=xgbmodel, seed=42)
cl.fit(X, ynoisy)


labelissues = cl.getlabelissues()
detected = labelissues[labelissues["islabelissue"]].index.tolist()

print(f"Terdeteksi {len(detected)} label issues")
print(f"Yang benar-benar noisy: {len(noiseidx)}")


tp = len(set(detected) & set(noiseidx))
precision = tp / len(detected) if detected else 0
recall = tp / len(noiseidx)

print(f"Precision: {precision:.2%}")
print(f"Recall: {recall:.2%}")

from cleanlab.count import computeconfidentjoint, estimatelatent

confidentjoint = computeconfidentjoint(

    labels=ynoisy,
    predprobs=predprobs
)

print("Confident Joint Matrix:")
print(confidentjoint)


py, noisematrix, invnoisematrix = estimatelatent(

    confidentjoint=confidentjoint,

    labels=ynoisy
)

print("\nNoise Matrix (P(noisy label | true label)):")
print(np.round(noisematrix, 3))

print("\nInverse Noise Matrix (P(true label | noisy label)):")
print(np.round(invnoisematrix, 3))

from cleanlab.rank import getlabelqualityscores

qualityscores = getlabelqualityscores(
    labels=ynoisy,

    predprobs=predprobs,

    method="selfconfidence"
)

print(f"Mean quality score: {qualityscores.mean():.4f}")

print(f"Min quality score: {qualityscores.min():.4f}")
print(f"Max quality score: {qualityscores.max():.4f}")


import matplotlib.pyplot as plt

fig, axes = plt.subplots(1, 2, figsize=(14, 5))

axes[0].hist(qualityscores, bins=50, edgecolor="black", alpha=0.7)
axes[0].setxlabel("Label Quality Score")

axes[0].setylabel("Frekuensi")
axes[0].settitle("Distribusi Label Quality Scores")

axes[0].axvline(x=0.5, color="red", linestyle="--", label="Threshold 0.5")
axes[0].legend()

cleanscores = qualityscores[~np.isin(np.arange(len(ynoisy)), noisyidx)]

noisyscores = qualityscores[noisyidx]

axes[1].hist(cleanscores, bins=30, alpha=0.6, label="Clean labels", color="green")

axes[1].hist(noisyscores, bins=30, alpha=0.6, label="Noisy labels", color="red")
axes[1].setxlabel("Label Quality Score")

axes[1].setylabel("Frekuensi")
axes[1].settitle("Quality Score: Clean vs Noisy Labels")

axes[1].legend()

plt.tightlayout()
plt.savefig("labelqualitydistribution.png", dpi=150)
plt.show()

from cleanlab.multilabelclassification.filter import findlabelissues as findmultilabelissues
from sklearn.multiclass import OneVsRestClassifier
from sklearn.linearmodel import LogisticRegression
from sklearn.preprocessing import MultiLabelBinarizer

np.random.seed(42)
nsamples = 500

nfeatures = 20
nclasses = 4


X = np.random.randn(nsamples, nfeatures)


truelabels = []
for i in range(nsamples):

    nlabels = np.random.randint(1, 3)
    labels = sorted(np.random.choice(nclasses, size=nlabels, replace=False).tolist())
    truelabels.append(labels)


mlb = MultiLabelBinarizer()
ybinary = mlb.fittransform(truelabels)

noiseindices = np.random.choice(nsamples, size=50, replace=False)
ynoisybinary = ybinary.copy()

for idx in noiseindices:
    flipcol = np.random.randint(nclasses)
    ynoisybinary[idx, flipcol] = 1 - ynoisybinary[idx, flipcol]

model = OneVsRestClassifier(LogisticRegression(maxiter=500, randomstate=42))
predprobsmultilabel = crossvalpredict(model, X, ynoisybinary, cv=5, method="predictproba")


issues = findmultilabelissues(

    labels=list(mlb.inversetransform(ynoisybinary)),
    predprobs=[predprobsmultilabel[:, i] for i in range(nclasses)],
)

print(f"Ditemukan {issues.sum()} label issues dalam multi-label setting")

# BENAR: Cross-validation
predprobs = crossvalpredict(model, X, y, cv=5, method="predictproba")

SALAH: In-sample predictions
model.fit(X, y)
predprobs = model.predictproba(X)  # Overfit, hasil tidak reliable

from sklearn.ensemble import GradientBoostingClassifier

model = GradientBoostingClassifier(
    nestimators=200,

    maxdepth=5,
    learningrate=0.1,

    randomstate=42
)
predprobs = crossvalpredict(model, X, y, cv=5, method="predictproba")

from cleanlab.filter import findlabelissues

issues = findlabelissues(
    labels=y, predprobs=predprobs,
    returnindicesrankedby="selfconfidence"
)

print("Top 20 data paling mencurigakan:")
for rank, idx in enumerate(issues[:20], 1):
    print(f"  #{rank}: Index {idx}")
    print(f"    Label saat ini: {y[idx]}")
    print(f"    Predicted probs: {predprobs[idx]}")

    print(f"    Suggested label: {np.argmax(predprobs[idx])}")
    print()

def iterativecleaning(X, y, model, maxiterations=3):
    """Melakukan pembersihan data secara iteratif."""
    currentX = X.copy()

    currenty = y.copy()
    totalremoved = 0


    for iteration in range(maxiterations):
        predprobs = crossvalpredict(

            model, currentX, currenty, cv=5, method="predictproba"
        )

        issues = findlabelissues(
            labels=currenty, predprobs=predprobs,

            returnindicesrankedby="selfconfidence"

        )

        if len(issues) == 0:
            print(f"Iterasi {iteration+1}: Tidak ada lagi label issues")
            break

        print(f"Iterasi {iteration+1}: Ditemukan {len(issues)} label issues")
        totalremoved += len(issues)

        mask = np.ones(len(currenty), dtype=bool)

        mask[issues] = False
        currentX = currentX[mask]

        currenty = currenty[mask]


    print(f"\nTotal data yang dihapus: {totalremoved}")
    print(f"Ukuran dataset akhir: {len(currenty)}")

    return currentX, currenty

# Metode paling konservatif (false positive rendah)
issuesprune = findlabelissues(
    labels=y, predprobs=predprobs,
    filterby="prunebyclass"

)

Metode paling agresif (recall tinggi)
issuesnoise = findlabelissues(
    labels=y, predprobs=predprobs,
    filterby="prunebynoiserate"
)

Metode balanced (default)
issuesboth = findlabelissues(

    labels=y, predprobs=predprobs,

    filterby="both"
)

Metode confident learning
issuescl = findlabelissues(

    labels=y, predprobs=predprobs,

    filterby="confidentlearning"

)

print(f"prunebyclass: {len(issuesprune)} issues")
print(f"prunebynoiserate: {len(issuesnoise)} issues")
print(f"both: {len(issuesboth)} issues")

print(f"confidentlearning: {len(issuescl)} issues")

import numpy as np
import pandas as pd
from sklearn.modelselection import crossvalpredict, crossvalscore
from sklearn.ensemble import GradientBoostingClassifier
from cleanlab import Datalab
from cleanlab.classification import CleanLearning

def dataqualitypipeline(X, y, featurenames=None):

    """Pipeline lengkap untuk analisis dan perbaikan kualitas data."""

    if featurenames is None:
        featurenames = [f"feature{i}" for i in range(X.shape[1])]

    df = pd.DataFrame(X, columns=featurenames)

    df["label"] = y

    print("="  60)
    print("STEP 1: Baseline Performance")
    print("="  60)


    model = GradientBoostingClassifier(nestimators=100, randomstate=42)

    baselinescores = crossvalscore(model, X, y, cv=5, scoring="accuracy")
    print(f"Baseline Accuracy: {baselinescores.mean():.4f} (+/- {baselinescores.std():.4f})")

    print("\n" + "="  60)
    print("STEP 2: Analisis Kualitas Data dengan Datalab")
    print("="  60)


    predprobs = crossvalpredict(model, X, y, cv=5, method="predictproba")
    lab = Datalab(data=df, labelname="label", task="classification")

    lab.findissues(predprobs=predprobs, features=X)
    lab.report()

    print("\n" + "="  60)
    print("STEP 3: Membersihkan Data dan Re-training")
    print("="  60)


    cl = CleanLearning(
        clf=GradientBoostingClassifier(nestimators=100, randomstate=42),
        seed=42
    )
    cl.fit(X, y)

    issuesdf = cl.getlabelissues()

    nissues = issuesdf["islabelissue"].sum()

    print(f"Label issues ditemukan: {nissues} ({nissues/len(y)100:.1f}%)")

    cleanmask = ~issuesdf["islabelissue"].values

    Xclean = X[cleanmask]

    yclean = y[cleanmask]


    cleanscores = crossvalscore(model, Xclean, yclean, cv=5, scoring="accuracy")
    print(f"Clean Accuracy: {cleanscores.mean():.4f} (+/- {cleanscores.std():.4f})")
    print(f"Improvement: {(cleanscores.mean() - baselinescores.mean())100:.2f}%")


    return {
        "baselineaccuracy": baselinescores.mean(),
        "cleanaccuracy": cleanscores.mean(),
        "nissues": nissues,
        "issuesdf": issuesdf,
        "Xclean": Xclean,
        "yclean": yclean
    }

from sklearn.datasets import loaddigits


digits = loaddigits()
X, y = digits.data, digits.target

np.random.seed(42)
nnoisy = int(0.05  len(y))
noisyidx = np.random.choice(len(y), size=nnoisy, replace=False)

ynoisy = y.copy()
for idx in noisyidx:

    ynoisy[idx] = np.random.choice([l for l in range(10) if l != y[idx]])

results = dataqualitypipeline(X, ynoisy, featurenames=[f"pixel_{i}" for i in range(64)])

Kesimpulan

Cleanlab adalah tool yang sangat berguna dalam toolkit data scientist modern. Dengan mengimplementasikan Confident Learning, Cleanlab memungkinkan Anda untuk:

Kunci keberhasilan menggunakan Cleanlab:

• Selalu gunakan out-of-sample predictions (cross-validation)
• Gunakan model yang cukup baik untuk menghasilkan predicted probabilities
• Verifikasi temuan secara manual, terutama untuk kasus yang ambigu
• Lakukan pembersihan secara iteratif untuk hasil yang optimal

Dengan data yang lebih bersih, model machine learning Anda akan lebih akurat, lebih robust, dan lebih dapat diandalkan di production. Ingat: garbage in, garbage out — dan Cleanlab membantu Anda memastikan bahwa yang masuk ke model Anda bukanlah garbage.