Claude Codeで学習データの前処理を検証する — panderaでJSONLのスキーマを固定

学習データのJSONLは、1行ずつ目で見るには多すぎ、壊れ方は静かです。同じ行が2回入る、必須の項目がnullになる、メールアドレスが回答に混ざる。どれも学習を回すまでエラーになりません。

この記事では、検証ルールをpanderaのスキーマとpytestに固定し、Claude Codeには「通るまで直す」役を任せます。スクリプトを書かせること自体より、合否が機械的に返る検査を先に置くことが要点です。コードは手元でpandera 0.35.0・Python 3.12.2を使って動かし、出力も載せています。

Claude Codeに渡すのは「合否が返る検査」

Claude Codeは、作業が終わったように見えた時点で止まります。実際に終わったかを判断する材料が無ければ、「終わったように見える」しか手がかりがありません。ベストプラクティスのページはこう書いたうえで、合否が返る何かを渡すよう勧めています。テストファイル、リンター、出力をフィクスチャと突き合わせるスクリプトなどです。

学習データの前処理は、この条件に向いています。正しいデータの定義を、列ごとの型・許す値・重複の扱いとして書き出せるからです。panderaはその定義をDataFrameの検証ルールとして持てるライブラリで、スキーマを一度書けば、pandasのほか、polarsやpysparkなど別のDataFrame種別にも使い回せます。

役割分担は次のとおりです。

重複除去のような処理そのものは、pandas全般の使い方としてClaude Codeでpandasを回す手順が扱っています。ここでは、処理の後に残す検査に絞ります。

JSONLの検証スキーマをDataFrameModelで書く

題材は、問い合わせ対応のやり取りを集めた学習データです。1行が1レコードで、id・prompt・response・sourceの4項目を持ちます。

{"id":"b1","prompt":"返品方法は?","response":"購入から14日以内に申請してください。","source":"manual"}

pandera 0.24.0以降、pandas向けのスキーマはpandera.pandasから読み込むのが推奨です。トップレベルのpanderaからDataFrameSchemaなどを使うとFutureWarningが出て、0.29.0で非推奨になる予定だと案内されています。インストールはpip install 'pandera[pandas]'です。

python -m venv .venv && source .venv/bin/activate
pip install 'pandera[pandas]' pytest

スキーマは、クラスで書くDataFrameModelにします。pydanticに近い書き方で、列の定義とデータ全体にかかるルールが1か所にまとまり、Claude Codeが差分を読みやすいからです。

import re
import unicodedata
 
import pandas as pd
import pandera.pandas as pa
 
EMAIL = r"[\w.+-]+@[\w-]+\.[\w.-]+"
PHONE = r"0\d{1,4}-\d{1,4}-\d{3,4}"
PII = f"(?:{EMAIL})|(?:{PHONE})"
 
 
def normalize(text: str) -> str:
    t = unicodedata.normalize("NFKC", text)
    return re.sub(r"\s+", " ", t).strip().lower()
 
 
class TrainRecord(pa.DataFrameModel):
    id: str = pa.Field(unique=True)
    prompt: str = pa.Field(str_length={"min_value": 1})
    response: str = pa.Field(str_length={"min_value": 1})
    source: str = pa.Field(isin=["manual", "synthetic", "crawl"])
 
    class Config:
        strict = True
        coerce = True
        unique = ["prompt", "response"]
 
    @pa.dataframe_check
    def no_near_duplicate(cls, df: pd.DataFrame) -> pd.Series:
        p = df["prompt"].fillna("")
        r = df["response"].fillna("")
        key = (p + "\n" + r).map(normalize)
        return ~key.duplicated(keep="first")
 
    @pa.check("prompt", "response", name="no_pii")
    def no_pii(cls, s: pd.Series) -> pd.Series:
        return ~s.fillna("").str.contains(PII, regex=True)
 
 
def load(path: str) -> pd.DataFrame:
    return pd.read_json(path, lines=True, dtype=False)

押さえる点は3つです。

• 列は既定でnullを許しません。許したい列にだけnullable=Trueを付けます。欠損の検査にコードは要らず、書かないことが検査になります
• strict=Trueはスキーマに無い列を落とします。データの出力側で列が増えた事故を拾えます
• uniqueに列名のリストを渡すと、その組み合わせの重複を検査します

完全一致と表記ゆれの2段で重複を見る

重複は2段で見ます。1段目はpromptとresponseの組が同じ行で、スキーマのuniqueが受け持ちます。2段目は、全角と半角、大文字と小文字、空白の連続といった違いだけで中身が同じ行です。こちらはnormalizeで正規化した文字列をキーに、dataframe_checkで拾います。

