Zenn
セリオ株式会社 テックブログPublicationへの投稿
🕵️‍♂️

【Presidio】LLM活用時の個人情報マスキング

くじら
に公開
AI
LLM
生成 AI
presidio
tech

背景

LLMを活用したサービスを社内で展開する際、個人情報（PII）を含むデータの取り扱いには特に注意が必要です。AWSやAzureなどのクラウドプロバイダは、利用者のデータをトレーニングに使用しないことを明記していますが、社内規定やコンプライアンスの観点からも慎重な対応が求められます。Presidio[1]は、Microsoftが提供する個人情報の検出やマスキングを行うOSSであり、今回はPresidioを使って個人情報をマスキングした後、そのマスキングを解除する方法についてまとめます。

環境

項目 バージョン
OS Windows11 Pro
ランタイム Python 3.10
主要ライブラリ presidio-analyzer, presidio-anonymizer
使用モデル GPT-4.1-mini

事前準備

  1. Pythonパッケージのインストール
$ pip install presidio-analyzer presidio-anonymizer fastapi uvicorn
  1. NLPモデルのダウンロード
$ python -m spacy download ja_core_news_trf

手順

1. NLPエンジンのセットアップ

NLPエンジンのセットアップを行います。今回はSpacyの日本語モデルであるja_core_news_trfを使用します。日本語で使用できるSpacyモデルはこちらから確認できます。また、PresidioはSpacy以外にも、Hugging FaceのTransformersモデルやStanza等のNLPエンジンをサポートしています。[2]

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Dict, List, Optional
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine, OperatorConfig
from presidio_anonymizer.operators import Operator, OperatorType
from presidio_analyzer.nlp_engine import NlpEngineProvider
import uvicorn

app = FastAPI()

# NLPエンジンセットアップ
configuration = {
    "nlp_engine_name": "spacy",
    "models": [
        {"lang_code": "ja", "model_name": "ja_core_news_trf"}
    ],
}
provider = NlpEngineProvider(nlp_configuration=configuration)
nlp_engine = provider.create_engine()

analyzer = AnalyzerEngine(nlp_engine=nlp_engine, supported_languages=["ja"])
anonymizer = AnonymizerEngine()

# 匿名化のためのマッピング辞書
entity_mapping = {}

2. カスタムオペレータの作成

デフォルトのAnonymizerエンジンでは、検出したPIIの種類（人名や電話番号等）によって対象の単語が全て<PERSON>や<PHONE_NUMBER>に置き換わります。
（例：ジョンの電話番号は000-0000-0000と111-1111-1111です。→ <PERSON>の電話番号は<PHONE_NUMBER>と<PHONE_NUMBER>です。）

しかし、これだとマスキングを解除する際、2つの電話番号を識別することができません。そのため、<PHONE_NUMBER_1>、<PHONE_NUMBER_2>のようなユニークな識別子に置き換えるカスタムAnonymizerを作成します。

class InstanceCounterAnonymizer(Operator):
    """
    エンティティ値をユニークな識別子に置き換える匿名化オペレータ
    """
    
    REPLACING_FORMAT = "<{entity_type}_{index}>"
    
    def operate(self, text: str, params: Dict = None) -> str:
        """テキストを匿名化する"""
        
        entity_type: str = params["entity_type"]
        
        # entity_mappingはエンティティタイプごとのマッピングを含む辞書
        entity_mapping: Dict[str, Dict[str, str]] = params["entity_mapping"]
        
        entity_mapping_for_type = entity_mapping.get(entity_type, {})
        if not entity_type in entity_mapping:
            entity_mapping[entity_type] = {}
            entity_mapping_for_type = entity_mapping[entity_type]
        
        # 元のテキストがエンティティマッピングの値に含まれるか確認
        for token, original in entity_mapping_for_type.items():
            if original == text:
                return token
            
        previous_index = self._get_last_index(entity_mapping_for_type)
        new_text = self.REPLACING_FORMAT.format(
            entity_type=entity_type, index=previous_index + 1
        )
        
        # キーを匿名化トークン、値を元のテキストにする
        entity_mapping[entity_type][new_text] = text
        return new_text
        
    @staticmethod
    def _get_last_index(entity_mapping_for_type: Dict) -> int:
        """指定されたエンティティタイプの最後のインデックスを取得"""
        return len(entity_mapping_for_type)
        
    def validate(self, params: Dict = None) -> None:
        """オペレータのパラメータを検証"""
        
        if "entity_mapping" not in params:
            raise ValueError("entity_mappingパラメータが必要です")
        if "entity_type" not in params:
            raise ValueError("entity_typeパラメータが必要です")
            
    def operator_name(self) -> str:
        return "entity_counter"
        
    def operator_type(self) -> OperatorType:
        return OperatorType.Anonymize

3. 匿名化APIの実装

PresidioとFastAPIを組み合わせて、個人情報の匿名化および復元をAPIとして利用します。ここでは、先ほど作成したカスタムAnonymizerをFastAPIのエンドポイントに組み込み、テキストの匿名化と復元処理を実装します。

匿名化API/anonymize_textは、入力されたテキストから人名や電話番号などの個人情報を検出し、ユニークなトークンに置き換えます。置き換えたトークンと元の値の対応関係はマッピング辞書として保持され、後から復元API/deanonymize_textで元のテキストに戻すことができます。

