로그 데이터 scrub하기 — Langfuse로 보내기 전에 PII를 가리자

관측성 도구(Langfuse 같은)에 로그를 보내다 보면 어느 순간 찜찜해져요. 여기에 우리 사용자의 이메일, 전화번호, 주소 같은 개인정보가 그대로 섞여 들어가고 있잖아. 로깅 통합으로 데이터를 보내기 전에 메시지를 수정하거나 PII를 마스킹할 수 있다면, 그 걱정이 확 줄어듭니다. LiteLLM은 그걸 위해 로깅 훅을 제공합니다.

출처: 공식문서 — Scrub Logged Data

이 가이드의 접근 방식은 커스텀 콜백을 하나 만들어서 요청/응답이 로깅되기 직전에 가리는 것입니다. LiteLLM이 실제로 PII 마스킹에 쓰는 Presidio 예시도 참고가 되니 함께 보면 좋아요. 핵심만 정리하면 동기·비동기 두 가지 훅이 있고, 요청이 로그로 남기 전에 kwargs["messages"]를 수정해 줍니다.

1. 커스텀 콜백 만들기

CustomLogger를 상속받은 핸들러를 만들고, 로깅 훅을 오버라이드합니다. 여기서 메시지를 원하는 값으로 치환해요.

from litellm.integrations.custom_logger import CustomLogger

class MyCustomHandler(CustomLogger):
    async def async_logging_hook(
        self, kwargs: dict, result: Any, call_type: str
    ) -> Tuple[dict, Any]:
        """
        로깅되는 요청/응답을 마스킹. 요청/result의 수정본을 반환.
        `async_log_success_event` 전에 호출됨.
        """
        if (
            call_type == "completion" or call_type == "acompletion"
        ):  # /chat/completions 요청
            messages = kwargs.get("messages", None)
            kwargs["messages"] = [{"role": "user", "content": "MASK_THIS_ASYNC_VALUE"}]
        return kwargs, responses

    def logging_hook(
        self, kwargs: dict, result: Any, call_type: str
    ) -> Tuple[dict, Any]:
        """
        로깅되는 요청/응답을 마스킹. 요청/result의 수정본을 반환.
        `log_success_event` 전에 호출됨.
        """
        if (
            call_type == "completion" or call_type == "acompletion"
        ):  # /chat/completions 요청
            messages = kwargs.get("messages", None)
            kwargs["messages"] = [{"role": "user", "content": "MASK_THIS_SYNC_VALUE"}]
        return kwargs, responses


customHandler = MyCustomHandler()

함수 시그니처 중 주석이 있는 두 줄 — "Called before async_log_success_event" / "Called before log_success_event" — 이 핵심이에요. 이 훅은 성공 이벤트가 로깅되기 바로 전에 불려서, 반환한 수정본이 실제로 로그에 남습니다.

2. 커스텀 핸들러 연결하기

만든 핸들러를 LiteLLM의 콜백으로 등록합니다.

import litellm

litellm.callbacks = [customHandler]

3. 테스트해 보기

Langfuse 콜백과 함께 실제 호출을 날려, 두 방식 모두 마스킹이 적용되는지 확인할 수 있어요.

# uv add langfuse

import os
import litellm
from litellm import completion

os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
os.environ["LANGFUSE_HOST"]  # 선택, 기본은 https://cloud.langfuse.com
# LLM API Keys
os.environ['OPENAI_API_KEY']=""

litellm.callbacks = [customHandler]
litellm.success_callback = ["langfuse"]

## sync
response = completion(model="gpt-3.5-turbo", messages=[{ "role": "user", "content": "Hi 👋 - i'm openai"}],
                              stream=True)
for chunk in response:
    continue

## async
import asyncio

def async completion():
    response = await acompletion(model="gpt-3.5-turbo", messages=[{ "role": "user", "content": "Hi 👋 - i'm openai"}],
                              stream=True)
    async for chunk in response:
        continue
asyncio.run(completion())

이 예시는 메시지를 고정 문자열로 바꾸는 단순한 형태지만, 실제로는 여기서 정규식이나 Presidio 같은 도구로 PII 패턴을 찾아 가리는 로직을 넣으면 됩니다. 핵심 흐름은 같아요 — 로그로 나가기 전에 요청을 손보는 거요.

로그 데이터 scrub하기 — Langfuse로 보내기 전에 PII를 가리자