NeMo Curator を使った日本語データのキュレーション – モデルベース編

mkdir curator-example
cd curator-example

sudo docker run --rm -it --gpus all --shm-size=64g --ulimit memlock=-1 --network=host -v ${PWD}:/workspace -w /workspace nvcr.io/nvidia/nemo:25.04 bash

mkdir -p data && cd data && wget -qO- --no-check-certificate "https://gitlab.llm-jp.nii.ac.jp/datasets/llm-jp-corpus-v4/-/raw/main/ja/ja_fineweb-2/0000.jsonl.gz?ref_type=heads" | gunzip > 0000.jsonl && cd ..

git lfs install
git clone https://huggingface.co/tokyotech-llm/edu-classifier

import argparse

import fasttext
import numpy as np
import pandas as pd

import nemo_curator as nc
from nemo_curator.datasets import DocumentDataset
from nemo_curator.filters.doc_filter import DocumentFilter
from nemo_curator.utils.decorators import batched
from nemo_curator.utils.distributed_utils import (
    get_client,
    read_data,
    write_to_disk,
    NoWorkerError,
    load_object_on_worker,
)
from nemo_curator.utils.file_utils import get_all_files_paths_under
from nemo_curator.utils.script_utils import ArgumentHelper


def load_dataset(input_data_dir: str) -> DocumentDataset:
    files = list(get_all_files_paths_under(input_data_dir, keep_extensions="jsonl"))
    raw_data = read_data(files, file_type="jsonl", backend="pandas", add_filename=True)
    return DocumentDataset(raw_data)


class LLMFastTextQualityFilter(DocumentFilter):
    def __init__(
        self,
        model_path: str | None = None,
        min_cutoff: float = -1.0,
    ):
        if model_path is None:
            msg = "Must provide a valid path to a FastText model to compute document scores with this filter"
            raise ValueError(msg)
        self._model_path = model_path
        self._cutoff = min_cutoff
        self._name = "fasttext_quality_filter"

    @batched
    def score_document(self, df: pd.Series) -> pd.Series:
        model_attr = f"{self._name}_{self._model_path}"
        try:
            model = load_object_on_worker(model_attr, self._load_model, {})
        except NoWorkerError:
            return pd.Series(np.ones(len(df)), dtype=float)

        def _score_document(text: str) -> float:
            text = text.replace("\n", " ").replace("__label__", " ")
            pred = model.predict(text, k=-1)
            return sum([int(label[-1]) * prob for label, prob in zip(pred[0], pred[1])])

        return df.apply(_score_document)

    @batched
    def keep_document(self, df: pd.Series) -> pd.Series:
        return df > np.full(len(df), self._cutoff)  # noqa: NPY002

    def _load_model(self) -> fasttext.FastText:
        return fasttext.load_model(self._model_path)


def main(args: argparse.Namespace) -> None:
    # Params
    model_path = "edu-classifier/llm_llama.bin"  # "edu-classifier/wiki.bin"
    data_path = "data/"
    filtered_output = "result/0000_scored.jsonl"

    # Prepare samples for the classifier
    client = get_client(**ArgumentHelper.parse_client_args(args), memory_limit="32GB")

    print("Loading dataset...")
    # Filter data
    target_dataset = load_dataset(data_path)
    print(f"Dataset loaded: {len(target_dataset.df)} rows")
    filter_pipeline = nc.ScoreFilter(
        LLMFastTextQualityFilter(model_path, min_cutoff=-1.0),
        score_field="quality_score",
        score_type=float,
    )
    print("Applying filter...")
    filtered_dataset = filter_pipeline(target_dataset)
    print("Writing filtered dataset...")
    write_to_disk(filtered_dataset.df, filtered_output, write_to_filename=True)
    print("Dataset written successfully!")

    client.close()


def attach_args(
    parser: argparse.ArgumentParser,
) -> argparse.ArgumentParser:
    return ArgumentHelper(parser).add_distributed_args()


