PII 데이터 처리

W&B Weave에서 PII 데이터 처리 기능을 사용하는 방법을 알아봅니다

이 노트북은 대화형입니다. 로컬에서 실행하거나 아래 링크를 통해 열 수 있습니다:

• Google Colab에서 열기
• GitHub에서 소스 보기

​

​

PII 데이터와 함께 Weave를 사용하는 방법

이 가이드에서는 W&B Weave를 사용할 때 개인 식별 정보(PII) 데이터가 비공개로 유지되도록 하는 방법을 설명합니다. 이 가이드는 PII 데이터를 식별, 마스킹 및 익명화하는 다음 방법을 보여줍니다:

1. PII 데이터를 식별하고 마스킹하기 위한 정규 표현식(regular expressions).
2. 파이썬 기반 데이터 보호 SDK인 Microsoft의 Presidio. 이 도구는 마스킹 및 대체 기능을 제공합니다.
3. 가짜 데이터를 생성하는 Python 라이브러리인 **Faker**를 Presidio와 함께 사용하여 PII 데이터를 익명화하는 방법.

또한 weave.op 입력/출력 로깅 커스터마이징 및 autopatch_settings 를 사용하여 PII 마스킹 및 익명화를 워크플로에 통합하는 방법도 배우게 됩니다. 자세한 내용은 로그되는 입력과 출력을 커스터마이즈하기를 참고하세요.

시작하려면 다음을 수행하세요:

1. 개요 섹션을 검토합니다.
2. 사전 준비 사항을 완료합니다.
3. PII 데이터를 식별, 마스킹 및 익명화하기 위한 사용 가능한 방법을 검토합니다.
4. Weave 호출에 이 방법들을 적용합니다.

​

개요

다음 섹션에서는 weave.op을 사용한 입력 및 출력 로깅에 대한 개요와, Weave에서 PII 데이터를 다룰 때의 모범 사례를 설명합니다.

​

weave.op을 사용해 입력 및 출력 로깅 사용자 지정하기

Weave op을 사용하면 입력 및 출력 후처리 함수를 정의할 수 있습니다. 이러한 함수를 사용해 LLM 호출에 전달되는 데이터나 Weave에 로깅되는 데이터를 수정할 수 있습니다.

다음 예제에서는 두 개의 후처리 함수를 정의하고, 이를 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 Traces로 PII 처리 프로세스를 추적하세요
• 실제 PII를 노출하지 않고 익명화 성능을 측정하세요

​

운영 환경에서

• 원본 PII는 절대 로그에 기록하지 마십시오
• 로그에 기록하기 전에 민감한 필드는 암호화하십시오

​

암호화 팁

• 나중에 복호화해야 하는 데이터에는 복호화 가능한 암호화를 사용하세요
• 다시 복원할 필요가 없는 고유 ID에는 단방향 해시를 적용하세요
• 암호화된 상태로 분석해야 하는 데이터에는 특수한 암호화 방식을 고려하세요

​

사전 준비 사항

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                          # PII 데이터를 가짜 데이터로 대체하기 위해 Faker를 사용합니다
!pip install weave                          # Traces 활용을 위해
!pip install set-env-colab-kaggle-dotenv -q # 환경 변수 설정용
!pip install anthropic                      # sonnet 사용을 위해
!pip install cryptography                   # 데이터 암호화를 위해

2. 다음에서 API 키를 생성합니다:

   • W&B 사용자 설정
   • Anthropic 콘솔

%%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 프로젝트를 초기화합니다.

import weave

# 새 Weave 프로젝트 시작
WEAVE_PROJECT = "pii_cookbook"
weave.init(WEAVE_PROJECT)

4. 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: 정규식을 사용한 필터링

정규 표현식(regular expression, regex)은 PII 데이터를 식별하고 마스킹하는 데 사용할 수 있는 가장 간단한 방법입니다. 정규식을 사용하면 전화번호, 이메일 주소, 사회보장 번호와 같은 민감한 정보의 다양한 형식을 매칭할 수 있는 패턴을 정의할 수 있습니다. 정규식을 활용하면 대용량 텍스트를 대상으로 검색을 수행하면서, 더 복잡한 NLP 기법 없이도 해당 정보를 치환하거나 마스킹할 수 있습니다.

