트레이스에서 민감한 데이터 기록 방지하기

트레이스에서 민감한 데이터 기록 방지하기

LangSmith 트레이스에서 민감한 정보가 기록되는 것을 방지하는 여러 방법을 알려드릴게요.

LangSmith 트레이스로 작업할 때 개인정보를 유지하고 보안 요구사항을 준수하기 위해 민감한 정보가 기록되는 것을 방지해야 할 수 있습니다. LangSmith는 데이터가 백엔드로 전송되기 전에 보호하는 여러 접근 방식을 제공합니다:

참고: 규정 준수나 개인정보 요구사항이 특정 작업을 절대 트레이싱해서는 안 된다고 정한다면(예: zero-retention 정책이 있는 클라이언트) 데이터를 마스킹하는 대신 조건부 트레이싱을 사용해 특정 요청에 대해 트레이싱을 선택적으로 비활성화하는 것을 고려하세요.

출처: 문서

본문

입력과 출력 숨기기

트레이스의 입력과 출력을 완전히 숨기려면 애플리케이션을 실행할 때 다음 환경 변수를 설정할 수 있습니다:

LANGSMITH_HIDE_INPUTS=true
LANGSMITH_HIDE_OUTPUTS=true

이것은 LangSmith SDK(Python과 TypeScript)와 LangChain 모두에서 동작합니다.

주어진 Client 인스턴스에 대해 이 동작을 사용자 지정하고 재정의할 수도 있습니다. Client 객체에 hide_inputs와 hide_outputs 파라미터를 설정하면 됩니다(TypeScript에서는 hideInputs와 hideOutputs).

다음 예제는 hide_inputs와 hide_outputs 모두에 빈 객체를 반환하지만 필요에 맞게 사용자 지정할 수 있습니다:

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

openai_client = wrap_openai(openai.Client())
langsmith_client = Client(
    hide_inputs=lambda inputs: {}, hide_outputs=lambda outputs: {}
)

# The trace produced will have its metadata present, but the inputs will be hidden
openai_client.chat.completions.create(
    model="gpt-5.4-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello!"},
    ],
    langsmith_extra={"client": langsmith_client},
)

# The trace produced will not have hidden inputs and outputs
openai_client.chat.completions.create(
    model="gpt-5.4-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello!"},
    ],
)

import OpenAI from "openai";
import { Client } from "langsmith";
import { wrapOpenAI } from "langsmith/wrappers";

const langsmithClient = new Client({
    hideInputs: (inputs) => ({}),
    hideOutputs: (outputs) => ({}),
});

// The trace produced will have its metadata present, but the inputs will be hidden
const filteredOAIClient = wrapOpenAI(new OpenAI(), {
    client: langsmithClient,
});
await filteredOAIClient.chat.completions.create({
    model: "gpt-5.4-mini",
    messages: [
        { role: "system", content: "You are a helpful assistant." },
        { role: "user", content: "Hello!" },
    ],
});

const openaiClient = wrapOpenAI(new OpenAI());
// The trace produced will not have hidden inputs and outputs
await openaiClient.chat.completions.create({
    model: "gpt-5.4-mini",
    messages: [
        { role: "system", content: "You are a helpful assistant." },
        { role: "user", content: "Hello!" },
    ],
});

메타데이터 숨기기

hide_metadata 파라미터는 LangSmith Python SDK로 트레이싱할 때 실행 메타데이터를 숨길지 변환할지 제어하게 합니다. 메타데이터는 실행을 만들 때 extra 파라미터로 전달됩니다(예: extra={"metadata": {...}}). hide_metadata는 민감한 정보 제거, 개인정보 요구사항 준수, LangSmith로 보내는 데이터 양 줄이기에 유용합니다. 메타데이터 숨기기는 두 가지 방식으로 구성할 수 있습니다:

  • SDK 사용:

    from langsmith import Client

client = Client(hide_metadata=True)

  • 환경 변수 사용:

    export LANGSMITH_HIDE_METADATA=true

hide_metadata 파라미터는 세 가지 유형의 값을 수락합니다:

  • True: 모든 메타데이터를 완전히 제거합니다(빈 딕셔너리 전송).
  • False 또는 None: 메타데이터를 그대로 보존합니다(기본 동작).
  • Callable: 메타데이터 딕셔너리를 변환하는 커스텀 함수.

