트레이스에서 민감한 데이터 로깅 방지

LANGSMITH_HIDE_INPUTS=true
LANGSMITH_HIDE_OUTPUTS=true

from langsmith import traceable

def process_inputs(inputs: dict) -> dict:
    # inputs는 키가 인수 이름이고 값이 제공된 인수인 딕셔너리입니다
    # 처리된 입력이 포함된 새 딕셔너리를 반환합니다
    return {
        "processed_key": inputs.get("my_cool_key", "default"),
        "length": len(inputs.get("my_cool_key", ""))
    }

def process_outputs(output: Any) -> dict:
    # output은 함수의 직접 반환 값입니다
    # 출력을 딕셔너리로 변환합니다
    # 이 경우 "output"은 정수입니다
    return {"processed_output": str(output)}

@traceable(process_inputs=process_inputs, process_outputs=process_outputs)
def my_function(my_cool_key: str) -> int:
    # 함수 구현
    return len(my_cool_key)

result = my_function("example")

@traceable(process_inputs=process_inputs, process_outputs=process_outputs)
async def async_function(key: str) -> int:
    # 비동기 구현
    return len(key)

import re
import openai
from langsmith import Client
from langsmith.wrappers import wrap_openai

# 다양한 PII에 대한 정규식 패턴 정의
SSN_PATTERN = re.compile(r'\b\d{3}-\d{2}-\d{4}\b')
CREDIT_CARD_PATTERN = re.compile(r'\b(?:\d[ -]*?){13,16}\b')
EMAIL_PATTERN = re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b')
PHONE_PATTERN = re.compile(r'\b(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b')
FULL_NAME_PATTERN = re.compile(r'\b([A-Z][a-z]*\s[A-Z][a-z]*)\b')

def regex_anonymize(text):
    """
    정규식 패턴을 사용하여 텍스트의 민감한 정보를 익명화합니다.
    Args:
        text (str): 익명화할 입력 텍스트입니다.
    Returns:
        str: 익명화된 텍스트입니다.
    """
    # 민감한 정보를 자리 표시자로 대체합니다
    text = SSN_PATTERN.sub('[REDACTED SSN]', text)
    text = CREDIT_CARD_PATTERN.sub('[REDACTED CREDIT CARD]', text)
    text = EMAIL_PATTERN.sub('[REDACTED EMAIL]', text)
    text = PHONE_PATTERN.sub('[REDACTED PHONE]', text)
    text = FULL_NAME_PATTERN.sub('[REDACTED NAME]', text)
    return text

def recursive_anonymize(data, depth=10):
    """
    데이터 구조를 재귀적으로 순회하여 민감한 정보를 익명화합니다.
    Args:
        data (any): 익명화할 입력 데이터입니다.
        depth (int): 과도한 재귀를 방지하기 위한 현재 재귀 깊이입니다.
    Returns:
        any: 익명화된 데이터입니다.
    """
    if depth == 0:
        return data
    if isinstance(data, dict):
        anonymized_dict = {}
        for k, v in data.items():
            anonymized_value = recursive_anonymize(v, depth - 1)
            anonymized_dict[k] = anonymized_value
        return anonymized_dict
    elif isinstance(data, list):
        anonymized_list = []
        for item in data:
            anonymized_item = recursive_anonymize(item, depth - 1)
            anonymized_list.append(anonymized_item)
        return anonymized_list
    elif isinstance(data, str):
        anonymized_data = regex_anonymize(data)
        return anonymized_data
    else:
        return data

openai_client = wrap_openai(openai.Client())

# 익명화 함수로 LangSmith 클라이언트 초기화
langsmith_client = Client(
    hide_inputs=recursive_anonymize, hide_outputs=recursive_anonymize
)

# 생성된 트레이스에는 메타데이터가 있지만 입력 및 출력은 익명화됩니다
response_with_anonymization = openai_client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "My name is John Doe, my SSN is 123-45-6789, my credit card number is 4111 1111 1111 1111, my email is [email protected], and my phone number is (123) 456-7890."},
    ],
    langsmith_extra={"client": langsmith_client},
)

# 생성된 트레이스에는 익명화되지 않은 입력 및 출력이 있습니다
response_without_anonymization = openai_client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "My name is John Doe, my SSN is 123-45-6789, my credit card number is 4111 1111 1111 1111, my email is [email protected], and my phone number is (123) 456-7890."},
    ],
)

import openai
from langsmith import Client
from langsmith.wrappers import wrap_openai
from presidio_anonymizer import AnonymizerEngine
from presidio_analyzer import AnalyzerEngine

anonymizer = AnonymizerEngine()
analyzer = AnalyzerEngine()

