> ## 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 마스킹 및 익명화를 워크플로에 통합하는 방법도 알아봅니다. 자세한 내용은 [로깅되는 입력 및 출력 사용자 지정하기](/ko/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에 로깅되는 데이터를 수정할 수 있습니다.
다음 예시에서는 후처리 함수 두 개를 정의하고 `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 Traces로 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: 정규 표현식을 사용한 필터링
[정규 표현식(정규식)](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 유형을 나타내는 자리 표시자로 대체합니다. 예를 들어 Presidio는 `"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를 함께 사용하려면 맞춤형 Operator에 대한 참조를 제공해야 합니다. 이 Operator는 Presidio가 PII를 가짜 데이터로 바꾸는 Faker 함수를 호출하도록 연결해 줍니다.
```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" 엔티티용 맞춤형 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()
# 위에서 정의한 Operator를 반드시 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}")
```
코드를 하나의 클래스로 통합하고, 앞서 확인한 추가 엔티티까지 포함하도록 엔티티 목록을 확장하려면 다음을 실행하세요.
```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()
# 각 엔티티에 대한 맞춤형 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={"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`를 사용하면 지원되는 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 Traces에서 결과를 미리 확인하는 방법을 보여 줍니다.
먼저 [Weave 모델](/ko/products/wandb/weave/guides/core-types/models)을 생성하세요. Weave 모델은 설정 값, 모델 가중치, 모델의 동작 방식을 정의하는 코드 등의 정보를 하나로 묶은 것입니다.
이 모델에는 Anthropic API를 호출하는 predict 함수가 포함되어 있습니다. Anthropic의 Claude Sonnet이 감성 분석을 수행하고, [Traces](/ko/products/wandb/weave/quickstart)로 LLM Call을 트레이싱합니다. 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"}
# 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"}
# 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 Model / 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"}
# 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"}
# 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:
# 키가 있으면 바이트 형식으로 변환합니다
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)
```