### 방법 2: Microsoft Presidio를 사용해 PII 마스킹하기
다음 방법은 [Microsoft Presidio](https://microsoft.github.io/presidio/)를 사용해 PII 데이터를 완전히 제거하는 것입니다. Presidio는 PII를 마스킹한 뒤 PII 유형을 나타내는 플레이스홀더로 대체합니다. 예를 들어, Presidio는 `"My name is Alex"`에서 `Alex`를 `
### 방법 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를 함께 사용하려면 맞춤형 operator에 대한 참조를 제공해야 합니다. 이 operator는 PII를 가짜 데이터로 바꾸는 역할을 하는 Faker 함수로 Presidio를 연결해 줍니다.
```python lines theme={null}
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 엔티티에 대한 맞춤형 operator 생성
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()
# 위에서 정의한 operators를 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}")
```
코드를 하나의 클래스로 통합하고 entity 목록에 앞서 식별한 추가 항목도 포함되도록 확장하려면 다음을 실행하세요:
```python lines theme={null}
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()
# entity에 대한 맞춤형 operator 생성
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={null}
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` 사용
하나 이상의 지원되는 LLM 인테그레이션에서 초기화 시점에 직접 PII 처리를 구성하려면 `autopatch_settings`를 사용할 수 있습니다. W\&B는 특정 인테그레이션의 모든 Call 전반에서 PII 처리를 중앙 집중식으로 관리하려는 경우 이 접근 방식을 권장합니다. 이 방법의 장점은 다음과 같습니다.
1. PII 처리 로직을 초기화 시점에 중앙에서 관리하고 범위를 지정할 수 있어, 여기저기 흩어진 맞춤형 로직의 필요성이 줄어듭니다.
2. 특정 인테그레이션에 대해 PII 처리 워크플로를 사용자 지정하거나 완전히 비활성화할 수 있습니다.
`autopatch_settings`를 사용해 PII 처리를 구성하려면, 지원되는 LLM 인테그레이션 중 하나의 `op_settings`에서 `postprocess_inputs` 또는 `postprocess_output`을 정의하세요.
```python lines theme={null}
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](https://docs.wandb.ai/weave/guides/core-types/models)을 만드세요. Weave Model은 설정, 모델 가중치, 그리고 모델이 어떻게 동작하는지를 정의하는 코드 같은 정보를 조합한 것입니다.
이 모델에는 Anthropic API를 호출하는 predict 함수가 포함되어 있습니다. Anthropic의 Claude Sonnet은 [트레이스](https://docs.wandb.ai/weave/quickstart)를 사용해 LLM Call을 트레이싱하면서 감성 분석을 수행합니다. Claude Sonnet은 텍스트 블록을 입력으로 받아 다음 감성 분류 중 하나를 출력합니다: *긍정*, *부정*, 또는 *중립*. 또한 이 모델에는 LLM으로 전송되기 전에 PII 데이터가 마스킹되거나 익명화되도록 하는 후처리 함수도 포함되어 있습니다.
이 코드를 실행하면 Weave 프로젝트 페이지 링크와 함께, 실행한 특정 트레이스(LLM Call) 링크도 받게 됩니다. 이 링크를 사용해 입력이 LLM에 도달하기 전에 후처리 함수가 예상대로 마스킹되거나 익명화되었는지 확인하세요.
### Regex 방법
시작점으로 정규식을 사용해 원본 텍스트에서 PII 데이터를 파악하고 마스킹할 수 있습니다.
```python lines theme={null}
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 모델 / 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={null}
# 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={null}
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={null}
# 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={null}
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={null}
# 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={null}
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={null}
# 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={null}
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)
```