설정되면 이 파라미터는 Client가 만들거나 업데이트하는 모든 실행의 extra 파라미터의 metadata 필드에 영향을 줍니다. 여기에는 @traceable 데코레이터나 LangChain 통합을 통해 만든 실행도 포함됩니다.

모든 메타데이터 숨기기

hide_metadata=True로 설정하면 LangSmith로 보내는 실행에서 모든 메타데이터를 완전히 제거합니다:

from langsmith import Client

# Hide all metadata completely
client = Client(hide_metadata=True)

# Now when you create runs, metadata will be empty
client.create_run(
    "my_run",
    inputs={"question": "What is 2+2?"},
    run_type="llm",
    extra={"metadata": {"user_id": "123", "session": "abc"}}
)
# The metadata sent to LangSmith will be {} instead of the provided metadata

커스텀 변환

호출 가능 함수를 사용해 LangSmith로 보내기 전에 메타데이터를 선택적으로 필터링, 삭제, 수정합니다:

# Remove sensitive keys
def hide_sensitive_metadata(metadata: dict) -> dict:
    return {k: v for k, v in metadata.items() if not k.startswith("_private")}

client = Client(hide_metadata=hide_sensitive_metadata)

# Redact specific values
def redact_emails(metadata: dict) -> dict:
    import re
    result = {}
    for k, v in metadata.items():
        if isinstance(v, str) and "@" in v:
            result[k] = "[REDACTED_EMAIL]"
        else:
            result[k] = v
    return result

client = Client(hide_metadata=redact_emails)

# Add transformation marker
def add_marker(metadata: dict) -> dict:
    return {**metadata, "transformed": True}

client = Client(hide_metadata=add_marker)

입력과 출력의 규칙 기반 마스킹

정보: 이 기능은 다음 LangSmith SDK 버전에서 사용할 수 있습니다:

  • Python: 0.1.81 이상
  • TypeScript: 0.1.33 이상

입력과 출력의 특정 데이터를 마스킹하려면 create_anonymizer / createAnonymizer 함수를 사용하고 새로 만든 익명화기를 Client를 인스턴스화할 때 전달합니다. 익명화기는 regex 패턴과 교체 값의 목록으로 구성하거나 문자열을 수락하고 반환하는 함수로 구성할 수 있습니다.

팁: API 키, 토큰 및 기타 자격 증명 삭제에 대해서는 즉시 사용 가능한 regex 패턴과 레시피가 있는 트레이스에서 시크릿 삭제를 참고하세요.

LANGSMITH_HIDE_INPUTS = true이면 익명화기는 입력에 대해 건너뜁니다. 출력에 대해서도 LANGSMITH_HIDE_OUTPUTS = true가 같습니다.

그러나 입력이나 출력이 Client로 보내질 경우 anonymizer 메서드는 hide_inputs와 hide_outputs에 있는 함수보다 우선합니다. 기본적으로 create_anonymizer는 최대 10개 중첩 수준까지만 보며, max_depth 파라미터로 구성할 수 있습니다.

from langsmith.anonymizer import create_anonymizer
from langsmith import Client, traceable
import re

# create anonymizer from list of regex patterns and replacement values
anonymizer = create_anonymizer([
    { "pattern": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}", "replace": "<email-address>" },
    { "pattern": r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", "replace": "<UUID>" }
])

# or create anonymizer from a function
email_pattern = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}")
uuid_pattern = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}")
anonymizer = create_anonymizer(
    lambda text: email_pattern.sub("<email-address>", uuid_pattern.sub("<UUID>", text))
)

client = Client(anonymizer=anonymizer)

@traceable(client=client)
def main(inputs: dict) -> dict:
    ...

import { createAnonymizer } from "langsmith/anonymizer"
import { traceable } from "langsmith/traceable"
import { Client } from "langsmith"

// create anonymizer from list of regex patterns and replacement values
const anonymizer = createAnonymizer([
    { pattern: /[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}/g, replace: "<email>" },
    { pattern: /[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}/g, replace: "<uuid>" }
])

// or create anonymizer from a function
const anonymizer = createAnonymizer((value) => value.replace("...", "<value>"))

const client = new Client({ anonymizer })

const main = traceable(async (inputs: any) => {
    // ...
}, { client })

익명화기는 처리를 위해 페이로드를 JSON으로 직렬화하므로, 복잡한 정규식이나 큰 페이로드에서 성능 저하가 발생할 수 있습니다.