if __name__ == "__main__":
    main(
        attach_args(
            argparse.ArgumentParser(
                formatter_class=argparse.ArgumentDefaultsHelpFormatter
            )
        ).parse_args()
    )

mkdir result
python classifier_filtering.py 

jq -s 'max_by(.quality_score)' result/0000_scored.jsonl

"text": "妊婦さんや母乳育児中のママさんがよく気になるのが\"カフェインの摂取\"。もともとコーヒーを飲むのが習慣になっている妊婦さんの中にも、「妊娠中はコーヒーが飲めないのが辛い…」という方が少なくありません。\nそこで今回の記事では、妊婦さんでも安心なカフェインレスコーヒーの選び方について、詳しくご紹介したいと思います。
=== 省略 ===
\n今回の記事を参考にしながら選べば、安心で味もおいしいカフェインレスコーヒーが見つけられるので、妊娠&授乳中でもリラックスできるコーヒータイムを楽しみましょう。",
"quality_score": 2.9894943333

'Adult', 'Arts_and_Entertainment', 'Autos_and_Vehicles', 'Beauty_and_Fitness', 'Books_and_Literature', 'Business_and_Industrial', 'Computers_and_Electronics', 'Finance', 'Food_and_Drink', 'Games', 'Health', 'Hobbies_and_Leisure', 'Home_and_Garden', 'Internet_and_Telecom', 'Jobs_and_Education', 'Law_and_Government', 'News', 'Online_Communities', 'People_and_Society', 'Pets_and_Animals', 'Real_Estate', 'Science', 'Sensitive_Subjects', 'Shopping', 'Sports', 'Travel_and_Transportation'

import argparse
import time

from nemo_curator.classifiers import MultilingualDomainClassifier
from nemo_curator.datasets import DocumentDataset
from nemo_curator.utils.distributed_utils import get_client
from nemo_curator.utils.script_utils import ArgumentHelper


def main(args: argparse.Namespace) -> None:
    global_st = time.time()

    # Input can be a string or list
    input_file_path = "data/0000.jsonl"
    output_file_path = "result/0000_domain_sports.jsonl"

    client_args = ArgumentHelper.parse_client_args(args)
    client_args["cluster_type"] = "gpu"
    client = get_client(**client_args)

    input_dataset = DocumentDataset.read_json(
        input_file_path, backend="cudf", add_filename=True
    )

    multilingual_domain_classifier = MultilingualDomainClassifier(
        filter_by=["Sports"]
    )
    result_dataset = multilingual_domain_classifier(dataset=input_dataset)

    result_dataset.to_json(output_path=output_file_path, write_to_filename=True)

    global_et = time.time()
    print(
        f"Total time taken for multilingual domain classifier inference: {global_et - global_st} s",
        flush=True,
    )

    client.close()


def attach_args(
    parser: argparse.ArgumentParser,
) -> argparse.ArgumentParser:
    arg_helper = ArgumentHelper(parser)
    arg_helper.add_distributed_classifier_cluster_args()

    return arg_helper.parser


if __name__ == "__main__":
    main(
        attach_args(
            argparse.ArgumentParser(
                formatter_class=argparse.ArgumentDefaultsHelpFormatter
            )
        ).parse_args()
    )

head -n1 result/0000_domain_sports.jsonl

"text":"スポーツをしていてこんなお悩みはありませんか?\n- もっと力強く動きたい\n- もっと素早く動きたい\n- フォームをきれいに改善したい\n- もっとボールを飛ばしたい\n- もっと持久力を上げたい\n- もっと体を柔軟にしたい\nこのような願望を叶えてくれるメソッド(技法)があります。
=== 省略 ===
\nあなたにはもっとスポーツを楽しんで、強くなっていただきたい、それが私の希望です。そのためにこの技術を大いに活用し、出し惜しみしないことをお約束します。","domain_pred":"Sports"

[
    "medical_record_number",
    "location",
    "address",
    "date_of_birth",
    "date_time",
    "name",
    "email",
    ...
]

import pandas as pd
import textwrap

