メインコンテンツへスキップ
これはインタラクティブなノートブックです。ローカルで実行することも、以下のリンクを使用することもできます。
このガイドでは、個人を特定できる情報 (PII) データのプライバシーを保ちながら、W&B Weave を使用する方法を学びます。PII を保護することで、LLM のトレースと評価のメリットを活かしながら、プライバシー要件とコンプライアンス要件を満たすことができます。このガイドは、機密性の高いユーザーデータを処理する LLM アプリケーションに Weave を統合する開発者を対象としています。 このガイドでは、PII データを特定し、マスクし、匿名化するための以下の方法を紹介します。
  1. PII データを特定してマスクするための正規表現。
  2. Python ベースのデータ保護 SDK である Microsoft の Presidio。このツールでは、マスキングと置換の機能を利用できます。
  3. 偽データを生成する Python ライブラリ Faker。Presidio と組み合わせることで、PII データを匿名化できます。
さらに、PII のマスキングと匿名化をワークフローに統合するために、weave.op input/output logging customization と autopatch_settings の使い方も学びます。詳細は、Customize logged inputs and outputsを参照してください。 開始するには、以下の手順に従ってください。
  1. 概要 セクションを確認します。
  2. prerequisites を完了します。
  3. PII データの特定、マスキング、匿名化に使用できる available methods を確認します。
  4. Apply the methods to Weave calls を確認します。

​
概要

以下のセクションでは、weave.op を使用した入出力のロギングの概要と、Weave で PII データを扱う際のベストプラクティスを紹介します。

​
weave.op を使用して入出力のロギングをカスタマイズする

Weave Ops では、入力と出力の後処理関数を定義できます。これらの関数を使うと、LLM Call に渡すデータや Weave にログするデータを変更できます。 次の例では、2 つの後処理関数を定義し、それらを weave.op() の引数として渡します。 
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. まず、必要なパッケージをインストールします。
%%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                   # データを暗号化するため
  1. 次の場所でAPIキーを作成します。 
%%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")
  1. Weave プロジェクトを初期化します。
import weave

# 新しい Weave プロジェクトを開始する
WEAVE_PROJECT = "pii_cookbook"
weave.init(WEAVE_PROJECT)
  1. 10 個のテキストブロックを含むデモ用の PII データセットを読み込みます。
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"] + '"')

​
マスキング方法の概要

前提条件 を完了したら、PII データを検出して保護するために、次のいずれかの方法を選択できます。各方法では PII データを特定してマスクし、必要に応じて匿名化します。
  1. 正規表現を使用して PII データを特定し、マスクします。
  2. Microsoft Presidio。マスキングと置換の機能を備えた、Python ベースのデータ保護 SDK です。
  3. Faker。偽データを生成するための Python ライブラリです。

​
方法 1: 正規表現を使用してフィルタリングする

正規表現 (regex) は、PII データを特定してマスクするためのシンプルな方法です。regex を使うと、電話番号、メールアドレス、社会保障番号などの機密情報のさまざまな形式に一致するパターンを定義できます。regex を使えば、より複雑な NLP 手法を使わなくても、大量のテキストをスキャンして情報を置換またはマスクできます。 
import re

# regexを使用して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", "<PHONE>", 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", "<EMAIL>", 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", "<SSN>", 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", "<NAME>", text)

    return text
関数をテストするには、サンプルテキストを使って次を実行します。 
# 関数をテストする
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 を使用して PII データを完全に除去します。Presidio は PII をマスクし、PII のタイプを表すプレースホルダーに置き換えます。たとえば、Presidio は "My name is Alex" 内の Alex を <PERSON> に置き換えます。 Presidio では、一般的なエンティティ が標準でサポートされています。以下の例では、PHONE_NUMBER、PERSON、LOCATION、EMAIL_ADDRESS、US_SSN のいずれかに該当するすべてのエンティティをマスクします。Presidio による処理は関数にまとめられています。 
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

