0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

OpenAI Privacy Filterを使ってテキスト内の個人情報を抽出してみる

0
Posted at

OpenAIが先日、Privacy Filterという個人情報検出のためのモデルを公開しました。

1.5Bと軽量で、ノートPCのCPUでも動作するとされています。

環境

  • Windows 11 WSL2
  • Snapdragon X CPU / 16GB RAM

動かしてみる

さっそく動かしてみましょう。モデルはHuggingfaceで公開されているのでtransformersを使えば簡単に動かせます。適当に動かすためのプログラムをAIに書かせたものが以下です。
テキストのチャンク処理とかお好みで変えてください。

# /// script
# requires-python = ">=3.13"
# dependencies = [
#     "torch==2.11.0",
#     "transformers==5.6.1",
#     "typer==0.24.1",
#     "accelerate==1.13.0",
#     "tqdm==4.67.3",
# ]
# ///
"""
usage:
  初回(モデルとパッケージのダウンロード):
    uv run --script main.py download

  以降(オフライン実行):
    uv run --offline --script main.py check sample.txt
    uv run --offline --script main.py check ./*.txt

  mise タスク:
    mise run download
    mise run check sample.txt
    mise run check ./*.txt
"""

import glob
import os
import sys
from pathlib import Path

# transformers のインポート前に設定しないとオフラインモードが有効にならない
if len(sys.argv) > 1 and sys.argv[1] == "check":
    os.environ.setdefault("HF_HUB_OFFLINE", "1")

import typer
import torch
from tqdm import tqdm
from transformers import AutoModelForTokenClassification, AutoTokenizer

MODEL_NAME = "openai/privacy-filter"

app = typer.Typer()


def split_into_chunks(text: str, max_chars: int = 800) -> list[str]:
    """テキストを段落ごとに分割し、max_chars 以下のチャンクにまとめる。"""
    paragraphs = text.split("\n\n")
    chunks: list[str] = []
    current = ""
    for para in paragraphs:
        # 1段落が max_chars を超える場合は強制分割
        while len(para) > max_chars:
            slice_, para = para[:max_chars], para[max_chars:]
            if current:
                chunks.append(current)
                current = ""
            chunks.append(slice_)
        if not para:
            continue
        candidate = (current + "\n\n" + para).lstrip() if current else para
        if len(candidate) > max_chars:
            chunks.append(current)
            current = para
        else:
            current = candidate
    if current:
        chunks.append(current)
    return chunks


@app.command()
def download():
    """モデルをHugging Faceからダウンロードしてローカルにキャッシュします。"""
    print("モデルをダウンロードしています...")
    AutoTokenizer.from_pretrained(MODEL_NAME)
    AutoModelForTokenClassification.from_pretrained(MODEL_NAME)
    print("ダウンロード完了しました。")


@app.command()
def check(patterns: list[str] = typer.Argument(..., help="チェック対象のテキストファイルまたはglobパターン(**/*.txt 等、複数指定可)")):
    """テキストファイルの個人情報をオフラインでチェックします(事前に download コマンドが必要)。"""

    # glob とディレクトリ指定を組み合わせてファイルリストを構築
    paths: list[str] = []
    for pattern in patterns:
        p = Path(pattern)
        if p.is_dir():
            # ディレクトリ指定時は再帰的に全ファイルを展開
            found = sorted(str(f) for f in p.rglob("*") if f.is_file())
            paths.extend(found) if found else paths.append(pattern)
        else:
            expanded = glob.glob(pattern, recursive=True)
            if expanded:
                # ディレクトリはスキップしファイルのみ追加
                file_list = sorted(f for f in expanded if not Path(f).is_dir())
                paths.extend(file_list) if file_list else paths.append(pattern)
            else:
                paths.append(pattern)  # glob にマッチしない場合はそのまま渡す(エラーは open 時に発生)

    # モデルの準備
    print("モデルの準備をしています...", file=sys.stderr)
    tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
    model = AutoModelForTokenClassification.from_pretrained(MODEL_NAME, device_map="auto")
    print("model.device:", model.device, file=sys.stderr)

    max_length = model.config.max_position_embeddings
    max_chars = 4096

    # ファイルごとに処理
    for path in paths:
        with open(path, encoding="utf-8") as file:
            print(f"\n--- {file.name} ---", file=sys.stderr)
            text = file.read()
            chunks = split_into_chunks(text, max_chars=max_chars)
            print(f"テキストを {len(chunks)} チャンクに分割しました", file=sys.stderr)

            print("チェック開始します...", file=sys.stderr)
            results: list[str] = []
            for chunk in tqdm(chunks, desc="チャンク処理", unit="chunk", file=sys.stderr):
                inputs = tokenizer(
                    chunk,
                    return_tensors="pt",
                    max_length=max_length,
                    truncation=True,
                ).to(model.device)

                with torch.no_grad():
                    outputs = model(**inputs)

                token_ids = inputs["input_ids"][0].tolist()
                predicted_token_classes = [
                    model.config.id2label[t.item()]
                    for t in outputs.logits[0].argmax(dim=-1)
                ]

                i = 0
                while i < len(token_ids):
                    label = predicted_token_classes[i]
                    if label == "O":
                        i += 1
                        continue

                    _, base_label = label.split("-", 1)
                    span_ids = [token_ids[i]]
                    i += 1

                    # 同じ base_label が連続する限り収集(S-の連続・孤立I-/E-にも対応)
                    while i < len(token_ids):
                        next_label = predicted_token_classes[i]
                        if next_label == "O" or next_label.startswith("B-"):
                            break
                        next_prefix, next_base = next_label.split("-", 1)
                        if next_base != base_label:
                            break
                        span_ids.append(token_ids[i])
                        i += 1
                        if next_prefix == "E":
                            break

                    decoded = tokenizer.decode(span_ids).strip()
                    results.append(f"  {base_label} -> {decoded}")

            print(f"\n=== {file.name} の個人情報候補 ===")
            if results:
                for line in results:
                    print(line)
            else:
                print("  (候補は見つかりませんでした)")


if __name__ == "__main__":
    app()

PythonのInline Script Metadata使っているので、このスクリプトをそのままuv run --scriptとかで動かせば依存関係がインストールされて実行できるようになっています。一応このプログラムのリポジトリもあります。
downloadコマンドでモデルのダウンロードだけできるようにしてあり、checkコマンドは完全オフラインでも動作できるようにしてあります。

適当にQiitaの拙稿をチェックしてみます。

$ wget -O qiita.txt https://qiita.com/suzuki_sh/items/9d6f77c6975c69485e08.md
$ uv run --offline --script main.py check qiita.txt
(中略)

=== qiita.txt の個人情報候補 ===
  private_person -> suzuki_sh
  private_url -> ://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50391
  secret -> /b589e029-6147-4b5d
  secret -> 04-445c
  secret -> 2bd0-87
  secret -> 6-40d8-95
  secret -> -59
  secret -> 792222

private_personとして私のアカウント名suzuki_shが検出されました。これは割と正しい挙動のように思います。
一方で、private_urlとsecretとして画像アップロード先のURLが検出されています。これは公開情報ですが、そういうのの判別は難しいですね。

その他、日本語のテキストでも普通に動作しましたが、「大阪」や「JR東海」のような通常の名詞がprivate_personやprivate_addressとして反応してしまうことがありました。英語でも「Gemini」や「Jira」などをprivate_personとして検出していました。適合率はまずまずのように思います。また再現率は評価できていません。
HTMLなどのソースコードを与えると、secretやprivate_phoneとして過剰に反応してしまうことがありました。入力するテキストの前処理は必要そうです。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?