壊れた行を混ぜたsample.jsonlを用意し、実行します。

{"id":"a1","prompt":"返品方法は?","response":"購入から14日以内に申請してください。","source":"manual"}
{"id":"a2","prompt":"返品方法は?","response":"購入から14日以内に申請してください。","source":"manual"}
{"id":"a3","prompt":"返品　方法は?","response":"購入から14日以内に 申請してください。","source":"crawl"}
{"id":"a4","prompt":"連絡先は?","response":"taro@example.com か 03-1234-5678 へ。","source":"synthetic"}
{"id":"a5","prompt":"営業時間は?","response":null,"source":"manual"}
{"id":"a5","prompt":"定休日は?","response":"水曜です。","source":"web"}

uniqueの違反は、重複した2行がどちらも報告されました。解説ページの本文は、重複の報告を最初の1件を除く形が既定だと書いています。一方、同じページのスキーマ表示はreport_duplicates=allです。手元の0.35.0で既定のまま実行すると、先頭の行を含めて全行が出ました。報告に出る行を基準に「どれを残すか」を決めるスクリプトを書かせるなら、report_duplicatesを明示して挙動を固定します。

2段目のno_near_duplicateは、a1と中身が同じa2だけを報告しました。a3は拾えていません。promptの「返品　方法は?」はNFKCで全角空白が半角になるものの、空白そのものは残るため、「返品方法は?」とは別の文字列になるからです。

つまり、このチェックが見るのは正規化後に一致する行までです。語順の違いや言い換えを含む近似重複は、別の手段(文字n-gramのMinHashなど)を、チェック関数の中から呼ぶ形で足す領域になります。Claude Codeに「近似重複も除去して」とだけ頼むと、どこまでを近似とみなしたかが曖昧なまま通ります。正規化の規則をスキーマ側に書いておけば、その規則がレビュー対象になります。

欠損・不正値・個人情報を1回の実行で全部読む

validateにlazy=Trueを渡すと、最初の違反で止まらず、すべてのルールを走らせてから例外を投げます。エラーはSCHEMA(列の有無・型・null)とDATA(値のチェック)に分けて集約されます。

try:
    TrainRecord.validate(df, lazy=True)
except pa.errors.SchemaErrors as exc:
    print(json.dumps(exc.message, ensure_ascii=False, indent=2))
    print(exc.failure_cases[["column", "check", "index"]])

sample.jsonlに対する出力を、見やすく抜粋します。

DATA / DUPLICATES                 multiple_fields_uniqueness  (prompt, response)
DATA / SERIES_CONTAINS_DUPLICATES field_uniqueness            id: a5 が2回
DATA / DATAFRAME_CHECK            no_pii                      response: taro@example.com か ...
DATA / DATAFRAME_CHECK            isin([...])                 source: web
DATA / DATAFRAME_CHECK            no_near_duplicate           行1 (a2)
SCHEMA / SERIES_CONTAINS_NULLS    not_nullable                response: 行4が欠損

5種類の問題が1回の実行で出そろいました。failure_casesはDataFrameで、column・check・indexを持つため、行番号を頼りに該当レコードを開けます。Claude Codeに渡す出力としても、このほうが1件ずつ直す往復より速く収束します。

個人情報のチェックは、メールアドレスと日本の固定電話の書式だけを見る簡易なものです。氏名や住所、携帯電話の番号は拾いません。PIIの正規表現は、扱うデータの種類に合わせて人間が決めてから渡します。検出に漏れがあるときは、通ったことを根拠に「個人情報は無い」とは言えません。

検証結果をpytestに固定する

スクリプトを1回動かして通っただけでは、次にデータを足したときに同じ検査が働く保証がありません。検査をテストにして残します。テストは2種類です。クリーンなファイルが通ること、そして壊れた行がそれぞれ想定した理由で落ちることです。

import json
from pathlib import Path
 
import pandas as pd
import pandera.pandas as pa
import pytest
 
from validate_jsonl import TrainRecord, load
 
 
def test_clean_file_passes():
    TrainRecord.validate(load("data/clean.jsonl"), lazy=True)
 
 
def row(**kw):
    base = {"id": "x1", "prompt": "q", "response": "a", "source": "manual"}
    return base | kw
 
 