참고: anonymizer API의 성능 개선은 로드맵에 있습니다! 성능 문제가 발생하면 support.langchain.com을 통해 지원에 문의하세요.

이전 버전의 LangSmith SDK는 hide_inputs와 hide_outputs 파라미터를 사용해 같은 효과를 낼 수 있습니다. 이 파라미터를 사용해 입력과 출력을 더 효율적으로 처리할 수도 있습니다.

import re
from langsmith import Client, traceable

# Define the regex patterns for email addresses and UUIDs
EMAIL_REGEX = r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}"
UUID_REGEX = r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}"

def replace_sensitive_data(data, depth=10):
    if depth == 0:
        return data
    if isinstance(data, dict):
        return {k: replace_sensitive_data(v, depth-1) for k, v in data.items()}
    elif isinstance(data, list):
        return [replace_sensitive_data(item, depth-1) for item in data]
    elif isinstance(data, str):
        data = re.sub(EMAIL_REGEX, "<email-address>", data)
        data = re.sub(UUID_REGEX, "<UUID>", data)
        return data
    else:
        return data

client = Client(
    hide_inputs=lambda inputs: replace_sensitive_data(inputs),
    hide_outputs=lambda outputs: replace_sensitive_data(outputs)
)

inputs = {"role": "user", "content": "Hello! My email is [email protected] and my ID is 123e4567-e89b-12d3-a456-426614174000."}
outputs = {"role": "assistant", "content": "Hi! I've noted your email as [email protected] and your ID as 123e4567-e89b-12d3-a456-426614174000."}

@traceable(client=client)
def child(inputs: dict) -> dict:
    return outputs

@traceable(client=client)
def parent(inputs: dict) -> dict:
    child_outputs = child(inputs)
    return child_outputs

parent(inputs)

import { Client } from "langsmith";
import { traceable } from "langsmith/traceable";

// Define the regex patterns for email addresses and UUIDs
const EMAIL_REGEX = /[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}/g;
const UUID_REGEX = /[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}/g;

function replaceSensitiveData(data: any, depth: number = 10): any {
    if (depth === 0) return data;
    if (typeof data === "object" && !Array.isArray(data)) {
        const result: Record<string, any> = {};
        for (const [key, value] of Object.entries(data)) {
            result[key] = replaceSensitiveData(value, depth - 1);
        }
        return result;
    } else if (Array.isArray(data)) {
        return data.map(item => replaceSensitiveData(item, depth - 1));
    } else if (typeof data === "string") {
        return data.replace(EMAIL_REGEX, "<email-address>").replace(UUID_REGEX, "<UUID>");
    } else {
        return data;
    }
}

const langsmithClient = new Client({
    hideInputs: (inputs) => replaceSensitiveData(inputs),
    hideOutputs: (outputs) => replaceSensitiveData(outputs)
});

const inputs = {
    role: "user",
    content: "Hello! My email is [email protected] and my ID is 123e4567-e89b-12d3-a456-426614174000."
};
const outputs = {
    role: "assistant",
    content: "Hi! I've noted your email as <email-address> and your ID as <UUID>."
};

const child = traceable(async (inputs: any) => {
    return outputs;
}, { name: "child", client: langsmithClient });

const parent = traceable(async (inputs: any) => {
    const childOutputs = await child(inputs);
    return childOutputs;
}, { name: "parent", client: langsmithClient });

await parent(inputs);

단일 함수의 입력 및 출력 처리

정보: process_outputs 파라미터는 Python용 LangSmith SDK 버전 0.1.98 이상에서 사용할 수 있습니다.

Client 수준 입력·출력 처리에 더해, LangSmith는 @traceable 데코레이터의 process_inputs와 process_outputs 파라미터를 통해 함수 수준 처리를 제공합니다.

이 파라미터들은 특정 함수의 입력과 출력을 LangSmith에 기록하기 전에 변환하게 하는 함수를 수락합니다. 페이로드 크기 줄이기, 민감한 정보 제거, 특정 함수가 LangSmith에서 객체를 직렬화·표현하는 방법 사용자 지정에 유용합니다.

process_inputs와 process_outputs 사용 예:

from langsmith import traceable

def process_inputs(inputs: dict) -> dict:
    # inputs is a dictionary where keys are argument names and values are the provided arguments
    # Return a new dictionary with processed inputs
    return {
        "processed_key": inputs.get("my_cool_key", "default"),
        "length": len(inputs.get("my_cool_key", ""))
    }

