PII 데이터 처리

W&B Weave의 데이터 마스킹 및 필터링 도구를 사용해 LLM 애플리케이션에서 개인 식별 정보(PII)를 처리합니다.

이 콘텐츠는 대화형 노트북입니다. 로컬에서 실행하거나 아래 링크를 사용할 수 있습니다:

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

​

​

PII 데이터로 Weave를 사용하는 방법

이 가이드에서는 Personally Identifiable Information (PII) 데이터의 프라이버시를 보호하면서 W&B Weave를 사용하는 방법을 알아봅니다. 이 가이드에서는 PII 데이터를 식별하고, 마스킹하고, 익명화하는 다음 방법을 보여줍니다:

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

또한 weave.op 입력/출력 로깅 사용자 지정 및 _autopatch_settings_를 사용해 워크플로에 PII 마스킹과 익명화를 통합하는 방법도 알아봅니다. 자세한 내용은 Customize logged inputs and outputs를 참조하세요.

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

1. Overview 섹션을 검토합니다.
2. 사전 요구 사항을 완료합니다.
3. PII 데이터를 식별하고, 마스킹하고, 익명화하는 사용 가능한 방법을 검토합니다.
4. 이 방법을 Weave calls에 적용합니다.

​

Overview

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

​

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

Weave Ops를 사용하면 입력값과 출력값에 대한 후처리 함수를 정의할 수 있습니다. 이러한 함수를 사용하면 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 트레이스로 PII 처리 과정을 추적합니다
• 실제 PII를 노출하지 않고도 익명화 성능을 측정합니다

​

프로덕션 환경에서

• 원본 PII는 절대 기록하지 마세요
• 기록하기 전에 민감한 필드는 암호화하세요

​

암호화 팁

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

​

사전 요구 사항

1. 먼저 필요한 패키지를 설치하세요.

%%capture
# @title 필수 파이썬 패키지:
!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                          # 트레이스 활용을 위해
!pip install set-env-colab-kaggle-dotenv -q # 환경 변수 설정용
!pip install anthropic                      # sonnet 사용을 위해
!pip install cryptography                   # 데이터 암호화를 위해

2. 다음 위치에서 API 키를 생성하세요:

   • W&B User Settings
   • Anthropic Console

%%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"] + '"')

​

마스킹 방법 Overview

설정을 완료했다면 다음을 수행할 수 있습니다

PII 데이터를 탐지하고 보호하기 위해 다음 방법으로 PII 데이터를 파악해 마스킹하고, 필요에 따라 익명화할 수 있습니다:

1. PII 데이터를 파악하고 마스킹하는 정규 표현식.
2. 마스킹 및 대체 기능을 제공하는 Python 기반 데이터 보호 SDK인 Microsoft Presidio.
3. 가짜 데이터를 생성하는 Python 라이브러리인 Faker.

​

방법 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._%+-]+ : 이메일 사용자 이름에 올 수 있는 하나 이상의 문자
    # @          : 리터럴 @ 기호
    # [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를 사용해 PII 마스킹하기

다음 방법은 Microsoft Presidio를 사용해 PII 데이터를 완전히 제거하는 것입니다. Presidio는 PII를 마스킹한 뒤 PII 유형을 나타내는 플레이스홀더로 대체합니다. 예를 들어, Presidio는 "My name is Alex"에서 Alex를 <PERSON>으로 바꿉니다.

Presidio는 일반적인 entity를 기본적으로 지원합니다. 아래 예시에서는 PHONE_NUMBER, PERSON, LOCATION, EMAIL_ADDRESS, US_SSN인 모든 entity를 마스킹합니다. 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를 함께 효과적으로 사용하려면 맞춤형 Operator에 대한 레퍼런스를 제공해야 합니다. 이러한 Operator는 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 entity에 대한 맞춤형 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_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}")

코드를 하나의 클래스로 통합하고, 앞서 식별한 추가 항목까지 포함하도록 entity 목록을 확장해 보겠습니다.

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

샘플 텍스트를 사용해 함수를 테스트해 보겠습니다:

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를 사용할 수 있습니다. 이 방법의 장점은 다음과 같습니다.

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 calls에 적용하기

다음 예제에서는 PII 마스킹 및 익명화 방법을 Weave Models에 통합하고, Weave Traces에서 결과를 미리 확인합니다.

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

이 모델에는 Anthropic API를 호출하는 predict 함수를 포함합니다. Anthropic의 Claude Sonnet은 Traces를 사용해 LLM calls를 트레이싱하면서 감성 분석을 수행합니다. Claude Sonnet은 텍스트 블록을 입력받아 다음 감성 분류 중 하나를 출력합니다: 긍정, 부정, 또는 중립. 또한 PII 데이터가 LLM으로 전송되기 전에 마스킹되거나 익명화되도록 후처리 함수도 포함합니다.

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

​

Regex 방법

가장 단순한 경우에는 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 모델 / 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("모델로부터 응답 없음")
        parsed = json.loads(result)
        return parsed
python
# 시스템 프롬프트로 LLM 모델 생성
model = SentimentAnalysisRegexPiiModel(
    name="claude-3-sonnet",
    model_name="claude-3-5-sonnet-20240620",
    system_prompt='당신은 감성 분석 분류기입니다. 텍스트를 감성에 따라 분류합니다. 입력은 텍스트 블록입니다. 다음 평가 옵션 중 하나로 답하세요["positive", "negative", "neutral"]. 답변은 json 형식의 한 단어여야 합니다: {classification}. 유효한 JSON인지 확인하세요.',
    temperature=0,
)

print("Model: ", model)
# 각 텍스트 블록에 대해 먼저 익명화한 후 예측 수행
for entry in pii_data:
    await model.predict(entry["text"])

​

Presidio 마스킹 방법

다음으로, 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("모델로부터 응답 없음")
        parsed = json.loads(result)
        return parsed
python
# 시스템 프롬프트로 LLM 모델 생성
model = SentimentAnalysisPresidioPiiModel(
    name="claude-3-sonnet",
    model_name="claude-3-5-sonnet-20240620",
    system_prompt='당신은 감성 분석 분류기입니다. 텍스트를 감성에 따라 분류합니다. 입력은 텍스트 블록이며, 다음 평가 옵션 중 하나로 답변하세요["positive", "negative", "neutral"]. 답변은 json 형식의 한 단어여야 합니다: {classification}. 유효한 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 함수는 방법 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("모델로부터 응답 없음")
        parsed = json.loads(result)
        return parsed
python
# 시스템 프롬프트로 LLM 모델 생성
model = SentimentAnalysisRegexPiiModel(
    name="claude-3-sonnet",
    model_name="claude-3-5-sonnet-20240620",
    system_prompt='당신은 감성 분석 분류기입니다. 텍스트를 감성에 따라 분류합니다. 입력은 텍스트 블록이며, 다음 평가 옵션 중 하나로 답변하세요["positive", "negative", "neutral"]. 답변은 json 형식의 한 단어여야 합니다: {classification}. 유효한 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="당신은 감성 분석 분류기입니다. 텍스트의 감성을 기반으로 분류합니다. 입력은 텍스트 블록이며, 다음 평가 옵션 중 하나로 답변하세요[\"positive\", \"negative\", \"neutral\"]. 답변은 키가 classification인 json 형식의 dict에서 한 단어여야 합니다.",
    temperature=0
)

for entry in pii_data:
    encrypted_input = EncryptedSentimentAnalysisInput.encrypt(entry["text"])
    await model.predict(encrypted_input)