# カスタムAnonymizerをエンジンに追加
anonymizer_engine = AnonymizerEngine()
anonymizer_engine.add_anonymizer(InstanceCounterAnonymizer)

class TextRequest(BaseModel):
    text: str
    
@app.post("/anonymize_text")
def anonymize_text(req: TextRequest):
    """テキストを匿名化し、後で復元できるようにマッピングを保持する"""
    print(req)
    results = analyzer.analyze(
        text=req.text,
        entities=["PERSON", "PHONE_NUMBER"],
        language="ja"
    )
    
    # 匿名化
    anonymized_result = anonymizer_engine.anonymize(
        text=req.text,
        analyzer_results=results,
        operators={
            "DEFAULT": OperatorConfig(
                "entity_counter",
                params={"entity_mapping": entity_mapping}
            )
        }
    )
    
    return { "text": anonymized_result.text }

@app.post("/deanonymize_text")
def deanonymize_text_content(req: TextRequest):
    """
    匿名化されたテキスト内のトークンを復元する
    """
    print(req)
    
    # マッピングが空の場合、エラーを返す
    if not entity_mapping:
        return {"error": "マッピングが存在しません。先に/anonymizeエンドポイントを呼び出してください。"}

    try:
        anonymized_text = req.text
        # トークンを検索して置換
        for _, mapping in entity_mapping.items():
            for token, original in mapping.items():
                anonymized_text = anonymized_text.replace(token, original)

        return {"original_text": anonymized_text}
    except Exception as e:
        return {"error": f"予期しないエラーが発生しました: {str(e)}"}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

結果

検証のために、/anonymize_textエンドポイントを呼び出して、個人情報をマスキングした後にLLMへクエリを送信し、LLMのレスポンスを/deanonymize_textエンドポイントで元のテキストに戻すという流れを実装しました。

検証コード
from langfuse.langchain import CallbackHandler
from langchain_openai import AzureChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from dotenv import load_dotenv
import os
import requests

load_dotenv()

langfuse_handler = CallbackHandler()

# Azure OpenAIの設定
model = AzureChatOpenAI(
    deployment_name=os.environ.get("AZURE_OPENAI_DEPLOYMENT_NAME"),
    api_key=os.environ.get("AZURE_OPENAI_API_KEY"),
    azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
    api_version="2024-12-01-preview",
    model_name="gpt-4.1-mini"
)

# プロンプトテンプレートの作成
prompt = ChatPromptTemplate.from_template("あなたとの対話で個人情報が含まれている場合は、事前にマスキングされた状態で送信されます。（クエリ例：私は<PERSON_1>です。）（回答例：こんにちは<PERSON_1>!）\
                                          もし、このようなマスキングされたものが送られた場合は、そのマスキングを保ったまま回答を出力してください。\
                                          Query: {input}")

chain = prompt | model

query = "ジョンの電話番号は090-0000-0000と090-0000-0001です。"

# テキストを/anonymize_textに送信
anonymize_url = "http://localhost:8000/anonymize_text"
anonymize_res = requests.post(anonymize_url, json={"text": query})
anonymized_text = anonymize_res.json()["text"]

# 匿名化テキストをLLMに送信
llm_response = chain.invoke({"input": anonymized_text}, config={"callbacks": [langfuse_handler]})
llm_text = llm_response.content if hasattr(llm_response, "content") else str(llm_response)

# LLMレスポンスを/deanonymize_textに送信
deanonymize_url = "http://localhost:8000/deanonymize_text"
deanonymize_res = requests.post(deanonymize_url, json={"text": llm_text})
deanonymized_text = deanonymize_res.json().get("original_text", deanonymize_res.text)

print(deanonymized_text)

元のテキスト

ジョンの電話番号は090-0000-0000と090-0000-0001です。

マスキング後のテキスト

<PERSON_1>の電話番号は<PHONE_NUMBER_2>と<PHONE_NUMBER_1>です。

LLMのアウトプット

ありがとうございます。<PERSON_1>の電話番号は<PHONE_NUMBER_2>と<PHONE_NUMBER_1>ですね。何かお手伝いできることがあれば教えてください。

マスキング解除後のテキスト

ありがとうございます。ジョンの電話番号は090-0000-0000と090-0000-0001ですね。何かお手伝いできることがあれば教えてください。

想定通り、マスキングされた状態でLLMへクエリを送信し、LLMのレスポンスもマスキングされた状態で返ってきました。さらに、マスキング解除APIを用いて元のテキストに戻すこともできました。

結論＆まとめ

  • Presidioを用いることで、PIIを含むテキストをマスキングすることができました。
  • カスタムAnonymizerを実装することで、マスキングされたテキストの復元が可能になりました。
  • 今回は、ja_core_news_trfを使用しましたが、次は他の日本語モデルを使用してPII検出の精度を比較してみたいと思います。
脚注

  1. Presidioの公式ドキュメント(https://microsoft.github.io/presidio/) ↩︎

  2. PresidioのNLPエンジンの詳細(https://deepwiki.com/microsoft/presidio/2.1.2-nlp-engine) ↩︎

セリオ株式会社 テックブログ
セリオ株式会社 テックブログPublication

未来の技術に挑戦し続ける岡山のソフトウェア企業です。製造業をはじめ、物流・介護などさまざまな業界のお客様と共創しDXを実現します。

Discussion