def process_outputs(output: Any) -> dict:
    # output is the direct return value of the function
    # Transform the output into a dictionary
    # In this case, "output" will be an integer
    return {"processed_output": str(output)}

@traceable(process_inputs=process_inputs, process_outputs=process_outputs)
def my_function(my_cool_key: str) -> int:
    # Function implementation
    return len(my_cool_key)

result = my_function("example")

이 예제에서 process_inputs는 처리된 입력 데이터로 새 딕셔너리를 만들고, process_outputs는 LangSmith에 기록하기 전에 출력을 특정 형식으로 변환합니다.

경고: 프로세서 함수에서 원본 객체를 변형하는 것은 피하는 것이 좋습니다. 대신 처리된 데이터로 새 객체를 만들고 반환하세요.

비동기 함수의 사용법도 비슷합니다:

@traceable(process_inputs=process_inputs, process_outputs=process_outputs)
async def async_function(key: str) -> int:
    # Async implementation
    return len(key)

이 함수 수준 프로세서는 둘 다 정의되면 Client 수준 프로세서(hide_inputs와 hide_outputs)보다 우선합니다.

예제

규칙 기반 마스킹을 다양한 익명화기와 결합해 입력과 출력에서 민감한 정보를 제거할 수 있습니다. 다음 예제는 regex, Microsoft Presidio, Amazon Comprehend 작업을 다룹니다.

Regex

정보: 아래 구현은 완전하지 않으며 일부 형식이나 가장자리 케이스를 놓칠 수 있습니다. 프로덕션에서 사용하기 전에 구현을 철저히 테스트하세요.

regex를 사용해 LangSmith로 보내기 전에 입력과 출력을 마스킹할 수 있습니다. 아래 구현은 이메일 주소, 전화번호, 전체 이름, 신용카드 번호, SSN을 마스킹합니다.

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

# Define regex patterns for various 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):
    """
    Anonymize sensitive information in the text using regex patterns.
    Args:
        text (str): The input text to be anonymized.
    Returns:
        str: The anonymized text.
    """
    # Replace sensitive information with placeholders
    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):
    """
    Recursively traverse the data structure and anonymize sensitive information.
    Args:
        data (any): The input data to be anonymized.
        depth (int): The current recursion depth to prevent excessive recursion.
    Returns:
        any: The anonymized data.
    """
    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())

# Initialize the LangSmith Client with the anonymization functions
langsmith_client = Client(
    hide_inputs=recursive_anonymize, hide_outputs=recursive_anonymize
)

# The trace produced will have its metadata present, but the inputs and outputs will be anonymized
response_with_anonymization = openai_client.chat.completions.create(
    model="gpt-5.4-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},
)

# The trace produced will not have anonymized inputs and outputs
response_without_anonymization = openai_client.chat.completions.create(
    model="gpt-5.4-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."},
    ],
)

Microsoft Presidio

정보: 아래 구현은 사용자와 LLM 사이에 교환되는 메시지의 민감한 정보를 익명화하는 일반적인 예를 제공합니다. 완전하지 않으며 모든 경우를 고려하지 않습니다. 프로덕션에서 사용하기 전에 구현을 철저히 테스트하세요.

Microsoft Presidio는 데이터 보호 및 비식별화 SDK입니다. 아래 구현은 Presidio를 사용해 LangSmith로 보내기 전에 입력과 출력을 익명화합니다. 최신 정보는 Presidio의 공식 문서를 참고하세요.

Presidio와 그 spaCy 모델을 사용하려면 다음을 설치합니다:

pip install presidio-analyzer
pip install presidio-anonymizer
python -m spacy download en_core_web_lg

uv add presidio-analyzer
uv add presidio-anonymizer
python -m spacy download en_core_web_lg

OpenAI도 설치합니다:

pip install openai

uv add openai

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):
    """
    Anonymize sensitive information sent by the user or returned by the model.
    Args:
        data (any): The data to be anonymized.
    Returns:
        any: The anonymized data.
    """
    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())

# initialize the langsmith Client with the anonymization functions
langsmith_client = Client(
  hide_inputs=presidio_anonymize, hide_outputs=presidio_anonymize
)