import re

# regex를 사용하여 PII 데이터를 redact하는 함수 정의
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._%+-]+ : 이메일 사용자 이름에 올 수 있는 하나 이상의 문자
    # @          : @ 기호 (리터럴)
    # [A-Za-z0-9.-]+ : 도메인 이름에 올 수 있는 하나 이상의 문자
    # \.         : 점 (리터럴)
    # [A-Z|a-z]{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]      : 대문자 한 글자
    # [a-z]+     : 하나 이상의 소문자
    # \s         : 공백 문자 하나
    # [A-Z]      : 대문자 한 글자
    # [a-z]+     : 하나 이상의 소문자
    # \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 [e-mail], 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 유형을 나타내는 플레이스홀더로 교체합니다. 예를 들어 "My name is Alex"에서 Alex를 <PERSON>으로 교체합니다.

Presidio에는 common entities에 대한 기본 지원이 포함되어 있습니다. 아래 예시에서는 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"원본 텍스트:\n\t{text}")
print(f"삭제된 텍스트:\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를 효과적으로 함께 사용하려면 사용자 정의 오퍼레이터에 대한 참조를 제공해야 합니다. 이 오퍼레이터들은 PII를 가짜 데이터로 교체하는 Faker 함수로 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()

# 위에서 정의한 operators를 익명화 엔진에 전달하는 것을 잊지 마세요
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}")

코드를 하나의 클래스로 통합하고, 앞에서 확인한 추가 항목을 포함하도록 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}")

​

방법 4: autopatch_settings 사용

autopatch_settings를 사용하면 지원되는 하나 이상의 LLM 인테그레이션에 대해 초기화 시점에 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 호출에 메서드 적용하기

다음 예제에서는 PII 마스킹 및 익명화 메서드를 Weave Models에 통합하고, 그 결과를 Weave Traces에서 확인합니다.

먼저 Weave Model을 생성합니다. Weave Model은 설정, 모델 가중치, 그리고 모델이 어떻게 동작하는지를 정의하는 코드와 같은 정보를 결합한 것입니다.

우리 모델에는 Anthropic API를 호출하는 predict 함수를 포함합니다. Anthropic의 Claude Sonnet은 Traces를 사용해 LLM 호출을 추적하면서 감성 분석을 수행하는 데 사용됩니다. Claude Sonnet은 텍스트 블록을 입력으로 받아 positive, negative, neutral 중 하나의 감성 분류를 출력합니다. 또한 PII 데이터가 LLM으로 전송되기 전에 마스킹되거나 익명화되도록 보장하기 위해 후처리 함수를 포함합니다.

이 코드를 실행하면 Weave 프로젝트 페이지에 대한 링크와 함께, 실행한 특정 트레이스(LLM 호출)에 대한 링크를 받게 됩니다.

​

Regex 방식

가장 단순한 방법은 정규식을 사용해 원본 텍스트에서 PII 데이터를 식별하고 마스킹하는 것입니다.

import json
from typing import Any

import anthropic

import weave

# 모델 예측 Weave Op에 regex 리덱션을 적용하는 입력 후처리 함수 정의
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
python
# 시스템 프롬프트로 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 비식별화(redaction) 방법

이제 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
python
# 시스템 프롬프트로 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 데이터를 식별해 대체합니다.

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
# 시스템 프롬프트로 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 함수는 Method 1: Regular Expression Filtering에 정의된 redact_with_regex 메서드를 적용합니다. 이제 redact_with_regex는 모든 anthropic 모델에 전달되는 모든 입력에 적용됩니다.

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
python
# 시스템 프롬프트로 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 대칭 키 암호화를 사용해 데이터에 추가적인 보안 계층을 더할 수 있습니다. 이 방식은 익명화된 데이터가 가로채이더라도 암호 키 없이는 읽을 수 없도록 보장합니다.

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("모델로부터 응답 없음")
        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)