from nemo_curator.datasets import DocumentDataset
from nemo_curator.modifiers.async_llm_pii_modifier import AsyncLLMPiiModifier
from nemo_curator.modules.modify import Modify
from nemo_curator.utils.distributed_utils import get_client
import multiprocessing as mp

def main():
    # 準備: Daskクライアント（分散実行／スケールアウト用、ローカルでもOK）
    client = get_client(cluster_type="gpu", n_workers=8)
    
    # LLMの出力を機械可読に保つため、返却形式を固定化
    JSON_SCHEMA = {
        "type": "array",
        "items": {
            "type": "object",
            "required": ["entity_type", "entity_text"],
            "properties": {"entity_type": {"type": "string"}, "entity_text": {"type": "string"}},
        },
    }
    
    # PIIエンティティを明示
    EVAL_LABELS = ["name","email","address","phone_number","employee_id","location","customer_id","ip_address","password","credit_card_number","api_key"]
    
    # ​​システムプロンプト: 役割指示（編集者）/ 出力制約（JSON）/ 非正規化（原文コピー）を明記
    # スキーマ/ラベルを文字列で埋め込み、LLM側の解釈余地を最小化
    SYSTEM_PROMPT = textwrap.dedent(f"""
        あなたは熟練の編集者です。ユーザーがテキストを提供しますので、
        その中から個人を特定できる情報（PII）を見つけてください。
    
        重要: 出力は必ず次のJSONスキーマに従ってください:
        {JSON_SCHEMA}
    
        厳守事項:
        - 実際に入力テキストに含まれている情報のみを返してください（捏造禁止）
        - entity_text は入力から正確にコピーし、変更・正規化をしてはいけません
        - entity_type は次のいずれかに限定します: {", ".join(EVAL_LABELS)}
    """ ).strip()
    
    # Input: テスト用ダミーデータの準備
    rows = [
        {
            "text": ("木下大地（ユーザー名: daichi.k、顧客ID: C23456）は、自宅のIPアドレス（172.16.0.51）からWebサービスにアクセスし、新しいパスワード「SafePass2025」に変更した後、アカウント情報を最新のものに更新しました。")
        },
        {
            "text": ("石川直樹（社員番号: EMP-33322、クレジットカード番号: 4000-1111-2222-3333）は、2025年6月18日午後3時ごろ、会社の専用システムにAPIキー「APIKEY-ISH2025」を入力し、オンライン経費精算用にアカウントにログインしました。その際、本人確認のため追加の認証も行いました。")
        },
    ]
    
    dataframe = pd.DataFrame(rows, columns=["text"]).reset_index(drop=False).rename(columns={"index": "id"})
    
    dataset = DocumentDataset.from_pandas(dataframe, npartitions=1)
    
    # LLMモディファイアの設定（非同期）
    modifier = AsyncLLMPiiModifier(
        # Endpoint for the user's NIM
        base_url="https://integrate.api.nvidia.com/v1",
        api_key="<your_nvidia_api_key>",  # 取得したAPIキーを入力してください
        model="tokyotech-llm/llama-3-swallow-70b-instruct-v0.1",
        system_prompt = SYSTEM_PROMPT,
        language="jpn",
        max_concurrent_requests=10,
    )
    
    modify = Modify(modifier)
    modified_dataset = modify(dataset)
    
    # Output
    modified_dataset.to_json("output.jsonl")
    client.close()


if __name__ == "__main__":
    mp.freeze_support()
    main()

# output.jsonl

{"id":0,"text":"{{name}}（ユーザー名: daichi.k、顧客ID: {{customer_id}}）は、自宅のIPアドレス（{{ip_address}}）からWebサービスにアクセスし、新しいパスワード「{{password}}」に変更した後、アカウント情報を最新のものに更新しました。"}

{"id":1,"text":"{{name}}（社員番号: {{employee_id}}、クレジットカード番号: {{credit_card_number}}）は、2025年6月18日午後3時ごろ、会社の専用システムにAPIキー「{{api_key}}」を入力し、オンライン経費精算用にアカウントにログインしました。その際、本人確認のため追加の認証も行いました。"}