# The trace produced will have its metadata present, but the inputs and outputs will be anonymized
response_with_anonymization = openai_client.chat.completions.create(
  model="gpt-5.4-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},
)

# The trace produced will not have anonymized inputs and outputs
response_without_anonymization = openai_client.chat.completions.create(
  model="gpt-5.4-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]"},
  ],
)

Amazon Comprehend

정보: 아래 구현은 Amazon Comprehend를 사용해 민감한 정보를 익명화하는 일반적인 예를 제공합니다. 완전하지 않으며 모든 실행 형태나 콘텐츠 유형을 고려하지 않습니다. 프로덕션에서 사용하기 전에 구현을 철저히 테스트하세요.

Comprehend는 개인 식별 정보를 감지할 수 있는 자연어 처리 서비스입니다. 아래 구현은 Comprehend의 DetectPiiEntities API를 사용해 LangSmith로 보내기 전에 입력과 출력을 익명화합니다.

경고: Amazon Comprehend는 일괄 PII 엔드포인트를 제공하지 않습니다. detect_pii_entities는 호출당 문서 하나를 처리하고, batch_detect_entities는 일반 엔티티(사람, 장소, 조직)를 감지합니다 — PII 전용 API가 아닙니다. 많은 문서를 처리하려면 detect_pii_entities를 루프로 호출하거나(아래와 같음) S3의 문서에 대해 비동기 start_pii_entities_detection_job을 사용하세요.

Comprehend를 사용하려면 boto3를 설치합니다:

pip install boto3

uv add boto3

OpenAI도 설치합니다:

pip install openai

uv add openai

AWS에서 자격 증명을 설정하고 AWS CLI로 인증해야 합니다. AWS Comprehend 설정 지침을 따르세요.

아래 예제는 전체 입력/출력 페이로드를 재귀적으로 순회하며 모든 문자열을 Comprehend에 통과시킵니다. 페이로드를 순회하는 것은(data["messages"]나 data["choices"][0]["message"]만 검사하는 대신) chat-completion 형태의 실행뿐 아니라 모든 형태의 span에서 PII가 삭제되게 합니다.

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

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

# Skip strings shorter than this — Comprehend rarely finds PII in them
# and skipping reduces API calls significantly.
MIN_LENGTH_FOR_DETECTION = 3
# Comprehend's per-call text limit is 100 KB of UTF-8.
MAX_BYTES_PER_CALL = 100_000

def redact_pii_entities(text: str, entities: list) -> str:
    """Replace each detected entity with `[TYPE]`, working back-to-front to preserve offsets."""
    for entity in sorted(entities, key=lambda e: e["BeginOffset"], reverse=True):
        begin, end = entity["BeginOffset"], entity["EndOffset"]
        text = text[:begin] + f"[{entity['Type']}]" + text[end:]
    return text

def detect_and_redact(text: str) -> str:
    """Detect PII in `text` with Comprehend and return the redacted version."""
    if not text or len(text) < MIN_LENGTH_FOR_DETECTION:
        return text
    if len(text.encode("utf-8")) > MAX_BYTES_PER_CALL:
        # Skip oversized strings — chunk them yourself if you need to redact them.
        return text
    try:
        response = comprehend.detect_pii_entities(Text=text, LanguageCode="en")
    except Exception as e:
        print(f"Comprehend error: {e}")
        return text
    entities = response.get("Entities", [])
    return redact_pii_entities(text, entities) if entities else text

def comprehend_anonymize(data, depth: int = 10):
    """
    Recursively walk a payload and redact PII in every string we find.
    This works on arbitrary run shapes — chat messages, tool inputs/outputs,
    retrieval results, custom run types, etc.
    """
    if depth == 0:
        return data
    if isinstance(data, dict):
        return {k: comprehend_anonymize(v, depth - 1) for k, v in data.items()}
    if isinstance(data, list):
        return [comprehend_anonymize(item, depth - 1) for item in data]
    if isinstance(data, str):
        return detect_and_redact(data)
    return data

openai_client = wrap_openai(openai.Client())

# initialize the langsmith Client with the anonymization functions
langsmith_client = Client(
    hide_inputs=comprehend_anonymize, hide_outputs=comprehend_anonymize
)

# The trace produced will have its metadata present, but the inputs and outputs will be anonymized
response_with_anonymization = openai_client.chat.completions.create(
    model="gpt-5.4-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},
)

