> ## Documentation Index
> Fetch the complete documentation index at: https://docs.coreweave.com/llms.txt
> Use this file to discover all available pages before exploring further.
# PII データの処理
> W&B Weave のデータマスキングおよびフィルタリングツールを使用して、LLM アプリケーションで個人を特定できる情報 (PII) を処理します。
これはインタラクティブなノートブックです。ローカルで実行するか、以下のリンクを使用してください。
* [Google Colab で開く](https://colab.research.google.com/github/wandb/docs/blob/main/weave/cookbooks/source/pii.ipynb)
* [GitHub でソースを表示](https://github.com/wandb/docs/blob/main/weave/cookbooks/source/pii.ipynb)
このガイドでは、個人を特定できる情報 (PII) を外部に漏らすことなく W\&B Weave を使用する方法を説明します。PII を保護すれば、LLM のトレースや評価を活用しながら、プライバシーおよびコンプライアンスの要件を満たせます。このガイドは、機密性の高いユーザーデータを扱う LLM アプリケーションに Weave を統合する開発者を対象としています。
このガイドでは、PII データを特定、マスク、匿名化するために、次の方法を紹介します。
1. **正規表現**: PII データを特定してマスクします。
2. **Microsoft の [Presidio](https://microsoft.github.io/presidio/)**: Python ベースのデータ保護 SDK で、マスキングと置換の機能を備えています。
3. **[Faker](https://faker.readthedocs.io/en/master/)**: 疑似データを生成する Python ライブラリです。Presidio と組み合わせて PII データを匿名化します。
また、\_`weave.op` の入力/出力ログのカスタマイズ\_と *`autopatch_settings`* を使用して、PII のマスキングと匿名化をワークフローに組み込む方法も説明します。詳細については、[ログする入力と出力のカスタマイズ](/ja/products/wandb/weave/guides/tracking/ops#customize-logged-inputs-and-outputs)を参照してください。
まずは、次の手順を実行してください。
1. [概要](#overview)セクションを確認します。
2. [前提条件](#prerequisites)を満たします。
3. PII データを特定、マスク、匿名化するための[利用可能なメソッド](#redaction-methods-overview)を確認します。
4. [Weave の Call にメソッドを適用します](#apply-the-methods-to-weave-calls)。
概要
このセクションでは、`weave.op` を使用した入出力のロギングの概要と、Weave で PII データを扱う際のベストプラクティスについて説明します。
Weave Op では、入力と出力の後処理関数を定義できます。これらの関数を使うと、LLM Call に渡すデータや Weave にログするデータを変更できます。
次の例では、2 つの後処理関数を定義し、それらを引数として `weave.op()` に渡しています。
```python lines theme={"system"}
from dataclasses import dataclass
from typing import Any
import weave
# 入力用ラッパークラス
@dataclass
class CustomObject:
x: int
secret_password: str
# まず、入力と出力の後処理を行う関数を定義します。
def postprocess_inputs(inputs: dict[str, Any]) -> dict[str, Any]:
return {k:v for k,v in inputs.items() if k != "hide_me"}
def postprocess_output(output: CustomObject) -> CustomObject:
return CustomObject(x=output.x, secret_password="REDACTED")
# 次に、`@weave.op` デコレーターを使用する際に、これらの処理関数を引数として渡します。
@weave.op(
postprocess_inputs=postprocess_inputs,
postprocess_output=postprocess_output,
)
def some_llm_call(a: int, hide_me: str) -> CustomObject:
return CustomObject(x=a, secret_password=hide_me)
```
PII データを扱う際の Weave のベストプラクティス
PII データを扱う環境で Weave を使用する前に、以下のベストプラクティスを確認してください。ベストプラクティスは開発ライフサイクルの段階ごとにまとめてあり、暗号化に関する補足ガイダンスも記載しています。
テスト時
* 匿名化したデータをログして、PII が検出されるかを確認します。
* Weave トレースを使用して、PII の処理プロセスをトラッキングします。
* 実際の PII を露出させずに、匿名化のパフォーマンスを測定します。
本番環境では
* 生の PII は絶対にログしないでください。
* 機密性の高いフィールドは、ログする前に暗号化してください。
暗号化のヒント
* 後で復号する必要があるデータには、可逆暗号化を使用します。
* 元に戻す必要のない一意の ID には、一方向ハッシュを適用します。
* 暗号化したまま分析する必要があるデータには、専用の暗号化方式の使用を検討してください。
前提条件
いずれかのマスキング方法を適用する前に、以下のセットアップ手順を実行して、依存関係のインストール、APIキーの設定、Weave プロジェクトの初期化、サンプルデータの読み込みを済ませておいてください。
1. まず、必要なパッケージをインストールします。
```python lines theme={"system"}
%%capture
# @title 必要な Python パッケージ:
!pip install cryptography
!pip install presidio_analyzer
!pip install presidio_anonymizer
!python -m spacy download en_core_web_lg # Presidio は spaCy の NLP エンジンを使用します
!pip install Faker # Faker を使用して PII データを疑似データに置き換えます
!pip install weave # トレースを活用するため
!pip install set-env-colab-kaggle-dotenv -q # 環境変数の設定用
!pip install anthropic # Sonnet を使用するため
!pip install cryptography # データを暗号化するため
```
2. 次のページでAPIキーを作成します。
* [W\&B User Settings](https://forge.coreweave.com/settings)
* [Anthropic Console](https://platform.claude.com/login?returnTo=%2Fsettings%2Fkeys)
```python lines theme={"system"}
%%capture
# @title APIキーを正しく設定する
# 使用方法については https://pypi.org/project/set-env-colab-kaggle-dotenv/ を参照してください。
from set_env import set_env
_ = set_env("ANTHROPIC_API_KEY")
_ = set_env("WANDB_API_KEY")
```
3. Weave プロジェクトを初期化します。
```python lines theme={"system"}
import weave
# 新しい Weave プロジェクトを開始
WEAVE_PROJECT = "pii_cookbook"
weave.init(WEAVE_PROJECT)
```
4. 10 件のテキストブロックを含む、デモ用の PII データセットを読み込みます。
```python lines theme={"system"}
import requests
url = "https://raw.githubusercontent.com/wandb/docs/main/weave/cookbooks/source/10_pii_data.json"
response = requests.get(url)
pii_data = response.json()
print('PII data first sample: "' + pii_data[0]["text"] + '"')
```
マスキング方法の概要
[前提条件](#prerequisites)を満たしたら、次のいずれかの方法で PII データを検出して保護できます。どの方法でも PII を特定してマスクでき、必要に応じて匿名化も行えます。
1. **正規表現**: PII データを特定してマスクします。
2. **Microsoft [Presidio](https://microsoft.github.io/presidio/)**: マスキングと置換の機能を備えた、Python ベースのデータ保護 SDK です。
3. **[Faker](https://faker.readthedocs.io/en/master/)**: 疑似データを生成する Python ライブラリです。
方法 1: 正規表現を使用してフィルターする
[正規表現 (regex)](https://docs.python.org/3/library/re.html) は、PII データを特定してマスクするためのシンプルな方法です。正規表現を使用すると、電話番号、メールアドレス、社会保障番号など、さまざまな形式の機密情報に一致するパターンを定義できます。より複雑な NLP 手法を使わなくても、大量のテキストをスキャンして情報を置換またはマスクできます。
```python lines theme={"system"}
import re
# 正規表現を使用して PII データをマスクする関数を定義します
def redact_with_regex(text):
# 電話番号のパターン
# \b : 単語境界
# \d{3} : 3 桁ちょうどの数字
# [-.]? : ハイフンまたはドット（省略可）
# \d{3} : さらに 3 桁の数字
# [-.]? : ハイフンまたはドット（省略可）
# \d{4} : 4 桁ちょうどの数字
# \b : 単語境界
text = re.sub(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b", "", text)
# メールアドレスのパターン
# \b : 単語境界
# [A-Za-z0-9._%+-]+ : メールのユーザー名に使用できる文字（1 文字以上）
# @ : @ 記号（リテラル）
# [A-Za-z0-9.-]+ : ドメイン名に使用できる文字（1 文字以上）
# \. : ドット（リテラル）
# [A-Z|a-z]{2,} : 2 文字以上の英大文字または英小文字（TLD）
# \b : 単語境界
text = re.sub(
r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "", text
)
# SSN のパターン
# \b : 単語境界
# \d{3} : 3 桁ちょうどの数字
# - : ハイフン（リテラル）
# \d{2} : 2 桁ちょうどの数字
# - : ハイフン（リテラル）
# \d{4} : 4 桁ちょうどの数字
# \b : 単語境界
text = re.sub(r"\b\d{3}-\d{2}-\d{4}\b", "", text)
# 名前の簡易パターン（すべてのケースには対応していません）
# \b : 単語境界
# [A-Z] : 英大文字 1 文字
# [a-z]+ : 1 文字以上の英小文字
# \s : 空白文字 1 文字
# [A-Z] : 英大文字 1 文字
# [a-z]+ : 1 文字以上の英小文字
# \b : 単語境界
text = re.sub(r"\b[A-Z][a-z]+ [A-Z][a-z]+\b", "", text)
return text
```
関数をテストするには、サンプルテキストを指定して次のコードを実行します。
```python lines theme={"system"}
# 関数をテストします
test_text = "My name is John Doe, my email is john.doe@example.com, my phone is 123-456-7890, and my SSN is 123-45-6789."
cleaned_text = redact_with_regex(test_text)
print(f"Raw text:\n\t{test_text}")
print(f"Redacted text:\n\t{cleaned_text}")
```
方法 2: Microsoft Presidio を使用してマスクする
次の方法では、[Microsoft Presidio](https://microsoft.github.io/presidio/) を使用して PII データを完全に削除します。Presidio は PII をマスクし、その PII のタイプを表すプレースホルダーに置き換えます。たとえば、`"My name is Alex"` の `Alex` は `` に置き換えられます。
Presidio は、[一般的なエンティティ](https://microsoft.github.io/presidio/supported_entities/)を標準でサポートしています。次の例では、`PHONE_NUMBER`、`PERSON`、`LOCATION`、`EMAIL_ADDRESS`、`US_SSN` のいずれかに該当するエンティティをすべてマスクします。Presidio による処理は関数にまとめてあります。
```python lines theme={"system"}
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
# Analyzer を設定します。Analyzer は NLP モジュール (デフォルトでは spaCy モデル) とその他の PII 認識器を読み込みます。
analyzer = AnalyzerEngine()
# Anonymizer を設定します。Anonymizer は analyzer の結果を使用してテキストを匿名化します。
anonymizer = AnonymizerEngine()
# Presidio によるマスキングのプロセスを関数にまとめます
def redact_with_presidio(text):
# テキストを分析して PII データを特定します
results = analyzer.analyze(
text=text,
entities=["PHONE_NUMBER", "PERSON", "LOCATION", "EMAIL_ADDRESS", "US_SSN"],
language="en",
)
# 特定された PII データを匿名化します
anonymized_text = anonymizer.anonymize(text=text, analyzer_results=results)
return anonymized_text.text
```
関数をテストするには、サンプルテキストを渡して次のコードを実行します。
```python lines theme={"system"}
text = "My phone number is 212-555-5555 and my name is alex"
# 関数をテスト
anonymized_text = redact_with_presidio(text)
print(f"Raw text:\n\t{text}")
print(f"Redacted text:\n\t{anonymized_text}")
```
方法 3: Faker と Presidio を使用して置換による匿名化を行う
テキストをマスクする代わりに、MS Presidio を使用して名前や電話番号などの PII を [Faker](https://faker.readthedocs.io/en/master/) Python ライブラリで生成した疑似データに置き換えることで、データを匿名化できます。たとえば、次のようなデータがあるとします。
`"My name is Raphael and I like to fish. My phone number is 212-555-5555"`
Presidio と Faker でデータを処理すると、たとえば次のようになります。
`"My name is Katherine Dixon and I like to fish. My phone number is 667.431.7379"`
Presidio と Faker を組み合わせて使用するには、カスタムオペレーターへの参照を指定する必要があります。これらのオペレーターによって、PII を疑似データに置き換える Faker の関数が Presidio に指定されます。
```python lines theme={"system"}
from faker import Faker
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
fake = Faker()
# faker 関数を作成します（関数は値を受け取る必要があります）
def fake_name(x):
return fake.name()
def fake_number(x):
return fake.phone_number()
# PERSON および PHONE_NUMBER" エンティティ用のカスタムオペレーターを作成します
operators = {
"PERSON": OperatorConfig("custom", {"lambda": fake_name}),
"PHONE_NUMBER": OperatorConfig("custom", {"lambda": fake_number}),
}
text_to_anonymize = (
"My name is Raphael and I like to fish. My phone number is 212-555-5555"
)
# Analyzer の出力
analyzer_results = analyzer.analyze(
text=text_to_anonymize, entities=["PHONE_NUMBER", "PERSON"], language="en"
)
anonymizer = AnonymizerEngine()
# 上で定義したオペレーターを必ず anonymizer に渡してください
anonymized_results = anonymizer.anonymize(
text=text_to_anonymize, analyzer_results=analyzer_results, operators=operators
)
print(f"Raw text:\n\t{text_to_anonymize}")
print(f"Anonymized text:\n\t{anonymized_results.text}")
```
コードを 1 つのクラスにまとめ、先ほど特定した追加のエンティティもエンティティのリストに含めるには、次のコードを実行します。
```python lines theme={"system"}
from typing import ClassVar
from faker import Faker
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
# Faker を拡張した、疑似データ生成用のカスタムクラス
class MyFaker(Faker):
# faker 関数を作成します（値を受け取る必要があることに注意してください）
def fake_address(self):
return fake.address()
def fake_ssn(self):
return fake.ssn()
def fake_name(self):
return fake.name()
def fake_number(self):
return fake.phone_number()
def fake_email(self):
return fake.email()
# 各エンティティ用のカスタムオペレーターを作成します
operators: ClassVar[dict[str, OperatorConfig]] = {
"PERSON": OperatorConfig("custom", {"lambda": fake_name}),
"PHONE_NUMBER": OperatorConfig("custom", {"lambda": fake_number}),
"EMAIL_ADDRESS": OperatorConfig("custom", {"lambda": fake_email}),
"LOCATION": OperatorConfig("custom", {"lambda": fake_address}),
"US_SSN": OperatorConfig("custom", {"lambda": fake_ssn}),
}
def redact_and_anonymize_with_faker(self, text):
anonymizer = AnonymizerEngine()
analyzer_results = analyzer.analyze(
text=text,
entities=["PHONE_NUMBER", "PERSON", "LOCATION", "EMAIL_ADDRESS", "US_SSN"],
language="en",
)
anonymized_results = anonymizer.anonymize(
text=text, analyzer_results=analyzer_results, operators=self.operators
)
return anonymized_results.text
```
関数をテストするには、サンプルテキストを指定して次のコードを実行します。
```python lines theme={"system"}
faker = MyFaker()
text_to_anonymize = (
"My name is Raphael and I like to fish. My phone number is 212-555-5555"
)
anonymized_text = faker.redact_and_anonymize_with_faker(text_to_anonymize)
print(f"Raw text:\n\t{text_to_anonymize}")
print(f"Anonymized text:\n\t{anonymized_text}")
```
方法 4: `autopatch_settings` を使用する
`autopatch_settings` を使用すると、サポートされる 1 つ以上の LLM インテグレーションについて、初期化時に PII の処理を直接設定できます。特定のインテグレーションに対するすべての Call で PII の処理を一元管理したい場合は、この方法をお勧めします。この方法には次の利点があります。
1. PII の処理ロジックが初期化時に一元化され、その適用範囲も初期化時に決まるため、カスタムロジックを各所に分散させる必要が少なくなります。
2. 特定のインテグレーションについて、PII の処理ワークフローをカスタマイズしたり、完全に無効にしたりできます。
`autopatch_settings` を使用して PII の処理を設定するには、サポートされるいずれかの LLM インテグレーションの `op_settings` で `postprocess_inputs` または `postprocess_output` を定義します。
```python lines theme={"system"}
def postprocess(inputs: dict) -> dict:
if "SENSITIVE_KEY" in inputs:
inputs["SENSITIVE_KEY"] = "REDACTED"
return inputs
client = weave.init(
...,
autopatch_settings={
"openai": {
"op_settings": {
"postprocess_inputs": postprocess,
"postprocess_output": ...,
}
},
"anthropic": {
"op_settings": {
"postprocess_inputs": ...,
"postprocess_output": ...,
}
}
},
)
```
メソッドを Weave の Call に適用する
ここまでで各マスキング方法を個別に確認しました。次のサンプルでは、これらを Weave モデルに統合し、Weave トレースで結果をプレビューする方法を紹介します。
まず、[Weave モデル](/ja/products/wandb/weave/guides/core-types/models)を作成します。Weave モデルは、設定、モデルの重み、モデルの動作を定義するコードなどの情報をまとめたものです。
このモデルには、Anthropic API を呼び出す predict 関数が含まれています。Anthropic の Claude Sonnet が感情分析を行い、その際の LLM Call は [トレース](/ja/products/wandb/weave/quickstart) を使用してトレースされます。Claude Sonnet はテキストを受け取り、*positive*、*negative*、*neutral* のいずれかの感情分類を出力します。また、このモデルには、PII データが LLM に送信される前に確実にマスクまたは匿名化されるよう、後処理関数も含まれています。
このコードを実行すると、Weave プロジェクトのページと、実行した特定のトレース (LLM Call) へのリンクが表示されます。これらのリンクから、入力が LLM に到達する前に後処理関数によって期待どおりマスクまたは匿名化されていることを確認してください。
正規表現による方法
手始めに、正規表現を使用して元のテキストから PII データを特定し、マスクできます。
```python lines theme={"system"}
import json
from typing import Any
import anthropic
import weave
# モデルの予測を行う Weave Op に正規表現によるマスキングを適用する、入力の後処理関数を定義します
def postprocess_inputs_regex(inputs: dict[str, Any]) -> dict:
inputs["text_block"] = redact_with_regex(inputs["text_block"])
return inputs
# Weave モデル / predict 関数
class SentimentAnalysisRegexPiiModel(weave.Model):
model_name: str
system_prompt: str
temperature: int
@weave.op(
postprocess_inputs=postprocess_inputs_regex,
)
async def predict(self, text_block: str) -> dict:
client = anthropic.AsyncAnthropic()
response = await client.messages.create(
max_tokens=1024,
model=self.model_name,
system=self.system_prompt,
messages=[
{"role": "user", "content": [{"type": "text", "text": text_block}]}
],
)
result = response.content[0].text
if result is None:
raise ValueError("No response from model")
parsed = json.loads(result)
return parsed
```
```python lines theme={"system"}
# システムプロンプトを指定して LLM モデルを作成します
model = SentimentAnalysisRegexPiiModel(
name="claude-3-sonnet",
model_name="claude-3-5-sonnet-20240620",
system_prompt='You are a Sentiment Analysis classifier. You will be classifying text based on their sentiment. Your input will be a block of text. You will answer with one the following rating option["positive", "negative", "neutral"]. Your answer should be one word in json format: {classification}. Ensure that it is valid JSON.',
temperature=0,
)
print("Model: ", model)
# 各テキストブロックを匿名化してから予測を実行します
for entry in pii_data:
await model.predict(entry["text"])
```
Presidio のマスキング方法
次に、Presidio を使用して、元のテキストから PII データを特定してマスクします。
```python lines theme={"system"}
from typing import Any
import weave
# モデルの予測を行う Weave Op の入力に Presidio によるマスキングを適用する後処理関数を定義します
def postprocess_inputs_presidio(inputs: dict[str, Any]) -> dict:
inputs["text_block"] = redact_with_presidio(inputs["text_block"])
return inputs
# Weave モデル / predict 関数
class SentimentAnalysisPresidioPiiModel(weave.Model):
model_name: str
system_prompt: str
temperature: int
@weave.op(
postprocess_inputs=postprocess_inputs_presidio,
)
async def predict(self, text_block: str) -> dict:
client = anthropic.AsyncAnthropic()
response = await client.messages.create(
max_tokens=1024,
model=self.model_name,
system=self.system_prompt,
messages=[
{"role": "user", "content": [{"type": "text", "text": text_block}]}
],
)
result = response.content[0].text
if result is None:
raise ValueError("No response from model")
parsed = json.loads(result)
return parsed
```
```python lines theme={"system"}
# システムプロンプトを指定して LLM モデルを作成します
model = SentimentAnalysisPresidioPiiModel(
name="claude-3-sonnet",
model_name="claude-3-5-sonnet-20240620",
system_prompt='You are a Sentiment Analysis classifier. You will be classifying text based on their sentiment. Your input will be a block of text. You will answer with one the following rating option["positive", "negative", "neutral"]. Your answer should be one word in json format: {classification}. Ensure that it is valid JSON.',
temperature=0,
)
print("Model: ", model)
# 各テキストブロックを匿名化してから予測を実行します
for entry in pii_data:
await model.predict(entry["text"])
```
Faker と Presidio による置換手法
この例では、Faker を使用して匿名化された置換用の PII データを生成し、Presidio を使用して元のテキスト内の PII データを特定して置換します。
```python lines theme={"system"}
from typing import Any
import weave
# モデルの予測用 Weave Op に対して、Faker による匿名化と Presidio によるマスキングを適用する入力後処理関数を定義します
faker = MyFaker()
def postprocess_inputs_faker(inputs: dict[str, Any]) -> dict:
inputs["text_block"] = faker.redact_and_anonymize_with_faker(inputs["text_block"])
return inputs
# Weave モデル / predict 関数
class SentimentAnalysisFakerPiiModel(weave.Model):
model_name: str
system_prompt: str
temperature: int
@weave.op(
postprocess_inputs=postprocess_inputs_faker,
)
async def predict(self, text_block: str) -> dict:
client = anthropic.AsyncAnthropic()
response = await client.messages.create(
max_tokens=1024,
model=self.model_name,
system=self.system_prompt,
messages=[
{"role": "user", "content": [{"type": "text", "text": text_block}]}
],
)
result = response.content[0].text
if result is None:
raise ValueError("No response from model")
parsed = json.loads(result)
return parsed
```
```python lines theme={"system"}
# システムプロンプトを指定して LLM モデルを作成します
model = SentimentAnalysisFakerPiiModel(
name="claude-3-sonnet",
model_name="claude-3-5-sonnet-20240620",
system_prompt='You are a Sentiment Analysis classifier. You will be classifying text based on their sentiment. Your input will be a block of text. You will answer with one the following rating option["positive", "negative", "neutral"]. Your answer should be one word in json format: {classification}. Ensure that it is valid JSON.',
temperature=0,
)
print("Model: ", model)
# テキストブロックごとに、まず匿名化してから予測を実行します
for entry in pii_data:
await model.predict(entry["text"])
```
`autopatch_settings` メソッド
次の例では、初期化時に `anthropic` の `postprocess_inputs` として `postprocess_inputs_regex()` 関数を設定しています。`postprocess_inputs_regex` 関数は、[方法 1: 正規表現を使用してフィルターする](#method-1-filter-using-regular-expressions) で定義した `redact_with_regex` メソッドを適用します。これにより、`anthropic` のすべてのモデルへの入力すべてに `redact_with_regex` が適用されます。
```python lines theme={"system"}
from typing import Any
import weave
client = weave.init(
...,
autopatch_settings={
"anthropic": {
"op_settings": {
"postprocess_inputs": postprocess_inputs_regex,
}
}
},
)
# モデル予測用の Weave Op に対して、正規表現によるマスキングを適用する入力後処理関数を定義します
def postprocess_inputs_regex(inputs: dict[str, Any]) -> dict:
inputs["text_block"] = redact_with_regex(inputs["text_block"])
return inputs
# Weave モデル / predict 関数
class SentimentAnalysisRegexPiiModel(weave.Model):
model_name: str
system_prompt: str
temperature: int
async def predict(self, text_block: str) -> dict:
client = anthropic.AsyncAnthropic()
response = await client.messages.create(
max_tokens=1024,
model=self.model_name,
system=self.system_prompt,
messages=[
{"role": "user", "content": [{"type": "text", "text": text_block}]}
],
)
result = response.content[0].text
if result is None:
raise ValueError("No response from model")
parsed = json.loads(result)
return parsed
```
```python lines theme={"system"}
# システムプロンプトを指定して LLM モデルを作成する
model = SentimentAnalysisRegexPiiModel(
name="claude-3-sonnet",
model_name="claude-3-5-sonnet-20240620",
system_prompt='You are a Sentiment Analysis classifier. You will be classifying text based on their sentiment. Your input will be a block of text. You will answer with one the following rating option["positive", "negative", "neutral"]. Your answer should be one word in json format: {classification}. Ensure that it is valid JSON.',
temperature=0,
)
print("Model: ", model)
# 各テキストブロックを先に匿名化してから予測する
for entry in pii_data:
await model.predict(entry["text"])
```
オプション: データを暗号化する
PII の匿名化に加えて、cryptography ライブラリの [Fernet](https://cryptography.io/en/latest/fernet/) 共通鍵暗号でデータを暗号化すると、セキュリティをさらに強化できます。こうしておけば、匿名化したデータが傍受されても、暗号化キーがない限り読み取られることはありません。次の例では、入力テキストをログする前に暗号化し、モデルの `predict` メソッド内で復号する方法を示します。
```python lines theme={"system"}
import os
from cryptography.fernet import Fernet
from pydantic import BaseModel, ValidationInfo, model_validator
def get_fernet_key():
# 環境変数にキーが存在するかを確認します
key = os.environ.get('FERNET_KEY')
if key is None:
# キーが存在しない場合は、新しいキーを生成します
key = Fernet.generate_key()
# キーを環境変数に保存します
os.environ['FERNET_KEY'] = key.decode()
else:
# キーが存在する場合は、bytes 型に変換します
key = key.encode()
return key
cipher_suite = Fernet(get_fernet_key())
class EncryptedSentimentAnalysisInput(BaseModel):
encrypted_text: str = None
@model_validator(mode="before")
def encrypt_fields(cls, values):
if "text" in values and values["text"] is not None:
values["encrypted_text"] = cipher_suite.encrypt(values["text"].encode()).decode()
del values["text"]
return values
@property
def text(self):
if self.encrypted_text:
return cipher_suite.decrypt(self.encrypted_text.encode()).decode()
return None
@text.setter
def text(self, value):
self.encrypted_text = cipher_suite.encrypt(str(value).encode()).decode()
@classmethod
def encrypt(cls, text: str):
return cls(text=text)
def decrypt(self):
return self.text
# 新しい EncryptedSentimentAnalysisInput を使用するよう変更した sentiment_analysis_model
class sentiment_analysis_model(weave.Model):
model_name: str
system_prompt: str
temperature: int
@weave.op()
async def predict(self, encrypted_input: EncryptedSentimentAnalysisInput) -> dict:
client = AsyncAnthropic()
decrypted_text = encrypted_input.decrypt() # カスタムクラスを使用してテキストを復号します
response = await client.messages.create(
max_tokens=1024,
model=self.model_name,
system=self.system_prompt,
messages=[
{ "role": "user",
"content":[
{
"type": "text",
"text": decrypted_text
}
]
}
]
)
result = response.content[0].text
if result is None:
raise ValueError("No response from model")
parsed = json.loads(result)
return parsed
model = sentiment_analysis_model(
name="claude-3-sonnet",
model_name="claude-3-5-sonnet-20240620",
system_prompt="You are a Sentiment Analysis classifier. You will be classifying text based on their sentiment. Your input will be a block of text. You will answer with one the following rating option[\"positive\", \"negative\", \"neutral\"]. Your answer should one word in json format dict where the key is classification.",
temperature=0
)
for entry in pii_data:
encrypted_input = EncryptedSentimentAnalysisInput.encrypt(entry["text"])
await model.predict(encrypted_input)
```