메인 콘텐츠로 건너뛰기
이것은 대화형 노트북입니다. 로컬에서 실행하거나 아래 링크를 사용할 수 있습니다:
이 가이드에서는 Personally Identifiable Information (PII) 데이터의 프라이버시를 보호하면서 W&B Weave를 사용하는 방법을 알아봅니다. PII를 보호하면 LLM Tracing 및 Evaluation의 이점을 계속 활용하면서도 개인정보 보호 및 규정 준수 요구 사항을 충족하는 데 도움이 됩니다. 이 가이드는 민감한 사용자 데이터를 처리하는 LLM 애플리케이션에 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 Call에 적용합니다.

​
Overview

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

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

Weave Ops를 사용하면 입력 및 출력 후처리 함수를 정의할 수 있습니다. 이 함수들을 통해 LLM Call에 전달되거나 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에는 단방향 해시를 적용하세요.
  • 암호화된 상태에서 분석해야 하는 데이터에는 특수한 암호화 방식을 고려하세요.

​
사전 요구 사항

마스킹 방법을 적용하기 전에 의존성을 설치하고, 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                          # PII 데이터를 가짜 데이터로 대체하기 위해 Faker를 사용합니다
!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"] + '"')

​
마스킹 방법 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._%+-]+ : 이메일 사용자 이름에 올 수 있는 문자 (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를 사용해 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 엔티티에 대한 맞춤형 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 목록에 앞서 식별한 추가 항목도 포함되도록 확장하려면 다음을 실행하세요: 
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를 사용할 수 있습니다. 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은 트레이스를 사용해 LLM Call을 트레이싱하면서 감성 분석을 수행합니다. Claude Sonnet은 텍스트 블록을 입력으로 받아 다음 감성 분류 중 하나를 출력합니다: 긍정, 부정, 또는 중립. 또한 이 모델에는 LLM으로 전송되기 전에 PII 데이터가 마스킹되거나 익명화되도록 하는 후처리 함수도 포함되어 있습니다. 이 코드를 실행하면 Weave 프로젝트 페이지 링크와 함께, 실행한 특정 트레이스(LLM Call) 링크도 받게 됩니다. 이 링크를 사용해 입력이 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 모델 / 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

# 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 데이터를 파악하고 마스킹하세요.
파악된 PII entity와 마스킹된 텍스트 출력을 보여주는 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 모델 / 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

# 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 데이터를 파악해 대체합니다.
원본 텍스트, 파악된 PII, 익명화된 대체 값이 포함된 Faker 및 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

# 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: 정규 표현식을 사용한 필터링에 정의된 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 모델 / 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

# 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 데이터 암호화 프로세스
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)