# The trace produced will not have anonymized inputs and outputs
response_without_anonymization = openai_client.chat.completions.create(
    model="gpt-5.4-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]"},
    ],
)

참고: comprehend_anonymize는 만나는 문자열마다 Comprehend API 호출을 하나씩 발생시키므로, 높은 처리량 트레이스에서 rate limit에 걸릴 수 있습니다. 더 높은 처리량에는 아래 일괄 처리 접근 방식을 사용하거나, contains_pii_entities로 사전 필터링하고 PII가 있을 때만 detect_pii_entities를 호출하세요.

Comprehend와 함께 create_anonymizer 사용

LangSmith의 내장 create_anonymizer 헬퍼(재귀 순회를 이미 처리하고 langchain 에이전트 미들웨어와 깔끔하게 통합됨)를 선호한다면, 직접 순회를 작성하는 대신 Comprehend 기반 callable을 전달할 수 있습니다:

from langsmith import Client
from langsmith.anonymizer import create_anonymizer

def comprehend_replacer(value: str, path: list) -> str:
    return detect_and_redact(value)

client = Client(anonymizer=create_anonymizer(comprehend_replacer))

이것은 위의 comprehend_anonymize와 동일한 삭제 동작을 보일러플레이트 없이 제공하며, 스택의 다른 곳에서도 create_anonymizer를 사용한다면 권장되는 접근 방식입니다.

높은 처리량 마스킹을 위한 일괄 처리

정보: process_buffered_run_ops는 Python SDK에서만 사용할 수 있습니다.

이 페이지의 이전 접근 방식들은 각 실행을 개별적으로 처리합니다. 마스킹 로직에 rate-limited API나 모델 추론(Presidio나 Amazon Comprehend 예제처럼)이 포함된다면, 실행을 한 번에 하나씩 처리하면 병목이 될 수 있습니다. process_buffered_run_ops는 직렬화되어 API로 보내지기 전에 원시 실행 dict 일괄을 가로채게 하므로, 비용을 여러 실행에 한 번에 분산할 수 있습니다. LangSmith는 이 실행들을 백그라운드 스레드에서 처리하므로 애플리케이션을 차단하지 않습니다.

LangSmith는 실행을 메모리 버퍼에 보관하고 다음 중 하나일 때 일괄로 플러시합니다:

  • run_ops_buffer_size 실행 작업이 누적되었거나,
  • 마지막 실행이 추가된 이후 run_ops_buffer_timeout_ms 밀리초가 경과했거나(기본값: 5000ms).

함수는 원시 실행 dict 목록으로 일괄을 받고, 같은 길이, 같은 순서, 실행 ID가 변경되지 않은 목록을 반환해야 합니다. 두 제약 중 하나라도 위반하면 ValueError가 발생합니다.

참고: run_ops_buffer_size는 고유 실행이 아니라 개별 실행 작업(operations) 을 셉니다. 각 트레이싱 호출은 일반적으로 두 개의 작업을 생성합니다: 생성(실행 시작 시)과 업데이트(출력과 함께 종료 시). 버퍼 크기를 그에 맞게 설정하세요. 예를 들어 run_ops_buffer_size=1000은 약 500번의 트레이싱 호출을 버퍼링합니다. 이 때문에 같은 실행 ID가 단일 일괄에서 두 번 나타날 수 있습니다: 입력이 있는 것 한 번, 출력이 있는 것 한 번.

경고: 버퍼는 크기 한도에 도달하거나 타임아웃이 경과했을 때만 자동 플러시됩니다. 버퍼링된 실행이 유실되지 않도록 프로그램이 종료되기 전에 항상 client.flush()를 호출하세요.

일괄의 각 실행 dict는 생성 작업(inputs 포함, 실행 시작 시 전송) 또는 업데이트 작업(outputs 포함, 종료 시 전송)입니다. 단일 트레이싱 호출에 대한 전형적인 쌍은 다음과 같습니다:

# Create op — sent when the run starts
{
    "id": "018f1b2c-...",
    "name": "my_llm_call",
    "run_type": "llm",
    "inputs": {"messages": [{"role": "user", "content": "My name is Jane Smith..."}]},
    "start_time": "2024-01-01T00:00:00.000Z",
    "trace_id": "018f1b2c-...",
    "dotted_order": "20240101T000000000000Z018f1b2c-...",
    "extra": {"metadata": {}, "runtime": {...}},
    "session_name": "default",
}

