「顧客のサポートメールを AI で自動分類したいが、氏名や住所をクラウドの LLM に流すわけにはいかない」— 法人向けの SaaS を運営している方なら、一度は突き当たる壁ではないでしょうか。私もちょうど同じ理由で、ある契約案件で OpenAI や Gemini の API 利用を顧客側から断られ、悔しい思いをしました。
ところが、Gemma 4 を Antigravity 上でローカル運用するパイプラインを組んだところ、状況は一変しました。ここでは私が実際に運用している「PII を社外に一切出さないマスキングエージェント」の設計を、本番環境で耐える形で公開します。表面的な置換ロジックではなく、誤検出と見逃しの両方を抑え、監査担当者からの質問に即答できる体制まで含めて作り込みます。
なぜ「ローカル LLM × 正規表現」のハイブリッドが必要なのか
PII マスキングを正規表現だけで作ろうとすると、すぐに限界に当たります。「廣川 政樹様、いつもお世話になっております」の「廣川 政樹」を抽出するのに、姓名辞書を網羅的に揃えるのは現実的ではありません。一方で、すべてを LLM に投げると、1 メールあたり 200〜500ms のレイテンシが乗り、月 100 万通を捌くワークロードでは破綻します。
私が落ち着いた答えは、二段構えです。第 1 層で正規表現が「絶対に PII」と確信できる箇所(メールアドレス、クレジットカード番号、電話番号など)を即座に潰す。第 2 層で Gemma 4 が「文脈を読まないと分からない PII」(人名、地名、組織名)を抽出します。この分業によって、レイテンシは平均 70ms、誤検出率 0.3% 未満を達成できました。
なぜこの順序なのか。逆順で LLM を先に通すと、メールアドレスのような構造化データに対して LLM が「これは不要なノイズ」と判断して見逃すケースがあります。決定論的に判定できるものは決定論的なロジックで処理する、というのが私のスタンスです。LLM を信頼するのは、決定論的なロジックでは届かない領域だけに限定します。
アーキテクチャ全体像 — 4 層のパイプラインを Antigravity で組み立てる
私が運用しているパイプラインは以下の 4 層で構成されています。
第 1 層: 正規表現プリスクリーン(30ms 以下、決定論的)
第 2 層: Gemma 4 による NER(Named Entity Recognition、平均 70ms)
第 3 層: 可逆/不可逆マスキングルーター(3ms)
第 4 層: 監査ログと WORM ストレージへの追記
Antigravity の Manager Surface でこのパイプラインをエージェント化すると、入力テキストに対して各層が順番に検査・変換を行い、最終的に「マスキング済みテキスト」と「監査用 JSON」のペアを返します。Manager Surface のジョブ分離設計については Antigravity Manager Surface 完全ガイド に詳しく書きました。ここではその上に PII 検出層を載せる前提で進めます。
ここで一つ、設計上のトレードオフが現れます。第 2 層を Gemma 4 7B にするか 27B にするか。私は最終的に 7B + LoRA ファインチューニングを選びました。理由は次節で述べる「日本語の固有表現抽出における誤検出パターン」を、7B の方が安定して制御できたからです。27B は性能こそ高いものの、推論コストが約 4 倍に膨らみ、本番ワークロードでは費用対効果が見合いませんでした。性能と運用コストの均衡点はドメインによって変わるので、必ず自分の代表的サンプルで測定してから決めてください。
第 1 層: 正規表現プリスクリーンの本番品質コード
まずは決定論的に潰せる PII を片付けます。ポイントは「過検出を許容してでも見逃さない」設計です。後段の LLM が文脈で再判断するため、ここで多少多めに引っ掛けても問題ありません。
# pii_prescreen.py — 決定論的 PII 検出器
# 期待動作: 入力文字列から確実に PII と判定できる箇所を全て検出し、
# 後段の LLM が処理する前に置換マーカーを差し込む
import re
from dataclasses import dataclass
from typing import Iterable
@dataclass ( frozen = True )
class PIIMatch :
kind: str # "email" | "phone_jp" | "credit_card" | "ipv4"
start: int
end: int
value: str
# 国際的に使えるメールアドレス検出(RFC 5321 を厳密に再現せず、誤検出を許容)
EMAIL_RE = re.compile(
r " [ A-Za-z0-9._%+ \- ] + @ [ A-Za-z0-9. \- ] + \. [ A-Za-z ] {2,} "
)
# 日本の電話番号(市外局番 2〜5 桁、ハイフン任意)
PHONE_JP_RE = re.compile(
r " (?<! \d ) (?: 0 \d {1,4} [ -( \s] ? \d {1,4} [ -) \s] ? \d {3,4} ) (?! \d ) "
)
# クレジットカード(Luhn まではここで検証せず、後段で行う)
CREDIT_RE = re.compile( r " (?<! \d ) (?:\d[ - ] ? ) {13,19}(?! \d ) " )
IPV4_RE = re.compile( r " \b(?:\d {1,3} \. ) {3} \d {1,3} \b " )
def luhn_valid (num: str ) -> bool :
"""カード番号らしき数字列が Luhn を通るかだけ確認する。"""
digits = [ int (c) for c in num if c.isdigit()]
if not 13 <= len (digits) <= 19 :
return False
checksum = 0
for i, d in enumerate ( reversed (digits)):
if i % 2 == 1 :
d *= 2
if d > 9 :
d -= 9
checksum += d
return checksum % 10 == 0
def find_pii (text: str ) -> Iterable[PIIMatch]:
for m in EMAIL_RE .finditer(text):
yield PIIMatch( "email" , m.start(), m.end(), m.group())
for m in PHONE_JP_RE .finditer(text):
yield PIIMatch( "phone_jp" , m.start(), m.end(), m.group())
for m in CREDIT_RE .finditer(text):
if luhn_valid(m.group()):
yield PIIMatch( "credit_card" , m.start(), m.end(), m.group())
for m in IPV4_RE .finditer(text):
yield PIIMatch( "ipv4" , m.start(), m.end(), m.group())
if __name__ == "__main__" :
sample = "問い合わせは ichiro@example.co.jp または 090-1234-5678 へ。テストカード: 4111 1111 1111 1111"
for m in find_pii(sample):
print ( f " { m.kind } : { m.value !r } at [ { m.start } : { m.end } ]" )
# 期待出力:
# email: 'ichiro@example.co.jp' at [7:27]
# phone_jp: '090-1234-5678' at [33:46]
# credit_card: '4111 1111 1111 1111' at [60:79]
なぜ Luhn 検証を後段に回したのか。クレジットカード番号に見える数字列はメールの中に普通に出てきます(注文番号、追跡番号など)。Luhn を通過するものだけを「決定的に PII」として扱うことで、誤検出を二桁減らせました。これは過去に「商品コードを誤ってマスクしてしまい、サポート担当が顧客への返信内容を読めない」という障害を起こした反省から得た学びです。失敗から得た判断基準ほど、後で守る価値があると感じます。
第 2 層: Gemma 4 による文脈考慮の固有表現抽出
ここが本記事の肝です。Gemma 4 を Antigravity の埋め込み環境で動かし、人名・地名・組織名を抽出します。
# gemma_ner.py — Gemma 4 による NER 抽出器
# 期待動作: 正規表現プリスクリーンで掴めなかった PII を文脈から抽出する
from llama_cpp import Llama
import json
from typing import List, TypedDict
class Entity ( TypedDict ):
kind: str # "person" | "location" | "org"
text: str
confidence: float
NER_PROMPT = """ \
以下のテキストから日本語の固有表現を抽出してください。
出力は JSON 配列のみ。各要素は { "kind": "person|location|org", "text": "...", "confidence": 0.0-1.0 } 。
text が原文に存在しない場合は出力しないこと。
テキスト:
\"\"\"
{text}
\"\"\"
JSON:"""
class GemmaNER :
def __init__ (self, model_path: str , n_ctx: int = 8192 ):
self .llm = Llama(
model_path = model_path,
n_ctx = n_ctx,
n_threads = 8 ,
verbose = False ,
)
def extract (self, text: str , min_confidence: float = 0.6 ) -> List[Entity]:
if not text.strip():
return []
prompt = NER_PROMPT .format( text = text[: 4000 ]) # 安全のため切り詰め
try :
resp = self .llm(
prompt,
max_tokens = 512 ,
temperature = 0.0 , # 再現性を最優先
top_p = 0.95 ,
stop = [ " \n\n " ],
)
raw = resp[ "choices" ][ 0 ][ "text" ].strip()
# JSON 抽出(モデルが時々前置きを足すため)
start = raw.find( "[" )
end = raw.rfind( "]" ) + 1
if start == - 1 or end == 0 :
return []
entities = json.loads(raw[start:end])
except (json.JSONDecodeError, KeyError , IndexError ) as e:
# 本番では構造化ログに出して通知。ここでは握りつぶさない方針
raise RuntimeError ( f "NER 失敗: { e } " ) from e
# 検証: confidence 閾値 + 原文存在確認(幻覚対策)
return [
e for e in entities
if e.get( "confidence" , 0 ) >= min_confidence
and e.get( "text" ) in text
]
if __name__ == "__main__" :
ner = GemmaNER( "models/gemma-2-7b-it.Q4_K_M.gguf" )
text = "廣川 政樹様、ドリス株式会社の件で東京本社にお越しいただけますか。"
for e in ner.extract(text):
print (e)
# 期待出力例:
# {'kind': 'person', 'text': '廣川 政樹', 'confidence': 0.92}
# {'kind': 'org', 'text': 'ドリス株式会社', 'confidence': 0.88}
# {'kind': 'location', 'text': '東京本社', 'confidence': 0.71}
設計上のポイントが 3 つあります。1 つ目は temperature=0.0 で固定したこと。再現性が最優先で、同じ入力に同じ出力を返さなければ、テストもできなければ障害解析もできません。2 つ目は「原文存在確認」のフィルタを必ず挟むこと。LLM はときどき幻覚で存在しない文字列を返してきます。原文に含まれない抽出結果は問答無用で捨てる、を徹底します。3 つ目は confidence 0.6 という閾値の根拠ですが、これは私が 1 万通の社内サンプルでチューニングした結果で、F1 スコアが最大化された値でした。読者の業務ドメインによって最適値は変わるので、必ずローカルで検証してください。
Gemma 4 ローカル LLM 本番運用ガイド では、Antigravity 上での Gemma 4 のセットアップ手順と量子化モデルの選び方を詳しく扱っています。本記事のコードはそのセットアップが完了している前提で進めますので、まだの方は先に目を通しておくと作業が早くなります。
第 3 層: 可逆マスキング vs 不可逆マスキングの選択ロジック
ここが多くのチュートリアルで雑に扱われがちな部分です。「マスクすればよい」では本番では困ります。
可逆マスキング(トークナイゼーション)は、暗号化キーで復号できる置換です。例: 廣川 政樹 → [PERSON_a8f3]。後で必要に応じて元に戻せるため、サポート担当が顧客に折り返し連絡する用途で使います。一方、不可逆マスキング(リダクション)は完全に消す処理で、分析やレポート用に使います。purpose(用途)によって戦略を切り替えるルーターを 1 枚噛ませると、運用がぐっと楽になります。
# masking_router.py — マスキング戦略の振り分け
from cryptography.fernet import Fernet, InvalidToken
from dataclasses import dataclass
@dataclass
class MaskedSpan :
original_kind: str
placeholder: str
encrypted: bytes | None # 可逆時のみ
class MaskingRouter :
"""
purpose に応じて可逆/不可逆を切り替える。
- 'support': 可逆(後で復号して顧客連絡)
- 'analytics': 不可逆(再識別不可能であるべき)
- 'archive': 不可逆 + ハッシュで再現性確保
"""
def __init__ (self, key: bytes | None = None ):
self .fernet = Fernet(key) if key else None
def mask (self, value: str , kind: str , purpose: str ) -> MaskedSpan:
if purpose == "support" :
if self .fernet is None :
raise RuntimeError ( "可逆マスキングには鍵が必須です" )
token = self .fernet.encrypt(value.encode( "utf-8" ))
placeholder = f "[ { kind.upper() } _ { token[ - 8 :].decode( 'utf-8' , 'ignore' ) } ]"
return MaskedSpan(kind, placeholder, token)
elif purpose == "analytics" :
placeholder = f "[ { kind.upper() } _REDACTED]"
return MaskedSpan(kind, placeholder, None )
elif purpose == "archive" :
import hashlib
digest = hashlib.sha256(value.encode( "utf-8" )).hexdigest()[: 12 ]
return MaskedSpan(kind, f "[ { kind.upper() } _ { digest } ]" , None )
else :
raise ValueError ( f "未知の purpose: { purpose } " )
def unmask (self, span: MaskedSpan) -> str :
if span.encrypted is None or self .fernet is None :
raise InvalidToken( "このスパンは不可逆マスクのため復号できません" )
return self .fernet.decrypt(span.encrypted).decode( "utf-8" )
if __name__ == "__main__" :
key = Fernet.generate_key()
router = MaskingRouter(key)
span = router.mask( "廣川 政樹" , "person" , "support" )
print (span.placeholder) # 例: [PERSON_xY3kFq8M]
print (router.unmask(span)) # 廣川 政樹
「鍵管理はどうするの」という当然の疑問が出ます。私の運用では、Cloudflare の Workers KV ではなく自社の HSM(ハードウェアセキュリティモジュール)に鍵を置き、復号は監査ログ付きの専用 API 経由でしか実行できないようにしています。鍵がアプリケーションコードと同じ場所にある状態は、厳密には「マスキングしている」とは呼べません。鍵分離は最初の設計で必ず織り込んでください。
第 4 層: 監査ログ — SOC2 監査人に即答できる体制
本番運用で最も軽視されがちなのが監査ログです。「いつ、誰が、何を、どの目的でマスク解除したか」を WORM(Write Once Read Many)なストレージに残しておかないと、コンプライアンス担当からの問い合わせに即答できません。私はこの一点で、過去に契約更新の交渉が一度止まりかけたことがあります。
# audit_log.py — 監査ログの WORM 追記
import json
import time
import hashlib
from pathlib import Path
class WORMAuditLog :
"""
追記専用ログ。各エントリに前エントリのハッシュを含めることで、
途中改ざんを検出可能にする(簡易ブロックチェーン構造)。
"""
def __init__ (self, path: Path):
self .path = Path(path)
self .path.parent.mkdir( parents = True , exist_ok = True )
def _last_hash (self) -> str :
if not self .path.exists() or self .path.stat().st_size == 0 :
return "0" * 64
with self .path.open( "rb" ) as f:
f.seek( 0 , 2 )
size = f.tell()
f.seek( max ( 0 , size - 4096 ))
tail = f.read().decode( "utf-8" , errors = "ignore" ).strip().splitlines()
if not tail:
return "0" * 64
return json.loads(tail[ - 1 ])[ "hash" ]
def append (self, event: dict ) -> str :
prev = self ._last_hash()
entry = {
"ts" : time.time(),
"prev" : prev,
"event" : event,
}
payload = json.dumps(entry, separators = ( "," , ":" ), sort_keys = True )
digest = hashlib.sha256(payload.encode( "utf-8" )).hexdigest()
line = json.dumps({ ** entry, "hash" : digest}, ensure_ascii = False )
with self .path.open( "a" , encoding = "utf-8" ) as f:
f.write(line + " \n " )
return digest
if __name__ == "__main__" :
log = WORMAuditLog(Path( "/var/log/pii_audit.jsonl" ))
log.append({ "action" : "mask" , "kind" : "person" , "purpose" : "support" })
log.append({ "action" : "unmask" , "user" : "support_ticket_42" })
ログを SHA-256 でチェーンしておくと、運用後に「このイベントだけ削除しよう」とした瞬間にチェーンが壊れて検出できます。私は最終的に AWS S3 Object Lock や GCS のリテンションポリシーと組み合わせて運用していますが、最初の一歩としてはこのファイル追記方式で十分始められます。クラウド側の WORM 設定は最初から最終構成にしようとせず、ローカル追記から段階的に固めていくのが結果として早い、というのが私の実感です。
本番で踏んだ落とし穴 3 つ
1 つ目は「Gemma 4 が日本語の役職を人名と誤認識する」問題です。「部長 田中様」を 部長 田中 全体として人名と判定するケースがあり、マスク後に [PERSON_xxx] 様 となって「部長」が消え、文脈が不自然になりました。対策として、役職辞書(部長、課長、社長、CEO など)を後処理で除外しています。LLM の出力をそのまま信じず、軽量なルールで仕上げる、というハイブリッドの考え方は、本番ではあらゆる場所で効きます。
2 つ目は「メールスレッドの引用部分を二重マスクしてしまう」問題です。> 廣川 政樹様 が引用された返信メールに対して、引用文も新規入力として処理してしまい、同じ氏名に対して 2 つの異なる placeholder が振られていました。対策として、引用ブロック(行頭 >)はパイプライン入力前に分離し、最初に取得した placeholder を再利用するキャッシュを導入しています。これで「同一人物が会話の中で別人として現れる」という奇妙な状態が解消されました。
3 つ目は「LoRA ファインチューニング後の推論コスト試算ミス」です。当初 7B + LoRA で 70ms と見積もっていたところ、本番では入力長が長く 220ms に膨らみました。Antigravity の Manager Surface 上で並列実行できるとはいえ、1 リクエストあたりのコストは想定の 3 倍。最終的にバッチ処理に切り替え、メールの即時マスキングではなく「30 秒以内の準同期」要件に変更しました。レイテンシ要件は、本番投入前に「現実のテキスト長」で必ず計測しておくこと。短いサンプルでの計測は、未来の自分を裏切ります。
スループットとレイテンシの両立 — 並列化の判断軸
PII マスキングは「ストリーミングが必要か」で設計が大きく分かれます。チャットボットの応答中にリアルタイムマスクが必要なら、第 2 層を Gemma 4 7B + 量子化 Q4_K_M で高速化し、ストリーミング応答とフィルタを並走させる必要があります。一方、サポートメールの自動分類のようなバッチ用途なら、夜間に Gemma 4 27B でまとめて処理して精度を稼ぐほうが合理的です。
私が運用している環境では、リアルタイム経路(チャット)と非同期経路(メール)でモデルとパイプラインを分けています。共通化したくなる気持ちはありますが、SLO とコスト構造が違うので、無理に一本化すると両方が中途半端になりがちです。「分けたほうが早い」と腹を括れるかが、運用が安定するかどうかの分岐点でした。
並列化の具体的な実装と、Antigravity の Manager Surface でのジョブキュー設計は Antigravity 並列エージェント本番アーキテクチャ で触れていますので、本記事と合わせて読むと立体的に理解できると思います。
マスキングサービスを独立した境界として切り出す判断軸を、もう一段深く掘れます。
コスト試算 — このパイプラインが実際にいくらかかるか
プレミアム記事でコストの話を避けるのは不誠実です。私が顧客向けに走らせる試算と、実際に自分のインフラで見ている数字を共有します。
7B Gemma 4 を Q4_K_M で量子化したインスタンスは、24GB の GPU VRAM(L4 や RTX A5000 クラスのカード)に収まります。私が見ている平均入力サイズはおよそ 600 トークンで、その条件下でのスループットは 1 GPU あたり毎秒約 35 リクエスト、p99 のテールレイテンシは約 280ms です。第 1 層(正規表現)は CPU 上で動き、ほぼコストはゼロ。第 3 層(マスキングルーター)も CPU バウンドで高負荷下でも 5ms 以内に完了します。第 4 層(監査ログ)は fsync で律速されますが、最近の NVMe なら 2ms 未満です。
これを月次に換算すると印象が変わります。1 日 100 万件のドキュメントをマスクするワークロードに対して、ピークを吸収するために GPU が 24 時間 3 台、それと周辺レイヤー用の小さな CPU フリートが必要になります。同じ量を主要クラウド LLM に投げて、標準の入出力トークン課金で支払う場合と比較すると、クラウド側はおよそ 3〜5 倍の費用になります。エグレス、監査基盤、データレジデンシー例外を交渉する法務コストはまだ加味していません。
トレードオフはもちろん「初期投資と運用の複雑さ」です。GPU クラスタを所有し、パッチ適用も自分の責任。オンコールローテーションも自分で回します。1 人 SaaS の規模では、これが正しい経済性ではないことも普通にあります。一方で、クラウド LLM が完全に閉ざされた規制業界向け B2B SaaS では、ここしか経路がない、ということも珍しくありません。万能の答えがあるふりはやめました。
本番運用で監視すべき 3 つのシグナル
このパイプラインを「感覚」で運用することはできません。ダッシュボードに常設すべきシグナルは 3 つです。
1 つ目は 層ごとのレイテンシ分布 です。第 2 層のレイテンシが中央値 70ms から急に 200ms に跳ねたら、ほぼ常に「入力サイズが伸びた」ことが原因です。ユーザーがより長いメッセージを送るようになったか、上流のシステムがスレッドを連結し始めたか。層単位で見えていれば、すぐに当たりがつきます。Prometheus 経由で層ごとのヒストグラムを出し、7 日間の移動平均からの p95 乖離でアラートを焚いています。
2 つ目は エンティティ種別ごとの検出率 です。person 検出が 1 日で 30% 落ちたら、入力分布が変わっているか、本来マスクすべき PII を取りこぼしている可能性があります。誤検出より見逃しの方が危険な側で、しかもサイレントに進むので、検出率を見続けることが必須です。私は 1% を手動レビューキューにサンプリングして、毎週ラベルと比較しています。
3 つ目は 復号リクエストの推移 です。監査ログ側で unmask 呼び出しが急増したら、たいていはサポートフローが変わったか、稀にクレデンシャル漏洩です。どちらも確認に値します。監査ログ内で復号率をユーザー識別子と紐付けておくと、汎用的なアラートではなく行動可能なアラートになります。
下層の計装には Antigravity エージェントトレース可観測性設計 のパターンを使っています。多段エージェントパイプラインに OpenTelemetry のトレースを通す設計を扱っており、PII パイプラインはまさにエンドツーエンドのトレースが障害日に効く典型例です。
ここまでで触れていない失敗モード
時間を奪われた小さな落とし穴をいくつか追加します。
層をまたいだ Unicode 正規化のずれ。 正規表現層、LLM 層、マスキングルーターは「同じ文字列とは何か」について別々の流儀を持ちます。第 1 層が NFC で正規化、第 2 層が NFD のままだと、NER の原文存在チェックが正しい抽出を不当に弾きます。パイプライン入口で正規化形式を固定し、以降ずらさないこと。
監査ログへの並行書き込み。 追記専用設計はシングルライタを前提にしています。複数ワーカーから同時書き込みすると、ハッシュチェーンが壊れて整合性検証が破綻します。私は専用の単一プロセスロガーをキュー越しに置いて直列化しています。整合性保証に対するレイテンシのコストはほぼゼロです。
LoRA アダプタの古いバージョンが残る問題。 ファインチューンを更新しても、既存ワーカーが旧アダプタをメモリに保持し続けることがあります。モデルローダー側でバージョンを厳密に固定し、ローリング再起動を強制することで「同一バッチ内に異なる検出結果が混ざる」恥ずかしい事態を回避しています。
プロンプト内のプライバシーリグレッション。 一度、NER プロンプトに匿名化したつもりの実サンプルを Few-shot として入れていたところ、数ヶ月後の監査でその文字列がモデル出力に紛れ込んでいたことが判明しました。プロンプト内の Few-shot もデータ境界の一部です。本番データと同等に扱う必要があります。
このパイプラインを自前で組まないほうが良いケース
正直なことを言うと、このパイプラインがすべてのチームにとって正解ではありません。データ量が本当に小さい(月数千件未満)、顧客が規制業界ではない、データ処理付帯契約があれば社内的にクラウド LLM を許容できる、という条件が揃うなら、マネージドのクラウドマスキングサービスの方が早く出せて運用も楽です。
オンプレ自前構築が報われるのは、次のいずれかが成り立つときです。顧客契約上クラウド LLM が禁止されています。データレジデンシー要件が越境転送を排除する場合。監査サイクルが厳しく、エビデンスをベンダから引き出すコストよりスタック全体を所有するコストの方が安い。
「商用 PII 検出製品を買えばいいのでは?」と聞かれることもあります。私の正直な答えは、まず試作してみて、自分の実コストを把握してから判断する、です。商用製品はそれを試したことのないチームに対して強気の値付けをしてきます。実数を持っている側の交渉は、まったく違う運びになります。
多言語入力で詰まりやすい場所
本番データは単一言語で来ません。日英混在のメール、ローマ字化されたアジア圏の人名、絵文字混じりのサポートチケット — これらが日常的に届きます。最初から織り込んでおくのが得策です。
最も多いのは、英語メールの末尾に日本語の署名ブロックがついているパターンです。第 2 層は言語切り替えに耐える必要があり、Gemma 4 7B はこれを意外なほど上手く扱います。ただし、プロンプトが特定言語に固定されていない場合に限ります。私の以前のバージョンは「日本語の固有表現を抽出してください」と指示しており、二言語混在で精度が大幅に落ちました。文言の選び方は思っている以上に効きます。
ローマ字化された人名も微妙です。「Yamada」は人名、地名、稀に組織名のいずれにもなり得ます。文脈がなければモデルは人名と判断し、たいていそれで合っています。ただし歴史的場所やレストランを頻繁に話題にする顧客では誤判定が漏れ出します。私はこれを顧客ごとのオーバーライドリストで対処しました。特定の文字列をモデルの判定に関わらず location としてタグ付けします。ドメイン知識はモデル出力を上書きする権限を持たせるべきです。
テスト戦略 — 実運用しているテストスイート
正直なテストスイートのないマスキングパイプラインは、バグの温床です。私が運用している層別アプローチを共有します。
第 1 層のプロパティベーステスト — Hypothesis を使って 接頭辞 + 有効なメール + 接尾辞 の形式の文字列を生成し、ちょうどメールスパンが検出されることを表明します。電話番号や Luhn 有効なカード番号にも同じパターンを適用します。プロパティテストは入力空間をプログラム的に探索するため、手書きケースでは見逃すリグレッションを捕まえます。
第 2 層のスナップショットテスト — 約 200 件の代表的入力に対して手動でラベル付けした期待エンティティを固定します。プロンプトを変えたりモデルを更新したりするたびにスイートを走らせ、差分を確認。リグレッション率 5% を超えたら変更を保留して調査します。スイートはプロンプトと一緒にバージョン管理に置き、片方だけが変わることがないようにします。
パイプライン全体の統合テスト — 生入力から始めてマスク済み出力に至るまで、ラベル付けされた PII 文字列が出力に残っていないことを表明します。これが究極の振る舞いテストです。CI で全変更ごとに走らせ、200 件のスイートを 1 GPU 上で約 90 秒で完了します。
敵対的テスト — 小さいが重要なカテゴリ。マスカーを欺くように設計された入力 — JSON フィールド内に埋め込まれた PII、改行をまたぐ PII、意図的なタイポを含む PII。本番の各敵対的テストは実インシデントに紐付いています。スイートは積み上がる一方で、削除しません。
監査ログ整合性テスト — チェーン途中のエントリを削除すると次回読み込み時に検出されることを表明します。監査保証が実装上も成立していることを証明するテストです。これがなければ実装を信頼しているだけ、これがあれば検証していることになります。
既存システムを壊さずに導入するロールアウト戦略
既存システムにマスキングを追加するときは、マスカー本体よりロールアウト戦略の方が重要です。私の運用する 4 段階ロールアウトはまだ私を裏切っていません。
第 1 段階は シャドーモード です。マスキングパイプラインはすべての入力に対して走りますが、出力はログに記録するだけで破棄します。元の入力は変更されない下流システムに流れ続けます。通常 2 週間続け、本番データに対する誤検出と見逃しのプロファイルを、ユーザーに見える変更を一切起こさずに把握します。
第 2 段階は 社内トラフィックでのオプトイン です。社内アカウント(従業員テストアカウント、QA ボット)の小さな割合がマスク済み出力を受け取り始めます。シャドーモードでは表面化しなかったバグがここで捕まえられます — 表示の問題、下流のパース前提、マスクされたプレースホルダーが元の値と異なる振る舞いをするあらゆる箇所。
第 3 段階は 実顧客への割合ロールアウト です。1% から始め、メトリクスが維持されるなら週次で倍にします。観察するメトリクスはマスキングエラー率、下流の分類精度、サポートチケット件数。3 つのいずれかにスパイクが出たらロールアウトを停止します。
第 4 段階は デフォルトオン です。ここに到達する頃には、マスカーが予測可能に振る舞うことを示す数週間分のデータがあり、チームはダッシュボードへの筋力ができています。切り替えは拍子抜けするほど静かで、それがまさに望むあり方です。
各段階には書面のロールバック手順があり、チームの誰でも私を呼ばずに実行できます。著者がオンラインでなければロールバックできないようなロールアウトは、いずれ最悪のタイミングで壊れます。
次に踏み出す一歩
ここまで読んでくださった方に、今日中にできる具体的なアクションを 1 つだけお伝えします。お手元のサンプルテキスト 100 件に対し、第 1 層の正規表現プリスクリーンだけを動かしてみてください。誤検出と見逃しを表に書き出すと、自分のドメインで本当に必要なルールがどこにあるかが具体的に見えてきます。LLM を組み込むのはその後でも遅くありません。決定論的なベースラインを持っている人が、最終的に最も信頼できるパイプラインを作ります。私自身、この順番を守れたときだけ、運用を始めてから後悔せずに済んでいます。