@pytest.mark.parametrize(
    "rows, check",
    [
        ([row(id="1"), row(id="2")], "multiple_fields_uniqueness"),
        ([row(id="1", prompt="Q "), row(id="2", prompt="q")],
         "no_near_duplicate"),
        ([row(response=None)], "not_nullable"),
        ([row(response="mail a@b.com")], "no_pii"),
        ([row(source="web")], "isin"),
    ],
)
def test_bad_rows_are_rejected(rows, check):
    with pytest.raises(pa.errors.SchemaErrors) as e:
        TrainRecord.validate(pd.DataFrame(rows), lazy=True)
    assert check in json.dumps(e.value.message)

2つ目のテストは、スキーマ自身の検査です。ルールを緩めたときに、検出されなくなった理由で落ちます。Claude Codeがテストを通すためにチェックを消したり緩めたりする編集をしたら、ここで気づけます。

pytest -q

手元の実行では6件(クリーンなファイル1件、壊れた行5件)がすべて通りました。

CLAUDE.mdとStopフックで「通るまで」を固定する

プロンプトで毎回「pytestが通るまで直して」と書く代わりに、検査コマンドをCLAUDE.mdに置きます。ベストプラクティスのページは、リポジトリに検査があるなら、CLAUDE.mdで最も役に立つ1行はそれを走らせるコマンドだと説明しています。

# Commands
- 検証: `pytest -q`(data/ の変更後は必ず実行)
- 単体実行: `python validate_jsonl.py data/<file>.jsonl`
 
# Conventions
- スキーマは validate_jsonl.py の TrainRecord に集約する
- チェックを消す・緩めるときは、理由を説明して承認を得る
- 本番データ(data/prod/)は読まない。ダミーで検証する

依頼文は、結果の証拠を求める形にします。同じページが、「全部通りました」だけでは情報が少ないとして、実行したコマンドと出力を求めるよう書いているためです。

data/sample.jsonl を検証する validate_jsonl.py を pandera の DataFrameModel で
書いて。重複(完全一致と正規化後一致)・欠損・メールと電話番号の混入を検査し、
壊れた行ごとのテストも tests/ に置いて。pytest が通るまで直して、最後に
実行したコマンドと出力を見せて

作業の終了条件まで機械に持たせたいなら、Stopフックにします。Stopフックは、Claudeが応答を終えようとするたびにスクリプトを走らせ、終了コード2を返すと作業に戻します。標準エラーに出した内容が、止められた理由としてClaudeに返ります。

{
  "hooks": {
    "Stop": [
      {
        "hooks": [
          { "type": "command", "command": "pytest -q 1>&2 || exit 2" }
        ]
      }
    ]
  }
}

この設定を.claude/settings.jsonに置くと、pytestが落ちている間はターンが終わりません。ただし、ツール呼び出しを挟まずに連続8回ブロックすると、Claude Codeが次のブロックを無視してターンを終えます。上限はCLAUDE_CODE_STOP_HOOK_BLOCK_CAPで変えられます。テストを一度の編集で直せない大きなデータでは、フックが無限ループになるわけではなく、8回で人間に返ってくると考えておくと運用の見通しが立ちます。

Pythonのコード自体の品質ゲートは、別にruffとmypyを併用する検証ループを足せます。データの中身を見るpanderaと、コードを見るruffとmypyは検査の対象が重ならないため、同じStopフックにコマンドを並べられます。dbtのモデルが相手なら、dbt testでデータテストを足す手順が同じ発想をSQL側に当てています。

つまずきやすい所

次の3点も、検査の抜けにつながります。

• 列チェックは既定でnullを取り除いてから走ります。nullも検査対象にしたいときはCheck(..., ignore_na=False)を指定します。これはnullableとは別の設定です
• .pandasを付けずにimport pandera as paで書かせると、FutureWarningの原因になります。CLAUDE.mdのConventionsにimport pandera.pandas as paを書いておけば防げます
• strict=Trueでは、データ側に新しい項目が増えただけで検証が落ちます。項目の追加が正当なら、スキーマに足すかstrict="filter"で余分な列を落とすかを、方針として決めます

まとめ

学習データの前処理の検証で任せる価値があるのは、スクリプトを書く作業より、検査を残す作業です。スキーマの一つ一つが、データ品質の約束事のレビュー対象になります。

panderaでunique・nullable・isin・strictを宣言し、重複の正規化規則と個人情報の検出パターンは人間が決める。pytestで「通る例」と「落ちる例」の両方を残し、CLAUDE.mdかStopフックでpytestを走らせる。この3点が揃うと、Claude Codeは通るまで自分で直せます。近似重複の検出と、氏名・住所の検出は、このスキーマの外側に残る課題です。