# Analyzer を設定します。NLP モジュール（デフォルトでは spaCy モデル）およびその他の PII 認識エンジンを読み込みます。
analyzer = AnalyzerEngine()

# Anonymizer を設定します。アナライザーの結果を使用してテキストを匿名化します。
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
この関数をテストするには、サンプルテキストを使って以下を実行します： 
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 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 を一緒に使用するには、custom operator への参照を指定する必要があります。これらの operator は、PII をダミーデータに置き換える Faker の function を Presidio が使用するように指定するものです。 
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_results = analyzer.analyze(
    text=text_to_anonymize, entities=["PHONE_NUMBER", "PERSON"], language="en"
)

anonymizer = AnonymizerEngine()

# 上記のオペレーターをアノニマイザーに渡すことを忘れずに
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 つのクラスに統合し、Entities のリストに先ほど特定した追加項目を含めるには、次を実行します。 
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
関数をテストするには、サンプルテキストを使って以下を実行します。 
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}")

​
Method 4: autopatch_settings を使用する

autopatch_settings を使用すると、サポートされる1つ以上のLLMインテグレーションについて、初期化時にPII処理を直接設定できます。W&Bでは、特定のインテグレーションのすべてのCallにわたってPII処理を一元化したい場合、この方法を推奨しています。この方法の利点は次のとおりです。
  1. PII処理ロジックを初期化時に一元化して適用できるため、各所に分散したカスタムロジックの必要性を減らせます。
  2. PII処理のワークフローは、特定のインテグレーションごとにカスタマイズしたり、完全に無効化したりできます。
autopatch_settings を使用してPII処理を設定するには、サポートされるLLMインテグレーションのいずれかについて、op_settings で postprocess_inputs または postprocess_output を定義します。 

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 Models に統合し、結果を Weave トレース で確認する方法を示します。 まず、Weave Model を作成します。Weave Model は、設定、モデルの重み、モデルの動作を定義するコードなどの情報を組み合わせたものです。 このモデルには、Anthropic API を呼び出す predict 関数が含まれています。Anthropic の Claude Sonnet は、Traces を使用して LLM Call をトレースしながら、感情分析を実行します。Claude Sonnet はテキストブロックを受け取り、positive、negative、neutral のいずれかの感情分類を出力します。また、このモデルには、PII データが LLM に送信される前にマスクまたは匿名化されるようにするための後処理関数も含まれています。 このコードを実行すると、Weave プロジェクトページへのリンクと、実行した特定のトレース (LLM Call) へのリンクが表示されます。これらのリンクを使用して、入力が LLM に到達する前に、後処理関数によって想定どおりにマスクまたは匿名化されていることを確認してください。

​
Regex による方法

まずは、Regex を使用して元のテキスト内の PII データを特定し、マスクできます。 
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 モデル / 予測関数
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

# システムプロンプトを使って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 データを特定し、マスクします。
特定された PII Entities と、マスクされたテキスト output を示す Presidio の PII マスキング プロセス
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 モデル / 予測関数
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

# システムプロンプトを使用して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 データを検出して置換します。
元のテキスト、検出された PII、匿名化された置換値を含む Faker と Presidio の PII 置換プロセス
from typing import Any

import weave

# Faker による匿名化と Presidio によるマスキングをモデル予測 Weave Op に適用する入力後処理関数を定義する
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

# システムプロンプトを使用して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: 正規表現を使用したフィルター で定義した redact_with_regex メソッドを適用します。その結果、すべての anthropic モデルへの入力に redact_with_regex が適用されます。 
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 モデル / 予測関数
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

# system promptを使って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 データ暗号化プロセス
PII の匿名化に加えて、cryptography ライブラリの Fernet 対称暗号化を使ってデータを暗号化すると、セキュリティをさらに強化できます。この方法では、匿名化したデータが傍受された場合でも、暗号化キーがなければ内容を読み取れません。次の例では、入力テキストをログする前に暗号化し、モデルの predict メソッド内で復号する方法を示します。 
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:
        # キーが存在する場合、バイト型であることを確認する
        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)