# Update op — sent when the run ends (same id, adds outputs)
{
    "id": "018f1b2c-...",
    "outputs": {"choices": [{"message": {"role": "assistant", "content": "Hello Jane..."}}]},
    "end_time": "2024-01-01T00:00:01.000Z",
    "trace_id": "018f1b2c-...",
    "dotted_order": "20240101T000000000000Z018f1b2c-...",
}

경고: Amazon Comprehend에는 일괄 PII 엔드포인트가 없습니다. batch_detect_entities는 PII가 아닌 일반 엔티티(사람, 장소, 조직)를 감지하며 detect_pii_entities의 대체물이 아닙니다. 아래의 일괄 최적화는 실행 간 동일한 문자열의 중복 제거와 실행당 버퍼링 오버헤드 분산에서 비롯됩니다 — 단일 대량 API 호출이 아닙니다. 진정한 대량 PII 감지가 필요하면 트레이싱 경로 밖에서 S3의 문서에 대해 비동기 start_pii_entities_detection_job을 실행하세요.

다음 예제는 버퍼 전체의 모든 문자열을 수집하고 중복을 제거한 뒤, 고유 문자열당 한 번씩 Comprehend의 detect_pii_entities 엔드포인트를 호출합니다. 공통 프롬프트, 시스템 메시지, 도구 입력을 반복하는 트래픽의 경우 실행별 접근 방식에 비해 Comprehend 호출 수를 크게 줄일 수 있습니다. 이 예제는 전체 페이로드를 재귀적으로 순회하므로 어떤 실행 형태(채팅 메시지, 도구 입력/출력, retriever span 등)에서도 동작합니다.

import boto3
from langsmith import Client, traceable

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

MIN_LENGTH_FOR_DETECTION = 3
MAX_BYTES_PER_CALL = 100_000

def redact_entities(text: str, entities: list) -> str:
    for entity in sorted(entities, key=lambda e: e["BeginOffset"], reverse=True):
        placeholder = f"[{entity['Type']}]"
        text = text[:entity["BeginOffset"]] + placeholder + text[entity["EndOffset"]:]
    return text

def detect_and_redact(text: str) -> str:
    if len(text) < MIN_LENGTH_FOR_DETECTION:
        return text
    if len(text.encode("utf-8")) > MAX_BYTES_PER_CALL:
        return text
    try:
        response = comprehend.detect_pii_entities(Text=text, LanguageCode="en")
    except Exception as e:
        print(f"Comprehend error: {e}")
        return text
    entities = response.get("Entities", [])
    return redact_entities(text, entities) if entities else text

def _redact_in_place(data, cache: dict, depth: int = 10):
    """Recursively redact every string in `data`, using `cache` to dedupe API calls."""
    if depth == 0:
        return data
    if isinstance(data, dict):
        return {k: _redact_in_place(v, cache, depth - 1) for k, v in data.items()}
    if isinstance(data, list):
        return [_redact_in_place(item, cache, depth - 1) for item in data]
    if isinstance(data, str):
        if data not in cache:
            cache[data] = detect_and_redact(data)
        return cache[data]
    return data

def comprehend_anonymize_batch(runs: list[dict]) -> list[dict]:
    # One shared cache per buffer flush. Identical strings (e.g. the same
    # system prompt repeated across runs) only hit Comprehend once.
    # Note: the same run ID may appear twice — once as a create (with inputs)
    # and once as an update (with outputs).
    cache: dict[str, str] = {}
    for run in runs:
        for field in ("inputs", "outputs"):
            data = run.get(field)
            if isinstance(data, (dict, list)):
                run[field] = _redact_in_place(data, cache)
    return runs

client = Client(
    process_buffered_run_ops=comprehend_anonymize_batch,
    run_ops_buffer_size=1000,        # ~500 traced calls (2 ops each: create + update)
    run_ops_buffer_timeout_ms=3000,  # or after 3 seconds, whichever comes first
)

@traceable(client=client)
def my_llm_call(messages: list) -> dict:
    # ... your LLM call ...
    pass

try:
    my_llm_call([{"role": "user", "content": "My name is Jane Smith, call me at 555-867-5309"}])
finally:
    client.flush()  # always flush before exit

process_buffered_run_ops와 run_ops_buffer_size는 항상 함께 설정되어야 합니다 — 하나만 제공하면 ValueError가 발생합니다.

더 알아보기