def presidio_anonymize(data):
    """
    사용자가 보내거나 모델이 반환한 민감한 정보를 익명화합니다.
    Args:
        data (any): 익명화할 데이터입니다.
    Returns:
        any: 익명화된 데이터입니다.
    """
    message_list = (
        data.get('messages') or [data.get('choices', [{}])[0].get('message')]
    )
    if not message_list or not all(isinstance(msg, dict) and msg for msg in message_list):
        return data

    for message in message_list:
        content = message.get('content', '')
        if not content.strip():
            print("Empty content detected. Skipping anonymization.")
            continue

        results = analyzer.analyze(
            text=content,
            entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "US_SSN"],
            language='en'
        )
        anonymized_result = anonymizer.anonymize(
            text=content,
            analyzer_results=results
        )
        message['content'] = anonymized_result.text

    return data

openai_client = wrap_openai(openai.Client())

# 익명화 함수로 langsmith 클라이언트 초기화
langsmith_client = Client(
  hide_inputs=presidio_anonymize, hide_outputs=presidio_anonymize
)

# 생성된 트레이스에는 메타데이터가 있지만 입력 및 출력은 익명화됩니다
response_with_anonymization = openai_client.chat.completions.create(
  model="gpt-4o-mini",
  messages=[
      {"role": "system", "content": "You are a helpful assistant."},
      {"role": "user", "content": "My name is Slim Shady, call me at 313-666-7440 or email me at [email protected]"},
  ],
  langsmith_extra={"client": langsmith_client},
)

# 생성된 트레이스에는 익명화되지 않은 입력 및 출력이 있습니다
response_without_anonymization = openai_client.chat.completions.create(
  model="gpt-4o-mini",
  messages=[
      {"role": "system", "content": "You are a helpful assistant."},
      {"role": "user", "content": "My name is Slim Shady, call me at 313-666-7440 or email me at [email protected]"},
  ],
)

import openai
import boto3
from langsmith import Client
from langsmith.wrappers import wrap_openai

comprehend = boto3.client('comprehend', region_name='us-east-1')

def redact_pii_entities(text, entities):
    """
    감지된 엔티티를 기반으로 텍스트의 PII 엔티티를 수정합니다.
    Args:
        text (str): PII를 포함하는 원본 텍스트입니다.
        entities (list): 감지된 PII 엔티티 목록입니다.
    Returns:
        str: PII 엔티티가 수정된 텍스트입니다.
    """
    sorted_entities = sorted(entities, key=lambda x: x['BeginOffset'], reverse=True)
    redacted_text = text
    for entity in sorted_entities:
        begin = entity['BeginOffset']
        end = entity['EndOffset']
        entity_type = entity['Type']
        # 엔티티 유형에 따라 수정 자리 표시자 정의
        placeholder = f"[{entity_type}]"
        # 텍스트의 PII를 자리 표시자로 대체
        redacted_text = redacted_text[:begin] + placeholder + redacted_text[end:]
    return redacted_text

def detect_pii(text):
    """
    AWS Comprehend를 사용하여 주어진 텍스트에서 PII 엔티티를 감지합니다.
    Args:
        text (str): 분석할 텍스트입니다.
    Returns:
        list: 감지된 PII 엔티티 목록입니다.
    """
    try:
        response = comprehend.detect_pii_entities(
            Text=text,
            LanguageCode='en',
        )
        entities = response.get('Entities', [])
        return entities
    except Exception as e:
        print(f"Error detecting PII: {e}")
        return []

def comprehend_anonymize(data):
    """
    사용자가 보내거나 모델이 반환한 민감한 정보를 익명화합니다.
    Args:
        data (any): 익명화할 입력 데이터입니다.
    Returns:
        any: 익명화된 데이터입니다.
    """
    message_list = (
        data.get('messages') or [data.get('choices', [{}])[0].get('message')]
    )
    if not message_list or not all(isinstance(msg, dict) and msg for msg in message_list):
        return data

    for message in message_list:
        content = message.get('content', '')
        if not content.strip():
            print("Empty content detected. Skipping anonymization.")
            continue

        entities = detect_pii(content)
        if entities:
            anonymized_text = redact_pii_entities(content, entities)
            message['content'] = anonymized_text
        else:
            print("No PII detected. Content remains unchanged.")

    return data

openai_client = wrap_openai(openai.Client())

# 익명화 함수로 langsmith 클라이언트 초기화
langsmith_client = Client(
  hide_inputs=comprehend_anonymize, hide_outputs=comprehend_anonymize
)

# 생성된 트레이스에는 메타데이터가 있지만 입력 및 출력은 익명화됩니다
response_with_anonymization = openai_client.chat.completions.create(
  model="gpt-4o-mini",
  messages=[
      {"role": "system", "content": "You are a helpful assistant."},
      {"role": "user", "content": "My name is Slim Shady, call me at 313-666-7440 or email me at [email protected]"},
  ],
  langsmith_extra={"client": langsmith_client},
)

# 생성된 트레이스에는 익명화되지 않은 입력 및 출력이 있습니다
response_without_anonymization = openai_client.chat.completions.create(
  model="gpt-4o-mini",
  messages=[
      {"role": "system", "content": "You are a helpful assistant."},
      {"role": "user", "content": "My name is Slim Shady, call me at 313-666-7440 or email me at [email protected]"},
  ],
)